Buckets:
| """Unit tests for data/build_kin_causal_v4b.py (hands_on / hand-event label v4b = v4 with the RIM-BAND distance convention): synthetic | |
| arrays, fast. Data-dependent checks (built files: inputs bit-identical to v2, ellipse identical to v4, 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_v4b as V4B # noqa: E402 | |
| 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 | |
| V4B_DIR = Path(os.environ.get("KIN_V4B_DIR", ROOT / "results" / "targets_kin_causal_routes_v4b")) | |
| V4_DIR = ROOT / "results" / "targets_kin_causal_routes_v4" | |
| V2_DIR = ROOT / "results" / "targets_kin_causal_routes_v2" | |
| G = (1.1, 0.7, 0.25, 0.15, 0.4) | |
| def _rim(phi, off=0.0): | |
| """Point at parametric angle phi (deg) on G, displaced by `off` along the outward normal (negative = inward).""" | |
| x, y = WE.ellipse_points(G, [phi]); x, y = float(x[0]), float(y[0]) | |
| if off == 0.0: return x, y | |
| cx, cy, a, b, th = G; p = np.radians(phi); c, s = np.cos(th), np.sin(th) | |
| nu, nv = np.cos(p) / a, np.sin(p) / b # gradient of the canonical form ~ outward normal | |
| nx, ny = nu * c - nv * s, nu * s + nv * c; nn = np.hypot(nx, ny) | |
| return x + off * nx / nn, y + off * ny / nn | |
| def _route(n, manual, seed=0, noise=0.006, interior_frames=(), lap_frames=()): | |
| """Synthetic route: both wrists slide along the rim of G on manual frames (left 120..260 deg, right -60..80 deg, one sweep per 100 frames); | |
| on non-manual frames the wrists rest far from the wheel (lap, outside the ellipse). interior_frames: manual frames on which both | |
| wrists sit at the ellipse CENTRE (deep interior = hands in the lap seen through the wheel in 2-D); lap_frames: both wrists outside.""" | |
| 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(lap_frames)] = True | |
| q[lap, 0] = 1.0 + r.normal(0, 0.01, lap.sum()); q[lap, 1] = 1.2 + r.normal(0, 0.01, lap.sum()); q[lap, 2] = 1.3 + r.normal(0, 0.01, lap.sum()); q[lap, 3] = 1.2 + r.normal(0, 0.01, lap.sum()) | |
| ii = np.zeros(n, bool); ii[list(interior_frames)] = True | |
| q[ii, 0] = G[0] + r.normal(0, 0.005, ii.sum()); q[ii, 1] = G[1] + r.normal(0, 0.005, ii.sum()); q[ii, 2] = G[0] + r.normal(0, 0.005, ii.sum()); q[ii, 3] = G[1] + r.normal(0, 0.005, ii.sum()) | |
| return q | |
| # --------------------------------------------------------------------------------------------------------- distance convention | |
| def test_rim_point_on_deep_interior_off_and_per_wrist_abs(): | |
| far = (2.0, 2.0) | |
| rows = [(*_rim(30.0), *far), # 0: L exactly on the rim, R far -> d_rim ~ 0 (ON) | |
| (G[0], G[1], G[0], G[1]), # 1: both wrists at the centre (deep interior) -> d_rim = b = 0.15 (OFF) | |
| (G[0], G[1], *_rim(200.0, +0.02)), # 2: L at the centre, R 0.02 outside the rim -> d_rim = 0.02 (ON; v4 would say -0.15) | |
| (*_rim(120.0, -0.02), *far), # 3: L 0.02 INSIDE the rim (rim band, inner side) -> d_rim = 0.02 (ON) | |
| (np.nan, np.nan, *_rim(0.0, +0.10)), # 4: L undefined, R 0.10 outside -> d_rim = 0.10 (OFF) | |
| (*_rim(300.0, -0.06), *far), # 5: L 0.06 inside the rim (dead band) -> d_rim = 0.06 (keep) | |
| (G[0], G[1], G[0], G[1])] # 6: ellipse undefined -> NaN | |
| q = np.array([list(r) + [0.0] for r in rows], np.float32); ell = np.tile(np.asarray(G, np.float32), (len(rows), 1)); ell[6] = np.nan | |
| d = V4B.wheel_rim_distance(q, ell); d4 = V4.wheel_signed_distance(q, ell) | |
| assert d.dtype == np.float32 and d.shape == (len(rows),) | |
| assert d[0] == pytest.approx(0.0, abs=2e-4) and d[1] == pytest.approx(G[3], abs=1e-5) and d[2] == pytest.approx(0.02, abs=2e-4) | |
| assert d[3] == pytest.approx(0.02, abs=2e-4) and d[4] == pytest.approx(0.10, abs=2e-4) and d[5] == pytest.approx(0.06, abs=5e-4) and np.isnan(d[6]) | |
| assert (d[np.isfinite(d)] >= 0).all(), "d_rim is never negative" | |
| # v4 signed convention on the same rows: the interior is negative (ON in v4); abs is taken per wrist, not after the v4 min | |
| assert d4[1] == pytest.approx(-G[3], abs=1e-5) and d4[2] == pytest.approx(-G[3], abs=1e-5) and abs(d4[2]) != pytest.approx(d[2], abs=1e-3) | |
| assert np.allclose(np.abs(d4[[0, 3, 4, 5]]), d[[0, 3, 4, 5]], atol=1e-6), "single-wrist rows: d_rim = |v4 signed d|" | |
| # state: rim / near-rim rows are ON, the deep interior and the far row are OFF (each row evaluated from a fresh state) | |
| states = [V4B.hyst_hands_v4b(np.array([v], np.float32))[0][0] for v in d[:6]] | |
| assert states == [1, 0, 1, 1, 0] + [states[5]] and np.isnan(states[5]), "dead band alone is undecided" | |
| def test_deep_interior_gives_leave_and_return_in_v4b_but_not_in_v4(): | |
| """Both hands move from the rim to the ellipse centre (hands in the lap behind the wheel in 2-D) for 4 s and come back.""" | |
| n = 3000; manual = np.ones(n, bool) | |
| q = _route(n, manual, interior_frames=range(2000, 2040)) | |
| ell, fits, _, _, _ = V4.causal_ellipse_series(np.zeros((0, 2)), np.zeros((0, 2)), q, manual, V4.device_rng("E2E")) | |
| assert np.isfinite(ell[300:]).all() | |
| d_rim = V4B.wheel_rim_distance(q, ell); s, h = V4B.hyst_hands_v4b(d_rim); ev_r, ev_l, unk = V3.confirmed_transitions(s, h) | |
| assert (d_rim[300:2000] < 0.04).all() and (s[300:2000] == 1).all(), "sliding along the rim stays in the ON band" | |
| assert (d_rim[2000:2040] > 0.08).all() and (s[2000:2040] == 0).all(), "the deep interior is OFF in v4b" | |
| 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 and not unk.any() | |
| # v4 (filled-region convention) on the same data: the interior is ON -> no event at all (the failure mode that motivated v4b) | |
| d4 = V4.wheel_signed_distance(q, ell); s4, h4 = V4.hyst_hands_v4(d4); r4, l4, _ = V3.confirmed_transitions(s4, h4) | |
| assert (s4[300:2040] == 1).all() and r4.sum() == 0 and l4.sum() == 0 | |
| # a genuine excursion OUTSIDE the wheel (lap beside the wheel) is an event in both conventions at the same frames | |
| q2 = _route(n, manual, lap_frames=range(1000, 1030)) | |
| ell2, _, _, _, _ = V4.causal_ellipse_series(np.zeros((0, 2)), np.zeros((0, 2)), q2, manual, V4.device_rng("E2E")) | |
| sb, hb = V4B.hyst_hands_v4b(V4B.wheel_rim_distance(q2, ell2)); rb, lb, _ = V3.confirmed_transitions(sb, hb) | |
| s4b, h4b = V4.hyst_hands_v4(V4.wheel_signed_distance(q2, ell2)); r4b, l4b, _ = V3.confirmed_transitions(s4b, h4b) | |
| assert int(np.flatnonzero(lb)[0]) == 1000 and int(np.flatnonzero(rb)[0]) == 1030 and np.array_equal(rb, r4b) and np.array_equal(lb, l4b) | |
| # ----------------------------------------------------------------------------------------------------- hysteresis / events | |
| def test_hysteresis_v4b_thresholds_and_reset(): | |
| d = np.array([0.20, 0.06, 0.03, 0.06, 0.079, 0.081, 0.06, 0.039, 0.0, 0.081, 0.05], np.float32) | |
| s, held = V4B.hyst_hands_v4b(d) | |
| # 0.20 OFF; 0.06 dead band keeps OFF; 0.03 ON; held; held; 0.081 OFF; held; 0.039 ON; 0.0 (on the rim) 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, _ = V4B.hyst_hands_v4b(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_rim == 0.04 / 0.08 do not switch" | |
| s3, _ = V4B.hyst_hands_v4b(np.array([0.20, 0.08, 0.04, 0.20, 0.04], np.float32)); assert s3.tolist() == [0, 0, 0, 0, 0] | |
| g = V4B.GAP_RESET | |
| d4 = np.r_[[0.0] * 3, [np.nan] * (g + 1), [0.06, 0.06, 0.03]].astype(np.float32); s4, h4 = V4B.hyst_hands_v4b(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 = V4B.hyst_hands_v4b(d5); assert s5[3 + g] == 1 and (h5[3:3 + g] == 1).all(), "gap of exactly 1 s keeps the state" | |
| # a deep-interior distance (b = 0.15 at the centre) is OFF: the v4b input is |d| so the interior arrives as a large positive value | |
| s6, _ = V4B.hyst_hands_v4b(np.array([0.15, 0.10, 0.0], np.float32)); assert s6.tolist() == [0, 0, 1] | |
| # identical to the v4 hysteresis on identical (non-negative) input | |
| s7, h7 = V4B.hyst_hands_v4b(d); s7v4, h7v4 = V4.hyst_hands_v4(d); assert np.array_equal(s7, s7v4) and np.array_equal(h7, h7v4) | |
| 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 = V4B.hyst_hands_v4b(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 = V4B.hyst_hands_v4b(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 = V4B.hyst_hands_v4b(d3); r3, l3, _ = V3.confirmed_transitions(s3, h3); assert r3.sum() == 1 and l3.sum() == 1 | |
| # rim sliding on either side of the rim: d_rim stays < 0.08 -> NO event | |
| d4 = ser((30, 0.02), (20, 0.06), (30, 0.01)); s4, h4 = V4B.hyst_hands_v4b(d4); r4, l4, _ = V3.confirmed_transitions(s4, h4); assert r4.sum() == 0 and l4.sum() == 0 and (s4 == 1).all() | |
| # route edge: a transition < 1 s before the end is unconfirmable (EM cleared), not an event | |
| d5 = ser((30, 0.20), (5, 0.0)); s5, h5 = V4B.hyst_hands_v4b(d5); r5, l5, unk5 = V3.confirmed_transitions(s5, h5); assert r5.sum() == 0 and unk5[30] | |
| def test_S0_causal_no_lookahead_and_prefix_invariance(): | |
| d = np.r_[[0.20] * 5, [np.nan] * 12, [0.06] * 3, [0.0] * 5].astype(np.float32); s, _ = V4B.hyst_hands_v4b(d) | |
| assert s[4] == 0 and np.isnan(s[17]) and np.isnan(s[19]) and s[20] == 1, "a later ON does not change an earlier NaN / OFF state" | |
| n = 1500; manual = np.ones(n, bool); q = _route(n, manual, interior_frames=range(1200, 1230)) | |
| ell, _, _, _, _ = V4.causal_ellipse_series(np.zeros((0, 2)), np.zeros((0, 2)), q, manual, V4.device_rng("P")) | |
| s_full, _ = V4B.hyst_hands_v4b(V4B.wheel_rim_distance(q, ell)) | |
| for K in (351, 700, 1210): | |
| ellK, _, _, _, _ = V4.causal_ellipse_series(np.zeros((0, 2)), np.zeros((0, 2)), q[:K], manual[:K], V4.device_rng("P")) | |
| sK, _ = V4B.hyst_hands_v4b(V4B.wheel_rim_distance(q[:K], ellK)) | |
| assert np.array_equal(sK, s_full[:K], equal_nan=True), K | |
| q2 = q.copy(); q2[1000:] = np.nan | |
| ell2, _, _, _, _ = V4.causal_ellipse_series(np.zeros((0, 2)), np.zeros((0, 2)), q2, manual, V4.device_rng("P")) | |
| s2, _ = V4B.hyst_hands_v4b(V4B.wheel_rim_distance(q2, ell2)); assert np.array_equal(s2[:1000], s_full[:1000], equal_nan=True) | |
| # ------------------------------------------------------------------------------------------------------------ override plumbing | |
| def test_patched_overrides_exactly_two_things_and_restores_v4(): | |
| orig = {k: getattr(V4, k) for k in ("wheel_signed_distance", "OUT_V4", "LABEL_VERSION", "WHEEL_RULE", "write_run_json", "hyst_hands_v4", "causal_ellipse_series", "HYST_ON", "HYST_OFF", "build_route_series", "main")} | |
| q = np.array([[G[0], G[1], G[0], G[1], 0.0]], np.float32); ell = np.asarray(G, np.float32)[None] | |
| with V4B.patched(): | |
| assert V4.wheel_signed_distance is V4B.wheel_rim_distance and V4.OUT_V4 == V4B.OUT_V4B and V4.LABEL_VERSION == "v4b" and V4.WHEEL_RULE == V4B.WHEEL_RULE | |
| assert V4.wheel_signed_distance(q, ell)[0] == pytest.approx(G[3], abs=1e-5), "inside the block the v4 module computes d_rim" | |
| for k in ("hyst_hands_v4", "causal_ellipse_series", "HYST_ON", "HYST_OFF", "build_route_series", "main"): assert getattr(V4, k) is orig[k], k | |
| for k, v in orig.items(): assert getattr(V4, k) is v, f"{k} not restored" | |
| assert V4.wheel_signed_distance(q, ell)[0] == pytest.approx(-G[3], abs=1e-5) and V4.LABEL_VERSION == "v4" | |
| with pytest.raises(RuntimeError): | |
| with V4B.patched(): raise RuntimeError("x") | |
| assert V4.wheel_signed_distance is orig["wheel_signed_distance"] and V4.OUT_V4 == orig["OUT_V4"], "restored on error too" | |
| assert V4B.run_name("build_kin_causal_routes_v4") == "build_kin_causal_routes_v4b" and V4B.run_name("other") == "other" | |
| assert "v4b" in V4B.WHEEL_RULE and "interior is NOT on" in V4B.WHEEL_RULE and "negative inside" not in V4B.WHEEL_RULE and "negative inside" in V4.WHEEL_RULE | |
| assert V4B.WHEEL_RULE.split("d_rim")[0] == V4.WHEEL_RULE.split("d = min over wrists")[0], "everything before the distance clause is the v4 text" | |
| def test_constants_match_preregistration_and_no_copied_machinery(): | |
| assert V4B.HYST_ON == 0.04 and V4B.HYST_OFF == 0.08 and V4B.PERSIST_S == 1.0 and V4B.GAP_RESET == 10 and V4B.LABEL_VERSION == "v4b" | |
| assert V4B.HYST_ON == V4.HYST_ON and V4B.HYST_OFF == V4.HYST_OFF and V4B.PERSIST_S == V4.PERSIST_S and V4B.GAP_RESET == V4.GAP_RESET | |
| assert V4B.OUT_V4B.name == "targets_kin_causal_routes_v4b" and V4B.OUT_V4B != V4.OUT_V4 | |
| assert WE.N_MIN_PTS == 200 and WE.COV_MIN_DEG == 90.0 and V4.REFIT_FRAMES == 50 and V4.POOL_CAP == 4000 | |
| src = (ROOT / "data" / "build_kin_causal_v4b.py").read_text() | |
| for token in ("def causal_ellipse_series", "def build_route_series", "def hyst_hands(", "def confirmed_transitions", "ransac", "def signed_distance", "savez"): | |
| assert token not in src, f"v4b must not copy v4/v3 machinery ({token})" | |
| assert "V4.main(" in src and "V4.wheel_signed_distance" in src and "V4.hyst_hands_v4" in src | |
| assert "hyst_hands_v4" in V4B.hyst_hands_v4b.__code__.co_names and "_V4_SIGNED_DISTANCE" in V4B.wheel_rim_distance.__code__.co_names | |
| assert V4B._V4_SIGNED_DISTANCE is V4.wheel_signed_distance, "the bound original is the v4 function (outside `patched`)" | |
| 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_v4b.py", Path(__file__), ROOT / "eval" / "kin_v4b_build_stats.py", | |
| ROOT / "tools" / "review_tool" / "build_relabel_v4b.py", ROOT / "tools" / "review_tool" / "apply_review_v4b.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) | |
| # ------------------------------------------------------------------------------------------------------------------ built files | |
| def test_built_files_inputs_identical_to_v2_ellipse_identical_to_v4(reg): | |
| pb = V4B_DIR / f"kin_causal_{reg}_test.npz"; p2 = V2_DIR / f"kin_causal_{reg}_test.npz" | |
| if not (pb.exists() and p2.exists()): pytest.skip("v4b test split not built") | |
| zb = np.load(pb, allow_pickle=True); z2 = np.load(p2, allow_pickle=True) | |
| assert np.array_equal(zb["uid"], z2["uid"]) and np.allclose(zb["t"], z2["t"]) | |
| E, EM = zb["E_c"], zb["EM_c"]; assert not (E.astype(bool) & ~EM).any() | |
| assert str(zb["label_version"]) == "v4b" and float(zb["hyst_on"]) == np.float32(0.04) and float(zb["hyst_off"]) == np.float32(0.08) and float(zb["persist_s"]) == 1.0 and int(zb["gap_reset_frames"]) == 10 | |
| assert "interior is NOT on" in str(zb["wheel_ref_rule"]) | |
| for key in ("K_c", "K0_c", "Kvalid", "Kframe", "K0valid", "row_ok", "route_decoded"): | |
| assert zb[key].dtype == z2[key].dtype and zb[key].shape == z2[key].shape, key | |
| assert np.array_equal(zb[key], z2[key], equal_nan=True) if zb[key].dtype.kind == "f" else np.array_equal(zb[key], z2[key]), f"{key} differs from v2 (inputs must be identical)" | |
| assert np.array_equal(zb["S_c"][:, :, 1], z2["S_c"][:, :, 1], equal_nan=True) and np.array_equal(zb["E_c"][:, :, 2], z2["E_c"][:, :, 2]) and np.array_equal(zb["EM_c"][:, :, 2], z2["EM_c"][:, :, 2]) | |
| S = zb["S_c"][:, :, 0]; S0 = zb["S0_c"][:, 0]; Sp = np.concatenate([S0[:, None], S[:, :-1]], 1) | |
| er = zb["E_c"][:, :, 0].astype(bool); assert (Sp[er] == 0).all() and (S[er] == 1).all() | |
| el = zb["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} | |
| D0 = zb["dwheel0_c"]; assert (D0[np.isfinite(D0)] >= 0).all(), "dwheel0_c holds d_rim >= 0" | |
| ok = zb["row_ok"]; on = ok & (S0 == 1); off = ok & (S0 == 0) | |
| assert not (D0[on] > 0.08).any() and not (D0[off] < 0.04).any(), "state consistent with d_rim at the anchor frame" | |
| assert zb["ellipse0_c"].shape == (len(S0), 5) and (np.isfinite(zb["ellipse0_c"]).all(1) == zb["wheel_ref_defined"]).all() | |
| hb = V4B_DIR / f"kinhist_causal_{reg}_test.npz"; h2 = V2_DIR / f"kinhist_causal_{reg}_test.npz" | |
| assert hb.exists() and (os.path.samefile(hb, h2) or hb.read_bytes() == h2.read_bytes()), "kinhist byte-identical to v2 (hard link)" | |
| p4 = V4_DIR / f"kin_causal_{reg}_test.npz" | |
| if p4.exists(): | |
| z4 = np.load(p4, allow_pickle=True) | |
| assert np.array_equal(zb["ellipse0_c"], z4["ellipse0_c"], equal_nan=True) and np.array_equal(zb["wheel_ref_defined"], z4["wheel_ref_defined"]), "same causal ellipse as v4" | |
| both = np.isfinite(D0) & np.isfinite(z4["dwheel0_c"]); assert (D0[both] <= np.abs(z4["dwheel0_c"][both]) + 1e-6).all(), "min over wrists of |d| <= |v4 signed min|" | |
| assert np.array_equal(np.isfinite(D0), np.isfinite(z4["dwheel0_c"])), "d_rim defined exactly where the v4 d is defined" | |
| def test_built_sample_route_recomputed_from_pose_matches_v4b(): | |
| pb = V4B_DIR / "kin_causal_engaged_test.npz" | |
| if not (pb.exists() and (V4B_DIR / "route_refs.npz").exists()): pytest.skip("v4b test split not built") | |
| zb = np.load(pb, allow_pickle=True); rr = np.load(V4B_DIR / "route_refs.npz", allow_pickle=True) | |
| assert str(rr["label_version"]) == "v4b" | |
| R = np.flatnonzero(zb["row_ok"] & zb["wheel_ref_defined"]); u = zb["uid"][R[len(R) // 2]]; rows = np.flatnonzero(zb["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); ell = rr[f"ell__{u}"] | |
| d = V4B.wheel_rim_distance(raw["q"], ell); d[~F] = np.nan | |
| assert np.allclose(d, rr[f"d4__{u}"], equal_nan=True, atol=1e-6) and (d[np.isfinite(d)] >= 0).all() | |
| k = np.round(t * 10).astype(np.int64); k0 = int(k.min()); M = int(k.max()) - k0 + 1; DD = np.full(M, np.nan, np.float32); DD[k - k0] = d | |
| s_raw, _ = V4B.hyst_hands_v4b(DD) | |
| ka = np.round(zb["t"][rows] * 10).astype(np.int64) - k0; ok = (ka >= 0) & (ka < M) | |
| assert np.array_equal(zb["S0_c"][rows, 0][ok], s_raw[ka[ok]], equal_nan=True), "S0_c hands_on recomputed causally from d_rim" | |
| assert np.allclose(zb["dwheel0_c"][rows][ok], DD[ka[ok]], equal_nan=True) | |
| if (V4_DIR / "route_refs.npz").exists(): | |
| rr4 = np.load(V4_DIR / "route_refs.npz", allow_pickle=True) | |
| assert np.array_equal(rr[f"ell__{u}"], rr4[f"ell__{u}"], equal_nan=True) and np.array_equal(rr[f"fits__{u}"], rr4[f"fits__{u}"], equal_nan=True), "ellipse series and fit log identical to v4" | |
| assert np.array_equal(rr[f"ref__{u}"], rr4[f"ref__{u}"], equal_nan=True) | |
Xet Storage Details
- Size:
- 20.4 kB
- Xet hash:
- 46406e2de060600473c5d18bc05202b0cd4b8a7b4596f1cc9014faf7aa4f56fd
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.