|
|
@@ -1,42 +1,87 @@ |
|
|
from typing import Optional, TYPE_CHECKING |
|
|
|
|
|
|
|
|
|
|
|
from fastf1.ergast import Ergast |
|
|
|
|
|
import slugify |
|
|
|
|
|
# Import FastF1 for dynamic circuit data |
|
|
|
|
|
import fastf1 |
|
|
|
|
|
|
|
|
import re |
|
|
import logging |
|
|
import logging |
|
|
|
|
|
import urllib.request |
|
|
|
|
|
import json |
|
|
|
|
|
from functools import lru_cache |
|
|
|
|
|
from typing import Optional, TYPE_CHECKING |
|
|
|
|
|
|
|
|
if TYPE_CHECKING: |
|
|
if TYPE_CHECKING: |
|
|
from circuits_service import CircuitService |
|
|
from circuits_service import CircuitService |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def slugify(text: str) -> str: |
|
|
|
|
|
text = text.lower().strip() |
|
|
|
|
|
replacements = { |
|
|
|
|
|
'ü': 'u', 'ö': 'o', 'ä': 'a', 'é': 'e', 'è': 'e', 'ê': 'e', 'ë': 'e', |
|
|
|
|
|
'à': 'a', 'â': 'a', 'î': 'i', 'ï': 'i', 'ô': 'o', 'û': 'u', 'ñ': 'n', |
|
|
|
|
|
'ã': 'a', 'á': 'a', 'í': 'i', 'ó': 'o', 'ú': 'u', |
|
|
|
|
|
} |
|
|
|
|
|
for k, v in replacements.items(): |
|
|
|
|
|
text = text.replace(k, v) |
|
|
|
|
|
text = re.sub(r'[^a-z0-9]+', '-', text) |
|
|
|
|
|
return text.strip('-') |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class ErgastService: |
|
|
class ErgastService: |
|
|
|
|
|
JOLPICA_BASE = 'https://api.jolpi.ca/ergast/f1' |
|
|
|
|
|
|
|
|
|
|
|
# Country name normalization: Jolpica country → our country slug |
|
|
country_map = { |
|
|
country_map = { |
|
|
"united-kingdom": "uk", |
|
|
|
|
|
|
|
|
'UK': 'uk', |
|
|
|
|
|
'UAE': 'united-arab-emirates', |
|
|
|
|
|
'USA': 'united-states', |
|
|
|
|
|
'Korea': 'korea', |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
def __init__(self, circuits_service: 'CircuitService' = None): |
|
|
def __init__(self, circuits_service: 'CircuitService' = None): |
|
|
self.circuits_service = circuits_service |
|
|
self.circuits_service = circuits_service |
|
|
# Configure logging |
|
|
|
|
|
logging.basicConfig(level=logging.INFO) |
|
|
|
|
|
self.logger = logging.getLogger(__name__) |
|
|
self.logger = logging.getLogger(__name__) |
|
|
|
|
|
self._circuit_name_index = self._build_circuit_name_index() |
|
|
|
|
|
|
|
|
# Initialize FastF1 cache |
|
|
|
|
|
fastf1.Cache.enable_cache('cache') |
|
|
|
|
|
|
|
|
def _build_circuit_name_index(self) -> dict[str, tuple[str, str, str]]: |
|
|
|
|
|
"""Build a lookup from slugified circuit name → (country_slug, locality_slug, circuit_slug).""" |
|
|
|
|
|
index = {} |
|
|
|
|
|
if not self.circuits_service: |
|
|
|
|
|
return index |
|
|
|
|
|
for country_slug, country in self.circuits_service.circuit_data.items(): |
|
|
|
|
|
for locality_slug, locality in country.localities.items(): |
|
|
|
|
|
for circuit_slug, circuit in locality.circuits.items(): |
|
|
|
|
|
# Index by our circuit slug (which is the slugified circuit name) |
|
|
|
|
|
index[circuit_slug] = (country_slug, locality_slug, circuit_slug) |
|
|
|
|
|
# Also index by ASCII-slugified version for unicode mismatches |
|
|
|
|
|
ascii_slug = slugify(circuit.name) |
|
|
|
|
|
if ascii_slug not in index: |
|
|
|
|
|
index[ascii_slug] = (country_slug, locality_slug, circuit_slug) |
|
|
|
|
|
return index |
|
|
|
|
|
|
|
|
def find_slugs_for_grand_prix(self, grand_prix_name: str, season: int) -> Optional[tuple[str, str, str]]: |
|
|
|
|
|
""" |
|
|
|
|
|
Find and return the circuit layout file path based on Ergast F1 API race data format. |
|
|
|
|
|
""" |
|
|
|
|
|
|
|
|
@lru_cache(maxsize=128) |
|
|
|
|
|
def _fetch_season_schedule(self, season: int) -> list[dict]: |
|
|
|
|
|
"""Fetch and cache the race schedule for a season from Jolpica.""" |
|
|
|
|
|
url = f'{self.JOLPICA_BASE}/{season}.json?limit=100' |
|
|
try: |
|
|
try: |
|
|
ergast = Ergast() |
|
|
|
|
|
f1_event = fastf1.events.get_event(int(season), grand_prix_name) |
|
|
|
|
|
f1_seasons = ergast.get_race_schedule(season=season) |
|
|
|
|
|
f1_season = f1_seasons[f1_seasons["raceName"] == grand_prix_name] |
|
|
|
|
|
|
|
|
|
|
|
country_slug = slugify.slugify(f1_event["Country"]) |
|
|
|
|
|
city_slug = slugify.slugify(f1_season["locality"].values[0]) |
|
|
|
|
|
circuit_slug = slugify.slugify(f1_season["circuitName"].values[0]) |
|
|
|
|
|
return self.country_map[country_slug] if country_slug in self.country_map else country_slug, city_slug, circuit_slug |
|
|
|
|
|
|
|
|
resp = urllib.request.urlopen(url, timeout=10) |
|
|
|
|
|
data = json.loads(resp.read()) |
|
|
|
|
|
return data['MRData']['RaceTable']['Races'] |
|
|
except Exception as e: |
|
|
except Exception as e: |
|
|
self.logger.error( |
|
|
|
|
|
f"Error finding circuit layout in Ergast for grand prix {grand_prix_name} in season {season}: {str(e)}") |
|
|
|
|
|
return None |
|
|
|
|
|
|
|
|
self.logger.error(f"Failed to fetch Jolpica schedule for {season}: {e}") |
|
|
|
|
|
return [] |
|
|
|
|
|
|
|
|
|
|
|
def find_slugs_for_grand_prix(self, grand_prix_name: str, season: int) -> Optional[tuple[str, str, str]]: |
|
|
|
|
|
"""Resolve a grand prix name + season to (country_slug, locality_slug, circuit_slug).""" |
|
|
|
|
|
races = self._fetch_season_schedule(season) |
|
|
|
|
|
|
|
|
|
|
|
# Find the matching race |
|
|
|
|
|
race = next((r for r in races if r['raceName'] == grand_prix_name), None) |
|
|
|
|
|
if not race: |
|
|
|
|
|
self.logger.error(f"Race '{grand_prix_name}' not found in {season} schedule") |
|
|
|
|
|
return None, None, None |
|
|
|
|
|
|
|
|
|
|
|
circuit = race['Circuit'] |
|
|
|
|
|
circuit_name_slug = slugify(circuit['circuitName']) |
|
|
|
|
|
|
|
|
|
|
|
# Try matching by slugified circuit name |
|
|
|
|
|
if circuit_name_slug in self._circuit_name_index: |
|
|
|
|
|
return self._circuit_name_index[circuit_name_slug] |
|
|
|
|
|
|
|
|
|
|
|
self.logger.warning(f"Circuit '{circuit['circuitName']}' (slug: {circuit_name_slug}) not found in index") |
|
|
|
|
|
return None, None, None |