f1-circuits/circuits_service.py

import json
import logging
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
from models.locality import Locality
from models.model_builder import parse_circuits_json
from models.track_layout import TrackLayout


# Define supported formats
class Format(str, Enum):
    PNG = "png"
    SVG = "svg"
    GEOJSON = "geojson"


class CircuitService:
    def __init__(self, circuit_data_path: str = "circuits/circuits.json"):
        self.circuit_data_path = circuit_data_path
        self.circuit_data = self._load_circuit_data()
        self._layout_cache = {}
        self._circuit_file_cache = {}
        self._grand_prix_cache = {}
        self.ergast_service = ErgastService(self)

        # Configure logging
        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"""
        with open(self.circuit_data_path) as f:
            return parse_circuits_json(json.load(f))

    def find_layout_slug_for_year(self, circuit: Circuit, year: int) -> Optional[str]:
        """Find the appropriate layout slug for a given year"""
        cache_key = f"{circuit.name}-{year}"

        if cache_key in self._layout_cache:
            return self._layout_cache[cache_key]

        for layout_slug, _ in circuit.layouts.items():
            # Check different layout slug formats
            if '-' in layout_slug:
                # Handle range format
                start_year, *end = layout_slug.split('-')
                start_year = int(start_year)
                if not end[0]:
                    # Format: 1967- (no end year)
                    if year >= start_year:
                        self._layout_cache[cache_key] = layout_slug
                        return layout_slug
                else:
                    # Format: 1967-1968
                    end_year = int(end[0])
                    if start_year <= year <= end_year:
                        self._layout_cache[cache_key] = layout_slug
                        return layout_slug
            else:
                # Format: 1967 (single year)
                if year == int(layout_slug):
                    self._layout_cache[cache_key] = layout_slug
                    return layout_slug
        return None

    def get_countries_list(self) -> Optional[list[Country]]:
        """Get list of all available countries with their slugs"""
        return [country for _, country in self.circuit_data.items()]

    def get_country_details(self, country_slug: str) -> Optional[Country]:
        """Get details for a specific country"""
        return self.circuit_data.get(country_slug)

    def get_localities_list(self, country_slug: str) -> Optional[list[Locality]]:
        """Get list of cities for a specific country"""
        return [locality for _, locality in self.circuit_data[country_slug].localities.items()]

    def get_locality_details(self, country_slug: str, locality_slug: str) -> Optional[Locality]:
        """Get details for a specific city"""
        return self.circuit_data[country_slug].localities.get(locality_slug)

    def get_circuits_list(self, country_slug: str, locality_slug: str) -> Optional[list[Circuit]]:
        """Get list of circuits for a specific city"""
        return [circuit_data for _, circuit_data in self.circuit_data[country_slug].localities.get(locality_slug).circuits.items()]

    def get_circuit_details(self, country_slug: str, locality_slug: str, circuit_slug: str) -> Optional[Circuit]:
        """Get circuit details and available formats"""
        return self.circuit_data[country_slug].localities.get(locality_slug).circuits.get(circuit_slug)

    def get_layout_details(self, country_slug: str, locality_slug: str, circuit_slug: str, layout_slug: str) -> Optional[TrackLayout]:
        """Get details for a specific layout"""
        return self.circuit_data[country_slug].localities.get(locality_slug).circuits.get(circuit_slug).layouts.get(
            layout_slug)

    def get_circuit_layout_by_ergast_data(self, grand_prix_name: str, season: int) -> Optional[TrackLayout]:
        """
        Find and return the circuit layout file path based on Ergast F1 API race data format.
        """
        country_slug, locality_slug, circuit_slug = self.ergast_service.find_slugs_for_grand_prix(grand_prix_name, season)
        if country_slug is None or locality_slug is None or circuit_slug is None:
            self.logger.error(f"Error finding circuit layout in Ergast for grand prix {grand_prix_name} in season {season}")
            return None

        circuit = self.get_circuit_details(country_slug, locality_slug, circuit_slug)
        if circuit is None:
            self.logger.error(f"Error finding circuit details for Ergast country, locality and circuit slug: {country_slug}, {locality_slug}, {circuit_slug}")
            return None

        layout_slug = self.find_layout_slug_for_year(circuit, season)
        if layout_slug is None:
            self.logger.error(f"Error finding layout slug for year {season} for circuit {circuit_slug}")
            return None

        return self.get_layout_details(country_slug, locality_slug, circuit_slug, layout_slug)