Spaces:

Arry7868
/

F1-Simulator-for-ML-Simulations

Running

App Files Files Community

F1-Simulator-for-ML-Simulations / tests /test_fetch_data.py

Arry7868

Claude Opus 5.5

Add tests, CI with auto-deploy to the Space, and a weekly data refresh

dec0e0c 25 days ago

Raw History Blame Contribute Delete

7.42 kB

"""Processing in fetch_data.py: no network, synthetic inputs with known answers."""
from datetime import datetime, timedelta, timezone

import numpy as np
import pytest

import fetch_data as fd

T0 = datetime(2025, 3, 1, 15, 0, tzinfo=timezone.utc)
CIRCUIT = "Testring"


def iso(seconds: float) -> str:
    return (T0 + timedelta(seconds=seconds)).isoformat()


# ── names and colours ───────────────────────────────────────────────────────

def test_race_filename_is_a_plain_name():
    assert fd.race_filename(2024, "São Paulo Grand Prix", "R") == "2024_são_paulo_grand_prix_r.json"
    # Separators from a crafted name must never reach the file system
    for name in ("../../etc/passwd", "..\\..\\serve", "a/b\\c"):
        f = fd.race_filename(2024, name, "R")
        assert "/" not in f and "\\" not in f


@pytest.mark.parametrize("raw, expected", [
    ("3671c6", "#3671c6"), ("#ff8000", "#ff8000"), (None, "#ffffff"), ("nan", "#ffffff"), ("", "#ffffff"),
])
def test_hex_color(raw, expected):
    assert fd.hex_color(raw) == expected


# ── track geometry: race distance comes from snapping onto the outline ──────

def circle(radius=1000.0, n=2000):
    a = np.linspace(0, 2 * np.pi, n, endpoint=False)
    return np.c_[radius * np.cos(a), radius * np.sin(a)]


def test_snap_gives_distance_along_the_lap():
    loop = fd._Loop(circle())
    assert loop.length == pytest.approx(2 * np.pi * 1000, rel=1e-3)
    # A quarter of the way round, 5 units off the line
    s, off = loop.snap(np.array([0.0]), np.array([1005.0]))
    assert s[0] == pytest.approx(loop.length / 4, rel=2e-3)
    assert off[0] == pytest.approx(5.0, abs=1.0)


def test_unwrap_counts_laps():
    loop = fd._Loop(circle())
    a = np.linspace(0, 3 * 2 * np.pi, 600)           # three laps, sampled densely
    s, _ = loop.snap(1000 * np.cos(a), 1000 * np.sin(a))
    cum = fd._unwrap(s, loop.length)
    assert cum[-1] - cum[0] == pytest.approx(3 * loop.length, rel=0.01)
    assert np.all(np.diff(cum) > -1)                  # never jumps back a lap


# ── race control -> FastF1-style track status codes ─────────────────────────

def test_track_status_codes():
    rc = [
        {"date": iso(10), "category": "SafetyCar", "message": "SAFETY CAR DEPLOYED"},
        {"date": iso(20), "category": "SafetyCar", "message": "SAFETY CAR IN THIS LAP"},
        {"date": iso(30), "category": "Flag", "flag": "CLEAR", "scope": "Track", "message": "TRACK CLEAR"},
        {"date": iso(40), "category": "SafetyCar", "message": "VIRTUAL SAFETY CAR DEPLOYED"},
        {"date": iso(50), "category": "SafetyCar", "message": "VIRTUAL SAFETY CAR ENDING"},
        {"date": iso(60), "category": "Flag", "flag": "CLEAR", "scope": "Sector", "message": "CLEAR IN TRACK SECTOR 4"},
        {"date": iso(70), "category": "Flag", "flag": "RED", "scope": "Track", "message": "RED FLAG"},
        {"date": iso(80), "category": "Flag", "flag": "DOUBLE YELLOW", "scope": "Track", "message": "DOUBLE YELLOW"},
    ]
    status = fd._track_status(rc, T0.timestamp())
    assert [(t, code) for t, code, _ in status] == [
        (10, "4"), (30, "1"), (40, "6"), (50, "1"), (70, "5"), (80, "2")]


# ── lap windows, including a retirement mid-lap ─────────────────────────────

