personal-website/scripts/generate-race-replay.py

"""Build the compact 2024 São Paulo replay snapshot from public race data.

Run from the repository root: python3 scripts/generate-race-replay.py
The website reads the checked-in snapshot and makes no live API requests.
"""

from __future__ import annotations

import json
import math
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from urllib.parse import urlencode
from urllib.request import Request, urlopen

JOLPICA = "https://api.jolpi.ca/ergast/f1/2024/21"
OPENF1 = "https://api.openf1.org/v1"
OUTPUT = Path(__file__).resolve().parents[1] / "src/lib/race-replay-data.json"


def fetch(url: str):
    for attempt in range(5):
        try:
            with urlopen(Request(url, headers={"User-Agent": "SeanOConnorRaceReplay/1.0"}), timeout=30) as response:
                return json.load(response)
        except Exception:
            if attempt == 4:
                raise
            time.sleep(2 ** attempt)


def seconds(value: str) -> float:
    total = 0.0
    for part in value.split(":"):
        total = total * 60 + float(part)
    return round(total, 3)


def jolpica(endpoint: str):
    payload = fetch(f"{JOLPICA}/{endpoint}/?limit=100")
    return payload["MRData"]


first_page = jolpica("laps")
total = int(first_page["total"])
offsets = range(100, total, 100)
with ThreadPoolExecutor(max_workers=4) as pool:
    more_pages = list(pool.map(lambda offset: fetch(f"{JOLPICA}/laps/?limit=100&offset={offset}")["MRData"], offsets))

all_laps = {}
for page in [first_page, *more_pages]:
    for lap in page["RaceTable"]["Races"][0]["Laps"]:
        all_laps.setdefault(int(lap["number"]), []).extend(lap["Timings"])

results = jolpica("results")["RaceTable"]["Races"][0]
pits = jolpica("pitstops")["RaceTable"]["Races"][0]["PitStops"]
openf1_drivers = fetch(f"{OPENF1}/drivers?session_key=9636")
driver_colors = {int(driver["driver_number"]): f"#{driver['team_colour']}" for driver in openf1_drivers}

drivers = []
for result in results["Results"]:
    driver = result["Driver"]
    number = int(result["number"])
    drivers.append({
        "id": driver["driverId"],
        "number": number,
        "code": driver.get("code", driver["familyName"][:3].upper()),
        "name": f"{driver['givenName']} {driver['familyName']}",
        "team": result["Constructor"]["name"],
        "color": driver_colors.get(number, "#ffffff"),
        "grid": int(result["grid"]),
        "finish": int(result["position"]),
        "status": result["status"],
    })

laps = []
for number, timings in sorted(all_laps.items()):
    laps.append({
        "number": number,
        "timings": [
            {"id": timing["driverId"], "position": int(timing["position"]), "seconds": seconds(timing["time"])}
            for timing in timings
        ],
    })

pit_stops = [
    {"id": pit["driverId"], "lap": int(pit["lap"]), "seconds": seconds(pit["duration"])}
    for pit in pits
]

# A complete lap from Verstappen's on-car coordinates, rather than an
# unrelated circuit image or generated illustration.
lap_eight = next(item for item in fetch(f"{OPENF1}/laps?session_key=9636&driver_number=1") if item["lap_number"] == 8)
from datetime import datetime, timedelta

start = datetime.fromisoformat(lap_eight["date_start"])
end = start + timedelta(seconds=lap_eight["lap_duration"])
params = urlencode({
    "session_key": 9636,
    "driver_number": 1,
    "date>": start.isoformat(timespec="milliseconds"),
    "date<": end.isoformat(timespec="milliseconds"),
})
location = fetch(f"{OPENF1}/location?{params}")
points = [(row["x"], row["y"]) for row in location if row["x"] is not None and row["y"] is not None]
if len(points) < 100:
    raise RuntimeError("Not enough location samples for the circuit trace")
if math.dist(points[0], points[-1]) > 400:
    raise RuntimeError("The sampled track lap does not close")
points.append(points[0])
# Mirror the telemetry Y axis to match the FIA's 2024 circuit map: the
# start/finish straight is on the left and the back straight is on the right.
min_x, max_x = min(x for x, _ in points), max(x for x, _ in points)
min_y, max_y = min(y for _, y in points), max(y for _, y in points)
scale = min(650 / (max_x - min_x), 630 / (max_y - min_y))
offset_x = (1000 - (max_x - min_x) * scale) / 2
offset_y = (740 - (max_y - min_y) * scale) / 2


def map_point(x: float, y: float) -> list[float]:
    return [round(offset_x + (x - min_x) * scale, 1), round(offset_y + (max_y - y) * scale, 1)]


track = [map_point(x, y) for x, y in points]
circuit = fetch("https://api.multiviewer.app/api/v1/circuits/14/2024")
corners = []
for corner in circuit["corners"]:
    x, y = map_point(corner["trackPosition"]["x"], corner["trackPosition"]["y"])
    corners.append({"number": corner["number"], "x": x, "y": y})


