Buckets:
| """cwm3/eval/pressure_eval.py — road-pressure ONSET evaluation of the E0 `*_P` layouts (p_onset_p / p_names = P_<chain>), paired device-cluster bootstrap. | |
| Populations per P chain f (exactly the P_dem rule of cwm2/eval/populations.py applied to results/targets_pressure): valid = Pvalid[:, :, f] & Moving & M | |
| (s12 targets), z = P[:, :, f], z0 = P0[:, f]; rows inactive AND valid at the anchor (P0valid & ~P0) with valid[:, :tau].all(1); y = first onset < tau (tau = 15 steps = 3 s). | |
| Scores: A = the layout's p_onset_p at 3 s for P_<chain>; B = another layout | --b-margin (the current-margin RULE score S0[:, f] of targets_pressure: the | |
| threshold-normalised margin ratio at t — ttc: thr/ttc, ttc_drop: -dttc/thr (ttc < 8 s), cut_in: thr/headway, lead_braking: aLead/thr, lane_excursion: | |
| half-width/lat_margin, badas: badas/thr; > 1 = fired, 0 where the signal is absent or not moving — the P analogue of current_signal_S0) | --b-persistence | |
| (constant score -> PR-AUC = base rate; recorded as a metric "base_rate" row). Identical rows (both finite), shared-resample device-cluster bootstrap | |
| (paired_compare.boot_pair, n_boot 1000, seed 0) -> rows appended to results/paired_compare.csv (cell p_onset|<chain>|3s, metric pr_auc, tag). CPU; no DMS field. | |
| python eval/pressure_eval.py --regime manual --a results/runs/E0_pooled_P_s0_v2 --b-margin --tag exp3P_s0_pooledP_vs_margin | |
| python eval/pressure_eval.py --regime manual --a results/runs/E0_regionpca256_P_s0_v2 --b results/runs/E0_pooled_P_s0_v2 --tag exp3P_s0_rp256P_vs_pooledP | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import csv | |
| import sys | |
| import time | |
| from pathlib import Path | |
| import numpy as np | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| EXPLORE = CWM3.parent | |
| for p in (EXPLORE / "cwm2" / "eval", EXPLORE / "s12_wm", CWM3 / "common", CWM3 / "eval"): | |
| sys.path.insert(0, str(p)) | |
| from pred_layout import load_pred, col # noqa: E402 | |
| import importlib.util as _ilu # noqa: E402 | |
| _spec = _ilu.spec_from_file_location("cwm3_runmeta", CWM3 / "common" / "runmeta.py") # 2026-09-11: cwm2/eval/populations imports cwm2/data/runmeta.py (unfixed stamp-collision | |
| _rm = _ilu.module_from_spec(_spec); _spec.loader.exec_module(_rm) # assert) into sys.modules["runmeta"] first; load cwm3's fixed writer by path | |
| write_run_json = _rm.write_run_json | |
| from paired_compare import boot_pair, COLS, OUT, layout_guard, guard_line # noqa: E402 | |
| from pagecache import drop_all as _drop_cache # noqa: E402 | |
| S12R = EXPLORE / "s12_wm" / "results" | |
| PRESS = CWM3 / "results" / "targets_pressure" | |
| P_CHAINS = ("ttc", "ttc_drop", "cut_in", "lead_braking", "lane_excursion", "badas") | |
| TAU = 15 | |
| def populations(regime, split): | |
| tg = np.load(S12R / f"targets_{regime}_{split}.npz", allow_pickle=True) | |
| pz = np.load(PRESS / f"pressure_{regime}_{split}.npz", allow_pickle=True) | |
| assert np.array_equal(pz["uid"], tg["uid"]) and np.allclose(pz["t"], tg["t"]), "pressure rows != targets rows" | |
| assert [str(c) for c in pz["families"]] == list(P_CHAINS) | |
| M = tg["M"].astype(bool) | |
| P, Pv, P0, P0v, Mv, S0 = pz["P"].astype(bool), pz["Pvalid"].astype(bool), pz["P0"].astype(bool), pz["P0valid"].astype(bool), pz["Moving"].astype(bool), pz["S0"].astype(np.float32) | |
| pops = {} | |
| for f, name in enumerate(P_CHAINS): | |
| valid = Pv[:, :, f] & Mv & M | |
| z = P[:, :, f] | |
| zp = np.concatenate([P0[:, f][:, None], z[:, :-1]], 1) | |
| onset = z & ~zp & valid | |
| first = np.where(onset.any(1), onset.argmax(1), -1) | |
| inact0 = P0v[:, f] & ~P0[:, f] | |
| sel = np.flatnonzero(inact0 & valid[:, :TAU].all(1)) | |
| y = (first[sel] >= 0) & (first[sel] < TAU) | |
| pops[name] = {"sel": sel, "y": y, "dongle": tg["dongle"][sel], "margin": np.nan_to_num(S0[sel, f], nan=0.0)} | |
| _drop_cache([S12R / f"targets_{regime}_{split}.npz", PRESS / f"pressure_{regime}_{split}.npz"]) | |
| return {"uid": tg["uid"], "t": tg["t"]}, pops | |
| def p_scores(layout_dir, regime, split, tg): | |
| pa = load_pred(Path(layout_dir) / f"pred_layout_{regime}_{split}.npz", tg) | |
| assert "p_onset_p" in pa, f"{layout_dir}: no p_onset_p block (not a *_P layout)" | |
| names = [str(x) for x in pa["p_names"]] | |
| ti = list(map(int, pa["onset_taus"])).index(TAU) | |
| def pcol(c): | |
| # E0 `*_P` layouts name the columns P_<chain> (eval/write_pred_layout.py); the JOINT layouts (eval/write_pred_layout_joint.py) use the bare chain name. | |
| # 2026-09-12 (robustness table): fall back to the bare name so joint layouts are scored too; E0 layouts resolve exactly as before. | |
| j = col(names, f"P_{c}") | |
| if j is None: | |
| j = col(names, c) | |
| assert j is not None, f"{layout_dir}: no P column for {c} in {names}" | |
| return j | |
| return {c: pa["p_onset_p"][:, ti, pcol(c)].astype(np.float64) for c in P_CHAINS} | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--regime", required=True, choices=["manual", "engaged"]) | |
| ap.add_argument("--a", required=True) | |
| ap.add_argument("--b", default="") | |
| ap.add_argument("--b-margin", action="store_true", help="B = the current-margin rule score S0 of targets_pressure") | |
| ap.add_argument("--b-persistence", action="store_true", help="only the base-rate (persistence) rows") | |
| ap.add_argument("--split", default="test") | |
| ap.add_argument("--tag", default="") | |
| ap.add_argument("--n-boot", type=int, default=1000) | |
| ap.add_argument("--min-rows", type=int, default=1000) | |
| ap.add_argument("--min-pos", type=int, default=20) | |
| ap.add_argument("--no-append", action="store_true", help="smoke: print the cells, append nothing to results/paired_compare.csv, no run.json") | |
| a = ap.parse_args() | |
| assert sum([bool(a.b), a.b_margin, a.b_persistence]) == 1, "exactly one of --b / --b-margin / --b-persistence" | |
| t0 = time.time() | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| guards = [layout_guard(a.a, a.regime, a.split)] + ([layout_guard(a.b, a.regime, a.split)] if a.b else []) | |
| for g in guards: | |
| print(guard_line(g), flush=True) | |
| tg, pops = populations(a.regime, a.split) | |
| sa_all = p_scores(a.a, a.regime, a.split, tg) | |
| na = Path(a.a).name | |
| sb_all, nb = None, "" | |
| if a.b: | |
| sb_all, nb = p_scores(a.b, a.regime, a.split, tg), Path(a.b).name | |
| elif a.b_margin: | |
| nb = "margin_rule_S0" | |
| else: | |
| nb = "persistence" | |
| rows = [] | |
| for c in P_CHAINS: | |
| pop = pops[c] | |
| sel, y = pop["sel"], pop["y"] | |
| if a.b_persistence: | |
| rows.append({"run_stamp": stamp, "tag": a.tag, "regime": a.regime, "split": a.split, "cell": f"p_onset|{c}|3s", "metric": "base_rate", "a": na, "b": "persistence", | |
| "value_a": float(y.mean()) if len(y) else "", "ci_a_lo": "", "ci_a_hi": "", "value_b": float(y.mean()) if len(y) else "", "ci_b_lo": "", "ci_b_hi": "", | |
| "delta": "", "delta_ci_lo": "", "delta_ci_hi": "", "frac_boot_d_gt0": "", "n": int(len(sel)), "n_pos": int(y.sum()), "n_dev": int(len(np.unique(pop["dongle"]))), "n_boot": 0}) | |
| print(f" p_onset|{c}|3s base_rate {float(y.mean()) if len(y) else float('nan'):.4f} n={len(sel)} pos={int(y.sum())}", flush=True) | |
| continue | |
| sa = sa_all[c][sel] | |
| sb = sb_all[c][sel] if sb_all is not None else pop["margin"].astype(np.float64) | |
| keep = np.isfinite(sa) & np.isfinite(sb) | |
| if keep.sum() < a.min_rows or y[keep].sum() < a.min_pos or y[keep].sum() >= keep.sum(): | |
| print(f" p_onset|{c}|3s skipped (n={int(keep.sum())}, pos={int(y[keep].sum())})", flush=True) | |
| continue | |
| res = boot_pair(y[keep], sa[keep], sb[keep], pop["dongle"][keep], a.n_boot, 0, "ap") | |
| rows.append({"run_stamp": stamp, "tag": a.tag, "regime": a.regime, "split": a.split, "cell": f"p_onset|{c}|3s", "metric": "pr_auc", "a": na, "b": nb, | |
| "value_a": res["value_a"], "ci_a_lo": res["ci_a"][0], "ci_a_hi": res["ci_a"][1], "value_b": res["value_b"], "ci_b_lo": res["ci_b"][0], "ci_b_hi": res["ci_b"][1], | |
| "delta": res["delta"], "delta_ci_lo": res["ci_d"][0], "delta_ci_hi": res["ci_d"][1], "frac_boot_d_gt0": res["frac_gt0"], "n": int(keep.sum()), "n_pos": int(y[keep].sum()), | |
| "n_dev": res["n_dev"], "n_boot": a.n_boot}) | |
| print(f" p_onset|{c}|3s A={res['value_a']:.4f} B={res['value_b']:.4f} Δ={res['delta']:+.4f} [{res['ci_d'][0]:+.4f},{res['ci_d'][1]:+.4f}] n={int(keep.sum())} pos={int(y[keep].sum())}", flush=True) | |
| if a.no_append: | |
| print(f"no-append: {len(rows)} rows computed, nothing written ({time.time() - t0:.0f}s)") | |
| return | |
| new = not OUT.exists() | |
| with open(OUT, "a", newline="") as f: | |
| w = csv.DictWriter(f, fieldnames=COLS) | |
| if new: | |
| w.writeheader() | |
| for r in rows: | |
| w.writerow(r) | |
| print(f"appended {len(rows)} rows to {OUT} ({time.time() - t0:.0f}s)") | |
| write_run_json(CWM3 / "results", "pressure_eval", [Path(__file__), CWM3 / "eval" / "paired_compare.py"], | |
| [Path(a.a) / f"pred_layout_{a.regime}_{a.split}.npz"] + ([Path(a.b) / f"pred_layout_{a.regime}_{a.split}.npz"] if a.b else []), seed=0, t_start=t0, | |
| input_listing=False, extra={"run_stamp": stamp, "tag": a.tag, "regime": a.regime, "a": a.a, "b": a.b or nb, "n_rows": len(rows), "dms_fields_used": [], | |
| "layout_guard": guards}) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 9.56 kB
- Xet hash:
- 9fc497c049d7e54cd2c4eeb9b6769b4eef7a1116de0e78e9abbd1903eb3c8103
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.