Buckets:
| #!/usr/bin/env python3 | |
| """cwm3/baselines_dms/fit_u_base.py — LAYER 1 of the DMS-baseline comparison (plan §2 第一层 / §3 fit_u_base, 2026-09-14): train the UNCHANGED U0 | |
| survivor recipe with a baseline's channels substituted for the 11 driver-kinematics channels, then pair it against the existing U0 base with the | |
| UNCHANGED eval/paired_compare_joint.py (the B1 protocol) and append a section to results/tables/dms_driver_onsets.md. | |
| INPUTS | |
| results/baselines_dms/summ/<b>/summ_{regime}_{split}.npz summarise.py output, row-aligned to s12_wm/results/targets_{regime}_{split}.npz: | |
| S0 f32[N,C] (latest valid 10 Hz value <= t, S0_TOL 0.15 s) + S0valid, channels[C]; window baselines: emb_anchor f16[N,D] (+ n_win_8s); row_ok | |
| results/baselines_dms/cache/<b>/<uid>.npz (optional, per-frame 10 Hz route caches) -> `--source cache` gives the EXACT 10 Hz history through | |
| dataset_adapter.write_substitute_kin_pack (route_hist / summarise.window_index, causal by assertion); `--source summ` (default when no cache | |
| directory exists, always for the window baselines) reconstructs the history from the anchor rows of the SAME route: the step at t - 0.1 j is the | |
| S0 of the anchor row whose time equals it (5 Hz anchor grid -> every odd 10 Hz step is INVALID, Khvalid False; recorded as history_source). | |
| ANCHOR SOURCE (2026-09-15, additive `--anchor-source {auto,S0,emb}`, layer-1 launch): the window baselines' summaries carry BOTH a non-empty S0 block (videomae2: | |
| 3 crop-geometry channels vmae2_crop_w / _h / lb_scale; ov_action: the 512 encoder units ova_enc_*) AND the window quantity emb_anchor (videomae2: 384-d ViT-S | |
| embedding; ov_action: the 9 action logits). The substitute pack of a window baseline is built from emb_anchor — resolve_anchor_source: auto = emb for | |
| videomae2 / ov_action, S0 for every per-frame face baseline (ov_gaze, mp_face, openface3, l2cs_gaze); an explicit S0 / emb overrides. C > 11 -> PCA-11 | |
| (fit_reducer_from_summaries: TRAIN-split summaries only — asserted on the file's `split` key — both regimes), applied to the anchor values BEFORE the history | |
| is assembled (a linear per-step map, so identical to reducing afterwards, without the [N, 80, 384] intermediate). Recorded in the provenance sidecar as | |
| anchor_source / history_source (summ_emb_anchor_... vs summ_S0_...). `--summ-root D` reads D/summ_{regime}_{split}.npz (the full-grid summaries | |
| results/baselines_dms/summ_full/<b>/ of the layer-1 launcher) instead of results/baselines_dms/summ/<b>/. | |
| OUTPUTS | |
| results/baselines_dms/kin_subst/<b>/kinhist_causal_{regime}_{split}.npz the substitute causal kin-history pack with EXACTLY the key set / dtypes / shapes | |
| of the real results/targets_kin_causal_routes_v2/kinhist_causal_*.npz (verify_pack asserts against it): | |
| uid <U, t f64, row_ok bool[N], Kh_c f16[N,40,11], Khvalid bool[N,40], Kh8_c f16[N,80,11], Kh8valid bool[N,80], channels <U[11], units <U, source <U | |
| channel count forced to 11 (DK): C < 11 -> the C channels, then `valid_flag` (1.0 on a valid step) and `pad_j` zeros (0.0 on a valid step); | |
| C == 11 -> as is; C > 11 -> the 11 leading components of a PCA fitted on TRAIN-split valid rows only (dataset_adapter.PCAReducer; saved to | |
| kin_subst/<b>/pca11.npz, its sha256 in the sidecar). Invalid steps are NaN on every channel (train/batch.std_kh validity = channels 0 and 10 finite). | |
| Sidecar kinhist_causal_{regime}_{split}.provenance.json: baseline, history_source, channels_used, reducer, source summary / cache sha256s, written_utc. | |
| results/baselines_dms/kin_subst/<b>/kin_causal_{regime}_{split}.npz SYMLINKS to the ORIGINAL causal targets (dataset_adapter.link_targets): the driver-onset | |
| targets stay those of the U0 reference run. | |
| results/runs/U0_dms_<b>_s<k>_kcv2_clip8_ln/ the trained run (train/train_cwm3.py writes ckpts, stats.npz, pred_*.npz, pred_layout_*.npz via | |
| eval/write_pred_layout.py, run_*.json, PRED_DONE) + u0_dms_provenance.json (this module's hook record). NEW dirs only (refuses an existing PRED_DONE). | |
| results/paired_compare.csv rows appended by eval/paired_compare_joint.py (tags dmsU_s<k>_<b>_vs_u0base and dmsU_s<k>_<b>_vs_k0) | |
| results/tables/dms_driver_onsets.md one appended section per baseline (A / B / Δ [CI] / verdict, eval/report_joint.render format; kill_checks.beats) | |
| HOW THE UNCHANGED TRAINER IS REPOINTED (file:line as of 2026-09-14; see also dataset_adapter.py header) | |
| train/train_cwm3.py main() :606 has no directory flag; :632 ds.set_kin_source(args.kin_source, args.kin_version); data/cwm3_dataset.py kin_dir() :82-84 returns | |
| the MODULE CONSTANT KINC_V2 :61 for --kin-version v2; files() :120-121 builds kinhist = kin_dir()/kinhist_causal_{r}_{s}.npz and kin = kin_dir()/kin_causal_{r}_{s}.npz; | |
| load_split :170-178 reads (uid, t, Kh_c, Khvalid, row_ok) and asserts the uid / t witness; the kinhist `channels` key is NOT asserted (:189 asserts the TARGETS file only). | |
| => train_u0() sets cwm3_dataset.KINC_V2 = kin_subst/<b> IN-PROCESS (both this module and train_cwm3 hold the same sys.modules['cwm3_dataset']), then calls | |
| train_cwm3.main() with sys.argv = the survivor argv. --no-lock runs the child inline (no os.fork), so the patched constant is what the trainer sees. | |
| Normalisation (--stats-version 2): train_cwm3.main :671 `st = bt.compute_stats(tr, target_stats(args), version=args.stats_version)` on the loaded TRAIN rows -> | |
| train/batch.py:82-91 nan-aware float64 mean / sd over Kh.reshape(-1, 11) -> kmu / ksd are RECOMPUTED from the substitute pack (never read from a file) and | |
| written to <run>/stats.npz :672-673. The --dry-run asserts this (pad channels kmu == 0, valid_flag kmu == 1). | |
| The exact argv (train_u0, identical to results/runs/U0_kin_can_noK_s<k>_kcv2_clip8_ln/run_train_*.json "command" + the explicit default --stats-version 2): | |
| train/train_cwm3.py --stage U0 --variant kin_can_noK_clip8_ln --seed <k> --kin-source causal_routes --kin-version v2 --stats-version 2 --out <run_dir> --no-lock | |
| The layout is written by the trainer itself (eval/write_pred_layout.py, arm cwm3_U; train_cwm3.py:10, :716) — eval/write_pred_layout_joint.py is the JOINT-run writer | |
| (12-chain pred packs) and does not apply to a U0 run; `ensure_layouts` calls the same write_pred_layout.write_layout if a pred pack lacks its layout. | |
| Pairing (exactly eval/run_joint_evals_final_k0.sh / run_joint_evals_final.sh): | |
| eval/paired_compare_joint.py --regime <r> --a results/runs/U0_dms_<b>_s<k>_kcv2_clip8_ln --b results/runs/U0_kin_can_noK_s<k>_kcv2_clip8_ln --tag dmsU_s<k>_<b>_vs_u0base --kin-targets causal_routes --kin-version v2 | |
| eval/paired_compare_joint.py --regime <r> --a results/runs/U0_dms_<b>_s<k>_kcv2_clip8_ln --b none --tag dmsU_s<k>_<b>_vs_k0 --kin-targets causal_routes --kin-version v2 --b-k0 | |
| ELIGIBILITY (plan §2): face-only baselines (ov_gaze, mp_face, openface3, l2cs_gaze, R0 lmface_rule) are scored on head_turn_onset only; their hand_leave / | |
| hand_return cells are printed as "n/a" (numbers kept in the CSV, descriptive); videomae2 and ov_action are scored on all three events. | |
| Nothing here reads a video, a model or any openpilot field. | |
| python baselines_dms/fit_u_base.py --baseline l2cs_gaze --build-packs [--source auto|summ|cache] # packs for every (regime, split) | |
| python baselines_dms/fit_u_base.py --baseline videomae2 --build-packs --source summ --anchor-source auto --summ-root results/baselines_dms/summ_full/videomae2 | |
| python baselines_dms/fit_u_base.py --baseline l2cs_gaze --coverage # CPU: kin_subst pack coverage section -> results/tables/dms_coverage.md | |
| python baselines_dms/fit_u_base.py --baseline l2cs_gaze --train --seed 0 # GPU (queue only; gate in run_queue_dms.sh) | |
| python baselines_dms/fit_u_base.py --baseline l2cs_gaze --pair --seeds 0 1 2 --table # CPU: pairing + table section | |
| python baselines_dms/fit_u_base.py --dry-run # CPU: synthetic summary -> pack -> 1 optimisation step | |
| """ | |
| 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 hashlib # noqa: E402 | |
| import json # noqa: E402 | |
| import subprocess # noqa: E402 | |
| import sys # noqa: E402 | |
| import time # noqa: E402 | |
| from pathlib import Path # noqa: E402 | |
| import numpy as np # noqa: E402 | |
| PKG = Path(__file__).resolve().parent | |
| CWM3 = PKG.parent | |
| EXPLORE = CWM3.parent | |
| CWM2 = EXPLORE / "cwm2" | |
| S12R = EXPLORE / "s12_wm" / "results" | |
| for _p in (PKG, CWM3 / "data", CWM3 / "train", CWM3 / "eval", CWM3 / "common", CWM2 / "eval"): | |
| if str(_p) not in sys.path: | |
| sys.path.insert(0, str(_p)) | |
| import cache_schema as cs # noqa: E402 | |
| import dataset_adapter as DA # noqa: E402 | |
| PY = sys.executable | |
| DK, HIST, HIST8, HZ = 11, 40, 80, 10 | |
| BASELINES = ("ov_gaze", "ov_action", "mp_face", "l2cs_gaze", "videomae2", "openface3") | |
| WINDOW_BASELINES = ("videomae2", "ov_action") # window caches: the anchor quantity is emb_anchor (summarise_windows), no per-frame semantic channels | |
| FACE_ONLY = ("ov_gaze", "mp_face", "openface3", "l2cs_gaze", "lmface_rule") | |
| ANCHOR_SOURCES = ("auto", "S0", "emb") # summary anchor block used for the substitute channels (see ANCHOR SOURCE in the module doc) | |
| COVERAGE_TABLE = CWM3 / "results" / "tables" / "dms_coverage.md" | |
| EVENTS = ("hand_leave", "hand_return", "head_turn_onset") | |
| CELLS = [(f"drv_onset|{e}|3s", "pr_auc") for e in EVENTS] + [(f"hazard_nll|{e}", "nll") for e in EVENTS] | |
| SUBST_ROOT = cs.RESULTS_ROOT / "kin_subst" | |
| RUNS = CWM3 / "results" / "runs" | |
| TABLE = CWM3 / "results" / "tables" / "dms_driver_onsets.md" | |
| PAIRED_CSV = CWM3 / "results" / "paired_compare.csv" | |
| REAL_KIN = CWM3 / "results" / "targets_kin_causal_routes_v2" | |
| REAL_KEYS = ("uid", "t", "row_ok", "Kh_c", "Khvalid", "Kh8_c", "Kh8valid", "channels", "units", "source") | |
| U0_BASE = "U0_kin_can_noK_s{seed}_kcv2_clip8_ln" | |
| U0_DMS = "U0_dms_{b}_s{seed}_kcv2_clip8_ln" | |
| VARIANT = "kin_can_noK_clip8_ln" | |
| TRAINER = CWM3 / "train" / "train_cwm3.py" | |
| PAIRER = CWM3 / "eval" / "paired_compare_joint.py" | |
| REGIMES, SPLITS = ("manual", "engaged"), ("train", "val", "test") | |
| def eligible_events(baseline): | |
| return ("head_turn_onset",) if baseline in FACE_ONLY else EVENTS | |
| def resolve_anchor_source(baseline, anchor_source="auto"): | |
| """'auto' -> 'emb' for the window baselines (videomae2, ov_action: S0 holds crop geometry / encoder units, the anchor quantity is emb_anchor), 'S0' otherwise.""" | |
| assert anchor_source in ANCHOR_SOURCES, anchor_source | |
| if anchor_source != "auto": | |
| return anchor_source | |
| return "emb" if baseline in WINDOW_BASELINES else "S0" | |
| def summ_path(baseline, regime, split, summ_root=None): | |
| """results/baselines_dms/summ/<b>/summ_{regime}_{split}.npz (summ_root overrides the <b> directory: tests / dry-run).""" | |
| return (Path(summ_root) if summ_root else cs.summ_dir(baseline)) / f"summ_{regime}_{split}.npz" | |
| def subst_dir(baseline): | |
| return SUBST_ROOT / baseline | |
| def sha16(p): | |
| return cs.sha256_file(p)[:16] | |
| # ----------------------------------------------------------------------------- summaries -> anchor-value block | |
| def load_anchor_block(path, anchor_source="auto", baseline=None): | |
| """summ npz -> dict(uid, t, row_ok, S f32[N,C] (NaN undefined), valid bool[N], channels[C], kind 'S0' | 'emb_anchor', anchor_source, baseline, split). | |
| anchor_source: 'S0' -> the S0 block (latest valid 10 Hz value <= t); 'emb' -> emb_anchor (last valid window ending <= t); 'auto' -> resolve_anchor_source | |
| of `baseline` (default: the file's `baseline` key) = emb for the window baselines, S0 otherwise. The requested block must be present and non-empty | |
| (no silent fallback: videomae2's S0 is crop geometry, its substitute channels must come from emb_anchor).""" | |
| with np.load(path, allow_pickle=True) as z: | |
| uid, t, row_ok = z["uid"], z["t"], z["row_ok"].astype(bool) | |
| base = str(z["baseline"]) if "baseline" in z.files else "" | |
| split = str(z["split"]) if "split" in z.files else "" | |
| kind = resolve_anchor_source(baseline if baseline is not None else base, anchor_source) | |
| S0 = z["S0"] if "S0" in z.files else np.zeros((len(uid), 0), np.float32) | |
| has_s0 = S0.ndim == 2 and S0.shape[1] > 0 | |
| if kind == "S0": | |
| if not has_s0: | |
| raise ValueError(f"{path}: anchor source S0 requested but the summary has no S0 channels" + (" (emb_anchor present: use --anchor-source emb)" if "emb_anchor" in z.files else "")) | |
| S = S0.astype(np.float32) | |
| valid = (z["S0valid"].astype(bool) if "S0valid" in z.files else np.isfinite(S).all(1)) & row_ok & np.isfinite(S).all(1) | |
| ch = [str(c) for c in z["channels"]] | |
| kind = "S0" | |
| else: | |
| if "emb_anchor" not in z.files: | |
| raise ValueError(f"{path}: anchor source emb requested but the summary has no emb_anchor (window baselines only)") | |
| S = z["emb_anchor"].astype(np.float32) | |
| valid = row_ok & np.isfinite(S).all(1) | |
| ch = [f"emb{j}" for j in range(S.shape[1])] | |
| kind = "emb_anchor" | |
| S = np.where(valid[:, None], S, np.nan).astype(np.float32) | |
| return {"uid": uid, "t": t, "row_ok": row_ok, "S": S, "valid": valid, "channels": ch, "kind": kind, "anchor_source": ("S0" if kind == "S0" else "emb"), "baseline": base, "split": split} | |
| def history_from_anchor_rows(uid, t, S, valid, hist=HIST, hz=HZ): | |
| """anchor-row values -> (Kh f32[N,hist,C] NaN where undefined, Kv bool[N,hist]) with step j (0..hist-1) at t - (hist-1-j)/hz taken from the SAME route's | |
| anchor row at exactly that time (causal: only rows with time <= t are ever indexed); no row at that time -> invalid. On the 5 Hz anchor grid every | |
| odd 10 Hz step is therefore invalid.""" | |
| N, C = S.shape | |
| Kh = np.full((N, hist, C), np.nan, np.float32) | |
| Kv = np.zeros((N, hist), bool) | |
| k_all = np.round(np.asarray(t, np.float64) * hz).astype(np.int64) | |
| offs = np.arange(-(hist - 1), 1) | |
| for u, rows in DA.SM._route_blocks(np.asarray(uid)): | |
| ks = k_all[rows] | |
| order = np.argsort(ks, kind="stable") | |
| ks_s, rows_s = ks[order], rows[order] | |
| target = ks[:, None] + offs[None, :] # [n, hist] absolute steps (all <= the anchor step) | |
| j = np.searchsorted(ks_s, target) | |
| hit = (j < len(ks_s)) | |
| jj = np.clip(j, 0, max(len(ks_s) - 1, 0)) | |
| hit &= ks_s[jj] == target | |
| src = rows_s[jj] | |
| ok = hit & valid[src] | |
| Kh[rows] = np.where(ok[..., None], S[src], np.nan) | |
| Kv[rows] = ok | |
| return Kh, Kv | |
| # ----------------------------------------------------------------------------- 11-channel forcing | |
| def force_dk(Kh, Kv, channels, reducer=None): | |
| """[N,T,C] (NaN invalid) -> ([N,T,11] f32, names[11]): C <= 11 -> channels + valid_flag (1.0) + pad_j (0.0) on valid steps; C > 11 -> reducer (PCA-11). | |
| Every channel is NaN exactly where the step is invalid (std_kh validity: channels 0 and 10 finite).""" | |
| N, T, C = Kh.shape | |
| if C > DK: | |
| assert reducer is not None, f"{C} channels > {DK}: a PCAReducer fitted on TRAIN rows is required" | |
| assert list(reducer.channels) == list(channels), "reducer fitted on different channels" | |
| Z = reducer.transform(Kh.reshape(-1, C)).reshape(N, T, DK) | |
| names = list(reducer.out_names) | |
| elif C == DK: | |
| Z, names = Kh.astype(np.float32).copy(), list(channels) | |
| else: | |
| Z = np.full((N, T, DK), np.nan, np.float32) | |
| Z[:, :, :C] = Kh | |
| Z[:, :, C] = 1.0 | |
| Z[:, :, C + 1:] = 0.0 | |
| names = list(channels) + ["valid_flag"] + [f"{DA.PAD_PREFIX}{j}" for j in range(C + 1, DK)] | |
| ok = Kv & np.isfinite(Z).all(-1) | |
| Z[~ok] = np.nan | |
| return Z.astype(np.float32), names, ok | |
| def fit_reducer_from_summaries(baseline, n_out=DK, max_rows=200_000, seed=0, out_dir=None, summ_root=None, regimes=REGIMES, anchor_source="auto", log=print): | |
| """PCA-n_out on the TRAIN-split valid anchor rows (both regimes) of the baseline's summaries (anchor block per resolve_anchor_source) -> (PCAReducer, path, sha256). | |
| Only summ_{regime}_train.npz files are read (asserted on the file's `split` key when present): val / test rows never enter the fit.""" | |
| parts, uids, chans = [], [], None | |
| rng = np.random.default_rng(seed) | |
| fit_files = [] | |
| for r in regimes: | |
| p = summ_path(baseline, r, "train", summ_root) | |
| if not p.exists(): | |
| continue | |
| blk = load_anchor_block(p, anchor_source, baseline) | |
| assert blk["split"] in ("", "train"), f"{p}: reducer fit requires TRAIN summaries (file split = {blk['split']!r})" | |
| chans = chans or blk["channels"] | |
| assert blk["channels"] == chans | |
| F = blk["S"][blk["valid"]] | |
| if len(F): | |
| parts.append(F[np.sort(rng.choice(len(F), min(max_rows // 2, len(F)), replace=False))]) | |
| uids += np.unique(blk["uid"][blk["valid"]]).tolist() | |
| fit_files.append(str(p)) | |
| assert parts, f"{baseline}: no TRAIN summaries with valid rows" | |
| red = DA.PCAReducer.fit(np.concatenate(parts), chans, n_out, baseline, sorted(set(uids))) | |
| out = (Path(out_dir) if out_dir else subst_dir(baseline)) / "pca11.npz" | |
| red.save(out) | |
| sha = cs.sha256_file(out) | |
| (out.parent / "pca11.fit.json").write_text(json.dumps({"baseline": baseline, "anchor_source": resolve_anchor_source(baseline, anchor_source), "fit_splits": ["train"], "fit_files": fit_files, | |
| "n_fit_rows": int(red.n_fit), "n_fit_uids": len(red.fit_uids), "evr": [float(x) for x in red.evr], "evr_sum": float(red.evr.sum()), | |
| "channels_in": chans, "sha256": sha, "written_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}, indent=1)) | |
| log(f"reducer {baseline}: {red.describe()} anchor_source {resolve_anchor_source(baseline, anchor_source)} evr {np.round(red.evr, 4).tolist()} -> {out} sha256 {sha[:16]}") | |
| return red, out, sha | |
| # ----------------------------------------------------------------------------- pack writer | |
| def _save_pack(path, uid, t, row_ok, Kh, Kv, Kh8, Kv8, names, baseline, provenance): | |
| path = Path(path) | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| np.savez_compressed(path, uid=uid, t=t, row_ok=row_ok, Kh_c=Kh.astype(np.float16), Khvalid=Kv, Kh8_c=Kh8.astype(np.float16), Kh8valid=Kv8, channels=np.array(names), | |
| units=np.array(f"raw {baseline} channels (standardised by the trainer's TRAIN statistics, train/batch.compute_stats v2); valid_flag 1, pads 0"), | |
| source=np.array(f"baselines_dms:{baseline}")) | |
| provenance = dict(provenance, pack_sha256=cs.sha256_file(path), written_utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())) | |
| (path.parent / (path.stem + ".provenance.json")).write_text(json.dumps(provenance, indent=1, default=str)) | |
| return path | |
| def build_pack_from_summary(baseline, regime, split, out_dir=None, reducer=None, summ_root=None, anchor_source="auto", log=print): | |
| """summ_{regime}_{split}.npz -> kinhist_causal_{regime}_{split}.npz (history_source summ_S0 / summ_emb_anchor: 5 Hz anchor rows; anchor block per | |
| resolve_anchor_source). A reducer (C > 11) is applied to the anchor values BEFORE the history is assembled: PCA is a linear per-step map, so the result equals | |
| reducing the assembled history, without the [N, 80, C] intermediate (videomae2: C = 384 -> 77 GB).""" | |
| t0 = time.time() | |
| out_dir = Path(out_dir) if out_dir else subst_dir(baseline) | |
| sp = summ_path(baseline, regime, split, summ_root) | |
| blk = load_anchor_block(sp, anchor_source, baseline) | |
| with np.load(S12R / f"targets_{regime}_{split}.npz", allow_pickle=True) as z: | |
| assert np.array_equal(z["uid"], blk["uid"]) and np.array_equal(z["t"], blk["t"]), f"{sp}: not aligned with the s12 pack" | |
| if isinstance(reducer, (str, Path)): | |
| reducer = DA.PCAReducer.load(reducer) | |
| S, ch_hist = blk["S"], list(blk["channels"]) | |
| if reducer is not None: | |
| assert len(blk["channels"]) > DK, f"reducer given for C = {len(blk['channels'])} <= {DK}" | |
| assert list(reducer.channels) == list(blk["channels"]), "reducer fitted on different channels" | |
| S = reducer.transform(S) # [N, 11] (NaN where the anchor value is undefined) | |
| ch_hist = list(reducer.out_names) | |
| Kh, Kv = history_from_anchor_rows(blk["uid"], blk["t"], S, blk["valid"], HIST) | |
| Kh8, Kv8 = history_from_anchor_rows(blk["uid"], blk["t"], S, blk["valid"], HIST8) | |
| Z, names, ok = force_dk(Kh, Kv, ch_hist, None) | |
| Z8, _, ok8 = force_dk(Kh8, Kv8, ch_hist, None) | |
| row_ok = blk["row_ok"] & ok.any(1) | |
| Z[~row_ok], Z8[~row_ok] = np.nan, np.nan | |
| ok[~row_ok], ok8[~row_ok] = False, False | |
| prov = {"baseline": baseline, "regime": regime, "split": split, "history_source": f"summ_{blk['kind']}_5Hz_anchor_rows (odd 10 Hz steps invalid)", "anchor_source": blk["anchor_source"], | |
| "anchor_source_rule": "resolve_anchor_source: auto = emb (emb_anchor) for the window baselines videomae2 / ov_action, S0 for the per-frame face baselines; explicit S0 / emb overrides", | |
| "n_channels_in": len(blk["channels"]), "channels_in": blk["channels"], "channels_out": names, "reducer": (reducer.describe() if reducer is not None else None), | |
| "reducer_applied": ("to the anchor values before the history (linear per-step map)" if reducer is not None else None), | |
| "reducer_sha256": (cs.sha256_file(out_dir / "pca11.npz") if (reducer is not None and (out_dir / "pca11.npz").exists()) else None), | |
| "source_summary": str(sp), "source_summary_sha256": cs.sha256_file(sp), "rows_ok": int(row_ok.sum()), "khvalid_share_given_ok": float(ok[row_ok].mean()) if row_ok.any() else None, | |
| "trainer_contract": "data/cwm3_dataset.load_split reads uid, t, Kh_c, Khvalid, row_ok; kin_dir() -> this directory via cwm3_dataset.KINC_V2 (fit_u_base.train_u0)"} | |
| p = _save_pack(out_dir / f"kinhist_causal_{regime}_{split}.npz", blk["uid"], blk["t"], row_ok, Z, ok, Z8, ok8, names, baseline, prov) | |
| log(f"{baseline} {regime}/{split}: N={len(row_ok):,} row_ok {row_ok.mean():.4f} Khvalid|ok {prov['khvalid_share_given_ok']} anchor {blk['anchor_source']} C_in {len(blk['channels'])} -> {names} ({time.time() - t0:.1f}s) {p}") | |
| return p | |
| def build_pack_from_cache(baseline, regime, split, out_dir=None, reducer=None, cache_dir=None, log=print): | |
| """per-frame route caches -> the same pack through dataset_adapter.write_substitute_kin_pack (exact 10 Hz history), re-saved with the real key set.""" | |
| out_dir = Path(out_dir) if out_dir else subst_dir(baseline) | |
| tmp = out_dir / "_tmp_cache_pack" | |
| p0 = DA.write_substitute_kin_pack(baseline, regime, split, tmp, cache_dir=cache_dir, reducer=reducer, with_hist8=True, log=log) | |
| with np.load(p0, allow_pickle=True) as z: | |
| Kh, Kv, Kh8, Kv8 = z["Kh_c"].astype(np.float32), z["Khvalid"], z["Kh8_c"].astype(np.float32), z["Kh8valid"] | |
| names = [str(c) for c in z["channels"]] | |
| uid, t, row_ok = z["uid"], z["t"], z["row_ok"] | |
| adapter = json.loads(str(z["adapter"])) | |
| if any(n.startswith(DA.PAD_PREFIX) for n in names): # the adapter pads with zeros only: make the first pad the valid_flag | |
| j = names.index(next(n for n in names if n.startswith(DA.PAD_PREFIX))) | |
| Kh[:, :, j] = np.where(Kv, 1.0, np.nan) | |
| Kh8[:, :, j] = np.where(Kv8, 1.0, np.nan) | |
| names[j] = "valid_flag" | |
| prov = {"baseline": baseline, "regime": regime, "split": split, "history_source": "cache_10Hz (dataset_adapter.route_hist / summarise.window_index)", "channels_out": names, | |
| "anchor_source": "cache_frames (per-frame channels of results/baselines_dms/cache/<b>/<uid>.npz; the summary anchor-source flag does not apply)", | |
| "rows_ok": int(np.asarray(row_ok).sum()), "khvalid_share_given_ok": float(Kv[np.asarray(row_ok).astype(bool)].mean()) if np.asarray(row_ok).any() else None, | |
| "adapter": adapter, "trainer_contract": adapter.get("trainer_contract")} | |
| p = _save_pack(out_dir / f"kinhist_causal_{regime}_{split}.npz", uid, t, row_ok, Kh, Kv, Kh8, Kv8, names, baseline, prov) | |
| p0.unlink() | |
| return p | |
| def build_packs(baseline, source="auto", out_dir=None, regimes=REGIMES, splits=SPLITS, summ_root=None, anchor_source="auto", cache_dir=None, log=print): | |
| """every (regime, split) pack + PCA reducer when C > 11 + target symlinks -> list of paths. source auto: cache (exact 10 Hz per-frame history through | |
| dataset_adapter.write_substitute_kin_pack; routes without a complete <uid>.npz stay row_ok False) when results/baselines_dms/cache/<b> holds complete route | |
| caches and the baseline is a per-frame baseline, else summ (5 Hz anchor rows; always for the window baselines, whose anchor block is chosen by anchor_source).""" | |
| out_dir = Path(out_dir) if out_dir else subst_dir(baseline) | |
| out_dir.mkdir(parents=True, exist_ok=True) | |
| cdir = Path(cache_dir) if cache_dir else cs.cache_dir(baseline) | |
| complete = sorted(p for p in cdir.glob("*.npz") if not p.name.endswith(".part.npz")) if cdir.exists() else [] | |
| use_cache = source == "cache" or (source == "auto" and baseline not in WINDOW_BASELINES and len(complete) > 0) | |
| assert not (use_cache and not complete), f"{cdir}: no complete route cache for --source cache" | |
| ref = summ_path(baseline, regimes[0], "train", summ_root) | |
| assert ref.exists() or use_cache, f"{ref} missing (summaries are produced by summarise.py)" | |
| asrc = resolve_anchor_source(baseline, anchor_source) | |
| log(f"{baseline}: history source = {'cache_10Hz (' + str(len(complete)) + ' complete route caches under ' + str(cdir) + ')' if use_cache else 'summ (5 Hz anchor rows, anchor block ' + asrc + ', summ root ' + str(ref.parent) + ')'}" | |
| + (f"; --source {source}" if source != "auto" else "; --source auto") + f"; --anchor-source {anchor_source} -> {asrc}") | |
| reducer = None | |
| if use_cache: | |
| rc = cs.RouteCache.load(complete[0]) | |
| if rc.C > DK: | |
| reducer = DA.fit_reducer(baseline, cdir, out_path=out_dir / "pca11.npz", log=log) | |
| else: | |
| C = len(load_anchor_block(ref, anchor_source, baseline)["channels"]) | |
| if C > DK: | |
| reducer, _, _ = fit_reducer_from_summaries(baseline, out_dir=out_dir, summ_root=summ_root, regimes=regimes, anchor_source=anchor_source, log=log) | |
| paths = [] | |
| for r in regimes: | |
| for s in splits: | |
| if not use_cache and not summ_path(baseline, r, s, summ_root).exists(): | |
| log(f"{baseline} {r}/{s}: summary missing -> skipped") | |
| continue | |
| paths.append(build_pack_from_cache(baseline, r, s, out_dir, reducer, cache_dir=(str(cdir) if cache_dir else None), log=log) if use_cache | |
| else build_pack_from_summary(baseline, r, s, out_dir, reducer, summ_root, anchor_source, log=log)) | |
| DA.link_targets(out_dir, "v2", regimes, splits) | |
| for p in paths: | |
| verify_pack(p) | |
| return paths | |
| def verify_pack(path, ref_dir=REAL_KIN): | |
| """the substitute pack has EXACTLY the real pack's keys, dtypes (kind + itemsize for the numeric arrays) and shapes (string widths excepted).""" | |
| path = Path(path) | |
| ref = Path(ref_dir) / path.name | |
| with np.load(path, allow_pickle=True) as a, np.load(ref, allow_pickle=True) as b: | |
| assert set(a.files) == set(b.files) == set(REAL_KEYS), (sorted(a.files), sorted(b.files)) | |
| for k in REAL_KEYS: | |
| va, vb = a[k], b[k] | |
| assert va.shape == vb.shape, (k, va.shape, vb.shape) | |
| if vb.dtype.kind in "US": | |
| assert va.dtype.kind == vb.dtype.kind, (k, va.dtype, vb.dtype) | |
| else: | |
| assert va.dtype == vb.dtype, (k, va.dtype, vb.dtype) | |
| assert np.array_equal(a["uid"], b["uid"]) and np.array_equal(a["t"], b["t"]), "uid / t differ from the real pack (s12 witness)" | |
| Kh, V = a["Kh_c"].astype(np.float32), a["Khvalid"] | |
| assert np.isfinite(Kh[V][:, [0, DK - 1]]).all() and np.isnan(Kh[~V]).all(), "std_kh validity rule violated" | |
| return True | |
| # ----------------------------------------------------------------------------- trainer hook | |
| def trainer_argv(seed, out_dir, extra=()): | |
| """the survivor recipe argv (== the recorded command of U0_kin_can_noK_s<k>_kcv2_clip8_ln + the explicit default --stats-version 2).""" | |
| return ["--stage", "U0", "--variant", VARIANT, "--seed", str(int(seed)), "--kin-source", "causal_routes", "--kin-version", "v2", "--stats-version", "2", | |
| "--out", str(out_dir), "--no-lock", *extra] | |
| def train_u0(baseline, seed, out_dir=None, sub_dir=None, extra_argv=(), quick_csv=None, log=print): | |
| """repoint cwm3_dataset.KINC_V2 at the substitute pack and run train_cwm3.main() with the survivor argv (the trainer recomputes kmu / ksd from the pack).""" | |
| import cwm3_dataset as ds | |
| import train_cwm3 as T | |
| sub_dir = Path(sub_dir) if sub_dir else subst_dir(baseline) | |
| out_dir = Path(out_dir) if out_dir else RUNS / U0_DMS.format(b=baseline, seed=seed) | |
| assert not (out_dir / "PRED_DONE").exists(), f"{out_dir} already finished: new dirs only" | |
| for r in REGIMES: | |
| for s in SPLITS: | |
| assert (sub_dir / f"kinhist_causal_{r}_{s}.npz").exists() or "--regimes" in extra_argv, f"pack {r}/{s} missing in {sub_dir}" | |
| DA.link_targets(sub_dir, "v2") | |
| argv = trainer_argv(seed, out_dir, extra_argv) | |
| saved_const, saved_src, saved_argv, saved_quick = ds.KINC_V2, dict(ds._KIN_SOURCE), list(sys.argv), T.QUICK_CSV | |
| hook = {"baseline": baseline, "seed": int(seed), "sub_dir": str(sub_dir), "hook": "cwm3_dataset.KINC_V2 = sub_dir (in-process; kin_dir() :82-84 -> files() :120-121)", | |
| "argv": [str(TRAINER), *argv], "stats": "recomputed by train_cwm3.main :671 -> train/batch.compute_stats v2 (:82-91) on the substitute TRAIN rows; written to <run>/stats.npz", | |
| "pack_sha256": {p.name: cs.sha256_file(p) for p in sorted(sub_dir.glob("kinhist_causal_*.npz"))}, "t_start_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())} | |
| t0 = time.time() | |
| try: | |
| ds.KINC_V2 = sub_dir | |
| ds.set_kin_source("causal_routes", "v2") | |
| assert ds.kin_dir() == sub_dir and ds.files("manual", "val")["kinhist"].parent == sub_dir | |
| if quick_csv is not None: | |
| T.QUICK_CSV = Path(quick_csv) | |
| sys.argv = [str(TRAINER), *argv] | |
| log(f"train_u0 {baseline} s{seed}: KINC_V2 -> {sub_dir}; argv {' '.join(argv)}") | |
| T.main() | |
| finally: | |
| ds.KINC_V2, sys.argv, T.QUICK_CSV = saved_const, saved_argv, saved_quick | |
| ds._KIN_SOURCE.update(saved_src) | |
| run_json = sorted(out_dir.glob("run_train_*_latest.json")) | |
| rec = json.loads(run_json[-1].read_text()) if run_json else {} | |
| assert Path(rec.get("kin_dir", "")) == sub_dir, f"run.json kin_dir {rec.get('kin_dir')} != {sub_dir}" | |
| st = dict(np.load(out_dir / "stats.npz")) | |
| hook.update(wall_s=round(time.time() - t0, 1), run_json=str(run_json[-1]) if run_json else "", kin_dir_recorded=rec.get("kin_dir"), stats_version=int(rec.get("stats_version", 0)), | |
| kmu=[float(x) for x in st["kmu"]], ksd=[float(x) for x in st["ksd"]], best_epoch=rec.get("best_epoch"), epochs_run=rec.get("epochs_run"), val_objective=rec.get("val_objective")) | |
| (out_dir / "u0_dms_provenance.json").write_text(json.dumps(hook, indent=1)) | |
| return out_dir | |
| def ensure_layouts(run_dir, regimes=REGIMES, splits=("test", "val"), log=print): | |
| """write_pred_layout.write_layout (the base-branch writer the trainer calls at :716) for any pred pack that lacks its layout — normally a no-op.""" | |
| import write_pred_layout as wpl | |
| run_dir = Path(run_dir) | |
| cfg = json.loads((run_dir / "config.json").read_text())["cfg"] | |
| run_cfg = {"stage": cfg["stage"], "variant": cfg["variant"], "cfg": cfg, "roi_streams": ["Kh", "K", "S"], "roi": "causal_clip_roi"} | |
| code_files = sorted((CWM3 / "models").glob("*.py")) + sorted((CWM3 / "train").glob("*.py")) + [CWM3 / "eval" / "write_pred_layout.py", CWM3 / "data" / "cwm3_dataset.py", EXPLORE / "s12_wm/eval_s13.py"] | |
| out = [] | |
| for r in regimes: | |
| for s in splits: | |
| if (run_dir / f"pred_{r}_{s}.npz").exists() and not (run_dir / f"pred_layout_{r}_{s}.npz").exists(): | |
| out.append(wpl.write_layout(str(run_dir), r, s, run_cfg, code_files, log=log)) | |
| return out | |
| # ----------------------------------------------------------------------------- label-row prediction (for fit_readout --with-u0) | |
| def predict_label_rows(baseline, seed, rows_from=None, run_dir=None, sub_dir=None, regimes=REGIMES, force=False, log=print): | |
| """run the U0_dms_<b>_s<k> BEST model on the warning head's TRAIN label rows (eval/predict_u0_rows.py pattern: train_cwm3.model_kwargs / batch.prep_U unchanged, CPU) | |
| -> <run>/pred_u0_{regime}_train_labelrows.npz: rows, uid, t, onset_p f32[n,4 events,4 taus] (hand_return = hands_on.on, hand_leave = hands_on.off, head_turn_onset = | |
| head_turned.on, head_return = head_turned.off; taus 5 / 10 / 15 / 25 steps; NaN where no causal / baseline row), h_on / h_off f32[n,25,2], z0 f32[n,2], row_ok.""" | |
| import torch | |
| import cwm3_dataset as ds | |
| import batch as bt | |
| import train_cwm3 as T | |
| from models import build_branch | |
| from models import heads as hd | |
| torch.set_num_threads(4) | |
| rd = Path(run_dir) if run_dir else RUNS / U0_DMS.format(b=baseline, seed=seed) | |
| prov = json.loads((rd / "u0_dms_provenance.json").read_text()) if (rd / "u0_dms_provenance.json").exists() else {} | |
| sub_dir = Path(sub_dir) if sub_dir else Path(prov.get("sub_dir", subst_dir(baseline))) | |
| rows_from = Path(rows_from) if rows_from else RUNS / f"I1_all_minus_road2drv_s{seed}_final" / "label_rows.npz" | |
| cfgj = json.loads((rd / "config.json").read_text()) | |
| cfg, seed_run = cfgj["cfg"], int(cfgj["args"]["seed"]) | |
| best = (rd / "BEST").read_text().strip() | |
| z = torch.load(rd / best, map_location="cpu", weights_only=False) | |
| model = build_branch("U", seed_run, **T.model_kwargs(cfg)) | |
| model.load_state_dict(z["model"]) | |
| model.eval() | |
| st = {k: v for k, v in np.load(rd / "stats.npz").items()} | |
| lr = np.load(rows_from) | |
| taus = (5, 10, 15, 25) | |
| events = ("hand_return", "hand_leave", "head_turn_onset", "head_return") | |
| saved_const, saved_src = ds.KINC_V2, dict(ds._KIN_SOURCE) | |
| outs = [] | |
| try: | |
| ds.KINC_V2 = sub_dir | |
| ds.set_kin_source("causal_routes", "v2") | |
| ds.set_pack_version(cfg.get("pack_version", "v1") or "v1") | |
| for regime in regimes: | |
| op = rd / f"pred_u0_{regime}_train_labelrows.npz" | |
| if op.exists() and not force: | |
| outs.append(op) | |
| continue | |
| rows = np.sort(lr[regime].astype(np.int64)) | |
| raw = ds.load_regimes([regime], "train", T.need_streams(cfg), verbose=lambda m: None) | |
| sub = ds.subset(raw, rows) | |
| del raw | |
| n = len(rows) | |
| store = bt.GpuStore(sub, bt.KEYS["U"], "cpu") | |
| with torch.no_grad(): | |
| b, tg = bt.prep_U(store, torch.arange(n), st, cfg, True) | |
| o = model(b) | |
| u = hd.split_U(o["U"]) | |
| h_on, h_off = hd.hazards(u["chains"]) | |
| h_on, h_off = h_on.float().clamp(1e-6, 1 - 1e-6), h_off.float().clamp(1e-6, 1 - 1e-6) | |
| src = {"hand_return": h_on[:, :, 0], "hand_leave": h_off[:, :, 0], "head_turn_onset": h_on[:, :, 1], "head_return": h_off[:, :, 1]} | |
| onset = torch.stack([torch.stack([1 - torch.prod(1 - src[e][:, :tau], 1) for tau in taus], 1) for e in events], 1) # [n,4 events,4 taus] | |
| z0 = tg["z0"].float() | |
| row_ok = sub["kin_row_ok"].astype(bool) if "kin_row_ok" in sub else np.ones(n, bool) | |
| onset, h_on, h_off, z0 = (x.numpy().astype(np.float32) for x in (onset, h_on, h_off, z0)) | |
| for arr in (onset, h_on, h_off, z0): | |
| arr[~row_ok] = np.nan | |
| np.savez_compressed(op, rows=rows, uid=sub["uid"], t=sub["t"], onset_p=onset, event_names=np.array(events), taus=np.array(taus), h_on=h_on, h_off=h_off, z0=z0, row_ok=row_ok, | |
| best_ckpt=np.array(best), run=np.array(rd.name), sub_dir=np.array(str(sub_dir)), source_rows=np.array(str(rows_from)), regime=np.array(regime), split=np.array("train")) | |
| log(f"{op.name}: {n} rows, row_ok {row_ok.mean():.3f}, onset_p@3s hand_leave mean {np.nanmean(onset[:, 1, 2]):.3f}") | |
| outs.append(op) | |
| del sub, store | |
| finally: | |
| ds.KINC_V2 = saved_const | |
| ds._KIN_SOURCE.update(saved_src) | |
| return outs | |
| # ----------------------------------------------------------------------------- pairing (unchanged evaluator) + table | |
| def pair_argv(baseline, seed, regime, k0=False): | |
| a = RUNS / U0_DMS.format(b=baseline, seed=seed) | |
| argv = [str(PAIRER), "--regime", regime, "--a", str(a.relative_to(CWM3))] | |
| if k0: | |
| argv += ["--b", "none", "--tag", f"dmsU_s{seed}_{baseline}_vs_k0", "--kin-targets", "causal_routes", "--kin-version", "v2", "--b-k0"] | |
| else: | |
| argv += ["--b", str((RUNS / U0_BASE.format(seed=seed)).relative_to(CWM3)), "--tag", f"dmsU_s{seed}_{baseline}_vs_u0base", "--kin-targets", "causal_routes", "--kin-version", "v2"] | |
| return argv | |
| def run_pairs(baseline, seeds=(0, 1, 2), regimes=REGIMES, k0=True, log=print, dry=False): | |
| cmds = [] | |
| for s in seeds: | |
| for r in regimes: | |
| for kk in ((False, True) if k0 else (False,)): | |
| argv = [PY, *pair_argv(baseline, s, r, kk)] | |
| cmds.append(argv) | |
| log("pair: " + " ".join(argv)) | |
| if not dry: | |
| rc = subprocess.call(argv, cwd=str(CWM3)) | |
| assert rc == 0, f"paired_compare_joint exited {rc}" | |
| return cmds | |
| def _latest_rows(tags): | |
| import pandas as pd | |
| df = pd.read_csv(PAIRED_CSV, low_memory=False) | |
| df = df[df.tag.isin(tags)].copy() | |
| if not len(df): | |
| return df | |
| df["seed"] = df.tag.str.extract(r"_s(\d+)_")[0].astype(int) | |
| return df.sort_values("run_stamp").groupby(["tag", "regime", "split", "cell", "metric"], as_index=False).tail(1) | |
| def render_section(baseline, seeds=(0, 1, 2), csv_path=PAIRED_CSV): | |
| """A / B / Δ [CI] / verdict lines (eval/report_joint.render format; verdict = kill_checks.beats over the seeds) for the driver cells of the two pairings.""" | |
| import pandas as pd | |
| import kill_checks as KC | |
| elig = set(eligible_events(baseline)) | |
| stamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) | |
| lines = [f"## {baseline} — `U0_dms_{baseline}_s<k>_kcv2_clip8_ln` (baseline channels in place of the 11 driver-kinematics channels; survivor U0 recipe) — {stamp}", "", | |
| f"Evaluator: `eval/paired_compare_joint.py` (unchanged; `--kin-targets causal_routes --kin-version v2`, n_boot 1000, identical rows = rows both layouts scored), " | |
| f"verdict = `kill_checks.beats` (≥ 2/3 seeds CI excl. 0, same sign, |mean Δ| > seed sd). Eligibility (plan §2): " | |
| + ("face-only baseline -> scored on head_turn_onset only; hand_leave / hand_return rows are n/a (numbers kept for the record, not a claim)." if baseline in FACE_ONLY | |
| else "all three driver events eligible.") + f" Every number is read from `{csv_path.relative_to(CWM3)}` (tags `dmsU_s<k>_{baseline}_vs_u0base`, `dmsU_s<k>_{baseline}_vs_k0`).", ""] | |
| for suffix, bname in (("vs_u0base", "U0_kin_can_noK_s<k>_kcv2_clip8_ln (the existing U0 base)"), ("vs_k0", "k0 heuristic (label-free current-state reference, as claim B1)")): | |
| tags = [f"dmsU_s{s}_{baseline}_{suffix}" for s in seeds] | |
| df = _latest_rows(tags) if csv_path.exists() else pd.DataFrame() | |
| lines += [f"### A = U0_dms_{baseline}, B = {bname}", ""] | |
| if not len(df): | |
| lines += [f"no rows yet (tags `{tags}`)", ""] | |
| continue | |
| lines += ["| regime | cell | metric | " + " | ".join(f"A s{s}" for s in seeds) + " | " + " | ".join(f"B s{s}" for s in seeds) + " | " + " | ".join(f"Δ s{s} [CI]" for s in seeds) + " | mean Δ ± sd (seeds, n rows, pos) | verdict |", | |
| "|---|---|---|" + "---|" * (3 * len(seeds)) + "---|---|"] | |
| for regime in REGIMES: | |
| for cell, metric in CELLS: | |
| g = df[(df.regime == regime) & (df.cell == cell) & (df.metric == metric)].sort_values("seed") | |
| if not len(g): | |
| continue | |
| ev = cell.split("|")[1] | |
| lower = cell.startswith("hazard_nll") | |
| pr = pd.DataFrame({"seed": g.seed, "delta": g.delta.astype(float) * (-1 if lower else 1), "delta_ci_lo": (g.delta_ci_lo if not lower else -g.delta_ci_hi).astype(float), | |
| "delta_ci_hi": (g.delta_ci_hi if not lower else -g.delta_ci_lo).astype(float), "status": "secondary"}) | |
| res = KC.beats(pr, pd.DataFrame({"value": g.value_a.astype(float)}), pd.DataFrame({"value": g.value_b.astype(float)}), min_seeds=3) | |
| by = {int(s): r for s, r in zip(g.seed, g.itertuples())} | |
| A = [f"{by[s].value_a:.4f}" if s in by else "—" for s in seeds] | |
| B = [f"{by[s].value_b:.4f}" if s in by else "—" for s in seeds] | |
| D = [f"{by[s].delta:+.4f} [{by[s].delta_ci_lo:+.4f},{by[s].delta_ci_hi:+.4f}]" if s in by else "—" for s in seeds] | |
| d = g.delta.astype(float).to_numpy() | |
| verdict = res["verdict"] if ev in elig else f"n/a (ineligible: face-only baseline; descriptive {res['verdict']})" | |
| n_rows = int(float(g.n.iloc[0])) if "n" in g.columns and np.isfinite(float(g.n.iloc[0])) else -1 | |
| n_posv = int(float(g.n_pos.iloc[0])) if np.isfinite(float(g.n_pos.iloc[0])) else -1 | |
| lines.append(f"| {regime} | {cell.replace('|', '/')} | {metric}{' (lower better)' if lower else ''} | " + " | ".join(A) + " | " + " | ".join(B) + " | " + " | ".join(D) | |
| + f" | {d.mean():+.4f} ± {d.std(ddof=1) if len(d) > 1 else 0.0:.4f} (seeds={len(d)}, n={n_rows:,} rows, pos={n_posv:,}) | {verdict} |") | |
| lines.append("") | |
| return "\n".join(lines) + "\n" | |
| def append_section(baseline, seeds=(0, 1, 2), table=TABLE): | |
| sec = render_section(baseline, seeds) | |
| if not table.exists(): | |
| table.write_text("# DMS baselines as the driver-kinematics input of the U0 base — driver onset PR-AUC@3s and hazard NLL vs the existing U0 base and the k0 heuristic " | |
| "(baselines_dms/fit_u_base.py; numbers from results/paired_compare.csv written by the unchanged eval/paired_compare_joint.py)\n") | |
| with open(table, "a", encoding="utf-8") as fh: | |
| fh.write("\n" + sec) | |
| return sec | |
| # ----------------------------------------------------------------------------- coverage of the substitute packs (results/tables/dms_coverage.md, append-only) | |
| def pack_coverage_rows(baseline, sub_dir=None, regimes=REGIMES, splits=SPLITS): | |
| """per (regime, split) of kin_subst/<b>: rows, rows with a substitute history (row_ok), share, valid history steps given row_ok, exposure hours (rows x 0.2 s), history source.""" | |
| sub_dir = Path(sub_dir) if sub_dir else subst_dir(baseline) | |
| rows = [] | |
| for r in regimes: | |
| for s in splits: | |
| p = sub_dir / f"kinhist_causal_{r}_{s}.npz" | |
| if not p.exists(): | |
| continue | |
| with np.load(p, allow_pickle=True) as z: | |
| ok, V = z["row_ok"].astype(bool), z["Khvalid"] | |
| N = len(ok) | |
| names = [str(c) for c in z["channels"]] | |
| pj = p.parent / (p.stem + ".provenance.json") | |
| prov = json.loads(pj.read_text()) if pj.exists() else {} | |
| rows.append({"baseline": baseline, "regime": r, "split": s, "rows": int(N), "rows_ok": int(ok.sum()), "share": float(ok.mean()) if N else 0.0, | |
| "khvalid_given_ok": float(V[ok].mean()) if ok.any() else float("nan"), "hours": N * 0.2 / 3600, "hours_ok": int(ok.sum()) * 0.2 / 3600, | |
| "history_source": prov.get("history_source", ""), "anchor_source": prov.get("anchor_source", ""), "channels": names, "pack_sha256": prov.get("pack_sha256", "")[:16]}) | |
| return rows | |
| def append_pack_coverage(baseline, table=COVERAGE_TABLE, sub_dir=None, note=""): | |
| """append ONE section for the layer-1 substitute pack of `baseline` to results/tables/dms_coverage.md (existing sections untouched) -> (section text, rows).""" | |
| rows = pack_coverage_rows(baseline, sub_dir) | |
| stamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) | |
| lines = [f"## coverage — kin_subst/{baseline} (layer-1 substitute driver-kinematics pack for `U0_dms_{baseline}_s<k>_kcv2_clip8_ln`) — {stamp}", "", | |
| "Missing policy (decisions.md 2026-09-14): a row without a substitute history (row_ok False: no cache / no frame at t / no valid step) is NaN on every channel and is neither trained on nor " | |
| "scored (the U0 trainer drops kin_row_ok False rows; paired_compare_joint scores the rows both layouts predicted, n printed in results/tables/dms_driver_onsets.md). " | |
| f"Source: results/baselines_dms/kin_subst/{baseline}/kinhist_causal_*.npz + provenance sidecars (baselines_dms/fit_u_base.py). Eligibility (plan §2): " | |
| + ("face-only baseline -> head_turn_onset only." if baseline in FACE_ONLY else "all three driver events.") + (f" {note}" if note else ""), "", | |
| "| baseline | regime | split | rows | rows with history | share | valid steps given history | hours | hours with history | history source | anchor source | channels |", "|---|---|---|---|---|---|---|---|---|---|---|---|"] | |
| for r in rows: | |
| lines.append(f"| {r['baseline']} | {r['regime']} | {r['split']} | {r['rows']:,} | {r['rows_ok']:,} | {r['share']:.3f} | {r['khvalid_given_ok']:.3f} | {r['hours']:.2f} | {r['hours_ok']:.2f} | " | |
| f"{r['history_source']} | {r['anchor_source']} | `{','.join(r['channels'])}` |") | |
| if not rows: | |
| lines.append(f"no packs under {Path(sub_dir) if sub_dir else subst_dir(baseline)}") | |
| sec = "\n".join(lines) + "\n" | |
| table = Path(table) | |
| table.parent.mkdir(parents=True, exist_ok=True) | |
| with open(table, "a", encoding="utf-8") as fh: | |
| fh.write(("\n" if table.exists() and table.stat().st_size else "") + sec) | |
| return sec, rows | |
| # ----------------------------------------------------------------------------- synthetic summaries (tests / dry-run) | |
| def synthetic_summary(baseline, regime, split, out_path, n_routes=2, channels=("yaw_deg", "pitch_deg", "closed_p", "conf"), seed=0, kind="S0", invalid_frac=0.05, spike=None, s0_channels=None, offset=0.0): | |
| """a summ_{regime}_{split}.npz with the s12 uid / t of the pack, S0 (or emb_anchor) on `n_routes` routes, plausible marginals; spike = (row, value) plants an outlier on one row. | |
| kind 'emb' + s0_channels: ALSO writes a non-empty S0 block of geometry-like channels (the real videomae2 / ov_action summaries carry both); offset shifts every value (tests).""" | |
| with np.load(S12R / f"targets_{regime}_{split}.npz", allow_pickle=True) as z: | |
| uid, t = z["uid"], z["t"] | |
| N, C = len(uid), len(channels) | |
| rng = np.random.default_rng(seed) | |
| routes = [str(u) for u in np.unique(uid)[:n_routes]] | |
| dec = np.isin(uid, routes) | |
| S = np.full((N, C), np.nan, np.float32) | |
| for j, c in enumerate(channels): | |
| S[dec, j] = (np.cumsum(rng.normal(0, 2.0, dec.sum())) * 0.3 + rng.normal(0, 3.0, dec.sum())) if c.endswith("_deg") else np.clip(rng.beta(1.2, 6.0, dec.sum()), 0, 1) if c.endswith("_p") else np.clip(rng.normal(0.9, 0.05, dec.sum()), 0, 1) | |
| valid = dec & (rng.random(N) >= invalid_frac) | |
| S[~valid] = np.nan | |
| S = (S + np.float32(offset)).astype(np.float32) | |
| if spike is not None: | |
| row, val = spike | |
| S[row] = val | |
| valid[row] = True | |
| row_ok = dec.copy() | |
| x_names = [f"{c}_{q}" for c in channels for q in ("mean_8s", "std_8s", "share_off_3s")] | |
| X = np.where(row_ok[:, None], rng.random((N, len(x_names))).astype(np.float32), np.nan) | |
| vs8 = np.where(row_ok, np.clip(rng.normal(0.9, 0.1, N), 0, 1), np.nan).astype(np.float32) | |
| rule0 = np.where(row_ok, rng.random(N), np.nan).astype(np.float32) | |
| out = Path(out_path) | |
| out.parent.mkdir(parents=True, exist_ok=True) | |
| common = dict(uid=uid, t=t, row_ok=row_ok, route_decoded=dec, X=X, x_names=np.array(x_names), valid_share_8s=vs8, n_valid_8s=np.round(np.nan_to_num(vs8) * 80).astype(np.int32), rule_names=np.array(["rule0"]), rule0=rule0, | |
| thresholds=np.array(json.dumps({"synthetic": True})), source_cache_sha256=np.array("{}"), hz=np.int64(HZ), hist_s=np.float32(8.0), baseline=np.array(baseline), spec=np.array("{}"), | |
| regime=np.array(regime), split=np.array(split), written_utc=np.array(time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()))) | |
| if kind == "S0": | |
| np.savez_compressed(out, S0=S, S0valid=valid, channels=np.array(list(channels)), **common) | |
| else: | |
| if s0_channels: | |
| G = np.where(valid[:, None], np.abs(rng.normal(300.0, 20.0, (N, len(s0_channels)))), np.nan).astype(np.float32) | |
| np.savez_compressed(out, S0=G, S0valid=valid, channels=np.array(list(s0_channels)), emb_anchor=S.astype(np.float16), emb_mean_8s=S.astype(np.float16), | |
| n_win_8s=np.where(valid, 16, 0).astype(np.int32), **common) | |
| else: | |
| np.savez_compressed(out, S0=np.zeros((N, 0), np.float32), S0valid=row_ok, channels=np.array([]), emb_anchor=S.astype(np.float16), emb_mean_8s=S.astype(np.float16), | |
| n_win_8s=np.where(valid, 16, 0).astype(np.int32), **common) | |
| return out, routes | |
| def dry_run(keep_dir=None, train=True, log=print): | |
| """synthetic summaries (manual train / val, 2 routes, C = 4 -> pad path) -> packs -> verify_pack vs the real pack -> train_cwm3.main() on CPU for ONE optimisation | |
| step (--cpu --no-amp --epochs 1 --max-anchors 4096 --regimes manual --pred-splits val) with QUICK_CSV redirected -> asserts kmu / ksd were recomputed from the pack.""" | |
| import tempfile | |
| root = Path(keep_dir) if keep_dir else Path(tempfile.mkdtemp(prefix="dms_u0_dryrun_")) | |
| b = "synthetic" | |
| summ_root = root / "summ" / b | |
| rep = {"root": str(root), "baseline": b} | |
| for split in ("train", "val"): | |
| synthetic_summary(b, "manual", split, summ_root / f"summ_manual_{split}.npz", n_routes=2, seed={"train": 0, "val": 1}[split]) | |
| sub = root / "kin_subst" / b | |
| paths = build_packs(b, "summ", sub, regimes=("manual",), splits=("train", "val"), summ_root=summ_root, log=log) | |
| rep["packs"] = [str(p) for p in paths] | |
| with np.load(paths[0], allow_pickle=True) as z: | |
| rep["channels"] = [str(c) for c in z["channels"]] | |
| rep["row_ok"] = int(z["row_ok"].sum()) | |
| rep["khvalid_share_given_ok"] = float(z["Khvalid"][z["row_ok"]].mean()) | |
| if not train: | |
| print(json.dumps(rep, indent=1)) | |
| return rep | |
| out = root / "runs" / U0_DMS.format(b=b, seed=0) | |
| extra = ["--cpu", "--no-amp", "--epochs", "1", "--max-anchors", "4096", "--regimes", "manual", "--pred-splits", "val", "--minutes", "8"] | |
| t0 = time.time() | |
| train_u0(b, 0, out, sub, extra_argv=extra, quick_csv=root / "quick_metrics_dryrun.csv", log=log) | |
| st = dict(np.load(out / "stats.npz")) | |
| names = rep["channels"] | |
| kmu, ksd = st["kmu"], st["ksd"] | |
| j_flag = names.index("valid_flag") | |
| pads = [j for j, n in enumerate(names) if n.startswith(DA.PAD_PREFIX)] | |
| assert abs(float(kmu[j_flag]) - 1.0) < 1e-5 and all(abs(float(kmu[j])) < 1e-6 for j in pads), (kmu, names) # stats came from the SUBSTITUTE pack | |
| assert np.isfinite(kmu).all() and np.isfinite(ksd).all() and (ksd > 0).all() | |
| for f in ("TRAIN_DONE", "PRED_DONE", "BEST", "stats.npz", "pred_manual_val.npz", "pred_layout_manual_val.npz", "u0_dms_provenance.json"): | |
| assert (out / f).exists(), f | |
| hist = [json.loads(l) for l in (out / "history.jsonl").read_text().splitlines()] | |
| prov = json.loads((out / "u0_dms_provenance.json").read_text()) | |
| rep.update(run_dir=str(out), epochs=len(hist), train_loss=hist[-1]["train_loss"], val_objective=hist[-1]["val_objective"], kmu=[round(float(x), 4) for x in kmu], ksd=[round(float(x), 4) for x in ksd], | |
| kin_dir_recorded=prov["kin_dir_recorded"], argv=prov["argv"], wall_s=round(time.time() - t0, 1), pair_argv_example=[PY, *pair_argv("l2cs_gaze", 0, "manual")], | |
| pair_argv_k0_example=[PY, *pair_argv("l2cs_gaze", 0, "manual", k0=True)]) | |
| print(json.dumps(rep, indent=1)) | |
| return rep | |
| def main(argv=None): | |
| ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) | |
| ap.add_argument("--baseline", default="", choices=["", *BASELINES]) | |
| ap.add_argument("--build-packs", action="store_true") | |
| ap.add_argument("--source", default="auto", choices=["auto", "summ", "cache"]) | |
| ap.add_argument("--anchor-source", default="auto", choices=list(ANCHOR_SOURCES), help="summary anchor block for the substitute channels: auto = emb (emb_anchor) for videomae2 / ov_action, S0 otherwise") | |
| ap.add_argument("--summ-root", default="", help="directory holding summ_{regime}_{split}.npz (default results/baselines_dms/summ/<b>; the layer-1 launcher passes results/baselines_dms/summ_full/<b>)") | |
| ap.add_argument("--coverage", action="store_true", help="CPU: append the kin_subst pack coverage section to results/tables/dms_coverage.md") | |
| ap.add_argument("--note", default="", help="free-text launch note (inert; the launcher passes gate_token=train_cwm3 so that every lane's trainer gate `pgrep -f 'python.*train_cwm3'` counts this process)") | |
| ap.add_argument("--train", action="store_true", help="GPU: train U0_dms_<b>_s<seed> (launch through run_queue_dms.sh only)") | |
| ap.add_argument("--seed", type=int, default=0) | |
| ap.add_argument("--pair", action="store_true", help="CPU: eval/paired_compare_joint.py vs the U0 base and vs k0 (both regimes)") | |
| ap.add_argument("--predict-label-rows", action="store_true", help="CPU: U0_dms_<b>_s<seed> outputs on the 979 warning-head label rows (fit_readout --with-u0)") | |
| ap.add_argument("--seeds", nargs="+", type=int, default=[0, 1, 2]) | |
| ap.add_argument("--no-k0", action="store_true") | |
| ap.add_argument("--table", action="store_true", help="append the section to results/tables/dms_driver_onsets.md") | |
| ap.add_argument("--print-argv", action="store_true") | |
| ap.add_argument("--dry-run", action="store_true") | |
| ap.add_argument("--dry-run-no-train", action="store_true") | |
| ap.add_argument("--keep-dir", default="") | |
| a = ap.parse_args(argv) | |
| if a.dry_run or a.dry_run_no_train: | |
| dry_run(a.keep_dir or None, train=not a.dry_run_no_train) | |
| return 0 | |
| assert a.baseline, "--baseline required" | |
| if a.build_packs: | |
| build_packs(a.baseline, a.source, summ_root=(a.summ_root or None), anchor_source=a.anchor_source) | |
| if a.train: | |
| train_u0(a.baseline, a.seed) | |
| ensure_layouts(RUNS / U0_DMS.format(b=a.baseline, seed=a.seed)) | |
| if a.predict_label_rows: | |
| predict_label_rows(a.baseline, a.seed) | |
| if a.pair or a.print_argv: | |
| run_pairs(a.baseline, tuple(a.seeds), k0=not a.no_k0, dry=a.print_argv and not a.pair) | |
| if a.table: | |
| print(append_section(a.baseline, tuple(a.seeds))) | |
| if a.coverage: | |
| print(append_pack_coverage(a.baseline)[0]) | |
| if a.print_argv: | |
| print("train argv:", " ".join([PY, str(TRAINER), *trainer_argv(a.seed, RUNS / U0_DMS.format(b=a.baseline, seed=a.seed))])) | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 56.6 kB
- Xet hash:
- 1d461e22415469479efbaecf3c507665787dac50b56e47cad68185b5dfdac79f
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.