Buckets:
| """cwm3/eval/gold_extended.py — the pre-registered SECONDARY gold stratum `clean_test_r2` for the warning-head claims (roadmap §B3 item 4; 2026-09-12). | |
| Primary claim: unchanged, on the 509-row / 63-positive clean gold (cwm2/eval/warning_gold.py). Secondary (pre-registered here, before any round-2 label | |
| exists): the round-2 human consensus labels on TEST-split devices (cwm2/annotation_human/h5_to_consensus.py -> results/W1d/h5_consensus_latest.csv; two | |
| blind raters + adjudication; unsure excluded; idx not already in the 568-episode gold) anchored with the UNCHANGED warning_gold.build_gold (legacy rows do | |
| not exist for them -> the s12-latest anchor inside [t0 − 2, t0 + 0.2]) form `clean_test_r2`; `clean_test_pooled` = the 509 gold rows ∪ r2. For every head | |
| column of the given W layouts (results/runs/<w_run>/pred_layout_{regime}_{test,val}.npz) and the DMS att@3s chain recomputed at the same anchors from the | |
| legacy marginals (compose.legacy_compose att_only — exactly as train_warning_head / operating_points do), AUROC with device-cluster CI (warning_gold.cluster_boot) | |
| on r2 alone and pooled, and the paired Δ head − DMS on identical resamples. Rows are appended to results/warning_gold/gold_extended.csv and a section to | |
| results/tables/power_gold.md. With no consensus file yet the script reports the empty stratum and exits 0; --self-test exercises the anchoring / scoring path on | |
| 40 EXISTING gold episodes re-labelled as `r2` (their AUROC must equal warning_gold's on the same rows within 1e-6) and writes nothing. | |
| python eval/gold_extended.py --w-runs W_I1_all_minus_road2drv_s0_final_dwfc --seeds 0 --cols W_learned_curhgb_t15_w10_n979dwfc [--self-test] | |
| No openpilot DMS field is read (the DMS chain enters as the labelled legacy comparator, 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 (CWM3 / "train", CWM3 / "data", CWM3 / "eval", CWM3 / "common", CWM2 / "eval", CWM2 / "data", CWM2, CWM2 / "baselines" / "legacy_cockpitwm"): | |
| sys.path.insert(0, str(p)) | |
| import train_warning_head as TW # noqa: E402 (LB, WG; nothing else) | |
| import warning_gold as WG # noqa: E402 | |
| from compose import legacy_compose # noqa: E402 | |
| from runmeta import write_run_json # noqa: E402 | |
| LB = TW.LB | |
| REGIMES = ("manual", "engaged") | |
| CONSENSUS = CWM2 / "results" / "W1d" / "h5_consensus_latest.csv" | |
| S3_ANCHORS = EXPLORE / "s3_can_baseline" / "results" / "anchors.parquet" | |
| DMS_DIR = CWM2 / "baselines" / "legacy_cockpitwm" / "results" | |
| OUT_CSV = CWM3 / "results" / "warning_gold" / "gold_extended.csv" | |
| TABLE = CWM3 / "results" / "tables" / "power_gold.md" | |
| TAU = 15 | |
| def split_of_dongles(dongles): | |
| a3 = pd.read_parquet(S3_ANCHORS, columns=["dongle", "split"]).drop_duplicates("dongle") | |
| known = dict(zip(a3.dongle, a3.split)) | |
| return np.array([known.get(d, "unknown") for d in dongles]) | |
| def r2_label_frame(EP, consensus_csv): | |
| """round-2 consensus rows on TEST devices, not in the 568-episode gold -> lab frame for warning_gold.build_gold.""" | |
| if not Path(consensus_csv).exists(): | |
| return pd.DataFrame(columns=["idx", "y", "dongle", "uid", "t0", "ep_regime", "clean_test", "clean_val", "label"]) | |
| C = pd.read_csv(consensus_csv) | |
| C = C[C.status.isin(["consensus", "adjudicated", "panel_consensus"]) & C.y.notna() & C.uid.notna() & C.t0.notna()] # panel_consensus: committee mode (2026-09-13 deviation) | |
| C = C[~C.idx.isin(EP.idx)] | |
| C["split_dev"] = split_of_dongles(C.dongle.astype(str)) | |
| C = C[C.split_dev == "test"] | |
| return pd.DataFrame(dict(idx=C.idx.astype(int), y=C.y.astype(int), dongle=C.dongle.astype(str), uid=C.uid.astype(str), t0=C.t0.astype(float), ep_regime=C.regime.fillna("manual").astype(str), | |
| clean_test=True, clean_val=False, label=np.where(C.y == 1, "warranted", "not_warranted"))) | |
| def dms_at_anchors(G, seed): | |
| """legacy att@3s chain at the anchors of G (test split rows) from the marginals files; NaN where unavailable.""" | |
| out = np.full(len(G), np.nan) | |
| for r in REGIMES: | |
| p = Path(DMS_DIR) / f"marginals_{r}_s{seed}.npz" | |
| if not p.exists(): | |
| continue | |
| z = np.load(p, allow_pickle=True) | |
| sc = legacy_compose(z["pA"].astype(np.float32), None, TAU)["att_only"].astype(np.float64) | |
| key = {(u, round(float(t), 1)): i for i, (u, t) in enumerate(zip(z["uid"], z["t"]))} | |
| g = G[(G.regime == r) & (G.split == "test")] | |
| for i, u, t in zip(g.index, g.uid, g.t_anchor): | |
| j = key.get((u, round(float(t), 1))) | |
| if j is not None: | |
| out[i] = sc[j] | |
| return out | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--w-runs", nargs="+", required=True) | |
| ap.add_argument("--seeds", nargs="+", type=int, required=True) | |
| ap.add_argument("--cols", nargs="+", required=True) | |
| ap.add_argument("--consensus", default=str(CONSENSUS)) | |
| ap.add_argument("--n-boot", type=int, default=2000) | |
| ap.add_argument("--self-test", action="store_true") | |
| a = ap.parse_args() | |
| assert len(a.w_runs) == len(a.seeds) | |
| t0 = time.time() | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| EP = LB.episode_labels() | |
| lab568 = EP[EP.label != "unsure"].reset_index(drop=True) | |
| packs = {(r, s): WG.load_pack(r, s) for r in REGIMES for s in ("test", "val")} | |
| if a.self_test: | |
| sub = lab568[lab568.clean_test].head(40).copy() | |
| r2 = pd.DataFrame(dict(idx=sub.idx, y=sub.y, dongle=sub.dongle, uid=sub.uid, t0=sub.t0, ep_regime=sub.ep_regime, clean_test=True, clean_val=False, label=sub.label)) | |
| print(f"self-test: {len(r2)} existing gold episodes used as a fake r2 stratum (nothing is written)") | |
| else: | |
| r2 = r2_label_frame(EP, a.consensus) | |
| print(f"round-2 consensus rows eligible for clean_test_r2: {len(r2)} ({int(r2.y.sum()) if len(r2) else 0} warranted) from {a.consensus}") | |
| rows = [] | |
| for w_run, seed in zip(a.w_runs, a.seeds): | |
| legacy = WG.merge_legacy(seed) | |
| G568 = WG.build_gold(lab568, packs, legacy, "legacy_first") | |
| G568 = G568[G568.clean_test].reset_index(drop=True) | |
| G568["stratum_src"] = "gold509" | |
| G568["att"] = legacy.reindex(G568.idx)["att@3s"].to_numpy(float) if legacy is not None else np.nan | |
| if len(r2): | |
| Gr2 = WG.build_gold(r2, packs, None, "s12") # no legacy rows exist for round-2 clips -> s12 anchor rule (unchanged function) | |
| Gr2 = Gr2[Gr2.split == "test"].reset_index(drop=True) | |
| Gr2["stratum_src"] = "r2" | |
| Gr2["att"] = dms_at_anchors(Gr2, seed) | |
| if a.self_test: | |
| # the fake r2 rows must reproduce the gold anchors of the same episodes (legacy_first may move an anchor; compare on the s12 rule) | |
| chk = WG.build_gold(lab568[lab568.idx.isin(r2.idx)], packs, None, "s12") | |
| assert len(Gr2) == len(chk) and np.array_equal(np.sort(Gr2.row.to_numpy()), np.sort(chk.row.to_numpy())), "self-test anchoring mismatch" | |
| print(f"self-test: {len(Gr2)} of {len(r2)} anchored on the test split; anchors identical to build_gold(s12) on the same episodes") | |
| else: | |
| Gr2 = G568.iloc[:0].copy() | |
| lay = {(r, s): np.load(CWM3 / "results" / "runs" / w_run / f"pred_layout_{r}_{s}.npz", allow_pickle=True) for r in REGIMES for s in ("test", "val")} | |
| for col in a.cols: | |
| def score(G): | |
| v = np.full(len(G), np.nan) | |
| for (r, s), g in G.groupby(["regime", "split"]): | |
| z = lay[(r, s)] | |
| names = list(map(str, z["w_names"])) | |
| if col not in names: | |
| continue | |
| vals = z["w"][g.row.to_numpy(), names.index(col)].astype(np.float64) | |
| if "w_mask" in z.files: | |
| vals = np.where(z["w_mask"][g.row.to_numpy()], vals, np.nan) | |
| v[g.index.to_numpy()] = vals | |
| return v | |
| for sname, G in (("clean_test_r2", Gr2), ("clean_test_pooled", pd.concat([G568, Gr2], ignore_index=True)), ("clean_test", G568)): | |
| if len(G) < 20 or G.y.nunique() < 2: | |
| rows.append(dict(run_stamp=stamp, w_run=w_run, seed=seed, column=col, stratum=sname, n=int(len(G)), n_pos=int(G.y.sum()) if len(G) else 0, note="stratum empty / too small")) | |
| continue | |
| s = score(G) | |
| att = G["att"].to_numpy(float) | |
| y, dong = G.y.to_numpy().astype(int), G.dongle.to_numpy() | |
| fin = np.isfinite(s) | |
| bs = WG.cluster_boot(y[fin], {"h": s[fin]}, dong[fin], a.n_boot, 0)["h"]["auroc"] | |
| rec = dict(run_stamp=stamp, w_run=w_run, seed=seed, column=col, stratum=sname, n=int(fin.sum()), n_pos=int(y[fin].sum()), n_dev=int(len(np.unique(dong[fin]))), | |
| n_r2=int((G.stratum_src == "r2").sum()), auroc=WG.fast_auroc(y[fin], s[fin]), ci_lo=WG.pct(bs)[0], ci_hi=WG.pct(bs)[1], note="self-test" if a.self_test else "") | |
| both = fin & np.isfinite(att) | |
| if both.sum() >= 20 and y[both].min() != y[both].max(): | |
| b2 = WG.cluster_boot(y[both], {"h": s[both], "d": att[both]}, dong[both], a.n_boot, 0) | |
| d = b2["h"]["auroc"] - b2["d"]["auroc"] | |
| rec.update(auroc_dms=WG.fast_auroc(y[both], att[both]), delta_vs_dms=WG.fast_auroc(y[both], s[both]) - WG.fast_auroc(y[both], att[both]), delta_ci_lo=WG.pct(d)[0], delta_ci_hi=WG.pct(d)[1], n_paired=int(both.sum())) | |
| rows.append(rec) | |
| print(f" {w_run} {col} {sname}: AUROC {rec['auroc']:.4f} [{rec['ci_lo']:.3f},{rec['ci_hi']:.3f}] n={rec['n']}/{rec['n_pos']}" | |
| + (f"; Δ vs DMS {rec['delta_vs_dms']:+.4f} [{rec['delta_ci_lo']:+.3f},{rec['delta_ci_hi']:+.3f}]" if "delta_vs_dms" in rec else "")) | |
| df = pd.DataFrame(rows) | |
| if a.self_test: | |
| ref = df[(df.stratum == "clean_test") & (df.column == a.cols[0])].auroc.iloc[0] | |
| print(f"self-test OK: clean_test AUROC of {a.cols[0]} on the 509 gold rows = {ref:.4f} (compare with warning_gold.csv); nothing written") | |
| return | |
| hdr = not OUT_CSV.exists() | |
| df.to_csv(OUT_CSV, mode="a", header=hdr, index=False) | |
| L = [f"## clean_test_r2 secondary stratum (eval/gold_extended.py, run {stamp})", "", | |
| f"Round-2 consensus rows eligible: {len(r2)} ({int(r2.y.sum()) if len(r2) else 0} warranted) from `{Path(a.consensus).name}`; primary claim unchanged on the 509-row gold; " | |
| "pooled = 509 ∪ r2 (pre-registered secondary). Anchoring by the unchanged `warning_gold.build_gold`; DMS att@3s at the same anchors from the legacy marginals.", "", | |
| "| layout | seed | column | stratum | n / pos (r2 rows) | AUROC [CI] | DMS AUROC | Δ vs DMS [CI] |", "|---|---|---|---|---|---|---|---|"] | |
| for r in df.itertuples(): | |
| if getattr(r, "note", "") == "stratum empty / too small": | |
| L.append(f"| {r.w_run} | {r.seed} | {r.column} | {r.stratum} | {r.n} / {r.n_pos} | — (empty: no round-2 consensus labels yet) | | |") | |
| else: | |
| L.append(f"| {r.w_run} | {r.seed} | {r.column} | {r.stratum} | {r.n} / {r.n_pos} ({getattr(r, 'n_r2', 0)}) | {r.auroc:.3f} [{r.ci_lo:.2f},{r.ci_hi:.2f}] | " | |
| f"{getattr(r, 'auroc_dms', np.nan):.3f} | {getattr(r, 'delta_vs_dms', np.nan):+.3f} [{getattr(r, 'delta_ci_lo', np.nan):+.3f},{getattr(r, 'delta_ci_hi', np.nan):+.3f}] |") | |
| L.append("") | |
| with open(TABLE, "a") as f: | |
| f.write("\n".join(L) + "\n") | |
| print("\n".join(L)) | |
| write_run_json(CWM3 / "results" / "warning_gold", "gold_extended", [Path(__file__), Path(WG.__file__)], [Path(a.consensus)] if Path(a.consensus).exists() else [], seed=0, t_start=t0, | |
| input_listing=False, extra={"w_runs": a.w_runs, "seeds": a.seeds, "cols": a.cols, "n_r2": int(len(r2)), "rows": rows, "dms_fields_used": []}) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 12.3 kB
- Xet hash:
- 7e245bdca983c7062f3d848e8a778feda922c02ff0812a9a25ec0689457c087d
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.