Buckets:
| """cwm3/eval/alert_timing_eval.py — EVENT-LEVEL TIMING EVALUATION of warning scores on the continuous TEST exposure (plan R2 WP2.3; pre-registration row | |
| 2026-09-16T06:46:35Z in docs/decisions.md). CPU only, numpy / pandas. | |
| python eval/alert_timing_eval.py [--seeds 0 1 2] [--budgets 5 10 20] [--deltas 0.5 1 2] [--n-boot 1000] [--extra-scores name=W_DIR:COLUMN ...] [--fs-dirs W_DIR ...] | |
| Events are defined AUTOMATICALLY on every anchor of both regimes' TEST splits (5 Hz s12 grid; NOT filtered by surrogate silence — surrogate-silent events are a | |
| labelled hard subset) from label-free causal state series and CAN margins: | |
| state at t hands_on / head_turned = S0_c of results/targets_kin_causal_routes_v2/kin_causal_{regime}_{split}.npz (causal pose, NaN invalid); | |
| gaze_off = S0_g of results/targets_gaze_causal_routes_v2/gaze_{regime}_{split}.npz (own face landmarks, NaN invalid) | |
| unmet state U_t hands_off ∨ eyes_off, eyes_off = head_turned ∨ gaze_off; an invalid component counts as met; U_t undefined when every component is | |
| invalid (forward-filled inside a continuous segment for <= FFILL_STEPS = 5 steps = 1 s, otherwise met) | |
| road demand D_t any road-pressure family high at t: P0 ∧ P0valid of results/targets_pressure/pressure_{regime}_{split}.npz (the CAN margins M0 of | |
| data/build_pressure_targets.py against its THR: ttc < 4 s, ttc drop, cut-in, lead braking, lane excursion, badas; moving gate > 2 m/s) | |
| continuous segment consecutive anchors of one route with dt <= SEG_GAP_S = 1.0 s (the grid is 0.2 s; 99.7 % of steps) | |
| state entry U_t 0 -> 1 inside a segment; dwell_t = t - t_entry of the current unbroken U run (0 when met) | |
| demand onset D_t 0 -> 1 inside a segment | |
| WARNING CONDITION WC_t = Moving0_t ∧ U_t ∧ (D_t ∨ dwell_t >= DWELL_S = 3 s) (state × demand overlap, or long distraction >= 3 s à la Euro NCAP) | |
| event onset o WC 0 -> 1 inside a segment (a segment that STARTS inside WC has no observed onset: counted as censored, not an event); onsets within | |
| EVENT_MERGE_S = 10 s of the previous kept onset on the same route are merged into that event (the S1 same-event guard of the human programme) | |
| recovery first anchor >= o in the segment with U_t = 0; NaN (censored) when the segment ends first | |
| Alerts: score >= threshold with the EXISTING 60 s per-route cooldown of train_warning_head.alerts_per_hour (rows in (uid, t) order); thresholds per score × | |
| regime fitted on the VAL split (all anchors) by bisection (operating_points.thr_for_rate semantics, 24 iterations) to budgets of 5 / 10 / 20 alerts per | |
| exposure hour (exposure = rows × 0.2 s); a score whose floor (alerts/h at threshold = max score) exceeds the budget is reported AT ITS FLOOR and flagged. | |
| Matching: an event is detected by the first alert in [o − 10, o + 3] s; lead = o − t_alert (negative = late); PRIMARY Recall_{>= δ s early} = share of events | |
| with lead >= δ, δ ∈ {0.5, 1, 2} at each budget; secondary: detected share, first-alert lead distribution (undetected = 0; late leads negative), TTA of late | |
| detections (t_alert − o), achieved alerts/h and FA/h (alerts not inside [o − 10, max(o + 3, t_recovery)] of any event of the route), per regime and pooled; | |
| device-cluster bootstrap (devices resampled with replacement, thresholds fixed) with paired differences between scores on identical resamples. | |
| Scores: `surrogate_att3s` (openpilot-derived surrogate: legacy attention marginals -> compose.legacy_compose att_only, tau 15; test marginals seed-matched, | |
| val marginals exist for seed 0 only -> every seed uses the s0 val thresholds, disclosed), `cur_dwfc` = W_<run>_dwfc::W_learned_curhgb_t15_w10_n979dwfc, | |
| `both_dwfc` = W_<run>_dwfcboth::W_learned_bothhgb_t15_w10_n979dwfc, `both_base` = W_<run>_merged::W_learned_bothhgb_t15_w10_n979 (human-programme system A, | |
| context), `persistence` = 1[U_t]·(0.5 + 0.5·1[D_t])·Moving0 (the current state assumed to persist: 1.0 = WC already reached, 0.5 = unmet without demand -> | |
| reaches the dwell threshold if it persists), `dwell_timer` = 1[U_t]·(min(dwell_t, 3)/3 + 1[D_t])/2·Moving0 (time-to-threshold extrapolation of the current dwell), | |
| plus any `fs_*` future-state columns of --fs-dirs layouts and --extra-scores (hook for later agents); `--phead-runs name=RUN_DIR` (2026-09-17, owner decision (7)) adds a joint | |
| run's learned DEMAND-ONSET forecast as road scores `ph_<name>_P_hard3s` = P(any hard P family ttc / ttc_drop / cut_in / lead_braking first switches high within 3 s) = closed-form first | |
| passage of the P head's onset hazards (union over families and the 15 steps), `_xU` = x 1[U_t], and `ph_<name>_D_ttc3s_xU` (the legacy D chains ttc / ttc_drop) — e.g. the ECAN joint | |
| runs vs the frozen main. Every non-oracle score at t reads anchor rows <= t only | |
| (tests/test_alert_timing.py leakage test); the only t+ quantity used anywhere is the event recovery time (declared). | |
| Confirmation set: human event records of cwm2/annotation_human/labels_newtest.jsonl (stage=event; usable = event_type != none with a valid event_onset_s; | |
| absolute onset = t0 − anchor_s + event_onset_s) — per record the nearest automatic onset and each score's first alert at the 10 / h val thresholds. | |
| Outputs (append-only, new files): results/tables/alert_timing.md (section appended) + results/tables/alert_timing_<stamp>.json; results/alert_timing/<stamp>/ | |
| {events_<regime>.csv, metrics.csv, paired.csv, thresholds.json, human_confirmation.csv, run.json}. No openpilot DMS field is read (the surrogate enters as the | |
| labelled comparator score, as in warning_gold.py). | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import sys | |
| import time | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| EXPLORE = CWM3.parent | |
| CWM2 = EXPLORE / "cwm2" | |
| for p in (CWM2, CWM2 / "eval", CWM3 / "common", CWM3 / "eval"): | |
| if str(p) not in sys.path: | |
| sys.path.insert(0, str(p)) | |
| import margin_rule as MR # noqa: E402 (label-free current-margin rule; the road-rule baseline) | |
| REGIMES = ("manual", "engaged") | |
| KINC = CWM3 / "results" / "targets_kin_causal_routes_v2" | |
| GAZE = CWM3 / "results" / "targets_gaze_causal_routes_v2" | |
| PRESS = CWM3 / "results" / "targets_pressure" | |
| RUNS = CWM3 / "results" / "runs" | |
| LEGACY = CWM2 / "baselines" / "legacy_cockpitwm" / "results" | |
| HUMAN = CWM2 / "annotation_human" / "labels_newtest.jsonl" | |
| GRID_DT, COOLDOWN, TAU_SURR = 0.2, 60.0, 15 | |
| MATCH_PRE, MATCH_POST = 10.0, 3.0 | |
| EVENT_MERGE_S, DWELL_S, SEG_GAP_S, FFILL_STEPS = 10.0, 3.0, 1.0, 5 | |
| RULE_SCORES = ("persistence", "dwell_timer") | |
| HARD_FAMILIES = ("ttc", "ttc_drop", "cut_in", "lead_braking") # secondary event family B: overlap with a kinematic-conflict demand family only | |
| FAMILY_LABEL = {"A": f"A (registered, PRIMARY): Moving ∧ U ∧ (D ∨ dwell ≥ {DWELL_S:g} s)", "B": "B (secondary, stricter): Moving ∧ U ∧ D_hard, D_hard = ttc / ttc_drop / cut_in / lead_braking high", | |
| "C1": "C1 (addendum): Moving ∧ U sustained ≥ 3.0 s — event time = the anchor at which the U run reaches 3.0 s (Euro NCAP long-distraction analogue)", | |
| "C2": "C2 (addendum): Moving ∧ U ∧ D_hard sustained ≥ 1.0 s — event time = the anchor at which the overlap run reaches 1.0 s", | |
| "C1_5s": "C1 sensitivity: U sustained ≥ 5.0 s", "C1_10s": "C1 sensitivity: U sustained ≥ 10.0 s", "C2_2s": "C2 sensitivity: U ∧ D_hard sustained ≥ 2.0 s", | |
| "D1": "D1 (addendum 2): onset of a HARD road demand (D_hard 0→1) while the driver is already unmet (U active at the onset anchor) — event time = the demand onset", | |
| "D2": "D2 (addendum 2): D1 with the U entry observed within the 3 s before the demand onset (lead margin ≤ 3 s; entry-censored runs excluded)"} | |
| # family spec: ("wc", grid key) = onset at WC 0 -> 1; ("sustained", grid key, min_s) = onset at the anchor where the unbroken run of the condition first lasts min_s | |
| FAMILY_SPEC = {"A": ("wc", "WC"), "B": ("wc", "WC_B"), "C1": ("sustained", "U_mov", 3.0), "C2": ("sustained", "UDh_mov", 1.0), | |
| "C1_5s": ("sustained", "U_mov", 5.0), "C1_10s": ("sustained", "U_mov", 10.0), "C2_2s": ("sustained", "UDh_mov", 2.0), | |
| "D1": ("demand_onset", "D_hard", None), "D2": ("demand_onset", "D_hard", 3.0)} | |
| ROAD_RULES = ("road_mr_max", "road_mr_hard", "cv_ttc_3s", "road_mr_max_xU", "road_mr_hard_xU", "cv_ttc_3s_xU") | |
| RD_SCORES = ("rd_ttc_min_3s", "rd_ttc_min_3s_xU") | |
| FS_SCORES = ("fs_occ_any_1s", "fs_occ_any_3s", "fs_EDoff_5s") | |
| PH_KINDS = ("P_hard3s", "P_hard3s_xU", "D_ttc3s_xU") # --phead-runs score kinds (learned demand-onset forecasts of a joint run) | |
| PH_D_TTC, PH_TAU = ("ttc", "ttc_drop"), 15 | |
| # ----------------------------------------------------------------------------- series | |
| def segments(uid, t, gap=SEG_GAP_S): | |
| """segment id per row (rows sorted by (uid, t)): a new segment at a route change or a gap > gap seconds.""" | |
| n = len(t) | |
| if n == 0: | |
| return np.zeros(0, np.int64) | |
| brk = np.ones(n, bool) | |
| brk[1:] = (uid[1:] != uid[:-1]) | (np.diff(t) > gap + 1e-9) | |
| return np.cumsum(brk) - 1 | |
| def ffill_segment(x, seg, max_steps=FFILL_STEPS): | |
| """forward-fill NaN inside a segment for at most max_steps consecutive steps (x float, NaN = undefined).""" | |
| n = len(x) | |
| out = x.copy() | |
| fin = np.isfinite(x) | |
| idx = np.where(fin, np.arange(n), -1) | |
| idx = np.maximum.accumulate(idx) | |
| ok = (idx >= 0) & (np.arange(n) - idx <= max_steps) & (seg[np.clip(idx, 0, n - 1)] == seg) | |
| fill = ~fin & ok | |
| out[fill] = x[idx[fill]] | |
| return out | |
| def unmet_series(hands_on, head, gaze): | |
| """U_t float {0, 1, NaN}: hands_off ∨ head_turned ∨ gaze_off; invalid component = met; NaN when every component is invalid.""" | |
| v_h, v_d, v_g = np.isfinite(hands_on), np.isfinite(head), np.isfinite(gaze) | |
| U = (v_h & (hands_on < 0.5)) | (v_d & (head > 0.5)) | (v_g & (gaze > 0.5)) | |
| return np.where(v_h | v_d | v_g, U.astype(np.float32), np.nan) | |
| def dwell_series(U, seg, t): | |
| """dwell_t (s) = t − t_entry of the current unbroken run of U = 1 inside the segment; 0 where U = 0.""" | |
| n = len(U) | |
| on = U > 0.5 | |
| start = on.copy() | |
| start[1:] = on[1:] & (~on[:-1] | (seg[1:] != seg[:-1])) | |
| run = np.cumsum(start) - 1 | |
| t_entry = np.full(max(run.max() + 1, 1), np.nan) | |
| t_entry[run[start]] = t[start] | |
| d = np.where(on, t - t_entry[np.clip(run, 0, len(t_entry) - 1)], 0.0) | |
| return d.astype(np.float64) | |
| def warning_condition(U, D, moving, dwell, dwell_s=DWELL_S): | |
| return (moving > 0.5) & (U > 0.5) & ((D > 0.5) | (dwell >= dwell_s - 1e-9)) | |
| def sustained_onsets(cond, seg, t, min_s): | |
| """onset mask of 'condition sustained >= min_s': the first anchor of each unbroken run of cond (inside a segment) whose duration t - t_run_start reaches min_s; | |
| runs that start at a segment start (entry unobserved) are censored -> (onset mask, n_censored).""" | |
| n = len(cond) | |
| on = np.asarray(cond, bool) | |
| start = on.copy() | |
| start[1:] = on[1:] & (~on[:-1] | (seg[1:] != seg[:-1])) | |
| run = np.cumsum(start) - 1 | |
| n_run = max(int(run.max()) + 1, 1) | |
| t_start = np.full(n_run, np.nan) | |
| t_start[run[start]] = t[start] | |
| dur = np.where(on, t - t_start[np.clip(run, 0, n_run - 1)], -1.0) | |
| reached = on & (dur >= min_s - 1e-9) | |
| first = reached.copy() | |
| first[1:] = reached[1:] & ~(reached[:-1] & (run[1:] == run[:-1])) | |
| seg_start = np.ones(n, bool) | |
| seg_start[1:] = seg[1:] != seg[:-1] | |
| run_cens = np.zeros(n_run, bool) | |
| run_cens[run[start]] = seg_start[start] | |
| censored = first & run_cens[np.clip(run, 0, n_run - 1)] | |
| return first & ~censored, int(censored.sum()) | |
| def entry_censored_rows(on, seg): | |
| """per row: True when the unbroken run of `on` containing the row started at a segment start (state entry unobserved).""" | |
| n = len(on) | |
| on = np.asarray(on, bool) | |
| start = on.copy() | |
| start[1:] = on[1:] & (~on[:-1] | (seg[1:] != seg[:-1])) | |
| run = np.cumsum(start) - 1 | |
| n_run = max(int(run.max()) + 1, 1) | |
| seg_start = np.ones(n, bool) | |
| seg_start[1:] = seg[1:] != seg[:-1] | |
| run_cens = np.zeros(n_run, bool) | |
| run_cens[run[start]] = seg_start[start] | |
| return on & run_cens[np.clip(run, 0, n_run - 1)] | |
| def demand_onsets(Dh, U, dwell, seg, max_dwell=None): | |
| """family D: onset mask = D_hard 0 -> 1 inside a segment with U active at that anchor; max_dwell (D2) keeps only onsets whose observed U entry lies within | |
| max_dwell s before (entry-censored runs excluded); -> (onset mask, n_censored, entry_censored row mask).""" | |
| n = len(Dh) | |
| Dh = np.asarray(Dh, bool) | |
| onset = np.zeros(n, bool) | |
| onset[1:] = Dh[1:] & ~Dh[:-1] & (seg[1:] == seg[:-1]) & (U[1:] > 0.5) | |
| cens = entry_censored_rows(U > 0.5, seg) | |
| if max_dwell is not None: | |
| keep = onset & (dwell <= max_dwell + 1e-9) & ~cens | |
| n_c = int((onset & cens & (dwell <= max_dwell + 1e-9)).sum()) | |
| return keep, n_c, cens | |
| return onset, int((onset & cens).sum()), cens | |
| def score_road_rule(name, M0, M0v, U=None): | |
| """label-free road rules from the CURRENT margins M0 (t only): road_mr_max = margin_rule.score; road_mr_hard = max over ttc / headway / lead_decel ratios; | |
| cv_ttc_3s = 1 / max(min_h ttc_cv(h), 0.1) with ttc_cv(h) = max(dRel + vRel h, 0) / max(-vRel, 0.1), h = 1, 2, 3 s (0 when no valid approaching lead; | |
| under constant velocity ttc_cv(h) = ttc_now - h, so the ranking equals the current-TTC rule's); `_xU` variants = rule × 1[U_t].""" | |
| base = name.replace("_xU", "") | |
| if base == "road_mr_max": | |
| v = MR.score(M0, M0v) | |
| elif base == "road_mr_hard": | |
| r = MR.rule_ratios(M0, M0v) | |
| v = np.maximum.reduce([r["ttc"], r["headway"], r["lead_decel"]]) | |
| elif base == "cv_ttc_3s": | |
| c = {n_: i for i, n_ in enumerate(MR.MARGIN_NAMES)} | |
| d, vr = M0[:, c["dRel_m"]].astype(np.float64), M0[:, c["vRel_mps"]].astype(np.float64) | |
| ok = M0v[:, c["dRel_m"]] & M0v[:, c["vRel_mps"]] & (M0v[:, c["lead_valid"]] & (M0[:, c["lead_valid"]] > 0.5)) & (d < 199.0) & (vr < -1e-6) | |
| closing = np.maximum(-vr, 0.1) | |
| ttc_h = np.stack([np.maximum(d + vr * h, 0.0) / closing for h in (1.0, 2.0, 3.0)], 1) | |
| v = np.where(ok, 1.0 / np.maximum(ttc_h.min(1), 0.1), 0.0) | |
| else: | |
| raise ValueError(name) | |
| v = np.asarray(v, np.float32) | |
| if name.endswith("_xU"): | |
| v = v * (U > 0.5) | |
| return v | |
| def events_from_wc(WC, U, D, seg, t, uid, dongle, hands_off, eyes_off, merge_s=EVENT_MERGE_S, onset=None, n_censored=None, entry_cens=None): | |
| """onsets (WC 0 -> 1 inside a segment, or an explicit `onset` mask from sustained_onsets), merged per route within merge_s of the previous kept onset -> | |
| DataFrame (lead_margin_s = onset - entry of the U run = the earliness that is physically available); also the censored-start count.""" | |
| n = len(WC) | |
| seg_start = np.ones(n, bool) | |
| seg_start[1:] = seg[1:] != seg[:-1] | |
| if onset is None: | |
| onset = np.zeros(n, bool) | |
| onset[1:] = WC[1:] & ~WC[:-1] & (seg[1:] == seg[:-1]) | |
| n_censored_start = int((WC & seg_start).sum()) | |
| else: | |
| n_censored_start = int(n_censored or 0) | |
| seg_end_t = np.full(n, np.nan) | |
| last = np.ones(n, bool) | |
| last[:-1] = seg[1:] != seg[:-1] | |
| ends = np.flatnonzero(last) | |
| starts = np.r_[0, ends[:-1] + 1] | |
| for s0, e0 in zip(starts, ends): | |
| seg_end_t[s0:e0 + 1] = t[e0] | |
| rows = [] | |
| last_uid, last_o = None, -1e9 | |
| for i in np.flatnonzero(onset): | |
| if uid[i] != last_uid: | |
| last_uid, last_o = uid[i], -1e9 | |
| if t[i] - last_o <= merge_s: | |
| continue | |
| last_o = t[i] | |
| # recovery: first j >= i in the segment with U = 0 | |
| j = i | |
| end = ends[np.searchsorted(ends, i)] | |
| rec = np.nan | |
| sub = U[i:end + 1] | |
| k = np.flatnonzero(sub < 0.5) | |
| if len(k): | |
| rec = t[i + k[0]] | |
| lm = float(dwell_at(i, U, seg, t)) | |
| rows.append(dict(uid=uid[i], dongle=dongle[i], row=int(i), onset_t=float(t[i]), trigger="overlap" if D[i] > 0.5 else "dwell", | |
| hands_off=bool(hands_off[i]), eyes_off=bool(eyes_off[i]), t_recovery=rec, seg_end_t=float(seg_end_t[i]), | |
| dwell_at_onset=lm, lead_margin_s=lm, entry_censored=bool(entry_cens[i]) if entry_cens is not None else False)) | |
| return pd.DataFrame(rows), n_censored_start | |
| def dwell_at(i, U, seg, t): | |
| j = i | |
| while j > 0 and U[j - 1] > 0.5 and seg[j - 1] == seg[i]: | |
| j -= 1 | |
| return t[i] - t[j] | |
| # ----------------------------------------------------------------------------- rule scores (read rows <= t only) | |
| def score_persistence(U, D, moving, dwell): | |
| return np.where(np.isfinite(U), (U > 0.5) * (0.5 + 0.5 * (D > 0.5)) * (moving > 0.5), np.nan).astype(np.float32) | |
| def score_dwell_timer(U, D, moving, dwell, dwell_s=DWELL_S): | |
| return np.where(np.isfinite(U), (U > 0.5) * (np.minimum(dwell, dwell_s) / dwell_s + (D > 0.5)) / 2.0 * (moving > 0.5), np.nan).astype(np.float32) | |
| # ----------------------------------------------------------------------------- alerts | |
| def alert_rows(uid, t, s, thr, cooldown=COOLDOWN): | |
| """row indices of the alerts (rows assumed sorted by (uid, t)): s >= thr and >= cooldown since the route's last alert — train_warning_head.alerts_per_hour semantics.""" | |
| fire = np.isfinite(s) & (s >= thr) | |
| out = [] | |
| last_uid, last_t = None, -1e9 | |
| for j in np.flatnonzero(fire): | |
| if uid[j] != last_uid: | |
| last_uid, last_t = uid[j], -1e9 | |
| elif t[j] - last_t < cooldown: | |
| continue | |
| out.append(j) | |
| last_t = t[j] | |
| return np.asarray(out, dtype=np.int64) | |
| def _alerts_fast(uid, t, s, thr, cooldown=COOLDOWN): | |
| """same as alert_rows but resets the route's last alert correctly when fire rows of a route are skipped: alert_rows only sees firing rows, so the | |
| first firing row of a route always fires (correct) and later ones need >= cooldown since the LAST ALERT (not the last firing row) — implemented above.""" | |
| return alert_rows(uid, t, s, thr, cooldown) | |
| def alerts_per_hour(uid, t, s, thr, cooldown=COOLDOWN): | |
| n = len(alert_rows(uid, t, s, thr, cooldown)) | |
| hours = len(t) * GRID_DT / 3600.0 | |
| return n, hours, n / hours if hours > 0 else np.nan | |
| def thr_for_rate(uid, t, s, rate, it=24): | |
| """bisection on the threshold so that alerts/h (60 s cooldown) matches `rate`; -> (thr, achieved, reachable, floor).""" | |
| fin = np.isfinite(s) | |
| if not fin.any(): | |
| return np.nan, np.nan, False, np.nan | |
| lo, hi = float(np.min(s[fin])), float(np.max(s[fin])) | |
| floor = alerts_per_hour(uid, t, s, hi, )[2] | |
| if floor > rate: | |
| return hi, floor, False, floor | |
| for _ in range(it): | |
| mid = 0.5 * (lo + hi) | |
| r_ = alerts_per_hour(uid, t, s, mid)[2] | |
| if r_ > rate: | |
| lo = mid | |
| else: | |
| hi = mid | |
| thr = hi | |
| return thr, alerts_per_hour(uid, t, s, thr)[2], True, floor | |
| # ----------------------------------------------------------------------------- matching | |
| def match_events(ev, uid, t, arows, pre=MATCH_PRE, post=MATCH_POST): | |
| """per event: first alert in [o − pre, o + post] -> lead = o − t_alert (NaN undetected); TTA of a late first alert; also the alert justification mask | |
| (alert inside [o − pre, max(o + post, t_recovery)] of any event of its route) -> (lead[n_ev], justified[n_alerts]).""" | |
| at = t[arows] | |
| au = uid[arows] | |
| by_route = {} | |
| for k, (u, tt) in enumerate(zip(au, at)): | |
| by_route.setdefault(u, []).append((tt, k)) | |
| lead = np.full(len(ev), np.nan) | |
| justified = np.zeros(len(arows), bool) | |
| for i, e in enumerate(ev.itertuples()): | |
| lst = by_route.get(e.uid) | |
| if not lst: | |
| continue | |
| o = e.onset_t | |
| hi_just = max(o + post, e.t_recovery if np.isfinite(e.t_recovery) else e.seg_end_t) | |
| first = None | |
| for tt, k in lst: | |
| if o - pre <= tt <= hi_just: | |
| justified[k] = True | |
| if o - pre <= tt <= o + post and first is None: | |
| first = tt | |
| if first is not None: | |
| lead[i] = o - first | |
| return lead, justified | |
| # ----------------------------------------------------------------------------- data | |
| def load_grid(regime, split): | |
| with np.load(KINC / f"kin_causal_{regime}_{split}.npz", allow_pickle=True) as z: | |
| uid, dongle, t = z["uid"], z["dongle"], z["t"].astype(np.float64) | |
| S0 = z["S0_c"].astype(np.float32) | |
| with np.load(GAZE / f"gaze_{regime}_{split}.npz", allow_pickle=True) as z: | |
| assert np.array_equal(z["uid"], uid) and np.allclose(z["t"], t) | |
| S0g = z["S0_g"].astype(np.float32) | |
| with np.load(PRESS / f"pressure_{regime}_{split}.npz", allow_pickle=True) as z: | |
| assert np.array_equal(z["uid"], uid) and np.allclose(z["t"], t) | |
| fam = [str(x) for x in z["families"]] | |
| hard = np.array([f in HARD_FAMILIES for f in fam]) | |
| PP = z["P0"] & z["P0valid"] | |
| D = PP.any(1).astype(np.float32) | |
| D_hard = PP[:, hard].any(1).astype(np.float32) | |
| moving = z["Moving0"].astype(np.float32) | |
| M0, M0v = z["M0"].astype(np.float32), z["M0valid"].astype(bool) | |
| order = np.lexsort((t, uid)) | |
| g = dict(uid=uid[order], dongle=dongle[order], t=t[order], hands_on=S0[order, 0], head=S0[order, 1], gaze=S0g[order], D=D[order], D_hard=D_hard[order], moving=moving[order], order=order, uid_raw=uid, t_raw=t) | |
| g["seg"] = segments(g["uid"], g["t"]) | |
| U = ffill_segment(unmet_series(g["hands_on"], g["head"], g["gaze"]), g["seg"]) | |
| g["U"] = np.where(np.isfinite(U), U, 0.0) | |
| g["U_defined"] = np.isfinite(U) | |
| g["dwell"] = dwell_series(g["U"], g["seg"], g["t"]) | |
| g["hands_off"] = np.isfinite(g["hands_on"]) & (g["hands_on"] < 0.5) | |
| g["eyes_off"] = (np.isfinite(g["head"]) & (g["head"] > 0.5)) | (np.isfinite(g["gaze"]) & (g["gaze"] > 0.5)) | |
| g["WC"] = warning_condition(g["U"], g["D"], g["moving"], g["dwell"]) | |
| g["WC_B"] = (g["moving"] > 0.5) & (g["U"] > 0.5) & (g["D_hard"] > 0.5) | |
| g["M0"], g["M0valid"] = M0[order], M0v[order] | |
| g["U_mov"] = (g["moving"] > 0.5) & (g["U"] > 0.5) | |
| g["UDh_mov"] = g["U_mov"] & (g["D_hard"] > 0.5) | |
| g["hours"] = len(t) * GRID_DT / 3600.0 | |
| return g | |
| def build_events(g, fam): | |
| """events of family `fam` (FAMILY_SPEC) on grid g -> (DataFrame, n_censored).""" | |
| spec = FAMILY_SPEC[fam] | |
| D_lab = g["D"] if fam == "A" or fam.startswith("C1") else g["D_hard"] | |
| if spec[0] == "wc": | |
| return events_from_wc(g[spec[1]], g["U"], D_lab, g["seg"], g["t"], g["uid"], g["dongle"], g["hands_off"], g["eyes_off"]) | |
| if spec[0] == "demand_onset": | |
| onset, n_c, cens = demand_onsets(g[spec[1]] > 0.5, g["U"], g["dwell"], g["seg"], spec[2]) | |
| return events_from_wc(g[spec[1]] > 0.5, g["U"], g["D_hard"], g["seg"], g["t"], g["uid"], g["dongle"], g["hands_off"], g["eyes_off"], onset=onset, n_censored=n_c, entry_cens=cens) | |
| onset, n_c = sustained_onsets(g[spec[1]], g["seg"], g["t"], spec[2]) | |
| return events_from_wc(g[spec[1]], g["U"], D_lab, g["seg"], g["t"], g["uid"], g["dongle"], g["hands_off"], g["eyes_off"], onset=onset, n_censored=n_c) | |
| def rd_column(rd_dir, regime, split, name, g): | |
| """B2-lane road-dynamics forecast results/runs/<rd_dir>/pred_rd_{regime}_{split}.npz (mu [N,25,4] = CV + residual, channels lead_dRel / lead_vRel / ...): | |
| rd_ttc_min_3s = 1 / max(min_{k=+1,+2,+3 s} max(mu_dRel_k, 0) / max(-mu_vRel_k, 0.1), 0.1) where both lead channels are valid (0 when the lead is not closing at | |
| every horizon; NaN where invalid); `_xU` = × 1[U_t].""" | |
| p = RUNS / rd_dir / f"pred_rd_{regime}_{split}.npz" | |
| if not p.exists(): | |
| return None, f"{p} missing" | |
| with np.load(p, allow_pickle=True) as z: | |
| if not (np.array_equal(z["uid"], g["uid_raw"]) and np.allclose(z["t"], g["t_raw"])): | |
| return None, f"{p}: rows differ from the causal grid (incompatible layout)" | |
| ch = [str(c) for c in z["channels"]] | |
| i_d, i_v = ch.index("lead_dRel"), ch.index("lead_vRel") | |
| ks = [4, 9, 14] | |
| d = z["mu"][:, ks, i_d].astype(np.float32) | |
| vr = z["mu"][:, ks, i_v].astype(np.float32) | |
| ok = z["row_ok"][:, i_d] & z["row_ok"][:, i_v] | |
| ttc = np.where(vr < -1e-6, np.maximum(d, 0.0) / np.maximum(-vr, 0.1), 1e3) | |
| v = np.where(ok, 1.0 / np.maximum(ttc.min(1), 0.1), np.nan).astype(np.float32) | |
| v = v[g["order"]] | |
| if name.endswith("_xU"): | |
| v = v * (g["U"] > 0.5) | |
| return v, str(p) | |
| def phead_column(run_dir, regime, split, kind, g): | |
| """learned DEMAND-ONSET forecast of a joint run (results/runs/<run_dir>/pred_{regime}_{split}.npz; its rows = the run's pred_layout = the causal grid, checked): | |
| P_hard3s = P(any HARD road-pressure family first switches high within 3 s) = 1 - prod_f prod_{k<=15} (1 - hz_on_P[k, f]) over the P chains ttc / ttc_drop / cut_in / | |
| lead_braking (= the families of D_hard; the closed-form first passage used by train_warning_head.onset3s_P_*); D_ttc3s = the same over the legacy D chains ttc / ttc_drop | |
| (hz_on); `_xU` = x 1[U_t]. NaN where valid_E is False. Reads the hazards emitted at t only (rows <= t).""" | |
| p = RUNS / run_dir / f"pred_{regime}_{split}.npz" | |
| lp = RUNS / run_dir / f"pred_layout_{regime}_{split}.npz" | |
| if not p.exists() or not lp.exists(): | |
| return None, f"{p} missing" | |
| with np.load(lp, allow_pickle=True) as z: | |
| if not (np.array_equal(z["uid"], g["uid_raw"]) and np.allclose(z["t"], g["t_raw"])): | |
| return None, f"{lp}: rows differ from the causal grid (incompatible layout)" | |
| base = kind.replace("_xU", "") | |
| with np.load(p, allow_pickle=True) as z: | |
| if base == "P_hard3s": | |
| names = [str(x) for x in z["p_chain_names"]] | |
| idx = [names.index(f) for f in HARD_FAMILIES] | |
| hz = z["hz_on_P"][:, :PH_TAU, :][:, :, idx].astype(np.float64) | |
| elif base == "D_ttc3s": | |
| names = [str(x) for x in z["chain_names"]] | |
| idx = [names.index(f) for f in PH_D_TTC] | |
| hz = z["hz_on"][:, :PH_TAU, :][:, :, idx].astype(np.float64) | |
| else: | |
| raise ValueError(kind) | |
| ok = z["valid_E"].astype(bool) | |
| assert len(ok) == len(g["uid_raw"]) == len(hz), (p, len(ok), len(hz), len(g["uid_raw"])) | |
| hz = np.clip(hz, 1e-6, 1 - 1e-6) | |
| v = 1.0 - np.prod(1.0 - hz, axis=(1, 2)) # union first passage over the horizon steps and the families | |
| v = np.where(ok, v, np.nan).astype(np.float32)[g["order"]] | |
| if kind.endswith("_xU"): | |
| v = v * (g["U"] > 0.5) | |
| return v, str(p) | |
| def overlap_spec(spec): | |
| """--overlap-dirs entry -> (score name, run dir): `name=DIR` -> ("ov_<name>", DIR); bare `DIR` -> ("ov_J", DIR) (the original hook, unchanged).""" | |
| if "=" in spec: | |
| name, d = spec.split("=", 1) | |
| return f"ov_{name}", d | |
| return "ov_J", spec | |
| def overlap_column(ov_dir, regime, split, g): | |
| """future F8 overlap head hook: results/runs/<ov_dir>/pred_overlap_{regime}_{split}.npz with uid / t and J[N] (or q[N,25] -> 0.2 * sum_k q_k).""" | |
| p = RUNS / ov_dir / f"pred_overlap_{regime}_{split}.npz" | |
| if not p.exists(): | |
| return None, f"{p} missing (no F8 overlap head output exists)" | |
| with np.load(p, allow_pickle=True) as z: | |
| if not (np.array_equal(z["uid"], g["uid_raw"]) and np.allclose(z["t"], g["t_raw"])): | |
| return None, f"{p}: rows differ from the causal grid (incompatible layout)" | |
| v = z["J"].astype(np.float32) if "J" in z.files else (0.2 * z["q"].astype(np.float32).sum(1)) | |
| return v[g["order"]], str(p) | |
| def fs_dwell_column(fs_dir, regime, split, name, g): | |
| """declared model scores from a WP4 future-state pack results/runs/<fs_dir>/fs_{regime}_{split}.npz (row-aligned with the grid; `valid` applied): | |
| fs_occ_any_1s / fs_occ_any_3s = 1 - prod_s (1 - pi_off[k, s]) at k = +1 s / +3 s; fs_EDoff_5s = sum_s ED_off_s.""" | |
| p = RUNS / fs_dir / f"fs_{regime}_{split}.npz" | |
| if not p.exists(): | |
| return None, f"{p} missing" | |
| with np.load(p, allow_pickle=True) as z: | |
| if not (np.array_equal(z["uid"], g["uid_raw"]) and np.allclose(z["t"], g["t_raw"])): | |
| return None, f"{p}: rows differ from the causal grid (incompatible layout)" | |
| valid = z["valid"].astype(bool) | |
| if name == "fs_EDoff_5s": | |
| v = z["ED_off_s"].astype(np.float32).sum(1) | |
| else: | |
| k = 4 if name == "fs_occ_any_1s" else 14 | |
| po = np.clip(z["pi_off"][:, k, :].astype(np.float32), 0.0, 1.0) | |
| v = 1.0 - np.prod(1.0 - po, 1) | |
| v = np.where(valid, v, np.nan).astype(np.float32) | |
| return v[g["order"]], str(p) | |
| def alert_early_share(ev, uid, t, arows, delta=1.0, pre=MATCH_PRE): | |
| """share of alerts that precede some event onset of their route by >= delta s (and <= pre s): alert-level 'early hit' share.""" | |
| if len(arows) == 0: | |
| return np.nan | |
| ons = {u: np.sort(g.onset_t.to_numpy()) for u, g in ev.groupby("uid")} if len(ev) else {} | |
| hit = 0 | |
| for k in arows: | |
| o = ons.get(uid[k]) | |
| if o is None: | |
| continue | |
| j = np.searchsorted(o, t[k] + delta) # first onset >= t_alert + delta | |
| if j < len(o) and o[j] - t[k] <= pre: | |
| hit += 1 | |
| return hit / len(arows) | |
| def layout_column(w_dir, regime, split, col, g): | |
| p = RUNS / w_dir / f"pred_layout_{regime}_{split}.npz" | |
| with np.load(p, allow_pickle=True) as z: | |
| assert np.array_equal(z["uid"], g["uid_raw"]) and np.allclose(z["t"], g["t_raw"]), f"{p}: rows differ from the causal grid" | |
| names = [str(x) for x in z["w_names"]] | |
| assert col in names, f"{p}: column {col} not in {names}" | |
| v = z["w"][:, names.index(col)].astype(np.float32) | |
| if "w_mask" in z.files: | |
| v = np.where(z["w_mask"], v, np.nan) | |
| return v[g["order"]] | |
| def surrogate_column(regime, split, seed, g): | |
| """legacy attention marginals -> att_only (first passage of the inattention union, tau 15) mapped onto the grid by (uid, t); NaN where absent.""" | |
| from compose import legacy_compose | |
| p = LEGACY / (f"marginals_{regime}_s{seed}.npz" if split == "test" else f"marginals_{regime}_val_s{seed}.npz") | |
| if not p.exists(): | |
| return None, str(p) | |
| with np.load(p, allow_pickle=True) as z: | |
| sc = legacy_compose(z["pA"].astype(np.float32), None, TAU_SURR)["att_only"].astype(np.float32) | |
| key = pd.DataFrame({"uid": z["uid"], "tk": np.round(z["t"].astype(np.float64) * 10).astype(np.int64), "s": sc}) | |
| grid = pd.DataFrame({"uid": g["uid"], "tk": np.round(g["t"] * 10).astype(np.int64)}) | |
| m = grid.merge(key.drop_duplicates(["uid", "tk"]), on=["uid", "tk"], how="left") | |
| return m["s"].to_numpy(np.float32), str(p) | |
| # ----------------------------------------------------------------------------- bootstrap | |
| def cluster_boot_events(dongle, ind, n_boot, seed=0): | |
| """ind [n_ev, K] float indicators -> resampled means [n_boot, K] (devices with replacement).""" | |
| rng = np.random.default_rng(seed) | |
| ug = np.unique(dongle) | |
| gidx = {d: np.flatnonzero(dongle == d) for d in ug} | |
| out = np.full((n_boot, ind.shape[1]), np.nan) | |
| for b in range(n_boot): | |
| pick = np.concatenate([gidx[d] for d in rng.choice(ug, len(ug), replace=True)]) | |
| out[b] = ind[pick].mean(0) | |
| return out | |
| def pct(v): | |
| v = v[np.isfinite(v)] | |
| return (float(np.percentile(v, 2.5)), float(np.percentile(v, 97.5))) if len(v) else (np.nan, np.nan) | |
| def f3(x, nd=3): | |
| try: | |
| x = float(x) | |
| except (TypeError, ValueError): | |
| return "—" | |
| return "—" if not np.isfinite(x) else f"{x:.{nd}f}" | |
| # ----------------------------------------------------------------------------- main | |
| def _recall_row(ld, deltas): | |
| return {f"recall_early_{d:g}s": float(np.mean(np.nan_to_num(ld, nan=-99) >= d)) if len(ld) else np.nan for d in deltas} | |
| def _lead_stats(ld): | |
| fin = np.isfinite(ld) | |
| late = fin & (ld < 0) | |
| return dict(detected=float(fin.mean()) if len(ld) else np.nan, lead0_mean=float(np.mean(np.where(fin, ld, 0.0))) if len(ld) else np.nan, | |
| lead0_median=float(np.median(np.where(fin, ld, 0.0))) if len(ld) else np.nan, lead_detected_median=float(np.nanmedian(ld)) if fin.any() else np.nan, | |
| tta_late_median=float(np.nanmedian(-ld[late])) if late.any() else np.nan, n_late=int(late.sum())) | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--seeds", nargs="+", type=int, default=[0, 1, 2]) | |
| ap.add_argument("--run-family", default="I1_all_minus_road2drv_s{seed}_final") | |
| ap.add_argument("--budgets", nargs="+", type=float, default=[5.0, 10.0, 20.0]) | |
| ap.add_argument("--deltas", nargs="+", type=float, default=[0.5, 1.0, 2.0]) | |
| ap.add_argument("--n-boot", type=int, default=1000) | |
| ap.add_argument("--extra-scores", nargs="*", default=[], help="name=W_DIR:COLUMN (W_DIR under results/runs; {seed} expands); hook for fs_* future-state scores") | |
| ap.add_argument("--fs-dirs", nargs="*", default=[], help="layout dirs whose fs_* columns are added as scores ({seed} expands)") | |
| ap.add_argument("--headline-budget", type=float, default=10.0) | |
| ap.add_argument("--thr-fits", nargs="+", default=["val", "test"], help="threshold fitting splits: val = PRIMARY (deployable), test = SECONDARY equal achieved test load (operating_points.py convention)") | |
| ap.add_argument("--families", nargs="+", default=["A", "B"], help=f"event families to evaluate, in order (first = headline): {list(FAMILY_SPEC)}") | |
| ap.add_argument("--fs-dwell-dirs", nargs="*", default=[], help="WP4 future-state pack dirs under results/runs ({seed} expands), e.g. FS_dwell_s{seed}: adds the fs_* declared model scores") | |
| ap.add_argument("--road-dyn-dirs", nargs="*", default=[], help="B2-lane road-dynamics run dirs under results/runs ({seed} expands), e.g. RD_dyn_s{seed}: adds rd_ttc_min_3s (+ _xU)") | |
| ap.add_argument("--overlap-dirs", nargs="*", default=[], help="F8 overlap-head run dirs under results/runs ({seed} expands): adds ov_J when pred_overlap_{regime}_{split}.npz exists; " | |
| "`name=DIR` adds the arm as ov_<name> (2026-09-17 F8 lane: joint / prod / margonly in one run)") | |
| ap.add_argument("--road-rules", action="store_true", help="add the label-free road-rule baselines (margin rule, hard-family margin rule, constant-velocity TTC extrapolation; plain and × U)") | |
| ap.add_argument("--phead-runs", nargs="*", default=[], help="name=RUN_DIR ({seed} expands), e.g. ecan=I1_all_minus_road2drv_s{seed}_final_ecan main=I1_all_minus_road2drv_s{seed}_final: the run's learned " | |
| "demand-onset forecast as road scores ph_<name>_P_hard3s (+ _xU) and ph_<name>_D_ttc3s_xU (2026-09-17, owner decision (7))") | |
| ap.add_argument("--out-table", default=str(CWM3 / "results" / "tables" / "alert_timing.md")) | |
| ap.add_argument("--scratch", default="", help="smoke runs: write every output (run dir, table, json) under this directory instead of results/") | |
| a = ap.parse_args() | |
| t_start = time.time() | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| tables_dir = Path(a.scratch) if a.scratch else CWM3 / "results" / "tables" | |
| if a.scratch: | |
| a.out_table = str(Path(a.scratch) / "alert_timing.md") | |
| out = (Path(a.scratch) if a.scratch else CWM3 / "results" / "alert_timing") / stamp | |
| out.mkdir(parents=True, exist_ok=True) | |
| log_p = out / "alert_timing.log" | |
| def log(m): | |
| line = f"[{time.strftime('%H:%M:%S')}] {m}" | |
| print(line, flush=True) | |
| with open(log_p, "a") as f: | |
| f.write(line + "\n") | |
| for f_ in a.families: | |
| assert f_ in FAMILY_SPEC, f"unknown family {f_}; known {list(FAMILY_SPEC)}" | |
| FAMS = {f_: FAMILY_SPEC[f_][1] for f_ in a.families} | |
| fam0 = a.families[0] | |
| grids = {(r, s): load_grid(r, s) for r in REGIMES for s in ("val", "test")} | |
| # ---- events on TEST (and VAL for reference), every requested family | |
| EV = {} | |
| for (r, s), g in grids.items(): | |
| for fam, key in FAMS.items(): | |
| ev, n_cens = build_events(g, fam) | |
| ev["regime"], ev["split"], ev["family"] = r, s, fam | |
| EV[(r, s, fam)] = (ev, n_cens) | |
| log(f"{r}/{s} family {fam}: {len(g['t'])} rows / {g['hours']:.2f} h; U defined {g['U_defined'].mean():.3f}, U share {np.mean(g['U'] > 0.5):.3f}, D share {np.mean(g['D'] > 0.5):.4f}, " | |
| f"D_hard share {np.mean(g['D_hard'] > 0.5):.4f}, cond share {np.mean(g[key]):.4f}; events {len(ev)} ({len(ev) / g['hours']:.1f} / h; overlap {int((ev.trigger == 'overlap').sum()) if len(ev) else 0} / " | |
| f"dwell {int((ev.trigger == 'dwell').sum()) if len(ev) else 0}; hands_off at onset {int(ev.hands_off.sum()) if len(ev) else 0}, eyes_off {int(ev.eyes_off.sum()) if len(ev) else 0}; " | |
| f"recovery observed {int(np.isfinite(ev.t_recovery).sum()) if len(ev) else 0}); censored WC segment starts {n_cens}") | |
| if s == "test": | |
| ev.to_csv(out / f"events_{fam}_{r}.csv", index=False) | |
| ev_test = {(r, fam): EV[(r, "test", fam)][0] for r in REGIMES for fam in FAMS} | |
| ev_all = {fam: pd.concat([ev_test[(r, fam)] for r in REGIMES], ignore_index=True) for fam in FAMS} | |
| hours_test = {r: grids[(r, "test")]["hours"] for r in REGIMES} | |
| # ---- scores per seed | |
| metrics, paired_rows, thr_rec, human_rows = [], [], {}, [] | |
| fs_status_all = {} | |
| for seed in a.seeds: | |
| run = a.run_family.format(seed=seed) | |
| specs = {"surrogate_att3s": ("surrogate", None), "cur_dwfc": (f"W_{run}_dwfc", "W_learned_curhgb_t15_w10_n979dwfc"), "both_dwfc": (f"W_{run}_dwfcboth", "W_learned_bothhgb_t15_w10_n979dwfc"), | |
| "both_base": (f"W_{run}_merged", "W_learned_bothhgb_t15_w10_n979"), "persistence": ("rule", None), "dwell_timer": ("rule", None)} | |
| for spec in a.extra_scores: | |
| name, rest = spec.split("=", 1) | |
| d, c = rest.split(":", 1) | |
| specs[name] = (d.format(seed=seed), c) | |
| for d in a.fs_dirs: | |
| d = d.format(seed=seed) | |
| p = RUNS / d / "pred_layout_manual_test.npz" | |
| if p.exists(): | |
| with np.load(p, allow_pickle=True) as z: | |
| for c in [str(x) for x in z["w_names"]]: | |
| if c.startswith("fs_"): | |
| specs[f"{d}::{c}"] = (d, c) | |
| for d in a.fs_dwell_dirs: | |
| d = d.format(seed=seed) | |
| for c in FS_SCORES: | |
| specs[c] = (f"fsdwell:{d}", c) | |
| if a.road_rules: | |
| for c in ROAD_RULES: | |
| specs[c] = ("rule", None) | |
| for d in a.road_dyn_dirs: | |
| d = d.format(seed=seed) | |
| for c in RD_SCORES: | |
| specs[c] = (f"rd:{d}", c) | |
| for d in a.overlap_dirs: | |
| nm, d = overlap_spec(d.format(seed=seed)) # F8 lane 2026-09-17T19:32:07Z: `name=DIR` -> score ov_<name> (several arms in one run); bare DIR -> ov_J as before | |
| specs[nm] = (f"ov:{d}", nm) | |
| for spec in a.phead_runs: | |
| nm, d = spec.split("=", 1) | |
| for kd in PH_KINDS: | |
| specs[f"ph_{nm}_{kd}"] = (f"ph:{d.format(seed=seed)}", kd) | |
| scores = {} | |
| fs_status = {} | |
| for name, (d, c) in specs.items(): | |
| ok = True | |
| for (r, s), g in grids.items(): | |
| if d == "surrogate": | |
| v, src = surrogate_column(r, s, seed if s == "test" else 0, g) | |
| if v is None: | |
| log(f" {name} {r}/{s}: {src} missing"); ok = False; break | |
| elif d.startswith("fsdwell:"): | |
| v, src = fs_dwell_column(d.split(":", 1)[1], r, s, c, g) | |
| if v is None: | |
| log(f" {name} {r}/{s}: {src} -> fs score not available for seed {seed}"); fs_status[name] = src; ok = False; break | |
| fs_status[name] = "compatible (row-aligned with the grid)" | |
| elif d == "rule" and name in ROAD_RULES: | |
| v = score_road_rule(name, g["M0"], g["M0valid"], g["U"]) | |
| elif d == "rule": | |
| v = (score_persistence if name == "persistence" else score_dwell_timer)(g["U"], g["D"], g["moving"], g["dwell"]) | |
| v = np.where(g["U_defined"] | (g["U"] > 0.5), v, np.nan) | |
| elif d.startswith("rd:"): | |
| v, src = rd_column(d.split(":", 1)[1], r, s, c, g) | |
| if v is None: | |
| log(f" {name} {r}/{s}: {src} -> road-dyn score not available for seed {seed}"); fs_status[name] = src; ok = False; break | |
| fs_status[name] = "compatible (row-aligned with the grid)" | |
| elif d.startswith("ov:"): | |
| v, src = overlap_column(d.split(":", 1)[1], r, s, g) | |
| if v is None: | |
| log(f" {name} {r}/{s}: {src}"); fs_status[name] = src; ok = False; break | |
| fs_status[name] = "compatible (row-aligned with the grid)" | |
| elif d.startswith("ph:"): | |
| v, src = phead_column(d.split(":", 1)[1], r, s, c, g) | |
| if v is None: | |
| log(f" {name} {r}/{s}: {src} -> demand-forecast score not available for seed {seed}"); fs_status[name] = src; ok = False; break | |
| fs_status[name] = f"compatible (row-aligned with the grid; {src})" | |
| else: | |
| p = RUNS / d / f"pred_layout_{r}_{s}.npz" | |
| if not p.exists(): | |
| log(f" {name} {r}/{s}: {p} missing -> score skipped for seed {seed}"); ok = False; break | |
| v = layout_column(d, r, s, c, g) | |
| scores[(name, r, s)] = v | |
| if not ok: | |
| for k in [k for k in scores if k[0] == name]: | |
| scores.pop(k) | |
| names = sorted({k[0] for k in scores}, key=lambda n: list(specs).index(n) if n in specs else 99) | |
| for k_, v_ in fs_status.items(): | |
| fs_status_all[f"s{seed}:{k_}"] = v_ | |
| log(f"seed {seed}: scores {names}; fs status {fs_status}") | |
| # thresholds per regime × budget: fitted on VAL (primary, deployable) and on TEST (secondary: equal ACHIEVED test load, the operating_points.py convention; not deployable) | |
| thr = {} | |
| for fit in a.thr_fits: | |
| for name in names: | |
| for r in REGIMES: | |
| g = grids[(r, fit)] | |
| for B in a.budgets: | |
| th, ach, reach, floor = thr_for_rate(g["uid"], g["t"], scores[(name, r, fit)], B) | |
| thr[(fit, name, r, B)] = (th, ach, reach, floor) | |
| thr_rec[f"s{seed}|{fit}|{name}|{r}|{B:g}"] = {"thr": th, "achieved_fit": ach, "reachable": bool(reach), "floor_fit": floor, "fit_split": fit, "seed_val_marginals": 0 if name == "surrogate_att3s" else seed} | |
| # alerts on TEST (family-independent), matching against both families | |
| per_event = {} # (fit, fam, name, B) -> lead array over ev_all[fam] | |
| for fit in a.thr_fits: | |
| for name in names: | |
| for B in a.budgets: | |
| leads = {fam: [] for fam in FAMS} | |
| tot = {fam: dict(fa=0, nal=0, early=0.0) for fam in FAMS} | |
| for r in REGIMES: | |
| g = grids[(r, "test")] | |
| th = thr[(fit, name, r, B)][0] | |
| ar = alert_rows(g["uid"], g["t"], scores[(name, r, "test")], th) | |
| for fam in FAMS: | |
| ev = ev_test[(r, fam)] | |
| ld, just = match_events(ev, g["uid"], g["t"], ar) | |
| es = alert_early_share(ev, g["uid"], g["t"], ar, 1.0) | |
| leads[fam].append(ld); tot[fam]["fa"] += int((~just).sum()); tot[fam]["nal"] += len(ar); tot[fam]["early"] += (es * len(ar) if np.isfinite(es) else 0.0) | |
| metrics.append(dict(seed=seed, thr_fit=fit, family=fam, score=name, budget=B, regime=r, thr=th, reachable=thr[(fit, name, r, B)][2], floor_val=thr[(fit, name, r, B)][3], achieved_val=thr[(fit, name, r, B)][1], | |
| alerts_test=len(ar), alerts_per_h_test=len(ar) / g["hours"], fa_per_h_test=int((~just).sum()) / g["hours"], early_share_1s=es, hours_test=g["hours"], | |
| n_events=len(ld), events_per_h=len(ld) / g["hours"], recall_ceiling=min(1.0, B * g["hours"] / max(len(ld), 1)), **_recall_row(ld, a.deltas), **_lead_stats(ld))) | |
| for fam in FAMS: | |
| ld = np.concatenate(leads[fam]) | |
| per_event[(fit, fam, name, B)] = ld | |
| H = sum(hours_test.values()) | |
| metrics.append(dict(seed=seed, thr_fit=fit, family=fam, score=name, budget=B, regime="pooled", thr=np.nan, reachable=all(thr[(fit, name, r, B)][2] for r in REGIMES), floor_val=np.nan, achieved_val=np.nan, | |
| alerts_test=tot[fam]["nal"], alerts_per_h_test=tot[fam]["nal"] / H, fa_per_h_test=tot[fam]["fa"] / H, early_share_1s=tot[fam]["early"] / max(tot[fam]["nal"], 1), hours_test=H, | |
| n_events=len(ld), events_per_h=len(ld) / H, recall_ceiling=min(1.0, B * H / max(len(ld), 1)), **_recall_row(ld, a.deltas), **_lead_stats(ld))) | |
| # surrogate-silent hard subset (no surrogate alert in the match window at the same budget) | |
| for fit in a.thr_fits: | |
| for fam in FAMS: | |
| for B in a.budgets: | |
| if (fit, fam, "surrogate_att3s", B) not in per_event: | |
| continue | |
| silent = ~np.isfinite(per_event[(fit, fam, "surrogate_att3s", B)]) | |
| for name in names: | |
| ld = per_event[(fit, fam, name, B)] | |
| for r in list(REGIMES) + ["pooled"]: | |
| m = silent & ((ev_all[fam].regime == r).to_numpy() if r != "pooled" else True) | |
| if m.sum() == 0: | |
| continue | |
| metrics.append(dict(seed=seed, thr_fit=fit, family=fam, score=name, budget=B, regime=f"{r}|surrogate_silent", n_events=int(m.sum()), **_recall_row(ld[m], a.deltas), **_lead_stats(ld[m]))) | |
| # bootstrap: device clusters over events, paired differences | |
| for fit in a.thr_fits: | |
| for fam in FAMS: | |
| for B in a.budgets: | |
| cols = [(name, d) for name in names for d in a.deltas] | |
| ind = np.stack([(np.nan_to_num(per_event[(fit, fam, name, B)], nan=-99) >= d).astype(float) for name, d in cols], 1) | |
| for r in list(REGIMES) + ["pooled"]: | |
| m = (ev_all[fam].regime == r).to_numpy() if r != "pooled" else np.ones(len(ev_all[fam]), bool) | |
| if m.sum() < 5: | |
| continue | |
| bs = cluster_boot_events(ev_all[fam].dongle.to_numpy()[m], ind[m], a.n_boot, 0) | |
| for j, (name, d) in enumerate(cols): | |
| lo, hi = pct(bs[:, j]) | |
| for row in metrics: | |
| if row["seed"] == seed and row["thr_fit"] == fit and row["family"] == fam and row["score"] == name and row["budget"] == B and row["regime"] == r: | |
| row[f"recall_early_{d:g}s_ci_lo"], row[f"recall_early_{d:g}s_ci_hi"] = lo, hi | |
| for na in names: | |
| for nb in names: | |
| if na == nb: | |
| continue | |
| for d in a.deltas: | |
| ja, jb = cols.index((na, d)), cols.index((nb, d)) | |
| dd = bs[:, ja] - bs[:, jb] | |
| lo, hi = pct(dd) | |
| paired_rows.append(dict(seed=seed, thr_fit=fit, family=fam, budget=B, regime=r, delta_s=d, a=na, b=nb, value_a=float(ind[m, ja].mean()), value_b=float(ind[m, jb].mean()), | |
| diff=float(ind[m, ja].mean() - ind[m, jb].mean()), ci_lo=lo, ci_hi=hi, frac_boot_gt0=float(np.mean(dd > 0)), n_events=int(m.sum()))) | |
| # human confirmation set at the headline budget (family A onsets for the nearest-automatic-onset column) | |
| if HUMAN.exists(): | |
| recs = [json.loads(l) for l in HUMAN.read_text().splitlines() if l.strip()] | |
| evr = [x for x in recs if x.get("stage") == "event"] | |
| n_usable = 0 | |
| for x in evr: | |
| f = x.get("fields", {}) | |
| et = f.get("event_type") or [] | |
| on = f.get("event_onset_s") | |
| usable = bool(et) and et != ["none"] and isinstance(on, (int, float)) and x.get("valid", {}).get("event_onset_s", False) | |
| r = x.get("regime") | |
| row = dict(seed=seed, idx=x.get("idx"), regime=r, uid=x.get("uid"), t0=x.get("t0"), event_type="+".join(et), event_onset_s=on, usable=usable, | |
| first_hand_to_wheel_s=f.get("first_hand_to_wheel_s"), first_gaze_to_road_s=f.get("first_gaze_to_road_s")) | |
| if usable and r in REGIMES: | |
| n_usable += 1 | |
| o_h = float(x["t0"]) - float(x.get("anchor_s", 5)) + float(on) | |
| row["onset_abs_s"] = o_h | |
| g = grids[(r, "test")] | |
| ev = ev_test[(r, fam0)] | |
| e_route = ev[ev.uid == x.get("uid")] | |
| if len(e_route): | |
| k = int(np.argmin(np.abs(e_route.onset_t.to_numpy() - o_h))) | |
| row["nearest_auto_onset_dt_s"] = float(e_route.onset_t.iloc[k] - o_h) | |
| row["nearest_auto_trigger"] = e_route.trigger.iloc[k] | |
| row["nearest_auto_recovery_abs_s"] = float(e_route.t_recovery.iloc[k]) | |
| for hk in ("first_hand_to_wheel_s", "first_gaze_to_road_s"): | |
| v = f.get(hk) | |
| row[hk + "_abs"] = float(x["t0"]) - float(x.get("anchor_s", 5)) + float(v) if isinstance(v, (int, float)) else np.nan | |
| rows_route = np.flatnonzero(g["uid"] == x.get("uid")) | |
| for name in names: | |
| th = thr[("val", name, r, a.headline_budget)][0] | |
| if len(rows_route) == 0: | |
| row[f"lead_{name}"] = np.nan | |
| continue | |
| sub = rows_route | |
| ar = alert_rows(g["uid"][sub], g["t"][sub], scores[(name, r, "test")][sub], th) | |
| ta = g["t"][sub][ar] | |
| hit = ta[(ta >= o_h - MATCH_PRE) & (ta <= o_h + MATCH_POST)] | |
| row[f"lead_{name}"] = float(o_h - hit.min()) if len(hit) else np.nan | |
| human_rows.append(row) | |
| log(f"seed {seed}: human event records {len(evr)}, usable (event != none with a valid onset) {n_usable}") | |
| M = pd.DataFrame(metrics); P = pd.DataFrame(paired_rows); Hm = pd.DataFrame(human_rows) | |
| M.to_csv(out / "metrics.csv", index=False); P.to_csv(out / "paired.csv", index=False); Hm.to_csv(out / "human_confirmation.csv", index=False) | |
| (out / "thresholds.json").write_text(json.dumps(thr_rec, indent=1)) | |
| names = list(dict.fromkeys(M.score)) | |
| seeds = sorted(int(x) for x in M.seed.unique()) | |
| M_all, P_all = M, P | |
| M, P = M_all[M_all.thr_fit == "val"], P_all[P_all.thr_fit == "val"] | |
| # ---- table | |
| T = [f"## Event-level alert timing on the continuous TEST exposure — seeds {seeds}; {stamp}; eval/alert_timing_eval.py", "", | |
| "**Definitions (pre-registered 2026-09-16T06:46:35Z; every number below is script output).** Rows = every anchor of both regimes' TEST splits (5 Hz s12 grid; exposure = rows × 0.2 s), " | |
| "NOT filtered by surrogate silence. State at t: hands_on / head_turned = causal pose states `S0_c`, gaze_off = own-landmark state `S0_g`; unmet U_t = hands_off ∨ head_turned ∨ gaze_off " | |
| "(invalid component = met; U forward-filled ≤ 1 s inside a continuous segment = consecutive anchors ≤ 1 s apart). Road demand D_t = any road-pressure family high at t (`P0 ∧ P0valid`: the CAN margins " | |
| "M0 of data/build_pressure_targets.py against its THR — ttc < 4 s, ttc drop, cut-in, lead braking, lane excursion, badas; moving > 2 m/s); D_hard = the kinematic-conflict families only " | |
| f"({', '.join(HARD_FAMILIES)}). State entry = U 0→1; demand onset = D 0→1; **event families in this section**: {'; '.join(FAMILY_LABEL[f_] for f_ in FAMS)}. Event onset o = WC 0→1 inside a segment (families A / B) or the anchor at which the sustained condition first reaches its duration (C families; runs starting at a segment start are censored), onsets within " | |
| f"{EVENT_MERGE_S:g} s of the previous kept onset merged (the S1 same-event guard); recovery = first anchor ≥ o with U = 0 (censored at the segment end). Alerts = score ≥ threshold with the existing " | |
| f"{COOLDOWN:g} s per-route cooldown; thresholds per score × regime fitted on the VAL split by bisection to {', '.join(f'{b:g}' for b in a.budgets)} alerts / exposure hour; a score whose floor " | |
| "(alerts/h at threshold = max score) exceeds the budget is reported AT ITS FLOOR (flag `floor`). Match window [o − 10, o + 3] s; lead = o − first alert (negative = late); " | |
| f"**PRIMARY Recall_{{≥ δ s early}}, δ ∈ {{{', '.join(f'{d:g}' for d in a.deltas)}}}, family A**; secondary: family B, detected share, lead distribution (undetected = 0), TTA of late detections, achieved alerts/h, " | |
| "FA/h (alerts outside [o − 10, max(o + 3, recovery)] of every event of the route), early share = share of alerts that precede some onset by ≥ 1 s (≤ 10 s). Sections 1–5 use the VAL-fitted " | |
| "thresholds (PRIMARY, deployable; the learned heads then reach a LOWER test load than the budget — achieved alerts/h is printed next to every recall); section 6 repeats the comparison at equal " | |
| "ACHIEVED test load (SECONDARY). **Recall ceiling** = min(1, budget × hours / n events): " | |
| "with tens of events per hour a 5–20 / h budget bounds every recall — read the PAIRED differences, not the levels. Device-cluster bootstrap (devices with replacement, thresholds fixed) with paired differences " | |
| "on identical resamples. Scores: `surrogate_att3s` = openpilot-derived surrogate (legacy attention marginals → att_only, τ 15; val marginals exist for seed 0 only → s0 val thresholds for every seed); " | |
| "`cur_dwfc` = current-state head (cur+dwell+face, HGB); `both_dwfc` = same + the model's predicted future; `both_base` = human-programme system A (context); `persistence` = current state assumed to persist " | |
| "(1.0 = WC now, 0.5 = unmet without demand); `dwell_timer` = time-to-threshold extrapolation of the current dwell (+ demand); `fs_*` = declared model scores of the WP4 future-state head " | |
| "(--fs-dwell-dirs: fs_occ_any_1s / fs_occ_any_3s = 1 − Π_s(1 − pi_off) at +1 / +3 s, fs_EDoff_5s = Σ_s expected off-task seconds over 5 s; causal by that trainer's construction, NOT covered by the rule-score " | |
| f"leakage test). Road-rule baselines (--road-rules, label-free, t only): `road_mr_max` = margin_rule.score (max threshold-normalised current margin ratio over ttc / headway / lead_decel / lateral / badas), " | |
| "`road_mr_hard` = the same over ttc / headway / lead_decel, `cv_ttc_3s` = constant-velocity TTC extrapolation 1 / max(min_h ttc_cv(h), 0.1), ttc_cv(h) = max(dRel + vRel·h, 0) / max(−vRel, 0.1), h = 1, 2, 3 s " | |
| "(0 without a valid closing lead; under constant velocity ttc_cv(h) = ttc_now − h, so its RANKING equals the current-TTC rule's); `*_xU` = rule × 1[U_t]. Hooks: `rd_ttc_min_3s` = the B2 road-dynamics " | |
| "forecast (pred_rd mu = CV + residual) as 1 / min predicted TTC over +1 / +2 / +3 s (+ _xU); `ov_J` = F8 overlap head (none exists); `ph_<run>_P_hard3s` (--phead-runs, 2026-09-17) = a joint run's " | |
| "learned DEMAND-ONSET forecast: P(any hard P family ttc / ttc_drop / cut_in / lead_braking first switches high within 3 s) = closed-form first passage of its P-head onset hazards over 15 steps, " | |
| "`_xU` = × 1[U_t], `ph_<run>_D_ttc3s_xU` = the same over the legacy D chains ttc / ttc_drop. " | |
| f"Hook status: {fs_status_all if fs_status_all else 'no fs / rd / ov dirs requested'}. Rule scores are seed-invariant. Not filtered by any human label; the " | |
| "head_turned / gaze_off states are percentile-defined automatic states (gaze validation pending), so U-based events are frequent and include ordinary glances. **Lead margin available** = " | |
| "onset − entry of the U run (the earliness that is physically possible for an alert keyed to the state entry).", ""] | |
| for fam in FAMS: | |
| for r in REGIMES: | |
| ev, n_cens = EV[(r, "test", fam)] | |
| T.append(f"- **family {fam}, {r} TEST**: {len(grids[(r, 'test')]['t'])} rows / {hours_test[r]:.2f} h / {ev.dongle.nunique() if len(ev) else 0} devices with events; **{len(ev)} events = {len(ev) / hours_test[r]:.1f} / h** " | |
| f"(recall ceiling at {', '.join(f'{B:g}/h: {min(1.0, B * hours_test[r] / max(len(ev), 1)):.2f}' for B in a.budgets)}); overlap-triggered {int((ev.trigger == 'overlap').sum())}, dwell-triggered {int((ev.trigger == 'dwell').sum())}; " | |
| f"hands_off at onset {int(ev.hands_off.sum())}, eyes_off {int(ev.eyes_off.sum())}; recovery observed {int(np.isfinite(ev.t_recovery).sum())}, median time to recovery {np.nanmedian(ev.t_recovery - ev.onset_t) if len(ev) else float('nan'):.1f} s; " | |
| f"**lead margin available** median {np.median(ev.lead_margin_s) if len(ev) else float('nan'):.1f} s [IQR {np.percentile(ev.lead_margin_s, 25) if len(ev) else float('nan'):.1f}, {np.percentile(ev.lead_margin_s, 75) if len(ev) else float('nan'):.1f}], " | |
| f"share ≥ 0.5 / 1 / 2 s = {np.mean(ev.lead_margin_s >= 0.5) if len(ev) else float('nan'):.2f} / {np.mean(ev.lead_margin_s >= 1.0) if len(ev) else float('nan'):.2f} / {np.mean(ev.lead_margin_s >= 2.0) if len(ev) else float('nan'):.2f}; " | |
| f"censored segment starts / excluded {n_cens}; entry-censored kept (lead margin = lower bound) {int(ev.entry_censored.sum()) if len(ev) else 0}; VAL: {len(EV[(r, 'val', fam)][0])} events / {grids[(r, 'val')]['hours']:.2f} h.") | |
| for fam in FAMS: | |
| T += ["", f"### 1.{list(FAMS).index(fam) + 1} {'PRIMARY' if fam in ('A', fam0) else 'secondary'} — family {fam}: Recall_{{≥ δ s early}} at each budget (val-fitted thresholds; per seed; [CI] = device-cluster 95 %; `floor` = budget unreachable, reported at the floor)", ""] | |
| for B in a.budgets: | |
| T += [f"#### Family {fam}, budget {B:g} alerts / h", "", "| score | regime | seed | n events (ceiling) | alerts/h test | FA/h | early share ≥1 s | detected | " + " | ".join(f"Recall ≥ {d:g} s [CI]" for d in a.deltas) + " | lead0 mean / median s | TTA late median s (n) |", | |
| "|---|---|---|---|---|---|---|---|" + "---|" * len(a.deltas) + "---|---|"] | |
| for name in names: | |
| for r in list(REGIMES) + ["pooled"]: | |
| g = M[(M.family == fam) & (M.score == name) & (M.budget == B) & (M.regime == r)].sort_values("seed") | |
| for row in g.to_dict("records"): | |
| flag = "" if bool(row["reachable"]) else " `floor`" | |
| cells = [f"{f3(row.get(f'recall_early_{d:g}s'))} [{f3(row.get(f'recall_early_{d:g}s_ci_lo', np.nan), 2)},{f3(row.get(f'recall_early_{d:g}s_ci_hi', np.nan), 2)}]" for d in a.deltas] | |
| T.append(f"| {name}{flag} | {r} | s{row['seed']} | {int(row['n_events'])} ({f3(row['recall_ceiling'], 2)}) | {f3(row['alerts_per_h_test'], 1)} | {f3(row['fa_per_h_test'], 1)} | {f3(row['early_share_1s'], 2)} | {f3(row['detected'])} | " + " | ".join(cells) + | |
| f" | {f3(row['lead0_mean'], 2)} / {f3(row['lead0_median'], 2)} | {f3(row['tta_late_median'], 2)} ({int(row['n_late'])}) |") | |
| T.append("") | |
| T += [f"### 2. Seed summary at {a.headline_budget:g} / h — Recall ≥ 1 s early, mean ± sd over seeds (rule scores and the surrogate's thresholds are seed-invariant)", "", | |
| "| family | score | " + " | ".join(f"{r} mean ± sd" for r in list(REGIMES) + ["pooled"]) + " |", "|---|---|---|---|---|"] | |
| for fam in FAMS: | |
| for name in names: | |
| cells = [] | |
| for r in list(REGIMES) + ["pooled"]: | |
| g = M[(M.family == fam) & (M.score == name) & (M.budget == a.headline_budget) & (M.regime == r)] | |
| v = g["recall_early_1s"].to_numpy(float) if "recall_early_1s" in g else np.array([]) | |
| cells.append(f"{v.mean():.3f} ± {v.std(ddof=1) if len(v) > 1 else 0:.3f}" if len(v) else "—") | |
| T.append(f"| {fam} | {name} | " + " | ".join(cells) + " |") | |
| T += ["", "### 3. Paired differences (Recall ≥ δ s early, identical device-cluster resamples) — decisive controls", "", | |
| "| family | A − B | budget | regime | δ | " + " | ".join(f"s{s} Δ [CI]" for s in seeds) + " | mean Δ | seeds CI excl. 0 |", "|---|---|---|---|---|" + "---|" * len(seeds) + "---|---|"] | |
| ctrl = [("both_dwfc", "cur_dwfc"), ("cur_dwfc", "persistence"), ("both_dwfc", "persistence"), ("cur_dwfc", "dwell_timer"), ("both_dwfc", "dwell_timer"), ("both_dwfc", "surrogate_att3s"), | |
| ("cur_dwfc", "surrogate_att3s"), ("both_base", "cur_dwfc"), ("dwell_timer", "persistence")] | |
| for fsn in [n for n in names if n.startswith("fs_")]: | |
| ctrl += [(fsn, "cur_dwfc"), (fsn, "both_dwfc"), (fsn, "persistence"), (fsn, "dwell_timer"), (fsn, "surrogate_att3s")] | |
| if any(n in names for n in ROAD_RULES): | |
| for L in [n for n in names if n in ("cur_dwfc", "both_dwfc") or n.startswith(("fs_", "rd_", "ov_"))]: | |
| ctrl += [(L, "road_mr_hard_xU"), (L, "cv_ttc_3s_xU"), (L, "road_mr_max_xU")] | |
| ctrl += [("road_mr_hard_xU", "road_mr_hard"), ("cv_ttc_3s_xU", "cv_ttc_3s"), ("cv_ttc_3s_xU", "road_mr_hard_xU"), ("road_mr_hard_xU", "persistence"), ("road_mr_hard_xU", "dwell_timer"), | |
| ("road_mr_hard_xU", "surrogate_att3s")] | |
| if "rd_ttc_min_3s" in names: | |
| ctrl += [("rd_ttc_min_3s", "cv_ttc_3s"), ("rd_ttc_min_3s_xU", "cv_ttc_3s_xU"), ("rd_ttc_min_3s_xU", "road_mr_hard_xU")] | |
| ph_names = [n for n in names if n.startswith("ph_")] | |
| for L in ph_names: # learned demand forecasts vs the physics rules, the driver-side scores and the surrogate | |
| ctrl += [(L, "cv_ttc_3s_xU"), (L, "road_mr_hard_xU"), (L, "cv_ttc_3s"), (L, "rd_ttc_min_3s_xU"), (L, "cur_dwfc"), (L, "surrogate_att3s"), (L, "persistence"), (L, "dwell_timer")] | |
| if L.endswith("_xU") and L[:-3] in names: | |
| ctrl.append((L, L[:-3])) | |
| for kd in PH_KINDS: # learned vs learned (e.g. ECAN vs the frozen main), same score kind, in --phead-runs order | |
| ks = [n for n in ph_names if n.endswith(f"_{kd}")] | |
| for i in range(len(ks)): | |
| for j in range(i + 1, len(ks)): | |
| ctrl.append((ks[i], ks[j])) | |
| ov_names = [n for n in names if n.startswith("ov_")] # F8 lane (2026-09-17T19:32:07Z): every overlap arm vs the read-outs / rules / surrogate, and the arms against each other | |
| for L in ov_names: | |
| ctrl += [(L, "cur_dwfc"), (L, "both_dwfc"), (L, "persistence"), (L, "dwell_timer"), (L, "surrogate_att3s")] | |
| for i in range(len(ov_names)): | |
| for j in range(i + 1, len(ov_names)): | |
| ctrl.append((ov_names[i], ov_names[j])) | |
| ctrl = [c for c in ctrl if c[0] in names and c[1] in names] | |
| for fam in FAMS: | |
| for na, nb in ctrl: | |
| for B in a.budgets: | |
| for r in list(REGIMES) + ["pooled"]: | |
| for d in a.deltas: | |
| g = P[(P.family == fam) & (P.a == na) & (P.b == nb) & (P.budget == B) & (P.regime == r) & (P.delta_s == d)].sort_values("seed") | |
| if not len(g): | |
| continue | |
| excl = int(((g.ci_lo > 0) | (g.ci_hi < 0)).sum()) | |
| T.append(f"| {fam} | {na} − {nb} | {B:g} | {r} | {d:g} | " + " | ".join(f"{x['diff']:+.3f} [{x['ci_lo']:+.3f},{x['ci_hi']:+.3f}]" for x in g.to_dict("records")) + f" | {g['diff'].mean():+.3f} | {excl}/{len(g)} |") | |
| T += ["", "### 4. Surrogate-silent hard subset (events with NO surrogate alert in the match window at the same budget) — Recall ≥ δ s early", "", | |
| "| family | score | budget | regime | seed | n events | detected | " + " | ".join(f"Recall ≥ {d:g} s" for d in a.deltas) + " |", "|---|---|---|---|---|---|---|" + "---|" * len(a.deltas)] | |
| for fam in FAMS: | |
| for B in a.budgets: | |
| for name in names: | |
| g = M[(M.family == fam) & (M.score == name) & (M.budget == B) & (M.regime.astype(str).str.endswith("surrogate_silent"))].sort_values(["regime", "seed"]) | |
| for row in g.to_dict("records"): | |
| T.append(f"| {fam} | {name} | {B:g} | {row['regime']} | s{row['seed']} | {int(row['n_events'])} | {f3(row['detected'])} | " + " | ".join(f3(row.get(f'recall_early_{d:g}s')) for d in a.deltas) + " |") | |
| T += ["", f"### 5. Human confirmation set — `labels_newtest.jsonl` stage=event records; usable = event_type ≠ none with a valid onset; leads at the {a.headline_budget:g} / h val thresholds (seed 0 rows shown; n is small — confirmation only; nearest automatic onset from family {fam0})", ""] | |
| if len(Hm): | |
| H0 = Hm[Hm.seed == seeds[0]] | |
| T.append(f"Records {len(H0)}, usable {int(H0.usable.sum())} (types: {H0[H0.usable].event_type.value_counts().to_dict()}).") | |
| lead_cols = [c for c in H0.columns if c.startswith("lead_")] | |
| T += ["", "| idx | regime | type | onset abs s | nearest auto onset Δ s (trigger) | human first hand / gaze back (abs s) | auto recovery abs s | " + " | ".join(c.replace("lead_", "lead ") for c in lead_cols) + " |", | |
| "|---|---|---|---|---|---|---|" + "---|" * len(lead_cols)] | |
| for x in H0[H0.usable].to_dict("records"): | |
| T.append(f"| {x['idx']} | {x['regime']} | {x['event_type']} | {f3(x.get('onset_abs_s'), 1)} | {f3(x.get('nearest_auto_onset_dt_s', np.nan), 1)} ({x.get('nearest_auto_trigger', '—')}) | " | |
| f"{f3(x.get('first_hand_to_wheel_s_abs', np.nan), 1)} / {f3(x.get('first_gaze_to_road_s_abs', np.nan), 1)} | {f3(x.get('nearest_auto_recovery_abs_s', np.nan), 1)} | " + | |
| " | ".join(f3(x.get(c), 1) for c in lead_cols) + " |") | |
| n_auto = int(H0[H0.usable].nearest_auto_onset_dt_s.abs().le(5.0).sum()) if "nearest_auto_onset_dt_s" in H0 else 0 | |
| T.append(f"\nAutomatic (family {fam0}) onset within ±5 s of the human onset: {n_auto} / {int(H0.usable.sum())} usable records.") | |
| # ---- secondary: thresholds fitted on TEST (equal ACHIEVED test load; not deployable — separates load differences from timing differences) | |
| Mt, Pt = M_all[M_all.thr_fit == "test"], P_all[P_all.thr_fit == "test"] | |
| if len(Mt): | |
| T += ["", "### 6. SECONDARY — thresholds fitted on the TEST split itself (equal achieved test alerts/h per score × regime; the `operating_points.py` convention; NOT deployable, shown to separate " | |
| "load differences from timing differences because the val-fitted thresholds give the learned heads a lower test load than the budget)", "", | |
| "| family | score | budget | regime | seed | alerts/h test | FA/h | detected | " + " | ".join(f"Recall ≥ {d:g} s [CI]" for d in a.deltas) + " |", "|---|---|---|---|---|---|---|---|" + "---|" * len(a.deltas)] | |
| for fam in FAMS: | |
| for B in a.budgets: | |
| for name in names: | |
| for r in list(REGIMES) + ["pooled"]: | |
| g = Mt[(Mt.family == fam) & (Mt.score == name) & (Mt.budget == B) & (Mt.regime == r)].sort_values("seed") | |
| for row in g.to_dict("records"): | |
| flag = "" if bool(row["reachable"]) else " `floor`" | |
| cells = [f"{f3(row.get(f'recall_early_{d:g}s'))} [{f3(row.get(f'recall_early_{d:g}s_ci_lo', np.nan), 2)},{f3(row.get(f'recall_early_{d:g}s_ci_hi', np.nan), 2)}]" for d in a.deltas] | |
| T.append(f"| {fam} | {name}{flag} | {B:g} | {r} | s{row['seed']} | {f3(row['alerts_per_h_test'], 1)} | {f3(row['fa_per_h_test'], 1)} | {f3(row['detected'])} | " + " | ".join(cells) + " |") | |
| T += ["", "#### 6b. Paired differences at equal achieved TEST load (Recall ≥ δ s early)", "", | |
| "| family | A − B | budget | regime | δ | " + " | ".join(f"s{s} Δ [CI]" for s in seeds) + " | mean Δ | seeds CI excl. 0 |", "|---|---|---|---|---|" + "---|" * len(seeds) + "---|---|"] | |
| for fam in FAMS: | |
| for na, nb in ctrl: | |
| for B in a.budgets: | |
| for r in list(REGIMES) + ["pooled"]: | |
| for d in a.deltas: | |
| g = Pt[(Pt.family == fam) & (Pt.a == na) & (Pt.b == nb) & (Pt.budget == B) & (Pt.regime == r) & (Pt.delta_s == d)].sort_values("seed") | |
| if not len(g): | |
| continue | |
| excl = int(((g.ci_lo > 0) | (g.ci_hi < 0)).sum()) | |
| T.append(f"| {fam} | {na} − {nb} | {B:g} | {r} | {d:g} | " + " | ".join(f"{x['diff']:+.3f} [{x['ci_lo']:+.3f},{x['ci_hi']:+.3f}]" for x in g.to_dict("records")) + f" | {g['diff'].mean():+.3f} | {excl}/{len(g)} |") | |
| T += ["", "### 7. Thresholds (`thresholds.json` in the run dir; fit = val PRIMARY / test SECONDARY)", "", "| seed | fit | score | regime | budget | thr | achieved alerts/h on the fit split | reachable | floor alerts/h on the fit split |", "|---|---|---|---|---|---|---|---|---|"] | |
| for k, v in thr_rec.items(): | |
| s_, f_, n_, r_, b_ = k.split("|") | |
| T.append(f"| {s_} | {f_} | {n_} | {r_} | {b_} | {f3(v['thr'], 4)} | {f3(v['achieved_fit'], 1)} | {v['reachable']} | {f3(v['floor_fit'], 1)} |") | |
| T.append("") | |
| md = "\n".join(T) + "\n" | |
| op = Path(a.out_table) | |
| if not op.exists(): | |
| op.write_text("# Event-level alert timing on the continuous test exposure (append-only; eval/alert_timing_eval.py)\n\n") | |
| with open(op, "a") as f: | |
| f.write(md) | |
| print(md) | |
| headline = {} | |
| for fam in FAMS: | |
| for name in names: | |
| for r in list(REGIMES) + ["pooled"]: | |
| g = M[(M.family == fam) & (M.score == name) & (M.budget == a.headline_budget) & (M.regime == r)] | |
| headline[f"{fam}|{name}|{r}|val-fitted"] = {"recall_early_1s_by_seed": {int(k): float(v) for k, v in g.sort_values("seed").set_index("seed")["recall_early_1s"].items()} if "recall_early_1s" in g else {}, | |
| "n_events": int(g.n_events.iloc[0]) if len(g) else 0, "recall_ceiling": float(g.recall_ceiling.iloc[0]) if len(g) else np.nan} | |
| js = {"stamp": stamp, "seeds": a.seeds, "budgets": a.budgets, "deltas": a.deltas, "n_boot": a.n_boot, "families": {f_: FAMILY_LABEL[f_] for f_ in FAMS}, "family_spec": {f_: list(FAMILY_SPEC[f_]) for f_ in FAMS}, | |
| "hard_families": list(HARD_FAMILIES), "fs_status": fs_status_all, "fs_scores": [n for n in names if n.startswith("fs_")], | |
| "lead_margin": {f"{fam}|{r}": {"median_s": float(np.median(EV[(r, 'test', fam)][0].lead_margin_s)) if len(EV[(r, 'test', fam)][0]) else None, | |
| "share_ge_1s": float(np.mean(EV[(r, 'test', fam)][0].lead_margin_s >= 1.0)) if len(EV[(r, 'test', fam)][0]) else None} for fam in FAMS for r in REGIMES}, | |
| "definitions": {"dwell_s": DWELL_S, "event_merge_s": EVENT_MERGE_S, "seg_gap_s": SEG_GAP_S, "ffill_steps": FFILL_STEPS, "cooldown_s": COOLDOWN, "match_window_s": [-MATCH_PRE, MATCH_POST], "grid_dt": GRID_DT, "surrogate_tau": TAU_SURR}, | |
| "events": {f"{fam}|{r}|{s}": {"n": int(len(EV[(r, s, fam)][0])), "n_censored_start": EV[(r, s, fam)][1], "hours": grids[(r, s)]["hours"]} for fam in FAMS for r in REGIMES for s in ("val", "test")}, | |
| "headline_budget": a.headline_budget, "headline_recall_early_1s": headline, "run_dir": str(out), "dms_fields_used": [], "wall_s": time.time() - t_start} | |
| (tables_dir / f"alert_timing_{stamp}.json").write_text(json.dumps(js, indent=1)) | |
| (out / "run.json").write_text(json.dumps(dict(js, command=" ".join(sys.argv)), indent=1)) | |
| log(f"DONE in {time.time() - t_start:.0f}s -> {out}") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 73.2 kB
- Xet hash:
- ee89888137771a313bd88d96e3acf5e73dcfbdcdf2712bca06e291e080279c50
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.