renaming of the fastf1 tools

app.py

@@ -1,7 +1,7 @@
 import gradio as gr
 
 # Local modules
-import tools
+import fastf1_tools
 from utils.constants import (
     DRIVER_NAMES,
     CONSTRUCTOR_NAMES,
@@ -12,7 +12,7 @@ from utils.constants import (
 )
 
 iface_driver_championship_standings = gr.Interface(
-    fn=tools.driver_championship_standings,
+    fn=fastf1_tools.driver_championship_standings,
     inputs=[
         gr.Number(label="Calendar year", value=CURRENT_YEAR, minimum=1950, maximum=CURRENT_YEAR),
         gr.Dropdown(label="Driver", choices=DRIVER_NAMES)
@@ -23,7 +23,7 @@ iface_driver_championship_standings = gr.Interface(
 )
 
 iface_constructor_championship_standings = gr.Interface(
-    fn=tools.constructor_championship_standings,
+    fn=fastf1_tools.constructor_championship_standings,
     inputs=[
         gr.Number(label="Calendar year", value=CURRENT_YEAR, minimum=1950, maximum=CURRENT_YEAR),
         gr.Dropdown(label="Constructor", choices=CONSTRUCTOR_NAMES)
@@ -34,7 +34,7 @@ iface_constructor_championship_standings = gr.Interface(
 )
 
 iface_event_info = gr.Interface(
-    fn=tools.get_event_info,
+    fn=fastf1_tools.get_event_info,
     inputs=[
         gr.Number(label="Calendar year", value=CURRENT_YEAR, minimum=1950, maximum=CURRENT_YEAR),
         gr.Textbox(label="Grand Prix", placeholder="Ex: Monaco", info="The name of the GP/country/location (Fuzzy matching supported)"),
@@ -46,7 +46,7 @@ iface_event_info = gr.Interface(
 )
 
 iface_season_calendar = gr.Interface(
-    fn=tools.get_season_calendar,
+    fn=fastf1_tools.get_season_calendar,
     inputs=[
         gr.Number(label="Calendar year", value=CURRENT_YEAR, minimum=1950, maximum=CURRENT_YEAR),
     ],
@@ -56,7 +56,7 @@ iface_season_calendar = gr.Interface(
 )
 
 iface_track_visualization = gr.Interface(
-    fn=tools.track_visualization,
+    fn=fastf1_tools.track_visualization,
     inputs=[
         gr.Number(label="Calendar year", value=CURRENT_YEAR, minimum=1950, maximum=CURRENT_YEAR),
         gr.Textbox(label="Grand Prix", placeholder="Ex: Monaco", info="The name of the GP/country/location (Fuzzy matching supported)"),
@@ -69,7 +69,7 @@ iface_track_visualization = gr.Interface(
 )
 
 iface_session_results = gr.Interface(
-    fn=tools.get_session_results,
+    fn=fastf1_tools.get_session_results,
     inputs=[
         gr.Number(label="Calendar year", value=CURRENT_YEAR, minimum=1950, maximum=CURRENT_YEAR),
         gr.Textbox(label="Grand Prix", placeholder="Ex: Monaco", info="The name of the GP/country/location (Fuzzy matching supported)"),
@@ -81,7 +81,7 @@ iface_session_results = gr.Interface(
 )
 
 iface_driver_info = gr.Interface(
-    fn=tools.get_driver_info,
+    fn=fastf1_tools.get_driver_info,
     inputs=[
         gr.Dropdown(label="Driver", choices=DRIVER_NAMES)
     ],
@@ -127,6 +127,7 @@ named_interfaces = {
     "Track Visualizations": iface_track_visualization,
     "Session Results": iface_session_results,
     "Driver Info": iface_driver_info,
+    "Test": iface_test
 }
 
 # Tab names and interfaces

fastf1_tools.py

@@ -0,0 +1,239 @@
+"""
+This module provides tools for interacting with Formula 1 data using the FastF1 library.
+
+Tools to implement
+- driver info
+- compare drivers
+"""
+
+import json
+import fastf1
+import gradio as gr
+import pandas as pd
+from PIL import Image
+from typing import Union
+from fastf1.core import Session
+
+# Local modules
+from utils import parser_utils, track_utils
+from utils.constants import (
+    AVAILABLE_SESSION_TYPES,
+    DRIVER_DETAILS
+)
+
+# Custom types
+gp = Union[str, int]
+session_type = Union[str, int, None]
+
+### FastF1 tools ###
+
+def get_session(year: int, round: gp, session_type: session_type) -> Session:
+    """Retrieve a specific Formula 1 session.
+    
+    Args:
+        year (int): The season year (e.g., 2024)
+        round (str | int): The race round number or name (e.g., 1 or 'Monaco')
+        session_type (str | int | None): Type of session (e.g., 'FP1', 'Q', 'R')
+        
+    Returns:
+        Session: A FastF1 Session object for the specified parameters
+        
+    Note:
+        If session_type is a string and not in AVAILABLE_SESSION_TYPES,
+        returns an error message string instead.
+    """
+
+    # Check if session type is valid
+    if isinstance(session_type, str):
+        if session_type.lower() not in AVAILABLE_SESSION_TYPES:
+            return f"Session type {session_type} is not available. Supported session types: {AVAILABLE_SESSION_TYPES}"
+
+    return fastf1.get_session(year, round, session_type)
+
+def get_season_calendar(year: int) -> str:
+    """Get the complete race calendar for a specific F1 season.
+    
+    Args:
+        year (int): The season year to get the calendar for
+        
+    Returns:
+        str: Formatted string containing the season calendar
+    """
+    season_calendar = fastf1.get_event_schedule(year)
+    return parser_utils.parse_season_calendar(season_calendar)
+
+def get_event_info(year: int, round: gp, format: str) -> str:
+    """Retrieve information about a specific Formula 1 event.
+    
+    Args:
+        year (int): The season year
+        round (str | int): The race round number or name
+        format (str): Output format ('human' for readable text, 'LLM' for structured data)
+        
+    Returns:
+        str: Formatted event information based on the specified format
+    """
+
+    event = fastf1.get_session(year, round, "race").event # Event object is the same for all sessions, so hardcode "race"
+    if format == "human":
+        data_interval = f"{event['Session1DateUtc'].date()} - {event['Session5DateUtc'].date()}"
+        event_string = f"Round {event['RoundNumber']} : {event['EventName']} - {event['Location']}, {event['Country']} ({data_interval})"
+        return event_string
+    elif format == "LLM":
+        return parser_utils.parse_event_info(event)
+
+
+def get_constructor_standings(year: int) -> str:
+    """Retrieve constructor championship standings for a given year.
+    
+    Args:
+        year (int): The season year
+        
+    Returns:
+        str: Constructor championship standings
+    """
+    pass
+
+def get_driver_standings(year: int) -> str:
+    """Retrieve driver championship standings for a given year.
+    
+    Args:
+        year (int): The season year
+        
+    Returns:
+        str: Driver championship standings
+    """
+    pass
+
+def driver_championship_standings(year: int, driver_name: str) -> str:
+    """Get the championship standing for a specific driver in a given year.
+    
+    Args:
+        year (int): The season year
+        driver_name (str): Full name of the driver (e.g., 'Lewis Hamilton')
+        
+    Returns:
+        str: Formatted string with driver's position, points, and wins
+    """
+
+    with open("assets/driver_abbreviations.json") as f:
+        driver_abbreviations = json.load(f)
+        driver_abbreviation = driver_abbreviations[driver_name]
+    ergast = fastf1.ergast.Ergast()
+    driver_standings = ergast.get_driver_standings(year).content[0]
+    driver_standing = driver_standings[["position", "points", "wins", "driverCode"]].reset_index(drop=True)
+    driver_standing = driver_standing[driver_standing["driverCode"] == driver_abbreviation]
+    suffix = "st" if driver_standing['position'].iloc[0] == 1 else "nd" if driver_standing['position'].iloc[0] == 2 else "rd" if driver_standing['position'].iloc[0] == 3 else "th"
+    standings_string = f"{driver_name} is {driver_standing['position'].iloc[0]}{suffix} with {driver_standing['points'].iloc[0]} points and {driver_standing['wins'].iloc[0]} wins"
+    return standings_string
+    
+def constructor_championship_standings(year: int, constructor_name: str) -> str:
+    """Get the championship standing for a specific constructor in a given year.
+    
+    Args:
+        year (int): The season year
+        constructor_name (str): Name of the constructor team (e.g., 'Mercedes')
+        
+    Returns:
+        str: Formatted string with constructor's position, points, and wins
+    """
+
+    team_mapping = {
+    "McLaren": "McLaren",
+    "Ferrari": "Ferrari",
+    "Red Bull Racing": "Red Bull",
+    "Mercedes": "Mercedes",
+    "Aston Martin": "Aston Martin",
+    "Alpine": "Alpine F1 Team",
+    "Haas": "Haas F1 Team",
+    "Racing Bulls": "RB F1 Team",
+    "Williams": "Williams",
+    "Kick Sauber": "Sauber"
+    }
+
+    ergast = fastf1.ergast.Ergast()
+    constructor_standings = ergast.get_constructor_standings(year).content[0]
+    constructor_standing = constructor_standings[["position", "points", "wins", "constructorName"]].reset_index(drop=True)
+    mapped_name = team_mapping[constructor_name]
+    constructor_standing = constructor_standing[constructor_standing["constructorName"] == mapped_name]
+    suffix = "st" if constructor_standing['position'].iloc[0] == 1 else "nd" if constructor_standing['position'].iloc[0] == 2 else "rd" if constructor_standing['position'].iloc[0] == 3 else "th"
+    standings_string = f"{constructor_name} are {constructor_standing['position'].iloc[0]}{suffix} with {constructor_standing['points'].iloc[0]} points and {constructor_standing['wins'].iloc[0]} wins"
+    return standings_string
+
+def track_visualization(year: int, round: gp, visualization_type: str, driver_name: str) -> Image.Image:
+    """Generate a visualization of the track with specified data.
+    
+    Args:
+        year (int): The season year
+        round (str | int): The race round number or name
+        visualization_type (str): Type of visualization ('speed', 'corners', or 'gear')
+        driver_name (str): Name of the driver for driver-specific visualizations
+        
+    Returns:
+        Image.Image: A PIL Image object containing the visualization
+    """
+
+    session = get_session(year, round, "race")
+    session.load()
+
+    if visualization_type == "speed":
+        return track_utils.create_track_speed_visualization(session, driver_name)
+    elif visualization_type == "corners":
+        return track_utils.create_track_corners_visualization(session)
+    elif visualization_type == "gear":
+        return track_utils.create_track_gear_visualization(session)
+
+def get_session_results(year: int, round: gp, session_type: session_type) -> pd.DataFrame:
+    """Retrieve and format the results of a specific session.
+    
+    Args:
+        year (int): The season year
+        round (str | int): The race round number or name
+        session_type (str | int | None): Type of session (e.g., 'Q', 'R', 'Sprint')
+        
+    Returns:
+        pd.DataFrame: DataFrame containing the session results
+        
+    Raises:
+        gr.Error: If the session type is not supported for the specified round
+    """
+
+    try:
+        session = get_session(year, round, session_type)
+        session.load(telemetry=False)
+        results = session.results
+    except ValueError as e:
+        raise gr.Error(f"Session type {session_type} is not supported for the specified round. This Grand Prix most likely did not include a sprint race/quali.")
+
+    df = results[["DriverNumber", "Abbreviation", "FullName", "Position", "GridPosition", "Points", "Status", "Q1", "Q2", "Q3"]]
+    df["Name"] = df.apply(lambda row: f"{row['FullName']} ({row['Abbreviation']} • {row['DriverNumber']})", axis=1)
+    df = df.drop(columns=["FullName", "Abbreviation", "DriverNumber"])
+    df = df.rename(columns={"Position": "Pos", "GridPosition": "Grid Pos"})
+
+    # Process results based on session type
+    if session_type in ["race", "sprint"]:
+        df = df[["Pos", "Name", "Points", "Grid Pos", "Status"]]
+    elif "qualifying" in session_type:
+        df[["Q1", "Q2", "Q3"]] = df[["Q1", "Q2", "Q3"]].apply(lambda x: x.dt.total_seconds().apply(lambda y: f"{int(y//60):02d}:{int(y%60):02d}.{int(y%1*1000):03d}" if pd.notna(y) else "-"))
+        df = df[["Pos", "Name", "Q1", "Q2", "Q3"]]
+    return df
+
+def get_driver_info(driver_name: str) -> str:
+    """Retrieve detailed information about a specific driver.
+    
+    Args:
+        driver_name (str): Full name of the driver (e.g., 'Max Verstappen')
+        
+    Returns:
+        str: Formatted string with driver's details including name, team, number,
+             nationality, and a brief summary
+    """
+    driver = DRIVER_DETAILS[driver_name]
+    driver_info_string = f"{driver_name} ({driver['birth_date']})\n{driver['team']} #{driver['number']}\n{driver['nationality']}\n\n{driver['summary']}"
+    return driver_info_string
+    
+
+
+if __name__ == "__main__":
+    session = get_session(2024, 1, "fp1")
+    session.load()

mcp_client.py

@@ -1,12 +1,49 @@
 import os
 import gradio as gr
-from mcp import StdioServerParameters
-from smolagents import InferenceClientModel, CodeAgent, ToolCollection, MCPClient
+from typing import List, Dict
+from smolagents import InferenceClientModel, LiteLLMModel, ToolCallingAgent, MCPClient
+
+SYSTEM_PROMPT = """You are a helpful Formula 1 assistant and strategist. You have access to various F1 data and tools to help answer questions about races, drivers, teams, and more. 
+Be concise and accurate in your responses. If you don't know something, use the available tools to find the information.
+In addition, you will be asked to act as a live race engineer strategist during a Formula 1 race, making crucial calls during the event."""
+
+def format_messages(history: List[List[str]], message: str) -> List[Dict[str, str]]:
+    """Format the conversation history and new message for the agent."""
+    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
+    
+    # Add conversation history
+    for user_msg, bot_msg in history:
+        messages.append({"role": "user", "content": user_msg})
+        messages.append({"role": "assistant", "content": bot_msg})
+    
+    # Add the new message
+    messages.append({"role": "user", "content": message})
+    return messages
+
+def agent_chat(message: str, history: List[List[str]]) -> str:
+    """Handle chat messages with conversation history."""
+
+    # Format messages with system prompt and history
+    formatted_messages = format_messages(history, message)
+    
+    # Convert messages to a single string with role indicators
+    chat_history = "\n".join(
+        f"{msg['role'].capitalize()}: {msg['content']}" 
+        for msg in formatted_messages[1:]  # Skip system prompt in the chat history
+    )
+    
+    # Include system prompt at the beginning
+    full_prompt = f"{SYSTEM_PROMPT}\n\n{chat_history}"
+    
+    # Get agent response
+    response = str(agent.run(full_prompt))
+    return response
 
 
 if __name__ == "__main__":
 
-    only_list_tools = True
+    only_list_tools = False # Set to True to only list tools (used for debugging)
+    local_model = True # If you have Ollama installed, set this to True
     
     try:
 
@@ -15,30 +52,42 @@ if __name__ == "__main__":
         tools = mcp_client.get_tools()
 
         print("### MCP tools ### ")
-        print("\n".join(f"{t.name}: {t.description}" for t in tools))
+        print("\n".join(f"{i}: {t.name}: {t.description}" for i,t in enumerate(tools)))
 
         if only_list_tools:
-            mcp_client.close()
+            mcp_client.disconnect()
             exit(0)
 
+
         # Define model
-        model = InferenceClientModel(
-            model_id="Qwen/Qwen2.5-32B-Instruct",
-            token=os.getenv("HF_TOKEN")
-        )
+        if local_model:
+            model = LiteLLMModel(
+                model_id="ollama_chat/qwen3:8b",
+                api_base="http://127.0.0.1:11434", # Default ollama server
+                num_ctx=32768,
+            )
+        else:
+            model = InferenceClientModel(
+                model_id="Qwen/Qwen2.5-32B-Instruct",
+                provider="",
+                token=os.getenv("HF_TOKEN")
+            )
 
-        agent = CodeAgent(tools=[*tools], model=model)
+        agent = ToolCallingAgent(tools=[*tools], model=model)
 
 
         chat_interface = gr.ChatInterface(
-            fn=lambda message, history: str(agent.run(message)),
+            fn=agent_chat,
             type="messages",
-            examples=["Prime factorization of 68"],
-            title="Agent with MCP Tools",
-            description="This is a simple agent that uses MCP tools to answer questions."
+            examples=[
+                "What are the driver standings for the 2024 Formula 1 season?",
+                "What is the calendar for the 2024 Formula 1 season?"
+            ],
+            title="🏎️ Formula 1 Assistant",
+            description="This is a simple agent that uses MCP tools to answer questions about Formula 1."
         )
 
         chat_interface.launch()
 
     finally:
-        mcp_client.close()
+        mcp_client.disconnect()