Buckets:
| """cwm3/eval/paired_compare.py — paired comparison of two pred_layout files on the rows BOTH scored (device-cluster bootstrap). | |
| eval_grid.py pairs only its own LEARNED arms and its per-row mean cells (y_mae / y_crps / nr_brier / f_cos) use each arm's own finite | |
| rows — so an arm that predicts a row subset (V0 can8s: x8_ok rows only) is not comparable there. This script restricts every cell to the | |
| intersection of finite rows of A and B and reports A, B, A − B with a shared-resample cluster bootstrap (n_boot 1000, seed 0): | |
| onset cells : populations.P_veh / P_drv / P_dem exactly as eval_grid (PR-AUC@3s, plus any tau present) for events both files emit | |
| mean cells : y_mae / y_crps (per channel), f_cos (per horizon), nr_p Brier vs NoResp (NoRespValid & k + 15 <= 24) | |
| Rows appended to results/paired_compare.csv (append-only). CPU only, reads targets + the two layouts. | |
| python paired_compare.py --regime manual --a results/runs/V0_can8s_s0 --b results/runs/V0_can4s_s0 [--split test] [--tag exp2] | |
| """ | |
| 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 | |
| CWM2 = EXPLORE / "cwm2" | |
| for p in (CWM2 / "eval", CWM2 / "data", EXPLORE / "s12_wm", CWM3 / "common"): | |
| sys.path.insert(0, str(p)) | |
| import populations as P # noqa: E402 | |
| from pred_layout import load_pred, col, TAU_NAME # 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 pagecache import drop_all as _drop_cache # noqa: E402 | |
| OUT = CWM3 / "results" / "paired_compare.csv" | |
| def layout_guard(run_dir, regime, split="test"): | |
| """Refuse stale / incomplete layouts (coordinator guard 2026-09-11, after the I1 v2 pairing rows were computed from first-attempt layouts): | |
| the run dir must hold TRAIN_DONE and PRED_DONE and the layout file; the layout meta's model_id checkpoint must equal the run's BEST, the meta | |
| cfg (recorded at layout time) must hash-match the run's config.json cfg, and for joint runs meta.base_ckpts must equal base_ckpts.json. | |
| -> provenance dict {layout, mtime_utc, source_sha256 (= meta.inputs_digest, the pred pack the layout was built from), model_id, cfg_sha, code_sha256}. | |
| Raises SystemExit with a clear message otherwise.""" | |
| import hashlib, json, os, time | |
| rd = Path(run_dir) | |
| lay = rd / f"pred_layout_{regime}_{split}.npz" | |
| problems = [f"{m} missing" for m, p in (("TRAIN_DONE", rd / "TRAIN_DONE"), ("PRED_DONE", rd / "PRED_DONE"), (lay.name, lay), ("config.json", rd / "config.json"), ("BEST", rd / "BEST")) if not p.exists()] | |
| if problems: | |
| raise SystemExit(f"LAYOUT GUARD: refusing {rd.name} ({regime}/{split}): " + ", ".join(problems) + " — the run is not finished (or was archived); nothing paired") | |
| meta = json.loads(str(np.load(lay, allow_pickle=True)["meta"])) | |
| sha = lambda o: hashlib.sha256(json.dumps(o, sort_keys=True, default=str).encode()).hexdigest() | |
| cfg_run = json.load(open(rd / "config.json"))["cfg"] | |
| best = (rd / "BEST").read_text().strip() | |
| mid = str(meta.get("model_id", "")) | |
| ck = mid.split(":", 1)[1] if ":" in mid else "" | |
| if ck != best: | |
| raise SystemExit(f"LAYOUT GUARD: {lay} was written from checkpoint {ck!r} but the run's BEST is {best!r} — stale layout; re-run the prediction / layout step") | |
| if sha(meta.get("cfg", {})) != sha(cfg_run): | |
| raise SystemExit(f"LAYOUT GUARD: {lay} meta.cfg (sha {sha(meta.get('cfg', {}))[:16]}) != {rd.name}/config.json cfg (sha {sha(cfg_run)[:16]}) — the layout belongs to another configuration of this dir") | |
| if "base_ckpts" in meta and (rd / "base_ckpts.json").exists(): | |
| rec = json.load(open(rd / "base_ckpts.json")).get("base_ckpts", {}) | |
| if meta["base_ckpts"] != rec: | |
| raise SystemExit(f"LAYOUT GUARD: {lay} meta.base_ckpts {meta['base_ckpts']} != {rd.name}/base_ckpts.json {rec}") | |
| return {"run": rd.name, "layout": str(lay), "mtime_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(os.path.getmtime(lay))), | |
| "source_sha256": str(meta.get("inputs_digest", "")), "model_id": mid, "cfg_sha": sha(cfg_run)[:16], "code_sha256": str(meta.get("code_sha256", ""))[:16], | |
| "base_ckpts": meta.get("base_ckpts", {})} | |
| def guard_line(g): | |
| return f"[layout guard] {g['run']}: layout mtime {g['mtime_utc']} source sha256 {g['source_sha256'][:16]} model_id {g['model_id']} cfg {g['cfg_sha']} code {g['code_sha256']}" | |
| COLS = ["run_stamp", "tag", "regime", "split", "cell", "metric", "a", "b", "value_a", "ci_a_lo", "ci_a_hi", "value_b", "ci_b_lo", "ci_b_hi", | |
| "delta", "delta_ci_lo", "delta_ci_hi", "frac_boot_d_gt0", "n", "n_pos", "n_dev", "n_boot"] | |
| VEH = {"gas_on": "gasPressed", "brake_on": "brakePressed", "steer_phase_on": "steer_phase"} | |
| def ap_weighted(y, s, w): | |
| order = np.argsort(-s, kind="mergesort") | |
| y, w = y[order], w[order] | |
| tp, fp = np.cumsum(w * y), np.cumsum(w * ~y) | |
| if tp[-1] <= 0: | |
| return np.nan | |
| prec = tp / np.maximum(tp + fp, 1e-12) | |
| rec = tp / tp[-1] | |
| drec = np.diff(np.r_[0.0, rec]) | |
| return float((prec * drec).sum()) | |
| def boot_pair(y, sa, sb, dongle, n_boot, seed, kind, wa=None): | |
| """kind 'ap' (y bool, sa/sb scores) or 'mean' (sa/sb per-row values, y unused). -> dict.""" | |
| uniq, dev = np.unique(dongle, return_inverse=True) | |
| rng = np.random.default_rng(seed) | |
| ones = np.ones(len(sa)) | |
| if kind == "ap": | |
| pa, pb = ap_weighted(y, sa, ones), ap_weighted(y, sb, ones) | |
| else: | |
| pa, pb = float(sa.mean()), float(sb.mean()) | |
| da, db = np.full(n_boot, np.nan), np.full(n_boot, np.nan) | |
| for i in range(n_boot): | |
| cnt = np.bincount(rng.integers(0, len(uniq), len(uniq)), minlength=len(uniq)).astype(np.float64) | |
| w = cnt[dev] | |
| if kind == "ap": | |
| if (w * y).sum() <= 0 or (w * y).sum() >= w.sum(): | |
| continue | |
| da[i], db[i] = ap_weighted(y, sa, w), ap_weighted(y, sb, w) | |
| else: | |
| da[i], db[i] = (w * sa).sum() / w.sum(), (w * sb).sum() / w.sum() | |
| d = da - db | |
| ok = np.isfinite(d) | |
| pct = lambda v: (float(np.nanpercentile(v, 2.5)), float(np.nanpercentile(v, 97.5))) | |
| return {"value_a": pa, "value_b": pb, "ci_a": pct(da), "ci_b": pct(db), "delta": pa - pb, "ci_d": pct(d), | |
| "frac_gt0": float((d[ok] > 0).mean()) if ok.any() else np.nan, "n_dev": int(len(uniq))} | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--regime", required=True, choices=["manual", "engaged"]) | |
| ap.add_argument("--a", required=True) | |
| ap.add_argument("--b", required=True) | |
| ap.add_argument("--split", default="test") | |
| ap.add_argument("--tag", default="") | |
| ap.add_argument("--n-boot", type=int, default=1000) | |
| ap.add_argument("--taus", default="3s") | |
| a = ap.parse_args() | |
| 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)] # abort on stale / unfinished layouts (2026-09-11) | |
| for g in guards: | |
| print(guard_line(g), flush=True) | |
| data = P.load_all(a.regime, a.split) | |
| tg, kin, dm = data["targets"], data["kin"], data["demand"] | |
| N = len(tg["t"]) | |
| pa = load_pred(Path(a.a) / f"pred_layout_{a.regime}_{a.split}.npz", tg) | |
| pb = load_pred(Path(a.b) / f"pred_layout_{a.regime}_{a.split}.npz", tg) | |
| na, nb = Path(a.a).name, Path(a.b).name | |
| _drop_cache(list(data.get("files", [])) + [Path(a.a) / f"pred_layout_{a.regime}_{a.split}.npz", Path(a.b) / f"pred_layout_{a.regime}_{a.split}.npz"]) | |
| pops = P.build_all(data, tuple(a.taus.split(","))) | |
| rows = [] | |
| def add(cell, metric, res, n, n_pos): | |
| rows.append({"run_stamp": stamp, "tag": a.tag, "regime": a.regime, "split": a.split, "cell": cell, "metric": metric, "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": n, "n_pos": n_pos, "n_dev": res["n_dev"], "n_boot": a.n_boot}) | |
| print(f" {cell:28s} {metric:7s} A={res['value_a']:.4f} B={res['value_b']:.4f} Δ={res['delta']:+.4f} [{res['ci_d'][0]:+.4f},{res['ci_d'][1]:+.4f}] " | |
| f"P(Δ>0)={res['frac_gt0']:.2f} n={n} pos={n_pos}", flush=True) | |
| # onset cells for events both files emit | |
| ev_a = [str(x) for x in pa["event_names"]] | |
| for name, pop in pops.items(): | |
| parts = name.split("|") | |
| if len(parts) != 3 or parts[0] not in ("drv_onset", "veh_onset", "d_onset"): | |
| continue | |
| kind, ev, tau = parts | |
| tau = int(tau) | |
| e = col(ev_a, f"D_{ev}" if kind == "d_onset" else ev) # demand families are stored as D_<family> in the layout | |
| if e is None or "onset_p" not in pa or "onset_p" not in pb: | |
| continue | |
| ti = list(map(int, pa["onset_taus"])).index(tau) | |
| sa = pa["onset_p"][pop.sel, ti, e].astype(np.float64) | |
| sb = pb["onset_p"][pop.sel, ti, e].astype(np.float64) | |
| keep = np.isfinite(sa) & np.isfinite(sb) | |
| if keep.sum() < 1000 or pop.y[keep].sum() < 20: | |
| continue | |
| res = boot_pair(pop.y[keep], sa[keep], sb[keep], pop.dongle[keep], a.n_boot, 0, "ap") | |
| add(f"{kind}|{ev}|{TAU_NAME.get(tau, tau)}", "pr_auc", res, int(keep.sum()), int(pop.y[keep].sum())) | |
| # per-row means | |
| dong = tg["dongle"] | |
| for key, names in (("y_mae", ("steeringAngleDeg", "vEgo", "aEgo")), ("y_crps", ("steeringAngleDeg", "vEgo", "aEgo")), ("f_cos", tuple(f"+{i}s" for i in range(1, 6)))): | |
| if key in pa and key in pb: | |
| for j, nm in enumerate(names): | |
| va, vb = pa[key][:, j].astype(np.float64), pb[key][:, j].astype(np.float64) | |
| keep = np.isfinite(va) & np.isfinite(vb) | |
| if keep.sum() < 1000: | |
| continue | |
| res = boot_pair(None, va[keep], vb[keep], dong[keep], a.n_boot, 0, "mean") | |
| add(f"{key}|{nm}", "mean", res, int(keep.sum()), "") | |
| if "nr_p" in pa and "nr_p" in pb: | |
| nr = dm["NoResp"].astype(np.float64) | |
| valid = dm["NoRespValid"] & (np.arange(25)[None, :] + 15 <= 24) | |
| va, vb = pa["nr_p"].astype(np.float64), pb["nr_p"].astype(np.float64) | |
| ok = valid & np.isfinite(va) & np.isfinite(vb) | |
| rows_ok = ok.any(1) | |
| ba = np.where(ok, (va - nr) ** 2, 0).sum(1)[rows_ok] / ok.sum(1)[rows_ok] | |
| bb = np.where(ok, (vb - nr) ** 2, 0).sum(1)[rows_ok] / ok.sum(1)[rows_ok] | |
| res = boot_pair(None, ba, bb, dong[rows_ok], a.n_boot, 0, "mean") | |
| add("nr_brier|NoResp", "brier", res, int(rows_ok.sum()), "") | |
| 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", "paired_compare", [Path(__file__), CWM2 / "eval" / "populations.py"], | |
| [Path(a.a) / f"pred_layout_{a.regime}_{a.split}.npz", Path(a.b) / f"pred_layout_{a.regime}_{a.split}.npz"], seed=0, t_start=t0, | |
| input_listing=False, extra={"run_stamp": stamp, "tag": a.tag, "regime": a.regime, "a": a.a, "b": a.b, "n_rows": len(rows), "dms_fields_used": [], | |
| "layout_guard": guards}) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 12.1 kB
- Xet hash:
- 3a601e38873786d8496339d9e638a0e0d78de3e0a7355539a4060c3932cc6cd0
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.