177 lines
7.0 KiB
Python
177 lines
7.0 KiB
Python
"""Tests for race-plan helpers: `personal_records` and `calibrate_tl_per_km`."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
import pytest
|
||
|
||
from openrun.model import calibrate_tl_per_km, personal_records, plan_to_daily_load
|
||
|
||
|
||
def _act(aid: int, dist_km: float, pace: float, when: str = "2026-01-01") -> dict:
|
||
"""Build a single-activity dict matching `load_activities` output columns."""
|
||
duration_s = pace * 60.0 * dist_km
|
||
return {
|
||
"activity_id": aid,
|
||
"start_time_local": pd.Timestamp(when),
|
||
"distance_km": dist_km,
|
||
"pace_min_per_km": pace,
|
||
"moving_duration_s": duration_s,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# personal_records
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_personal_records_picks_fastest_in_bin() -> None:
|
||
"""Three 10K candidates: returned row must be the one with the lowest pace."""
|
||
acts = pd.DataFrame([
|
||
_act(1, 10.0, pace=5.0),
|
||
_act(2, 10.2, pace=4.5), # winner
|
||
_act(3, 9.8, pace=4.8),
|
||
])
|
||
pr = personal_records(acts, distance_bins_km=(10.0,))
|
||
assert len(pr) == 1
|
||
assert pr.loc[0, "activity_id"] == 2
|
||
assert pr.loc[0, "pace_min_per_km"] == 4.5
|
||
|
||
|
||
def test_personal_records_respects_tolerance_band() -> None:
|
||
"""±5% band on 10K = [9.5, 10.5]. An 11K activity is too far out to qualify."""
|
||
acts = pd.DataFrame([
|
||
_act(1, 10.0, pace=5.0),
|
||
_act(2, 11.0, pace=3.5), # faster but outside tolerance
|
||
])
|
||
pr = personal_records(acts, distance_bins_km=(10.0,))
|
||
assert pr.loc[0, "activity_id"] == 1
|
||
|
||
|
||
def test_personal_records_skips_bins_with_no_qualifying_activity() -> None:
|
||
acts = pd.DataFrame([_act(1, 5.0, pace=5.0)])
|
||
pr = personal_records(acts, distance_bins_km=(5.0, 10.0, 21.0975))
|
||
assert list(pr["bin_km"]) == [5.0]
|
||
|
||
|
||
def test_personal_records_ignores_nan_pace() -> None:
|
||
acts = pd.DataFrame([
|
||
{**_act(1, 10.0, pace=5.0), "pace_min_per_km": np.nan}, # masked by loader
|
||
_act(2, 10.0, pace=5.5),
|
||
])
|
||
pr = personal_records(acts, distance_bins_km=(10.0,))
|
||
assert pr.loc[0, "activity_id"] == 2
|
||
|
||
|
||
def test_personal_records_empty_input_returns_empty_frame() -> None:
|
||
pr = personal_records(pd.DataFrame())
|
||
assert pr.empty
|
||
assert set(pr.columns) >= {"bin_km", "activity_id", "pace_min_per_km"}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# calibrate_tl_per_km
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _insert_activity(conn, aid: int, distance_m: float, training_load: float,
|
||
activity_type: str = "running", when: str = "2026-05-01") -> None:
|
||
conn.execute(
|
||
"""INSERT INTO activities
|
||
(activity_id, start_time_local, activity_type, distance_m, training_load, raw, fetched_at)
|
||
VALUES (?, ?, ?, ?, ?, '{}', 'now')""",
|
||
(aid, when, activity_type, distance_m, training_load),
|
||
)
|
||
|
||
|
||
def test_calibrate_tl_per_km_returns_median_iqr(tmp_conn) -> None:
|
||
"""Median of synthetic tl/km values equals the constructed truth."""
|
||
# 5 runs, each 10 km, with training loads chosen so tl/km = 8, 9, 10, 11, 12.
|
||
for i, tl_per_km in enumerate([8.0, 9.0, 10.0, 11.0, 12.0], start=1):
|
||
_insert_activity(tmp_conn, aid=i, distance_m=10_000, training_load=10 * tl_per_km)
|
||
tmp_conn.commit()
|
||
|
||
result = calibrate_tl_per_km(tmp_conn, lookback_days=None)
|
||
assert result["n"] == 5
|
||
assert result["median"] == 10.0
|
||
assert result["q1"] == 9.0
|
||
assert result["q3"] == 11.0
|
||
|
||
|
||
def test_calibrate_tl_per_km_excludes_short_distance(tmp_conn) -> None:
|
||
_insert_activity(tmp_conn, aid=1, distance_m=10_000, training_load=100)
|
||
_insert_activity(tmp_conn, aid=2, distance_m=500, training_load=50) # too short
|
||
tmp_conn.commit()
|
||
assert calibrate_tl_per_km(tmp_conn, lookback_days=None)["n"] == 1
|
||
|
||
|
||
def test_calibrate_tl_per_km_excludes_other_activity_types(tmp_conn) -> None:
|
||
_insert_activity(tmp_conn, aid=1, distance_m=10_000, training_load=100, activity_type="running")
|
||
_insert_activity(tmp_conn, aid=2, distance_m=10_000, training_load=100, activity_type="cycling")
|
||
tmp_conn.commit()
|
||
assert calibrate_tl_per_km(tmp_conn, lookback_days=None)["n"] == 1
|
||
|
||
|
||
def test_calibrate_tl_per_km_returns_nan_when_empty(tmp_conn) -> None:
|
||
result = calibrate_tl_per_km(tmp_conn, lookback_days=None)
|
||
assert result["n"] == 0
|
||
assert np.isnan(result["median"])
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# plan_to_daily_load
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _plan_row(week_start: str, weekly_km: float, long_run_km: float) -> dict:
|
||
return {"week_start": pd.Timestamp(week_start), "weekly_km": weekly_km, "long_run_km": long_run_km}
|
||
|
||
|
||
def test_plan_distributes_weekly_km_across_7_days() -> None:
|
||
"""A single 70 km / 28 km long-run week should emit 7 daily entries summing to 70·tl_per_km."""
|
||
plan = pd.DataFrame([_plan_row("2026-05-04", 70.0, 28.0)]) # Mon 2026-05-04
|
||
series = plan_to_daily_load(plan, tl_per_km=10.0)
|
||
assert len(series) == 7
|
||
assert series.sum() == pytest.approx(700.0)
|
||
# Saturday (default 40 %) gets the biggest single load.
|
||
saturday = pd.Timestamp("2026-05-09")
|
||
assert series.idxmax() == saturday
|
||
assert series[saturday] == pytest.approx(700.0 * 0.40)
|
||
|
||
|
||
def test_plan_race_week_uses_race_day_tl_per_km() -> None:
|
||
"""If a `race_dates` entry is inside the week, race day gets long_run_km × race_tl,
|
||
and the remaining km distribute across Mon/Wed/Fri at training rate."""
|
||
plan = pd.DataFrame([_plan_row("2026-05-04", 30.0, 20.0)])
|
||
series = plan_to_daily_load(
|
||
plan,
|
||
tl_per_km=11.0,
|
||
race_day_tl_per_km=7.0,
|
||
race_dates=("2026-05-09",), # Saturday
|
||
)
|
||
sat = pd.Timestamp("2026-05-09")
|
||
assert series[sat] == pytest.approx(20.0 * 7.0) # race load
|
||
# Remaining km = 30 - 20 = 10. Spread across Mon/Wed/Fri @ 11.0 → 110 / 3 each.
|
||
for d in [pd.Timestamp("2026-05-04"), pd.Timestamp("2026-05-06"), pd.Timestamp("2026-05-08")]:
|
||
assert series[d] == pytest.approx(110.0 / 3, rel=1e-6)
|
||
|
||
|
||
def test_plan_empty_returns_empty_series() -> None:
|
||
assert plan_to_daily_load(pd.DataFrame(columns=["week_start", "weekly_km", "long_run_km"]),
|
||
tl_per_km=10.0).empty
|
||
|
||
|
||
def test_plan_zero_weekly_km_yields_zeros() -> None:
|
||
"""A rest week (weekly_km=0) shouldn't crash and should produce 7 zero-load days."""
|
||
plan = pd.DataFrame([_plan_row("2026-05-04", 0.0, 0.0)])
|
||
series = plan_to_daily_load(plan, tl_per_km=10.0)
|
||
assert (series == 0).all()
|
||
assert len(series) == 7
|
||
|
||
|
||
def test_plan_custom_weekday_weights() -> None:
|
||
"""Pass a flat 1/7 split, expect equal load each day."""
|
||
plan = pd.DataFrame([_plan_row("2026-05-04", 70.0, 28.0)])
|
||
weights = {i: 1 / 7 for i in range(7)}
|
||
series = plan_to_daily_load(plan, tl_per_km=10.0, weekday_weights=weights)
|
||
assert series.std() == pytest.approx(0.0, abs=1e-9)
|
||
assert series.iloc[0] == pytest.approx(100.0)
|