From 15af1627fbda3702829be5c657bc7340a14a5fc3 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 8 May 2026 14:54:02 +0200 Subject: [PATCH] Replace FastF1 with Jolpica API for circuit resolution MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FastF1's get_event() couldn't resolve several grand prix names (Qatar, Brazilian, Mexican, Barcelona), causing 500/404 errors. Now uses Jolpica's race schedule directly — slugifies circuitName to match against circuits.json with ASCII fallback for unicode. --- circuits_service.py | 6 --- ergast_service.py | 99 ++++++++++++++++++++++++++++++++------------- 2 files changed, 72 insertions(+), 33 deletions(-) diff --git a/circuits_service.py b/circuits_service.py index b70d6cc..2f3cb05 100644 --- a/circuits_service.py +++ b/circuits_service.py @@ -4,9 +4,6 @@ from typing import Optional import functools from enum import Enum -# Import FastF1 for dynamic circuit data -import fastf1 - from ergast_service import ErgastService from models.circuit import Circuit from models.country import Country @@ -35,9 +32,6 @@ class CircuitService: logging.basicConfig(level=logging.INFO) self.logger = logging.getLogger(__name__) - # Initialize FastF1 cache - fastf1.Cache.enable_cache('cache') - @functools.lru_cache() def _load_circuit_data(self) -> dict[str, Country]: """Load and cache circuit data""" diff --git a/ergast_service.py b/ergast_service.py index e16449a..e0fa5b2 100644 --- a/ergast_service.py +++ b/ergast_service.py @@ -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 urllib.request +import json +from functools import lru_cache +from typing import Optional, TYPE_CHECKING if TYPE_CHECKING: 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: + JOLPICA_BASE = 'https://api.jolpi.ca/ergast/f1' + + # Country name normalization: Jolpica country → our country slug country_map = { - "united-kingdom": "uk", + 'UK': 'uk', + 'UAE': 'united-arab-emirates', + 'USA': 'united-states', + 'Korea': 'korea', } + def __init__(self, circuits_service: 'CircuitService' = None): self.circuits_service = circuits_service - # Configure logging - logging.basicConfig(level=logging.INFO) 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: - 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: - self.logger.error( - f"Error finding circuit layout in Ergast for grand prix {grand_prix_name} in season {season}: {str(e)}") - return None \ No newline at end of file + 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