Buckets:
| """cwm3/train/train_warning_head.py — the learned warning head W (plan §3 W; claims B4 / B5). CPU only, scikit-learn. | |
| python train_warning_head.py --run I1_all_s0 [--sources consensus[,legacy_v1]] [--models lr,hgb,mlp] [--curve 0,100,300,979 --n-sub 20] | |
| Features (per anchor row, from the joint model's prediction pack + label-free anchor state; `feature_names()` lists them, ≈ 150 dims): | |
| driver chain marginals pi_hands[25], pi_head[25]; K̂ at +1 / +3 / +5 s (standardised, 3 x 11); onset_p@3s of the driver (4) and vehicle (4) | |
| events; vehicle pi_resp[25]; anchor states z0 (6 vehicle / driver chains); DEMAND SIDE from the road-pressure family P (targets agent, | |
| 2026-09-10: cwm2's ego-motion demand families are subsets of the response rule): P onset_p@3s (6), P marginals at +1/+3/+5 s (18), P0 (6), | |
| the 10 current margins (NaN where invalid) + validity bits, the hard-response chain R >= 2 (onset_p@3s, marginals, z0); `--demand D` restores | |
| the legacy D features (onset_p, z0, S0, D0valid), `both` uses both; current kinematics K0 (11); visibility bits (K0 valid, front PCA ok, | |
| all-16 road slots present, V history available); mode; vEgo. | |
| Feature-set ablation (project owner, 2026-09-10; --feature-sets current,future,both): the same head / labels / CV on (i) current-state-only | |
| features (no predicted quantity), (ii) predicted-future-only (marginals, onset_p, K̂, P / R predictions), (iii) both; per set gold AUROC / | |
| PR-AUC with device-cluster CIs, alerts / hour, lead time; paired Δ (iii − i) on shared resamples -> feature_sets.csv. Vehicle-side features are NaN where the V history is missing (HGB handles NaN natively; LR / MLP get | |
| 0-imputation + the V-valid indicator) so W_learned is defined on EVERY row. | |
| Labels (data/labels.py, source column kept): clean TRAIN consensus (979 / 182) [+ legacy_v1 1,250 / 412 as an explicitly labelled extension] | |
| [+ H5 when present]; the clean gold (test / val) never enters a fit; the clean-VAL anchors fit the temperature only. | |
| Models: L2 logistic regression (C grid) / HistGradientBoosting (depth 3, monotone constraints: pi_hands −, pi_head +, demand onset_p +, | |
| hand_leave / head_turn onset_p +) / 2-layer MLP (64 hidden, alpha grid); selection = mean AUROC over device-grouped 5-fold CV (GroupKFold | |
| on dongle); the selected configuration is refit on all labels. Temperature fitted on clean-val (warning_gold.temperature_fit grid). | |
| Label curve: n ∈ {0, 100, 300, 979[, +legacy]} x --n-sub stratified subsamples of the selected model -> clean-test gold AUROC (device-cluster | |
| bootstrap CI for the full fit; mean ± sd over subsamples); n = 0 is the model's own W_direct_t15_w10_RV_6fam (no labels). | |
| Operating points: threshold at clean-val recall 0.5 -> alerts / hour on the full test split with a 60 s per-route cooldown (exposure = rows x 0.2 s), | |
| plus DMS-rate-matched rows (operating_dms_rate.csv): the legacy DMS att@3s chain (cwm2 legacy marginals on the test split, first passage of the | |
| inattention union over 15 steps) at its own clean-val-recall-0.5 threshold gives the reference alerts/hour per regime; every score is then | |
| re-thresholded (bisection) to that alert load and precision / recall / lead time are compared at equal load. --oof reads the out-of-fold | |
| label-row packs (train/train_cwm3_oof.py); the column tag gains `oof`. | |
| precision / recall on gold test at that threshold, precision at fixed recall {0.3, 0.5, 0.7} on gold test, lead time = t0 − first alert in | |
| [t0 − 10 s, t_anchor] on detected gold positives. W_direct / driver_only / demand_only get the same treatment (same val recall). | |
| Outputs (append-only): results/warning_head/<run>_<stamp>/{cv.csv,selected.json,label_curve.csv,operating.csv,gold_quick.csv,features.json,run_*.json} | |
| and results/runs/W_<run>/pred_layout_{regime}_{split}.npz (arm cwm3_W; w block = W_learned_<model>_t15_w10_n<labels> for every model type + the | |
| selected one as W_learned_sel_...; uid / t witness) for the UNCHANGED cwm2/eval/warning_gold.py (eval/warning_head_eval.py). | |
| 2026-09-18 (owner question 'front video + driver only, CAN missing'; pre-registration docs/decisions.md 2026-09-18T03:34:28Z): `--no-can` (default off = every | |
| older argv bit for bit) drops the CAN-derived columns from the feature table before the feature-set masks (nocan_keep; tag `nocan` in every column / arm name): | |
| current-state DROP z0_gas / z0_brake / z0_steer_phase / z0_resp (vehicle anchor states), P0_* (road-pressure anchor bits from CAN margins), margin_* + margin_valid_*, | |
| z0_R_hard, V_valid, vEgo, mode (regime = cs_enabled OR cruiseState_enabled: CAN-derived) and the legacy D / mr blocks when present; predicted-future DROP the vehicle | |
| onset3s_* (gas_on / brake_on / steer_phase_on / resp_on), pi_resp_*, onset3s_R_hard / pi_R_hard_*, pi_P_* (forward marginals seeded from the CAN anchor z0_P). | |
| KEPT: z0_hands_on / z0_head_turned, K0_* + K0_valid (pose), vf_ok / tok_all16 (vision availability), dw_* / fc_* (pose dwell, face landmarks), pi_hands_* / pi_head_*, | |
| K_{1,3,5}s_* (K-hat), the four driver onset3s, onset3s_P_* (hazard-only road-pressure forecasts of the E trunk). | |
| No openpilot DMS field is read anywhere. | |
| """ | |
| 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] | |
| EXPLORE = CWM3.parent | |
| CWM2 = EXPLORE / "cwm2" | |
| S12R = EXPLORE / "s12_wm" / "results" | |
| for p in (CWM3 / "data", CWM3 / "eval", CWM3 / "common", CWM2 / "eval", CWM2 / "data", CWM2, CWM2 / "baselines" / "legacy_cockpitwm"): | |
| sys.path.insert(0, str(p)) | |
| import cwm3_dataset as ds # noqa: E402 | |
| import importlib.util as _ilu # noqa: E402 | |
| _spec = _ilu.spec_from_file_location("cwm3_labels", CWM3 / "data" / "labels.py") # explicit path: cwm2/baselines/legacy_cockpitwm/labels.py shadows the name | |
| LB = _ilu.module_from_spec(_spec) | |
| _spec.loader.exec_module(LB) | |
| import write_pred_layout_joint as WJ # noqa: E402 | |
| import warning_gold as WG # noqa: E402 (temperature_fit, cluster_boot, fast_auroc, build_gold — unchanged cwm2 code) | |
| from pred_layout import EVENT_NAMES, FAMILIES # noqa: E402 | |
| from runmeta import write_run_json, code_hash # noqa: E402 | |
| REGIMES = ("manual", "engaged") | |
| CHAINS = WJ.CHAINS | |
| # 2026-09-11 (coordinator, E7 K0-source sensitivity): module-level switches set by main() — default = the run's own kin_source (unchanged behaviour) | |
| K0_OPTS = {"mask_uids": frozenset(), "source": "run"} # source: run (the joint run's kin_source) | cwm2 (the route-median K0 of the v1 tables; NON-causal, diagnostic) | |
| # 2026-09-12 (warning-head agent, roadmap §B3 1a / 1b / 3): --extra-features dwell,mr,face (train/warning_features_ext.py blocks appended to the CURRENT-state | |
| # features), --mask-routes (rows on the driver-box-suspect routes removed from the training table and from the gold / layout evaluation; tag rm) and | |
| # --per-regime (one HGB per regime; tag pr). Defaults = unchanged behaviour (no extra block, no mask, pooled head). | |
| # 2026-09-17 (owner decision (6)): --extra-features fs appends the F2′ future-state head's PREDICTED quantities (warning_features_ext.fs_block, 22 dims; pack --fs-run, | |
| # default FS_dwell_s<seed>) to the extra block — like the gaze block a predicted quantity carried by the column tag (fs), not by is_future(). | |
| EXTRA_OPTS = {"sets": (), "mask_uids": frozenset(), "per_regime": False} | |
| sys.path.insert(0, str(CWM3 / "train")) | |
| import warning_features_ext as WFX # noqa: E402 | |
| TAU, WIN = 15, 10 | |
| K_STEPS = ("1s", "3s", "5s") | |
| KIN_CH = ds.KIN_CHANNELS | |
| # ----------------------------------------------------------------------------- features | |
| P_CHAINS = ("ttc", "ttc_drop", "cut_in", "lead_braking", "lane_excursion", "badas") | |
| P_MARGINS = ("ttc_s", "headway_s", "dRel_m", "vRel_mps", "aLead_mps2", "lat_margin_m", "lane_width_m", "badas", "lead_valid", "lane_valid") | |
| DRIVER_VEH_EVENTS = ("hand_return", "hand_leave", "head_turn_onset", "head_return", "gas_on", "brake_on", "steer_phase_on", "resp_on") | |
| def feature_names(demand="P", extra=None): | |
| """demand = P (road-pressure chains + current margins; default per the targets agent: the D ego-motion families are subsets of the response | |
| rule) | D (legacy demand chains) | both. extra: tuple of warning_features_ext sets appended at the end (None -> EXTRA_OPTS['sets'], default none).""" | |
| names = [f"pi_hands_{k}" for k in range(25)] + [f"pi_head_{k}" for k in range(25)] | |
| names += [f"K_{s}_{c}" for s in K_STEPS for c in KIN_CH] | |
| names += [f"onset3s_{ev}" for ev in DRIVER_VEH_EVENTS] | |
| names += [f"pi_resp_{k}" for k in range(25)] | |
| names += [f"z0_{c}" for c in CHAINS[:6]] | |
| if demand in ("D", "both"): | |
| names += [f"onset3s_D_{f}" for f in FAMILIES] + [f"z0_{f}" for f in FAMILIES] + [f"S0_{f}" for f in FAMILIES] + [f"D0valid_{f}" for f in FAMILIES] | |
| if demand in ("P", "both"): | |
| names += [f"onset3s_P_{f}" for f in P_CHAINS] + [f"pi_P_{f}_{s}" for s in K_STEPS for f in P_CHAINS] + [f"P0_{f}" for f in P_CHAINS] | |
| names += [f"margin_{m}" for m in P_MARGINS] + [f"margin_valid_{m}" for m in P_MARGINS] | |
| names += ["onset3s_R_hard"] + [f"pi_R_hard_{s}" for s in K_STEPS] + ["z0_R_hard"] | |
| names += [f"K0_{c}" for c in KIN_CH] | |
| names += ["K0_valid", "vf_ok", "tok_all16", "V_valid", "mode", "vEgo"] | |
| extra = EXTRA_OPTS["sets"] if extra is None else extra | |
| if extra: | |
| names += WFX.names_for(extra) # dw_* / mr_* / fc_* blocks (current-state; never `is_future`) | |
| return names | |
| FEATURE_SETS = ("current", "future", "both") | |
| # --no-can (2026-09-18): the CAN-derived columns. Exact names and prefixes; every z0_* except the two driver anchor states is CAN-derived (vehicle chains, legacy D families). | |
| NOCAN_OPTS = {"on": False} | |
| NOCAN_DROP_EXACT = frozenset({"z0_R_hard", "onset3s_R_hard", "V_valid", "vEgo", "mode", "onset3s_gas_on", "onset3s_brake_on", "onset3s_steer_phase_on", "onset3s_resp_on"}) | |
| NOCAN_DROP_PREFIX = ("P0_", "margin_", "pi_resp_", "pi_P_", "pi_R_hard_", "onset3s_D_", "S0_", "D0valid_", "mr_") | |
| NOCAN_KEEP_Z0 = frozenset({"z0_hands_on", "z0_head_turned"}) | |
| def is_can_derived(n): | |
| """True for a feature column that carries whitelisted-CAN information (vehicle anchor states / kinematics, road margins and their anchor bits, the CAN-seeded | |
| forward marginals, the regime flag); False for the pose / face / vision columns and the hazard-only forecasts of the driver and P chains.""" | |
| if n in NOCAN_DROP_EXACT or n.startswith(NOCAN_DROP_PREFIX): | |
| return True | |
| if n.startswith("z0_"): | |
| return n not in NOCAN_KEEP_Z0 | |
| return False | |
| def nocan_keep(names): | |
| return np.array([not is_can_derived(n) for n in names], bool) | |
| def is_future(n): | |
| """predicted-future quantity (chain marginals, onset_p, K̂, P / R marginals & onset_p) vs current-state (anchor states, margins, K0, bits).""" | |
| return n.startswith(("pi_hands_", "pi_head_", "K_1s_", "K_3s_", "K_5s_", "onset3s_", "pi_resp_", "pi_P_", "pi_R_hard_")) | |
| def feature_mask(names, fs): | |
| fut = np.array([is_future(n) for n in names]) | |
| return {"current": ~fut, "future": fut, "both": np.ones(len(names), bool)}[fs] | |
| V_FEATURES = {f"onset3s_{ev}" for ev in ("gas_on", "brake_on", "steer_phase_on", "resp_on")} | {f"pi_resp_{k}" for k in range(25)} | \ | |
| {f"z0_{c}" for c in ("gas", "brake", "steer_phase", "resp")} | {"onset3s_R_hard", "z0_R_hard"} | {f"pi_R_hard_{s}" for s in K_STEPS} | |
| def monotone_constraints(demand="P", names=None): | |
| cst = [] | |
| for n in (names if names is not None else feature_names(demand)): | |
| if n.startswith("pi_hands_"): | |
| cst.append(-1) | |
| elif n.startswith("pi_head_") or n.startswith("onset3s_D_") or n.startswith("onset3s_P_") or n in ("onset3s_hand_leave", "onset3s_head_turn_onset"): | |
| cst.append(1) | |
| elif n.startswith(("dw_", "mr_", "fc_", "gz_", "fs_", "kp_")): | |
| cst.append(WFX.monotone(n)) # +1 on dwell / share / pressure / predicted gaze-off / predicted off-task features, 0 on validity shares and signed face channels | |
| else: | |
| cst.append(0) | |
| return cst | |
| def _aux(regime, split, rows): | |
| """label-free anchor-state features from the targets / demand / pressure / kin / pack files for `rows` of (regime, split).""" | |
| f = ds.files(regime, split) | |
| with np.load(f["targets"], allow_pickle=True) as z: | |
| vego = z["X"][rows, -1, 3].astype(np.float32) | |
| with np.load(f["demand"], allow_pickle=True) as z: | |
| S0, D0v = z["S0"][rows].astype(np.float32), z["D0valid"][rows].astype(np.float32) | |
| pf = CWM3 / "results" / "targets_pressure" / f"pressure_{regime}_{split}.npz" | |
| with np.load(pf, allow_pickle=True) as z: | |
| assert [str(m) for m in z["margin_names"]] == list(P_MARGINS), list(z["margin_names"]) | |
| M0, M0v = z["M0"][rows].astype(np.float32), z["M0valid"][rows].astype(np.float32) | |
| M0 = np.where(M0v > 0.5, M0, np.nan) | |
| kin_path = f["kin"] if K0_OPTS["source"] == "run" else ds.KIN / f"kin_{regime}_{split}.npz" # --k0-source cwm2: the v1 tables' route-median K0 (diagnostic) | |
| with np.load(kin_path, allow_pickle=True) as z: # cwm2 route-median K0, or the causal pack's K0_c when the run's kin_source is causal (ds.set_kin_source; 2026-09-11) | |
| kk = "K0_c" if "K0_c" in z.files else "K0" | |
| K0 = z[kk][rows].astype(np.float32)[:, :len(KIN_CH)] # kin v2p packs (2026-09-18) carry 3 extra proxy columns: the K0_* block stays the 11 channels (the `kp` extra set reads them) | |
| k0v = z["K0valid"][rows].astype(np.float32) if "K0valid" in z.files else np.isfinite(K0).all(1).astype(np.float32) | |
| if "row_ok" in z.files: | |
| ok = z["row_ok"][rows].astype(bool) | |
| K0[~ok] = np.nan | |
| k0v = k0v * ok.astype(np.float32) | |
| if K0_OPTS["mask_uids"]: # --k0-mask-routes: K0 of the driver-box-suspect routes -> NaN (HGB handles NaN; LR / MLP impute 0) + K0_valid 0 | |
| with np.load(f["targets"], allow_pickle=True) as z: | |
| bad = np.isin(z["uid"][rows], list(K0_OPTS["mask_uids"])) | |
| K0[bad] = np.nan | |
| k0v = k0v * (~bad).astype(np.float32) | |
| with np.load(f["vf256"]) as z: | |
| vf_ok = z["row_ok"][rows].astype(np.float32) | |
| ai = np.load(f["anchor_index"]) | |
| tok16 = (ai[rows] >= 0).all(1).astype(np.float32) | |
| return vego, S0, D0v, k0v, vf_ok, tok16, M0, M0v, K0 | |
| def _chain_feats(hz_on, hz_off, z0, names): | |
| """onset_p@3s [n,C], marginals at +1/+3/+5 s [n,3C] (flattened step-major), z0 [n,C].""" | |
| on3 = 1 - np.prod(1 - hz_on[:, :TAU], 1) | |
| pim = np.stack([WJ.forward_marginal(hz_on[:, :, c], hz_off[:, :, c], z0[:, c]) for c in range(len(names))], -1) # [n,25,C] | |
| return on3, pim[:, [4, 14, 24], :].reshape(len(z0), -1), z0 | |
| def features_from_pred(pred, regime, split, rows, pos=None, demand="P"): | |
| """pred = np.load of a joint prediction pack; rows = rows of the (regime, split) targets file; pos = positions inside the pack | |
| (None -> pos == rows, i.e. a full-split pack). -> X [n, F] float32 (NaN for V features where the V history is missing).""" | |
| pos = rows if pos is None else pos | |
| hz_on, hz_off = pred["hz_on"][pos].astype(np.float64), pred["hz_off"][pos].astype(np.float64) | |
| hz_on, hz_off = np.clip(hz_on, 1e-6, 1 - 1e-6), np.clip(hz_off, 1e-6, 1 - 1e-6) | |
| z0 = pred["z0"][pos].astype(np.float64) | |
| ci = {c: i for i, c in enumerate(CHAINS)} | |
| cols = [] | |
| for c in ("hands_on", "head_turned"): | |
| cols.append(WJ.forward_marginal(hz_on[:, :, ci[c]], hz_off[:, :, ci[c]], z0[:, ci[c]])) | |
| cols.append(pred["K_hat"][pos].astype(np.float32).reshape(len(pos), -1)) | |
| on3 = [] | |
| for ev in DRIVER_VEH_EVENTS: | |
| c, kind = WJ.EVENT_SRC[ev] | |
| hh = (hz_on if kind == "on" else hz_off)[:, :, ci[c]] | |
| on3.append(1 - np.prod(1 - hh[:, :TAU], 1)) | |
| cols.append(np.stack(on3, 1)) | |
| cols.append(pred["pi_resp"][pos].astype(np.float32)) | |
| cols.append(z0[:, :6]) | |
| vego, S0, D0v, k0v, vf_ok, tok16, M0, M0v, K0 = _aux(regime, split, rows) | |
| if demand in ("D", "both"): | |
| onD = np.stack([1 - np.prod(1 - hz_on[:, :TAU, ci[f]], 1) for f in FAMILIES], 1) | |
| cols += [onD, z0[:, 6:], S0, D0v] | |
| if demand in ("P", "both"): | |
| assert "hz_on_P" in pred.files, "prediction pack has no P head outputs" | |
| onP, piP, z0P = _chain_feats(np.clip(pred["hz_on_P"][pos].astype(np.float64), 1e-6, 1 - 1e-6), np.clip(pred["hz_off_P"][pos].astype(np.float64), 1e-6, 1 - 1e-6), | |
| pred["z0_P"][pos].astype(np.float64), P_CHAINS) | |
| onR, piR, z0R = _chain_feats(np.clip(pred["hz_on_R"][pos].astype(np.float64), 1e-6, 1 - 1e-6), np.clip(pred["hz_off_R"][pos].astype(np.float64), 1e-6, 1 - 1e-6), | |
| pred["z0_R"][pos].astype(np.float64), ("hard",)) | |
| cols += [onP, piP, z0P, M0, M0v, onR, piR, z0R] | |
| vV = pred["valid_V"][pos].astype(np.float32) | |
| cols += [K0, k0v[:, None], vf_ok[:, None], tok16[:, None], vV[:, None], np.full((len(pos), 1), 0.0 if regime == "manual" else 1.0, np.float32), vego[:, None]] | |
| X = np.concatenate([np.asarray(c, np.float32).reshape(len(pos), -1) for c in cols], 1) | |
| names = feature_names(demand, extra=()) | |
| assert X.shape[1] == len(names), (X.shape, len(names)) | |
| vmask = np.array([n in V_FEATURES for n in names]) | |
| X[:, vmask] = np.where(vV[:, None] > 0.5, X[:, vmask], np.nan) | |
| if EXTRA_OPTS["sets"]: # --extra-features: label-free current-state blocks (dwell from Kh8_c, margin rule, face landmarks) | |
| Xe, ne = WFX.extra_block(regime, split, np.asarray(rows, np.int64), EXTRA_OPTS["sets"]) | |
| X = np.concatenate([X, Xe.astype(np.float32)], 1) | |
| assert X.shape[1] == len(feature_names(demand)), (X.shape, len(feature_names(demand))) | |
| return X | |
| class PerRegimeHGB: | |
| """--per-regime: one HistGradientBoosting per regime (feature `mode` 0 = manual / 1 = engaged) with the pooled fit as the fallback for a regime whose | |
| training rows carry a single class or fewer than 40 rows; the CV selection / hp grid are those of the pooled head.""" | |
| def __init__(self, hp, seed): | |
| self.hp, self.seed = dict(hp), seed | |
| self.mode_col = list(hp["names"]).index("mode") | |
| def _one(self): | |
| return make_model("hgb", dict(self.hp, per_regime=False), self.seed) | |
| def fit(self, X, y): | |
| self.pooled = self._one().fit(X, y) | |
| self.models = {} | |
| for v in (0.0, 1.0): | |
| m = X[:, self.mode_col] == v | |
| if m.sum() >= 40 and y[m].min() != y[m].max(): | |
| self.models[v] = self._one().fit(X[m], y[m]) | |
| return self | |
| def predict_proba(self, X): | |
| p = self.pooled.predict_proba(X) | |
| for v, mdl in self.models.items(): | |
| m = X[:, self.mode_col] == v | |
| if m.any(): | |
| p[m] = mdl.predict_proba(X[m]) | |
| return p | |
| # ----------------------------------------------------------------------------- models | |
| def make_model(kind, hp, seed=0): | |
| from sklearn.pipeline import make_pipeline | |
| from sklearn.preprocessing import StandardScaler | |
| from sklearn.impute import SimpleImputer | |
| from sklearn.linear_model import LogisticRegression | |
| from sklearn.ensemble import HistGradientBoostingClassifier | |
| from sklearn.neural_network import MLPClassifier | |
| if kind == "lr": | |
| return make_pipeline(SimpleImputer(strategy="constant", fill_value=0.0), StandardScaler(), | |
| LogisticRegression(C=hp["C"], penalty="l2", max_iter=2000, class_weight=hp.get("cw"))) | |
| if kind == "hgb": | |
| if hp.get("per_regime", False): | |
| return PerRegimeHGB(hp, seed) | |
| return HistGradientBoostingClassifier(max_depth=3, learning_rate=hp["lr"], max_iter=hp.get("iters", 200), l2_regularization=1.0, | |
| min_samples_leaf=20, monotonic_cst=monotone_constraints(hp.get("demand", "P"), hp.get("names")), random_state=seed, early_stopping=False) | |
| if kind == "mlp": | |
| return make_pipeline(SimpleImputer(strategy="constant", fill_value=0.0), StandardScaler(), | |
| MLPClassifier(hidden_layer_sizes=(64,), alpha=hp["alpha"], max_iter=500, early_stopping=True, validation_fraction=0.15, | |
| n_iter_no_change=20, random_state=seed)) | |
| raise ValueError(kind) | |
| GRID = {"lr": [{"C": c, "cw": cw} for c in (0.01, 0.1, 1.0) for cw in (None, "balanced")], | |
| "hgb": [{"lr": lr, "iters": it} for lr in (0.05, 0.1) for it in (100, 200)], | |
| "mlp": [{"alpha": a} for a in (1e-3, 1e-2, 1e-1)]} | |
| def cv_auroc(kind, hp, X, y, groups, n_splits=5, seed=0): | |
| from sklearn.model_selection import GroupKFold | |
| gkf = GroupKFold(n_splits=n_splits) | |
| aucs = [] | |
| for tr, te in gkf.split(X, y, groups): | |
| if y[te].min() == y[te].max() or y[tr].min() == y[tr].max(): | |
| continue | |
| m = make_model(kind, hp, seed).fit(X[tr], y[tr]) | |
| aucs.append(WG.fast_auroc(y[te], m.predict_proba(X[te])[:, 1])) | |
| return float(np.mean(aucs)) if aucs else np.nan, aucs | |
| def boot_auroc(y, s, groups, n_boot=1000, seed=0): | |
| if len(y) == 0 or y.min() == y.max(): | |
| # 2026-09-18: a score undefined on every gold row (e.g. the model's own W_direct when the vehicle branch is a missing modality, valid_V = 0) -> NaN row instead of a crash | |
| return np.nan, (np.nan, np.nan) | |
| b = WG.cluster_boot(y, {"s": s}, groups, n_boot, seed)["s"]["auroc"] | |
| return WG.fast_auroc(y, s), WG.pct(b) | |
| def safe_ap(y, s): | |
| return WG.fast_ap(y, s) if len(y) and y.min() != y.max() else np.nan | |
| # ----------------------------------------------------------------------------- operating points | |
| def alerts_per_hour(uid, t, score, thr, cooldown=60.0): | |
| """per-route alert count with a cooldown; exposure = rows x 0.2 s.""" | |
| order = np.lexsort((t, uid)) | |
| u, tt, s = uid[order], t[order], score[order] | |
| n_alert = 0 | |
| last_uid, last_t = None, -1e9 | |
| fire = np.isfinite(s) & (s >= thr) | |
| for i in range(len(u)): | |
| if u[i] != last_uid: | |
| last_uid, last_t = u[i], -1e9 | |
| if fire[i] and tt[i] - last_t >= cooldown: | |
| n_alert += 1 | |
| last_t = tt[i] | |
| hours = len(u) * 0.2 / 3600.0 | |
| return n_alert, hours, n_alert / hours if hours > 0 else np.nan | |
| def threshold_at_recall(y, s, target): | |
| pos = np.sort(s[(y == 1) & np.isfinite(s)])[::-1] | |
| if len(pos) == 0: | |
| return np.nan | |
| k = int(np.ceil(target * len(pos))) - 1 | |
| return float(pos[max(k, 0)]) | |
| def precision_at_recall(y, s, target): | |
| order = np.argsort(-s, kind="mergesort") | |
| ys = y[order] | |
| tp = np.cumsum(ys) | |
| prec = tp / np.arange(1, len(ys) + 1) | |
| rec = tp / max(ys.sum(), 1) | |
| ok = rec >= target | |
| return float(prec[ok].max()) if ok.any() else np.nan | |
| def lead_times(G_pos, packs_scores, thr, lookback=10.0): | |
| """G_pos rows: regime, split, uid, t0, t_anchor. packs_scores[(regime, split)] = (uid, t, score). -> list of lead seconds (NaN = not detected).""" | |
| out = [] | |
| for e in G_pos.itertuples(): | |
| uid, t, s = packs_scores[(e.regime, e.split)] | |
| m = (uid == e.uid) & (t >= e.t0 - lookback) & (t <= e.t_anchor + 1e-6) & np.isfinite(s) | |
| if not m.any(): | |
| out.append(np.nan) | |
| continue | |
| tt, ss = t[m], s[m] | |
| hit = tt[ss >= thr] | |
| out.append(float(e.t0 - hit.min()) if len(hit) else np.nan) | |
| return np.array(out) | |
| # ----------------------------------------------------------------------------- main | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--run", required=True, help="joint run dir name under results/runs (e.g. I1_all_s0)") | |
| ap.add_argument("--sources", default="consensus") | |
| ap.add_argument("--models", default="lr,hgb,mlp") | |
| ap.add_argument("--feature-sets", default="current,future,both") | |
| ap.add_argument("--curve", default="0,100,300,979") | |
| ap.add_argument("--n-sub", type=int, default=20) | |
| ap.add_argument("--n-boot", type=int, default=1000) | |
| ap.add_argument("--seed", type=int, default=0) | |
| ap.add_argument("--val-recall", type=float, default=0.5) | |
| ap.add_argument("--out-root", default=str(CWM3 / "results" / "warning_head")) | |
| ap.add_argument("--demand", default="P", choices=["P", "D", "both"], help="demand-side features: road-pressure P (default) | legacy D | both") | |
| ap.add_argument("--oof", action="store_true", help="train features from the OUT-OF-FOLD label-row packs (train/train_cwm3_oof.py) instead of the in-sample ones") | |
| ap.add_argument("--dms-marginals", default=str(CWM2 / "baselines" / "legacy_cockpitwm" / "results"), help="legacy marginals_{regime}_s{seed}.npz dir (DMS att@3s on the test split)") | |
| ap.add_argument("--k0-mask-routes", default="", help="json with `flagged` [{uid}] (driver-box suspects): their rows' current-kinematics features K0 -> NaN, K0_valid -> 0 (train, gold and full-split " | |
| "scores alike); column tag += k0m (2026-09-11 K0-source sensitivity row of warning_head_final.md)") | |
| ap.add_argument("--k0-source", default="run", choices=["run", "cwm2"], help="run (default): K0 from the joint run's kin_source; cwm2: the route-median K0 of the v1 tables (NON-causal ROI, diagnostic); tag += k0old") | |
| ap.add_argument("--w-suffix", default="", help="suffix of the W layout dir results/runs/W_<run><suffix> (default '' ; --oof implies _oof); the sensitivity rows use _k0m / _k0old") | |
| ap.add_argument("--extra-features", default="", help="comma list of train/warning_features_ext blocks appended to the CURRENT-state features: dwell (Kh8_c dwell / share), " | |
| "mr (eval/margin_rule 7 dims), face (facehist_causal_routes_v2), gaze (the gaze_off CHAIN outputs of the U0 --gaze-run: predicted onset / marginal, " | |
| "10 dims; 2026-09-12 warning_head_final_gaze.md), fs (the F2′ future-state pack of --fs-run: predicted off-task occupancy @1/2/3/5 s per state, " | |
| "expected off-task seconds, expected remaining off dwell, P(leave within 3 s), 22 dims; 2026-09-17 owner decision (6)), kp (the kin v2p keypoint proxies' current values " | |
| "head_pitch_kp / wrist_face_kp / hands_lap_kp + pitch dwell + valid, 5 dims; 2026-09-18 plan A arm); column tag += dw / fc / mr / gz / fs / kp") | |
| ap.add_argument("--gaze-run", default="", help="U0 gaze-chain run dir (name under results/runs) whose layouts / pred_gaze_*_train_labelrows.npz feed the `gaze` block; " | |
| "default U0_kin_can_noK_s<seed of --run>_kcv2_clip8_ln_gzfc (+chain+face)") | |
| ap.add_argument("--fs-run", default="", help="F2′ future-state run dir (name under results/runs) whose fs_{regime}_{split}.npz packs feed the `fs` block; default FS_dwell_s<seed of --run>") | |
| ap.add_argument("--mask-routes", default="", help="json with `flagged` [{uid}]: rows on these routes are REMOVED from the training table and from the gold / layout evaluation " | |
| "(w_mask False) — the row-masked sensitivity rows; tag += rm") | |
| ap.add_argument("--mask-tag", default="rm", help="tag appended to the column names / arm / ext_tag when --mask-routes is given (default rm = the 64-route SUSPECT mask; " | |
| "2026-09-16: rmv = the 3-route human-VERIFIED wrong-occupant mask driver_box_verified_mask.json). Default unchanged.") | |
| ap.add_argument("--per-regime", action="store_true", help="HGB fitted per regime (mode 0 / 1; pooled fallback) under the pooled CV selection; tag += pr") | |
| ap.add_argument("--no-can", action="store_true", help="2026-09-18 (pre-registration docs/decisions.md 2026-09-18T03:34:28Z): drop every CAN-derived column (is_can_derived) from the " | |
| "feature table before the feature-set masks; tag += nocan (columns W_learned_<set><model>_t15_w10_n979<ext>nocan, arm abl_W_<ext>nocan)") | |
| a = ap.parse_args() | |
| NOCAN_OPTS["on"] = bool(a.no_can) | |
| assert not (a.no_can and a.per_regime), "--no-can drops the `mode` column the per-regime HGB splits on" | |
| if a.k0_mask_routes: | |
| K0_OPTS["mask_uids"] = frozenset(r["uid"] for r in json.loads(Path(a.k0_mask_routes).read_text())["flagged"]) | |
| K0_OPTS["source"] = a.k0_source | |
| EXTRA_OPTS["sets"] = tuple(s for s in ("dwell", "face", "mr", "gaze", "fs", "kp") if s in a.extra_features.split(",")) # kp (2026-09-18): kin v2p keypoint proxies + pitch dwell | |
| if a.mask_routes: | |
| EXTRA_OPTS["mask_uids"] = frozenset(r["uid"] for r in json.loads(Path(a.mask_routes).read_text())["flagged"]) | |
| EXTRA_OPTS["per_regime"] = bool(a.per_regime) | |
| ext_tag = WFX.tag_for(EXTRA_OPTS["sets"]) + (a.mask_tag if a.mask_routes else "") + ("pr" if a.per_regime else "") + ("nocan" if a.no_can else "") # --mask-tag (default rm) 2026-09-16; nocan 2026-09-18 | |
| w_suffix = a.w_suffix or ("_oof" if a.oof else "") | |
| t0 = time.time() | |
| rd = CWM3 / "results" / "runs" / a.run | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| out = Path(a.out_root) / f"{a.run}_{stamp}" | |
| out.mkdir(parents=True, exist_ok=True) | |
| sources = tuple(a.sources.split(",")) | |
| src_tag = ("n979" if sources == ("consensus",) else "n" + "".join(s[:3] for s in sources)) + ("" if a.demand == "P" else a.demand) + ("oof" if a.oof else "") \ | |
| + ("k0m" if a.k0_mask_routes else "") + ("k0old" if a.k0_source == "cwm2" else "") + ext_tag | |
| if a.per_regime: | |
| for hp0 in GRID["hgb"]: | |
| hp0["per_regime"] = True | |
| fsets = [f for f in a.feature_sets.split(",") if f in FEATURE_SETS] | |
| log_p = out / "train_warning_head.log" | |
| def log(m): | |
| line = f"[{time.strftime('%H:%M:%S')}] {m}" | |
| print(line, flush=True) | |
| with open(log_p, "a") as f: | |
| f.write(line + "\n") | |
| run_cfg = json.loads((rd / "config.json").read_text()) | |
| seed_run = int(run_cfg["args"]["seed"]) | |
| base_ckpts = json.loads((rd / "base_ckpts.json").read_text()) if (rd / "base_ckpts.json").exists() else {} | |
| kin_source, kin_version = run_cfg["cfg"].get("kin_source", "cwm2") or "cwm2", run_cfg["cfg"].get("kin_version", "v1") or "v1" | |
| gaze_run = "" | |
| if "gaze" in EXTRA_OPTS["sets"]: # 2026-09-12: the gaze block reads a U0 gaze-chain run of the SAME seed (default +chain+face) | |
| gaze_run = a.gaze_run or f"U0_kin_can_noK_s{seed_run}_kcv2_clip8_ln_gzfc" | |
| WFX.set_gaze_run(CWM3 / "results" / "runs" / gaze_run) | |
| fs_run = "" | |
| if "fs" in EXTRA_OPTS["sets"]: # 2026-09-17: the F2′ future-state pack trained on the FINAL run of the SAME seed | |
| fs_run = a.fs_run or f"FS_dwell_s{seed_run}" | |
| WFX.set_fs_run(CWM3 / "results" / "runs" / fs_run) | |
| ds.set_kin_source(kin_source, kin_version) # the current-kinematics features K0 come from the inputs the joint model was trained on (2026-09-11) | |
| ds.set_pack_version(run_cfg["cfg"].get("pack_version", "v1") or "v1") | |
| log(f"run {a.run} (seed {seed_run}); base checkpoints {base_ckpts.get('base_ckpts')}; kin_source {kin_source}/{kin_version}; pack {ds.pack_version()}; " | |
| f"K0 source {a.k0_source}; K0 masked routes {len(K0_OPTS['mask_uids'])}; W dir suffix '{w_suffix}'; extra features {EXTRA_OPTS['sets']}; " | |
| f"row-masked routes {len(EXTRA_OPTS['mask_uids'])}; per-regime HGB {a.per_regime}; gaze run {gaze_run or '-'}; fs run {fs_run or '-'}; no-can {a.no_can}") | |
| # ---- training features / labels (all columns; feature sets are column masks) | |
| tab = LB.train_label_table(sources=sources) | |
| if EXTRA_OPTS["mask_uids"]: # --mask-routes: suspect-route rows leave the training table (and, below, the gold evaluation) | |
| n_before = len(tab) | |
| tab = tab[~tab.uid.isin(EXTRA_OPTS["mask_uids"])].reset_index(drop=True) | |
| log(f"row mask: {n_before - len(tab)} training label rows on {len(EXTRA_OPTS['mask_uids'])} flagged routes removed -> {len(tab)} rows / {int(tab.y.sum())} pos") | |
| Xs, ys, gs, srcs = [], [], [], [] | |
| for regime in REGIMES: | |
| pf = rd / (f"pred_{regime}_train_labelrows_oof.npz" if a.oof else f"pred_{regime}_train_labelrows.npz") | |
| assert pf.exists(), f"{pf} missing: run the joint trainer with label-row prediction" + (" (train/train_cwm3_oof.py)" if a.oof else "") | |
| pr = np.load(pf, allow_pickle=True) | |
| prow = pr["rows"].astype(np.int64) | |
| posmap = {int(r): i for i, r in enumerate(prow)} | |
| sub = tab[tab.regime == regime] | |
| pos = np.array([posmap[int(r)] for r in sub.row], np.int64) | |
| Xs.append(features_from_pred(pr, regime, "train", sub.row.to_numpy().astype(np.int64), pos, a.demand)) | |
| ys.append(sub.y.to_numpy().astype(int)) | |
| gs.append(sub.dongle.to_numpy()) | |
| srcs.append(sub.source.to_numpy()) | |
| X_all, y, groups, src = np.concatenate(Xs), np.concatenate(ys), np.concatenate(gs), np.concatenate(srcs) | |
| names_all = feature_names(a.demand) | |
| keep_nocan = nocan_keep(names_all) if a.no_can else np.ones(len(names_all), bool) | |
| nocan_info = {"on": bool(a.no_can), "dropped": [n for n, k in zip(names_all, keep_nocan) if not k], "kept": [n for n, k in zip(names_all, keep_nocan) if k]} | |
| if a.no_can: | |
| log(f"--no-can: {len(nocan_info['dropped'])} CAN-derived columns dropped ({', '.join(nocan_info['dropped'][:12])}, ...); {len(nocan_info['kept'])} kept: {', '.join(nocan_info['kept'])}") | |
| log(f"train labels: {len(y)} rows / {int(y.sum())} pos / {len(np.unique(groups))} devices; sources {dict(zip(*np.unique(src, return_counts=True)))}; features {X_all.shape[1]}; " | |
| f"train features {'OUT-OF-FOLD' if a.oof else 'in-sample'}") | |
| # ---- gold anchors (evaluation / temperature only) | |
| 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")} | |
| legacy = WG.merge_legacy(seed_run) | |
| G = WG.build_gold(lab, packs, legacy, "legacy_first") | |
| if legacy is not None and "att@3s" in legacy: | |
| G["att@3s"] = legacy.reindex(G.idx)["att@3s"].to_numpy(float) | |
| assert not set(G[G.clean_test].idx) & set(tab.idx), "gold test episode inside the training table" | |
| preds = {(r, s): np.load(rd / f"pred_{r}_{s}.npz", allow_pickle=True) for r in REGIMES for s in ("test", "val")} | |
| Xg_all = np.full((len(G), X_all.shape[1]), np.nan, np.float32) | |
| for (r, s), g in G.groupby(["regime", "split"]): | |
| Xg_all[g.index.to_numpy()] = features_from_pred(preds[(r, s)], r, s, g.row.to_numpy().astype(np.int64), None, a.demand) | |
| gt, gv = G.clean_test.to_numpy(), G.clean_val.to_numpy() | |
| if EXTRA_OPTS["mask_uids"]: | |
| on_mask = G.uid.isin(EXTRA_OPTS["mask_uids"]).to_numpy() | |
| log(f"row mask: gold rows on flagged routes excluded from evaluation / temperature: clean_test {int((gt & on_mask).sum())} ({int(G.y.to_numpy()[gt & on_mask].sum())} pos), " | |
| f"clean_val {int((gv & on_mask).sum())}") | |
| gt, gv = gt & ~on_mask, gv & ~on_mask | |
| yg = G.y.to_numpy().astype(int) | |
| dong_g = G.dongle.to_numpy() | |
| log(f"gold: clean_test {int(gt.sum())} rows / {int(yg[gt].sum())} pos / {len(np.unique(dong_g[gt]))} dev; clean_val {int(gv.sum())} / {int(yg[gv].sum())}") | |
| lay = {(r, s): np.load(rd / f"pred_layout_{r}_{s}.npz", allow_pickle=True) for r in REGIMES for s in ("test", "val")} | |
| def ref_col(wn): | |
| v = np.full(len(G), np.nan) | |
| for (r, s), g in G.groupby(["regime", "split"]): | |
| z = lay[(r, s)] | |
| j = list(map(str, z["w_names"])).index(wn) | |
| v[g.index.to_numpy()] = z["w"][g.row.to_numpy(), j].astype(np.float64) | |
| return v | |
| REFS = ("W_direct_t15_w10_RV_6fam", "driver_only_t15", "demand_only_t15_6fam") | |
| ref_scores = {wn: ref_col(wn) for wn in REFS} | |
| full_rows = {(r, s): features_from_pred(pr, r, s, np.arange(len(pr["valid_V"])), None, a.demand) for (r, s), pr in preds.items()} | |
| # ---- per feature set: CV selection, full fits, gold, label curve, all-row scores | |
| cv_rows, gold_rows, curve_rows, selected, scores_all, gold_scores = [], [], [], {}, {}, {} | |
| rng = np.random.default_rng(a.seed) | |
| for fs in fsets: | |
| mask = feature_mask(names_all, fs) & keep_nocan # --no-can (2026-09-18): the CAN-derived columns leave every feature set | |
| names = [n for n, m in zip(names_all, mask) if m] | |
| X, Xg = X_all[:, mask], Xg_all[:, mask] | |
| log(f"=== feature set {fs}: {len(names)} features") | |
| best = {} | |
| for kind in a.models.split(","): | |
| for hp0 in GRID[kind]: | |
| hp = dict(hp0, demand=a.demand, names=names) | |
| m, folds = cv_auroc(kind, hp, X, y, groups, 5, a.seed) | |
| cv_rows.append({"feature_set": fs, "model": kind, "hp": json.dumps({k: v for k, v in hp.items() if k != "names"}), "cv_auroc": m, "folds": json.dumps([round(f, 4) for f in folds])}) | |
| log(f" cv {kind} {hp0}: AUROC {m:.4f}") | |
| if kind not in best or m > best[kind][0]: | |
| best[kind] = (m, hp) | |
| sel = max(best, key=lambda k: best[k][0]) | |
| selected[fs] = {"selected": sel, "hp": {k: v for k, v in best[sel][1].items() if k != "names"}, "cv_auroc": best[sel][0], "n_features": len(names), | |
| "per_model": {k: {"cv_auroc": v[0], "hp": {kk: vv for kk, vv in v[1].items() if kk != "names"}} for k, v in best.items()}} | |
| log(f" selected {sel} (cv AUROC {best[sel][0]:.4f}); per-model best {{{', '.join(f'{k}: {v[0]:.4f}' for k, v in best.items())}}}") | |
| fits = {kind: make_model(kind, hp, a.seed).fit(X, y) for kind, (cvm, hp) in best.items()} | |
| tag_fs = {"current": "cur", "future": "fut", "both": "both"}[fs] | |
| for kind, m in fits.items(): | |
| s_all = m.predict_proba(Xg)[:, 1] | |
| au, ci = boot_auroc(yg[gt], s_all[gt], dong_g[gt], a.n_boot, 0) | |
| b = WG.cluster_boot(yg[gt], {"s": s_all[gt]}, dong_g[gt], a.n_boot, 0)["s"]["prauc"] | |
| au_v = WG.fast_auroc(yg[gv], s_all[gv]) if yg[gv].min() != yg[gv].max() else np.nan | |
| Tm, pinned = WG.temperature_fit(s_all[gv], yg[gv].astype(float)) | |
| gold_rows.append({"feature_set": fs, "model": kind, "hp": json.dumps(selected[fs]["per_model"][kind]["hp"]), "cv_auroc": best[kind][0], "gold_test_auroc": au, "ci_lo": ci[0], "ci_hi": ci[1], | |
| "gold_test_prauc": WG.fast_ap(yg[gt], s_all[gt]), "prauc_ci_lo": WG.pct(b)[0], "prauc_ci_hi": WG.pct(b)[1], "n": int(gt.sum()), "n_pos": int(yg[gt].sum()), | |
| "clean_val_auroc": au_v, "temperature": Tm, "T_pinned": pinned, "selected": int(kind == sel), "labels": src_tag, "n_labels": int(len(y)), | |
| "base_ckpts": json.dumps(base_ckpts.get("base_ckpts", {}))}) | |
| log(f" gold clean_test AUROC {fs}/{kind}: {au:.4f} [{ci[0]:.3f}, {ci[1]:.3f}] PR-AUC {gold_rows[-1]['gold_test_prauc']:.3f} (val {au_v:.3f}, T {Tm:.2f})") | |
| gold_scores.setdefault(fs, {})[kind] = s_all | |
| if kind == sel: | |
| gold_scores[fs]["sel"] = s_all | |
| # label curve (selected model type + hp) | |
| ns = [int(x) for x in a.curve.split(",")] | |
| for n in ns: | |
| if n == 0: | |
| v = ref_scores["W_direct_t15_w10_RV_6fam"] | |
| fin = gt & np.isfinite(v) | |
| curve_rows.append({"feature_set": fs, "n": 0, "sub": 0, "model": "W_direct_t15_w10_RV_6fam", "gold_test_auroc": WG.fast_auroc(yg[fin], v[fin]), "n_pos_train": 0}) | |
| continue | |
| n_eff = min(n, len(y)) | |
| for sub in range(a.n_sub if n_eff < len(y) else 1): | |
| idx = np.concatenate([rng.choice(np.flatnonzero(y == c), size=int(round(n_eff * (y == c).mean())), replace=False) for c in (0, 1)]) | |
| if y[idx].sum() < 5: | |
| continue | |
| m = make_model(sel, best[sel][1], a.seed + sub).fit(X[idx], y[idx]) | |
| curve_rows.append({"feature_set": fs, "n": n_eff, "sub": sub, "model": sel, "gold_test_auroc": WG.fast_auroc(yg[gt], m.predict_proba(Xg)[:, 1][gt]), "n_pos_train": int(y[idx].sum())}) | |
| # scores on every test / val row | |
| for (r, s), Xrs in full_rows.items(): | |
| cols = scores_all.setdefault((r, s), {}) | |
| for kind, m in fits.items(): | |
| cols[f"W_learned_{tag_fs}{kind}_t{TAU}_w{WIN}_{src_tag}"] = m.predict_proba(Xrs[:, mask])[:, 1].astype(np.float32) | |
| cols[f"W_learned_{tag_fs}sel_t{TAU}_w{WIN}_{src_tag}"] = cols[f"W_learned_{tag_fs}{sel}_t{TAU}_w{WIN}_{src_tag}"] | |
| pd.DataFrame(cv_rows).to_csv(out / "cv.csv", index=False) | |
| pd.DataFrame(gold_rows).to_csv(out / "gold_quick.csv", index=False) | |
| (out / "selected.json").write_text(json.dumps({"feature_sets": selected, "sources": sources, "run": a.run, "seed_run": seed_run, "n_labels": int(len(y)), "n_pos": int(y.sum()), | |
| "base_ckpts": base_ckpts, "demand": a.demand, "src_tag": src_tag, "headline_feature_set": "both" if "both" in selected else fsets[-1], | |
| "kin_source": kin_source, "kin_version": kin_version, "pack_version": ds.pack_version(), | |
| "w_suffix": w_suffix, "k0_source": a.k0_source, "k0_mask_routes": a.k0_mask_routes, "n_k0_mask_routes": len(K0_OPTS["mask_uids"]), | |
| "extra_features": list(EXTRA_OPTS["sets"]), "mask_routes": a.mask_routes, "n_mask_routes": len(EXTRA_OPTS["mask_uids"]), | |
| "per_regime": bool(a.per_regime), "ext_tag": ext_tag, "gaze_run": gaze_run, "fs_run": fs_run, "no_can": nocan_info, | |
| "selected": selected.get("both", selected[fsets[-1]])["selected"]}, indent=1)) | |
| cdf = pd.DataFrame(curve_rows) | |
| cdf.to_csv(out / "label_curve.csv", index=False) | |
| summ = cdf.groupby(["feature_set", "n"]).gold_test_auroc.agg(["mean", "std", "count"]) | |
| log("label curve (gold clean_test AUROC, mean ± sd over subsamples):\n" + summ.to_string()) | |
| # references on the same anchors | |
| for wn, v in ref_scores.items(): | |
| fin = gt & np.isfinite(v) | |
| au, ci = boot_auroc(yg[fin], v[fin], dong_g[fin], a.n_boot, 0) | |
| gold_rows.append({"feature_set": "reference", "model": wn, "hp": "", "cv_auroc": np.nan, "gold_test_auroc": au, "ci_lo": ci[0], "ci_hi": ci[1], "gold_test_prauc": safe_ap(yg[fin], v[fin]), | |
| "prauc_ci_lo": np.nan, "prauc_ci_hi": np.nan, "n": int(fin.sum()), "n_pos": int(yg[fin].sum()), "clean_val_auroc": np.nan, "temperature": np.nan, "T_pinned": "", "selected": 0, | |
| "labels": "none", "n_labels": 0, "base_ckpts": json.dumps(base_ckpts.get("base_ckpts", {}))}) | |
| log(f" gold clean_test AUROC {wn}: {au:.4f} [{ci[0]:.3f}, {ci[1]:.3f}] n={int(fin.sum())}") | |
| pd.DataFrame(gold_rows).to_csv(out / "gold_quick.csv", index=False) | |
| # ---- paired deltas between feature sets (shared device-cluster resamples on the clean-test gold) | |
| # paired deltas between feature sets on shared resamples: (a) the CV-selected head per set ("sel"; model types may differ), (b) the SAME model | |
| # type across sets (lr / hgb / mlp) — the like-for-like test of "does the predicted future add value beyond the current state" | |
| fs_rows = [] | |
| if len(gold_scores) >= 2: | |
| kinds = ["sel"] + [k for k in a.models.split(",") if all(k in gold_scores[fs] for fs in gold_scores)] | |
| S = {f"{fs}|{k}": gold_scores[fs][k][gt] for fs in gold_scores for k in kinds if k in gold_scores[fs]} | |
| bs = WG.cluster_boot(yg[gt], S, dong_g[gt], a.n_boot, 0) | |
| for k in kinds: | |
| for fa, fb in (("both", "current"), ("future", "current"), ("both", "future")): | |
| ka, kb = f"{fa}|{k}", f"{fb}|{k}" | |
| if ka in S and kb in S: | |
| for met, key in (("auroc", "auroc"), ("pr_auc", "prauc")): | |
| fn = WG.fast_auroc if met == "auroc" else WG.fast_ap | |
| d = bs[ka][key] - bs[kb][key] | |
| lo, hi = WG.pct(d) | |
| ma = selected[fa]["selected"] if k == "sel" else k | |
| mb = selected[fb]["selected"] if k == "sel" else k | |
| fs_rows.append({"a": fa, "b": fb, "model_kind": k, "metric": met, "value_a": fn(yg[gt], S[ka]), "value_b": fn(yg[gt], S[kb]), "delta": fn(yg[gt], S[ka]) - fn(yg[gt], S[kb]), | |
| "delta_ci_lo": lo, "delta_ci_hi": hi, "frac_boot_d_gt0": float(np.mean(d[np.isfinite(d)] > 0)), "n": int(gt.sum()), "n_pos": int(yg[gt].sum()), | |
| "model_a": ma, "model_b": mb, "n_boot": a.n_boot, "base_ckpts": json.dumps(base_ckpts.get("base_ckpts", {}))}) | |
| log(f" paired [{k}] {fa} − {fb} {met}: {fs_rows[-1]['delta']:+.4f} [{lo:+.4f}, {hi:+.4f}]") | |
| pd.DataFrame(fs_rows).to_csv(out / "feature_sets.csv", index=False) | |
| # ---- W_learned layouts (arm cwm3_W) + operating points | |
| wout = CWM3 / "results" / "runs" / f"W_{a.run}{w_suffix}" | |
| wout.mkdir(parents=True, exist_ok=True) | |
| for (r, s), cols in scores_all.items(): | |
| z = lay[(r, s)] | |
| N = len(z["uid"]) | |
| meta = json.loads(str(z["meta"])) | |
| # extended heads (2026-09-12) get their own results_schema-legal arm `abl_W_<tag>`: warning_gold.py keys layouts by (arm, regime, seed, split), so several | |
| # cwm3_W dirs in one item would overwrite each other; the base head keeps arm cwm3_W (unchanged) | |
| meta.update({"arm": "cwm3_W" if not ext_tag else f"abl_W_{ext_tag}", "score_source": "independent_marginals", "model_id": f"W_{a.run}:" + "/".join(f"{fs}={selected[fs]['selected']}" for fs in selected), | |
| "warning_head": {"feature_sets": {fs: {"selected": v["selected"], "hp": v["hp"], "cv_auroc": v["cv_auroc"], "n_features": v["n_features"]} for fs, v in selected.items()}, | |
| "labels": src_tag, "n_labels": int(len(y)), "demand": a.demand, "base_ckpts": base_ckpts.get("base_ckpts", {}), | |
| "k0_source": a.k0_source, "k0_mask_routes": a.k0_mask_routes, "kin_source": kin_source, "kin_version": kin_version, | |
| "extra_features": list(EXTRA_OPTS["sets"]), "mask_routes": a.mask_routes, "per_regime": bool(a.per_regime), "gaze_run": gaze_run, "fs_run": fs_run, | |
| "no_can": bool(a.no_can), "no_can_dropped": nocan_info["dropped"]}, | |
| "code_sha256": code_hash([Path(__file__), Path(WJ.__file__)]), | |
| "notes": "cwm3 learned warning head on the joint model's outputs; w block = W_learned columns only (cur* current-state features, fut* predicted-future, both*; " | |
| "the joint model's own W_direct is in the cwm3_I layout)"}) | |
| wn = list(cols) | |
| w_mask = np.ones(N, bool) if not EXTRA_OPTS["mask_uids"] else ~np.isin(z["uid"], list(EXTRA_OPTS["mask_uids"])) # --mask-routes: flagged rows carry no score | |
| np.savez_compressed(wout / f"pred_layout_{r}_{s}.npz", meta=np.array(json.dumps(meta)), uid=z["uid"], t=z["t"], w_names=np.array(wn), | |
| w=np.stack([cols[k] for k in wn], 1).astype(np.float16), w_mask=w_mask) | |
| log(f" wrote {wout / f'pred_layout_{r}_{s}.npz'} ({N} rows, {len(wn)} columns)") | |
| op_rows = [] | |
| G_pos = G[gt & (G.y == 1)] | |
| getters = [] | |
| for fs in fsets: | |
| tag_fs = {"current": "cur", "future": "fut", "both": "both"}[fs] | |
| col = f"W_learned_{tag_fs}sel_t{TAU}_w{WIN}_{src_tag}" | |
| getters.append((f"W_learned[{fs}:{selected[fs]['selected']}]", (lambda col: lambda r, s: scores_all[(r, s)][col].astype(np.float64))(col))) | |
| for wn in REFS: | |
| getters.append((wn, (lambda wn: lambda r, s: lay[(r, s)]["w"][:, list(map(str, lay[(r, s)]["w_names"])).index(wn)].astype(np.float64))(wn))) | |
| for name, getter in getters: | |
| sc = {(r, s): getter(r, s) for r in REGIMES for s in ("test", "val")} | |
| gsc = np.full(len(G), np.nan) | |
| for (r, s), g in G.groupby(["regime", "split"]): | |
| gsc[g.index.to_numpy()] = sc[(r, s)][g.row.to_numpy()] | |
| fin_v = gv & np.isfinite(gsc) | |
| thr = threshold_at_recall(yg[fin_v], gsc[fin_v], a.val_recall) | |
| fin_t = gt & np.isfinite(gsc) | |
| rec_t = float(((gsc[fin_t] >= thr) & (yg[fin_t] == 1)).sum() / max(yg[fin_t].sum(), 1)) | |
| prec_t = float(((gsc[fin_t] >= thr) & (yg[fin_t] == 1)).sum() / max((gsc[fin_t] >= thr).sum(), 1)) | |
| aph = {} | |
| for r in REGIMES: | |
| uid, t = packs[(r, "test")]["uid"], packs[(r, "test")]["t"].astype(float) | |
| aph[r] = alerts_per_hour(uid, t, sc[(r, "test")], thr) | |
| leads = lead_times(G_pos, {k: (packs[k]["uid"], packs[k]["t"].astype(float), v) for k, v in sc.items()}, thr) | |
| det = np.isfinite(leads) | |
| op_rows.append({"score": name, "thr_val_recall": a.val_recall, "thr": thr, "gold_test_recall": rec_t, "gold_test_precision": prec_t, "n_gold": int(fin_t.sum()), | |
| "p_at_r30": precision_at_recall(yg[fin_t], gsc[fin_t], 0.3), "p_at_r50": precision_at_recall(yg[fin_t], gsc[fin_t], 0.5), "p_at_r70": precision_at_recall(yg[fin_t], gsc[fin_t], 0.7), | |
| "alerts_per_hour_manual": aph["manual"][2], "alerts_per_hour_engaged": aph["engaged"][2], "n_alerts_manual": aph["manual"][0], "n_alerts_engaged": aph["engaged"][0], | |
| "hours_manual": aph["manual"][1], "hours_engaged": aph["engaged"][1], "lead_detected_frac": float(det.mean()) if len(det) else np.nan, | |
| "lead_median_s": float(np.nanmedian(leads)) if det.any() else np.nan, "lead_mean_s": float(np.nanmean(leads)) if det.any() else np.nan, "n_gold_pos": int(len(leads)), | |
| "base_ckpts": json.dumps(base_ckpts.get("base_ckpts", {}))}) | |
| log(f" operating {name}: thr {thr:.4f} gold recall {rec_t:.2f} precision {prec_t:.2f}; alerts/h manual {aph['manual'][2]:.1f} engaged {aph['engaged'][2]:.1f}; " | |
| f"lead median {op_rows[-1]['lead_median_s']:.1f} s on {det.mean():.0%} detected") | |
| # ---- DMS attention chain (legacy att@3s = first passage of the inattention union, tau 15) on the test split: its alert rate at ITS | |
| # clean-val-recall-0.5 threshold (val scores from the legacy episodes CSV), then every score re-thresholded to the SAME alerts/hour | |
| dms_rows = [] | |
| try: | |
| sys.path.insert(0, str(CWM2)) | |
| from compose import legacy_compose | |
| att_val = G["att@3s"].to_numpy(float) if "att@3s" in G else np.full(len(G), np.nan) | |
| fin_v = gv & np.isfinite(att_val) | |
| thr_dms = threshold_at_recall(yg[fin_v], att_val[fin_v], a.val_recall) | |
| dms = {} | |
| for r in REGIMES: | |
| z = np.load(Path(a.dms_marginals) / f"marginals_{r}_s{seed_run}.npz", allow_pickle=True) | |
| sc_dms = legacy_compose(z["pA"].astype(np.float32), None, TAU)["att_only"] | |
| dms[r] = (z["uid"], z["t"].astype(float), sc_dms) | |
| aph_dms = {r: alerts_per_hour(dms[r][0], dms[r][1], dms[r][2], thr_dms) for r in REGIMES} | |
| fin_t = gt & np.isfinite(att_val) | |
| rec_t = float(((att_val[fin_t] >= thr_dms) & (yg[fin_t] == 1)).sum() / max(yg[fin_t].sum(), 1)) | |
| prec_t = float(((att_val[fin_t] >= thr_dms) & (yg[fin_t] == 1)).sum() / max((att_val[fin_t] >= thr_dms).sum(), 1)) | |
| target_rate = {r: aph_dms[r][2] for r in REGIMES} | |
| dms_rows.append({"score": "DMS att@3s (legacy_att)", "rule": f"own threshold at clean-val recall {a.val_recall}", "thr": thr_dms, "gold_test_recall": rec_t, "gold_test_precision": prec_t, | |
| "alerts_per_hour_manual": aph_dms["manual"][2], "alerts_per_hour_engaged": aph_dms["engaged"][2], "n_rows_manual": len(dms["manual"][2]), "n_rows_engaged": len(dms["engaged"][2]), | |
| "lead_median_s": np.nan, "lead_detected_frac": np.nan, "n_gold": int(fin_t.sum())}) | |
| log(f" DMS att@3s: thr {thr_dms:.4f} gold recall {rec_t:.2f} precision {prec_t:.2f}; alerts/h manual {aph_dms['manual'][2]:.1f} engaged {aph_dms['engaged'][2]:.1f} (legacy marginals rows)") | |
| def thr_for_rate(uid, t, s, rate, lo=None, hi=None, it=18): | |
| """bisection on the threshold so that alerts/hour (60 s cooldown) matches `rate`.""" | |
| fin = np.isfinite(s) | |
| if not fin.any(): | |
| return np.nan # 2026-09-18: score undefined on every row (missing-modality reference) -> no threshold | |
| lo, hi = float(np.nanmin(s[fin])), float(np.nanmax(s[fin])) | |
| for _ in range(it): | |
| mid = 0.5 * (lo + hi) | |
| r_ = alerts_per_hour(uid, t, s, mid)[2] | |
| if r_ > rate: | |
| lo = mid | |
| else: | |
| hi = mid | |
| return 0.5 * (lo + hi) | |
| for name, getter in getters: | |
| sc = {(r, s): getter(r, s) for r in REGIMES for s in ("test", "val")} | |
| thr_r = {r: thr_for_rate(packs[(r, "test")]["uid"], packs[(r, "test")]["t"].astype(float), sc[(r, "test")], target_rate[r]) for r in REGIMES} | |
| gsc = np.full(len(G), np.nan) | |
| thr_row = np.full(len(G), np.nan) | |
| for (r, s), g in G.groupby(["regime", "split"]): | |
| gsc[g.index.to_numpy()] = sc[(r, s)][g.row.to_numpy()] | |
| thr_row[g.index.to_numpy()] = thr_r[r] | |
| fin_t = gt & np.isfinite(gsc) | |
| hit = gsc[fin_t] >= thr_row[fin_t] | |
| rec_t = float((hit & (yg[fin_t] == 1)).sum() / max(yg[fin_t].sum(), 1)) | |
| prec_t = float((hit & (yg[fin_t] == 1)).sum() / max(hit.sum(), 1)) | |
| leads = np.array([lead_times(G_pos[G_pos.regime == r], {k: (packs[k]["uid"], packs[k]["t"].astype(float), v) for k, v in sc.items()}, thr_r[r]) for r in REGIMES], dtype=object) | |
| leads = np.concatenate([np.asarray(l, float) for l in leads]) if len(G_pos) else np.array([]) | |
| det = np.isfinite(leads) | |
| aph = {r: alerts_per_hour(packs[(r, "test")]["uid"], packs[(r, "test")]["t"].astype(float), sc[(r, "test")], thr_r[r])[2] for r in REGIMES} | |
| dms_rows.append({"score": name, "rule": "threshold per regime matched to the DMS alerts/hour", "thr": json.dumps({r: round(v, 5) for r, v in thr_r.items()}), "gold_test_recall": rec_t, | |
| "gold_test_precision": prec_t, "alerts_per_hour_manual": aph["manual"], "alerts_per_hour_engaged": aph["engaged"], "n_rows_manual": "", "n_rows_engaged": "", | |
| "lead_median_s": float(np.nanmedian(leads)) if det.any() else np.nan, "lead_detected_frac": float(det.mean()) if len(det) else np.nan, "n_gold": int(fin_t.sum())}) | |
| log(f" @DMS rate {name}: gold recall {rec_t:.2f} precision {prec_t:.2f}; alerts/h manual {aph['manual']:.1f} engaged {aph['engaged']:.1f}; lead median {dms_rows[-1]['lead_median_s']:.1f} s on {det.mean() if len(det) else float('nan'):.0%}") | |
| except Exception as ex: # noqa: BLE001 | |
| log(f" DMS-rate-matched rows unavailable: {ex!r}") | |
| pd.DataFrame(dms_rows).to_csv(out / "operating_dms_rate.csv", index=False) | |
| pd.DataFrame(op_rows).to_csv(out / "operating.csv", index=False) | |
| write_run_json(out, "train_warning_head", [Path(__file__), Path(LB.__file__), Path(WJ.__file__)], | |
| [rd / f"pred_{r}_{s}.npz" for r in REGIMES for s in ("test", "val")] + [rd / f"pred_{r}_train_labelrows.npz" for r in REGIMES], seed=a.seed, t_start=t0, | |
| input_listing=False, extra={"run": a.run, "seed_run": seed_run, "sources": sources, "feature_sets": selected, "n_labels": int(len(y)), "n_pos": int(y.sum()), | |
| "extra_features": list(EXTRA_OPTS["sets"]), "mask_routes": a.mask_routes, "per_regime": bool(a.per_regime), "ext_tag": ext_tag, "gaze_run": gaze_run, "fs_run": fs_run, | |
| "no_can": nocan_info, "gold_quick": gold_rows, "feature_set_deltas": fs_rows, "label_curve_summary": summ.reset_index().to_dict("records"), "operating": op_rows, | |
| "layout_dir": str(wout), "base_ckpts": base_ckpts, "dms_fields_used": [], "h5_note": getattr(tab, "attrs", {}).get("h5_note", "")}) | |
| log(f"DONE in {time.time() - t0:.0f}s -> {out}") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 57.7 kB
- Xet hash:
- e4542406c239df2f9239554cc1563c431241db4a8492bfe2d762b7770f5b3a54
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.