Buckets:
| """cwm3/eval/phone_episode_alerts.py — EVALUATION-ONLY alert share of warning read-out scores on the phone / object / gaze-away episodes of the TEST routes | |
| (kin v2p arm; pre-registration docs/decisions.md 2026-09-18T05:34:40Z). Nothing is fitted here: the episodes are the merged positives of | |
| results/tables/pitch_wristface_diag.json (labels used only to select evaluation windows), the thresholds are the VAL-fitted per-seed / regime / budget thresholds of an | |
| eval/alert_timing_eval.py run (thresholds.json key `s<seed>|val|<score>|<regime>|<budget>`), alerts are score >= thr on the TEST layout rows in (uid, t) order with the | |
| existing 60 s per-route cooldown (train/train_warning_head.alerts_per_hour semantics), and an episode counts as alerted when >= 1 alert falls inside [t - 5, t + 20] s | |
| (the window of results/tables/phone_episode_audit.md). CPU, numpy only; read-only on every run dir; no openpilot DMS field is read. | |
| python eval/phone_episode_alerts.py --timing-dir results/alert_timing/<stamp> --scores name=W_DIR_PATTERN:COLUMN ... [--pairs a-b ...] [--budgets 20 40] [--seeds 0 1 2] | |
| [--episodes results/tables/pitch_wristface_diag.json] [--out DIR] | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import sys | |
| import time | |
| from pathlib import Path | |
| import numpy as np | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| RUNS = CWM3 / "results" / "runs" | |
| WIN = (-5.0, 20.0) | |
| COOLDOWN = 60.0 | |
| REGIMES = ("manual", "engaged") | |
| def alert_times(uid, t, score, thr, cooldown=COOLDOWN): | |
| """rows of one regime's TEST layout -> dict uid -> sorted array of alert times (score >= thr; per-route cooldown; rows in (uid, t) order).""" | |
| order = np.lexsort((t, uid)) | |
| u, tt, s = uid[order], t[order], score[order] | |
| fire = np.isfinite(s) & (s >= thr) | |
| out = {} | |
| last_uid, last_t = None, -1e9 | |
| for i in np.flatnonzero(fire): | |
| if u[i] != last_uid: | |
| last_uid, last_t = u[i], -1e9 | |
| # rows between fires of another route do not matter: the cooldown is per route, and we iterate fires only -> reset on route change | |
| if tt[i] - last_t >= cooldown: | |
| out.setdefault(str(u[i]), []).append(float(tt[i])) | |
| last_t = tt[i] | |
| return {k: np.asarray(v) for k, v in out.items()} | |
| def load_layout(wdir, col, regime, split="test"): | |
| p = RUNS / wdir / f"pred_layout_{regime}_{split}.npz" | |
| with np.load(p, allow_pickle=True) as z: | |
| names = [str(x) for x in z["w_names"]] | |
| assert col in names, (p, col, names) | |
| w = z["w"][:, names.index(col)].astype(np.float32) | |
| mask = z["w_mask"].astype(bool) if "w_mask" in z.files else np.ones(len(w), bool) | |
| w = np.where(mask, w, np.nan) | |
| return z["uid"].astype(str), z["t"].astype(np.float64), w | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--timing-dir", required=True) | |
| ap.add_argument("--scores", nargs="+", required=True, help="name=W_DIR_PATTERN:COLUMN ({seed} expands); `name` must match a score of the timing run's thresholds.json") | |
| ap.add_argument("--pairs", nargs="*", default=[], help="a-b pairs (score names) for the discordant-episode counts") | |
| ap.add_argument("--budgets", nargs="+", type=float, default=[20.0, 40.0]) | |
| ap.add_argument("--seeds", nargs="+", type=int, default=[0, 1, 2]) | |
| ap.add_argument("--episodes", default=str(CWM3 / "results" / "tables" / "pitch_wristface_diag.json")) | |
| ap.add_argument("--out", default="") | |
| a = ap.parse_args() | |
| t0 = time.time(); stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| out = Path(a.out) if a.out else CWM3 / "results" / "kin_v2p_arm" / f"phone_alerts_{stamp}" | |
| out.mkdir(parents=True, exist_ok=True) | |
| thr = json.loads((Path(a.timing_dir) / "thresholds.json").read_text()) | |
| D = json.loads(Path(a.episodes).read_text()) | |
| eps = D["episodes"] | |
| assert len(eps) > 0, "no episodes" | |
| specs = {} | |
| for s in a.scores: | |
| name, rest = s.split("=", 1); pat, col = rest.rsplit(":", 1); specs[name] = (pat, col) | |
| per = {} # (name, seed, budget) -> bool array over episodes | |
| rows = [] | |
| for seed in a.seeds: | |
| for name, (pat, col) in specs.items(): | |
| lay = {r: load_layout(pat.format(seed=seed), col, r) for r in REGIMES} | |
| for B in a.budgets: | |
| al = {} | |
| for r in REGIMES: | |
| key = f"s{seed}|val|{name}|{r}|{B:g}" | |
| assert key in thr, f"{key} not in {a.timing_dir}/thresholds.json" | |
| al[r] = alert_times(*lay[r], thr[key]["thr"]) | |
| hit = np.zeros(len(eps), bool); first = np.full(len(eps), np.nan) | |
| for i, e in enumerate(eps): | |
| ts = al[e["regime"]].get(e["uid"], np.zeros(0)) | |
| m = (ts >= e["t"] + WIN[0]) & (ts <= e["t"] + WIN[1]) | |
| hit[i] = bool(m.any()) | |
| if m.any(): | |
| first[i] = float(ts[m].min() - e["t"]) | |
| per[(name, seed, B)] = hit | |
| tags = sorted({tg for e in eps for tg in e["tags"]}) | |
| rec = {"score": name, "seed": seed, "budget": B, "n_episodes": len(eps), "n_alerted": int(hit.sum()), "share": float(hit.mean()), | |
| "share_by_regime": {r: float(hit[[e["regime"] == r for e in eps]].mean()) for r in REGIMES}, | |
| "share_by_tag": {tg: float(hit[[tg in e["tags"] for e in eps]].mean()) for tg in tags}, | |
| "n_by_tag": {tg: int(sum(tg in e["tags"] for e in eps)) for tg in tags}, | |
| "first_alert_rel_s_median": float(np.nanmedian(first)) if np.isfinite(first).any() else None, | |
| "thr": {r: thr[f"s{seed}|val|{name}|{r}|{B:g}"]["thr"] for r in REGIMES}, "reachable": {r: thr[f"s{seed}|val|{name}|{r}|{B:g}"]["reachable"] for r in REGIMES}} | |
| rows.append(rec) | |
| print(f"s{seed} {name:14s} @{B:g}/h: {rec['n_alerted']}/{len(eps)} = {rec['share']:.3f} manual {rec['share_by_regime']['manual']:.3f} engaged {rec['share_by_regime']['engaged']:.3f}", flush=True) | |
| pairs = [] | |
| for pr in a.pairs: | |
| na, nb = pr.split("-", 1) | |
| for B in a.budgets: | |
| ds_ = [] | |
| for seed in a.seeds: | |
| ha, hb = per[(na, seed, B)], per[(nb, seed, B)] | |
| ds_.append({"seed": seed, "a_only": int((ha & ~hb).sum()), "b_only": int((~ha & hb).sum()), "both": int((ha & hb).sum()), "neither": int((~ha & ~hb).sum()), | |
| "share_a": float(ha.mean()), "share_b": float(hb.mean()), "delta": float(ha.mean() - hb.mean())}) | |
| pairs.append({"a": na, "b": nb, "budget": B, "per_seed": ds_, "mean_delta": float(np.mean([d["delta"] for d in ds_])), "mean_share_a": float(np.mean([d["share_a"] for d in ds_])), | |
| "mean_share_b": float(np.mean([d["share_b"] for d in ds_])), "a_higher_all_seeds": all(d["delta"] > 0 for d in ds_)}) | |
| summary = {} | |
| for name in specs: | |
| for B in a.budgets: | |
| sh = [r["share"] for r in rows if r["score"] == name and r["budget"] == B] | |
| summary[f"{name}|{B:g}"] = {"shares": sh, "mean": float(np.mean(sh)), "n_episodes": len(eps)} | |
| J = {"stamp": stamp, "script": "eval/phone_episode_alerts.py", "timing_dir": str(a.timing_dir), "episodes_file": str(a.episodes), "n_episodes": len(eps), "window_s": list(WIN), "cooldown_s": COOLDOWN, | |
| "thresholds": "VAL-fitted per seed / score / regime / budget (alert_timing_eval.py thresholds.json)", "scores": {k: {"pattern": v[0], "column": v[1]} for k, v in specs.items()}, | |
| "rows": rows, "pairs": pairs, "summary": summary, "episode_hits": {f"{n}|s{s}|{B:g}": h.astype(int).tolist() for (n, s, B), h in per.items()}, | |
| "episodes": [{"id": e["id"], "uid": e["uid"], "regime": e["regime"], "t": e["t"], "tags": e["tags"]} for e in eps], "sec": round(time.time() - t0, 1)} | |
| (out / "phone_alerts.json").write_text(json.dumps(J, indent=1)) | |
| print(f"wrote {out / 'phone_alerts.json'} ({J['sec']} s)") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 8.1 kB
- Xet hash:
- 381f30e1b86d5927d094b37f7124edcc152d3cda807b8c5baf0a1060149c4a64
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.