Buckets:
| """cwm3/eval/predict_holdout.py — predictions of a FROZEN joint run on the BATON-holdout split (WP5; pre-registration 2026-09-16T06:46:35Z: | |
| evaluated ONCE, no model or threshold changes after seeing it). Mechanism = eval/robustness_perturb.py without a perturbation: | |
| * the frozen run dir results/runs/<run> is READ-ONLY; results/runs/<run>_holdout/ receives copies of config.json / BEST / TRAIN_DONE / | |
| base_ckpts.json / stats.npz / history.jsonl / params.json / label_rows.npz and a symlink to the BEST checkpoint (never copied); | |
| * the data layer is pointed at the holdout artefacts of results/holdout/ (data/holdout_pipeline.py, data/build_pack_v2_holdout.py): | |
| cwm3_dataset.PACK -> results/holdout/pack (xpre_{r}_holdout) | |
| cwm3_dataset.PACK_V2 -> results/holdout/pack_v2 (vf256 / vc256 / future_{r}_holdout; frozen PCA) | |
| cwm3_dataset.KINC_V2 -> results/holdout/targets_kin_causal_routes_v2 | |
| cwm3_dataset.PRESS -> results/holdout/targets_pressure ; cwm3_dataset.DEM -> results/holdout/targets_demand | |
| cwm3_dataset.BANKS -> results/holdout/banks (front_tok.npz + anchor_index_{r}_holdout.npy) | |
| cwm3_dataset.FACE / GAZE -> results/holdout/targets_{face,gaze}_causal_routes_v2 (only read by u_face / u_gaze_chain bases) | |
| train_cwm3_joint.PRESSURE / RESPONSE and write_pred_layout_joint.KINC['v2'] / DEM likewise; | |
| s12_wm/results/targets_{r}_holdout.npz and target_stats.npz (frozen) stay where the pipeline reads them; | |
| * prediction re-uses train_cwm3_joint.load_raw / predict_split and write_pred_layout_joint.write_layout unchanged (imported); | |
| pred_{regime}_holdout.npz + pred_layout_{regime}_holdout.npz + PRED_DONE + run json are written into the new dir. | |
| Then the paired comparison (eval/paired_compare_joint.py, A = main, B = indep, split holdout) and the warning-timing protocol run on | |
| those layouts exactly as on test. This script is NOT run by the data lane; the coordinator runs it once per frozen run. | |
| python eval/predict_holdout.py --run I1_all_minus_road2drv_s0_final [--regimes manual,engaged] [--check-only] | |
| 2026-09-18 (candidate (iii), pre-registration docs/decisions.md 2026-09-18T03:41:14Z): runs whose config carries `u_kpf` (LANE F1 kpf joint models) need the | |
| forecaster channel kpf_{r}_holdout.npz; `--u-kpf-dir <dir>` points cwm3_dataset.set_kpf_dir at the hold-out prediction dir of the FROZEN forecaster | |
| (eval/predict_kpf_holdout.py -> results/runs/KPF_<arm>_s<k>_holdout/); without the flag the dir is cfg["u_kpf"] as in training (default unchanged; runs | |
| without `u_kpf` are untouched — tests/test_holdout_kpf_overrides.py). The kpf file is part of --check-only. | |
| """ | |
| import argparse, json, os, shutil, sys, time | |
| from pathlib import Path | |
| from types import SimpleNamespace | |
| import numpy as np | |
| CWM3 = Path(__file__).resolve().parents[1]; EXPLORE = CWM3.parent; H = CWM3 / "results" / "holdout" | |
| sys.path.insert(0, str(CWM3 / "common")); sys.path.insert(0, str(CWM3 / "data")); sys.path.insert(0, str(CWM3 / "train")); sys.path.insert(0, str(CWM3 / "eval")) | |
| from runmeta import write_run_json # noqa: E402 | |
| import cwm3_dataset as ds # noqa: E402 | |
| import train_cwm3_joint as TJ # noqa: E402 | |
| import train_cwm3 as T # noqa: E402 | |
| import write_pred_layout_joint as wplj # noqa: E402 | |
| SPLIT = "holdout" | |
| COPY = ("config.json", "BEST", "TRAIN_DONE", "base_ckpts.json", "stats.npz", "history.jsonl", "params.json", "label_rows.npz") | |
| REDIRECT = {"PACK": H / "pack", "PACK_V2": H / "pack_v2", "KINC_V2": H / "targets_kin_causal_routes_v2", "PRESS": H / "targets_pressure", "DEM": H / "targets_demand", | |
| "BANKS": H / "banks", "FACE": H / "targets_face_causal_routes_v2", "GAZE": H / "targets_gaze_causal_routes_v2", | |
| "ROAD_DYN": H / "targets_road_dyn"} # 2026-09-17 confirmation batch: F5b side streams vfcan256 / can256 of the ECAN runs (cwm3_dataset.ROAD_DYN) redirected too | |
| def kpf_dir_for(cfg, override=""): | |
| """the forecaster dir of the `kpf` stream: None when the run has no u_kpf (stream unused), else --u-kpf-dir if given, else cfg['u_kpf'] (the training dir).""" | |
| if not cfg.get("u_kpf"): | |
| return None | |
| return Path(override) if override else Path(cfg["u_kpf"]) | |
| def redirect(): | |
| for k, v in REDIRECT.items(): setattr(ds, k, v) | |
| if SPLIT not in ds.SPLITS: ds.SPLITS = tuple(ds.SPLITS) + (SPLIT,) # 2026-09-17: load_split asserts split in SPLITS = (train, val, test); the holdout split is added here only (additive, this process) | |
| TJ.PRESSURE, TJ.RESPONSE = H / "targets_pressure", H / "targets_response" | |
| wplj.KINC = dict(wplj.KINC, v2=H / "targets_kin_causal_routes_v2"); wplj.DEM = H / "targets_demand" | |
| def check_inputs(cfg, regimes): | |
| need = TJ.need_streams(cfg); missing = [] | |
| for r in regimes: | |
| f = ds.files(r, SPLIT) | |
| keys = ["targets", "xpre" if "X8" in need else None, "kin", "kinhist", "demand", "future", "vf256", "vc256" if "vc256" in need else None, "anchor_index", "bank", "pressure", "face" if "face" in need else None, "gaze" if "gaze" in need else None, | |
| "vfcan256" if "vfcan256" in need else None, "can256" if "can256" in need else None, "vfcan256s" if "vfcan256s" in need else None, | |
| "kpf" if "kpf" in need else None] # pressure / response: attach_aux_targets (P / R heads); 2026-09-17: E side streams (ECAN runs) checked too; 2026-09-18: kpf channel (F1 runs) | |
| for k in keys: | |
| if k == "kpf" and f[k] is None: missing.append(f"kpf_{r}_{SPLIT}.npz: cwm3_dataset.KPF_DIR unset (cfg.u_kpf / --u-kpf-dir)"); continue | |
| if k and not Path(f[k]).exists(): missing.append(str(f[k])) | |
| for p in (TJ.PRESSURE / f"pressure_{r}_{SPLIT}.npz", TJ.RESPONSE / f"response_{r}_{SPLIT}.npz"): | |
| if not p.exists(): missing.append(str(p)) | |
| return sorted(set(missing)) | |
| def main(): | |
| ap = argparse.ArgumentParser(); ap.add_argument("--run", required=True, help="frozen joint run dir name under results/runs (read-only)"); ap.add_argument("--regimes", default="manual,engaged") | |
| ap.add_argument("--cpu", action="store_true"); ap.add_argument("--check-only", action="store_true", help="list missing holdout artefacts and exit") | |
| ap.add_argument("--u-kpf-dir", default="", help="2026-09-18 candidate (iii): dir holding kpf_{r}_holdout.npz of the FROZEN forecaster (results/runs/KPF_<arm>_s<k>_holdout); default = cfg['u_kpf']; ignored for runs without u_kpf") | |
| a = ap.parse_args(); t0 = time.time() | |
| src = CWM3 / "results" / "runs" / a.run; assert (src / "PRED_DONE").exists() and (src / "TRAIN_DONE").exists(), f"{src}: not a finished run" | |
| cfgj = json.loads((src / "config.json").read_text()); cfg, cargs = cfgj["cfg"], cfgj["args"]; seed = int(cargs["seed"]) | |
| redirect(); ds.set_pack_version(cfg.get("pack_version", cargs.get("pack_version", "v1"))); ds.set_kin_source(cfg.get("kin_source", "cwm2"), cfg.get("kin_version", "v1")) | |
| kpf_dir = kpf_dir_for(cfg, a.u_kpf_dir) | |
| if kpf_dir is not None: ds.set_kpf_dir(kpf_dir) # 2026-09-18: F1 kpf runs — the frozen forecaster's hold-out channel (see module doc) | |
| regimes = a.regimes.split(","); miss = check_inputs(cfg, regimes) | |
| print(json.dumps(dict(run=a.run, pack_version=ds.pack_version(), kin_source=ds.kin_source(), kin_version=ds.kin_version(), need=sorted(TJ.need_streams(cfg)), kpf_dir=(str(kpf_dir) if kpf_dir else None), missing_inputs=miss), indent=1), flush=True) | |
| if a.check_only or miss: | |
| sys.exit(0 if not miss else 6) | |
| out = CWM3 / "results" / "runs" / f"{a.run}_{SPLIT}"; out.mkdir(parents=True, exist_ok=True) | |
| for f in COPY: | |
| if (src / f).exists() and not (out / f).exists(): shutil.copy2(src / f, out / f) | |
| best = (src / "BEST").read_text().strip() | |
| if best != "base" and not (out / best).exists(): os.symlink(src / best, out / best) | |
| args = SimpleNamespace(out=str(out), seed=seed, cpu=a.cpu, no_amp=False, minutes=600.0, synthetic=0, max_anchors=0, no_label_rows=True, no_lock=True) | |
| TJ.log(f"=== predict_holdout {a.run} split={SPLIT} seed={seed} regimes={regimes} out={out} redirect={ {k: str(v) for k, v in REDIRECT.items()} } kpf_dir={kpf_dir} ===", args.out) | |
| info = TJ.base_info_numpy(cfg, seed); rec_bc = json.loads((out / "base_ckpts.json").read_text())["base_ckpts"] | |
| assert info["base_ckpts"] == rec_bc, f"base checkpoints changed since the run: {info['base_ckpts']} vs {rec_bc}" | |
| st = {k: v for k, v in np.load(out / "stats.npz").items()}; bank = TJ.load_bank(args, cfg, info) | |
| for regime in regimes: | |
| pf = out / f"pred_{regime}_{SPLIT}.npz" | |
| if pf.exists() and not T.pred_ok(pf): TJ.log(f"{pf.name} was corrupt -> re-predicting", args.out) | |
| if pf.exists(): TJ.log(f"{pf.name} exists — skipped", args.out) | |
| else: | |
| raw = TJ.load_raw(args, cfg, SPLIT, subset=False, regimes=[regime]); code = TJ.predict_split(args, cfg, raw, st, bank, info, regime, SPLIT); assert code == 0, f"predict_split exit {code}"; del raw | |
| lay = out / f"pred_layout_{regime}_{SPLIT}.npz" | |
| if not lay.exists(): | |
| code_files = sorted((CWM3 / "models").glob("*.py")) + [CWM3 / "train" / "train_cwm3_joint.py", CWM3 / "train" / "train_cwm3.py", CWM3 / "train" / "batch.py", CWM3 / "eval" / "write_pred_layout_joint.py", CWM3 / "data" / "cwm3_dataset.py", EXPLORE / "s12_wm/eval_s13.py"] | |
| wplj.write_layout(str(out), regime, SPLIT, {"stage": cfg["stage"], "variant": cfg["variant"], "cfg": cfg}, code_files, log=lambda m: TJ.log(m, args.out)) | |
| (out / "PRED_DONE").write_text(json.dumps({"splits": [SPLIT], "regimes": regimes, "source_run": a.run, "holdout_draw": str(H / "draw_v1.json")})) | |
| write_run_json(str(out), f"predict_{SPLIT}", [Path(__file__), CWM3 / "train" / "train_cwm3_joint.py", CWM3 / "eval" / "write_pred_layout_joint.py", CWM3 / "data" / "cwm3_dataset.py"], ([src / best] if best != "base" else []) + [H / "draw_v1.json"], seed=seed, t_start=t0, input_listing=False, | |
| extra={"source_run": a.run, "split": SPLIT, "base_ckpts": info["base_ckpts"], "redirect": {k: str(v) for k, v in REDIRECT.items()}, "kpf_dir": (str(kpf_dir) if kpf_dir else None), "u_kpf_train": cfg.get("u_kpf", ""), | |
| "dms_fields_used": [], "note": "frozen checkpoint, holdout artefacts of results/holdout; evaluated once (pre-registration 2026-09-16T06:46:35Z)"}) | |
| TJ.log(f"PREDICT_HOLDOUT_DONE {a.run} ({time.time() - t0:.0f}s)", args.out) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 10.6 kB
- Xet hash:
- c06f462739caaceddbb9ce0532412bc671561ca81895fc3209c3b3a325b7eaf0
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.