Buckets:
| """Unit tests for data/build_kin_causal_v4.py (hands_on / hand-event label v4): synthetic arrays, fast. | |
| Data-dependent checks (inputs bit-identical to v2 on built files, S0 causal recomputation from the pose) 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_v4 as V4 # noqa: E402 | |
| import build_kin_causal_v3 as V3 # noqa: E402 | |
| import build_kin_causal as BK # noqa: E402 | |
| import wheel_ellipse as WE # noqa: E402 | |
| V4_DIR = Path(os.environ.get("KIN_V4_DIR", ROOT / "results" / "targets_kin_causal_routes_v4")) | |
| V2_DIR = ROOT / "results" / "targets_kin_causal_routes_v2" | |
| rng = np.random.default_rng(0) | |
| G = (1.1, 0.7, 0.25, 0.15, 0.4) | |
| def _route(n, manual, seed=0, noise=0.006, off_frames=()): | |
| """Synthetic route: both wrists slide along the rim of G on manual frames (left wrist 120..260 deg, right wrist -60..80 deg, one sweep per | |
| 100 frames); on non-manual frames the wrists rest far from the wheel (lap). off_frames: manual frames where both hands leave the wheel.""" | |
| r = np.random.default_rng(seed); q = np.full((n, 5), np.nan, np.float32); k = np.arange(n) | |
| phL = 190 + 70 * np.sin(2 * np.pi * k / 100); phR = 10 + 70 * np.cos(2 * np.pi * k / 100) | |
| xL, yL = WE.ellipse_points(G, phL); xR, yR = WE.ellipse_points(G, phR) | |
| q[:, 0] = xL + r.normal(0, noise, n); q[:, 1] = yL + r.normal(0, noise, n); q[:, 2] = xR + r.normal(0, noise, n); q[:, 3] = yR + r.normal(0, noise, n) | |
| lap = ~manual; lap[list(off_frames)] = True | |
| q[lap, 0] = 1.0 + r.normal(0, 0.01, lap.sum()); q[lap, 1] = 0.95 + r.normal(0, 0.01, lap.sum()); q[lap, 2] = 1.2 + r.normal(0, 0.01, lap.sum()); q[lap, 3] = 0.95 + r.normal(0, 0.01, lap.sum()) | |
| return q | |
| # ------------------------------------------------------------------------------------------------------- ellipse series | |
| def test_causal_ellipse_series_prefix_invariance_and_manual_only(): | |
| n = 1500; manual = np.ones(n, bool); manual[900:] = False # engaged (lap) from frame 900 | |
| q = _route(n, manual) | |
| ell, fits, pL, pR, prev = V4.causal_ellipse_series(np.zeros((0, 2)), np.zeros((0, 2)), q, manual, V4.device_rng("DEV")) | |
| assert ell.shape == (n, 5) and ell.dtype == np.float32 and fits.shape[1] == len(V4.FIT_COLS) | |
| # undefined until >= 200 points are available (100 manual frames x 2 wrists), defined afterwards, held constant on engaged frames | |
| assert np.isnan(ell[:99]).all() and np.isfinite(ell[200:]).all() | |
| assert np.array_equal(ell[900:], np.repeat(ell[899:900], n - 900, 0)), "engaged frames never change the ellipse" | |
| first = int(np.flatnonzero(np.isfinite(ell[:, 0]))[0]); assert first in fits[:, 0].astype(int) and first % V4.REFIT_FRAMES == 49 or first == 99 | |
| g = ell[-1].astype(np.float64); assert abs(g[0] - G[0]) < 0.01 and abs(g[1] - G[1]) < 0.01 and abs(g[2] - G[2]) < 0.01 and abs(g[3] - G[3]) < 0.01 | |
| # refit cadence: refits happen every REFIT_FRAMES new manual frames (frame index of the k-th refit = k*REFIT_FRAMES - 1 while manual) | |
| fr = fits[:, 0].astype(int); assert fr[0] == V4.REFIT_FRAMES - 1 and np.array_equal(np.diff(fr[:10]), np.full(9, V4.REFIT_FRAMES)) | |
| # gate columns consistent: passed <=> n_inlier >= 200 and coverage >= 90 | |
| ok = (fits[:, 2] >= WE.N_MIN_PTS) & (fits[:, 3] >= WE.COV_MIN_DEG) & np.isfinite(fits[:, 7]); assert np.array_equal(ok, fits[:, 6] > 0.5) | |
| # PREFIX INVARIANCE (causality): the series on frames < K is identical when the route is truncated at K | |
| for K in (120, 351, 700, 949): | |
| ell_p, fits_p, _, _, _ = V4.causal_ellipse_series(np.zeros((0, 2)), np.zeros((0, 2)), q[:K], manual[:K], V4.device_rng("DEV")) | |
| assert np.array_equal(ell_p, ell[:K], equal_nan=True), K | |
| assert np.array_equal(fits_p, fits[fits[:, 0] < K], equal_nan=True), K | |
| # the pool returned is the most recent POOL_CAP points per wrist and is what the next route starts from (pool-initialised fit at frame 0) | |
| assert len(pL) == min(V4.POOL_CAP, manual.sum()) and len(pR) == len(pL) | |
| ell2, fits2, _, _, _ = V4.causal_ellipse_series(pL, pR, q, manual, V4.device_rng("DEV"), prev=prev) | |
| assert fits2[0, 0] == 0 and np.isfinite(ell2[0]).all(), "with an earlier-route pool the ellipse is defined from the first frame" | |
| # a route whose future frames are altered leaves the past untouched (explicit non-lookahead check) | |
| q2 = q.copy(); q2[1000:] = np.nan | |
| ell3, _, _, _, _ = V4.causal_ellipse_series(np.zeros((0, 2)), np.zeros((0, 2)), q2, manual, V4.device_rng("DEV")); assert np.array_equal(ell3[:1000], ell[:1000], equal_nan=True) | |
| def test_ellipse_series_undefined_when_gate_fails(): | |
| n = 800; manual = np.ones(n, bool); r = np.random.default_rng(1); q = np.full((n, 5), np.nan, np.float32) | |
| # both wrists sit in two fixed grip clusters (no sliding): < 90 deg of arc -> the fit must stay undefined (NaN), never a spurious ellipse | |
| xL, yL = WE.ellipse_points(G, 180 + r.uniform(-8, 8, n)); xR, yR = WE.ellipse_points(G, r.uniform(-8, 8, n)) | |
| q[:, 0] = xL + r.normal(0, 0.005, n); q[:, 1] = yL + r.normal(0, 0.005, n); q[:, 2] = xR + r.normal(0, 0.005, n); q[:, 3] = yR + r.normal(0, 0.005, n) | |
| ell, fits, _, _, _ = V4.causal_ellipse_series(np.zeros((0, 2)), np.zeros((0, 2)), q, manual, V4.device_rng("X")) | |
| assert np.isnan(ell).all() and len(fits) > 0 and not (fits[:, 6] > 0.5).any() | |
| d = V4.wheel_signed_distance(q, ell); assert np.isnan(d).all() | |
| def test_wheel_signed_distance_min_over_wrists_and_sign(): | |
| ell = np.tile(np.asarray(G, np.float32), (4, 1)); ell[3] = np.nan | |
| c, s = np.cos(G[4]), np.sin(G[4]); xv, yv = WE.ellipse_points(G, [0.0]) | |
| q = np.array([[G[0], G[1], xv[0] + 0.1 * c, yv[0] + 0.1 * s, 0], # L at the centre (-b), R 0.1 outside -> min = -b | |
| [np.nan, np.nan, xv[0] + 0.1 * c, yv[0] + 0.1 * s, 0], # L undefined -> R | |
| [xv[0] + 0.02 * c, yv[0] + 0.02 * s, xv[0] + 0.1 * c, yv[0] + 0.1 * s, 0], | |
| [G[0], G[1], G[0], G[1], 0]], np.float32) # ellipse undefined | |
| d = V4.wheel_signed_distance(q, ell) | |
| assert d[0] == pytest.approx(-G[3], abs=1e-6) and d[1] == pytest.approx(0.1, abs=1e-6) and d[2] == pytest.approx(0.02, abs=1e-6) and np.isnan(d[3]) | |
| # ----------------------------------------------------------------------------------------------------- hysteresis / events | |
| def test_hysteresis_v4_thresholds_and_reset(): | |
| d = np.array([0.20, 0.06, 0.03, 0.06, 0.079, 0.081, 0.06, 0.039, -0.02, 0.081, 0.05], np.float32) | |
| s, held = V4.hyst_hands_v4(d) | |
| # 0.20 OFF; 0.06 dead band keeps OFF; 0.03 ON; held; held; 0.081 OFF; held; 0.039 ON; -0.02 (inside) ON; 0.081 OFF; held | |
| assert s.tolist() == [0, 0, 1, 1, 1, 0, 0, 1, 1, 0, 0] and np.array_equal(held, s) | |
| s2, _ = V4.hyst_hands_v4(np.array([0.02, 0.04, 0.08, 0.02, 0.08, 0.04], np.float32)); assert s2.tolist() == [1, 1, 1, 1, 1, 1], "d == 0.04 / 0.08 do not switch" | |
| s3, _ = V4.hyst_hands_v4(np.array([0.20, 0.08, 0.04, 0.20, 0.04], np.float32)); assert s3.tolist() == [0, 0, 0, 0, 0] | |
| g = V4.GAP_RESET | |
| d4 = np.r_[[0.0] * 3, [np.nan] * (g + 1), [0.06, 0.06, 0.03]].astype(np.float32); s4, h4 = V4.hyst_hands_v4(d4) | |
| assert np.isnan(h4[3 + g]) and np.isnan(s4[4 + g]) and np.isnan(s4[5 + g]) and s4[6 + g] == 1, "> 1 s gap resets: undecided in the dead band" | |
| d5 = np.r_[[0.0] * 3, [np.nan] * g, [0.06]].astype(np.float32); s5, h5 = V4.hyst_hands_v4(d5); assert s5[3 + g] == 1 and (h5[3:3 + g] == 1).all(), "gap of exactly 1 s keeps the state" | |
| # negative (inside the ellipse) is ON; the deep interior is ON too | |
| s6, _ = V4.hyst_hands_v4(np.array([-0.3, -0.1, 0.0], np.float32)); assert s6.tolist() == [1, 1, 1] | |
| def test_persistence_events_reuse_v3_machinery(): | |
| def ser(*segs): return np.concatenate([np.full(n, v, np.float32) for n, v in segs]) | |
| d = ser((30, 0.20), (3, 0.0), (30, 0.20), (12, 0.0), (30, 0.20)); s, h = V4.hyst_hands_v4(d); ev_r, ev_l, unk = V3.confirmed_transitions(s, h) | |
| assert ev_r.sum() == 1 and int(np.flatnonzero(ev_r)[0]) == 63 and ev_l.sum() == 1 and int(np.flatnonzero(ev_l)[0]) == 75 and not unk.any(), "0.3 s blip rejected, 1.2 s return + leave accepted" | |
| d2 = ser((30, 0.20), (9, 0.0), (30, 0.20)); s2, h2 = V4.hyst_hands_v4(d2); r2, l2, _ = V3.confirmed_transitions(s2, h2); assert r2.sum() == 0 and l2.sum() == 0 | |
| d3 = ser((30, 0.20), (10, 0.0), (30, 0.20)); s3, h3 = V4.hyst_hands_v4(d3); r3, l3, _ = V3.confirmed_transitions(s3, h3); assert r3.sum() == 1 and l3.sum() == 1 | |
| # rim sliding: d stays inside the ON band (< 0.08) while the hand moves along the rim -> NO event (the v3 failure mode) | |
| d4 = ser((30, -0.02), (20, 0.06), (30, -0.02)); s4, h4 = V4.hyst_hands_v4(d4); r4, l4, _ = V3.confirmed_transitions(s4, h4); assert r4.sum() == 0 and l4.sum() == 0 and (s4 == 1).all() | |
| def test_end_to_end_synthetic_route_rim_sliding_gives_no_event_but_lap_excursion_does(): | |
| n = 3000; manual = np.ones(n, bool) | |
| q = _route(n, manual, off_frames=range(2000, 2040)) # both hands in the lap for 4 s, then back on the rim | |
| ell, fits, _, _, _ = V4.causal_ellipse_series(np.zeros((0, 2)), np.zeros((0, 2)), q, manual, V4.device_rng("E2E")) | |
| d = V4.wheel_signed_distance(q, ell); s, h = V4.hyst_hands_v4(d); ev_r, ev_l, unk = V3.confirmed_transitions(s, h) | |
| assert np.isfinite(ell[300:]).all() | |
| assert (np.abs(d[300:2000]) < 0.04).all() and (s[300:2000] == 1).all(), "sliding along the rim never leaves the ON band" | |
| assert ev_l.sum() == 1 and int(np.flatnonzero(ev_l)[0]) == 2000 and ev_r.sum() == 1 and int(np.flatnonzero(ev_r)[0]) == 2040 | |
| # v3's point reference on the same data DOES chatter (documents the mechanism): the sliding wrist leaves the point by > 0.12 | |
| wref = V3.wheel_reference([np.zeros(0, np.float32)] * 4, q, manual); d3 = V3.wheel_distance(q, wref); s3, h3 = V3.hyst_hands(d3); r3, l3, _ = V3.confirmed_transitions(s3, h3) | |
| assert (np.nanmax(d3[300:2000]) > 0.12) and (r3.sum() + l3.sum() > 2) | |
| def test_hand_events_on_grid_and_S0_causal(): | |
| N, H = 3, 25; S0 = np.array([0, 0, 1], np.float32); S = np.zeros((N, H), np.float32); S[0, 3:] = 1; S[1, 3:] = 1; S[2] = 1 | |
| ev = np.zeros((N, H, 2), bool); ev[:2, 3, 0] = True; unk = np.zeros((N, H), bool); unk[1, 3] = True; Mc = np.ones((N, H), bool) | |
| E, EM = V3.hand_events_on_grid(S0, S, ev, unk, Mc) | |
| assert E[0, 3] == 1 and E[0].sum() == 1 and E[1, 3] == 0 and not EM[1, 3] and E[2].sum() == 0 | |
| # S0 at the anchor is the raw causal state at the anchor frame (no lookahead): a later ON does not change an earlier NaN/OFF | |
| d = np.r_[[0.20] * 5, [np.nan] * 12, [0.06] * 3, [0.0] * 5].astype(np.float32); s, _ = V4.hyst_hands_v4(d) | |
| assert s[4] == 0 and np.isnan(s[17]) and np.isnan(s[19]) and s[20] == 1 | |
| # ------------------------------------------------------------------------------------------------------------ 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_v4.py", ROOT / "data" / "wheel_ellipse.py", Path(__file__), ROOT / "tests" / "test_wheel_ellipse.py", ROOT / "eval" / "kin_v4_build_stats.py", | |
| ROOT / "tools" / "review_tool" / "build_relabel_v4.py", ROOT / "tools" / "review_tool" / "apply_review_v4.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 V4.HYST_ON == 0.04 and V4.HYST_OFF == 0.08 and V4.PERSIST_S == 1.0 and V4.P_FRAMES == 10 and V4.GAP_RESET == 10 and V4.LABEL_VERSION == "v4" | |
| assert WE.N_MIN_PTS == 200 and WE.COV_MIN_DEG == 90.0 and V4.REFIT_FRAMES == 50 and V4.POOL_CAP == 4000 | |
| assert V4.hyst_hands_v4.__code__.co_names and "hyst_hands" in V4.hyst_hands_v4.__code__.co_names, "v4 reuses the v3 hysteresis" | |
| src = (ROOT / "data" / "build_kin_causal_v4.py").read_text() | |
| assert "V3.confirmed_transitions" in src and "V3.hand_events_on_grid" in src and "V3.engaged_flags_for_pose" in src and "V3._link_or_copy" in src | |
| def test_built_files_inputs_identical_to_v2_and_events_subset(reg): | |
| p4 = V4_DIR / f"kin_causal_{reg}_test.npz"; p2 = V2_DIR / f"kin_causal_{reg}_test.npz" | |
| if not (p4.exists() and p2.exists()): pytest.skip("v4 test split not built") | |
| z4 = np.load(p4, allow_pickle=True); z2 = np.load(p2, allow_pickle=True) | |
| assert np.array_equal(z4["uid"], z2["uid"]) and np.allclose(z4["t"], z2["t"]) | |
| E, EM = z4["E_c"], z4["EM_c"]; assert not (E.astype(bool) & ~EM).any() | |
| assert str(z4["label_version"]) == "v4" and float(z4["hyst_on"]) == np.float32(0.04) and float(z4["hyst_off"]) == np.float32(0.08) and float(z4["persist_s"]) == 1.0 | |
| for key in ("K_c", "K0_c", "Kvalid", "Kframe", "K0valid", "row_ok", "route_decoded"): | |
| assert z4[key].dtype == z2[key].dtype and z4[key].shape == z2[key].shape, key | |
| assert np.array_equal(z4[key], z2[key], equal_nan=True) if z4[key].dtype.kind == "f" else np.array_equal(z4[key], z2[key]), f"{key} differs from v2 (inputs must be identical)" | |
| assert np.array_equal(z4["S_c"][:, :, 1], z2["S_c"][:, :, 1], equal_nan=True) and np.array_equal(z4["E_c"][:, :, 2], z2["E_c"][:, :, 2]) and np.array_equal(z4["EM_c"][:, :, 2], z2["EM_c"][:, :, 2]) | |
| S = z4["S_c"][:, :, 0]; S0 = z4["S0_c"][:, 0]; Sp = np.concatenate([S0[:, None], S[:, :-1]], 1) | |
| er = z4["E_c"][:, :, 0].astype(bool); assert (Sp[er] == 0).all() and (S[er] == 1).all() | |
| el = z4["E_c"][:, :, 1].astype(bool); assert (Sp[el] == 1).all() and (S[el] == 0).all() | |
| assert set(np.unique(S[np.isfinite(S)]).tolist()) <= {0.0, 1.0} | |
| assert z4["ellipse0_c"].shape == (len(S0), 5) and (np.isfinite(z4["ellipse0_c"]).all(1) == z4["wheel_ref_defined"]).all() | |
| h4 = V4_DIR / f"kinhist_causal_{reg}_test.npz"; h2 = V2_DIR / f"kinhist_causal_{reg}_test.npz" | |
| assert h4.exists() and (os.path.samefile(h4, h2) or h4.read_bytes() == h2.read_bytes()), "kinhist byte-identical to v2 (hard link)" | |
| def test_built_sample_route_recomputed_from_pose_matches_v4_and_v2(): | |
| """Recompute one TEST route from the pose npz: K0_c (inputs, v2 reference) and S0_c hands_on (v4 causal state from the stored ellipse series).""" | |
| p4 = V4_DIR / "kin_causal_engaged_test.npz" | |
| if not (p4.exists() and (V4_DIR / "route_refs.npz").exists()): pytest.skip("v4 test split not built") | |
| z4 = np.load(p4, allow_pickle=True); rr = np.load(V4_DIR / "route_refs.npz", allow_pickle=True) | |
| R = np.flatnonzero(z4["row_ok"] & z4["wheel_ref_defined"]); u = z4["uid"][R[len(R) // 2]]; rows = np.flatnonzero(z4["uid"] == u) | |
| pp = V4.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}"]; ell = rr[f"ell__{u}"] | |
| G_ = BK.clip_series(raw, ref).astype(np.float16).astype(np.float32); d = V4.wheel_signed_distance(raw["q"], ell); d[~F] = np.nan | |
| assert np.allclose(d, rr[f"d4__{u}"], equal_nan=True, atol=1e-6) | |
| 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, _ = V4.hyst_hands_v4(DD) | |
| ka = np.round(z4["t"][rows] * 10).astype(np.int64) - k0; ok = (ka >= 0) & (ka < M) | |
| assert np.array_equal(z4["K0_c"][rows][ok], GG[ka[ok]], equal_nan=True), "K0_c (input) recomputed from pose with the v2 reference" | |
| assert np.array_equal(z4["S0_c"][rows, 0][ok], s_raw[ka[ok]], equal_nan=True), "S0_c hands_on recomputed causally" | |
| assert np.allclose(z4["dwheel0_c"][rows][ok], DD[ka[ok]], equal_nan=True) | |
| # the stored ellipse only changes at refit frames that are manual+moving (or frame 0), never on engaged frames | |
| eng, m = V3.engaged_flags_for_pose(u, t); mm = z["moving"].astype(bool) & m & ~eng | |
| chg = np.flatnonzero(~np.all((ell[1:] == ell[:-1]) | (np.isnan(ell[1:]) & np.isnan(ell[:-1])), 1)) + 1 | |
| fits = rr[f"fits__{u}"]; assert set(chg.tolist()) <= set(fits[:, 0].astype(int).tolist()), "the ellipse changes only at recorded refits" | |
| assert all(mm[int(f)] or int(f) == 0 for f in fits[:, 0]), "refits happen on manual moving frames (or at route start from the earlier-route pool)" | |
| assert int(rr[f"n_manual_moving__{u}"]) == int(mm.sum()) | |
Xet Storage Details
- Size:
- 16.9 kB
- Xet hash:
- 25a0936f1f1ce9b56d25fa48560af4e0dc386009e53548b66885886bdcd3adcb
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.