Function bodies 217 total

def _extract_driver_position_data(
    driver_abbr: str,
    session: Session,
    ax: plt.Axes,
) -> dict | None:
    """Extract and plot position data for a single driver.

    Args:
        driver_abbr: Driver abbreviation (e.g., 'VER', 'HAM')
        session: FastF1 session object
        ax: Matplotlib axes to plot on

    Returns:
        Dictionary with driver statistics, or None if driver should be excluded
    """
    # Compute stats using shared utility
    stats = compute_driver_position_stats(driver_abbr, session)
    if stats is None:
        return None

    driver_laps = session.laps.pick_drivers(driver_abbr)
    position_data = driver_laps[["LapNumber", "Position"]].copy()
    position_data = position_data.dropna(subset=["Position"])

    # Get driver color from FastF1
    color = get_driver_color_safe(driver_abbr, session)

    # Plot position evolution
    ax.plot(
        position_data["LapNumber"],
        position_data["Position"],
        label=driver_abbr,
      

def _configure_position_plot(ax: plt.Axes, session: Session, year: int) -> None:
    """Configure plot axes, labels, and styling.

    Args:
        ax: Matplotlib axes to configure
        session: FastF1 session object
        year: Season year
    """
    ax.set_xlabel("Lap Number")
    ax.set_ylabel("Position")
    ax.set_title(f"{session.event['EventName']} {year} - Position Changes")

    # Invert y-axis so P1 is at the top
    ax.invert_yaxis()

    # Set y-axis to show all positions as integers
    max_position = int(ax.get_ylim()[0])  # After inversion, top limit is maximum
    ax.set_yticks(range(1, max_position + 1))

    # Add legend outside plot area
    ax.legend(
        loc="center left",
        bbox_to_anchor=(1, 0.5),
        framealpha=ALPHA_VALUE,
        title="Driver (▼ = Pit Stop)",
    )

    ax.grid(True, alpha=GRID_ALPHA, axis="both")

def _calculate_aggregate_statistics(stats: list[dict]) -> dict:
    """Calculate aggregate statistics from per-driver stats.

    Args:
        stats: List of per-driver statistics dictionaries

    Returns:
        Dictionary with aggregate statistics
    """
    total_overtakes = sum(s["overtakes"] for s in stats)
    total_position_changes = sum(abs(s["net_change"]) for s in stats)
    avg_volatility = sum(s["volatility"] for s in stats) / len(stats) if stats else 0
    mean_pit_stops = sum(s["pit_stops"] for s in stats) / len(stats) if stats else 0

    return {
        "total_overtakes": total_overtakes,
        "total_position_changes": total_position_changes,
        "average_volatility": round(avg_volatility, 2),
        "mean_pit_stops": round(mean_pit_stops, 2),
        "drivers": stats,
    }

def generate_position_changes_chart(
    year: int,
    gp: str,
    session_type: str,
    drivers: list[str] | None = None,
    top_n: int | None = None,
    workspace_dir: Path | None = None,
    test_number: int | None = None,
    session_number: int | None = None,
) -> dict:
    """Generate a position changes visualization.

    Args:
        year: Season year
        gp: Grand Prix name (ignored for testing sessions)
        session_type: Session identifier (typically 'R' for race, ignored for testing)
        drivers: Optional list of driver abbreviations to filter
        top_n: Optional number of top finishing drivers to show
        workspace_dir: Workspace directory for outputs and cache
        test_number: Testing event number (e.g., 1 or 2)
        session_number: Session within testing event (e.g., 1, 2, or 3)

    Returns:
        Dictionary with chart metadata and position change statistics
    """
    # Determine paths from workspace
    if test_number is not None and

def _fetch_per_round_points(
    year: int,
    schedule: pd.DataFrame,
    summary_type: str,
) -> tuple[pd.DataFrame, pd.DataFrame, pd.Series, list[str], pd.Series]:
    """Fetch per-round points (and finishing positions) via FastF1.

    Args:
        year: Championship year
        schedule: FastF1 event schedule DataFrame (from fastf1.get_event_schedule)
        summary_type: "drivers" or "constructors"

    Returns:
        Tuple of:
        - points_df: pivot of competitor × round → points, sorted ascending by
          total (champion sits at the top of the heatmap)
        - position_df: pivot of driver × round → finishing position (drivers
          mode only; empty DataFrame for constructors)
        - total_points: Series of total season points per competitor, same order
        - short_event_names: short race name strings in round-number order
        - competitor_teams: Series mapping competitor name → team name
          (for drivers mode: driver abbreviation → team; for

def _build_season_heatmap(
    points_df: pd.DataFrame,
    position_df: pd.DataFrame,
    total_points: pd.Series,
    short_event_names: list[str],
    year: int,
    summary_type: str,
) -> go.Figure:
    """Build the two-panel Plotly heatmap and return the figure.

    Left panel (85 %): per-round points with hover showing finishing position.
    Right panel (15 %): total season points.
    """
    n_rounds = len(points_df.columns)
    x_labels = short_event_names[:n_rounds]
    type_label = "Drivers'" if summary_type == "drivers" else "Constructors'"
    vmax_main = float(points_df.values.max()) if points_df.size > 0 else 25.0

    # Build per-cell hover customdata (list-of-lists of dicts with position)
    if not position_df.empty:
        hover_info = [
            [
                {"position": position_df.at[driver, race] if race in position_df.columns else "N/A"}
                for race in points_df.columns
            ]
            for driver in points_df.index
        ]
  

def generate_season_summary_chart(
    year: int,
    workspace_dir: Path,
    summary_type: str = "drivers",
) -> dict:
    """Generate an interactive season summary heatmap with per-round points.

    Saves the chart as an HTML file so the Plotly hover tooltips are preserved.

    Args:
        year: Championship year (e.g., 2024)
        summary_type: "drivers" or "constructors"
        workspace_dir: Workspace directory for output files

    Returns:
        Dictionary with chart metadata and season statistics

    Raises:
        ValueError: If summary_type is invalid or no data is available
    """
    summary_type = summary_type.lower()
    if summary_type not in ("drivers", "constructors"):
        raise ValueError(f"Invalid summary_type: {summary_type!r}. Must be 'drivers' or 'constructors'.")

    setup_fastf1_cache()
    schedule = fastf1.get_event_schedule(year, include_testing=False)
    total_race_events = int((schedule["RoundNumber"] > 0).sum())

    points_df, position_

def _format_sector_time(sector_time: pd.Timedelta) -> str | None:
    """Format a sector time as SS.mmm or M:SS.mmm, or None if not available."""
    if pd.isna(sector_time):
        return None
    total_seconds = sector_time.total_seconds()
    minutes = int(total_seconds // 60)
    secs = total_seconds % 60
    if minutes > 0:
        return f"{minutes}:{secs:06.3f}"
    return f"{secs:.3f}"

def _build_merged_telemetry(
    all_telemetry: list[dict],
    channel_keys: list[str],
) -> pd.DataFrame | None:
    """Merge all entries' telemetry onto a common distance grid.

    Returns a DataFrame with columns: Distance, {channel}_{key} for each
    entry/channel pair. Returns None if fewer than 2 entries have data.

    Entry dicts support both new-style (with 'key'/'label') and old-style
    (with 'driver') for backward compatibility.
    """
    if len(all_telemetry) < 2:
        return None

    cols_needed = ["Distance"] + channel_keys
    key0 = _entry_key(all_telemetry[0])
    ref = all_telemetry[0]["telemetry"][cols_needed].copy()
    ref.columns = ["Distance"] + [f"{c}_{key0}" for c in channel_keys]

    for entry in all_telemetry[1:]:
        k = _entry_key(entry)
        other = entry["telemetry"][cols_needed].copy()
        other.columns = ["Distance"] + [f"{c}_{k}" for c in channel_keys]
        ref = pd.merge_asof(
            ref.sort_values("Distance"),
        

def _build_customdata(
    label: str,
    channel_key: str,
    tel: pd.DataFrame,
    merged: pd.DataFrame,
    other_labels: list[str],
    *,
    key: str | None = None,
    other_keys: list[str] | None = None,
) -> np.ndarray:
    """Build customdata array with deltas vs other entries for tooltip display.

    Args:
        label: Display name of this entry (used only for context; column ops use key)
        channel_key: The telemetry channel key (e.g. "Speed")
        tel: This entry's telemetry DataFrame
        merged: Merged telemetry DataFrame from _build_merged_telemetry
        other_labels: Display labels of other entries (for fallback if other_keys not given)
        key: Column-naming key for this entry (defaults to label if not provided)
        other_keys: Column-naming keys of other entries (defaults to other_labels if not provided)
    """
    if merged is None or not other_labels:
        return np.zeros((len(tel), 1))

    own_key = key if key is not None else labe

def _build_hover_template(
    label: str,
    channel: dict,
    other_labels: list[str],
) -> str:
    """Build Plotly hovertemplate string with delta lines for each other entry."""
    fmt = channel["fmt"]
    unit = channel["unit"]
    lines = [
        f"<b>{label}</b>",
        "Distance: %{x:.0f}m",
        f"{channel['label']}: %{{y:{fmt}}}{unit}",
    ]
    for i, other in enumerate(other_labels):
        lines.append(f"\u0394 vs {other}: %{{customdata[{i}]:+.1f}}{unit}")
    lines.append("<extra></extra>")
    return "<br>".join(lines)

def _render_telemetry_chart(
    entries: list[dict],
    circuit_info,
    output_path: Path,
    annotate_corners: bool,
    title: str,
) -> bool:
    """Render a Plotly telemetry chart to HTML and return whether corners were drawn.

    Args:
        entries: List of entry dicts, each with:
            - 'key': column-naming identifier (no spaces)
            - 'label': display label (may contain spaces)
            - 'telemetry': DataFrame with Distance, Speed, RPM, nGear, Throttle, Brake, SuperClip
            - 'color': hex color string
            - 'style': line style dict with 'linestyle' and 'linewidth' keys
        circuit_info: FastF1 circuit info object (for corner annotations), or None
        output_path: Path to write the HTML file
        annotate_corners: Whether to draw corner annotations
        title: Chart title text

    Returns:
        True if corner annotations were successfully drawn, False otherwise
    """
    channel_keys = [ch["key"] for ch in CHANNELS]

def generate_telemetry_chart(
    year: int,
    gp: str,
    session_type: str,
    drivers: list[str],
    workspace_dir: Path,
    annotate_corners: bool = False,
    test_number: int | None = None,
    session_number: int | None = None,
) -> dict:
    """Generate an interactive telemetry comparison chart for fastest laps.

    Args:
        year: Season year
        gp: Grand Prix name (ignored for testing sessions)
        session_type: Session identifier (ignored for testing sessions)
        drivers: List of 2-5 driver abbreviations to compare
        workspace_dir: Workspace directory for outputs and cache
        annotate_corners: Whether to add corner markers and labels
        test_number: Testing event number (e.g., 1 or 2)
        session_number: Session within testing event (e.g., 1, 2, or 3)

    Returns:
        Dictionary with chart metadata and telemetry statistics

    Raises:
        ValueError: If drivers list has <2 or >5 entries
    """
    if len(drivers) < MIN_

def _rotate(xy: np.ndarray, *, angle: float) -> np.ndarray:
    """Rotate 2D coordinates by the given angle in radians.

    Args:
        xy: Array-like of shape (2,) or (N, 2) with X/Y coordinates.
        angle: Rotation angle in radians.

    Returns:
        Rotated coordinates with the same shape as input.
    """
    rot_mat = np.array([[np.cos(angle), np.sin(angle)], [-np.sin(angle), np.cos(angle)]])
    return np.matmul(xy, rot_mat)

def generate_track_map_chart(
    year: int,
    gp: str,
    session_type: str,
    workspace_dir: Path,
    test_number: int | None = None,
    session_number: int | None = None,
) -> dict:
    """Generate a track map with numbered corner labels.

    Args:
        year: Season year
        gp: Grand Prix name (ignored for testing sessions)
        session_type: Session identifier (ignored for testing sessions)
        workspace_dir: Workspace directory for outputs and cache
        test_number: Testing event number (e.g., 1 or 2)
        session_number: Session within testing event (e.g., 1, 2, or 3)

    Returns:
        Dictionary with chart metadata and corner statistics

    Raises:
        ValueError: If position data is unavailable for the session
    """
    # Build output path (no drivers for circuit-level chart)
    output_path = build_chart_path(
        workspace_dir,
        "track_map",
        year,
        gp,
        session_type,
        test_number=test_number,
   

def generate_tyre_strategy_chart(
    year: int,
    gp: str,
    session_type: str,
    workspace_dir: Path,
    test_number: int | None = None,
    session_number: int | None = None,
) -> dict:
    """Generate a tyre strategy visualization.

    Args:
        year: Season year
        gp: Grand Prix name (ignored for testing sessions)
        session_type: Session identifier (typically 'R' for race, ignored for testing)
        workspace_dir: Workspace directory for outputs and cache
        test_number: Testing event number (e.g., 1 or 2)
        session_number: Session within testing event (e.g., 1, 2, or 3)

    Returns:
        Dictionary with chart metadata and strategy info
    """
    # Build output path
    output_path = build_chart_path(
        workspace_dir,
        "tyre_strategy",
        year,
        gp,
        session_type,
        test_number=test_number,
        session_number=session_number,
    )

    # Load session with laps data
    session = load_session_or_te

def get_constructor_standings(
    year: int,
    round_number: int | None = None,
) -> dict:
    """Load constructor championship standings from FastF1 Ergast API.

    Points reflect the system used in that season. Sprint race points (where
    applicable) are included in the total. Historical seasons use their
    contemporary points systems.

    Args:
        year: Championship year (e.g., 2024)
        round_number: Optional specific round number (default: final standings)

    Returns:
        Dictionary with constructor standings and metadata

    Raises:
        Exception: If Ergast API request fails
    """
    # Initialize Ergast API
    ergast_api = get_ergast_client()

    # Fetch constructor standings
    # Use FINAL_ROUND for final standings if round_number not specified
    round_param = round_number if round_number is not None else FINAL_ROUND
    response = ergast_api.get_constructor_standings(season=year, round=round_param)

    # Parse and return response
    return

def get_driver_info(
    driver_code: str | None = None,
    season: int | None = None,
    limit: int = 100,
    offset: int = 0,
) -> dict:
    """Load driver information from FastF1 Ergast API.

    Args:
        driver_code: Optional 3-letter driver code (e.g., "VER", "HAM")
        season: Optional season year to filter drivers (e.g., 2024)
        limit: Maximum number of results to return
        offset: Number of results to skip for pagination

    Returns:
        Dictionary with driver information and metadata
    """
    # Initialize Ergast API
    ergast_api = get_ergast_client()

    # Call get_driver_info with appropriate filters
    if driver_code:
        # Ergast API expects driver ID, not driver code
        # Try driver code as-is first (in case it's actually an ID like "hamilton")
        driver_data = ergast_api.get_driver_info(driver=driver_code.lower())

        # If no results and it looks like a 3-letter code,
        # fetch recent season data and filter by dr

def get_driver_standings(
    year: int,
    round_number: int | None = None,
) -> dict:
    """Load driver championship standings from FastF1 Ergast API.

    Points reflect the system used in that season. Sprint race points (where
    applicable) are included in the total. Historical seasons use their
    contemporary points systems.

    Args:
        year: Championship year (e.g., 2024)
        round_number: Optional specific round number (default: final standings)

    Returns:
        Dictionary with driver standings and metadata

    Raises:
        Exception: If Ergast API request fails
    """
    # Initialize Ergast API
    ergast_api = get_ergast_client()

    # Fetch driver standings
    # Use FINAL_ROUND for final standings if round_number not specified
    round_param = round_number if round_number is not None else FINAL_ROUND
    response = ergast_api.get_driver_standings(season=year, round=round_param)

    # Parse and return response
    return parse_driver_standings_r

def get_event_schedule(
    year: int,
    round_number: int | None = None,
    country: str | None = None,
    include_testing: bool = True,
) -> dict:
    """Load event schedule from FastF1 and return as dict.

    Args:
        year: Championship year (e.g., 2024)
        round_number: Optional filter for specific round
        country: Optional filter for country name (case-insensitive)
        include_testing: Whether to include testing sessions (default: True)

    Returns:
        Dictionary with schedule data and event information
    """
    # Enable FastF1 cache
    setup_fastf1_cache()

    # Get the event schedule
    schedule = fastf1.get_event_schedule(year, include_testing=include_testing)

    # Convert DataFrame to list of dicts
    events = []
    for _, event in schedule.iterrows():
        # Apply filters if specified
        if round_number is not None and event["RoundNumber"] != round_number:
            continue
        if country is not None and event["Country"]

def _is_high_impact_message(row: pd.Series) -> bool:
    """Determine if a message is high-impact (race-changing event).

    High-impact messages include:
    - RED flags (race stoppages)
    - SAFETY CAR / VSC messages
    - CHEQUERED flag (race finish)
    - First GREEN light (race start)
    - Major collisions (messages containing "COLLISION")

    Args:
        row: Single row from race_control_messages DataFrame

    Returns:
        True if message is high-impact, False otherwise
    """
    category = row.get("Category")
    flag = row.get("Flag")
    message = str(row.get("Message", "")).upper()

    # RED flag - race stoppage
    if flag == FLAG_RED:
        return True

    # Safety Car category messages
    if category == MESSAGE_CATEGORY_SAFETY_CAR:
        return True

    # Chequered flag - race finish
    if flag == FLAG_CHEQUERED:
        return True

    # First green light - race start (check if it's the pit exit open before race)
    if flag == FLAG_GREEN and "PIT E

def _is_medium_impact_message(row: pd.Series) -> bool:
    """Determine if a message is medium-impact (situational awareness).

    Medium-impact messages include:
    - DOUBLE YELLOW flags
    - YELLOW flags (single)
    - DRS status changes
    - Significant penalties (time penalties, DSQ, etc.)
    - Retirements and DNFs

    Args:
        row: Single row from race_control_messages DataFrame

    Returns:
        True if message is medium-impact, False otherwise
    """
    category = row.get("Category")
    flag = row.get("Flag")
    message = str(row.get("Message", "")).upper()

    # DOUBLE YELLOW flags
    if flag == FLAG_DOUBLE_YELLOW:
        return True

    # Single YELLOW flags
    if flag == "YELLOW":
        return True

    # DRS status changes
    if category == MESSAGE_CATEGORY_DRS:
        return True

    # Penalties (but filter out "NO FURTHER INVESTIGATION")
    if "PENALTY" in message and "NO FURTHER" not in message:
        return True

    # Disqualifications
  

def _filter_by_detail_level(
    messages_df: pd.DataFrame,
    detail: str,
) -> pd.DataFrame:
    """Filter messages based on detail level (progressive disclosure).

    Args:
        messages_df: DataFrame of race control messages
        detail: Detail level ("high", "medium", or "full")

    Returns:
        Filtered DataFrame based on detail level

    Raises:
        ValueError: If detail level is not one of "high", "medium", or "full"
    """
    # Validate detail level
    valid_levels = ("high", "medium", "full")
    if detail not in valid_levels:
        raise ValueError(f"Invalid detail level: '{detail}'. Must be one of {valid_levels}")

    if detail == "full":
        return messages_df

    if detail == "high":
        # Only high-impact messages
        mask = messages_df.apply(_is_high_impact_message, axis=1)
        return messages_df[mask]

    # detail == "medium"
    # High-impact OR medium-impact messages
    mask = messages_df.apply(
        lambda row: _is_high_

def _filter_by_category(
    messages_df: pd.DataFrame,
    category: str | None,
) -> pd.DataFrame:
    """Filter messages by category.

    Args:
        messages_df: DataFrame of race control messages
        category: Category to filter by (Flag, Other, DRS, SafetyCar) or None

    Returns:
        Filtered DataFrame
    """
    if category is None:
        return messages_df

    return messages_df[messages_df["Category"] == category]

def _filter_by_flag_type(
    messages_df: pd.DataFrame,
    flag_type: str | None,
) -> pd.DataFrame:
    """Filter messages by flag type.

    Args:
        messages_df: DataFrame of race control messages
        flag_type: Flag type to filter by (RED, YELLOW, etc.) or None

    Returns:
        Filtered DataFrame
    """
    if flag_type is None:
        return messages_df

    return messages_df[messages_df["Flag"] == flag_type.upper()]

def _filter_by_driver(
    messages_df: pd.DataFrame,
    driver: str | None,
) -> pd.DataFrame:
    """Filter messages by driver racing number.

    Args:
        messages_df: DataFrame of race control messages
        driver: Racing number to filter by or None

    Returns:
        Filtered DataFrame
    """
    if driver is None:
        return messages_df

    return messages_df[messages_df["RacingNumber"] == driver]

def _filter_by_lap_range(
    messages_df: pd.DataFrame,
    lap_start: int | None,
    lap_end: int | None,
) -> pd.DataFrame:
    """Filter messages by lap range.

    Args:
        messages_df: DataFrame of race control messages
        lap_start: Start lap (inclusive) or None
        lap_end: End lap (inclusive) or None

    Returns:
        Filtered DataFrame
    """
    if lap_start is None and lap_end is None:
        return messages_df

    result = messages_df.copy()

    if lap_start is not None:
        result = result[result["Lap"] >= lap_start]

    if lap_end is not None:
        result = result[result["Lap"] <= lap_end]

    return result

def _filter_by_sector(
    messages_df: pd.DataFrame,
    sector: int | None,
) -> pd.DataFrame:
    """Filter messages by track sector.

    Args:
        messages_df: DataFrame of race control messages
        sector: Sector number to filter by or None

    Returns:
        Filtered DataFrame
    """
    if sector is None:
        return messages_df

    return messages_df[messages_df["Sector"] == sector]

def get_race_control_messages(
    year: int,
    gp: str | None = None,
    session_type: str | None = None,
    detail: str = "high",
    category: str | None = None,
    flag_type: str | None = None,
    driver: str | None = None,
    lap_start: int | None = None,
    lap_end: int | None = None,
    sector: int | None = None,
    test_number: int | None = None,
    session_number: int | None = None,
) -> RaceControlData:
    """Load race control messages from FastF1 with optional filtering.

    For regular GP sessions, provide gp and session_type.
    For testing sessions, provide test_number and session_number.

    Args:
        year: Season year (e.g., 2024)
        gp: Grand Prix name (e.g., "Monaco")
        session_type: Session identifier (R, Q, FP1, FP2, FP3, S, SQ)
        detail: Detail level for progressive disclosure ("high", "medium", "full")
        category: Filter by category (Flag, Other, DRS, SafetyCar) or None
        flag_type: Filter by flag type (RED, YELLOW, 

def _count_track_interruptions(session: Session) -> tuple[int, int, int]:
    """Count safety cars, VSCs, and red flags from track status data.

    Returns:
        Tuple of (safety_cars, virtual_safety_cars, red_flags)
    """
    try:
        track_status = session.track_status
        safety_cars = len(track_status[track_status["Status"] == TRACK_STATUS_SAFETY_CAR])
        vscs = len(track_status[track_status["Status"] == TRACK_STATUS_VSC_DEPLOYED])
        red_flags = len(track_status[track_status["Status"] == TRACK_STATUS_RED_FLAG])
        return safety_cars, vscs, red_flags
    except DataNotLoadedError:
        return 0, 0, 0

def _compute_wildness_score(
    race_summary: RaceSummaryStats,
    num_safety_cars: int,
    num_red_flags: int,
    race_distance_km: float,
    max_overtakes_per_km: float,
    max_volatility: float,
) -> float:
    """Compute a composite wildness score for a race.

    The score is a weighted combination of normalized overtake density,
    volatility, and safety car/red flag bonuses.  Overtakes are expressed
    as a per-km rate so that shortened races and sprints are not penalised
    for covering fewer laps.

    Args:
        race_summary: Aggregate race statistics
        num_safety_cars: Number of safety car deployments
        num_red_flags: Number of red flags
        race_distance_km: Completed race distance (total_laps * circuit_length_km,
            or total_laps * AVG_CIRCUIT_LENGTH_KM when circuit length is unavailable)
        max_overtakes_per_km: Maximum overtakes-per-km across comparable races
        max_volatility: Maximum average volatility across comparable ra

def get_season_summary(year: int) -> SeasonSummary:
    """Load all races in a season and rank them by wildness.

    This loads each race session individually, which can be slow on first run.
    Subsequent calls benefit from FastF1's cache.

    Args:
        year: Championship year (e.g., 2024)

    Returns:
        Dictionary with races sorted by wildness score (descending)
        and season-wide averages.
    """
    setup_fastf1_cache()
    schedule = fastf1.get_event_schedule(year, include_testing=False)

    # Collect raw data for all sessions first (need max values for normalization)
    raw_races = []

    for _, event in schedule.iterrows():
        round_number = int(event["RoundNumber"])
        if round_number == 0:
            continue

        event_name = event["EventName"]
        country = event["Country"]
        event_date = event["EventDate"]
        date_str = event_date.isoformat()[:10] if pd.notna(event_date) else None
        event_format = event.get("EventFo

def _extract_track_status(session: Session) -> RaceConditions | None:
    """Extract race condition counts from track status data.

    Args:
        session: FastF1 session object

    Returns:
        Dict with num_safety_cars, num_virtual_safety_cars, num_red_flags
        or None if track status data is unavailable
    """
    try:
        track_status = session.track_status

        # Count occurrences of each status code
        num_safety_cars = len(track_status[track_status["Status"] == TRACK_STATUS_SAFETY_CAR])
        num_red_flags = len(track_status[track_status["Status"] == TRACK_STATUS_RED_FLAG])
        # Count VSC as deployments, not endings
        num_virtual_safety_cars = len(track_status[track_status["Status"] == TRACK_STATUS_VSC_DEPLOYED])

        return {
            "num_safety_cars": num_safety_cars,
            "num_virtual_safety_cars": num_virtual_safety_cars,
            "num_red_flags": num_red_flags,
        }
    except DataNotLoadedError:
        return 

def _extract_weather_data(session: Session) -> WeatherData | None:
    """Extract weather statistics from session weather data.

    Args:
        session: FastF1 session object

    Returns:
        Dict with min/max/avg for air_temp, humidity, pressure, wind_speed
        or None if weather data is unavailable
    """
    try:
        weather: pd.DataFrame = session.weather_data

        # Check if DataFrame is empty
        if weather.empty:
            return None

        # Calculate statistics for each metric
        def get_stats(column_name: str):
            """Get min, max, avg for a column, handling missing data."""
            if column_name not in weather.columns:
                return {"min": None, "max": None, "avg": None}

            col = weather[column_name]
            # Filter out NaN values
            col_clean = col.dropna()

            if col_clean.empty:
                return {"min": None, "max": None, "avg": None}

            return {
                "min

def get_session_info(
    year: int,
    gp: str | None = None,
    session_type: str | None = None,
    test_number: int | None = None,
    session_number: int | None = None,
) -> SessionInfo:
    """Load session info from FastF1 and return as dict.

    For regular GP sessions, provide gp and session_type.
    For testing sessions, provide test_number and session_number.

    Args:
        year: Season year (e.g., 2024)
        gp: Grand Prix name (e.g., "Monaco", "Silverstone")
        session_type: Session identifier (R, Q, FP1, FP2, FP3, S, SQ)
        test_number: Testing event number (e.g., 1 or 2)
        session_number: Session within testing event (e.g., 1, 2, or 3)

    Returns:
        Dictionary with session metadata, driver info, race conditions, and weather statistics.
        Race conditions include counts of safety cars, virtual safety cars, and red flags.
        Weather includes min/max/avg for air temperature, humidity, pressure, and wind speed.
    """
    # Load s

def get_workspace_base() -> Path:
    """Get the base directory for all workspaces.

    Returns:
        Path to ~/.pitlane/workspaces/
    """
    return Path.home() / ".pitlane" / "workspaces"

def get_cache_dir() -> Path:
    """Get the shared cache directory for FastF1 data.

    Returns:
        Path to ~/.pitlane/cache/fastf1/
    """
    return Path.home() / ".pitlane" / "cache" / "fastf1"

def generate_workspace_id() -> str:
    """Generate a unique workspace ID.

    Returns:
        UUID string to use as workspace identifier.
    """
    return str(uuid.uuid4())

def get_workspace_path(workspace_id: str) -> Path:
    """Get the workspace path for a given workspace ID.

    Args:
        workspace_id: The workspace identifier.

    Returns:
        Path to the workspace directory.
    """
    return get_workspace_base() / workspace_id

def workspace_exists(workspace_id: str) -> bool:
    """Check if a workspace exists for the given workspace ID.

    Args:
        workspace_id: The workspace identifier.

    Returns:
        True if workspace exists, False otherwise.
    """
    workspace_path = get_workspace_path(workspace_id)
    return workspace_path.exists() and workspace_path.is_dir()

def create_workspace(workspace_id: str | None = None, description: str | None = None, max_retries: int = 3) -> dict:
    """Create a new workspace directory structure.

    Implements collision retry logic for auto-generated workspace IDs.

    Args:
        workspace_id: Workspace identifier. Auto-generated if None.
        description: Optional description for the workspace.
        max_retries: Maximum retry attempts for UUID collision (default: 3).

    Returns:
        Dictionary with workspace information:
        {
            "workspace_id": str,
            "workspace_path": str,
            "created_at": str,
        }

    Raises:
        ValueError: If workspace already exists for the given workspace_id.
        RuntimeError: If failed to generate unique workspace ID after max_retries.
    """
    if workspace_id is not None:
        # Explicit workspace_id provided - no retry logic
        if workspace_exists(workspace_id):
            raise ValueError(f"Workspace already 

def _create_workspace_internal(workspace_id: str, description: str | None) -> dict:
    """Internal workspace creation logic.

    Args:
        workspace_id: The workspace identifier.
        description: Optional description for the workspace.

    Returns:
        Dictionary with workspace information.
    """
    workspace_path = get_workspace_path(workspace_id)

    # Create directory structure
    workspace_path.mkdir(parents=True, exist_ok=False)
    (workspace_path / "data").mkdir(exist_ok=True)
    (workspace_path / "charts").mkdir(exist_ok=True)

    # Ensure shared cache exists
    cache_dir = get_cache_dir()
    cache_dir.mkdir(parents=True, exist_ok=True)

    now = datetime.now(UTC)

    # Write metadata
    metadata = {
        "workspace_id": workspace_id,
        "created_at": now.isoformat() + "Z",
        "last_accessed": now.isoformat() + "Z",
    }

    if description:
        metadata["description"] = description

    metadata_path = workspace_path / ".metadata.js

def update_workspace_metadata(workspace_id: str) -> None:
    """Update the last_accessed timestamp in workspace metadata.

    Args:
        workspace_id: The workspace identifier.

    Raises:
        ValueError: If workspace doesn't exist.
    """
    if not workspace_exists(workspace_id):
        raise ValueError(f"Workspace does not exist for workspace ID: {workspace_id}")

    workspace_path = get_workspace_path(workspace_id)
    metadata_path = workspace_path / ".metadata.json"

    # Read existing metadata or create new
    if not metadata_path.exists():
        # Metadata missing, recreate it
        now = datetime.now(UTC)
        metadata = {
            "workspace_id": workspace_id,
            "created_at": now.isoformat() + "Z",
            "last_accessed": now.isoformat() + "Z",
        }
    else:
        with open(metadata_path) as f:
            metadata = json.load(f)

        metadata["last_accessed"] = datetime.now(UTC).isoformat() + "Z"

    # Atomic write using t

def get_workspace_info(workspace_id: str) -> dict:
    """Get information about a workspace.

    Args:
        workspace_id: The workspace identifier.

    Returns:
        Dictionary with workspace metadata including:
        {
            "workspace_id": str,
            "workspace_path": str,
            "created_at": str,
            "last_accessed": str,
            "description": str (optional),
            "data_files": list[str],
            "chart_files": list[str],
        }

    Raises:
        ValueError: If workspace doesn't exist.
    """
    if not workspace_exists(workspace_id):
        raise ValueError(f"Workspace does not exist for workspace ID: {workspace_id}")

    workspace_path = get_workspace_path(workspace_id)
    metadata_path = workspace_path / ".metadata.json"

    if metadata_path.exists():
        with open(metadata_path) as f:
            metadata = json.load(f)
    else:
        metadata = {
            "workspace_id": workspace_id,
            "created_

def list_workspaces(show_all: bool = False) -> list[dict]:
    """List all workspaces.

    Args:
        show_all: If True, include all workspaces. If False, only recent ones.

    Returns:
        List of workspace information dictionaries sorted by last_accessed (newest first).
    """
    workspace_base = get_workspace_base()

    if not workspace_base.exists():
        return []

    workspaces = []

    for workspace_dir in workspace_base.iterdir():
        if not workspace_dir.is_dir():
            continue

        workspace_id = workspace_dir.name
        try:
            info = get_workspace_info(workspace_id)
            workspaces.append(info)
        except Exception:
            # Skip corrupted workspaces
            continue

    # Sort by last_accessed (newest first)
    def get_last_accessed(ws):
        try:
            return datetime.fromisoformat(ws["last_accessed"].rstrip("Z"))
        except Exception:
            return datetime.min

    workspaces.sort(key=get

def remove_workspace(workspace_id: str) -> None:
    """Remove a workspace and all its contents.

    Args:
        workspace_id: The workspace identifier.

    Raises:
        ValueError: If workspace doesn't exist.
    """
    if not workspace_exists(workspace_id):
        raise ValueError(f"Workspace does not exist for workspace ID: {workspace_id}")

    workspace_path = get_workspace_path(workspace_id)
    shutil.rmtree(workspace_path)

def clean_workspaces(older_than_days: int | None = None, remove_all: bool = False) -> dict:
    """Clean up old workspaces.

    Args:
        older_than_days: Remove workspaces not accessed in this many days.
        remove_all: If True, remove all workspaces (overrides older_than_days).

    Returns:
        Dictionary with cleanup statistics:
        {
            "removed_count": int,
            "removed_workspaces": list[str],
        }
    """
    workspace_base = get_workspace_base()

    if not workspace_base.exists():
        return {"removed_count": 0, "removed_workspaces": []}

    removed_workspaces = []

    for workspace_dir in workspace_base.iterdir():
        if not workspace_dir.is_dir():
            continue

        workspace_id = workspace_dir.name

        # Check if should remove
        should_remove = remove_all

        if not should_remove and older_than_days is not None:
            try:
                info = get_workspace_info(workspace_id)
               

def get_conversations_path(workspace_id: str) -> Path:
    """Get path to conversations.json for a workspace.

    Args:
        workspace_id: The workspace identifier.

    Returns:
        Path to the conversations.json file.
    """
    return get_workspace_path(workspace_id) / "conversations.json"

def load_conversations(workspace_id: str) -> dict:
    """Load conversation metadata for a workspace.

    Args:
        workspace_id: The workspace identifier.

    Returns:
        Dictionary with version, active_conversation_id, and conversations list.
        Returns empty structure if file doesn't exist.
    """
    conversations_path = get_conversations_path(workspace_id)

    if not conversations_path.exists():
        return {
            "version": 1,
            "active_conversation_id": None,
            "conversations": [],
        }

    try:
        with open(conversations_path) as f:
            return json.load(f)
    except (json.JSONDecodeError, OSError):
        # Return empty structure on corruption
        return {
            "version": 1,
            "active_conversation_id": None,
            "conversations": [],
        }

def save_conversations(workspace_id: str, data: dict) -> None:
    """Save conversation metadata atomically.

    Args:
        workspace_id: The workspace identifier.
        data: The conversation data dictionary to save.

    Raises:
        ValueError: If workspace doesn't exist.
    """
    if not workspace_exists(workspace_id):
        raise ValueError(f"Workspace does not exist for workspace ID: {workspace_id}")

    conversations_path = get_conversations_path(workspace_id)
    workspace_path = get_workspace_path(workspace_id)

    # Atomic write using tempfile + rename
    fd, temp_path = tempfile.mkstemp(
        dir=workspace_path,
        prefix=".conversations.tmp.",
        suffix=".json",
    )

    try:
        with os.fdopen(fd, "w") as f:
            json.dump(data, f, indent=2)
        os.replace(temp_path, conversations_path)
    except Exception:
        with suppress(Exception):
            os.unlink(temp_path)
        raise