Add all local circuit work — year-by-year SVGs and tooling
This commit is contained in:
+175
@@ -0,0 +1,175 @@
|
||||
from fastapi import FastAPI, HTTPException, Response
|
||||
from fastapi.responses import FileResponse
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from pathlib import Path
|
||||
import logging
|
||||
|
||||
# Import our CircuitService
|
||||
from circuits_service import CircuitService, Format
|
||||
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()
|
||||
|
||||
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}"
|
||||
]
|
||||
}
|
||||
|
||||
@app.get("/circuits")
|
||||
async def get_countries():
|
||||
"""Get list of all available countries with their slugs"""
|
||||
return {
|
||||
"countries": 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": 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": 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
|
||||
|
||||
@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
|
||||
|
||||
@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)
|
||||
Reference in New Issue
Block a user