Buckets:
| """cwm3/tests/test_baselines_dms_layers.py — contracts of the layer 1-3 driver code of the DMS-baseline comparison (baselines_dms/{fit_u_base,chain_style,fit_readout,eval_dms}.py; 2026-09-14). | |
| (a) substitute pack : from a synthetic summary (manual/val, s12 uid / t) fit_u_base.build_packs writes kinhist_causal_manual_val.npz whose KEYS, DTYPES and SHAPES equal the real | |
| results/targets_kin_causal_routes_v2 pack (verify_pack); pad path (C = 4 -> 4 channels + valid_flag + 6 pads) and PCA path (C = 16 -> pca0..pca10, pca11.npz | |
| + sha in the sidecar); the std_kh validity rule (channels 0 / 10 finite <=> valid step); target symlinks point at the ORIGINAL causal targets. | |
| causality : a spike planted on every anchor row LATER than t (the summary path) and on the 10 Hz frame at t + 0.1 s (the cache path, dataset_adapter.route_hist) | |
| changes NO history step of the anchor at t. | |
| (b) fit_readout : the dry-run fit on synthetic X / labels (TW.make_model / GRID / cv_auroc, unchanged) writes a layout that passes layout.validate_layout and | |
| cwm2/eval/results_schema.validate; undefined rows are NaN with w_mask False; arm / score_source vocabulary. | |
| (c) chain_style : CHANNEL_ROLES covers EVERY channel declared by EVERY extractor (l2cs_gaze, mp_face, ov_gaze, openface3, ov_action, videomae2) and nothing else; the | |
| mapped yaw / pitch / closure channels exist; the hazard-chain replica has the legacy architecture (state-dict shapes == score_val_marginals.build_net for | |
| 7 input channels); compose_att == cwm2/compose.legacy_compose; channel states follow the documented thresholds (0.4 rad head yaw, 20 / 15 deg, 0.5). | |
| (d) hygiene : no forbidden DMS token in the new files (the list of tests/test_no_dms_inputs.py); queue_dms.txt lines parse and name existing scripts; run_queue_dms.sh | |
| keeps the three launch guards; eval_dms argv builders call the unchanged evaluators with --item dms_<b> and the 1 / 3 / 5 alerts/h loads. | |
| Run: cd cwm3 && python -m pytest -q tests/test_baselines_dms_layers.py tests/test_no_dms_inputs.py | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import re | |
| import sys | |
| import tempfile | |
| from pathlib import Path | |
| os.environ.setdefault("OMP_NUM_THREADS", "4") | |
| import numpy as np # noqa: E402 | |
| import pytest # noqa: E402 | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| EXPLORE = CWM3.parent | |
| CWM2 = EXPLORE / "cwm2" | |
| PKG = CWM3 / "baselines_dms" | |
| S12R = EXPLORE / "s12_wm" / "results" | |
| for p in (PKG, PKG / "extractors", CWM3 / "data", CWM3 / "train", CWM2, CWM2 / "eval", CWM2 / "baselines" / "legacy_cockpitwm"): | |
| 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 | |
| import layout as LY # noqa: E402 | |
| import fit_u_base as FU # noqa: E402 | |
| import chain_style as CS # noqa: E402 | |
| import fit_readout as FR # noqa: E402 | |
| import eval_dms as ED # noqa: E402 | |
| NEW_FILES = [PKG / f for f in ("fit_u_base.py", "chain_style.py", "fit_readout.py", "eval_dms.py", "run_queue_dms.sh", "queue_dms.txt")] | |
| FORBIDDEN = ["driver_state", "faceProb", "isDistracted", "face_yaw", "face_pitch", "face_roll", "blinkProb", "dmonitoring", "driverState", "awarenessStatus", "distractedPct", "dstate", "leftEyeProb", "rightEyeProb"] | |
| REAL = CWM3 / "results" / "targets_kin_causal_routes_v2" | |
| needs_real_pack = pytest.mark.skipif(not (REAL / "kinhist_causal_manual_val.npz").exists(), reason="real causal kin pack missing") | |
| def _same(a, b): | |
| a, b = np.asarray(a, np.float64), np.asarray(b, np.float64) | |
| return a.shape == b.shape and np.array_equal(np.isnan(a), np.isnan(b)) and np.array_equal(a[~np.isnan(a)], b[~np.isnan(b)]) | |
| # ----------------------------------------------------------------------------- (a) substitute pack | |
| def test_substitute_pack_equals_real_pack_contract(C): | |
| with tempfile.TemporaryDirectory() as d: | |
| root = Path(d) | |
| chans = ("yaw_deg", "pitch_deg", "closed_p", "conf") if C == 4 else tuple(f"e{j}_p" for j in range(C)) | |
| FU.synthetic_summary("synthetic", "manual", "train", root / "summ" / "summ_manual_train.npz", n_routes=2, channels=chans, seed=0) | |
| FU.synthetic_summary("synthetic", "manual", "val", root / "summ" / "summ_manual_val.npz", n_routes=2, channels=chans, seed=1) | |
| paths = FU.build_packs("synthetic", "summ", root / "kin_subst", regimes=("manual",), splits=("train", "val"), summ_root=root / "summ", log=lambda m: None) | |
| assert len(paths) == 2 and all(FU.verify_pack(p) for p in paths) | |
| with np.load(paths[1], allow_pickle=True) as a, np.load(REAL / "kinhist_causal_manual_val.npz", allow_pickle=True) as b: | |
| assert set(a.files) == set(b.files) == set(FU.REAL_KEYS) | |
| for k in FU.REAL_KEYS: | |
| assert a[k].shape == b[k].shape, k | |
| assert a[k].dtype == b[k].dtype or (a[k].dtype.kind == b[k].dtype.kind == "U"), (k, a[k].dtype, b[k].dtype) | |
| names = [str(c) for c in a["channels"]] | |
| Kh, V, ok = a["Kh_c"].astype(np.float32), a["Khvalid"], a["row_ok"] | |
| assert len(names) == 11 and ok.any() | |
| assert np.isfinite(Kh[V][:, [0, 10]]).all() and np.isnan(Kh[~V]).all() # std_kh validity rule | |
| # 5 Hz anchor rows -> the odd 10 Hz steps (t - 0.1, t - 0.3, ...) are never valid; step 39 (= t) is valid on defined rows | |
| assert not V[:, 38::-2].any(), "odd 10 Hz steps must be invalid on the anchor-row history path" | |
| assert V[ok, 39].mean() > 0.8 | |
| if C == 4: | |
| assert names == ["yaw_deg", "pitch_deg", "closed_p", "conf", "valid_flag"] + [f"pad_{j}" for j in range(5, 11)] | |
| assert np.all(Kh[V][:, 4] == 1.0) and np.all(Kh[V][:, 5:] == 0.0) | |
| else: | |
| assert names == [f"pca{j}" for j in range(11)] and (root / "kin_subst" / "pca11.npz").exists() | |
| red = DA.PCAReducer.load(root / "kin_subst" / "pca11.npz") | |
| assert red.channels == list(chans) and red.W.shape == (16, 11) | |
| prov = json.loads((paths[1].parent / "kinhist_causal_manual_val.provenance.json").read_text()) | |
| assert prov["history_source"].startswith("summ_S0") and prov["channels_out"] == names and "pack_sha256" in prov | |
| if C == 16: | |
| assert prov["reducer"]["kind"] == "pca" and prov["reducer_sha256"] == cs.sha256_file(root / "kin_subst" / "pca11.npz") | |
| for r in ("manual",): | |
| for s in ("train", "val"): | |
| lnk = root / "kin_subst" / f"kin_causal_{r}_{s}.npz" | |
| assert lnk.is_symlink() and lnk.resolve() == (REAL / f"kin_causal_{r}_{s}.npz").resolve() # targets = the ORIGINAL causal targets | |
| def test_history_from_anchor_rows_is_causal(): | |
| with np.load(S12R / "targets_manual_val.npz", allow_pickle=True) as z: | |
| uid, t = z["uid"], z["t"] | |
| routes = [str(u) for u in np.unique(uid)[:2]] | |
| dec = np.isin(uid, routes) | |
| rng = np.random.default_rng(0) | |
| S = np.where(dec[:, None], rng.normal(0, 1, (len(uid), 3)), np.nan).astype(np.float32) | |
| valid = dec.copy() | |
| Kh, Kv = FU.history_from_anchor_rows(uid, t, S, valid, 40) | |
| rows = np.flatnonzero(dec) | |
| i = rows[len(rows) // 2] | |
| later = dec & (uid == uid[i]) & (t > t[i] + 1e-9) | |
| assert later.any() | |
| S2 = S.copy() | |
| S2[later] = 1e3 # spike every later anchor row of the same route | |
| Kh2, Kv2 = FU.history_from_anchor_rows(uid, t, S2, valid, 40) | |
| assert _same(Kh[i], Kh2[i]) and np.array_equal(Kv[i], Kv2[i]) | |
| assert Kv[i, 39] and (not Kv[i, 38]) and Kv[i, 37] == bool(np.any((uid == uid[i]) & np.isclose(t, t[i] - 0.2))) # step 39 = t, 38 = t - 0.1 (no anchor), 37 = t - 0.2 | |
| j = rows[rows > i][0] # the next anchor row sees the spike (teeth of the test) | |
| assert not _same(Kh[j], Kh2[j]) | |
| def test_cache_history_spike_after_anchor_changes_nothing(): | |
| rc = cs.synthetic_route_cache(channels=("yaw_deg", "pitch_deg", "closed_p", "conf"), t1=80.0) | |
| anchors = np.array([20.0, 40.0, 60.0]) | |
| Gh, Vh, ok = DA.route_hist(rc, anchors, 40) | |
| k = np.round(rc.t.astype(np.float64) * 10).astype(np.int64) | |
| pos = {int(kk): i for i, kk in enumerate(k)} | |
| for i, a in enumerate(anchors): | |
| hit = pos[int(round(a * 10)) + 1] | |
| feat = rc.feat.astype(np.float32) | |
| valid, fail = rc.valid.copy(), rc.fail.copy() | |
| valid[hit], fail[hit] = True, 0 | |
| feat[hit] = 1e3 | |
| feat[~valid] = np.nan | |
| rc2 = cs.RouteCache(rc.uid, rc.t, rc.frame_idx, rc.box, valid, fail, feat.astype(np.float16), rc.channels, rc.spans, rc.meta, rc.runtime, rc.moving).validate() | |
| Gh2, Vh2, ok2 = DA.route_hist(rc2, anchors[i:i + 1], 40) | |
| assert _same(Gh[i], Gh2[0]) and np.array_equal(Vh[i], Vh2[0]) and ok[i] == ok2[0] | |
| Gh3, _, _ = DA.route_hist(rc2, anchors[i:i + 1] + 0.1, 40) | |
| assert not _same(Gh[i], Gh3[0]) | |
| def test_trainer_argv_matches_survivor_recipe(): | |
| argv = FU.trainer_argv(1, "/x/out") | |
| rec = json.loads((CWM3 / "results" / "runs" / "U0_kin_can_noK_s1_kcv2_clip8_ln" / "run_train_U0_kin_can_noK_clip8_ln_s1_latest.json").read_text())["command"].split() | |
| want = [t for t in rec if t not in ("python", "train/train_cwm3.py")] | |
| want = [("/x/out" if w.startswith("results/runs/") else w) for w in want] | |
| assert [a for a in argv if a not in ("--stats-version", "2")] == want, (argv, want) # identical up to the explicit default --stats-version 2 | |
| assert argv[argv.index("--stats-version") + 1] == "2" | |
| a = FU.pair_argv("l2cs_gaze", 0, "manual", k0=True) | |
| assert a[-1] == "--b-k0" and "--kin-targets" in a and a[a.index("--kin-version") + 1] == "v2" and a[a.index("--tag") + 1] == "dmsU_s0_l2cs_gaze_vs_k0" | |
| b = FU.pair_argv("l2cs_gaze", 2, "engaged") | |
| assert b[b.index("--b") + 1] == "results/runs/U0_kin_can_noK_s2_kcv2_clip8_ln" and b[b.index("--a") + 1] == "results/runs/U0_dms_l2cs_gaze_s2_kcv2_clip8_ln" | |
| assert FU.eligible_events("l2cs_gaze") == ("head_turn_onset",) and FU.eligible_events("videomae2") == FU.EVENTS and FU.eligible_events("ov_action") == FU.EVENTS | |
| # ----------------------------------------------------------------------------- (b) fit_readout | |
| def test_fit_readout_layout_contract_on_synthetic_data(): | |
| with tempfile.TemporaryDirectory() as d: | |
| root = Path(d) | |
| summ_root = root / "summ" | |
| for split, sd in (("train", 0), ("val", 1)): | |
| FU.synthetic_summary("synthetic", "manual", split, summ_root / f"summ_manual_{split}.npz", n_routes=3, seed=sd) | |
| X, names, defined, uid, t, info = FR.summary_features("synthetic", "manual", "train", summ_root) | |
| assert X.shape == (len(uid), len(names)) and "valid_share_8s" in names and "rule0" in names and np.isnan(X[~defined]).all() | |
| with np.load(S12R / "targets_manual_train.npz", allow_pickle=True) as z: | |
| dongle = z["dongle"] | |
| tab = FR.synthetic_label_table(uid, t, defined, dongle, n=200) | |
| out = FR.fit("synthetic", 0, models=("lr", "hgb"), label_table=tab, splits=("val",), regimes=("manual",), summ_root=summ_root, out_root=root / "runs", res_root=root / "readout", gold=False, log=lambda m: None) | |
| assert out["arm"] == "abl_dms_synthetic_learned" and len(out["layouts"]) == 1 and out["selected"]["selected"] in ("lr", "hgb") | |
| m = LY.validate_layout(out["layouts"][0]) | |
| L = LY.read_layout(out["layouts"][0]) | |
| assert m["score_source"] == "" and m["dms_fields_used"] == [] and m["roi"] == "fullframe_causal" and m["seed"] == 0 and m["split"] == "val" | |
| cols = ["dms_synthetic_learned_lr_t15_w10_n979", "dms_synthetic_learned_hgb_t15_w10_n979", "dms_synthetic_learned_sel_t15_w10_n979"] | |
| assert L["w_names"] == cols | |
| _, _, dv, _, _, _ = FR.summary_features("synthetic", "manual", "val", summ_root) | |
| assert np.array_equal(L["w_mask"], dv) and np.isnan(L["w"][~dv].astype(np.float32)).all() and np.isfinite(L["w"][dv].astype(np.float32)).all() | |
| assert (root / "runs" / "abl_dms_synthetic_learned_s0" / "PRED_DONE").exists() | |
| import results_schema as RS | |
| RS.validate(LY.schema_row(m)) | |
| RS.validate(dict(LY.schema_row(m), score_source=m["score_source"])) | |
| sel = json.loads(Path(out["res_dir"], "selected.json").read_text()) | |
| assert sel["n_labels"] == len(tab) and set(sel["per_model"]) == {"lr", "hgb"} | |
| # the protocol pieces are the unchanged train_warning_head ones | |
| T = FR.TW() | |
| assert {"lr", "hgb", "mlp"} <= set(T.GRID) and len(T.GRID["hgb"]) == 4 and callable(T.cv_auroc) and callable(T.make_model) and callable(T.WG.temperature_fit) | |
| assert FR.arm_name("x", ctx=True) == "abl_dms_x_ctx" and FR.arm_name("x", with_u0=True, impute="prior") == "abl_dms_x_learnedu0imp" | |
| # ----------------------------------------------------------------------------- (c) chain_style | |
| def _extractor_channels(): | |
| out = {} | |
| import l2cs_gaze, mp_face, ov_gaze, openface3, ov_action, videomae2 # noqa: E401 (extractors import cache_schema / video_io / base only at module level) | |
| out["l2cs_gaze"] = list(l2cs_gaze.CHANNELS) | |
| out["mp_face"] = list(mp_face.CHANNELS) | |
| out["ov_gaze"] = list(ov_gaze.CHANNELS) | |
| out["openface3"] = list(openface3.CHANNELS) | |
| out["ov_action"] = list(ov_action.ENC_CHANNELS) + list(ov_action.LOGIT_CHANNELS) | |
| out["videomae2"] = list(videomae2.VideoMAE2Extractor.channels) + [videomae2.EMB_CH] | |
| return out | |
| def test_chain_mapping_covers_every_extractor_channel(): | |
| chans = _extractor_channels() | |
| assert set(chans) == set(CS.CHANNEL_ROLES) == set(CS.SPEC) == set(FU.BASELINES) | |
| for b, cl in chans.items(): | |
| assert set(CS.CHANNEL_ROLES[b]) == set(cl), (b, set(CS.CHANNEL_ROLES[b]) ^ set(cl)) | |
| sp = CS.SPEC[b] | |
| if sp is None: | |
| assert not CS.applicable(b) and all(("n/a" in r) for r in CS.CHANNEL_ROLES[b].values()) | |
| continue | |
| assert sp["yaw"] in cl and (sp["pitch"] is None or sp["pitch"] in cl) and all(c in cl for c in (sp["closed"] or [])) | |
| assert CS.CHANNEL_ROLES[b][sp["yaw"]].startswith("distracted.pose_yaw+gaze_off.head_yaw") | |
| for c in (sp["closed"] or []): | |
| assert CS.CHANNEL_ROLES[b][c].startswith("distracted.eye_") | |
| assert CS.channels_undefined(b) == ([] if sp["closed"] else ["distracted.eye_closure"]) | |
| assert CS.TH["gaze_off_head_yaw_rad"] == 0.4 and abs(CS.TH["gaze_off_head_yaw_deg"] - 22.918) < 1e-2 and CS.TH["distracted_pose_yaw_deg"] == 20.0 and CS.TH["distracted_pose_pitch_deg"] == 15.0 and CS.TH["eye_closed_p"] == 0.5 | |
| def test_chain_states_follow_documented_thresholds(): | |
| G = np.array([[0.0, 0.0, 0.1], [21.0, 0.0, 0.1], [23.0, 0.0, 0.1], [0.0, 16.0, 0.1], [0.0, 0.0, 0.9], [np.nan, 0.0, 0.9], [-179.0, 0.0, 0.1]], np.float32) | |
| A = CS.channel_states(G, ["yaw", "pitch", "closed"], {"yaw": "yaw", "pitch": "pitch", "closed": ["closed"], "closed_kind": "closed_p"}) | |
| assert A.shape == (7, 2) | |
| assert A[0].tolist() == [0, 0] and A[1].tolist() == [1, 0] and A[2].tolist() == [1, 1] and A[3].tolist() == [1, 0] and A[4].tolist() == [1, 0] | |
| assert np.isnan(A[5]).all() # face invalid -> both channels NaN | |
| assert A[6].tolist() == [1, 1] # circular deviation: -179 deg is 179 deg off axis -> off | |
| A2 = CS.channel_states(G[:, :2], ["yaw", "pitch"], {"yaw": "yaw", "pitch": "pitch", "closed": None, "closed_kind": None}) | |
| assert A2[4].tolist() == [0, 0] # no eye-closure output -> pose only | |
| def test_chain_replica_architecture_and_composition_match_legacy(): | |
| torch = pytest.importorskip("torch") | |
| import score_val_marginals as SVM | |
| ours = CS.build_net("ds,dyn,cab", 7).state_dict() | |
| ref = SVM.build_net("ds,dyn,cab").state_dict() | |
| assert list(ours) == list(ref) and all(ours[k].shape == ref[k].shape for k in ref), "hazard-chain replica differs from train_att / score_val_marginals.build_net" | |
| ours2 = CS.build_net("ds", 11).state_dict() | |
| assert ours2["ds.p.weight"].shape == (128, 11, 1) and "dyn.p.weight" not in ours2 | |
| from compose import legacy_compose, marginal | |
| rng = np.random.default_rng(0) | |
| n = 50 | |
| hon, hoff = rng.uniform(0.01, 0.3, (n, 25, 3)).astype(np.float32), rng.uniform(0.01, 0.4, (n, 25, 3)).astype(np.float32) | |
| hon2, hoff2 = rng.uniform(0.01, 0.3, (n, 25, 3)).astype(np.float32), rng.uniform(0.01, 0.4, (n, 25, 3)).astype(np.float32) | |
| A0 = (rng.random((n, 3)) < 0.3).astype(np.float32) | |
| pA, att, p_in = CS.compose_att(hon, hoff, hon2, hoff2, A0) | |
| for j, (a, b) in enumerate(((hon, hoff), (hon, hoff), (hon2, hoff2))): | |
| assert np.allclose(pA[:, :, j], marginal(a[:, :, j], b[:, :, j], A0[:, j]), atol=1e-6) | |
| assert np.allclose(att, legacy_compose(pA, None, 15, None)["att_only"], atol=1e-6) | |
| assert np.allclose(CS.state_att(A0), 1 - (1 - (1 - np.prod(1 - A0, -1))) ** 15) | |
| # one CPU epoch of the replica on synthetic arrays runs end to end and predicts hazards in (0, 1) | |
| A0s = (rng.random((256, 3)) < 0.2).astype(np.float32) | |
| A_s = np.repeat(A0s[:, None, :], 25, 1) | |
| flip = rng.random(256) < 0.3 # 30 % of the rows carry a state flip (as real chains), the rest persist | |
| A_s[flip, 12:, 0] = 1 - A_s[flip, 12:, 0] | |
| d = {"D": rng.normal(0, 1, (256, 40, 4)).astype(np.float32), "X": rng.normal(0, 1, (256, 40, 5)).astype(np.float32), "A": A_s, "M": np.ones((256, 25, 3), bool), "A0": A0s} | |
| pred = CS.train_chain(d, d, "ds", 0, epochs=1, device="cpu", rows_per_epoch=128, batch=32, log=lambda m: None) | |
| h_on, h_off = pred(d) | |
| assert h_on.shape == (256, 25, 3) and np.all((h_on > 0) & (h_on < 1)) and np.isfinite(pred.best_val) | |
| # ----------------------------------------------------------------------------- (d) hygiene / launcher / evaluator argv | |
| def test_new_files_have_no_forbidden_tokens(): | |
| pat = re.compile("|".join(map(re.escape, FORBIDDEN))) | |
| hits = [f"{p.name}:{i}" for p in NEW_FILES for i, l in enumerate(p.read_text().splitlines(), 1) if pat.search(l)] | |
| assert not hits, hits | |
| for p in NEW_FILES: | |
| assert p.exists() | |
| def test_queue_and_launcher(): | |
| q = (PKG / "queue_dms.txt").read_text().splitlines() | |
| lines = [l for l in q if l.strip() and not l.startswith("#")] | |
| assert len(lines) >= 100 | |
| markers = [] | |
| for l in lines: | |
| parts = l.split("||") | |
| assert len(parts) in (2, 3), l | |
| marker, args = parts[0], parts[1] | |
| script = args.split()[0] | |
| assert (CWM3 / script).exists(), script | |
| assert marker.startswith("results/") and "abl_dms_" in marker or marker.startswith("results/baselines_dms/") or "U0_dms_" in marker, marker | |
| markers.append(marker) | |
| if len(parts) == 3: | |
| assert parts[2].isdigit() | |
| if "--train" in args: | |
| assert parts[2] != "0", "GPU training lines must carry a VRAM need" | |
| assert len(markers) == len(set(markers)), "duplicate done markers" | |
| for b in FU.BASELINES: | |
| assert any(f"--baseline {b} --train --seed 2" in l for l in lines) and any(f"--baseline {b} --pair" in l for l in lines) and any(f"fit_readout.py --baseline {b} --seed 2 --ctx" in l for l in lines) | |
| sh = (PKG / "run_queue_dms.sh").read_text() | |
| assert "MemAvailable" in sh and "28 * 1024 * 1024" in sh and "ntr -lt 4" in sh and "(need + 4) * 1024" in sh and "setsid nohup" in sh and "pids.txt" in sh | |
| assert os.access(PKG / "run_queue_dms.sh", os.X_OK) | |
| def test_eval_dms_argv_and_coverage(): | |
| argv = ED.operating_argv("l2cs_gaze", ["abl_dms_l2cs_gaze_learned"] * 3, [0, 1, 2], "dms_l2cs_gaze_learned_sel_t15_w10_n979", "L2CS") | |
| assert argv[1].endswith("eval/operating_points.py") and argv[argv.index("--item") + 1] == "dms_l2cs_gaze" | |
| loads = argv[argv.index("--loads") + 1].split(",") | |
| assert {"dms_matched", "low10", "same_recall", "low1", "low3", "low5"} <= set(loads) | |
| src = (CWM3 / "eval" / "operating_points.py").read_text() | |
| assert 'tok.startswith("low")' in src, "operating_points.py low<N> token expected (the 1/3/5 alerts/h sweep relies on it)" | |
| with tempfile.TemporaryDirectory() as d: | |
| root = Path(d) | |
| puid, pt = LY.pack_uid_t("manual", "val") | |
| s = np.full(len(puid), np.nan, np.float32) | |
| routes = np.unique(puid)[:4] | |
| sel = np.isin(puid, routes) | |
| s[sel] = 0.5 | |
| LY.write_layout("abl_dms_synthetic_chain", "manual", 0, "val", puid, pt, {"dms_synthetic_chain_att3s": s}, out_root=root / "runs", score_source="legacy_chain") | |
| rows = ED.coverage_rows("abl_dms_synthetic_chain", 0, "dms_synthetic_chain_att3s", root / "runs") | |
| assert len(rows) == 1 and rows[0]["finite_rows"] == int(sel.sum()) and rows[0]["decoded_routes"] == 4 and abs(rows[0]["hours"] - len(puid) * 0.2 / 3600) < 1e-9 | |
| sec, _ = ED.append_coverage(["abl_dms_synthetic_chain"], [0], "dms_synthetic_chain_att3s", table=root / "t.md", root=root / "runs") | |
| assert "| abl_dms_synthetic_chain | 0 |" in sec and (root / "t.md").read_text().count("## coverage") == 1 | |
| g, gd = ED.gold_argv("x", ["abl_dms_lmface_rule"], [0], "dms_lmface_rule0", stamp="s") if (CWM3 / "results" / "runs" / "abl_dms_lmface_rule_s0" / "pred_layout_manual_test.npz").exists() else (None, None) | |
| if g is not None: | |
| assert g[1].endswith("cwm2/eval/warning_gold.py") and g[g.index("--item") + 1] == "dms_x" and len([a for a in g if a.endswith(".npz")]) == 4 and gd.name == "dms_x_s" | |
| # ----------------------------------------------------------------------------- (e) anchor-source rule of the substitute pack (2026-09-15, layer-1 launch; new tests only) | |
| def test_anchor_source_rule_is_explicit_per_baseline(): | |
| assert FU.resolve_anchor_source("videomae2") == "emb" and FU.resolve_anchor_source("ov_action") == "emb" | |
| for b in ("ov_gaze", "mp_face", "openface3", "l2cs_gaze"): | |
| assert FU.resolve_anchor_source(b) == "S0", b | |
| assert FU.resolve_anchor_source("videomae2", "S0") == "S0" and FU.resolve_anchor_source("l2cs_gaze", "emb") == "emb" | |
| with pytest.raises(AssertionError): | |
| FU.resolve_anchor_source("videomae2", "geometry") | |
| def test_videomae2_pack_from_emb_anchor_and_face_pack_from_S0_pca_on_train_only(): | |
| """a videomae2-like summary carries BOTH a non-empty S0 (3 crop-geometry channels) and emb_anchor (D = 16): the pack must be built from emb_anchor (PCA-11 fitted on the | |
| TRAIN rows only, applied before the history), never from the geometry S0; a face summary (l2cs_gaze-like S0) is built from S0.""" | |
| with tempfile.TemporaryDirectory() as d: | |
| root = Path(d) | |
| emb = tuple(f"e{j}_p" for j in range(16)) | |
| geo = ("vmae2_crop_w", "vmae2_crop_h", "vmae2_lb_scale") | |
| # train and val emb rows live on DIFFERENT distributions (val shifted by +7) so a val leak into the PCA fit is detectable on the mean | |
| FU.synthetic_summary("videomae2", "manual", "train", root / "summ" / "summ_manual_train.npz", n_routes=2, channels=emb, seed=0, kind="emb", s0_channels=geo) | |
| FU.synthetic_summary("videomae2", "manual", "val", root / "summ" / "summ_manual_val.npz", n_routes=2, channels=emb, seed=1, kind="emb", s0_channels=geo, offset=7.0) | |
| with np.load(root / "summ" / "summ_manual_train.npz", allow_pickle=True) as z: | |
| assert z["S0"].shape[1] == 3 and z["emb_anchor"].shape[1] == 16 # both blocks present, S0 non-empty (the real videomae2 layout) | |
| E_tr = z["emb_anchor"].astype(np.float32) | |
| v_tr = z["row_ok"].astype(bool) & np.isfinite(E_tr).all(1) | |
| uids_tr = set(np.unique(z["uid"][v_tr]).tolist()) | |
| with np.load(root / "summ" / "summ_manual_val.npz", allow_pickle=True) as z: | |
| E_va = z["emb_anchor"].astype(np.float32) | |
| v_va = z["row_ok"].astype(bool) & np.isfinite(E_va).all(1) | |
| # auto -> emb for videomae2 | |
| blk = FU.load_anchor_block(root / "summ" / "summ_manual_train.npz") | |
| assert blk["kind"] == "emb_anchor" and blk["anchor_source"] == "emb" and blk["channels"] == [f"emb{j}" for j in range(16)] and blk["split"] == "train" | |
| blk_s0 = FU.load_anchor_block(root / "summ" / "summ_manual_train.npz", anchor_source="S0") | |
| assert blk_s0["kind"] == "S0" and blk_s0["channels"] == list(geo) | |
| paths = FU.build_packs("videomae2", "summ", root / "kin_subst", regimes=("manual",), splits=("train", "val"), summ_root=root / "summ", log=lambda m: None) | |
| assert len(paths) == 2 and all(FU.verify_pack(p) for p in paths) | |
| red = DA.PCAReducer.load(root / "kin_subst" / "pca11.npz") | |
| assert red.channels == [f"emb{j}" for j in range(16)] and red.W.shape == (16, 11) # fitted on the embedding, NOT on the 3 geometry channels | |
| # PCA fitted on TRAIN rows only: mean == mean of the TRAIN valid emb rows (the fit samples <= 100k rows per regime; here every row), far from the val mean (+7) | |
| mu_tr = E_tr[v_tr].astype(np.float64).mean(0) | |
| mu_va = E_va[v_va].astype(np.float64).mean(0) | |
| assert np.allclose(red.mu, mu_tr, atol=1e-3), (red.mu[:3], mu_tr[:3]) | |
| assert np.abs(red.mu - mu_va).max() > 5.0 | |
| assert set(red.fit_uids) <= uids_tr and red.n_fit == int(v_tr.sum()) | |
| fit = json.loads((root / "kin_subst" / "pca11.fit.json").read_text()) | |
| assert fit["fit_splits"] == ["train"] and all(p.endswith("summ_manual_train.npz") for p in fit["fit_files"]) and fit["anchor_source"] == "emb" | |
| for p in paths: | |
| prov = json.loads((p.parent / (p.stem + ".provenance.json")).read_text()) | |
| assert prov["anchor_source"] == "emb" and prov["history_source"].startswith("summ_emb_anchor") and prov["channels_in"] == [f"emb{j}" for j in range(16)] | |
| assert prov["channels_out"] == [f"pca{j}" for j in range(11)] and prov["reducer"]["n_in"] == 16 and "before the history" in prov["reducer_applied"] | |
| with np.load(p, allow_pickle=True) as z: | |
| assert [str(c) for c in z["channels"]] == [f"pca{j}" for j in range(11)] | |
| # reduce-before-history == reduce-after-history on the val pack (linear per-step map) | |
| blk_va = FU.load_anchor_block(root / "summ" / "summ_manual_val.npz", "emb", "videomae2") | |
| Kh, Kv = FU.history_from_anchor_rows(blk_va["uid"], blk_va["t"], blk_va["S"], blk_va["valid"], 40) | |
| Z_after, _, ok_after = FU.force_dk(Kh, Kv, blk_va["channels"], red) | |
| with np.load(paths[1], allow_pickle=True) as z: | |
| Z_pack, ok_pack = z["Kh_c"].astype(np.float32), z["Khvalid"] | |
| assert np.array_equal(ok_pack, ok_after & (ok_after.any(1) & blk_va["row_ok"])[:, None]) | |
| assert np.allclose(np.nan_to_num(Z_pack[ok_pack]), np.nan_to_num(Z_after[ok_pack]), atol=2e-2, rtol=1e-2) # f16 storage of the pack | |
| # explicit --anchor-source S0 on the SAME videomae2 file -> the 3 geometry channels + valid_flag + pads (the override path) | |
| p_s0 = FU.build_packs("videomae2", "summ", root / "kin_subst_s0", regimes=("manual",), splits=("val",), summ_root=root / "summ", anchor_source="S0", log=lambda m: None)[0] | |
| with np.load(p_s0, allow_pickle=True) as z: | |
| assert [str(c) for c in z["channels"]] == list(geo) + ["valid_flag"] + [f"pad_{j}" for j in range(4, 11)] | |
| assert json.loads((p_s0.parent / (p_s0.stem + ".provenance.json")).read_text())["anchor_source"] == "S0" | |
| # a face baseline (l2cs_gaze-like S0 block, no emb_anchor): auto -> S0 | |
| FU.synthetic_summary("l2cs_gaze", "manual", "train", root / "summ_face" / "summ_manual_train.npz", n_routes=2, channels=("l2cs_face_conf", "l2cs_yaw", "l2cs_pitch", "l2cs_box_w"), seed=2) | |
| FU.synthetic_summary("l2cs_gaze", "manual", "val", root / "summ_face" / "summ_manual_val.npz", n_routes=2, channels=("l2cs_face_conf", "l2cs_yaw", "l2cs_pitch", "l2cs_box_w"), seed=3) | |
| pf = FU.build_packs("l2cs_gaze", "summ", root / "kin_subst_face", regimes=("manual",), splits=("train", "val"), summ_root=root / "summ_face", log=lambda m: None) | |
| prov_f = json.loads((pf[1].parent / (pf[1].stem + ".provenance.json")).read_text()) | |
| assert prov_f["anchor_source"] == "S0" and prov_f["history_source"].startswith("summ_S0") and prov_f["channels_out"][:4] == ["l2cs_face_conf", "l2cs_yaw", "l2cs_pitch", "l2cs_box_w"] | |
| # requesting emb on a face summary (no emb_anchor) fails loudly instead of silently falling back | |
| with pytest.raises(ValueError): | |
| FU.load_anchor_block(root / "summ_face" / "summ_manual_val.npz", anchor_source="emb") | |
| def test_fit_u_base_cli_is_additive(): | |
| import argparse | |
| src = (PKG / "fit_u_base.py").read_text() | |
| for flag in ("--anchor-source", "--summ-root", "--coverage", "--note"): | |
| assert f'"{flag}"' in src, flag | |
| assert 'choices=list(ANCHOR_SOURCES)' in src and FU.ANCHOR_SOURCES == ("auto", "S0", "emb") | |
| # --source auto keeps the per-route caches for the per-frame baselines (exact 10 Hz history) and the summaries for the window baselines | |
| assert "use_cache = source == \"cache\" or (source == \"auto\" and baseline not in WINDOW_BASELINES and len(complete) > 0)" in src | |
| assert "--note" in (PKG / "chain_style.py").read_text() | |
Xet Storage Details
- Size:
- 29.9 kB
- Xet hash:
- ee5d6f19d1fd97cac4d282c7206ce9bd1dc3aa6f679ca852a04a4a9ee395f079
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.