Buckets:
| """tests for eval/pitch_wristface_diag.py (plan A diagnosis): proxy geometry, causality, statistics.""" | |
| import sys | |
| from pathlib import Path | |
| import numpy as np | |
| ROOT = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(ROOT / "eval")) | |
| import pitch_wristface_diag as D # noqa: E402 | |
| def synth(T=400, W=1928, H=1208, pitch_down_from=None, hand_to_face_from=None, seed=0): | |
| """synthetic HALPE-136 series: face at image centre, eyes 0.05 W apart, nose tip 0.03 H below the eye line, shoulders 0.3 W apart, wrists on a wheel.""" | |
| rng = np.random.default_rng(seed) | |
| k = np.full((T, 136, 3), np.nan, np.float32) | |
| cx, cy = 0.5, 0.4 | |
| for t in range(T): | |
| down = pitch_down_from is not None and t >= pitch_down_from | |
| near = hand_to_face_from is not None and t >= hand_to_face_from | |
| ex = 0.025 * W / H # eye half-separation in H units -> back to x/W below | |
| re_c = np.array([cx - 0.025, cy]); le_c = np.array([cx + 0.025, cy]) | |
| nose = np.array([cx, cy + (0.06 if down else 0.03)]) | |
| for j, c in zip(D.RE, [re_c + rng.normal(0, 0.0005, 2) for _ in range(6)]): k[t, j] = [c[0], c[1], 0.9] | |
| for j, c in zip(D.LE, [le_c + rng.normal(0, 0.0005, 2) for _ in range(6)]): k[t, j] = [c[0], c[1], 0.9] | |
| k[t, D.NT] = [nose[0], nose[1], 0.9]; k[t, D.NOSE] = [nose[0], nose[1], 0.9] | |
| k[t, D.LSHO] = [cx - 0.06, cy + 0.25, 0.9]; k[t, D.RSHO] = [cx + 0.06, cy + 0.25, 0.9] # shoulder width 0.12 W = 0.19 H units | |
| if near: k[t, D.LWRI] = [cx - 0.02, cy + 0.05, 0.9] | |
| else: k[t, D.LWRI] = [cx - 0.12, cy + 0.55, 0.9] | |
| k[t, D.RWRI] = [cx + 0.12, cy + 0.55, 0.9] # 0.30 H below the shoulder line > 1 shoulder width | |
| return k, W, H | |
| def test_frame_proxies_geometry_and_sign(): | |
| k, W, H = synth(T=50) | |
| p = D.frame_proxies(k, W, H) | |
| assert p["face_valid"].all() and p["body_valid"].all() and p["box"].all() | |
| iod_h = 0.05 * W / H # IOD on the H-unit scale | |
| assert np.allclose(p["head_pitch"], 0.03 / iod_h, atol=0.02) | |
| k2, _, _ = synth(T=50, pitch_down_from=0) | |
| p2 = D.frame_proxies(k2, W, H) | |
| assert (p2["head_pitch"] > p["head_pitch"]).all(), "nose lower relative to the eyes must INCREASE head_pitch_kp (head-down direction)" | |
| k3, _, _ = synth(T=50, hand_to_face_from=0) | |
| p3 = D.frame_proxies(k3, W, H) | |
| assert (p3["wrist_face"] < p["wrist_face"]).all() | |
| # both wrists far below the shoulder line -> lap flag true when wrists on the lap, false when one wrist is at the face | |
| assert p["lap"].all() and not p3["lap"].any() | |
| def test_invalid_frames_are_nan(): | |
| k, W, H = synth(T=20) | |
| k[5, D.NT, 2] = 0.1 # nose tip low confidence | |
| k[6, D.LE[0]:D.LE[0] + 2, 2] = 0.0 # only 4 valid left-eye landmarks | |
| k[7] = np.nan # no box | |
| k[8, D.LWRI, 2] = 0.0 | |
| p = D.frame_proxies(k, W, H) | |
| assert not p["face_valid"][5] and not p["face_valid"][6] and not p["face_valid"][7] | |
| assert np.isnan(p["head_pitch"][[5, 6, 7]]).all() | |
| assert not p["body_valid"][8] and np.isnan(p["wrist_face"][8]) and not p["lap"][8] | |
| assert not p["box"][7] and p["box"][5] | |
| def test_causal_smooth_and_median_mad_are_causal(): | |
| rng = np.random.default_rng(1) | |
| x = rng.normal(0, 1, 300).astype(np.float32); x[10] = np.nan | |
| moving = np.ones(300, bool) | |
| s1 = D.causal_smooth(x); m1, d1 = D.causal_median_mad(s1, moving) | |
| x2 = x.copy(); x2[200:] += 50.0 # change the FUTURE only | |
| s2 = D.causal_smooth(x2); m2, d2 = D.causal_median_mad(s2, moving) | |
| assert np.allclose(s1[:200], s2[:200], equal_nan=True) and np.allclose(m1[:200], m2[:200], equal_nan=True) and np.allclose(d1[:200], d2[:200], equal_nan=True) | |
| assert np.isnan(m1[:D.N_MIN - 1]).all() and np.isfinite(m1[D.N_MIN:]).all() | |
| # smoothing: trailing mean of the last 5 valid values | |
| assert np.isclose(s1[20], np.nanmean(x[16:21])) | |
| assert np.isclose(s1[10], np.nanmean(x[6:11])) # NaN frame skipped, not propagated | |
| # non-moving frames do not enter the median | |
| mv = moving.copy(); mv[100:] = False | |
| m3, _ = D.causal_median_mad(s1, mv) | |
| assert np.allclose(m3[100:], m3[99]) | |
| def test_route_series_flags_head_down_only_after_onset(): | |
| k, W, H = synth(T=600, pitch_down_from=400, hand_to_face_from=450, seed=2) | |
| moving = np.ones(600, bool) | |
| S = D.route_series(k, moving, W, H) | |
| assert S["headdown"][D.N_MIN + 5:395].mean() < 0.10 # Gaussian noise: dev > 3 MAD ~ 2 % of frames one-sided; no sustained flag before onset | |
| assert (np.convolve(S["headdown"][D.N_MIN + 5:395], np.ones(10), "valid") < 10).all() | |
| assert S["headdown"][410:600].mean() > 0.9 | |
| assert S["lap_headdown"][410:450].mean() > 0.9 and not S["lap_headdown"][450:].any() # one wrist at the face -> not both on the lap | |
| pos = D.window_scores(S, 400, 500); neg = D.window_scores(S, 200, 300) | |
| assert pos["pitch_dev_max"] > neg["pitch_dev_max"] and pos["neg_wrist_face_max"] > neg["neg_wrist_face_max"] and pos["lap_headdown_share"] > neg["lap_headdown_share"] | |
| assert pos["cov_face"] == 1.0 and pos["n_frames"] == 100 | |
| e = D.window_scores(S, 590, 700); assert e["n_frames"] == 10 # truncated at the series end | |
| def test_auroc_rank_bootstrap(): | |
| assert D.auroc([1, 2, 3], [0, 0, 0]) == 1.0 and D.auroc([0, 0], [1, 1]) == 0.0 and D.auroc([1, 1], [1, 1]) == 0.5 | |
| assert np.isnan(D.auroc([], [1])) | |
| r = D.rankdata(np.array([3.0, np.nan, 1.0, 3.0])) | |
| assert list(r) == [3.5, 1.0, 2.0, 3.5] | |
| rng = np.random.default_rng(0) | |
| score = np.r_[rng.normal(2, 1, 40), rng.normal(0, 1, 120)]; y = np.r_[np.ones(40, bool), np.zeros(120, bool)] | |
| cl = np.tile(np.arange(8), 20) | |
| out = D.cluster_boot_auroc(score, y, cl, rng, 200) | |
| assert 0.8 < out["auroc"] < 1.0 and out["ci"][0] <= out["auroc"] <= out["ci"][1] and out["n_clusters"] == 8 | |
| score[0] = np.nan | |
| o2 = D.cluster_boot_auroc(score, y, cl, rng, 50); assert o2["n_pos"] == 39 | |
| o3 = D.cluster_boot_auroc(score, y, cl, rng, 50, drop_nan=False); assert o3["n_pos"] == 40 | |
| def test_negative_drawing_respects_clearance(): | |
| moving = np.ones(3000, bool) | |
| grid = np.arange(0, 3000, 2) | |
| excl = [(1000.0 / 10, 1000.0 / 10)] # event at 100 s | |
| pos_windows = [(200.0, 210.0)] | |
| negs, n_c = D.draw_negatives("u", "manual", pos_windows, grid, excl, moving, np.random.default_rng(0), 3) | |
| for t in negs: | |
| assert D.interval_dist(t, t + 10, 100, 100) >= 60 and D.interval_dist(t, t + 10, 200, 210) >= 60 | |
| assert moving[int(t * 10):int(t * 10) + 100].mean() >= 0.8 | |
| assert len(negs) <= 3 and all(abs(a - b) >= 10 for i, a in enumerate(negs) for b in negs[i + 1:]) | |
| moving[:] = False | |
| assert D.draw_negatives("u", "manual", pos_windows, grid, excl, moving, np.random.default_rng(0), 3) == ([], 0) | |
| def test_no_dms_identifiers_in_script_or_outputs(): | |
| """the hard rule: no openpilot DMS field name anywhere in the diagnosis code or its outputs (words assembled from fragments so this file stays clean).""" | |
| words = ["driver" + "_state", "face" + "Prob", "is" + "Distracted", "face" + "_yaw", "face" + "_pitch", "face" + "_roll", "blink" + "Prob", "d" + "monitoring", | |
| "driver" + "State", "awareness" + "Status", "distracted" + "Pct", "d" + "state", "leftEye" + "Prob", "rightEye" + "Prob"] | |
| texts = [(ROOT / "eval" / "pitch_wristface_diag.py").read_text()] | |
| for f in (ROOT / "results" / "tables" / "pitch_wristface_diag.md", ROOT / "results" / "tables" / "pitch_wristface_diag.json"): | |
| if f.exists(): texts.append(f.read_text()) | |
| for txt in texts: | |
| for w in words: | |
| assert w not in txt, w | |
Xet Storage Details
- Size:
- 7.89 kB
- Xet hash:
- c4ca868c4db977699569d25b3378b930b443592eb8af40ee75fb285c40a456fc
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.