Buckets:
| """Layout contract of cwm3/results/targets_kin_causal/{kin,kinhist}_causal_{regime}_{split}.npz (build_kin_causal.py): | |
| row alignment with s12 targets, shapes, NaN policy outside row_ok, interpolation parity, event/mask consistency.""" | |
| import json | |
| from pathlib import Path | |
| import numpy as np, pytest | |
| ROOT = Path(__file__).resolve().parents[1] | |
| KIN = ROOT / "results" / "targets_kin_causal" | |
| S12 = Path("/home/USER/s12_wm/results") | |
| CLIP_MAP = Path("/home/USER/cwm2/private/clip_map_full.json") | |
| CASES = [(r, s) for r in ("manual", "engaged") for s in ("train", "val", "test")] | |
| def _load(reg, sp): | |
| p = KIN / f"kin_causal_{reg}_{sp}.npz" | |
| if not p.exists(): pytest.skip("targets not built yet") | |
| return np.load(p, allow_pickle=True), np.load(KIN / f"kinhist_causal_{reg}_{sp}.npz", allow_pickle=True) | |
| def test_alignment_and_shapes(reg, sp): | |
| k, h = _load(reg, sp); s = np.load(S12 / f"targets_{reg}_{sp}.npz", allow_pickle=True); N = len(s["t"]) | |
| assert np.array_equal(k["uid"], s["uid"]) and np.allclose(k["t"], s["t"]) and np.array_equal(h["uid"], s["uid"]) | |
| assert k["K_c"].shape == (N, 25, 11) and k["K_c"].dtype == np.float16 and h["Kh_c"].shape == (N, 20, 11) | |
| assert k["S_c"].shape == (N, 25, 2) and k["E_c"].shape == (N, 25, 3) and k["EM_c"].shape == (N, 25, 3) and k["K0_c"].shape == (N, 11) | |
| assert k["row_ok"].shape == (N,) and k["clip_id"].shape == (N,) and k["dt_anchor"].shape == (N,) and k["interp"].shape == (N, 25) and h["interp_h"].shape == (N, 20) | |
| assert list(k["channels"]) == ["head_cx", "head_cy", "Lwri_x", "Lwri_y", "Rwri_x", "Rwri_y", "head_turn", "head_turn_dev", "Lhand_rest_dist", "Rhand_rest_dist", "min_hand_rest_dist"] | |
| def test_nan_policy_outside_windows(reg, sp): | |
| k, h = _load(reg, sp); no = ~k["row_ok"] | |
| assert (k["clip_id"][no] == "").all() and np.isnan(k["dt_anchor"][no]).all() | |
| assert np.isnan(k["K_c"][no].astype(np.float32)).all() and np.isnan(h["Kh_c"][no].astype(np.float32)).all() and np.isnan(k["S_c"][no]).all() | |
| assert not k["Kvalid"][no].any() and not k["EM_c"][no].any() and not k["K0valid"][no].any() and not h["Khvalid"][no].any() | |
| def test_window_and_interp_rules(reg, sp): | |
| k, h = _load(reg, sp); R = np.flatnonzero(k["row_ok"]) | |
| if not len(R): pytest.skip("no coverage on this split") | |
| dt = k["dt_anchor"][R]; assert (dt >= -1 - 1e-6).all() and (dt <= 5 + 1e-6).all() | |
| assert (k["clip_id"][R] != "").all() | |
| # anchor grid parity: all 25 future / 20 history steps share the anchor's parity (0.2 s steps on a 0.1 s grid) | |
| assert np.array_equal(k["interp"][R], np.repeat(k["interp0"][R][:, None], 25, 1)) and np.array_equal(h["interp_h"][R], np.repeat(k["interp0"][R][:, None], 20, 1)) | |
| cm = json.load(open(CLIP_MAP)); sub = R[:200] | |
| t0 = np.array([cm[c]["t0"] for c in k["clip_id"][sub]]); assert np.allclose(k["t"][sub] - t0, dt[:200], atol=1e-6) | |
| assert np.array_equal(k["interp0"][sub], (np.round((k["t"][sub] - (t0 - 5)) * 10).astype(int) % 2) == 1) | |
| def test_validity_and_events(reg, sp): | |
| k, _ = _load(reg, sp); R = np.flatnonzero(k["row_ok"]) | |
| if not len(R): pytest.skip("no coverage on this split") | |
| K = k["K_c"][R].astype(np.float32); V = k["Kvalid"][R] | |
| assert np.isfinite(K[V][:, [0, 2, 4, 6]]).all(), "Kvalid cells carry head/wrists/head_turn" | |
| S = k["S_c"][R]; E, EM = k["E_c"][R], k["EM_c"][R] | |
| assert not (E.astype(bool) & ~EM).any(), "onsets only where the mask is valid" | |
| assert set(np.unique(S[np.isfinite(S)]).tolist()) <= {0.0, 1.0} | |
| hon = S[:, :, 0]; md = K[:, :, 10]; m = np.isfinite(hon) & np.isfinite(md) | |
| assert np.array_equal(hon[m] > 0.5, md[m] < 0.10), "hands_on == min_hand_rest_dist < 0.10" | |
| mc = np.load(S12 / f"targets_{reg}_{sp}.npz", allow_pickle=True)["M"].astype(bool)[R] | |
| assert not (V & ~mc).any() and not (EM & ~mc[:, :, None]).any(), "s12 censoring applied" | |
Xet Storage Details
- Size:
- 4.01 kB
- Xet hash:
- 6f73b7f2cb39f05b7b827cc6708ef77a6366059864a1da1c01dd3ee199108c8e
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.