Buckets:
| """cwm3/eval/warning_head_variant_diag.py — FIXED-HGB feature-variant diagnosis of the learned warning head on the final joint runs (roadmap §B3 item 0 | |
| "E7 degradation" + the univariate check of item 1a / 1b; 2026-09-12). CPU, minutes. | |
| The same HistGradientBoosting head (the hp the CV selected for the base current-state set: read from the latest base warning-head dir's selected.json, | |
| default lr 0.05 / 100 iters, depth 3, monotone constraints of train_warning_head.monotone_constraints) is fitted on the 979 clean TRAIN consensus labels | |
| for every variant and scored on the clean-test gold (unchanged warning_gold.build_gold anchors, legacy_first). Variants are column selections / | |
| substitutions of ONE feature matrix per seed (train/train_warning_head.features_from_pred with the dwell / mr [/ face] blocks appended), so every paired | |
| Δ (variant − base) is on identical rows and identical device-cluster resamples (warning_gold.cluster_boot), per stratum clean_test / |manual / |engaged: | |
| base current-state features of the final head (K0 from the run's causal v2 pack; K0_valid bit; HGB handles NaN) | |
| K0old K0 (+ K0_valid) replaced by the cwm2 route-median pack of the v1 tables (NON-causal ROI; diagnostic only) | |
| K0nan_suspect K0 -> NaN and K0_valid -> 0 on the 64 driver-box-suspect routes (= the final-assembly agent's `_k0m` row, fixed head) | |
| rowmask rows on the suspect routes removed from train AND gold (Δ vs base computed on the same reduced gold rows) | |
| noK0 K0 block dropped | |
| K0rel only the reference-relative K0 channels kept (head_turn, head_turn_dev, L/R/min hand rest distance; absolute head / wrist positions dropped) | |
| dw / mr / dwmr / fc / dwfc / dwfcmr base + the extended blocks (train/warning_features_ext.py) when present | |
| Univariate: AUROC (device-cluster CI) of every extended feature on the 979 train labels (pooled / per regime) and, descriptively, on the gold. | |
| Outputs (append-only): results/warning_head_ext/variant_diag_<stamp>/{variants.csv,paired.csv,univariate.csv,run_*.json} and a markdown block printed | |
| for results/tables/warning_head_final_ext.md. No openpilot DMS field is read. | |
| python eval/warning_head_variant_diag.py --runs I1_all_minus_road2drv_s0_final I1_all_minus_road2drv_s1_final I1_all_minus_road2drv_s2_final --sets dwell,mr | |
| """ | |
| from __future__ import annotations | |
| import os | |
| os.environ.setdefault("OMP_NUM_THREADS", "4") | |
| os.environ.setdefault("MKL_NUM_THREADS", "4") | |
| import argparse # noqa: E402 | |
| import json # noqa: E402 | |
| import sys # noqa: E402 | |
| import time # noqa: E402 | |
| from pathlib import Path # noqa: E402 | |
| import numpy as np # noqa: E402 | |
| import pandas as pd # noqa: E402 | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| for p in (CWM3 / "train", CWM3 / "data", CWM3 / "eval", CWM3 / "common"): | |
| sys.path.insert(0, str(p)) | |
| import train_warning_head as TW # noqa: E402 | |
| import warning_features_ext as WFX # noqa: E402 | |
| import cwm3_dataset as ds # noqa: E402 | |
| from runmeta import write_run_json # noqa: E402 | |
| WG, LB = TW.WG, TW.LB | |
| REGIMES = ("manual", "engaged") | |
| SUSPECTS = CWM3 / "results" / "targets_kin_causal_routes_v2" / "driver_box_suspect_routes.json" | |
| OUT_ROOT = CWM3 / "results" / "warning_head_ext" | |
| K0_ABS = {"head_cx", "head_cy", "Lwri_x", "Lwri_y", "Rwri_x", "Rwri_y"} | |
| def f3(x, nd=3): | |
| return "—" if x is None or not np.isfinite(x) else f"{x:.{nd}f}" | |
| def base_hp(run): | |
| """the CV-selected HGB hp of the base current-state head of this run (latest base dir), default lr 0.05 / 100 iters.""" | |
| for d in sorted(CWM3.glob(f"results/warning_head/{run}_*"), reverse=True): | |
| try: | |
| sel = json.loads((d / "selected.json").read_text()) | |
| except Exception: # noqa: BLE001 | |
| continue | |
| if sel.get("w_suffix", "") == "" and sel.get("k0_source", "run") == "run" and not sel.get("k0_mask_routes") and not sel.get("extra_features"): | |
| hp = sel["feature_sets"]["current"]["per_model"]["hgb"]["hp"] | |
| return {"lr": hp["lr"], "iters": hp["iters"]}, str(d) | |
| return {"lr": 0.05, "iters": 100}, "default" | |
| def load_run(run, sets, demand="P"): | |
| """-> dict with train X / y / groups / uid, gold frame G + Xg, names, K0old block (train + gold), suspect flags.""" | |
| rd = CWM3 / "results" / "runs" / run | |
| cfg = json.loads((rd / "config.json").read_text()) | |
| seed_run = int(cfg["args"]["seed"]) | |
| ds.set_kin_source(cfg["cfg"].get("kin_source", "cwm2") or "cwm2", cfg["cfg"].get("kin_version", "v1") or "v1") | |
| ds.set_pack_version(cfg["cfg"].get("pack_version", "v1") or "v1") | |
| TW.EXTRA_OPTS["sets"] = tuple(sets) | |
| TW.K0_OPTS["source"], TW.K0_OPTS["mask_uids"] = "run", frozenset() | |
| tab = LB.train_label_table(sources=("consensus",)) | |
| Xs, ys, gs, us, k0o, rg = [], [], [], [], [], [] | |
| for regime in REGIMES: | |
| pr = np.load(rd / f"pred_{regime}_train_labelrows.npz", allow_pickle=True) | |
| posmap = {int(r): i for i, r in enumerate(pr["rows"].astype(np.int64))} | |
| sub = tab[tab.regime == regime] | |
| rows = sub.row.to_numpy().astype(np.int64) | |
| pos = np.array([posmap[int(r)] for r in rows], np.int64) | |
| Xs.append(TW.features_from_pred(pr, regime, "train", rows, pos, demand)) | |
| ys.append(sub.y.to_numpy().astype(int)) | |
| gs.append(sub.dongle.to_numpy()) | |
| us.append(sub.uid.to_numpy()) | |
| rg.append(np.full(len(sub), regime)) | |
| TW.K0_OPTS["source"] = "cwm2" | |
| _, _, _, k0v_o, _, _, _, _, K0_o = TW._aux(regime, "train", rows) | |
| TW.K0_OPTS["source"] = "run" | |
| k0o.append(np.concatenate([K0_o, k0v_o[:, None]], 1)) | |
| X, y, groups, uid_tr, K0old_tr = np.concatenate(Xs), np.concatenate(ys), np.concatenate(gs), np.concatenate(us), np.concatenate(k0o) | |
| names = TW.feature_names(demand) | |
| EP = LB.episode_labels() | |
| lab = 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")} | |
| G = WG.build_gold(lab, packs, WG.merge_legacy(seed_run), "legacy_first") | |
| assert not set(G[G.clean_test].idx) & set(tab.idx) | |
| preds = {(r, s): np.load(rd / f"pred_{r}_{s}.npz", allow_pickle=True) for r in REGIMES for s in ("test", "val")} | |
| Xg = np.full((len(G), X.shape[1]), np.nan, np.float32) | |
| K0old_g = np.full((len(G), 12), np.nan, np.float32) | |
| for (r, s), g in G.groupby(["regime", "split"]): | |
| rows = g.row.to_numpy().astype(np.int64) | |
| Xg[g.index.to_numpy()] = TW.features_from_pred(preds[(r, s)], r, s, rows, None, demand) | |
| TW.K0_OPTS["source"] = "cwm2" | |
| _, _, _, k0v_o, _, _, _, _, K0_o = TW._aux(r, s, rows) | |
| TW.K0_OPTS["source"] = "run" | |
| K0old_g[g.index.to_numpy()] = np.concatenate([K0_o, k0v_o[:, None]], 1) | |
| sus = frozenset(r["uid"] for r in json.loads(SUSPECTS.read_text())["flagged"]) | |
| return dict(run=run, seed=seed_run, X=X, y=y, groups=groups, uid_tr=uid_tr, reg_tr=np.concatenate(rg), K0old_tr=K0old_tr, names=names, G=G, Xg=Xg, K0old_g=K0old_g, | |
| sus_tr=np.isin(uid_tr, list(sus)), sus_g=G.uid.isin(sus).to_numpy(), n_sus=len(sus)) | |
| def variants(D, sets): | |
| """-> {name: (train_mask_rows, X_train, Xg, names)} for the fixed-head variants.""" | |
| names = np.array(D["names"]) | |
| cur = ~np.array([TW.is_future(n) for n in names]) | |
| ext = np.array([n.startswith(("dw_", "mr_", "fc_")) for n in names]) | |
| base = cur & ~ext | |
| k0 = np.array([n.startswith("K0_") for n in names]) # K0_<channel> + K0_valid | |
| k0abs = np.array([n.startswith("K0_") and n[3:] in K0_ABS for n in names]) | |
| all_tr = np.ones(len(D["y"]), bool) | |
| out = {} | |
| def sel(mask, rows=all_tr, X=None, Xg=None): | |
| X = D["X"] if X is None else X | |
| Xg = D["Xg"] if Xg is None else Xg | |
| return rows, X[:, mask], Xg[:, mask], [n for n, m in zip(names, mask) if m] | |
| out["base"] = sel(base) | |
| Xo, Xgo = D["X"].copy(), D["Xg"].copy() | |
| Xo[:, k0], Xgo[:, k0] = D["K0old_tr"], D["K0old_g"] | |
| out["K0old"] = sel(base, X=Xo, Xg=Xgo) | |
| Xm, Xgm = D["X"].copy(), D["Xg"].copy() | |
| kv = list(names).index("K0_valid") | |
| kch = k0 & (names != "K0_valid") | |
| Xm[np.ix_(D["sus_tr"], kch)] = np.nan | |
| Xm[D["sus_tr"], kv] = 0.0 | |
| Xgm[np.ix_(D["sus_g"], kch)] = np.nan | |
| Xgm[D["sus_g"], kv] = 0.0 | |
| out["K0nan_suspect"] = sel(base, X=Xm, Xg=Xgm) | |
| out["rowmask"] = sel(base, rows=~D["sus_tr"]) | |
| out["noK0"] = sel(base & ~k0) | |
| out["K0rel"] = sel(base & ~k0abs) | |
| have = {s for s in ("dwell", "mr", "face") if s in sets} | |
| dw = np.array([n.startswith("dw_") for n in names]) | |
| mr = np.array([n.startswith("mr_") for n in names]) | |
| fc = np.array([n.startswith("fc_") for n in names]) | |
| if "dwell" in have: | |
| out["dw"] = sel(base | dw) | |
| if "mr" in have: | |
| out["mr"] = sel(base | mr) | |
| if {"dwell", "mr"} <= have: | |
| out["dwmr"] = sel(base | dw | mr) | |
| if "face" in have: | |
| out["fc"] = sel(base | fc) | |
| if "dwell" in have: | |
| out["dwfc"] = sel(base | dw | fc) | |
| if {"dwell", "mr"} <= have: | |
| out["dwfcmr"] = sel(base | dw | fc | mr) | |
| return out | |
| def fit_score(hp, rows, X, Xg, names, y, seed): | |
| m = TW.make_model("hgb", dict(hp, demand="P", names=names), seed).fit(X[rows], y[rows]) | |
| return m.predict_proba(Xg)[:, 1] | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--runs", nargs="+", default=[f"I1_all_minus_road2drv_s{k}_final" for k in range(3)]) | |
| ap.add_argument("--sets", default="dwell,mr", help="extended blocks to append (face requires the face pack)") | |
| ap.add_argument("--n-boot", type=int, default=1000) | |
| ap.add_argument("--seed", type=int, default=0) | |
| ap.add_argument("--label", default="") | |
| a = ap.parse_args() | |
| t0 = time.time() | |
| sets = tuple(s for s in ("dwell", "face", "mr") if s in a.sets.split(",")) | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| out = OUT_ROOT / f"variant_diag_{stamp}" | |
| out.mkdir(parents=True, exist_ok=True) | |
| rows_v, rows_p, rows_u = [], [], [] | |
| for run in a.runs: | |
| hp, hp_src = base_hp(run) | |
| D = load_run(run, sets) | |
| G, y_g = D["G"], D["G"].y.to_numpy().astype(int) | |
| gt = G.clean_test.to_numpy() | |
| dong, reg = G.dongle.to_numpy(), G.regime.to_numpy() | |
| print(f"[{run}] seed {D['seed']}: train {len(D['y'])} / {int(D['y'].sum())} pos ({int(D['sus_tr'].sum())} on suspect routes); gold clean_test {int(gt.sum())} / {int(y_g[gt].sum())} " | |
| f"({int((gt & D['sus_g']).sum())} on suspect routes); features {len(D['names'])}; fixed HGB hp {hp} from {hp_src}", flush=True) | |
| V = variants(D, sets) | |
| scores = {} | |
| for vn, (rows, X, Xg, names) in V.items(): | |
| scores[vn] = fit_score(hp, rows, X, Xg, names, D["y"], a.seed) | |
| for stratum in ("clean_test", "clean_test|manual", "clean_test|engaged"): | |
| m = gt if stratum == "clean_test" else gt & (reg == stratum.split("|")[1]) | |
| for vn in V: | |
| mm = m & ~D["sus_g"] if vn == "rowmask" else m | |
| S = {"v": scores[vn][mm], "base": scores["base"][mm]} | |
| bs = WG.cluster_boot(y_g[mm], S, dong[mm], a.n_boot, 0) | |
| au = WG.fast_auroc(y_g[mm], S["v"]) | |
| lo, hi = WG.pct(bs["v"]["auroc"]) | |
| d = bs["v"]["auroc"] - bs["base"]["auroc"] | |
| dlo, dhi = WG.pct(d) | |
| dv = au - WG.fast_auroc(y_g[mm], S["base"]) | |
| rows_v.append(dict(run=run, seed=D["seed"], stratum=stratum, variant=vn, auroc=au, ci_lo=lo, ci_hi=hi, prauc=WG.fast_ap(y_g[mm], S["v"]), n=int(mm.sum()), n_pos=int(y_g[mm].sum()), | |
| n_dev=int(len(np.unique(dong[mm]))), n_features=len(V[vn][3]), n_train=int(V[vn][0].sum()), hp=json.dumps(hp), | |
| delta_vs_base=dv, delta_ci_lo=dlo, delta_ci_hi=dhi, frac_boot_d_gt0=float(np.mean(d[np.isfinite(d)] > 0)), | |
| base_auroc_same_rows=WG.fast_auroc(y_g[mm], S["base"]))) | |
| print(f" {stratum:20s} {vn:14s} AUROC {au:.4f} [{lo:.3f},{hi:.3f}] Δ vs base {dv:+.4f} [{dlo:+.4f},{dhi:+.4f}] n={int(mm.sum())}/{int(y_g[mm].sum())}", flush=True) | |
| # univariate AUROC of the extended features (train labels, device-cluster CI; gold descriptive) | |
| names = np.array(D["names"]) | |
| ext_idx = [i for i, n in enumerate(names) if n.startswith(("dw_", "mr_", "fc_"))] | |
| if ext_idx and run == a.runs[0]: | |
| grp_tr, reg_tr = D["groups"], D["reg_tr"] | |
| for i in ext_idx: | |
| x = D["X"][:, i].astype(np.float64) | |
| fin = np.isfinite(x) | |
| rec = dict(run=run, feature=names[i], n_train_finite=int(fin.sum()), finite_frac_train=float(fin.mean())) | |
| for pop, pm in (("train", fin), ("train_manual", fin & (reg_tr == "manual")), ("train_engaged", fin & (reg_tr == "engaged"))): | |
| if pm.sum() > 50 and D["y"][pm].min() != D["y"][pm].max(): | |
| b = WG.cluster_boot(D["y"][pm], {"f": x[pm]}, grp_tr[pm], a.n_boot, 0)["f"]["auroc"] | |
| rec.update({f"auroc_{pop}": WG.fast_auroc(D["y"][pm], x[pm]), f"ci_lo_{pop}": WG.pct(b)[0], f"ci_hi_{pop}": WG.pct(b)[1]}) | |
| xg = D["Xg"][:, i].astype(np.float64) | |
| for stratum in ("clean_test", "clean_test|manual", "clean_test|engaged"): | |
| m = gt if stratum == "clean_test" else gt & (reg == stratum.split("|")[1]) | |
| mm = m & np.isfinite(xg) | |
| rec[f"auroc_gold_{stratum.split('|')[-1] if '|' in stratum else 'pooled'}"] = WG.fast_auroc(y_g[mm], xg[mm]) if mm.sum() > 20 and y_g[mm].min() != y_g[mm].max() else np.nan | |
| rows_u.append(rec) | |
| Vdf, Udf = pd.DataFrame(rows_v), pd.DataFrame(rows_u) | |
| Vdf.to_csv(out / "variants.csv", index=False) | |
| Udf.to_csv(out / "univariate.csv", index=False) | |
| # ---- markdown block | |
| L = [f"### Fixed-HGB feature-variant diagnosis of the E7 head (eval/warning_head_variant_diag.py, run {stamp}{' — ' + a.label if a.label else ''})", "", | |
| f"Runs {', '.join(a.runs)}; the SAME HGB (CV-selected hp of the base current-state head, monotone constraints) refit per variant on the 979 clean train " | |
| f"consensus labels; clean-test gold via the unchanged `warning_gold.build_gold`; AUROC with device-cluster bootstrap ({a.n_boot}); paired Δ vs `base` on identical rows / " | |
| f"resamples (rowmask: both scored on the non-suspect gold rows). Raw rows: `results/warning_head_ext/{out.name}/variants.csv`.", ""] | |
| for stratum in ("clean_test", "clean_test|manual", "clean_test|engaged"): | |
| sub = Vdf[Vdf.stratum == stratum] | |
| seeds = sorted(sub.seed.unique()) | |
| L += [f"**{stratum}** (n / pos = {int(sub.n.max())} / {int(sub.n_pos.max())})", "", "| variant | features | " + " | ".join(f"s{s} AUROC [CI]" for s in seeds) + " | mean | " + " | ".join(f"s{s} Δ vs base [CI]" for s in seeds) + " | mean Δ | CI excl. 0 |", | |
| "|---|---|" + "---|" * len(seeds) + "---|" + "---|" * len(seeds) + "---|---|"] | |
| for vn in sub.variant.unique(): | |
| g = sub[sub.variant == vn].sort_values("seed") | |
| cells = [f"{f3(r.auroc)} [{f3(r.ci_lo, 2)},{f3(r.ci_hi, 2)}]" for r in g.itertuples()] | |
| dcells = [f"{r.delta_vs_base:+.3f} [{r.delta_ci_lo:+.3f},{r.delta_ci_hi:+.3f}]" for r in g.itertuples()] | |
| excl = int(((g.delta_ci_lo > 0) | (g.delta_ci_hi < 0)).sum()) | |
| L.append(f"| {vn} | {int(g.n_features.iloc[0])} | " + " | ".join(cells) + f" | {f3(g.auroc.mean())} | " + " | ".join(dcells) + f" | {g.delta_vs_base.mean():+.3f} | {excl}/{len(g)} |") | |
| L.append("") | |
| if len(Udf): | |
| L += ["**Univariate AUROC of the extended features** (979 train labels, device-cluster CI, pooled / manual 562 / engaged 417; gold columns descriptive; first run's features)", "", | |
| "| feature | finite (train) | AUROC train [CI] | train manual | train engaged | gold pooled | gold manual | gold engaged |", "|---|---|---|---|---|---|---|---|"] | |
| for r in Udf.itertuples(): | |
| g = lambda k: getattr(r, k, np.nan) # noqa: E731 | |
| L.append(f"| {r.feature} | {r.finite_frac_train:.3f} | {f3(g('auroc_train'))} [{f3(g('ci_lo_train'), 2)},{f3(g('ci_hi_train'), 2)}] | {f3(g('auroc_train_manual'))} | " | |
| f"{f3(g('auroc_train_engaged'))} | {f3(r.auroc_gold_pooled)} | {f3(r.auroc_gold_manual)} | {f3(r.auroc_gold_engaged)} |") | |
| L.append("") | |
| md = "\n".join(L) + "\n" | |
| (out / "block.md").write_text(md) | |
| print(md) | |
| write_run_json(out, "warning_head_variant_diag", [Path(__file__), Path(TW.__file__), Path(WFX.__file__)], [CWM3 / "results" / "runs" / r / "config.json" for r in a.runs], seed=a.seed, | |
| t_start=t0, input_listing=False, extra={"runs": a.runs, "sets": list(sets), "n_boot": a.n_boot, "dms_fields_used": [], "variants": sorted(Vdf.variant.unique().tolist())}) | |
| print(f"DONE in {time.time() - t0:.0f}s -> {out}") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 17.5 kB
- Xet hash:
- 79622beb40c904ffdc8772f8a8e9417c61e12482db9b5ed4cabb340ca36498cf
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.