Buckets:
| """cwm3/eval/f2_readout_verdict.py — dated READING sections for the two 2026-09-17 owner decisions, computed ONLY from published outputs (nothing is re-scored): | |
| PART A (decision 6, F2′ into the warning read-out): gold levels (AUROC pooled / manual / engaged, Brier / ECE raw and temperature-calibrated) of the cur+dwell+face+fs arm | |
| and its cur+dwell+face reference from results/warning_gold/warning_gold.csv (items cwm3_final_dwfcfs / cwm3_final_dwfcfsrmv), the paired Δ per stratum from the regime | |
| report's pair CSV (results/warning_head_ext/regime_pairs_<item>_*.csv, identical rows / resamples), the paired Δ of both arms vs the openpilot-derived surrogate (legacy_att | |
| rows of warning_gold.csv), and the event-level timing endpoints of the fs read-out vs cur+dwell+face (results/alert_timing/<stamp A>/paired.csv, families A / B / C1 / C2 / D1 / D2, | |
| budgets 5 / 10 / 20, δ 0.5 / 1 / 2 s). ADOPTION RULE (pre-registered in the WP2 / WP4 rows, restated by the coordinator): paired gold AUROC not worse than the reference | |
| (pooled Δ ≥ −0.005 on ≥ 2/3 seeds) AND at least one timing endpoint significantly better (paired CI > 0) on ≥ 2/3 seeds → 'adoptable'; otherwise 'not adopted'. | |
| Both the VAL-fitted (primary, deployable) and the TEST-fitted (equal achieved load, secondary) threshold rows are read; the number of endpoints tested is printed (multiplicity). | |
| PART B (decision 7, timing family D with the ECAN demand forecasts): levels and paired differences of the learned demand-onset forecasts `ph_ecan_*` (P / D heads of the ECAN | |
| joint runs) vs the U-gated constant-velocity TTC rule `cv_ttc_3s_xU`, the frozen main's P head `ph_main_*`, the margin rule, the B2 road-dynamics forecast, the current-state | |
| read-out and the surrogate, from results/alert_timing/<stamp B>/{metrics,paired}.csv (families D1 / D2, budgets 10 / 20 / 40). Verdict: the ECAN forecast BEATS the physics | |
| rule at ≤ 3 s iff `ph_ecan_P_hard3s_xU − cv_ttc_3s_xU` has a paired CI > 0 on ≥ 2/3 seeds (pooled, δ = 1 s primary; every δ / budget / fit shown). | |
| python eval/f2_readout_verdict.py [--timing-a <stamp>] [--timing-b <stamp>] [--out-a results/tables/warning_head_final_dwfcfs.md] [--out-b results/tables/alert_timing.md] [--dry] | |
| No openpilot DMS field is read (the surrogate enters as the labelled comparator rows already published). | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import time | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| WG = CWM3 / "results" / "warning_gold" / "warning_gold.csv" | |
| PAIRS = CWM3 / "results" / "warning_head_ext" | |
| AT = CWM3 / "results" / "alert_timing" | |
| STRATA = [("clean_test", "pooled"), ("clean_test|manual", "manual"), ("clean_test|engaged", "engaged")] | |
| COL, REF = "W_learned_curhgb_t15_w10_n979dwfcfs", "W_learned_curhgb_t15_w10_n979dwfc" | |
| NOT_WORSE_MARGIN = -0.005 | |
| def f3(x, nd=3): | |
| try: | |
| x = float(x) | |
| return "—" if not np.isfinite(x) else f"{x:.{nd}f}" | |
| except (TypeError, ValueError): | |
| return "—" | |
| def sgn(x, nd=3): | |
| return "—" if not np.isfinite(x) else f"{x:+.{nd}f}" | |
| def gold_block(item, col, ref, W): | |
| """levels + calibration + paired vs surrogate for one item -> (lines, summary dict).""" | |
| w = W[W.item == item] | |
| assert len(w), f"no warning_gold rows for item {item}" | |
| stamp = w.run_stamp.max() | |
| w = w[w.run_stamp == stamp] | |
| seeds = sorted(int(s) for s in w.seed.unique() if str(s).isdigit()) | |
| arm_col = sorted({a for a in w.arm.unique() if str(a).startswith("abl_W") and len(w[(w.arm == a) & (w.event_or_family == col)])}) | |
| arm_ref = sorted({a for a in w.arm.unique() if str(a).startswith("abl_W") and len(w[(w.arm == a) & (w.event_or_family == ref)])}) | |
| assert len(arm_col) == 1 and len(arm_ref) == 1, (arm_col, arm_ref) | |
| arm_col, arm_ref = arm_col[0], arm_ref[0] | |
| L = [f"Item `{item}` (warning_gold run {stamp}; arms `{arm_col}` = cur+dwell+face+fs, `{arm_ref}` = cur+dwell+face reference; seeds {seeds}).", "", | |
| "| column | stratum | " + " | ".join(f"AUROC s{s} [CI]" for s in seeds) + " | mean | n / pos |", "|---|---|" + "---|" * len(seeds) + "---|---|"] | |
| summ = {"item": item, "stamp": stamp, "levels": {}, "cal": {}, "vs_surrogate": {}} | |
| lev = w[(w.metric == "auroc") & (w.calib == "raw") & (w.comparator.isna())] | |
| for arm, c in ((arm_col, col), (arm_ref, ref)): | |
| for st, lab in STRATA: | |
| g = lev[(lev.arm == arm) & (lev.event_or_family == c) & (lev.stratum == st)] | |
| vals = {int(s): (float(v), float(lo), float(hi)) for s, v, lo, hi in zip(g.seed, g.value, g.ci_lo, g.ci_hi) if str(s).isdigit()} | |
| v = np.array([vals[s][0] for s in seeds if s in vals]) | |
| summ["levels"][f"{c}|{lab}"] = v.tolist() | |
| L.append(f"| {c} | {lab} | " + " | ".join(f"{f3(vals[s][0])} [{f3(vals[s][1], 2)},{f3(vals[s][2], 2)}]" if s in vals else "—" for s in seeds) | |
| + f" | {f3(v.mean()) if len(v) else '—'} | {int(g.n.iloc[0]) if len(g) else '—'} / {int(g.n_pos.iloc[0]) if len(g) else '—'} |") | |
| L += ["", "Calibration on the clean-test gold (pooled; Brier with device-cluster CI, ECE 15 bins; `_cal` = after the temperature fitted on clean-val; seed 0 shown — current-state heads are " | |
| "seed-invariant and the F2′ pack is one head per FINAL seed, so the seeds are listed when they differ):", "", | |
| "| column | Brier raw [CI] | ECE raw | Brier temp-cal [CI] | ECE temp-cal | seeds identical |", "|---|---|---|---|---|---|"] | |
| for arm, c in ((arm_col, col), (arm_ref, ref)): | |
| cells, ident = [], True | |
| for met, cb in (("brier", "raw"), ("ece", "raw"), ("brier_cal", "temperature"), ("ece_cal", "temperature")): | |
| g = w[(w.arm == arm) & (w.event_or_family == c) & (w.stratum == "clean_test") & (w.metric == met) & (w.calib == cb) & (w.comparator.isna())] | |
| g = g[g.seed.astype(str).str.isdigit()].drop_duplicates(["seed", "metric", "calib"]) | |
| vals = {int(s): float(v) for s, v in zip(g.seed, g.value)} | |
| ident &= len({round(v, 6) for v in vals.values()}) <= 1 | |
| summ["cal"][f"{c}|{met}|{cb}"] = vals | |
| v0 = vals.get(seeds[0], np.nan) | |
| ci = g[g.seed.astype(int) == seeds[0]] | |
| ci_txt = f" [{f3(ci.ci_lo.iloc[0], 3)},{f3(ci.ci_hi.iloc[0], 3)}]" if len(ci) and met.startswith("brier") and np.isfinite(float(ci.ci_lo.iloc[0])) else "" | |
| cells.append(f3(v0) + ci_txt + ("" if len({round(v, 6) for v in vals.values()}) <= 1 else " (" + " / ".join(f3(vals[s]) for s in seeds if s in vals) + ")")) | |
| L.append(f"| {c} | " + " | ".join(cells) + f" | {ident} |") | |
| L += ["", "Paired Δ vs the openpilot-derived surrogate (`legacy_att` att@3s; warning_gold pairs on identical device-cluster resamples):", "", | |
| "| column | stratum | " + " | ".join(f"s{s} Δ [CI]" for s in seeds) + " | mean Δ | seeds CI excl. 0 (+ / −) |", "|---|---|" + "---|" * len(seeds) + "---|---|"] | |
| for arm, c in ((arm_col, col), (arm_ref, ref)): | |
| for st, lab in STRATA: | |
| g = w[(w.arm == arm) & (w.event_or_family == c) & (w.stratum == st) & (w.metric == "auroc") & (w.comparator == "legacy_att")] | |
| g = g[g.seed.astype(str).str.isdigit()].drop_duplicates("seed").sort_values("seed") | |
| d = g.delta.astype(float).to_numpy() | |
| pos, neg = int((g.delta_ci_lo.astype(float) > 0).sum()), int((g.delta_ci_hi.astype(float) < 0).sum()) | |
| summ["vs_surrogate"][f"{c}|{lab}"] = {"delta": d.tolist(), "pos": pos, "neg": neg} | |
| L.append(f"| {c} | {lab} | " + " | ".join(f"{sgn(float(x.delta))} [{sgn(float(x.delta_ci_lo))},{sgn(float(x.delta_ci_hi))}]" for x in g.itertuples()) | |
| + f" | {sgn(d.mean()) if len(d) else '—'} | {pos} / {neg} of {len(g)} |") | |
| return L, summ, seeds | |
| def pair_block(item, effect): | |
| files = sorted(PAIRS.glob(f"regime_pairs_{item}_*.csv")) | |
| assert files, f"no regime_pairs CSV for {item}" | |
| P = pd.read_csv(files[-1]) | |
| P = P[P.effect == effect] | |
| assert len(P), f"effect {effect} missing in {files[-1].name}" | |
| seeds = sorted(int(s) for s in P.seed.unique()) | |
| L = [f"Paired Δ = cur+dwell+face+fs − cur+dwell+face on identical rows and device-cluster resamples (regime report pair `{effect}`, {files[-1].name}):", "", | |
| "| stratum | " + " | ".join(f"s{s} Δ [CI]" for s in seeds) + " | mean Δ ± sd | seeds Δ ≥ −0.005 | seeds CI excl. 0 (+ / −) |", "|---|" + "---|" * len(seeds) + "---|---|---|"] | |
| res = {} | |
| for st, lab in STRATA: | |
| g = P[P.stratum == st].sort_values("seed") | |
| d = g.delta.to_numpy(float) | |
| nw = int((d >= NOT_WORSE_MARGIN).sum()) | |
| pos, neg = int((g.delta_ci_lo > 0).sum()), int((g.delta_ci_hi < 0).sum()) | |
| res[lab] = {"delta": d.tolist(), "ci_lo": g.delta_ci_lo.tolist(), "ci_hi": g.delta_ci_hi.tolist(), "n_not_worse": nw, "pos": pos, "neg": neg, "n": int(len(g))} | |
| L.append(f"| {lab} | " + " | ".join(f"{x.delta:+.3f} [{x.delta_ci_lo:+.3f},{x.delta_ci_hi:+.3f}]" for x in g.itertuples()) | |
| + f" | {d.mean():+.3f} ± {d.std(ddof=1) if len(d) > 1 else 0:.3f} | {nw}/{len(g)} | {pos} / {neg} |") | |
| return L, res | |
| def timing_a_block(stamp, a_name="fs_head_dwfcfs", b_name="cur_dwfc"): | |
| d = AT / stamp | |
| P = pd.read_csv(d / "paired.csv") | |
| M = pd.read_csv(d / "metrics.csv") | |
| P = P[(P.a == a_name) & (P.b == b_name)] | |
| assert len(P), f"no {a_name} − {b_name} pairs in {d}" | |
| seeds = sorted(int(s) for s in P.seed.unique()) | |
| L, res = [], {} | |
| for fit in ("val", "test"): | |
| Q = P[P.thr_fit == fit] | |
| if not len(Q): | |
| continue | |
| rows = [] | |
| for (fam, B, dl, reg), g in Q.groupby(["family", "budget", "delta_s", "regime"]): | |
| g = g.sort_values("seed") | |
| pos, neg = int((g.ci_lo > 0).sum()), int((g.ci_hi < 0).sum()) | |
| rows.append(dict(family=fam, budget=B, delta_s=dl, regime=reg, mean=float(g["diff"].mean()), pos=pos, neg=neg, n=int(len(g)), cells=[f"{x.diff:+.3f} [{x.ci_lo:+.3f},{x.ci_hi:+.3f}]" for x in g.itertuples()], | |
| va=float(g.value_a.mean()), vb=float(g.value_b.mean()))) | |
| R = pd.DataFrame(rows) | |
| better = R[(R.pos >= 2) & (R.n >= 3)] | |
| worse = R[(R.neg >= 2) & (R.n >= 3)] | |
| n_ep = int(len(R)) | |
| res[fit] = {"n_endpoints": n_ep, "n_pooled": int((R.regime == "pooled").sum()), "better": better[["family", "budget", "delta_s", "regime", "mean", "pos"]].to_dict("records"), | |
| "worse": worse[["family", "budget", "delta_s", "regime", "mean", "neg"]].to_dict("records"), | |
| "n_better_pooled": int((better.regime == "pooled").sum()), "n_worse_pooled": int((worse.regime == "pooled").sum()), | |
| "better_budgets": sorted(better.budget.unique().tolist()), "max_gain": float(better["mean"].max()) if len(better) else float("nan")} | |
| # achieved alert load of A and B at these thresholds (alerts are family-independent; first family, pooled + per regime) | |
| Mf = M[(M.thr_fit == fit) & (M.family == sorted(M.family.unique())[0]) & (M.score.isin([a_name, b_name])) & (~M.regime.astype(str).str.contains("silent"))] | |
| load = {(B, reg, sc): float(Mf[(Mf.budget == B) & (Mf.regime == reg) & (Mf.score == sc)].alerts_per_h_test.mean()) for B in sorted(Mf.budget.unique()) for reg in ("manual", "engaged", "pooled") for sc in (a_name, b_name)} | |
| res[fit]["load"] = {f"{B:g}|{reg}": [load[(B, reg, a_name)], load[(B, reg, b_name)]] for B in sorted(Mf.budget.unique()) for reg in ("manual", "engaged", "pooled")} | |
| L += ["", f"**{'VAL-fitted thresholds (PRIMARY, deployable)' if fit == 'val' else 'TEST-fitted thresholds (SECONDARY, equal achieved test load)'}** — `{a_name} − {b_name}`, Recall_{{≥ δ s early}}; " | |
| f"{n_ep} endpoints (family × budget × δ × regime); endpoints with the paired CI > 0 on ≥ 2/3 seeds: **{len(better)}** (pooled {int((better.regime == 'pooled').sum())}); CI < 0 on ≥ 2/3 seeds: **{len(worse)}** (pooled {int((worse.regime == 'pooled').sum())}).", "", | |
| "Achieved TEST alert load at these thresholds (alerts/h, mean over seeds; A = fs read-out / B = cur+dwell+face) — a load gap at equal budget means the val-fitted comparison is load-confounded:", "", | |
| "| budget | manual A / B | engaged A / B | pooled A / B |", "|---|---|---|---|"] | |
| for B in sorted(Mf.budget.unique()): | |
| L.append(f"| {B:g} | " + " | ".join(f"{load[(B, reg, a_name)]:.1f} / {load[(B, reg, b_name)]:.1f}" for reg in ("manual", "engaged", "pooled")) + " |") | |
| L += ["", "| family | budget | δ | regime | " + " | ".join(f"s{s} Δ [CI]" for s in seeds) + " | mean Δ | A / B level (mean) | seeds CI excl. 0 (+ / −) |", "|---|---|---|---|" + "---|" * len(seeds) + "---|---|---|"] | |
| for x in R[R.regime == "pooled"].sort_values(["family", "budget", "delta_s"]).itertuples(): | |
| L.append(f"| {x.family} | {x.budget:g} | {x.delta_s:g} | {x.regime} | " + " | ".join(x.cells) + f" | {x.mean:+.3f} | {x.va:.3f} / {x.vb:.3f} | {x.pos} / {x.neg} |") | |
| if len(better) or len(worse): | |
| L += ["", "Per-regime endpoints with ≥ 2/3 seeds CI excl. 0:", "", "| family | budget | δ | regime | " + " | ".join(f"s{s} Δ [CI]" for s in seeds) + " | mean Δ | direction |", "|---|---|---|---|" + "---|" * len(seeds) + "---|---|"] | |
| for x in pd.concat([better, worse]).sort_values(["family", "budget", "delta_s", "regime"]).itertuples(): | |
| L.append(f"| {x.family} | {x.budget:g} | {x.delta_s:g} | {x.regime} | " + " | ".join(x.cells) + f" | {x.mean:+.3f} | {'BETTER' if x.pos >= 2 else 'WORSE'} ({max(x.pos, x.neg)}/{x.n}) |") | |
| # levels at 10 / h, recall >= 1 s, pooled, per family (val fit) | |
| Mv = M[(M.thr_fit == "val") & (M.regime == "pooled") & (M.budget == 10.0) & (M.score.isin([a_name, b_name, "surrogate_att3s", "fs_occ_any_1s", "persistence", "dwell_timer"]))] | |
| L += ["", "Levels at 10 / h (val-fitted), pooled, Recall ≥ 1 s early, mean ± sd over seeds (achieved alerts/h test in brackets):", "", "| family | " + " | ".join(sorted(Mv.score.unique(), key=lambda n: [a_name, b_name].index(n) if n in (a_name, b_name) else 9)) + " |", | |
| "|---|" + "---|" * Mv.score.nunique()] | |
| order = sorted(Mv.score.unique(), key=lambda n: [a_name, b_name].index(n) if n in (a_name, b_name) else 9) | |
| for fam, g in Mv.groupby("family"): | |
| cells = [] | |
| for n_ in order: | |
| h = g[g.score == n_] | |
| v = h.recall_early_1s.to_numpy(float) | |
| cells.append(f"{v.mean():.3f} ± {v.std(ddof=1) if len(v) > 1 else 0:.3f} ({h.alerts_per_h_test.mean():.1f}/h)" if len(v) else "—") | |
| L.append(f"| {fam} | " + " | ".join(cells) + " |") | |
| return L, res, seeds | |
| def timing_b_block(stamp): | |
| d = AT / stamp | |
| P = pd.read_csv(d / "paired.csv") | |
| M = pd.read_csv(d / "metrics.csv") | |
| seeds = sorted(int(s) for s in M.seed.unique()) | |
| scores_show = [s for s in ["ph_ecan_P_hard3s_xU", "ph_main_P_hard3s_xU", "ph_ecan_P_hard3s", "ph_main_P_hard3s", "ph_ecan_D_ttc3s_xU", "ph_main_D_ttc3s_xU", "cv_ttc_3s_xU", "cv_ttc_3s", "road_mr_hard_xU", | |
| "rd_ttc_min_3s_xU", "fs_occ_any_1s", "fs_head_dwfcfs", "cur_dwfc", "both_dwfc", "surrogate_att3s", "persistence", "dwell_timer"] if s in set(M.score)] | |
| L = [] | |
| for fit in ("val", "test"): | |
| Mm = M[(M.thr_fit == fit) & (M.regime == "pooled")] | |
| if not len(Mm): | |
| continue | |
| L += ["", f"**Levels, {'VAL-fitted (PRIMARY)' if fit == 'val' else 'TEST-fitted = equal achieved test load (SECONDARY)'}**, pooled, Recall ≥ 0.5 / 1 / 2 s early, mean ± sd over seeds; achieved alerts/h test in brackets (`floor` = budget unreachable):", "", | |
| "| score | " + " | ".join(f"{fam} @ {B:g}/h" for fam in sorted(Mm.family.unique()) for B in sorted(Mm.budget.unique())) + " |", "|---|" + "---|" * (Mm.family.nunique() * Mm.budget.nunique())] | |
| for sc in scores_show: | |
| cells = [] | |
| for fam in sorted(Mm.family.unique()): | |
| for B in sorted(Mm.budget.unique()): | |
| h = Mm[(Mm.score == sc) & (Mm.family == fam) & (Mm.budget == B)] | |
| if not len(h): | |
| cells.append("—"); continue | |
| fl = "" if bool(h.reachable.all()) else " `floor`" | |
| cells.append(" / ".join(f"{h[f'recall_early_{dl}s'].mean():.3f}" for dl in ("0.5", "1", "2")) + f" (±{h.recall_early_1s.std(ddof=1) if len(h) > 1 else 0:.3f}; {h.alerts_per_h_test.mean():.1f}/h){fl}") | |
| L.append(f"| {sc} | " + " | ".join(cells) + " |") | |
| key_pairs = [("ph_ecan_P_hard3s_xU", "cv_ttc_3s_xU"), ("ph_ecan_P_hard3s_xU", "ph_main_P_hard3s_xU"), ("ph_main_P_hard3s_xU", "cv_ttc_3s_xU"), ("ph_ecan_P_hard3s_xU", "road_mr_hard_xU"), | |
| ("ph_ecan_P_hard3s_xU", "rd_ttc_min_3s_xU"), ("ph_ecan_P_hard3s", "cv_ttc_3s"), ("ph_ecan_D_ttc3s_xU", "cv_ttc_3s_xU"), ("ph_ecan_P_hard3s_xU", "cur_dwfc"), ("ph_ecan_P_hard3s_xU", "surrogate_att3s"), | |
| ("ph_ecan_P_hard3s_xU", "ph_ecan_P_hard3s"), ("fs_head_dwfcfs", "cv_ttc_3s_xU"), ("fs_head_dwfcfs", "cur_dwfc")] | |
| res = {} | |
| for fit in ("val", "test"): | |
| Q = P[P.thr_fit == fit] | |
| if not len(Q): | |
| continue | |
| L += ["", f"**Paired Δ (Recall ≥ δ s early, identical device-cluster resamples), pooled, {'VAL-fitted (PRIMARY)' if fit == 'val' else 'TEST-fitted equal load (SECONDARY)'}**:", "", | |
| "| A − B | family | budget | δ | " + " | ".join(f"s{s} Δ [CI]" for s in seeds) + " | mean Δ | seeds CI excl. 0 (+ / −) |", "|---|---|---|---|" + "---|" * len(seeds) + "---|---|"] | |
| for na, nb in key_pairs: | |
| for (fam, B, dl), g in Q[(Q.a == na) & (Q.b == nb) & (Q.regime == "pooled")].groupby(["family", "budget", "delta_s"]): | |
| g = g.sort_values("seed") | |
| pos, neg = int((g.ci_lo > 0).sum()), int((g.ci_hi < 0).sum()) | |
| res[f"{fit}|{na}-{nb}|{fam}|{B:g}|{dl:g}"] = {"mean": float(g["diff"].mean()), "pos": pos, "neg": neg, "n": int(len(g)), "diff": g["diff"].tolist(), "ci_lo": g.ci_lo.tolist(), "ci_hi": g.ci_hi.tolist()} | |
| L.append(f"| {na} − {nb} | {fam} | {B:g} | {dl:g} | " + " | ".join(f"{x.diff:+.3f} [{x.ci_lo:+.3f},{x.ci_hi:+.3f}]" for x in g.itertuples()) + f" | {g['diff'].mean():+.3f} | {pos} / {neg} |") | |
| # per-regime rows of the headline pair at 20 / h, δ = 1 s | |
| L += ["", f"Headline pair per regime ({fit}-fitted): `ph_ecan_P_hard3s_xU − cv_ttc_3s_xU`, budget 20 / h, δ ∈ {{0.5, 1, 2}} s:", "", | |
| "| family | regime | δ | " + " | ".join(f"s{s} Δ [CI]" for s in seeds) + " | mean Δ | seeds CI excl. 0 (+ / −) |", "|---|---|---|" + "---|" * len(seeds) + "---|---|"] | |
| for (fam, reg, dl), g in Q[(Q.a == "ph_ecan_P_hard3s_xU") & (Q.b == "cv_ttc_3s_xU") & (Q.budget == 20.0)].groupby(["family", "regime", "delta_s"]): | |
| g = g.sort_values("seed") | |
| L.append(f"| {fam} | {reg} | {dl:g} | " + " | ".join(f"{x.diff:+.3f} [{x.ci_lo:+.3f},{x.ci_hi:+.3f}]" for x in g.itertuples()) + f" | {g['diff'].mean():+.3f} | {int((g.ci_lo > 0).sum())} / {int((g.ci_hi < 0).sum())} |") | |
| return L, res, seeds | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--timing-a", default="", help="results/alert_timing/<stamp> of the families A/B/C1/C2/D1/D2 run with fs_head_dwfcfs") | |
| ap.add_argument("--timing-b", default="", help="results/alert_timing/<stamp> of the D1/D2 run with --phead-runs ecan / main") | |
| ap.add_argument("--out-a", default=str(CWM3 / "results" / "tables" / "warning_head_final_dwfcfs.md")) | |
| ap.add_argument("--out-b", default=str(CWM3 / "results" / "tables" / "alert_timing.md")) | |
| ap.add_argument("--dry", action="store_true", help="print only, append nothing") | |
| a = ap.parse_args() | |
| now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) | |
| W = pd.read_csv(WG, low_memory=False) | |
| out = {} | |
| # ------------------------------------------------------------------ PART A | |
| A = [f"## {now} — READING: F2′ future-state block in the warning read-out (owner decision (6), 2026-09-17) — eval/f2_readout_verdict.py (published outputs only)", "", | |
| "Arm `cur+dwell+face+fs` = the cur+dwell+face read-out (HGB, same 979 TRAIN labels, GRID, device-grouped 5-fold CV, clean-val temperature) + 22 F2′ features of the same seed's `FS_dwell_s<k>` pack " | |
| "(predicted off-task occupancy of hands_on / head_turned / gaze_off at +1 / 2 / 3 / 5 s, expected off-task seconds over 5 s, expected remaining off dwell, P(first leave within 3 s), validity; " | |
| "`train/warning_features_ext.fs_block`). The F2′ head was trained on ALL train anchors of the frozen FINAL run (no out-of-fold twin exists), so its features on the 979 label rows are in-sample " | |
| "for the F2′ head — an OOF variant of THIS arm is not defined (the cur+dwell+face part contains no joint-model output and `--oof` would only swap that part, giving a bit-identical head); " | |
| "the in-sample optimism of predicted features on the final model was ≈ 0.02 AUROC (`warning_head_final_oof.md`). Gold = 512 clean-test rows / 63 pos / 47 devices (unmasked), " | |
| "device-cluster bootstrap 2,000; reference numbers of cur+dwell+face: unmasked 0.716 / 0.725 / 0.664, verified mask 0.725 / 0.726 / 0.679 (pooled / manual / engaged).", ""] | |
| verdict_a = {} | |
| for item, col, ref, effect, lab in (("cwm3_final_dwfcfs", COL, REF, "fs_beyond_dwell_face", "unmasked"), ("cwm3_final_dwfcfsrmv", COL + "rmv", REF + "rmv", "fs_beyond_dwell_face_rmv", "verified 3-route mask (rmv)")): | |
| try: | |
| Lg, summ, seeds = gold_block(item, col, ref, W) | |
| Lp, pres = pair_block(item, effect) | |
| except AssertionError as ex: | |
| A += [f"### Gold, {lab}: NOT AVAILABLE ({ex})", ""] | |
| continue | |
| A += [f"### Gold, {lab}", ""] + Lg + [""] + Lp + [""] | |
| pooled = pres["pooled"] | |
| not_worse = pooled["n_not_worse"] >= 2 and pooled["n"] >= 3 | |
| verdict_a[lab] = {"not_worse": bool(not_worse), "pooled": pooled, "manual": pres["manual"], "engaged": pres["engaged"], "levels": summ["levels"], "vs_surrogate": summ["vs_surrogate"], "cal": summ["cal"]} | |
| A += [f"Gold criterion ({lab}): pooled paired Δ ≥ −0.005 on ≥ 2/3 seeds → **{'NOT WORSE' if not_worse else 'WORSE'}** (pooled Δ {', '.join(f'{d:+.3f}' for d in pooled['delta'])}; " | |
| f"CI excl. 0: {pooled['pos']} positive / {pooled['neg']} negative of {pooled['n']}; manual {np.mean(pres['manual']['delta']):+.3f}, engaged {np.mean(pres['engaged']['delta']):+.3f}).", ""] | |
| timing_ok = None | |
| if a.timing_a: | |
| Lt, tres, _ = timing_a_block(a.timing_a) | |
| A += [f"### Timing endpoints (`results/alert_timing/{a.timing_a}`; families A / B / C1 / C2 / D1 / D2; budgets 5 / 10 / 20 per h; δ 0.5 / 1 / 2 s; 1,000 device-cluster resamples; the fs read-out " | |
| f"`fs_head_dwfcfs` = `W_<run>_dwfcfs::{COL}` vs `cur_dwfc`)", ""] + Lt + [""] | |
| timing_ok = {fit: len(v["better"]) for fit, v in tres.items()} | |
| verdict_a["timing"] = tres | |
| # adoption | |
| if verdict_a: | |
| nw_un = verdict_a.get("unmasked", {}).get("not_worse") | |
| nw_rm = verdict_a.get("verified 3-route mask (rmv)", {}).get("not_worse") | |
| t_val = timing_ok.get("val", 0) if timing_ok else None | |
| t_test = timing_ok.get("test", 0) if timing_ok else None | |
| if timing_ok is None: | |
| verdict = "PENDING (timing run not yet read)" | |
| else: | |
| tv, tt = verdict_a["timing"].get("val", {}), verdict_a["timing"].get("test", {}) | |
| adopt = bool(nw_un) and (t_val or 0) > 0 | |
| qual = (f" — by the LETTER of the pre-registered rule on the val-fitted primary ({t_val} endpoints better, {tv.get('n_better_pooled', 0)} pooled, budgets {tv.get('better_budgets')}, largest mean gain {tv.get('max_gain', float('nan')):+.3f}); " | |
| f"at EQUAL ACHIEVED TEST LOAD the gain is {'NOT confirmed' if tt.get('n_better_pooled', 0) == 0 else 'confirmed'} ({t_test} endpoints better, {tt.get('n_better_pooled', 0)} pooled, {tt.get('n_worse_pooled', 0)} pooled worse)" | |
| if adopt else "") | |
| verdict = ("adoptable" if adopt else "not adopted") + qual | |
| A += ["### Adoption verdict (pre-registered rule: paired gold AUROC not worse (−0.005 margin) AND a timing endpoint significantly better on ≥ 2/3 seeds)", "", | |
| f"- gold not worse: unmasked **{nw_un}**, verified mask **{nw_rm}**", | |
| f"- timing endpoints better on ≥ 2/3 seeds (paired CI > 0): val-fitted (primary) **{t_val}**, test-fitted equal load (secondary) **{t_test}**" | |
| + (f" of {verdict_a['timing']['val']['n_endpoints']} / {verdict_a['timing']['test']['n_endpoints']} endpoints tested (multiplicity: with ≈ 200 endpoints a handful of 2/3-seed exclusions is expected under no effect; " | |
| "the direction and the size of every listed endpoint are shown above)" if timing_ok else ""), | |
| f"- **VERDICT: {verdict}**", ""] | |
| out["A"] = {"verdict": verdict, "gold": {k: v for k, v in verdict_a.items() if k != "timing"}, "timing_better_counts": timing_ok, "timing_stamp": a.timing_a} | |
| # ------------------------------------------------------------------ PART B | |
| B = [] | |
| if a.timing_b: | |
| Lb, bres, seeds_b = timing_b_block(a.timing_b) | |
| B = [f"## {now} — READING: timing family D with the ECAN demand forecasts (owner decision (7), 2026-09-17) — run `{a.timing_b}` above; eval/f2_readout_verdict.py (published outputs only)", "", | |
| "Road scores added by `--phead-runs` (2026-09-17): `ph_ecan_P_hard3s` = P(any hard P family ttc / ttc_drop / cut_in / lead_braking first switches high within 3 s) from the P-head onset hazards of " | |
| "`I1_all_minus_road2drv_s<k>_final_ecan` (CAN road margins as E anchor features, `stage_final_ecan.md`), `ph_main_*` = the same from the FROZEN main `I1_all_minus_road2drv_s<k>_final`; `_xU` = × 1[U_t] " | |
| "(the driver-state gate that carried the physics rule's gain); `ph_*_D_ttc3s_xU` = the legacy D chains ttc / ttc_drop. Controls: `cv_ttc_3s_xU` (U-gated constant-velocity TTC rule; 0.129 Recall ≥ 1 s " | |
| "pooled at 20 / h equal load in the 2026-09-16 run), `road_mr_hard_xU`, `rd_ttc_min_3s_xU` (B2 road-dynamics forecast), the current-state read-out, the F2′ scores and the surrogate. Same protocol: events on " | |
| "continuous TEST segments (D1 = hard-demand onset while U active, D2 = U entry within 3 s before), budgets 10 / 20 / 40 per h, δ 0.5 / 1 / 2 s, 60 s cooldown, match window [o − 10, o + 3] s, " | |
| "paired device-cluster bootstrap (1,000). **Question: does the learned ECAN demand forecast beat the physics rule at ≤ 3 s?** Rule: `ph_ecan_P_hard3s_xU − cv_ttc_3s_xU` paired CI > 0 on ≥ 2/3 seeds " | |
| "(pooled; δ = 1 s primary; every δ / budget / fit shown).", ""] + Lb + [""] | |
| hits, verd = [], [] | |
| for fit in ("test", "val"): | |
| for fam in ("D1", "D2"): | |
| for Bd in (10.0, 20.0, 40.0): | |
| for dl in (0.5, 1.0, 2.0): | |
| k = f"{fit}|ph_ecan_P_hard3s_xU-cv_ttc_3s_xU|{fam}|{Bd:g}|{dl:g}" | |
| if k in bres: | |
| r = bres[k] | |
| tag = "BEATS" if r["pos"] >= 2 and r["n"] >= 3 else ("WORSE" if r["neg"] >= 2 and r["n"] >= 3 else "no difference") | |
| verd.append(f"{fit} {fam} {Bd:g}/h δ{dl:g}: {r['mean']:+.3f} ({r['pos']}+/{r['neg']}− of {r['n']}) → {tag}") | |
| if tag == "BEATS": | |
| hits.append(k) | |
| k20 = {fit: bres.get(f"{fit}|ph_ecan_P_hard3s_xU-cv_ttc_3s_xU|D1|20|1") for fit in ("test", "val")} | |
| kmain = {fit: bres.get(f"{fit}|ph_ecan_P_hard3s_xU-ph_main_P_hard3s_xU|D1|20|1") for fit in ("test", "val")} | |
| B += ["### Verdict", "", "- ECAN P-head (U-gated) vs the U-gated CV-TTC rule, every cell: " + "; ".join(verd), "", | |
| f"- headline D1 @ 20 / h, δ = 1 s: equal load Δ {k20['test']['mean']:+.3f} ({k20['test']['pos']}+/{k20['test']['neg']}− of {k20['test']['n']}); val-fitted Δ {k20['val']['mean']:+.3f} ({k20['val']['pos']}+/{k20['val']['neg']}−)" if k20["test"] and k20["val"] else "- headline cell missing", | |
| f"- ECAN vs the frozen main's P head (both U-gated), D1 @ 20 / h, δ = 1 s: equal load Δ {kmain['test']['mean']:+.3f} ({kmain['test']['pos']}+/{kmain['test']['neg']}−); val-fitted Δ {kmain['val']['mean']:+.3f} ({kmain['val']['pos']}+/{kmain['val']['neg']}−)" if kmain["test"] and kmain["val"] else "- ECAN-vs-main cell missing", | |
| f"- **ANSWER: {'YES — the learned ECAN demand forecast beats the physics rule at ≤ 3 s in ' + str(len(hits)) + ' of ' + str(len(verd)) + ' cells: ' + ', '.join(h.replace('|ph_ecan_P_hard3s_xU-cv_ttc_3s_xU', '') for h in hits) if hits else 'NO — in no family × budget × δ cell (val- or test-fitted) does the U-gated ECAN P-head forecast beat the U-gated constant-velocity TTC rule on ≥ 2/3 seeds'}**", ""] | |
| out["B"] = {"stamp": a.timing_b, "cells": verd, "beats_cells": hits, "headline_D1_20_1s": k20, "ecan_vs_main_D1_20_1s": kmain} | |
| mdA, mdB = "\n".join(A) + "\n", ("\n".join(B) + "\n" if B else "") | |
| print(mdA) | |
| print(mdB) | |
| if not a.dry: | |
| with open(a.out_a, "a") as f: | |
| f.write(mdA) | |
| if mdB: | |
| with open(a.out_b, "a") as f: | |
| f.write(mdB) | |
| (CWM3 / "results" / "warning_head_ext" / f"f2_readout_verdict_{now.replace(':', '').replace('-', '')}.json").write_text(json.dumps(out, indent=1, default=str)) | |
| print(f"appended -> {a.out_a}" + (f" and {a.out_b}" if mdB else "")) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 30.1 kB
- Xet hash:
- 114620e8d7780ab5e4649586afe02ef65ac8574679a27bcd303e61125d8d2a60
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.