Pitstop

Instructions

Implementation Reference

• tools/telemetry/lap_telemetry.py:10-62 (handler)

  Core implementation of the get_lap_telemetry tool. Fetches FastF1 session data, loads lap telemetry, processes DataFrame into list of TelemetryPoint models, and returns structured LapTelemetryResponse.

  def get_lap_telemetry(year: int, gp: Union[str, int], session: str, driver: Union[str, int], lap_number: int) -> LapTelemetryResponse:
      """
      Get high-frequency telemetry for a lap - speed, throttle, brake, gear, RPM, DRS.
  
      Args:
          year: Season year (2018+)
          gp: Grand Prix name or round
          session: 'FP1', 'FP2', 'FP3', 'Q', 'S', 'R'
          driver: Driver code or number
          lap_number: Specific lap number
  
      Returns:
          LapTelemetryResponse with telemetry points
  
      Example:
          get_lap_telemetry(2024, "Monza", "R", "VER", 15) → VER lap 15 telemetry
      """
      session_obj = fastf1_client.get_session(year, gp, session)
      session_obj.load(laps=True, telemetry=True, weather=False, messages=False)
  
      event = session_obj.event
  
      driver_laps = session_obj.laps.pick_drivers(driver)
      lap = driver_laps[driver_laps['LapNumber'] == lap_number].iloc[0]
      telemetry_df = lap.get_telemetry()
  
      # Convert to Pydantic models
      telemetry_points = []
      for idx, row in telemetry_df.iterrows():
          point = TelemetryPoint(
              session_time=str(row['SessionTime']) if pd.notna(row.get('SessionTime')) else None,
              distance=float(row['Distance']) if pd.notna(row.get('Distance')) else None,
              speed=float(row['Speed']) if pd.notna(row.get('Speed')) else None,
              rpm=float(row['RPM']) if pd.notna(row.get('RPM')) else None,
              n_gear=int(row['nGear']) if pd.notna(row.get('nGear')) else None,
              throttle=float(row['Throttle']) if pd.notna(row.get('Throttle')) else None,
              brake=float(row['Brake']) if pd.notna(row.get('Brake')) else None,
              drs=int(row['DRS']) if pd.notna(row.get('DRS')) else None,
              x=float(row['X']) if pd.notna(row.get('X')) else None,
              y=float(row['Y']) if pd.notna(row.get('Y')) else None,
              z=float(row['Z']) if pd.notna(row.get('Z')) else None,
          )
          telemetry_points.append(point)
  
      return LapTelemetryResponse(
          session_name=session_obj.name,
          event_name=event['EventName'],
          driver=str(lap['Driver']),
          lap_number=lap_number,
          lap_time=str(lap['LapTime']) if pd.notna(lap.get('LapTime')) else None,
          telemetry=telemetry_points,
          total_points=len(telemetry_points)
      )

• models/telemetry/telemetry.py:21-30 (schema)

  Pydantic model defining the output schema for get_lap_telemetry response.

  class LapTelemetryResponse(BaseModel):
      """Lap telemetry response."""
  
      session_name: str = Field(description="Session name")
      event_name: str = Field(description="Grand Prix name")
      driver: str = Field(description="Driver abbreviation")
      lap_number: int = Field(description="Lap number")
      lap_time: Optional[str] = Field(None, description="Lap time")
      telemetry: list[TelemetryPoint] = Field(description="Telemetry data points")
      total_points: int = Field(description="Total number of telemetry points")

• models/telemetry/telemetry.py:5-19 (schema)

  Pydantic model for individual telemetry data points used in lap telemetry response.

  class TelemetryPoint(BaseModel):
      """Single telemetry data point."""
  
      session_time: Optional[str] = Field(None, description="Session time")
      distance: Optional[float] = Field(None, description="Distance in meters")
      speed: Optional[float] = Field(None, description="Speed in km/h")
      rpm: Optional[float] = Field(None, description="Engine RPM")
      n_gear: Optional[int] = Field(None, description="Current gear (1-8)")
      throttle: Optional[float] = Field(None, description="Throttle position (0-100%)")
      brake: Optional[float] = Field(None, description="Brake application (0-100% or boolean)")
      drs: Optional[int] = Field(None, description="DRS status")
      x: Optional[float] = Field(None, description="X position coordinate")
      y: Optional[float] = Field(None, description="Y position coordinate")
      z: Optional[float] = Field(None, description="Z position coordinate")

• server.py:158-158 (registration)

  MCP tool registration for get_lap_telemetry function.

  mcp.tool()(get_lap_telemetry)

• tools/__init__.py:13-17 (registration)

  Re-export of get_lap_telemetry from tools/telemetry submodule for convenient import in server.py.

  from .telemetry import (
      get_lap_telemetry,
      compare_driver_telemetry,
  )