Buckets:
| """cwm3/tests/test_holdout_kpf_overrides.py — the additive hold-out overrides of candidate (iii) (pre-registration docs/decisions.md 2026-09-18T03:41:14Z): | |
| (1) data/build_kp_residual_store.py: the module defaults (benchmark pose dirs, kin-v2 dir, output, splits train / val / test) are unchanged; `holdout_override()` | |
| switches ONLY the pose / kin / output dirs and the split list to the hold-out artefacts and `restore()` puts the defaults back. | |
| (2) eval/predict_holdout.py: `kpf_dir_for` returns None for a config without `u_kpf` (stream unused — older runs untouched), cfg['u_kpf'] without the flag | |
| (default unchanged) and the override with it; the REDIRECT map keeps its nine entries; `check_inputs` lists the kpf file only for a kpf config and reports | |
| an unset KPF_DIR instead of raising. | |
| (3) eval/predict_kpf_holdout.py: `build_frozen` rebuilds a forecaster from a synthetic frozen run dir with the recorded scale_fallback and refuses a | |
| checkpoint / config scale_fallback mismatch (the "never refit" guard). | |
| Synthetic data only. Run: cd cwm3 && python -m pytest tests/test_holdout_kpf_overrides.py -q | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import sys | |
| from pathlib import Path | |
| os.environ.setdefault("OMP_NUM_THREADS", "4") | |
| import pytest # noqa: E402 | |
| import torch # noqa: E402 | |
| torch.set_num_threads(4) | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| for p in (CWM3, CWM3 / "data", CWM3 / "train", CWM3 / "eval", CWM3 / "common"): | |
| if str(p) not in sys.path: | |
| sys.path.insert(0, str(p)) | |
| import build_kp_residual_store as bks # noqa: E402 | |
| def test_store_builder_defaults_and_override(): | |
| assert bks.POSE_V2 == CWM3 / "results" / "pose_causal_routes_v2" | |
| assert bks.POSE_V1 == CWM3 / "results" / "pose_causal_routes" | |
| assert bks.KIN_V2 == CWM3 / "results" / "targets_kin_causal_routes_v2" | |
| assert bks.OUT == CWM3 / "results" / "kp_residual" | |
| assert bks.SPLITS == ("train", "val", "test") and bks.REGIMES == ("manual", "engaged") | |
| prev = bks.holdout_override(CWM3 / "results" / "holdout") | |
| try: | |
| assert bks.POSE_V2 == CWM3 / "results" / "holdout" / "pose_causal_routes" | |
| assert bks.POSE_V1 == CWM3 / "results" / "pose_causal_routes" # fallback = the kin-v2 hold-out build's pose_fallback_dir | |
| assert bks.KIN_V2 == CWM3 / "results" / "holdout" / "targets_kin_causal_routes_v2" | |
| assert bks.OUT == CWM3 / "results" / "holdout" / "kp_residual" | |
| assert bks.SPLITS == ("holdout",) | |
| assert bks.S12R == CWM3.parent / "s12_wm" / "results" and bks.JOINTS[0] == "nose" and bks.NJ == 11 and bks.STATE_STEPS == (4, 9, 14) # untouched | |
| finally: | |
| bks.restore(prev) | |
| assert bks.OUT == CWM3 / "results" / "kp_residual" and bks.SPLITS == ("train", "val", "test") and bks.POSE_V2 == CWM3 / "results" / "pose_causal_routes_v2" | |
| def test_predict_holdout_kpf_dir_default_unchanged(): | |
| import predict_holdout as PH | |
| assert set(PH.REDIRECT) == {"PACK", "PACK_V2", "KINC_V2", "PRESS", "DEM", "BANKS", "FACE", "GAZE", "ROAD_DYN"} | |
| assert PH.kpf_dir_for({"u_kpf": ""}) is None and PH.kpf_dir_for({}) is None and PH.kpf_dir_for({}, "results/runs/X_holdout") is None | |
| assert PH.kpf_dir_for({"u_kpf": "/x/KPF_mb_full_s0"}) == Path("/x/KPF_mb_full_s0") | |
| assert PH.kpf_dir_for({"u_kpf": "/x/KPF_mb_full_s0"}, "results/runs/KPF_mb_full_s0_holdout") == Path("results/runs/KPF_mb_full_s0_holdout") | |
| import cwm3_dataset as ds | |
| base = {"base": {"V": ["V0", "can4s_stats2"], "U": ["U0", "x"], "E": ["E0", "x"]}, "e_side": "vf256"} | |
| prev = ds.KPF_DIR | |
| try: | |
| ds.set_kpf_dir(None) | |
| miss0 = PH.check_inputs(dict(base), ["manual"]) | |
| assert not any("kpf" in m for m in miss0), miss0 # a run without u_kpf never asks for the kpf channel | |
| miss1 = PH.check_inputs(dict(base, u_kpf="/x/KPF_mb_full_s0"), ["manual"]) | |
| assert any(m.startswith("kpf_manual_holdout.npz: cwm3_dataset.KPF_DIR unset") for m in miss1), miss1 | |
| ds.set_kpf_dir(CWM3 / "results" / "runs" / "KPF_mb_full_s0") # the training dir has no hold-out file -> listed as missing by path | |
| miss2 = PH.check_inputs(dict(base, u_kpf="/x/KPF_mb_full_s0"), ["manual"]) | |
| assert any(m.endswith("KPF_mb_full_s0/kpf_manual_holdout.npz") for m in miss2), miss2 | |
| finally: | |
| ds.set_kpf_dir(prev) | |
| def test_build_frozen_uses_recorded_scale_fallback(tmp_path): | |
| import predict_kpf_holdout as PK | |
| import train_kp_residual as TK | |
| from models import kp_forecaster as kf | |
| torch.manual_seed(0) | |
| arm = "mb_full" | |
| model = kf.KpForecaster(**TK.ARMS[arm], scale_fallback=0.1234) | |
| cfg = {"arm": arm, **TK.ARMS[arm], "seed": 0, "amp": False, "scale_fallback": 0.1234, "params": {"total": kf.n_params(model), "backbone": kf.n_params(model.backbone), "heads": 0}} | |
| src = tmp_path / "KPF_mb_full_s0" | |
| src.mkdir() | |
| (src / "config.json").write_text(json.dumps(cfg)) | |
| (src / "BEST").write_text("ckpt_best.pt") | |
| torch.save({"model": model.state_dict(), "cfg": cfg, "epoch": 3}, src / "ckpt_best.pt") | |
| m2, cfg2, ckpt, ep = PK.build_frozen(src, torch.device("cpu")) | |
| assert ckpt == src / "ckpt_best.pt" and ep == 3 and abs(m2.scale_fallback - 0.1234) < 1e-12 and not m2.training | |
| for (k1, v1), (k2, v2) in zip(model.state_dict().items(), m2.state_dict().items()): | |
| assert k1 == k2 and torch.equal(v1, v2) | |
| # the guard: a config whose scale_fallback differs from the checkpoint's is refused (nothing may be refit on the hold-out) | |
| (src / "config.json").write_text(json.dumps(dict(cfg, scale_fallback=0.2))) | |
| with pytest.raises(AssertionError): | |
| PK.build_frozen(src, torch.device("cpu")) | |
Xet Storage Details
- Size:
- 5.85 kB
- Xet hash:
- 31a5c9621383051aabbfe62f51a155219499ce875dd6920b6a95d12b9a8fa07a
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.