from fastapi import FastAPI, HTTPException, Response from fastapi.responses import FileResponse, Response as FastAPIResponse, PlainTextResponse from fastapi.middleware.cors import CORSMiddleware from pathlib import Path import logging # Import our CircuitService from circuits_service import CircuitService, Format from ergast_service import ErgastService from posters_service import PostersService, BackgroundStyle from models.track_layout import TrackLayout # Initialize FastAPI app app = FastAPI( title="F1 Track Layouts API", description="API serving F1 track layout images, vectors and GeoJSON files", version="1.0.0" ) # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Initialize CircuitService circuit_service = CircuitService() ergast_service = ErgastService(circuit_service) posters_service = PostersService(circuit_service, ergast_service) def get_content_disposition(format: Format) -> str: if format == Format.PNG: return "inline" # SVG and GeoJSON files as attachments for security/convention return "attachment" def get_media_type(format: Format) -> str: media_types = { Format.PNG: "image/png", Format.SVG: "image/svg+xml", Format.GEOJSON: "application/geo+json" } return media_types.get(format, "application/octet-stream") # fallback for unknown formats @app.get("/") async def root(): """Root endpoint returning API information""" return { "name": "F1 Track Layouts API", "version": "1.0.0", "endpoints": [ "/circuits", "/circuits/{country_slug}", "/circuits/{country_slug}/{city_slug}", "/circuits/{country_slug}/{city_slug}/{circuit_slug}", "/circuits/{country_slug}/{city_slug}/{circuit_slug}/layout/{layout_slug}", "/tracks/{country_slug}/{city_slug}/{circuit_slug}/year/{year}/{format}", "/tracks/{country_slug}/{city_slug}/{circuit_slug}/layout/{layout_slug}/{format}", "/grand-prix/{grand_prix_name}/year/{year}", "/posters/race/{grand_prix_name}/{season}.png", "/posters/season/{season}.png", ] } @app.get("/posters/race/{grand_prix_name}/{season}.png") async def get_race_poster(grand_prix_name: str, season: int, bg: BackgroundStyle = BackgroundStyle.AERIAL): """Plex-ready 1000x1500 race poster (PNG). Cached on disk per (race, bg). `?bg=` selects the backdrop: - aerial (default) — Esri World Imagery satellite at the circuit's coords - osm — OpenStreetMap Carto street-map at the circuit's coords - none — plain black background (skip tile fetch) """ png = posters_service.render_race_poster(grand_prix_name, season, bg=bg) if not png: raise HTTPException(status_code=404, detail=f"No poster available for {grand_prix_name} {season}") return FastAPIResponse( content=png, media_type="image/png", headers={"Cache-Control": "public, max-age=31536000"}, ) @app.get("/posters/season/{season}.png") async def get_season_poster(season: int): """Plex-ready 1000x1500 season poster (PNG). Cached on disk.""" png = posters_service.render_season_poster(season) return FastAPIResponse( content=png, media_type="image/png", headers={"Cache-Control": "public, max-age=31536000"}, ) @app.get("/posters/session/{grand_prix_name}/{season}/{session_slug}.png") async def get_session_thumb(grand_prix_name: str, season: int, session_slug: str): """Plex episode-thumb (1920x1080) for a specific session of a race weekend. `session_slug` is one of: fp1, fp2, fp3, pre-qualifying, qualifying, post-qualifying, pre-race, race, post-race, highlights, pre-sprint, sprint, post-sprint, pre-sprint-qualifying, sprint-qualifying, post-sprint-qualifying, bonus. """ png = posters_service.render_session_thumb(grand_prix_name, season, session_slug) if not png: raise HTTPException(status_code=404, detail=f"No session thumb for {grand_prix_name} {season} {session_slug}") return FastAPIResponse( content=png, media_type="image/png", headers={"Cache-Control": "public, max-age=31536000"}, ) @app.get("/kometa/formula-1-metadata.yml") async def get_kometa_metadata(): """Generate a Kometa-compatible metadata YAML for the Formula 1 library. One entry per F1 season from 1950..current+1. Each entry: - uses Kometa's f1_season builder for round/episode names - sets url_poster on the show (per-season hero from files-api) - sets url_poster on each Plex 'season' (== each F1 round) from the f1-circuits race poster endpoint Kometa harmlessly skips shows / seasons that don't exist in the Plex library, so we can emit the full schedule and let Kometa do the matching. """ yaml = posters_service.render_kometa_metadata() return PlainTextResponse( content=yaml, media_type="text/yaml; charset=utf-8", # YAML changes only when the Jolpica schedule changes — cache 1h. headers={"Cache-Control": "public, max-age=3600"}, ) @app.get("/circuits") async def get_countries(): """Get list of all available countries with their slugs""" return { "countries": [c.to_dict() for c in circuit_service.get_countries_list()] } @app.get("/circuits/{country_slug}") async def get_cities(country_slug: str): """Get list of cities for a specific country""" cities = circuit_service.get_localities_list(country_slug) if not cities: raise HTTPException(status_code=404, detail="Country not found") return {"cities": [c.to_dict() for c in cities]} @app.get("/circuits/{country_slug}/{city_slug}") async def get_circuits(country_slug: str, city_slug: str): """Get list of circuits for a specific city""" circuits = circuit_service.get_circuits_list(country_slug, city_slug) if not circuits: raise HTTPException(status_code=404, detail="City not found") return {"circuits": [c.to_dict() for c in circuits]} @app.get("/circuits/{country_slug}/{city_slug}/{circuit_slug}") async def get_circuit_details(country_slug: str, city_slug: str, circuit_slug: str): """Get circuit details and available formats""" circuit_details = circuit_service.get_circuit_details(country_slug, city_slug, circuit_slug) if not circuit_details: raise HTTPException(status_code=404, detail="Circuit not found") return circuit_details.to_dict() @app.get("/circuits/{country_slug}/{city_slug}/{circuit_slug}/layout/{layout_slug}") async def get_layout_details(country_slug: str, city_slug: str, circuit_slug: str, layout_slug: str): """Get details for a specific layout""" layout_details = circuit_service.get_layout_details(country_slug, city_slug, circuit_slug, layout_slug) if not layout_details: raise HTTPException(status_code=404, detail="Layout not found") return layout_details.to_dict() @app.get("/tracks/{country_slug}/{city_slug}/{circuit_slug}/year/{year}/{image_format}") async def get_track_by_year( country_slug: str, city_slug: str, circuit_slug: str, year: int, image_format: Format, response: Response ): """Get track layout for a specific year""" circuit_details = circuit_service.get_circuit_details(country_slug, city_slug, circuit_slug) if not circuit_details: raise HTTPException(status_code=404, detail="Circuit not found") try: layout_slug = circuit_service.find_layout_slug_for_year(circuit_details, year) return get_file_response(circuit_service.get_layout_details(country_slug, city_slug, circuit_slug, layout_slug), image_format, response) except KeyError as e: logger.error(f"KeyError in get_track_by_year: {str(e)}") raise HTTPException(status_code=404, detail="Circuit or layout not found") @app.get("/circuits/{country_slug}/{city_slug}/{circuit_slug}/layout/{layout_slug}/{image_format}") async def get_track_by_layout( country_slug: str, city_slug: str, circuit_slug: str, layout_slug: str, image_format: Format, response: Response ): """Get specific track layout""" return get_file_response(circuit_service.get_layout_details(country_slug, city_slug, circuit_slug, layout_slug), image_format, response) @app.get("/grand-prix/{grand_prix_name}/{season}/{image_format}") async def get_grand_prix_circuit(grand_prix_name: str, season: int, image_format: Format, response: Response): layout = circuit_service.get_circuit_layout_by_ergast_data(grand_prix_name, season) if not layout: raise HTTPException(status_code=404, detail="Circuit not found") try: return get_file_response(layout, image_format, response) except KeyError as e: logger.error(f"KeyError in get_track_by_year: {str(e)}") raise HTTPException(status_code=404, detail="Circuit or layout not found") def get_file_response(layout: TrackLayout, image_format: Format, response: Response) -> FileResponse: # layout = circuit_service.get_layout_details(country_slug, city_slug, circuit_slug, layout_slug) file_path = Path(f"./circuits").joinpath(layout.relative_svg_filepath) if not file_path or not file_path.exists(): logger.error(f"File not found: {file_path}") logger.info(f"Looking for {layout.slug} in format {image_format}") raise HTTPException(status_code=404, detail=f"File not found in {image_format} format") # Get actual filename from the path for the Content-Disposition header filename = file_path.name logger.info(f"Serving file: {file_path}") return FileResponse( file_path, media_type=get_media_type(image_format), content_disposition_type=get_content_disposition(image_format), filename=filename, headers={ "Cache-Control": "public, max-age=31536000", "ETag": f"{layout.circuit.locality.country.slug}/{layout.circuit.locality.slug}/{layout.circuit.slug}-{layout.slug}-{image_format}" } ) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)