Buckets:
| """Unit tests for data/build_kin_causal_v3.py (hands_on / hand-event label v3): synthetic arrays, fast. | |
| Data-dependent checks (inputs bit-identical to v2 on a sample route, E_c <= EM_c on built files) skip when the files are absent.""" | |
| import os, re, sys | |
| from pathlib import Path | |
| import numpy as np, pytest | |
| ROOT = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(ROOT / "data")) | |
| import build_kin_causal_v3 as V3 # noqa: E402 | |
| import build_kin_causal as BK # noqa: E402 | |
| V3_DIR = Path(os.environ.get("KIN_V3_DIR", ROOT / "results" / "targets_kin_causal_routes_v3")) | |
| V2_DIR = ROOT / "results" / "targets_kin_causal_routes_v2" | |
| rng = np.random.default_rng(0) | |
| # ------------------------------------------------------------------------------------------------------------- wheel reference | |
| def test_wheel_reference_uses_only_manual_moving_frames_up_to_t(): | |
| n = 400; q = rng.normal(0.5, 0.05, (n, 4)).astype(np.float32) | |
| moving = np.ones(n, bool); moving[50:80] = False | |
| engaged = np.zeros(n, bool); engaged[200:] = True # second half engaged | |
| q[engaged] += 5.0 # engaged frames far away: must never enter the reference | |
| q[10, :] = np.nan # an invalid frame | |
| manual_moving = moving & ~engaged | |
| pools = [np.zeros(0, np.float32) for _ in range(4)] | |
| ref = V3.wheel_reference(pools, q, manual_moving) | |
| assert ref.shape == (n, 4) and ref.dtype == np.float32 | |
| # brute force: median over manual moving finite values <= t, NaN while count < N_MIN | |
| for k in (0, 5, 24, 25, 26, 100, 199, 200, 250, 399): | |
| sel = manual_moving[: k + 1] & np.isfinite(q[: k + 1, 0]) | |
| if sel.sum() < BK.N_MIN: assert np.isnan(ref[k]).all(), k | |
| else: assert np.allclose(ref[k], np.median(q[: k + 1][sel], 0), atol=1e-6), k | |
| # engaged frames never change the reference (the reference is frozen once only engaged frames follow) | |
| assert np.allclose(ref[200:], ref[199], atol=0) and np.all(np.abs(ref[199] - 0.5) < 0.1) | |
| # prefix property (causality): the reference at t does not depend on frames > t | |
| ref_pre = V3.wheel_reference([np.zeros(0, np.float32) for _ in range(4)], q[:150], manual_moving[:150]) | |
| assert np.array_equal(ref_pre, ref[:150], equal_nan=True) | |
| # a device pool (earlier routes) initialises the reference from frame 0 | |
| pool = [np.full(30, 0.7, np.float32) for _ in range(4)] | |
| ref_p = V3.wheel_reference(pool, q, manual_moving) | |
| assert np.isfinite(ref_p[0]).all() and np.allclose(ref_p[0], 0.7) and np.isfinite(ref_p).all() | |
| def test_engaged_flag_alignment_by_integer_key(tmp_path): | |
| t_grid = np.arange(0, 100, 0.1); eng = np.zeros(len(t_grid), bool); eng[300:600] = True # engaged 30.0..59.9 s | |
| np.savez(tmp_path / "ROUTE.npz", t=t_grid, G=np.zeros((len(t_grid), 32), np.float32), engaged=eng) | |
| t_pose = np.arange(20.6, 70.0, 0.1, dtype=np.float32) # float32 pose grid, offset start | |
| e, m = V3.engaged_flags_for_pose("ROUTE", t_pose, tmp_path) | |
| kt = np.round(t_pose.astype(np.float64) * 10).astype(int) | |
| assert m.all() and np.array_equal(e, (kt >= 300) & (kt < 600)) | |
| t_pose2 = np.arange(90.0, 110.0, 0.1, dtype=np.float32) # partly outside the grid10 range | |
| e2, m2 = V3.engaged_flags_for_pose("ROUTE", t_pose2, tmp_path) | |
| assert m2.sum() == int((np.round(t_pose2.astype(np.float64) * 10) < 1000).sum()) and not e2[~m2].any() | |
| e3, m3 = V3.engaged_flags_for_pose("MISSING", t_pose, tmp_path); assert not m3.any() and not e3.any() | |
| def test_wheel_distance_min_over_wrists_ignores_nan_wrist(): | |
| q = np.array([[0.0, 0.0, 1.0, 1.0], [np.nan, np.nan, 1.0, 1.0], [np.nan, np.nan, np.nan, np.nan]], np.float32) | |
| wref = np.array([[0.03, 0.04, 1.0, 1.5]] * 3, np.float32) | |
| d = V3.wheel_distance(q, wref) | |
| assert np.isclose(d[0], 0.05) and np.isclose(d[1], 0.5) and np.isnan(d[2]) | |
| # ----------------------------------------------------------------------------------------------------------------- hysteresis | |
| def test_hysteresis_bands_and_gap_reset(): | |
| d = np.array([0.20, 0.10, 0.07, 0.10, 0.11, 0.13, 0.10, 0.079, 0.121, 0.081], np.float32) | |
| s, held = V3.hyst_hands(d) | |
| # 0.20 -> OFF; 0.10 dead band keeps OFF; 0.07 -> ON; held through 0.10/0.11; 0.13 -> OFF; held; 0.079 -> ON; 0.121 -> OFF; held | |
| exp = [0, 0, 1, 1, 1, 0, 0, 1, 0, 0] | |
| assert np.array_equal(s, np.array(exp, np.float32), equal_nan=True) and np.array_equal(held, s, equal_nan=True) | |
| # exact thresholds: d == 0.08 and d == 0.12 do not switch | |
| s2, _ = V3.hyst_hands(np.array([0.05, 0.08, 0.12, 0.05, 0.12, 0.08], np.float32)); assert s2.tolist() == [1, 1, 1, 1, 1, 1] | |
| s3, _ = V3.hyst_hands(np.array([0.20, 0.12, 0.08, 0.20, 0.08], np.float32)); assert s3.tolist() == [0, 0, 0, 0, 0] | |
| # gap of exactly GAP_RESET missing frames keeps the state (held carries it); GAP_RESET+1 resets (undecided in the dead band) | |
| g = V3.GAP_RESET | |
| d4 = np.r_[[0.05] * 3, [np.nan] * g, [0.10, 0.10, 0.20]].astype(np.float32); s4, h4 = V3.hyst_hands(d4) | |
| assert np.isnan(s4[3:3 + g]).all() and (h4[3:3 + g] == 1).all() and s4[3 + g] == 1 and s4[4 + g] == 1 and s4[5 + g] == 0 | |
| d5 = np.r_[[0.05] * 3, [np.nan] * (g + 1), [0.10, 0.10, 0.07]].astype(np.float32); s5, h5 = V3.hyst_hands(d5) | |
| assert np.isnan(h5[3 + g]) and np.isnan(s5[4 + g]) and np.isnan(s5[5 + g]) and s5[6 + g] == 1 | |
| # undecided start: NaN until the first decisive value | |
| s6, _ = V3.hyst_hands(np.array([0.09, 0.10, 0.11, 0.13, 0.09], np.float32)); assert np.array_equal(s6, [np.nan, np.nan, np.nan, 0, 0], equal_nan=True) | |
| def test_state_is_causal_prefix_property(): | |
| d = rng.uniform(0.0, 0.25, 600).astype(np.float32); d[rng.random(600) < 0.05] = np.nan; d[300:320] = np.nan | |
| s_full, h_full = V3.hyst_hands(d) | |
| for k in (1, 50, 299, 310, 330, 599): | |
| s_pre, h_pre = V3.hyst_hands(d[:k]) | |
| assert np.array_equal(s_pre, s_full[:k], equal_nan=True) and np.array_equal(h_pre, h_full[:k], equal_nan=True), k | |
| # ---------------------------------------------------------------------------------------------------------------- persistence | |
| def _series(*segs): | |
| return np.concatenate([np.full(n, v, np.float32) for n, v in segs]) | |
| def test_persistence_rejects_blip_accepts_1p2s_return(): | |
| # 3.0 s OFF | 0.3 s ON blip | 3.0 s OFF | 1.2 s ON | 3.0 s OFF | |
| d = _series((30, 0.20), (3, 0.05), (30, 0.20), (12, 0.05), (30, 0.20)) | |
| s, h = V3.hyst_hands(d); ev_r, ev_l, unk = V3.confirmed_transitions(s, h) | |
| assert s[30:33].tolist() == [1, 1, 1], "the blip does switch the causal state" | |
| assert ev_r.sum() == 1 and int(np.flatnonzero(ev_r)[0]) == 63, "one confirmed return, stamped at the first ON frame" | |
| assert ev_l.sum() == 1 and int(np.flatnonzero(ev_l)[0]) == 75, "the 1.2 s ON run ends with a confirmed leave" | |
| assert not unk.any() | |
| # 0.9 s return is rejected; exactly 1.0 s (10 frames) is accepted; previous state of only 0.9 s is rejected | |
| d2 = _series((30, 0.20), (9, 0.05), (30, 0.20)); s2, h2 = V3.hyst_hands(d2); r2, l2, _ = V3.confirmed_transitions(s2, h2); assert r2.sum() == 0 and l2.sum() == 0 | |
| d3 = _series((30, 0.20), (10, 0.05), (30, 0.20)); s3, h3 = V3.hyst_hands(d3); r3, l3, _ = V3.confirmed_transitions(s3, h3); assert r3.sum() == 1 and l3.sum() == 1 | |
| d4 = _series((9, 0.20), (12, 0.05), (30, 0.20)); s4, h4 = V3.hyst_hands(d4); r4, l4, u4 = V3.confirmed_transitions(s4, h4) | |
| assert r4.sum() == 0 and l4.sum() == 1 and u4[9], "prev state truncated by the series start -> unknown, not an event" | |
| # dead-band chatter inside a run does not break it (hysteresis holds the state) | |
| d5 = _series((12, 0.20), (5, 0.05), (4, 0.10), (5, 0.05), (12, 0.20)); s5, h5 = V3.hyst_hands(d5); r5, l5, _ = V3.confirmed_transitions(s5, h5) | |
| assert r5.sum() == 1 and l5.sum() == 1 and int(np.flatnonzero(l5)[0]) == 26 | |
| # a short gap (<= GAP_RESET missing frames) inside the run neither breaks nor confirms it | |
| d6 = _series((12, 0.20), (4, 0.05), (5, np.nan), (4, 0.05), (12, 0.20)); s6, h6 = V3.hyst_hands(d6); r6, l6, _ = V3.confirmed_transitions(s6, h6) | |
| assert r6.sum() == 1 and int(np.flatnonzero(r6)[0]) == 12 and l6.sum() == 1 and int(np.flatnonzero(l6)[0]) == 25 | |
| def test_confirmed_transitions_unknown_at_series_end_and_empty(): | |
| d = _series((30, 0.20), (5, 0.05)); s, h = V3.hyst_hands(d); r, l, u = V3.confirmed_transitions(s, h) | |
| assert r.sum() == 0 and u[30] and not u[:30].any() | |
| r0, l0, u0 = V3.confirmed_transitions(np.zeros(1, np.float32), np.zeros(1, np.float32)); assert len(r0) == 1 and not u0.any() | |
| def test_hand_events_on_grid_subset_of_mask_and_v2_rule(): | |
| N, H = 6, 25 | |
| S0 = np.array([0, 0, 0, np.nan, 0, 1], np.float32); S = np.zeros((N, H), np.float32) | |
| S[0, 3:] = 1; S[1, 3:] = 1; S[2, 3:] = 1; S[3, 3:] = 1; S[4, 3:] = 1; S[5, :] = 1 | |
| S[2, 2] = np.nan # unknown state at the step before the event | |
| ev = np.zeros((N, H, 2), bool); ev[:5, 3, 0] = True # confirmed 10 Hz return at the frame of step h=3 (rows 0..4) | |
| unk = np.zeros((N, H), bool); unk[1, 3] = True # row 1: transition unconfirmable | |
| Mc = np.ones((N, H), bool); Mc[4, 3] = False; S[~Mc] = np.nan # row 4: censored by s12 | |
| E, EM = V3.hand_events_on_grid(S0, S, ev, unk, Mc) | |
| assert E.dtype == np.uint8 and not (E.astype(bool) & ~EM).any() and not (EM & ~Mc).any() | |
| assert E[0, 3] == 1 and E[0].sum() == 1 | |
| assert E[1, 3] == 0 and not EM[1, 3], "unconfirmable transition -> label unknown" | |
| assert E[2, 3] == 0 and not EM[2, 3] and EM[2, 4], "v2 rule: state at h-1 unknown -> step unobservable" | |
| assert not EM[3, 0] and EM[3, 1], "S0 NaN -> first step unobservable" | |
| assert not EM[4, 3] and E[4, 3] == 0 | |
| assert E[5].sum() == 0 and EM[5].all() | |
| # ------------------------------------------------------------------------------------------------------------ hygiene / data | |
| FORBIDDEN = ["driver_state", "faceProb", "isDistracted", "face_yaw", "face_pitch", "face_roll", "blinkProb", "dmonitoring", "driverState", "awarenessStatus", "distractedPct", "dstate", "leftEyeProb", "rightEyeProb"] | |
| def test_forbidden_tokens_absent(): | |
| files = [ROOT / "data" / "build_kin_causal_v3.py", Path(__file__), ROOT / "eval" / "kin_v3_build_stats.py"] | |
| for f in files: | |
| if not f.exists(): continue | |
| src = f.read_text() | |
| body = src if f != Path(__file__) else re.sub(r"FORBIDDEN = \[.*?\]", "", src, flags=re.S) | |
| for tok in FORBIDDEN: assert tok not in body, (f, tok) | |
| def test_constants_match_preregistration(): | |
| assert V3.HYST_ON == 0.08 and V3.HYST_OFF == 0.12 and V3.PERSIST_S == 1.0 and V3.P_FRAMES == 10 and V3.GAP_RESET == 10 | |
| assert BK.POOL_CAP == 4000 and BK.N_MIN == 25 | |
| assert "expanding_median_moving" in V3.wheel_reference.__code__.co_names, "v3 reuses the v2 expanding median" | |
| def test_built_files_inputs_identical_to_v2_and_events_subset(reg): | |
| p3 = V3_DIR / f"kin_causal_{reg}_test.npz"; p2 = V2_DIR / f"kin_causal_{reg}_test.npz" | |
| if not (p3.exists() and p2.exists()): pytest.skip("v3 test split not built") | |
| z3 = np.load(p3, allow_pickle=True); z2 = np.load(p2, allow_pickle=True) | |
| assert np.array_equal(z3["uid"], z2["uid"]) and np.allclose(z3["t"], z2["t"]) | |
| E, EM = z3["E_c"], z3["EM_c"]; assert not (E.astype(bool) & ~EM).any() | |
| for key in ("row_ok", "route_decoded", "wheel_ref_defined"): assert z3[key].dtype == bool and z3[key].shape == z2["row_ok"].shape | |
| assert float(z3["persist_s"]) == 1.0 and float(z3["hyst_on"]) == np.float32(0.08) and float(z3["hyst_off"]) == np.float32(0.12) and str(z3["label_version"]) == "v3" | |
| for key in ("K_c", "K0_c", "Kvalid", "Kframe", "K0valid", "row_ok", "route_decoded", "channels", "event_names", "state_names"): | |
| assert z3[key].dtype == z2[key].dtype and z3[key].shape == z2[key].shape, key | |
| # inputs identical on a sample route (all rows of the first decoded uid) + head channel unchanged | |
| R = np.flatnonzero(z3["row_ok"]); u = z3["uid"][R[0]]; rows = np.flatnonzero(z3["uid"] == u) | |
| for key in ("K_c", "K0_c"): assert np.array_equal(z3[key][rows], z2[key][rows], equal_nan=True), key | |
| for key in ("Kvalid", "Kframe", "K0valid"): assert np.array_equal(z3[key][rows], z2[key][rows]), key | |
| assert np.array_equal(z3["S_c"][rows, :, 1], z2["S_c"][rows, :, 1], equal_nan=True) and np.array_equal(z3["E_c"][rows, :, 2], z2["E_c"][rows, :, 2]) | |
| # E ==> S transition on the grid (return: S[h-1]=0, S[h]=1), states binary | |
| S = z3["S_c"][rows]; S0 = z3["S0_c"][rows]; Sp = np.concatenate([S0[:, None, 0], S[:, :-1, 0]], 1) | |
| er = z3["E_c"][rows, :, 0].astype(bool); assert (Sp[er] == 0).all() and (S[:, :, 0][er] == 1).all() | |
| el = z3["E_c"][rows, :, 1].astype(bool); assert (Sp[el] == 1).all() and (S[:, :, 0][el] == 0).all() | |
| assert set(np.unique(S[np.isfinite(S)]).tolist()) <= {0.0, 1.0} | |
| # kinhist: byte-identical to v2 (hard link or copy) | |
| h3 = V3_DIR / f"kinhist_causal_{reg}_test.npz"; h2 = V2_DIR / f"kinhist_causal_{reg}_test.npz" | |
| assert h3.exists() and (os.path.samefile(h3, h2) or h3.read_bytes() == h2.read_bytes()) | |
| def test_built_sample_route_recomputed_from_pose_matches_v3_and_v2(): | |
| """Recompute one TEST route from the pose npz with the v3 module and compare K0_c (inputs) and S0_c hands_on (v3 state) with the files.""" | |
| p3 = V3_DIR / "kin_causal_engaged_test.npz" | |
| if not (p3.exists() and (V3_DIR / "route_refs.npz").exists()): pytest.skip("v3 test split not built") | |
| z3 = np.load(p3, allow_pickle=True); rr = np.load(V3_DIR / "route_refs.npz", allow_pickle=True) | |
| R = np.flatnonzero(z3["row_ok"] & z3["wheel_ref_defined"]); u = z3["uid"][R[len(R) // 2]]; rows = np.flatnonzero(z3["uid"] == u) | |
| pp = V3.POSE_PREF / f"{u}.npz" | |
| if not pp.exists(): pp = BK.POSE_R / f"{u}.npz" | |
| z = np.load(pp); t = z["t"].astype(np.float64); F = np.isfinite(z["box"][:, 0]) & np.isfinite(z["kpts"][:, 0, 2].astype(np.float32)) | |
| raw = BK.raw_from_kpts(z["kpts"], F); ref = rr[f"ref__{u}"]; wref = rr[f"wheel_ref__{u}"] | |
| G = BK.clip_series(raw, ref).astype(np.float16).astype(np.float32); d = V3.wheel_distance(raw["q"], wref); d[~F] = np.nan | |
| k = np.round(t * 10).astype(np.int64); k0 = int(k.min()); M = int(k.max()) - k0 + 1 | |
| GG = np.full((M, BK.D), np.nan, np.float32); DD = np.full(M, np.nan, np.float32); GG[k - k0] = G; DD[k - k0] = d | |
| s_raw, _ = V3.hyst_hands(DD) | |
| ka = np.round(z3["t"][rows] * 10).astype(np.int64) - k0; ok = (ka >= 0) & (ka < M) | |
| assert np.array_equal(z3["K0_c"][rows][ok], GG[ka[ok]], equal_nan=True), "K0_c (input) recomputed from pose with the v2 reference" | |
| assert np.array_equal(z3["S0_c"][rows, 0][ok], s_raw[ka[ok]], equal_nan=True), "S0_c hands_on recomputed causally" | |
| assert np.allclose(z3["dwheel0_c"][rows][ok], DD[ka[ok]], equal_nan=True) | |
| # the stored manual-moving frame count matches the grid10 alignment rule | |
| eng, m = V3.engaged_flags_for_pose(u, t); mm = z["moving"].astype(bool) & m & ~eng | |
| assert int(rr[f"n_manual_moving__{u}"]) == int(mm.sum()) | |
| # the wheel reference never moves on engaged frames (only manual moving frames update it) | |
| wd = np.abs(np.diff(wref, axis=0)); upd = np.isfinite(wd).all(1) & (wd > 0).any(1) | |
| assert not (upd & ~mm[1:]).any(), "wheel reference changed on a frame that is not manual+moving" | |
Xet Storage Details
- Size:
- 15.5 kB
- Xet hash:
- 825a2d3e67dd60c276a4960fb2e80a14017b5409e38c1932b47d0fff4897bb4c
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.