Buckets:
| """tests/test_alert_timing.py — unit tests of eval/alert_timing_eval.py (plan R2 WP2.3 / WP7): the event matcher on synthetic sequences, the automatic | |
| warning-condition event definition, the alert cooldown semantics, the threshold bisection, and the LEAKAGE test that no non-oracle score at t reads | |
| anchor rows after t (only the declared oracle quantity — the event recovery time — may). | |
| python -m pytest -q tests/test_alert_timing.py | |
| """ | |
| import sys | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(CWM3 / "eval")) | |
| import alert_timing_eval as AT # noqa: E402 | |
| def _grid(T=100.0, dt=0.2): | |
| t = np.round(np.arange(0.0, T, dt), 1) | |
| n = len(t) | |
| return dict(uid=np.array(["r1"] * n), dongle=np.array(["d1"] * n), t=t, hands_on=np.ones(n, np.float32), head=np.zeros(n, np.float32), gaze=np.zeros(n, np.float32), | |
| D=np.zeros(n, np.float32), moving=np.ones(n, np.float32)) | |
| def _derive(g): | |
| seg = AT.segments(g["uid"], g["t"]) | |
| U = AT.ffill_segment(AT.unmet_series(g["hands_on"], g["head"], g["gaze"]), seg) | |
| U = np.where(np.isfinite(U), U, 0.0) | |
| dwell = AT.dwell_series(U, seg, g["t"]) | |
| WC = AT.warning_condition(U, g["D"], g["moving"], dwell) | |
| hands_off = np.isfinite(g["hands_on"]) & (g["hands_on"] < 0.5) | |
| eyes_off = (np.isfinite(g["head"]) & (g["head"] > 0.5)) | (np.isfinite(g["gaze"]) & (g["gaze"] > 0.5)) | |
| ev, n_cens = AT.events_from_wc(WC, U, g["D"], seg, g["t"], g["uid"], g["dongle"], hands_off, eyes_off) | |
| return seg, U, dwell, WC, ev, n_cens | |
| def _synthetic(): | |
| g = _grid() | |
| t = g["t"] | |
| g["hands_on"][(t >= 20.0) & (t < 30.0)] = 0.0 # hands off 20..30 | |
| g["D"][(t >= 22.0) & (t < 24.0)] = 1.0 # demand 22..24 -> overlap onset at 22.0 | |
| g["D"][(t >= 40.0) & (t < 45.0)] = 1.0 # demand alone (no unmet state) -> no event | |
| g["head"][(t >= 60.0) & (t < 70.0)] = 1.0 # head turned 60..70, no demand -> dwell onset at 63.0 | |
| g["hands_on"][(t >= 80.0) & (t < 82.0)] = 0.0 # 2 s hands off, no demand -> no event | |
| return g | |
| def test_event_definition_synthetic(): | |
| g = _synthetic() | |
| seg, U, dwell, WC, ev, n_cens = _derive(g) | |
| assert len(seg) == len(g["t"]) and seg.max() == 0 | |
| assert np.isclose(dwell[np.searchsorted(g["t"], 23.0)], 3.0) | |
| assert len(ev) == 2, ev | |
| assert np.allclose(ev.onset_t.to_numpy(), [22.0, 63.0]), ev.onset_t.tolist() | |
| assert list(ev.trigger) == ["overlap", "dwell"] | |
| assert bool(ev.hands_off.iloc[0]) and not bool(ev.eyes_off.iloc[0]) and bool(ev.eyes_off.iloc[1]) | |
| assert np.allclose(ev.t_recovery.to_numpy(), [30.0, 70.0]) | |
| assert n_cens == 0 | |
| def test_event_merge_and_censoring(): | |
| g = _grid(T=60.0) | |
| t = g["t"] | |
| g["hands_on"][t < 10.0] = 0.0 # segment starts inside WC (hands off + demand at t = 0) -> censored, not an event | |
| g["D"][t < 10.0] = 1.0 # demand for the whole hands-off spell, so WC never drops and re-enters | |
| g["hands_on"][(t >= 20.0) & (t < 40.0)] = 0.0 # long hands-off: dwell onset at 23.0; demand at 30 would re-trigger WC only if WC dropped — it does not | |
| g["D"][(t >= 30.0) & (t < 31.0)] = 1.0 | |
| seg, U, dwell, WC, ev, n_cens = _derive(g) | |
| assert n_cens == 1 | |
| assert len(ev) == 1 and np.isclose(ev.onset_t.iloc[0], 23.0) | |
| # two WC onsets 6 s apart on the same route merge into one event (10 s guard) | |
| g = _grid(T=60.0) | |
| t = g["t"] | |
| g["hands_on"][(t >= 10.0) & (t < 12.0)] = 0.0 | |
| g["D"][(t >= 10.0) & (t < 12.0)] = 1.0 | |
| g["hands_on"][(t >= 16.0) & (t < 18.0)] = 0.0 | |
| g["D"][(t >= 16.0) & (t < 18.0)] = 1.0 | |
| g["hands_on"][(t >= 30.0) & (t < 32.0)] = 0.0 | |
| g["D"][(t >= 30.0) & (t < 32.0)] = 1.0 | |
| seg, U, dwell, WC, ev, n_cens = _derive(g) | |
| assert np.allclose(ev.onset_t.to_numpy(), [10.0, 30.0]), ev.onset_t.tolist() | |
| def test_matcher_synthetic_alerts(): | |
| g = _synthetic() | |
| seg, U, dwell, WC, ev, _ = _derive(g) | |
| t = g["t"] | |
| s = np.zeros(len(t), np.float32) | |
| for ta in (5.0, 21.0, 25.0, 64.4): # 5.0 = false alarm; 21.0 = 1.0 s early for event 22; 25.0 = inside event 1 (justified, but < 60 s after 21 -> suppressed by cooldown); | |
| s[np.searchsorted(t, ta)] = 1.0 # 64.4 = 1.4 s late for event 63 | |
| ar = AT.alert_rows(g["uid"], t, s, 0.5) | |
| # with the 60 s cooldown: 5.0 fires, 21.0 is < 60 s after 5.0 -> suppressed, 25.0 suppressed, 64.4 fires (59.4 s after 5.0? no: 64.4 - 5.0 = 59.4 < 60 -> suppressed) -> only 5.0 | |
| assert np.allclose(t[ar], [5.0]), t[ar] | |
| # without cooldown every alert fires; matcher semantics | |
| ar = AT.alert_rows(g["uid"], t, s, 0.5, cooldown=0.0) | |
| assert np.allclose(t[ar], [5.0, 21.0, 25.0, 64.4]) | |
| lead, justified = AT.match_events(ev, g["uid"], t, ar) | |
| assert np.allclose(lead, [1.0, -1.4], atol=1e-6), lead | |
| assert justified.tolist() == [False, True, True, True] # 5.0 is a false alarm; 25.0 lies inside [22 - 10, max(25, recovery 30)] | |
| for d, exp in ((0.5, 0.5), (1.0, 0.5), (2.0, 0.0)): | |
| assert np.isclose(np.mean(np.nan_to_num(lead, nan=-99) >= d), exp) | |
| # an alert 12 s before the onset does not count (window starts at o - 10) | |
| s2 = np.zeros(len(t), np.float32) | |
| s2[np.searchsorted(t, 9.8)] = 1.0 | |
| lead2, _ = AT.match_events(ev, g["uid"], t, AT.alert_rows(g["uid"], t, s2, 0.5, cooldown=0.0)) | |
| assert np.isnan(lead2).all() | |
| def test_cooldown_semantics_match_trainer(): | |
| """alert_rows must agree with train_warning_head.alerts_per_hour on random scores over several routes.""" | |
| sys.path.insert(0, str(CWM3 / "train")) | |
| import train_warning_head as TW | |
| rng = np.random.default_rng(0) | |
| uid = np.repeat(np.array(["a", "b", "c"]), 400) | |
| t = np.concatenate([np.arange(400) * 0.2] * 3) | |
| s = rng.random(1200).astype(np.float32) | |
| s[rng.random(1200) < 0.1] = np.nan | |
| for thr in (0.5, 0.9, 0.99): | |
| n_ref = TW.alerts_per_hour(uid, t, s, thr)[0] | |
| assert len(AT.alert_rows(uid, t, s, thr)) == n_ref | |
| def test_threshold_bisection_reaches_budget_or_floor(): | |
| rng = np.random.default_rng(1) | |
| uid = np.repeat(np.array([f"r{i}" for i in range(5)]), 3600) | |
| t = np.concatenate([np.arange(3600) * 0.2] * 5) # 5 routes x 12 min = 1 h | |
| s = rng.random(len(t)).astype(np.float32) | |
| thr, ach, reach, floor = AT.thr_for_rate(uid, t, s, 10.0) | |
| assert reach and ach <= 10.0 + 1e-9 and ach > 8.0, (thr, ach, floor) | |
| binary = (s > 0.99).astype(np.float32) # saturating score: floor above a 1 / h budget | |
| thr_b, ach_b, reach_b, floor_b = AT.thr_for_rate(uid, t, binary, 1.0) | |
| assert not reach_b and thr_b == 1.0 and ach_b == floor_b and floor_b > 1.0 | |
| def test_no_future_leakage_in_rule_scores_and_events(): | |
| """perturb every array strictly after row i: scores, U, dwell and WC at rows <= i are unchanged; only the recovery time (declared oracle) may change.""" | |
| rng = np.random.default_rng(2) | |
| g = _grid(T=200.0) | |
| n = len(g["t"]) | |
| for arr in ("hands_on", "head", "gaze", "D"): | |
| g[arr] = (rng.random(n) < 0.15).astype(np.float32) if arr != "hands_on" else (rng.random(n) > 0.15).astype(np.float32) | |
| g[arr][rng.random(n) < 0.05] = np.nan | |
| seg, U, dwell, WC, ev, _ = _derive(g) | |
| sp = AT.score_persistence(U, g["D"], g["moving"], dwell) | |
| sd = AT.score_dwell_timer(U, g["D"], g["moving"], dwell) | |
| i = n // 2 | |
| g2 = {k: (v.copy() if isinstance(v, np.ndarray) else v) for k, v in g.items()} | |
| for arr in ("hands_on", "head", "gaze", "D", "moving"): | |
| g2[arr][i + 1:] = rng.random(n - i - 1) > 0.5 | |
| seg2, U2, dwell2, WC2, ev2, _ = _derive(g2) | |
| sp2 = AT.score_persistence(U2, g2["D"], g2["moving"], dwell2) | |
| sd2 = AT.score_dwell_timer(U2, g2["D"], g2["moving"], dwell2) | |
| assert np.array_equal(np.nan_to_num(sp[:i + 1], nan=-1), np.nan_to_num(sp2[:i + 1], nan=-1)) | |
| assert np.array_equal(np.nan_to_num(sd[:i + 1], nan=-1), np.nan_to_num(sd2[:i + 1], nan=-1)) | |
| assert np.array_equal(U[:i + 1], U2[:i + 1]) and np.allclose(dwell[:i + 1], dwell2[:i + 1]) and np.array_equal(WC[:i + 1], WC2[:i + 1]) | |
| o1 = ev[ev.onset_t <= g["t"][i]].onset_t.to_numpy() | |
| o2 = ev2[ev2.onset_t <= g["t"][i]].onset_t.to_numpy() | |
| assert np.array_equal(o1, o2) # onsets up to t_i identical; recovery of the last one may differ (declared t+ quantity) | |
| # the rule scores take only (U, D, moving, dwell): no future array is even passed | |
| import inspect | |
| for fn in (AT.score_persistence, AT.score_dwell_timer): | |
| assert [p for p in inspect.signature(fn).parameters][:4] == ["U", "D", "moving", "dwell"] | |
| def test_unmet_and_ffill(): | |
| hands = np.array([1, 0, np.nan, np.nan, 1, np.nan], np.float32) | |
| head = np.array([0, 0, np.nan, 1, 0, np.nan], np.float32) | |
| gaze = np.array([np.nan] * 6, np.float32) | |
| U = AT.unmet_series(hands, head, gaze) | |
| assert np.array_equal(np.nan_to_num(U, nan=-1), [0, 1, -1, 1, 0, -1]) | |
| seg = np.zeros(6, np.int64) | |
| Uf = AT.ffill_segment(U, seg, max_steps=1) | |
| assert np.array_equal(np.nan_to_num(Uf, nan=-1), [0, 1, 1, 1, 0, 0]) | |
| Uf0 = AT.ffill_segment(U, seg, max_steps=0) | |
| assert np.isnan(Uf0[2]) and np.isnan(Uf0[5]) | |
| def test_sustained_onsets_and_lead_margin(): | |
| """family C: onset = the anchor at which the unbroken run first reaches min_s; runs starting at a segment start are censored; lead margin = onset - U entry.""" | |
| g = _grid(T=120.0) | |
| t = g["t"] | |
| g["hands_on"][t < 6.0] = 0.0 # run from the segment start -> censored (entry unobserved) | |
| g["hands_on"][(t >= 20.0) & (t < 30.0)] = 0.0 # 10 s run -> C1 onset at 23.0 (dwell reaches 3.0), lead margin 3.0 | |
| g["head"][(t >= 50.0) & (t < 52.0)] = 1.0 # 2 s run -> no C1 event | |
| g["hands_on"][(t >= 70.0) & (t < 80.0)] = 0.0 # U from 70; hard demand 74..78 -> C2 onset when the overlap lasts 1.0 s = 75.0; lead margin 5.0 | |
| D_hard = np.zeros(len(t), np.float32) | |
| D_hard[(t >= 74.0) & (t < 78.0)] = 1.0 | |
| seg, U, dwell, WC, ev_A, _ = _derive(g) | |
| U_mov = (g["moving"] > 0.5) & (U > 0.5) | |
| onset, n_c = AT.sustained_onsets(U_mov, seg, t, 3.0) | |
| assert n_c == 1, n_c | |
| assert np.allclose(t[onset], [23.0, 73.0]), t[onset] | |
| hands_off = np.isfinite(g["hands_on"]) & (g["hands_on"] < 0.5) | |
| eyes_off = np.isfinite(g["head"]) & (g["head"] > 0.5) | |
| ev, n_cens = AT.events_from_wc(U_mov, U, g["D"], seg, t, g["uid"], g["dongle"], hands_off, eyes_off, onset=onset, n_censored=n_c) | |
| assert n_cens == 1 and np.allclose(ev.onset_t, [23.0, 73.0]) and np.allclose(ev.lead_margin_s, [3.0, 3.0]) and np.allclose(ev.t_recovery, [30.0, 80.0]) | |
| onset2, n_c2 = AT.sustained_onsets(U_mov & (D_hard > 0.5), seg, t, 1.0) | |
| assert n_c2 == 0 and np.allclose(t[onset2], [75.0]), t[onset2] | |
| ev2, _ = AT.events_from_wc(U_mov & (D_hard > 0.5), U, D_hard, seg, t, g["uid"], g["dongle"], hands_off, eyes_off, onset=onset2, n_censored=n_c2) | |
| assert np.allclose(ev2.lead_margin_s, [5.0]) and list(ev2.trigger) == ["overlap"] | |
| # an alert at 72.0 is 1.0 s early for the C1 event at 73.0 and 3.0 s early for the C2 event at 75.0 | |
| s = np.zeros(len(t), np.float32) | |
| s[np.searchsorted(t, 72.0)] = 1.0 | |
| ar = AT.alert_rows(g["uid"], t, s, 0.5) | |
| lead1, _ = AT.match_events(ev, g["uid"], t, ar) | |
| lead2, _ = AT.match_events(ev2, g["uid"], t, ar) | |
| assert np.allclose(np.nan_to_num(lead1, nan=-99), [-99, 1.0]) and np.allclose(lead2, [3.0]) | |
| # future leakage: perturbing rows after the C1 onset leaves the onset mask up to that row unchanged | |
| g2 = {k: (v.copy() if isinstance(v, np.ndarray) else v) for k, v in g.items()} | |
| i = np.searchsorted(t, 23.0) | |
| g2["hands_on"][i + 1:] = 1.0 | |
| seg2, U2, _, _, _, _ = _derive(g2) | |
| onset3, _ = AT.sustained_onsets((g2["moving"] > 0.5) & (U2 > 0.5), seg2, t, 3.0) | |
| assert np.array_equal(onset[:i + 1], onset3[:i + 1]) | |
| def test_family_d_onsets_and_road_rules(): | |
| """family D: onset = hard-demand onset with U active; D2 needs an observed U entry <= 3 s before; lead margin = onset - U entry; road rules read t only.""" | |
| g = _grid(T=120.0) | |
| t = g["t"] | |
| g["hands_on"][t < 5.0] = 0.0 # U from the segment start; demand onset at 3.0 -> D1 kept (entry censored), D2 excluded | |
| g["hands_on"][(t >= 20.0) & (t < 30.0)] = 0.0 # U entry 20; demand onset at 24 -> D1 (margin 4.0), not D2 | |
| g["hands_on"][(t >= 48.0) & (t < 60.0)] = 0.0 # U entry 48; demand onset at 50 -> D1 and D2 (margin 2.0) | |
| Dh = np.zeros(len(t), np.float32) | |
| for a_, b_ in ((3.0, 4.0), (24.0, 26.0), (50.0, 52.0), (70.0, 72.0)): # 70: demand while met -> no event | |
| Dh[(t >= a_) & (t < b_)] = 1.0 | |
| seg, U, dwell, WC, _, _ = _derive(g) | |
| onset1, n_c1, cens = AT.demand_onsets(Dh > 0.5, U, dwell, seg, None) | |
| assert np.allclose(t[onset1], [3.0, 24.0, 50.0]) and n_c1 == 1 | |
| onset2, n_c2, _ = AT.demand_onsets(Dh > 0.5, U, dwell, seg, 3.0) | |
| assert np.allclose(t[onset2], [50.0]) and n_c2 == 1 | |
| hands_off = np.isfinite(g["hands_on"]) & (g["hands_on"] < 0.5) | |
| eyes_off = np.zeros(len(t), bool) | |
| ev1, _ = AT.events_from_wc(Dh > 0.5, U, Dh, seg, t, g["uid"], g["dongle"], hands_off, eyes_off, onset=onset1, n_censored=n_c1, entry_cens=cens) | |
| assert np.allclose(ev1.lead_margin_s, [3.0, 4.0, 2.0]) and list(ev1.entry_censored) == [True, False, False] and list(ev1.trigger) == ["overlap"] * 3 | |
| assert np.allclose(ev1.t_recovery, [5.0, 30.0, 60.0]) | |
| # a persistence detector fires at the demand onset (lead 0) -> not early; an alert 1 s before the onset is early by 1 s | |
| sp = AT.score_persistence(U, Dh, g["moving"], dwell) | |
| ar = AT.alert_rows(g["uid"], t, sp, 0.75, cooldown=0.0) | |
| lead, _ = AT.match_events(ev1, g["uid"], t, ar) | |
| assert np.allclose(lead, [0.0, 0.0, 0.0]) | |
| # road rules: constant-velocity TTC and margin rule from M0 at t only | |
| M0 = np.zeros((4, len(AT.MR.MARGIN_NAMES)), np.float32) | |
| M0v = np.ones_like(M0, bool) | |
| c = {n_: i for i, n_ in enumerate(AT.MR.MARGIN_NAMES)} | |
| M0[:, c["lead_valid"]] = 1.0 | |
| M0[:, c["ttc_s"]] = [4.0, 30.0, 8.0, 2.0] | |
| M0[:, c["dRel_m"]] = [40.0, 200.0, 40.0, 20.0] | |
| M0[:, c["vRel_mps"]] = [-10.0, 0.0, -5.0, -10.0] # ttc_cv(3 s) = 1 / 3 -> 1.0 / 1.0; no closing -> 0; (40-15)/5 = 5 -> 0.2; (20-30->0)/10 = 0 -> 10 | |
| M0[:, c["headway_s"]] = 20.0 | |
| M0[:, c["lat_margin_m"]] = 2.0 | |
| v = AT.score_road_rule("cv_ttc_3s", M0, M0v) | |
| assert np.allclose(v, [1.0, 0.0, 0.2, 10.0]), v | |
| u = np.array([1.0, 1.0, 0.0, 1.0], np.float32) | |
| assert np.allclose(AT.score_road_rule("cv_ttc_3s_xU", M0, M0v, u), [1.0, 0.0, 0.0, 10.0]) | |
| hard = AT.score_road_rule("road_mr_hard", M0, M0v) | |
| assert np.allclose(hard, [1.0, 0.0, 0.5, 2.0]), hard # THR.ttc / ttc_s (30 s sentinel -> 0) | |
| assert np.all(AT.score_road_rule("road_mr_max", M0, M0v) >= hard - 1e-6) | |
| import inspect | |
| assert [q for q in inspect.signature(AT.score_road_rule).parameters] == ["name", "M0", "M0v", "U"] # current margins only, no future array | |
| def test_phead_demand_forecast_union_and_gating(tmp_path, monkeypatch): | |
| """--phead-runs (2026-09-17): P_hard3s = 1 - prod over the 15 steps and the 4 hard P families of (1 - hz_on_P); _xU gates by U_t; NaN where valid_E is False; | |
| D_ttc3s uses the legacy D chains ttc / ttc_drop of hz_on; a run whose layout rows differ from the grid is rejected.""" | |
| n = 40 | |
| g = _grid(T=n * 0.2) | |
| seg = AT.segments(g["uid"], g["t"]) | |
| g["U"] = np.zeros(n, np.float32); g["U"][n // 2:] = 1.0 | |
| g["order"] = np.arange(n); g["uid_raw"] = g["uid"]; g["t_raw"] = g["t"] | |
| rd = tmp_path / "RUN_X" | |
| rd.mkdir() | |
| np.savez(rd / "pred_layout_manual_test.npz", uid=g["uid"], t=g["t"]) | |
| p_names = np.array(["ttc", "ttc_drop", "cut_in", "lead_braking", "lane_excursion", "badas"]) | |
| c_names = np.array(AT_CHAINS) | |
| hzP = np.zeros((n, 25, 6), np.float32); hzP[:, :, 0] = 0.05; hzP[:, :, 4] = 0.9 # lane_excursion is NOT a hard family -> must not enter | |
| hzD = np.zeros((n, 25, 12), np.float32); hzD[:, :, c_names.tolist().index("ttc_drop")] = 0.1 | |
| vE = np.ones(n, bool); vE[3] = False | |
| np.savez(rd / "pred_manual_test.npz", hz_on_P=hzP, p_chain_names=p_names, hz_on=hzD, chain_names=c_names, valid_E=vE) | |
| monkeypatch.setattr(AT, "RUNS", tmp_path) | |
| v, src = AT.phead_column("RUN_X", "manual", "test", "P_hard3s", g) | |
| assert v is not None and np.isnan(v[3]) and np.isfinite(v[[0, 1, 2] + list(range(4, n))]).all() | |
| assert np.allclose(v[0], 1 - (1 - 0.05) ** 15, atol=1e-4) # the zero-hazard families are clipped to 1e-6 (45 factors -> +2e-5) | |
| vx, _ = AT.phead_column("RUN_X", "manual", "test", "P_hard3s_xU", g) | |
| assert np.allclose(vx[n // 2:], v[n // 2:]) and np.all(vx[: n // 2][np.isfinite(vx[: n // 2])] == 0.0) | |
| vd, _ = AT.phead_column("RUN_X", "manual", "test", "D_ttc3s_xU", g) | |
| assert np.allclose(vd[n // 2:], 1 - (1 - 0.1) ** 15, atol=1e-4) | |
| # incompatible layout -> rejected | |
| np.savez(rd / "pred_layout_manual_test.npz", uid=g["uid"], t=g["t"] + 0.2) | |
| v2, msg = AT.phead_column("RUN_X", "manual", "test", "P_hard3s", g) | |
| assert v2 is None and "differ" in msg | |
| AT_CHAINS = ["gas", "brake", "steer_phase", "resp", "hands_on", "head_turned", "steer_rate", "decel", "ttc", "ttc_drop", "lane", "badas"] | |
Xet Storage Details
- Size:
- 17.4 kB
- Xet hash:
- 6531f34303fbdf5962818427779e5c3ad3e14dab3ab8f633fdddb73c820dd1a8
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.