Buckets:
| """cwm3/tests/test_baselines_dms.py — contracts of the DMS-baseline package (plan §11, 2026-09-14). | |
| (a) causality : a synthetic 10 Hz cache with a spike planted at t_anchor + 0.1 s leaves EVERY summary / rule / chain column of that anchor unchanged | |
| (anchors >= 10 s apart so no window contains another anchor's future); summarise.window_index asserts t_frame <= t_anchor + 1e-6 and | |
| raises on an anchor off the grid; window caches: win_t_end <= t and a spike in a later window changes nothing. | |
| (a2) angles : angle deviations are circular (summarise.circ_dev): 179° vs -179° differ by 2°, not 358°; angle_block / rules.off_state are not `off` for that pair. | |
| (b) alignment : every summ_*.npz under results/baselines_dms/summ and every pred_layout under results/runs/abl_dms_* has uid / t equal to the s12 pack | |
| (skipped when none exists). | |
| (c) hygiene : no forbidden DMS token (the list of tests/test_no_dms_inputs.py) anywhere in baselines_dms/**/*.py; models/ train/ eval/ data/ common/ | |
| never import baselines_dms; arm names start with abl_dms_. | |
| (d) layout : results/runs/abl_dms_lmface_rule_s0 (R0, exists): w f16 in [0, 1] or NaN, w_mask bool, meta keys (arm, regime, seed, split, score_source, | |
| model_id, dms_fields_used == []), arm passes cwm2/eval/results_schema.validate; the layout writer round-trips and rejects misaligned uid / t. | |
| (e) rules : constants equal decisions.md (20/12°, 15°, 0.5, 2.0 s, 1.5 s, 0.25); chain_style == cwm2/compose.legacy_compose att_only. | |
| (f) adapter : write_substitute_kin_pack -> Kh_c f16[N,40,11] row-aligned; the UNCHANGED data layer (cwm3_dataset.load_split with KINC_V2 pointed at the | |
| pack) loads it; build_kx_numpy == train/batch.prep_U on CPU. | |
| Run: cd cwm3 && python -m pytest -q tests/test_baselines_dms.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 | |
| PKG = CWM3 / "baselines_dms" | |
| S12R = EXPLORE / "s12_wm" / "results" | |
| for p in (CWM3, PKG, CWM3 / "data", CWM3 / "train", CWM3 / "tests"): | |
| if str(p) not in sys.path: | |
| sys.path.insert(0, str(p)) | |
| import cache_schema as cs # noqa: E402 | |
| import summarise as SM # noqa: E402 | |
| import rules as RL # noqa: E402 | |
| import layout as LY # noqa: E402 | |
| SPEC = {"angle": {"yaw_deg": (20, 12), "pitch_deg": (15, 9)}, "eye_state": {"closed_p": {"kind": "closed_p", "th": 0.5}}, "prob": ["conf"], | |
| "rule": {"yaw": "yaw_deg", "pitch": "pitch_deg", "closed": "closed_p"}, "chain": {"head_dev": ("yaw_deg", "pitch_deg"), "closed": "closed_p", "gaze_off": None}} | |
| 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)]) | |
| def _cache(): | |
| return cs.synthetic_route_cache(channels=("yaw_deg", "pitch_deg", "closed_p", "conf"), t1=120.0, seed=3) | |
| # ----------------------------------------------------------------------------- (a) causality | |
| def test_causality_spike_after_anchor_changes_nothing(): | |
| rc = _cache() | |
| anchors = np.arange(12.0, 118.0, 10.0) | |
| sm = SM.summarise_route(rc, anchors, SPEC) | |
| cols = RL.rule_columns(sm, sm["spec"]) | |
| assert {"rule0", "ruleref", "chain_att3s_rule0", "chain_att3s_ruleref", "longest_closure_s"} <= set(cols) | |
| k = np.round(rc.t.astype(np.float64) * SM.HZ).astype(np.int64) | |
| pos = {int(kk): i for i, kk in enumerate(k)} | |
| for i, a in enumerate(anchors[:6]): | |
| hit = pos[int(round(a * SM.HZ)) + 1] # the 10 Hz step right after the anchor | |
| feat = rc.feat.astype(np.float32) | |
| valid, fail, mov = rc.valid.copy(), rc.fail.copy(), rc.moving.copy() | |
| valid[hit], fail[hit], mov[hit] = True, 0, True # valid AND moving: it would enter any non-causal reference | |
| feat[hit] = 1e3 | |
| feat[hit, 2] = 1.0 | |
| 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, mov).validate() | |
| sm2 = SM.summarise_route(rc2, anchors[i:i + 1], SPEC) | |
| cols2 = RL.rule_columns(sm2, sm2["spec"]) | |
| for key in ("X", "S0", "S0valid", "valid_share_8s", "n_valid_8s", "row_ok"): | |
| assert _same(sm[key][i], sm2[key][0]), (key, a) | |
| num = [c for c in cols if c != "chain_absent_channels"] | |
| assert sorted(cols) == sorted(cols2) and all(_same(cols[c][i], cols2[c][0]) for c in num), [c for c in num if not _same(cols[c][i], cols2[c][0])] | |
| assert cols["chain_absent_channels"] == ["gaze_off"] | |
| sm3 = SM.summarise_route(rc2, anchors[i:i + 1] + 0.2, SPEC) # teeth: 0.2 s later the spike is inside the window | |
| assert not _same(sm["X"][i], sm3["X"][0]) | |
| def test_window_index_is_causal_and_on_grid(): | |
| rc = _cache() | |
| D = SM.dense_series(rc) | |
| anchors = np.arange(10.0, 100.0, 0.2) | |
| idx = SM.window_index(D["k0"], D["M"], anchors) | |
| ok = idx >= 0 | |
| t_frame = (idx + D["k0"]) / SM.HZ | |
| assert idx.shape == (len(anchors), SM.HIST) and np.all(t_frame[ok] <= (anchors[:, None] + 1e-6).repeat(SM.HIST, 1)[ok]) | |
| assert np.allclose(t_frame[:, -1], anchors) # the last step IS the anchor frame | |
| with pytest.raises(AssertionError): | |
| SM.window_index(D["k0"], D["M"], np.array([10.05])) # off the 10 Hz grid | |
| def test_window_cache_causality(): | |
| rng = np.random.default_rng(0) | |
| M, D = 200, 5 | |
| wv = rng.random(M) < 0.9 | |
| E = rng.normal(0, 1, (M, D)).astype(np.float32) | |
| E[~wv] = np.nan | |
| meta = cs.make_meta("synthetic", ["x"], code_sha="0" * 64, video_source="synthetic") | |
| wc = cs.WindowCache("u", (np.arange(M) * 0.5 + 1.5).astype(np.float32), E.astype(np.float16), wv, 16, np.array([[0.0, 101.0]], np.float32), meta, {}) | |
| anchors = np.arange(10.0, 90.0, 0.2) | |
| ew = SM.summarise_windows(wc, anchors) | |
| te = wc.win_t_end.astype(np.float64) | |
| for i in (0, 17, 233): | |
| j = int(np.searchsorted(te, anchors[i] + 1e-6, side="right")) - 1 | |
| assert te[j] <= anchors[i] + 1e-6 | |
| a = anchors[50] | |
| j_next = int(np.searchsorted(te, a + 1e-6, side="right")) # first window ending after the anchor | |
| E2 = E.copy() | |
| E2[j_next] = 99.0 | |
| wc2 = cs.WindowCache("u", wc.win_t_end, E2.astype(np.float16), wv | (np.arange(M) == j_next), 16, wc.spans, meta, {}) | |
| ew2 = SM.summarise_windows(wc2, anchors[50:51]) | |
| assert _same(ew2["emb_anchor"][0], ew["emb_anchor"][50]) and _same(ew2["emb_mean_8s"][0], ew["emb_mean_8s"][50]) and ew2["n_win_8s"][0] == ew["n_win_8s"][50] | |
| def test_angle_deviation_is_circular(): | |
| assert float(SM.circ_dev(179.0, -179.0)) == pytest.approx(-2.0) and abs(float(SM.circ_dev(-179.0, 179.0))) == pytest.approx(2.0) | |
| assert np.allclose(np.abs(SM.circ_dev(np.array([179.0, -179.0, 0.0, 90.0, 10.0]), np.array([-179.0, 179.0, 180.0, -90.0, 3.0]))), [2.0, 2.0, 180.0, 180.0, 7.0]) | |
| assert np.isnan(SM.circ_dev(np.nan, 0.0)) and np.isnan(SM.circ_dev(5.0, np.nan)) | |
| assert np.allclose(SM.circ_dev(np.array([-100.0, 170.0]), 0.0), [-100.0, 170.0]) # inside (-180, 180) the deviation is the plain difference | |
| T = SM.HIST | |
| xh = np.full((1, T), 179.0, np.float32) | |
| Vh = np.ones((1, T), bool) | |
| ref = np.full((1, T), -179.0, np.float32) | |
| X, names, st = SM.angle_block(xh, Vh, 20.0, 12.0, ref, "yaw_") # 179 vs a -179 reference: 2 deg -> never off | |
| assert not st["off"].any() and X[0, names.index("yaw_max_abs_3s")] == pytest.approx(2.0) and X[0, names.index("yaw_share_off_8s")] == 0.0 | |
| X0, names0, st0 = SM.angle_block(xh, Vh, 20.0, 12.0, None, "yaw_") # vs the camera axis it IS 179 deg off | |
| assert st0["off"].all() and X0[0, names0.index("yaw_max_abs_3s")] == pytest.approx(179.0) | |
| off, sv = RL.off_state(xh, None, Vh, ref_yaw=ref) | |
| assert sv.all() and not off.any() | |
| off0, _ = RL.off_state(xh, None, Vh) | |
| assert off0.all() | |
| # ----------------------------------------------------------------------------- (b) alignment of anything already on disk | |
| def _pack(regime, split): | |
| with np.load(S12R / f"targets_{regime}_{split}.npz", allow_pickle=True) as z: | |
| return z["uid"], z["t"] | |
| def test_existing_summaries_and_layouts_are_aligned(): | |
| files = sorted((CWM3 / "results" / "baselines_dms" / "summ").glob("*/summ_*_*.npz")) + sorted((CWM3 / "results" / "runs").glob("abl_dms_*/pred_layout_*_*.npz")) | |
| if not files: | |
| pytest.skip("no summ / layout on disk yet") | |
| for p in files: | |
| with np.load(p, allow_pickle=True) as z: | |
| m = re.match(r"(summ|pred_layout)_(manual|engaged)_(train|val|test)\.npz", p.name) | |
| assert m, p | |
| uid, t = _pack(m.group(2), m.group(3)) | |
| assert np.array_equal(z["uid"], uid) and np.array_equal(z["t"], t), p | |
| # ----------------------------------------------------------------------------- (c) hygiene | |
| def test_package_has_no_forbidden_tokens_and_is_not_imported_by_core(): | |
| import test_no_dms_inputs as TN | |
| pat = re.compile("|".join(map(re.escape, TN.FORBIDDEN))) | |
| hits = [] | |
| for p in sorted(PKG.rglob("*.py")): | |
| if "__pycache__" in p.parts or "vendor" in p.parts: | |
| continue | |
| for i, line in enumerate(p.read_text().splitlines(), 1): | |
| if pat.search(line): | |
| hits.append(f"{p.relative_to(CWM3)}:{i}: {line.strip()[:80]}") | |
| assert not hits, "forbidden DMS tokens in baselines_dms:\n" + "\n".join(hits) | |
| imp = re.compile(r"^\s*(from\s+baselines_dms|import\s+baselines_dms)|baselines_dms[\"'/]") | |
| bad = [] | |
| for d in ("models", "train", "eval", "data", "common"): | |
| for p in sorted((CWM3 / d).glob("*.py")): | |
| for i, line in enumerate(p.read_text().splitlines(), 1): | |
| if imp.search(line): | |
| bad.append(f"{p.relative_to(CWM3)}:{i}: {line.strip()[:80]}") | |
| assert not bad, "core packages must never import baselines_dms:\n" + "\n".join(bad) | |
| assert LY.ARM_PREFIX == "abl_dms_" | |
| # ----------------------------------------------------------------------------- (d) layout contract | |
| R0 = CWM3 / "results" / "runs" / "abl_dms_lmface_rule_s0" | |
| def test_r0_layout_contract(regime, split): | |
| p = R0 / f"pred_layout_{regime}_{split}.npz" | |
| if not p.exists(): | |
| pytest.skip("R0 layout not written") | |
| m = LY.validate_layout(p, check_pack=True) | |
| assert m["arm"] == "abl_dms_lmface_rule" and m["regime"] == regime and m["split"] == split and m["seed"] == 0 and m["dms_fields_used"] == [] | |
| assert m["score_source"] and m["model_id"] | |
| sys.path.insert(0, str(EXPLORE / "cwm2" / "eval")) | |
| RS = pytest.importorskip("results_schema") | |
| RS.validate(LY.schema_row(m)) | |
| def test_layout_writer_roundtrip_and_alignment_assert(): | |
| uid, t = _pack("engaged", "val") | |
| rng = np.random.default_rng(1) | |
| s = rng.random(len(uid)).astype(np.float32) | |
| s[rng.random(len(uid)) < 0.3] = np.nan | |
| with tempfile.TemporaryDirectory() as d: | |
| p = LY.write_layout("abl_dms_unittest_rule0", "engaged", 2, "val", uid, t, {"dms_unittest_rule0": s}, out_root=d, meta_extra={"baseline": {"id": "unittest"}}) | |
| m = LY.validate_layout(p) | |
| L = LY.read_layout(p) | |
| assert m["seed"] == 2 and L["w"].dtype == np.float16 and np.array_equal(L["w_mask"], np.isfinite(s)) and L["uid"].tobytes() == uid.tobytes() and L["t"].tobytes() == t.tobytes() | |
| with pytest.raises(AssertionError): | |
| LY.write_layout("abl_dms_unittest_rule0", "engaged", 2, "val", uid, t + 0.2, {"x": s}, out_root=d) | |
| with pytest.raises(AssertionError): | |
| LY.write_layout("abl_dms_unittest_rule0", "engaged", 2, "val", uid, t, {"x": s * 3}, out_root=d) | |
| with pytest.raises(AssertionError): | |
| LY.write_layout("abl_unittest", "engaged", 2, "val", uid, t, {"x": s}, out_root=d) | |
| # ----------------------------------------------------------------------------- (e) rules | |
| def test_rule_constants_and_chain_formula(): | |
| R = RL.RULE | |
| assert (R["yaw_on_deg"], R["yaw_off_deg"], R["pitch_on_deg"], R["closed_th"], R["dwell_scale_s"], R["closure_scale_s"], R["min_valid_share"], R["hist_s"]) == (20.0, 12.0, 15.0, 0.5, 2.0, 1.5, 0.25, 8.0) | |
| s = RL.rule_score([0.0, 1.0, 4.0, np.nan], [0.0, 3.0, np.nan, np.nan], [0.1, np.nan, 0.5, np.nan], [1.0, 1.0, 1.0, 1.0]) | |
| assert np.allclose(s[:3], [0.1, 1.0, 1.0]) and np.isnan(s[3]) and np.isnan(RL.rule_score([4.0], [0.0], [0.5], [0.2])[0]) | |
| rng = np.random.default_rng(0) | |
| hd = (rng.random(300) < 0.3).astype(np.float32) | |
| ec = rng.random(300).astype(np.float32) | |
| go = (rng.random(300) < 0.2).astype(np.float32) | |
| r = RL.chain_style({"head_dev": hd, "eye_closed": ec, "gaze_off": go}) | |
| sys.path.insert(0, str(EXPLORE / "cwm2")) | |
| compose = pytest.importorskip("compose") | |
| ref = compose.legacy_compose(r["pA"].astype(np.float64), None, 15, None)["att_only"] | |
| assert np.allclose(r["att_only"], ref, atol=1e-6) | |
| assert np.allclose(r["att_only"], 1 - (1 - (1 - (1 - hd) * (1 - ec) * (1 - go))) ** 15, atol=1e-6) | |
| r0 = RL.chain_style({"head_dev": hd, "eye_closed": None, "gaze_off": None}) | |
| assert r0["meta"]["absent_channels"] == ["eye_closed", "gaze_off"] and set(np.unique(r0["att_only"]).tolist()) <= {0.0, 1.0} | |
| # ----------------------------------------------------------------------------- (f) dataset adapter -> unchanged data layer / prep_U | |
| def test_substitute_kin_pack_loads_through_unchanged_data_layer(): | |
| import dataset_adapter as DA | |
| import cwm3_dataset as ds | |
| uid, t = _pack("manual", "val") | |
| routes = [str(u) for u in np.unique(uid)[:2]] | |
| with tempfile.TemporaryDirectory() as d: | |
| cdir = Path(d) / "cache" / "synthetic" | |
| for i, u in enumerate(routes): | |
| ts = t[uid == u].astype(np.float64) | |
| rc = cs.synthetic_route_cache(uid=u, t0=max(0.0, ts.min() - 8.2), t1=ts.max() + 0.2, channels=("yaw_deg", "pitch_deg", "closed_p", "conf"), seed=i, meta={"baseline": "synthetic"}) | |
| rc.save(cdir / f"{u}.npz") | |
| out_dir = Path(d) / "kin_sub" | |
| p = DA.write_substitute_kin_pack("synthetic", "manual", "val", out_dir, cache_dir=cdir, log=lambda m: None) | |
| with np.load(p, allow_pickle=True) as z: | |
| assert np.array_equal(z["uid"], uid) and np.array_equal(z["t"], t) | |
| assert z["Kh_c"].dtype == np.float16 and z["Kh_c"].shape == (len(uid), 40, DA.DK) and z["Khvalid"].shape == (len(uid), 40) and z["row_ok"].dtype == np.bool_ | |
| ok = z["row_ok"] | |
| assert ok.any() and not ok[~np.isin(uid, routes)].any() | |
| Kh = z["Kh_c"].astype(np.float32) | |
| V = z["Khvalid"] | |
| assert np.isfinite(Kh[V][:, [0, DA.DK - 1]]).all() and np.isnan(Kh[~V]).all() | |
| assert [str(c) for c in z["channels"]][:4] == ["yaw_deg", "pitch_deg", "closed_p", "conf"] and len(z["channels"]) == DA.DK | |
| links = DA.link_targets(out_dir, "v2") | |
| assert all(l.exists() for l in links) | |
| # the UNCHANGED data layer loads the substitute when KINC_V2 points at it | |
| raw = DA.load_via_dataset(out_dir, "manual", "val", kin_version="v2") | |
| assert raw["Kh"].shape == (len(uid), 40, DA.DK) and raw["kin_row_ok"].sum() == ok.sum() | |
| # numpy mirror of train/batch.prep_U on the rows of the decoded routes | |
| torch = pytest.importorskip("torch") | |
| import batch as bt | |
| rows = np.flatnonzero(raw["kin_row_ok"])[:256] | |
| X = raw["X"][rows] | |
| Kh32 = raw["Kh"][rows].astype(np.float32) | |
| kmu, ksd = DA.compute_k_stats(raw["Kh"][raw["kin_row_ok"]]) | |
| xmu, xsd = DA.compute_x_stats(raw["X"]) | |
| st = {"kmu": kmu, "ksd": ksd, "xmu": xmu, "xsd": xsd} | |
| Kx_np = DA.build_kx_numpy(Kh32, X, st, clip=8.0) | |
| store = bt.GpuStore({"t": raw["t"][rows], "Kh": raw["Kh"][rows], "X": X, "mode": raw["mode"][rows]}, ["Kh", "X", "mode"], torch.device("cpu")) | |
| b, _ = bt.prep_U(store, torch.arange(len(rows)), st, {"kclip": 8.0}, need_targets=False) | |
| assert b["Kx"].shape == (len(rows), 40, DA.DK + 1 + 5 + 1) and np.allclose(b["Kx"].numpy(), Kx_np, atol=1e-5) | |
Xet Storage Details
- Size:
- 16.8 kB
- Xet hash:
- 32d5a162537c7cd44ff3e05d1c2934d586355188f6db8eb376185fed0344ba3f
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.