arre99 commited on
Commit
7206ec7
·
1 Parent(s): 35aeee0

implemented architecture for LLM to dynamically create approprite API requests to OpenF1 server

Browse files
openf1_api_playground.ipynb CHANGED
The diff for this file is too large to render. See raw diff
 
openf1_registry.py ADDED
@@ -0,0 +1,208 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from enum import Enum
2
+ from dataclasses import dataclass
3
+ from typing import Dict, Set, List, Optional
4
+
5
+ class FilterType(Enum):
6
+ EQUALITY = "equality" # exact match
7
+ COMPARISON = "comparison" # >, <, >=, <=
8
+
9
+ class DataType(Enum):
10
+ STRING = "string"
11
+ INTEGER = "integer"
12
+ DATETIME = "datetime"
13
+ BINARY = "binary" # true/false filters
14
+
15
+ @dataclass
16
+ class FilterSpec:
17
+ """Specification for an API filter parameter"""
18
+ name: str
19
+ filter_type: FilterType
20
+ data_type: DataType
21
+ description: str = ""
22
+ allowed_values: Optional[List[str]] = None # For equality filters with restricted values
23
+
24
+ def get_query_examples(self) -> List[str]:
25
+ """Generate example query parameters for this filter"""
26
+ examples = []
27
+
28
+ if self.data_type == DataType.BINARY:
29
+ examples = [f"{self.name}=true", f"{self.name}=false"]
30
+
31
+ elif self.filter_type == FilterType.EQUALITY:
32
+ if self.allowed_values:
33
+ examples = [f"{self.name}={val}" for val in self.allowed_values[:2]]
34
+ elif self.data_type == DataType.STRING:
35
+ examples = [f"{self.name}=example_value"]
36
+ elif self.data_type == DataType.INTEGER:
37
+ examples = [f"{self.name}=42"]
38
+ elif self.data_type == DataType.DATETIME:
39
+ examples = [f"{self.name}=2024-01-01T00:00:00Z", f"{self.name}=2024-01-01T10:30:00Z"]
40
+
41
+ elif self.filter_type == FilterType.COMPARISON:
42
+ if self.data_type == DataType.INTEGER:
43
+ examples = [f"{self.name}>=10", f"{self.name}<100"]
44
+ elif self.data_type == DataType.DATETIME:
45
+ examples = [f"{self.name}>=2024-01-01T00:00:00Z", f"{self.name}<2024-12-31T00:00:00Z"]
46
+ elif self.data_type == DataType.STRING:
47
+ examples = [f"{self.name}>M", f"{self.name}<Z"] # alphabetical comparison
48
+
49
+ return examples
50
+
51
+ def help_text(self) -> str:
52
+ """Generate help text for this filter"""
53
+ text = f"Filter: {self.name}\n"
54
+ text += f" Type: {self.filter_type.value} ({self.data_type.value})\n"
55
+
56
+ if self.description:
57
+ text += f" Description: {self.description}\n"
58
+
59
+ if self.allowed_values:
60
+ text += f" Allowed values: {', '.join(self.allowed_values)}\n"
61
+
62
+ examples = self.get_query_examples()
63
+ if examples:
64
+ text += f" Examples: {', '.join(examples)}"
65
+
66
+ return text
67
+
68
+
69
+
70
+ class APIEndpointRegistry:
71
+ """Registry for API endpoints and their supported filters"""
72
+
73
+ def __init__(self, base_url: str = ""):
74
+ self.base_url = base_url
75
+ self.endpoints: Dict[str, Set[str]] = {} # endpoint -> filter names
76
+ self.filters: Dict[str, FilterSpec] = {} # global filter definitions
77
+
78
+ def define_filter(
79
+ self,
80
+ name: str,
81
+ filter_type: FilterType,
82
+ data_type: DataType,
83
+ description: str = "",
84
+ allowed_values: Optional[List[str]] = None
85
+ ) -> 'APIEndpointRegistry':
86
+ """Define a filter that can be used by endpoints"""
87
+ filter_spec = FilterSpec(name, filter_type, data_type, description, allowed_values)
88
+ self.filters[name] = filter_spec
89
+ return self
90
+
91
+ def register_endpoint(self, endpoint: str, *filter_names: str) -> 'APIEndpointRegistry':
92
+ """Register an API endpoint with its supported filters"""
93
+
94
+ # Validate all filters exist
95
+ for filter_name in filter_names:
96
+ if filter_name not in self.filters:
97
+ raise ValueError(f"Filter '{filter_name}' not defined. Use define_filter() first.")
98
+
99
+ # Store endpoint -> filters mapping
100
+ self.endpoints[endpoint] = set(filter_names)
101
+
102
+ return self
103
+
104
+
105
+
106
+
107
+
108
+ def get_endpoint_filters(self, endpoint: str) -> Dict[str, FilterSpec]:
109
+ """Get all filters supported by an endpoint"""
110
+ filter_names = self.endpoints.get(endpoint, set())
111
+ return {name: self.filters[name] for name in filter_names}
112
+
113
+ def get_filter_help(self, filter_name: str) -> str:
114
+ """Get help text for a specific filter"""
115
+
116
+ if filter_name in self.filters:
117
+ return self.filters[filter_name].help_text()
118
+ return f"Filter '{filter_name}' not found."
119
+
120
+ def get_endpoint_help(self, endpoint: str) -> str:
121
+ """Get help text for all filters supported by an endpoint"""
122
+
123
+ filters = self.get_endpoint_filters(endpoint)
124
+ if not filters:
125
+ return f"Endpoint '{endpoint}' has no registered filters."
126
+
127
+ help_text = f"API Endpoint: {self.base_url}{endpoint}\n"
128
+ help_text += f"Supported filters ({len(filters)}):\n\n"
129
+
130
+ for name, spec in sorted(filters.items()):
131
+ help_text += spec.help_text() + "\n\n"
132
+
133
+ return help_text.strip()
134
+
135
+ def list_all_endpoints(self) -> List[str]:
136
+ """Get list of all registered endpoints"""
137
+ return sorted(self.endpoints.keys())
138
+
139
+ def list_all_filters(self) -> List[str]:
140
+ """Get list of all defined filters"""
141
+ return sorted(self.filters.keys())
142
+
143
+
144
+ # Create registry with base URL
145
+ f1_api = APIEndpointRegistry("https://api.openf1.org/v1/")
146
+
147
+ # Define filters with their specifications
148
+ f1_api.define_filter("date", FilterType.COMPARISON, DataType.DATETIME, "The UTC date and time, in ISO 8601 format.")
149
+ f1_api.define_filter("driver_number", FilterType.EQUALITY, DataType.INTEGER, "The unique number assigned to an F1 driver")
150
+ f1_api.define_filter("meeting_key", FilterType.EQUALITY, DataType.STRING, "The unique identifier for the meeting. Use 'latest' to identify the latest or current meeting.")
151
+ f1_api.define_filter("session_key", FilterType.EQUALITY, DataType.STRING, "The unique identifier for the session. Use 'latest' to identify the latest or current session.")
152
+ f1_api.define_filter("speed", FilterType.COMPARISON, DataType.INTEGER, "Velocity of the car in km/h.")
153
+ f1_api.define_filter("country_code", FilterType.EQUALITY, DataType.STRING, "A code that uniquely identifies the country.")
154
+ f1_api.define_filter("first_name", FilterType.EQUALITY, DataType.STRING, "The first name of the driver.")
155
+ f1_api.define_filter("last_name", FilterType.EQUALITY, DataType.STRING, "The last name of the driver.")
156
+ f1_api.define_filter("full_name", FilterType.EQUALITY, DataType.STRING, "The full name of the driver.")
157
+ f1_api.define_filter("name_acronym", FilterType.EQUALITY, DataType.STRING, "Three-letter acronym of the driver's name.")
158
+ f1_api.define_filter("team_name", FilterType.EQUALITY, DataType.STRING, "The name of the driver's team.")
159
+ f1_api.define_filter("gap_to_leader", FilterType.COMPARISON, DataType.INTEGER, "The time gap to the race leader in seconds, +1 LAP if lapped, or null for the race leader.")
160
+ f1_api.define_filter("interval", FilterType.COMPARISON, DataType.INTEGER, "The time gap to the car ahead in seconds, +1 LAP if lapped, or null for the race leader.")
161
+ f1_api.define_filter("date_start", FilterType.COMPARISON, DataType.DATETIME, "The UTC starting date and time, in ISO 8601 format.")
162
+ f1_api.define_filter("date_end", FilterType.COMPARISON, DataType.DATETIME, "The UTC ending date and time, in ISO 8601 format.")
163
+ f1_api.define_filter("is_pit_out_lap", FilterType.EQUALITY, DataType.BINARY, "A boolean value indicating whether the lap is an out lap from the pit (true if it is, false otherwise).")
164
+ f1_api.define_filter("lap_duration", FilterType.COMPARISON, DataType.INTEGER, "The total time taken, in seconds, to complete the entire lap.")
165
+ f1_api.define_filter("lap_number", FilterType.EQUALITY, DataType.INTEGER, "The sequential number of the lap within the session (starts at 1).")
166
+ f1_api.define_filter("circuit_key", FilterType.EQUALITY, DataType.STRING, "The unique identifier for the circuit where the event takes place.")
167
+ f1_api.define_filter("circuit_short_name", FilterType.EQUALITY, DataType.STRING, "The short or common name of the circuit where the event takes place.")
168
+ f1_api.define_filter("country_key", FilterType.EQUALITY, DataType.STRING, "The unique identifier for the country where the event takes place.")
169
+ f1_api.define_filter("country_name", FilterType.EQUALITY, DataType.STRING, "The name of the country where the event takes place.")
170
+ f1_api.define_filter("location", FilterType.EQUALITY, DataType.STRING, "The city or geographical location where the event takes place.")
171
+ f1_api.define_filter("meeting_name", FilterType.EQUALITY, DataType.STRING, "The name of the meeting.")
172
+ f1_api.define_filter("meeting_official_name", FilterType.EQUALITY, DataType.STRING, "The official name of the meeting.")
173
+ f1_api.define_filter("year", FilterType.EQUALITY, DataType.INTEGER, "The year of the event.")
174
+ f1_api.define_filter("pit_duration", FilterType.COMPARISON, DataType.INTEGER, "The time spent in the pit, from entering to leaving the pit lane, in seconds.")
175
+ f1_api.define_filter("position", FilterType.EQUALITY, DataType.INTEGER, "Position of the driver (starts at 1).", [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20])
176
+ f1_api.define_filter("category", FilterType.EQUALITY, DataType.STRING, "The category of the event (CarEvent, Drs, Flag, SafetyCar)", ["CarEvent", "Drs", "Flag", "SafetyCar"])
177
+ f1_api.define_filter("flag", FilterType.EQUALITY, DataType.STRING, "The flag displayed to the drivers.", ["Green", "Yellow", "Red", "Black", "White", "Blue", "Checkered", "White"])
178
+ f1_api.define_filter("message", FilterType.EQUALITY, DataType.STRING, "Description of the event or action.")
179
+ f1_api.define_filter("session_name", FilterType.EQUALITY, DataType.STRING, "The name of the session (Practice 1, Qualifying, Race, ...).")
180
+ f1_api.define_filter("session_type", FilterType.EQUALITY, DataType.STRING, "The type of the session (Practice, Qualifying, Race, ...).")
181
+ f1_api.define_filter("compound", FilterType.EQUALITY, DataType.STRING, "The specific compound of tyre used during the stint (SOFT, MEDIUM, HARD, ...).", ["SOFT", "MEDIUM", "HARD", "INTERMEDIATE", "WET"])
182
+ f1_api.define_filter("lap_end", FilterType.COMPARISON, DataType.INTEGER, "Number of the last completed lap in this stint.")
183
+ f1_api.define_filter("lap_start", FilterType.COMPARISON, DataType.INTEGER, "Number of the initial lap in this stint (starts at 1).")
184
+ f1_api.define_filter("stint_number", FilterType.EQUALITY, DataType.INTEGER, "The sequential number of the stint within the session (starts at 1).")
185
+ f1_api.define_filter("tyre_age_at_start", FilterType.COMPARISON, DataType.INTEGER, "The age of the tyres at the start of the stint, in laps completed.")
186
+ f1_api.define_filter("air_temperature", FilterType.COMPARISON, DataType.INTEGER, "Air temperature (°C).")
187
+ f1_api.define_filter("humidity", FilterType.COMPARISON, DataType.INTEGER, "Humidity percentage.")
188
+ f1_api.define_filter("pressure", FilterType.COMPARISON, DataType.INTEGER, "Air pressure (mbar).")
189
+ f1_api.define_filter("rainfall", FilterType.COMPARISON, DataType.INTEGER, "Whether there is rainfall.")
190
+ f1_api.define_filter("track_temperature", FilterType.COMPARISON, DataType.INTEGER, "Track temperature (°C).")
191
+ f1_api.define_filter("wind_direction", FilterType.COMPARISON, DataType.INTEGER, "Wind direction (°), from 0° to 359°.")
192
+ f1_api.define_filter("wind_speed", FilterType.COMPARISON, DataType.INTEGER, "Wind speed (m/s).")
193
+
194
+
195
+ # Register API endpoints with their supported filters
196
+ f1_api.register_endpoint("car_data", "date", "driver_number", "meeting_key", "session_key", "speed")
197
+ f1_api.register_endpoint("drivers", "session_key", "meeting_key", "country_code", "driver_number", "first_name", "last_name", "full_name", "name_acronym", "team_name")
198
+ f1_api.register_endpoint("intervals", "date", "driver_number", "meeting_key", "session_key", "gap_to_leader", "interval")
199
+ f1_api.register_endpoint("laps", "date_start", "driver_number", "meeting_key", "session_key", "lap_duration", "lap_number", "is_pit_out_lap")
200
+ f1_api.register_endpoint("location", "date", "driver_number", "meeting_key", "session_key")
201
+ f1_api.register_endpoint("meetings", "circuit_key", "circuit_short_name", "country_code", "country_key", "country_name", "date_start", "location", "meeting_key", "meeting_name", "meeting_official_name", "year")
202
+ f1_api.register_endpoint("pit", "date", "driver_number", "lap_number", "meeting_key", "session_key", "pit_duration")
203
+ f1_api.register_endpoint("position", "date", "driver_number", "meeting_key", "session_key", "position")
204
+ f1_api.register_endpoint("race_control", "category", "date", "driver_number", "meeting_key", "session_key", "flag", "message", "lap_number")
205
+ f1_api.register_endpoint("sessions", "circuit_key", "circuit_short_name", "country_code", "country_key", "country_name", "date_start", "date_end", "location", "session_name", "session_type", "session_key", "meeting_key", "year")
206
+ f1_api.register_endpoint("stints", "compound", "driver_number", "lap_end", "lap_start", "meeting_key", "session_key", "stint_number", "tyre_age_at_start")
207
+ f1_api.register_endpoint("team_radio", "date", "driver_number", "meeting_key", "session_key")
208
+ f1_api.register_endpoint("weather", "air_temperature", "date", "humidity", "pressure", "rainfall", "track_temperature", "wind_direction", "wind_speed", "meeting_key", "session_key")
openf1_tools.py ADDED
@@ -0,0 +1,169 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import json
2
+ from urllib.request import urlopen
3
+ from openf1_registry import f1_api
4
+
5
+
6
+ api_endpoints = {
7
+ "sessions": "sessions?",
8
+ "weather": "weather?",
9
+ "locations": "locations?",
10
+ "drivers": "drivers?",
11
+ "intervals": "intervals?",
12
+ "laps": "laps?",
13
+ "car_data": "car_data?",
14
+ "pit": "pit?",
15
+ "position": "position?",
16
+ "race_control": "race_control?",
17
+ "stints": "stints?",
18
+ "team_radio": "team_radio?",
19
+ }
20
+
21
+
22
+ ### Essential tools ###
23
+
24
+ def get_api_endpoint(endpoint: str) -> dict:
25
+ """
26
+ Retrieve the API endpoint URL and filter metadata for a given OpenF1 endpoint.
27
+
28
+ Args:
29
+ endpoint (str): The name of the OpenF1 API endpoint (e.g., 'sessions', 'laps').
30
+
31
+ Returns:
32
+ dict: A dictionary containing:
33
+ - status (str): 'success' if endpoint exists, 'error' otherwise
34
+ - api_string (str): The full API URL for the endpoint
35
+ - filter_metadata (dict): Available filters for the endpoint
36
+ """
37
+ try:
38
+ url = f1_api.base_url + api_endpoints.get(endpoint, None)
39
+ return {
40
+ "status": "success",
41
+ "api_string": url,
42
+ "filter_metadata": f1_api.get_endpoint_filters(endpoint)
43
+ }
44
+ except:
45
+ return {
46
+ "status": "error",
47
+ "api_string": f"Endpoint {endpoint} not found. Available endpoints: {f1_api.list_all_endpoints()}",
48
+ "filter_metadata": dict()
49
+ }
50
+
51
+ def get_filter_string(filter_name: str, filter_value: str, operator: str = "=") -> str:
52
+ """
53
+ Create a filter string for OpenF1 API requests.
54
+
55
+ Args:
56
+ filter_name (str): The name of the filter to apply.
57
+ filter_value (str): The value to filter by.
58
+ operator (str, optional): The comparison operator. Defaults to "=".
59
+
60
+ Returns:
61
+ str: Formatted filter string that can be appended to an API request.
62
+ """
63
+ return f"{filter_name}{operator}{filter_value}&"
64
+
65
+ def apply_filters(api_string: str, *filters: str) -> str:
66
+ """
67
+ Apply one or more filter strings to an API endpoint URL.
68
+
69
+ Args:
70
+ api_string (str): The base API endpoint URL.
71
+ *filters (str): Variable number of filter strings to apply.
72
+
73
+ Returns:
74
+ str: The complete API URL with all filters applied.
75
+ """
76
+ if filters:
77
+ for filter in filters:
78
+ api_string += filter
79
+ return api_string.rstrip("&") # Remove trailing & for last filter
80
+
81
+ def send_request(api_string: str) -> dict:
82
+ """
83
+ Send an HTTP GET request to the specified API endpoint and return the JSON response.
84
+
85
+ Args:
86
+ api_string (str): The complete API URL to send the request to.
87
+
88
+ Returns:
89
+ dict: The JSON response parsed as a Python dictionary.
90
+
91
+ Raises:
92
+ Exception: If there's an error during the HTTP request or JSON parsing.
93
+ """
94
+ try:
95
+ response = urlopen(api_string)
96
+ data = json.loads(response.read().decode('utf-8'))
97
+ return data
98
+ except Exception as e:
99
+ print(f"Error: {e}")
100
+ raise
101
+
102
+ ### LLM helper functions ###
103
+
104
+ def get_api_endpoints() -> dict:
105
+ """
106
+ Retrieve a list of all available OpenF1 API endpoints.
107
+
108
+ Returns:
109
+ dict: A dictionary containing a single key 'endpoints' with a list of
110
+ available endpoint names as strings.
111
+ """
112
+ return {
113
+ "endpoints": f1_api.list_all_endpoints(),
114
+ }
115
+
116
+ def get_endpoint_info(endpoint: str) -> dict:
117
+ """
118
+ Retrieve detailed information about a specific OpenF1 API endpoint.
119
+
120
+ Args:
121
+ endpoint (str): The name of the endpoint to get information about.
122
+
123
+ Returns:
124
+ dict: A dictionary containing:
125
+ - endpoint (str): The name of the endpoint
126
+ - endpoint_filters (list): Available filters for this endpoint
127
+ - endpoint_help (str): Help text describing the endpoint's purpose and usage
128
+ """
129
+ return {
130
+ "endpoint": endpoint,
131
+ "endpoint_filters": f1_api.get_endpoint_filters(endpoint),
132
+ "endpoint_help": f1_api.get_endpoint_help(endpoint)
133
+ }
134
+
135
+ def get_filter_info(filter_name: str) -> dict:
136
+ """
137
+ Retrieve detailed information about a specific OpenF1 API filter.
138
+
139
+ Args:
140
+ filter_name (str): The name of the filter to get information about.
141
+
142
+ Returns:
143
+ dict: A dictionary containing:
144
+ - filter_name (str): The name of the filter
145
+ - filter_metadata (dict): Metadata about the filter, including
146
+ description, valid values, and usage examples
147
+ """
148
+ return {
149
+ "filter_name": filter_name,
150
+ "filter_metadata": f1_api.get_filter_help(filter_name)
151
+ }
152
+
153
+
154
+
155
+
156
+
157
+ if __name__ == "__main__":
158
+
159
+ from pprint import pprint
160
+
161
+ data = get_api_endpoint("sessions")
162
+ pprint(data)
163
+
164
+
165
+ pprint(get_endpoint_info("sessions"))
166
+
167
+
168
+ # response = send_request(data["api_string"])
169
+ # pprint(response)
tools.py DELETED
@@ -1,243 +0,0 @@
1
- """
2
- This module provides tools for interacting with Formula 1 data using the FastF1 library.
3
-
4
- Tools to implement
5
- - driver info
6
- - compare drivers
7
- """
8
-
9
- import json
10
- import fastf1
11
- import gradio as gr
12
- import pandas as pd
13
- from PIL import Image
14
- from typing import Union
15
- from fastf1.core import Session
16
-
17
- # Local modules
18
- from utils import parser_utils, track_utils
19
- from utils.constants import (
20
- AVAILABLE_SESSION_TYPES,
21
- DRIVER_DETAILS
22
- )
23
-
24
- # Custom types
25
- gp = Union[str, int]
26
- session_type = Union[str, int, None]
27
-
28
- ### FastF1 tools ###
29
-
30
- def get_session(year: int, round: gp, session_type: session_type) -> Session:
31
- """Retrieve a specific Formula 1 session.
32
-
33
- Args:
34
- year (int): The season year (e.g., 2024)
35
- round (str | int): The race round number or name (e.g., 1 or 'Monaco')
36
- session_type (str | int | None): Type of session (e.g., 'FP1', 'Q', 'R')
37
-
38
- Returns:
39
- Session: A FastF1 Session object for the specified parameters
40
-
41
- Note:
42
- If session_type is a string and not in AVAILABLE_SESSION_TYPES,
43
- returns an error message string instead.
44
- """
45
-
46
- # Check if session type is valid
47
- if isinstance(session_type, str):
48
- if session_type.lower() not in AVAILABLE_SESSION_TYPES:
49
- return f"Session type {session_type} is not available. Supported session types: {AVAILABLE_SESSION_TYPES}"
50
-
51
- return fastf1.get_session(year, round, session_type)
52
-
53
- def get_season_calendar(year: int) -> str:
54
- """Get the complete race calendar for a specific F1 season.
55
-
56
- Args:
57
- year (int): The season year to get the calendar for
58
-
59
- Returns:
60
- str: Formatted string containing the season calendar
61
- """
62
- season_calendar = fastf1.get_event_schedule(year)
63
- return parser_utils.parse_season_calendar(season_calendar)
64
-
65
- def get_event_info(year: int, round: gp, format: str) -> str:
66
- """Retrieve information about a specific Formula 1 event.
67
-
68
- Args:
69
- year (int): The season year
70
- round (str | int): The race round number or name
71
- format (str): Output format ('human' for readable text, 'LLM' for structured data)
72
-
73
- Returns:
74
- str: Formatted event information based on the specified format
75
- """
76
-
77
- event = fastf1.get_session(year, round, "race").event # Event object is the same for all sessions, so hardcode "race"
78
- if format == "human":
79
- data_interval = f"{event['Session1DateUtc'].date()} - {event['Session5DateUtc'].date()}"
80
- event_string = f"Round {event['RoundNumber']} : {event['EventName']} - {event['Location']}, {event['Country']} ({data_interval})"
81
- return event_string
82
- elif format == "LLM":
83
- return parser_utils.parse_event_info(event)
84
-
85
-
86
- def get_constructor_standings(year: int) -> str:
87
- """Retrieve constructor championship standings for a given year.
88
-
89
- Args:
90
- year (int): The season year
91
-
92
- Returns:
93
- str: Constructor championship standings
94
- """
95
- pass
96
-
97
- def get_driver_standings(year: int) -> str:
98
- """Retrieve driver championship standings for a given year.
99
-
100
- Args:
101
- year (int): The season year
102
-
103
- Returns:
104
- str: Driver championship standings
105
- """
106
- pass
107
-
108
- def driver_championship_standings(year: int, driver_name: str) -> str:
109
- """Get the championship standing for a specific driver in a given year.
110
-
111
- Args:
112
- year (int): The season year
113
- driver_name (str): Full name of the driver (e.g., 'Lewis Hamilton')
114
-
115
- Returns:
116
- str: Formatted string with driver's position, points, and wins
117
- """
118
-
119
- with open("assets/driver_abbreviations.json") as f:
120
- driver_abbreviations = json.load(f)
121
- driver_abbreviation = driver_abbreviations[driver_name]
122
- ergast = fastf1.ergast.Ergast()
123
- driver_standings = ergast.get_driver_standings(year).content[0]
124
- driver_standing = driver_standings[["position", "points", "wins", "driverCode"]].reset_index(drop=True)
125
- driver_standing = driver_standing[driver_standing["driverCode"] == driver_abbreviation]
126
- 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"
127
- 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"
128
- return standings_string
129
-
130
- def constructor_championship_standings(year: int, constructor_name: str) -> str:
131
- """Get the championship standing for a specific constructor in a given year.
132
-
133
- Args:
134
- year (int): The season year
135
- constructor_name (str): Name of the constructor team (e.g., 'Mercedes')
136
-
137
- Returns:
138
- str: Formatted string with constructor's position, points, and wins
139
- """
140
-
141
- team_mapping = {
142
- "McLaren": "McLaren",
143
- "Ferrari": "Ferrari",
144
- "Red Bull Racing": "Red Bull",
145
- "Mercedes": "Mercedes",
146
- "Aston Martin": "Aston Martin",
147
- "Alpine": "Alpine F1 Team",
148
- "Haas": "Haas F1 Team",
149
- "Racing Bulls": "RB F1 Team",
150
- "Williams": "Williams",
151
- "Kick Sauber": "Sauber"
152
- }
153
-
154
- ergast = fastf1.ergast.Ergast()
155
- constructor_standings = ergast.get_constructor_standings(year).content[0]
156
- constructor_standing = constructor_standings[["position", "points", "wins", "constructorName"]].reset_index(drop=True)
157
- mapped_name = team_mapping[constructor_name]
158
- constructor_standing = constructor_standing[constructor_standing["constructorName"] == mapped_name]
159
- 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"
160
- 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"
161
- return standings_string
162
-
163
- def track_visualization(year: int, round: gp, visualization_type: str, driver_name: str) -> Image.Image:
164
- """Generate a visualization of the track with specified data.
165
-
166
- Args:
167
- year (int): The season year
168
- round (str | int): The race round number or name
169
- visualization_type (str): Type of visualization ('speed', 'corners', or 'gear')
170
- driver_name (str): Name of the driver for driver-specific visualizations
171
-
172
- Returns:
173
- Image.Image: A PIL Image object containing the visualization
174
- """
175
-
176
- session = get_session(year, round, "race")
177
- session.load()
178
-
179
- if visualization_type == "speed":
180
- return track_utils.create_track_speed_visualization(session, driver_name)
181
- elif visualization_type == "corners":
182
- return track_utils.create_track_corners_visualization(session)
183
- elif visualization_type == "gear":
184
- return track_utils.create_track_gear_visualization(session)
185
-
186
- def get_session_results(year: int, round: gp, session_type: session_type) -> pd.DataFrame:
187
- """Retrieve and format the results of a specific session.
188
-
189
- Args:
190
- year (int): The season year
191
- round (str | int): The race round number or name
192
- session_type (str | int | None): Type of session (e.g., 'Q', 'R', 'Sprint')
193
-
194
- Returns:
195
- pd.DataFrame: DataFrame containing the session results
196
-
197
- Raises:
198
- gr.Error: If the session type is not supported for the specified round
199
- """
200
-
201
- try:
202
- session = get_session(year, round, session_type)
203
- session.load(telemetry=False)
204
- results = session.results
205
- except ValueError as e:
206
- 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.")
207
-
208
- df = results[["DriverNumber", "Abbreviation", "FullName", "Position", "GridPosition", "Points", "Status", "Q1", "Q2", "Q3"]]
209
- df["Name"] = df.apply(lambda row: f"{row['FullName']} ({row['Abbreviation']} • {row['DriverNumber']})", axis=1)
210
- df = df.drop(columns=["FullName", "Abbreviation", "DriverNumber"])
211
- df = df.rename(columns={"Position": "Pos", "GridPosition": "Grid Pos"})
212
-
213
- # Process results based on session type
214
- if session_type in ["race", "sprint"]:
215
- df = df[["Pos", "Name", "Points", "Grid Pos", "Status"]]
216
- elif "qualifying" in session_type:
217
- 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 "-"))
218
- df = df[["Pos", "Name", "Q1", "Q2", "Q3"]]
219
- return df
220
-
221
- def get_driver_info(driver_name: str) -> str:
222
- """Retrieve detailed information about a specific driver.
223
-
224
- Args:
225
- driver_name (str): Full name of the driver (e.g., 'Max Verstappen')
226
-
227
- Returns:
228
- str: Formatted string with driver's details including name, team, number,
229
- nationality, and a brief summary
230
- """
231
- driver = DRIVER_DETAILS[driver_name]
232
- driver_info_string = f"{driver_name} ({driver['birth_date']})\n{driver['team']} #{driver['number']}\n{driver['nationality']}\n\n{driver['summary']}"
233
- return driver_info_string
234
-
235
- ### OpenF1 tools ###
236
-
237
-
238
-
239
-
240
-
241
- if __name__ == "__main__":
242
- session = get_session(2024, 1, "fp1")
243
- session.load()
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
utils/request_utils.py DELETED
@@ -1 +0,0 @@
1
- """ Utils file for handling OpenF1 requests """