Buckets:
| """Unit tests for the kin v2p arm (data/build_kin_causal_v2p.py + the additive n_kin plumbing; pre-registration docs/decisions.md 2026-09-18T05:34:40Z). | |
| Synthetic tests are fast; data-dependent checks (shared arrays bit-identical to v2, proxies NaN where the face is invalid) skip when the pack is absent.""" | |
| import os, re, sys | |
| from pathlib import Path | |
| import numpy as np, pytest, torch | |
| ROOT = Path(__file__).resolve().parents[1] | |
| for p in (ROOT, ROOT / "data", ROOT / "eval", ROOT / "train"): | |
| sys.path.insert(0, str(p)) | |
| import build_kin_causal_v2p as V2P # noqa: E402 | |
| import pitch_wristface_diag as PW # noqa: E402 | |
| import cwm3_dataset as ds # noqa: E402 | |
| import batch as bt # noqa: E402 | |
| from models import cwm3 as M # noqa: E402 | |
| from models.branches import DriverBranch, build_branch # noqa: E402 | |
| import warning_features_ext as WFX # noqa: E402 | |
| V2P_DIR = Path(os.environ.get("KIN_V2P_DIR", ROOT / "results" / "targets_kin_causal_routes_v2p")) | |
| V2_DIR = ROOT / "results" / "targets_kin_causal_routes_v2" | |
| rng = np.random.default_rng(0) | |
| FORBIDDEN = ["driver_state", "faceProb", "isDistracted", "face_yaw", "face_pitch", "face_roll", "blinkProb", "dmonitoring", "driverState", "awarenessStatus", "distractedPct", "dstate", "leftEyeProb", "rightEyeProb"] | |
| def synth_kpts(T=300, W=1928, H=1208, seed=0): | |
| """plausible HALPE-136 driver: nose (0), shoulders (5/6), wrists (9/10), nose tip (56), eyes 62-73; normalised coords, conf 0.9.""" | |
| r = np.random.default_rng(seed) | |
| k = np.full((T, 136, 3), np.nan, np.float32); k[:, :, 2] = 0.9 | |
| cx, cy = 0.5, 0.45 | |
| k[:, 0, :2] = [cx, cy]; k[:, 56, :2] = [cx, cy + 0.02] | |
| k[:, 5, :2] = [cx - 0.08, cy + 0.20]; k[:, 6, :2] = [cx + 0.08, cy + 0.20] | |
| k[:, 9, :2] = [cx - 0.10, cy + 0.45]; k[:, 10, :2] = [cx + 0.10, cy + 0.45] | |
| for j in range(62, 68): k[:, j, :2] = [cx + 0.02 + 0.002 * (j - 62), cy - 0.02] | |
| for j in range(68, 74): k[:, j, :2] = [cx - 0.02 - 0.002 * (j - 68), cy - 0.02] | |
| k[:, :, :2] += r.normal(0, 1e-4, (T, 136, 2)).astype(np.float32) | |
| return k | |
| # ------------------------------------------------------------------------------------------------------------- proxies | |
| def test_lap_depth_sign_and_validity(): | |
| k = synth_kpts(50); W, H = 1928, 1208 | |
| d = V2P.lap_depth(k, W, H) | |
| assert d.shape == (50,) and np.isfinite(d).all() and (d > 0).all() # wrists below the shoulder line -> positive | |
| k2 = k.copy(); k2[:, [9, 10], 1] = k2[:, [5, 6], 1] - 0.3 # wrists raised above the shoulders -> negative | |
| assert (V2P.lap_depth(k2, W, H) < 0).all() | |
| k3 = k.copy(); k3[:10, 9, 2] = 0.1 # left wrist low confidence -> body invalid -> NaN | |
| d3 = V2P.lap_depth(k3, W, H); assert np.isnan(d3[:10]).all() and np.isfinite(d3[10:]).all() | |
| def test_pitch_dwell_semantics(): | |
| hd = np.zeros(60, bool); hd[10:25] = True; hd[40:45] = True | |
| valid = np.ones(60, bool); valid[12] = False; valid[30:50] = False # one short gap inside the run, a long gap before the second run | |
| pd_ = V2P.pitch_dwell(hd, valid) | |
| assert pd_[9] == 0 and np.isclose(pd_[10], 0.1) and np.isnan(pd_[12]) and np.isclose(pd_[13], 0.3) and np.isclose(pd_[24], 1.4) and pd_[25] == 0 | |
| assert np.isnan(pd_[30:50]).all() and pd_[50] == 0 # long undefined gap -> reset; second run (40:45) is entirely undefined | |
| hd2 = np.ones(500, bool); assert V2P.pitch_dwell(hd2, np.ones(500, bool)).max() <= V2P.PD_CAP | |
| def test_route_proxy_series_causal_prefix_and_face_invalid(): | |
| T = 400; k = synth_kpts(T); t = np.arange(T) * 0.1; moving = np.ones(T, bool) | |
| k[200:260, 56, 1] += 0.03 # head pitched down for 6 s | |
| k[300:320, 62:74, 2] = 0.1 # eyes invalid -> pitch undefined | |
| z = dict(native_wh=np.array([1928, 1208]), kpts=k.astype(np.float16), moving=moving, t=t.astype(np.float32)) | |
| S = V2P.route_proxy_series(z) | |
| assert S["P"].shape == (T, 3) and S["HD"].shape == (T,) and S["PD"].shape == (T,) | |
| assert np.isnan(S["P"][304:320, 0]).all() and np.isnan(S["HD"][304:320]).all() and np.isfinite(S["P"][299, 0]) # face invalid -> pitch / flag undefined (the 0.5 s trailing mean carries 4 frames) | |
| assert np.isfinite(S["P"][300:320, 1]).all() # wrist-face needs the body only | |
| assert np.nanmean(S["P"][210:260, 0]) > np.nanmean(S["P"][100:200, 0]) # pitched-down segment: larger deviation | |
| assert S["HD"][220:260].sum() > 0 and S["PD"][259] > 1.0 | |
| # causality: the series up to tick 250 does not depend on frames after it | |
| z2 = dict(z, kpts=z["kpts"][:251], moving=moving[:251], t=z["t"][:251]) | |
| S2 = V2P.route_proxy_series(z2) | |
| for key in ("P", "HD", "PD"): | |
| assert np.array_equal(S2[key], S[key][:251], equal_nan=True), key | |
| # ------------------------------------------------------------------------------------------------------------- data layer / model plumbing | |
| def test_dataset_v2p_version_and_widths(): | |
| ds.set_kin_source("causal_routes", "v2p") | |
| try: | |
| assert ds.kin_dir().name == "targets_kin_causal_routes_v2p" and ds.n_kin() == 14 and ds.KIN_CHANNELS_V2P[:11] == ds.KIN_CHANNELS | |
| assert ds.KIN_CHANNELS_V2P[11:] == ["head_pitch_kp", "wrist_face_kp", "hands_lap_kp"] == V2P.PROXY_CHANNELS | |
| finally: | |
| ds.set_kin_source("cwm2", "v1") | |
| assert ds.n_kin() == 11 | |
| with pytest.raises(AssertionError): | |
| ds.set_kin_source("causal_routes", "v9") | |
| def test_driver_branch_widths_default_unchanged(): | |
| torch.manual_seed(0); u11 = DriverBranch(in_ln=True) | |
| torch.manual_seed(0); u14 = DriverBranch(in_ln=True, n_kin=14) | |
| assert u11.gru.input_size == 18 and u11.N_KIN_IN == 12 and u11.in_ln.normalized_shape == (12,) | |
| assert u14.gru.input_size == 21 and u14.N_KIN_IN == 15 and u14.in_ln.normalized_shape == (15,) | |
| n = lambda m: sum(p.numel() for p in m.parameters()) | |
| assert n(u11.gru) == 56832 and n(u14.gru) == 56832 + 3 * 3 * 128 # only the input weights widen (3 gates x 3 columns x 128) | |
| out = u14({"Kx": torch.randn(4, 40, 21), "mode": torch.tensor([0, 1, 0, 1])}) | |
| assert out["U"].shape == (4, 25, 26) | |
| with pytest.raises(RuntimeError): | |
| u14({"Kx": torch.randn(4, 40, 18), "mode": torch.tensor([0, 1, 0, 1])}) | |
| with pytest.raises(AssertionError): | |
| DriverBranch(n_kin=14, can_gate=True) | |
| def test_model_kwargs_forward_n_kin_only_when_present(): | |
| base = {"branch": "U", "variant": "kin_can_noK_clip8_ln", "model": {"hid": 128, "hidden": 128, "dropout": 0.0}, "recipe": {"u_in_ln": True}, "cabin": "none", "kin_source": "causal_routes", "kin_version": "v2"} | |
| assert "n_kin" not in M._model_kwargs(base) | |
| v2p = dict(base, kin_version="v2p", n_kin=14) | |
| kw = M._model_kwargs(v2p); assert kw["n_kin"] == 14 and kw["in_ln"] is True | |
| assert build_branch("U", 0, **kw).gru.input_size == 21 | |
| assert M.base_cfg_summary(v2p)["n_kin"] == 14 and "n_kin" not in M.base_cfg_summary(base) | |
| import train_cwm3 as TC | |
| assert TC.model_kwargs(v2p)["n_kin"] == 14 and "n_kin" not in TC.model_kwargs(base) | |
| def test_prep_u_kx_width_and_k0std_targets_only(): | |
| """prep_U with a 14-wide Kh store -> Kx [B,40,21]; K0std keeps the 11 target columns; an 11-wide store reproduces the old layout.""" | |
| class Store: | |
| def __init__(self, dk): | |
| N = 6; self.device = torch.device("cpu") | |
| Kh = torch.randn(N, 40, dk); Kh[0, :5, :] = float("nan") | |
| self.t = {"Kh": Kh, "X": torch.randn(N, 40, 5), "mode": torch.zeros(N, dtype=torch.long), "M": torch.ones(N, 25, dtype=torch.bool), | |
| "K": torch.randn(N, 25, 11), "Kvalid": torch.ones(N, 25, dtype=torch.bool), "S": torch.zeros(N, 25, 2), "S0": torch.zeros(N, 2), "EM": torch.ones(N, 25, 3, dtype=torch.bool)} | |
| def has(self, k): return k in self.t | |
| for dk, width in ((11, 18), (14, 21)): | |
| st = {"xmu": np.zeros(5, np.float32), "xsd": np.ones(5, np.float32), "kmu": np.zeros(dk, np.float32), "ksd": np.ones(dk, np.float32)} | |
| b, tg = bt.prep_U(Store(dk), torch.arange(6), st, {"kclip": 8.0}, need_targets=True) | |
| assert b["Kx"].shape == (6, 40, width) and tg["K0std"].shape == (6, 11) and tg["Kstd"].shape == (6, 25, 11) | |
| assert torch.isfinite(b["Kx"]).all() | |
| def test_compute_stats_width_follows_kh(): | |
| tr = {"Kh": rng.normal(0, 1, (500, 40, 14)).astype(np.float16)} | |
| st = bt.compute_stats(tr, {"mu": np.zeros(3), "sd": np.ones(3)}) | |
| assert st["kmu"].shape == (14,) and st["ksd"].shape == (14,) | |
| st1 = bt.compute_stats_v1(tr, {"mu": np.zeros(3), "sd": np.ones(3)}); assert st1["kmu"].shape == (14,) | |
| def test_readout_kp_block_names_monotone_tag(): | |
| assert WFX.names_for(["kp"]) == WFX.KP_NAMES and WFX.tag_for(("dwell", "face", "kp")) == "dwfckp" and WFX.tag_for(("dwell", "face")) == "dwfc" | |
| assert WFX.monotone("kp_head_pitch_kp") == 1 and WFX.monotone("kp_wrist_face_kp") == -1 and WFX.monotone("kp_pitch_dwell_s") == 1 and WFX.monotone("kp_hands_lap_kp") == 1 and WFX.monotone("kp_valid") == 0 | |
| import train_warning_head as TW | |
| names = TW.feature_names("P", extra=("dwell", "face", "kp")) | |
| assert names[-5:] == WFX.KP_NAMES and not any(TW.is_future(n) for n in WFX.KP_NAMES) | |
| cst = TW.monotone_constraints("P", names); assert cst[-5:] == [1, -1, 1, 1, 0] | |
| def test_no_dms_identifiers_in_new_code(): | |
| files = [ROOT / "data" / "build_kin_causal_v2p.py"] + [p for p in (ROOT / "eval" / "run_joint_evals_v2p.sh", ROOT / "eval" / "report_kin_v2p_arm.py", ROOT / "eval" / "phone_episode_alerts.py") if p.exists()] | |
| for f in files: | |
| s = f.read_text() | |
| for w in FORBIDDEN: | |
| assert not re.search(r"\b" + re.escape(w) + r"\b", s), (f.name, w) | |
| # the new kp read-out block and the widened Kh path read our own keypoint pack only | |
| src = (ROOT / "train" / "warning_features_ext.py").read_text() | |
| kp_src = src[src.index("def kp_block"):src.index("BLOCKS = {")] | |
| for w in FORBIDDEN: | |
| assert w not in kp_src, w | |
| # ------------------------------------------------------------------------------------------------------------- built pack (data-dependent) | |
| def _need_pack(): | |
| if not (V2P_DIR / "kin_causal_manual_test.npz").exists() or not (V2_DIR / "kin_causal_manual_test.npz").exists(): | |
| pytest.skip("v2p / v2 packs not built") | |
| def test_built_pack_shared_arrays_identical_to_v2_and_layout(): | |
| _need_pack() | |
| for r, s in (("manual", "test"), ("engaged", "val")): | |
| A = np.load(V2P_DIR / f"kin_causal_{r}_{s}.npz", allow_pickle=True); B = np.load(V2_DIR / f"kin_causal_{r}_{s}.npz", allow_pickle=True) | |
| for k in ("uid", "t", "row_ok", "K_c", "Kvalid", "K0valid", "S_c", "S0_c", "E_c", "EM_c", "E0_c", "channels"): | |
| assert np.array_equal(A[k], B[k], equal_nan=True) if A[k].dtype.kind in "fc" else np.array_equal(A[k], B[k]), (r, s, k) | |
| assert A["K0_c"].shape == (len(A["t"]), 14) and np.array_equal(A["K0_c"][:, :11], B["K0_c"], equal_nan=True) | |
| assert [str(c) for c in A["channels_k0"]] == ds.KIN_CHANNELS_V2P and [str(c) for c in A["channels"]] == ds.KIN_CHANNELS | |
| assert A["HD0_c"].shape == A["PD0_c"].shape == (len(A["t"]),) | |
| Ah = np.load(V2P_DIR / f"kinhist_causal_{r}_{s}.npz", allow_pickle=True); Bh = np.load(V2_DIR / f"kinhist_causal_{r}_{s}.npz", allow_pickle=True) | |
| assert Ah["Kh_c"].shape[2] == 14 and Ah["Kh8_c"].shape[2] == 14 and [str(c) for c in Ah["channels"]] == ds.KIN_CHANNELS_V2P | |
| assert np.array_equal(Ah["Kh_c"][:, :, :11], Bh["Kh_c"], equal_nan=True) and np.array_equal(Ah["Kh8_c"][:, :, :11], Bh["Kh8_c"], equal_nan=True) | |
| assert np.array_equal(Ah["Khvalid"], Bh["Khvalid"]) and np.array_equal(Ah["Kh8valid"], Bh["Kh8valid"]) and np.array_equal(Ah["row_ok"], Bh["row_ok"]) | |
| ok = A["row_ok"] & A["K0valid"] | |
| assert np.isfinite(A["K0_c"][ok, 12]).mean() > 0.95 and np.isfinite(A["K0_c"][ok, 11]).mean() > 0.9 # proxies defined on nearly every valid anchor | |
| pd_ = A["PD0_c"][ok]; assert np.nanmin(pd_) >= 0 and np.nanmax(pd_) <= V2P.PD_CAP | |
| def test_built_pack_loader_and_kp_block(): | |
| _need_pack() | |
| ds.set_kin_source("causal_routes", "v2p") | |
| try: | |
| raw = ds.load_regimes(["manual"], "val", frozenset({"Kh", "kin"}), 3000) | |
| assert raw["Kh"].shape[1:] == (40, 14) and raw["K"].shape[1:] == (25, 11) and raw["S"].shape[1:] == (25, 2) | |
| X, names = WFX.kp_block("manual", "val", np.arange(3000)) | |
| assert X.shape == (3000, 5) and names == WFX.KP_NAMES and set(np.unique(X[:, 4])) <= {0.0, 1.0} | |
| # dwell / K0 blocks of the paper recipe still read the 11-channel contract from the v2p pack | |
| Xd, nd = WFX.dwell_block("manual", "val", np.arange(200)); assert Xd.shape == (200, 11) and nd == WFX.DWELL_NAMES | |
| finally: | |
| ds.set_kin_source("cwm2", "v1") | |
Xet Storage Details
- Size:
- 13 kB
- Xet hash:
- d411bc68b8785de40eba4a2910ae4dec79a43c3b4571fb31094e473560aadc58
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.