def test_lap_windows_end_a_retirement_when_the_car_stops():
    laps = [{"lap_number": 1, "date_start": iso(0), "lap_duration": 90.0},
            {"lap_number": 2, "date_start": iso(90), "lap_duration": None}]
    t = T0.timestamp() + np.arange(0, 200, 1.0)
    x = np.where(t - T0.timestamp() < 130, (t - T0.timestamp()) * 50, 130 * 50)   # stops at 130 s
    y = np.zeros_like(x)
    w = fd._lap_windows(laps, t, x, y)
    assert [n for n, _, _ in w] == [1, 2]
    assert w[0][2] - T0.timestamp() == pytest.approx(90)
    assert w[1][2] - T0.timestamp() == pytest.approx(130, abs=1.5)


# ── pit stops: OpenF1's pit time is the exit, not the entry ─────────────────

def test_pit_window_ends_at_the_pit_time():
    stops = [{"driver_number": 1, "date": iso(1000), "lane_duration": 24.2},
             {"driver_number": 1, "date": iso(3000), "pit_duration": 30.0},   # older field name
             {"driver_number": 1, "date": iso(5000)},                         # no duration: 25 s
             {"driver_number": 2, "date": iso(1500), "lane_duration": 20.0}]
    assert fd.pit_windows(stops, 1, T0.timestamp()) == [[975.8, 1000.0], [2970.0, 3000.0], [4975.0, 5000.0]]


# ── pit lane traced from a real stop ────────────────────────────────────────

def test_pit_lane_is_traced_during_the_stop_window():
    t = T0.timestamp() + np.arange(0, 100, 0.25)
    raw = {44: (t, (t - T0.timestamp()) * 10, np.zeros_like(t))}
    stops = [{"driver_number": 44, "date": iso(60), "lane_duration": 20.0}]   # pit exit at 60 s
    lane = fd._pit_lane_from_stops(stops, raw)
    xs = [p[0] for p in lane]
    assert min(xs) >= 400 - 1e-6 and max(xs) <= 600 + 1e-6   # x = 10 * t for t in [40, 60]
    assert len(lane) >= 70
    assert fd._max_jump(lane) < 5


def test_no_pit_lane_without_timed_stops():
    raw = {1: (np.arange(10.0), np.arange(10.0), np.arange(10.0))}
    assert fd._pit_lane_from_stops([{"driver_number": 1, "date": iso(5)}], raw) is None


# ── per-lap table used by the strategy model ────────────────────────────────