def nearest_track_distance(x: float, y: float) -> float:
    """Project a mapped circuit marker onto the same polyline as the cars."""
    best_distance = float("inf")
    best_progress = 0.0
    progress = 0.0
    for (ax, ay), (bx, by) in zip(track, track[1:]):
        length_sq = (bx - ax) ** 2 + (by - ay) ** 2
        fraction = max(0.0, min(1.0, ((x - ax) * (bx - ax) + (y - ay) * (by - ay)) / length_sq)) if length_sq else 0.0
        px, py = ax + fraction * (bx - ax), ay + fraction * (by - ay)
        distance = math.dist((x, y), (px, py))
        if distance < best_distance:
            best_distance = distance
            best_progress = progress + math.sqrt(length_sq) * fraction
        progress += math.sqrt(length_sq)
    return best_progress


track_length = sum(math.dist(a, b) for a, b in zip(track, track[1:]))
corner_distance = {corner["number"]: nearest_track_distance(corner["x"], corner["y"]) for corner in corners}


def point_at(distance: float) -> list[float]:
    remaining = distance % track_length
    for (ax, ay), (bx, by) in zip(track, track[1:]):
        segment = math.dist((ax, ay), (bx, by))
        if remaining <= segment:
            fraction = remaining / segment if segment else 0
            return [round(ax + (bx - ax) * fraction, 1), round(ay + (by - ay) * fraction, 1)]
        remaining -= segment
    return track[0]


def zone_path(start: float, end: float) -> list[list[float]]:
    if end < start:
        end += track_length
    # Follow the same polyline as the animated dots, including the lap seam.
    samples = max(2, math.ceil((end - start) / 3))
    return [point_at(start + (end - start) * index / samples) for index in range(samples + 1)]


# FIA 2024 São Paulo Circuit Map: activation 1 is 30m after T3;
# activation 2 is 160m before T15. Zones end at the next braking turn.
metres_to_track = track_length / 4309
drs_zones = [
    {"number": 1, "track": zone_path(corner_distance[3] + 30 * metres_to_track, corner_distance[4] - 30 * metres_to_track)},
    {"number": 2, "track": zone_path(corner_distance[15] - 160 * metres_to_track, corner_distance[1] - 30 * metres_to_track)},
]

control = fetch(f"{OPENF1}/race_control?session_key=9636")
event_specs = [
    (1, "start", "A race before the race", "An aborted start delays lights out on a wet Interlagos grid.", "ABORTED START"),
    (28, "vsc", "Virtual safety car", "A VSC changes the pit-stop calculation as the rain intensifies.", "VIRTUAL SAFETY CAR DEPLOYED"),
    (30, "safety", "Safety car deployed", "The field bunches up in difficult conditions.", "SAFETY CAR DEPLOYED"),
    (32, "red", "Red flag", "The race stops, and the order is reset for a rolling restart.", "RED FLAG"),
    (33, "restart", "Rolling restart", "The field gets going again behind Ocon.", "ROLLING START"),
    (39, "safety", "Safety car returns", "Another interruption bunches the field.", "SAFETY CAR DEPLOYED"),
    (42, "restart", "Safety car in", "The race returns to green-flag running.", "SAFETY CAR IN THIS LAP"),
    (69, "finish", "Chequered flag", "Verstappen wins from P17; Alpine finishes second and third.", "CHEQUERED FLAG"),
]
events = []
for lap, kind, title, description, message in event_specs:
    if not any(item.get("lap_number") == lap and message in item["message"] for item in control):
        raise RuntimeError(f"Race-control event not found: lap {lap}, {message}")
    events.append({"lap": lap, "kind": kind, "title": title, "description": description})

data = {
    "race": {"title": "São Paulo Grand Prix", "year": 2024, "date": "3 November 2024", "circuit": "Autódromo José Carlos Pace", "location": "Interlagos · Brazil", "laps": 69},
    "drivers": drivers,
    "laps": laps,
    "pitStops": pit_stops,
    "events": events,
    "track": track,
    "corners": corners,
    "drsZones": drs_zones,
    "sources": {
        "laps": f"{JOLPICA}/laps/",
        "results": f"{JOLPICA}/results/",
        "pitStops": f"{JOLPICA}/pitstops/",
        "raceControl": f"{OPENF1}/race_control?session_key=9636",
        "track": f"{OPENF1}/location?session_key=9636&driver_number=1",
        "corners": "https://api.multiviewer.app/api/v1/circuits/14/2024",
        "circuitMap": "https://www.fia.com/sites/default/files/decision-document/2024%20S%C3%A3o%20Paulo%20Grand%20Prix%20-%20Event%20Notes%20-%20Circuit%20Map,%20Pit%20Lane%20Drawing%20and%20Red%20Zones.pdf",
    },
}
OUTPUT.write_text(json.dumps(data, ensure_ascii=False, separators=(",", ":")) + "\n")
print(f"Wrote {OUTPUT}: {len(drivers)} drivers, {len(laps)} laps, {len(track)} track points")