def test_lap_rows():
    session = {"session_key": 7, "year": 2025, "date_start": iso(0), "circuit_short_name": CIRCUIT}
    laps = [{"driver_number": 1, "lap_number": n, "date_start": iso(90 * (n - 1)), "lap_duration": 90.0,
             "is_pit_out_lap": n == 3} for n in (1, 2, 3)]
    stints = [{"driver_number": 1, "stint_number": 1, "lap_start": 1, "lap_end": 2, "compound": "SOFT", "tyre_age_at_start": 3},
              {"driver_number": 1, "stint_number": 2, "lap_start": 3, "lap_end": 3, "compound": "HARD", "tyre_age_at_start": 0}]
    pits = [{"driver_number": 1, "lap_number": 2, "date": iso(185), "lane_duration": 22.0}]
    weather = [{"date": iso(0), "track_temperature": 31.0, "air_temperature": 22.0, "rainfall": 0},
               {"date": iso(150), "track_temperature": 33.0, "air_temperature": 22.0, "rainfall": 0}]
    rc = [{"date": iso(200), "category": "SafetyCar", "message": "SAFETY CAR DEPLOYED"},
          {"date": iso(400), "category": "Flag", "flag": "CLEAR", "scope": "Track", "message": "TRACK CLEAR"}]
    rows = fd.lap_rows(session, laps, stints, pits, weather, rc, {1: {"team_name": "Team"}})
    by = {r["lap"]: r for r in rows}
    assert [by[n]["compound"] for n in (1, 2, 3)] == ["SOFT", "SOFT", "HARD"]
    assert [by[n]["tyre_age"] for n in (1, 2, 3)] == [3, 4, 0]
    assert [by[n]["pit_in_lap"] for n in (1, 2, 3)] == [0, 1, 0]
    assert [by[n]["pit_out_lap"] for n in (1, 2, 3)] == [0, 0, 1]
    assert [by[n]["neutralised"] for n in (1, 2, 3)] == [0, 0, 1]      # SC from 200 s, lap 3 = 180-270 s
    assert [by[n]["track_temp"] for n in (1, 2, 3)] == [31.0, 31.0, 33.0]   # weather at each lap start
    assert by[2]["start"] == 90 and by[1]["circuit"] == CIRCUIT

	"""Processing in fetch_data.py: no network, synthetic inputs with known answers."""
	from datetime import datetime, timedelta, timezone

	import numpy as np
	import pytest

	import fetch_data as fd

	T0 = datetime(2025, 3, 1, 15, 0, tzinfo=timezone.utc)
	CIRCUIT = "Testring"


	def iso(seconds: float) -> str:
	return (T0 + timedelta(seconds=seconds)).isoformat()


	# ── names and colours ───────────────────────────────────────────────────────

	def test_race_filename_is_a_plain_name():
	assert fd.race_filename(2024, "São Paulo Grand Prix", "R") == "2024_são_paulo_grand_prix_r.json"
	# Separators from a crafted name must never reach the file system
	for name in ("../../etc/passwd", "..\\..\\serve", "a/b\\c"):
	f = fd.race_filename(2024, name, "R")
	assert "/" not in f and "\\" not in f


	@pytest.mark.parametrize("raw, expected", [
	("3671c6", "#3671c6"), ("#ff8000", "#ff8000"), (None, "#ffffff"), ("nan", "#ffffff"), ("", "#ffffff"),
	])
	def test_hex_color(raw, expected):
	assert fd.hex_color(raw) == expected


	# ── track geometry: race distance comes from snapping onto the outline ──────

	def circle(radius=1000.0, n=2000):
	a = np.linspace(0, 2 * np.pi, n, endpoint=False)
	return np.c_[radius * np.cos(a), radius * np.sin(a)]


	def test_snap_gives_distance_along_the_lap():
	loop = fd._Loop(circle())
	assert loop.length == pytest.approx(2 * np.pi * 1000, rel=1e-3)
	# A quarter of the way round, 5 units off the line
	s, off = loop.snap(np.array([0.0]), np.array([1005.0]))
	assert s[0] == pytest.approx(loop.length / 4, rel=2e-3)
	assert off[0] == pytest.approx(5.0, abs=1.0)


	def test_unwrap_counts_laps():
	loop = fd._Loop(circle())
	a = np.linspace(0, 3 * 2 * np.pi, 600) # three laps, sampled densely
	s, _ = loop.snap(1000 * np.cos(a), 1000 * np.sin(a))
	cum = fd._unwrap(s, loop.length)
	assert cum[-1] - cum[0] == pytest.approx(3 * loop.length, rel=0.01)
	assert np.all(np.diff(cum) > -1) # never jumps back a lap


	# ── race control -> FastF1-style track status codes ─────────────────────────

	def test_track_status_codes():
	rc = [
	{"date": iso(10), "category": "SafetyCar", "message": "SAFETY CAR DEPLOYED"},
	{"date": iso(20), "category": "SafetyCar", "message": "SAFETY CAR IN THIS LAP"},
	{"date": iso(30), "category": "Flag", "flag": "CLEAR", "scope": "Track", "message": "TRACK CLEAR"},
	{"date": iso(40), "category": "SafetyCar", "message": "VIRTUAL SAFETY CAR DEPLOYED"},
	{"date": iso(50), "category": "SafetyCar", "message": "VIRTUAL SAFETY CAR ENDING"},
	{"date": iso(60), "category": "Flag", "flag": "CLEAR", "scope": "Sector", "message": "CLEAR IN TRACK SECTOR 4"},
	{"date": iso(70), "category": "Flag", "flag": "RED", "scope": "Track", "message": "RED FLAG"},
	{"date": iso(80), "category": "Flag", "flag": "DOUBLE YELLOW", "scope": "Track", "message": "DOUBLE YELLOW"},
	]
	status = fd._track_status(rc, T0.timestamp())
	assert [(t, code) for t, code, _ in status] == [
	(10, "4"), (30, "1"), (40, "6"), (50, "1"), (70, "5"), (80, "2")]


	# ── lap windows, including a retirement mid-lap ─────────────────────────────

	def test_lap_windows_end_a_retirement_when_the_car_stops():
	laps = [{"lap_number": 1, "date_start": iso(0), "lap_duration": 90.0},
	{"lap_number": 2, "date_start": iso(90), "lap_duration": None}]
	t = T0.timestamp() + np.arange(0, 200, 1.0)
	x = np.where(t - T0.timestamp() < 130, (t - T0.timestamp()) * 50, 130 * 50) # stops at 130 s
	y = np.zeros_like(x)
	w = fd._lap_windows(laps, t, x, y)
	assert [n for n, _, _ in w] == [1, 2]
	assert w[0][2] - T0.timestamp() == pytest.approx(90)
	assert w[1][2] - T0.timestamp() == pytest.approx(130, abs=1.5)


	# ── pit stops: OpenF1's pit time is the exit, not the entry ─────────────────

	def test_pit_window_ends_at_the_pit_time():
	stops = [{"driver_number": 1, "date": iso(1000), "lane_duration": 24.2},
	{"driver_number": 1, "date": iso(3000), "pit_duration": 30.0}, # older field name
	{"driver_number": 1, "date": iso(5000)}, # no duration: 25 s
	{"driver_number": 2, "date": iso(1500), "lane_duration": 20.0}]
	assert fd.pit_windows(stops, 1, T0.timestamp()) == [[975.8, 1000.0], [2970.0, 3000.0], [4975.0, 5000.0]]


	# ── pit lane traced from a real stop ────────────────────────────────────────

	def test_pit_lane_is_traced_during_the_stop_window():
	t = T0.timestamp() + np.arange(0, 100, 0.25)
	raw = {44: (t, (t - T0.timestamp()) * 10, np.zeros_like(t))}
	stops = [{"driver_number": 44, "date": iso(60), "lane_duration": 20.0}] # pit exit at 60 s
	lane = fd._pit_lane_from_stops(stops, raw)
	xs = [p[0] for p in lane]
	assert min(xs) >= 400 - 1e-6 and max(xs) <= 600 + 1e-6 # x = 10 * t for t in [40, 60]
	assert len(lane) >= 70
	assert fd._max_jump(lane) < 5


	def test_no_pit_lane_without_timed_stops():
	raw = {1: (np.arange(10.0), np.arange(10.0), np.arange(10.0))}
	assert fd._pit_lane_from_stops([{"driver_number": 1, "date": iso(5)}], raw) is None


	# ── per-lap table used by the strategy model ────────────────────────────────

	def test_lap_rows():
	session = {"session_key": 7, "year": 2025, "date_start": iso(0), "circuit_short_name": CIRCUIT}
	laps = [{"driver_number": 1, "lap_number": n, "date_start": iso(90 * (n - 1)), "lap_duration": 90.0,
	"is_pit_out_lap": n == 3} for n in (1, 2, 3)]
	stints = [{"driver_number": 1, "stint_number": 1, "lap_start": 1, "lap_end": 2, "compound": "SOFT", "tyre_age_at_start": 3},
	{"driver_number": 1, "stint_number": 2, "lap_start": 3, "lap_end": 3, "compound": "HARD", "tyre_age_at_start": 0}]
	pits = [{"driver_number": 1, "lap_number": 2, "date": iso(185), "lane_duration": 22.0}]
	weather = [{"date": iso(0), "track_temperature": 31.0, "air_temperature": 22.0, "rainfall": 0},
	{"date": iso(150), "track_temperature": 33.0, "air_temperature": 22.0, "rainfall": 0}]
	rc = [{"date": iso(200), "category": "SafetyCar", "message": "SAFETY CAR DEPLOYED"},
	{"date": iso(400), "category": "Flag", "flag": "CLEAR", "scope": "Track", "message": "TRACK CLEAR"}]
	rows = fd.lap_rows(session, laps, stints, pits, weather, rc, {1: {"team_name": "Team"}})
	by = {r["lap"]: r for r in rows}
	assert [by[n]["compound"] for n in (1, 2, 3)] == ["SOFT", "SOFT", "HARD"]
	assert [by[n]["tyre_age"] for n in (1, 2, 3)] == [3, 4, 0]
	assert [by[n]["pit_in_lap"] for n in (1, 2, 3)] == [0, 1, 0]
	assert [by[n]["pit_out_lap"] for n in (1, 2, 3)] == [0, 0, 1]
	assert [by[n]["neutralised"] for n in (1, 2, 3)] == [0, 0, 1] # SC from 200 s, lap 3 = 180-270 s
	assert [by[n]["track_temp"] for n in (1, 2, 3)] == [31.0, 31.0, 33.0] # weather at each lap start
	assert by[2]["start"] == 90 and by[1]["circuit"] == CIRCUIT