Buckets:
| """Layout contract of results/targets_face_causal_routes_v2/facehist_causal_{regime}_{split}.npz (data/build_face_causal.py --source routes): row alignment | |
| with the s12 targets, shapes, channel names, NaN / validity policy, f16 history == F0 at the last step, gaze_off hysteresis semantics, stored thresholds.""" | |
| import json | |
| import os | |
| from pathlib import Path | |
| import numpy as np | |
| import pytest | |
| ROOT = Path(__file__).resolve().parents[1] | |
| FACE = Path(os.environ.get("FACE_ROUTES_DIR", ROOT / "results" / "targets_face_causal_routes_v2")) | |
| S12 = Path("/home/USER/s12_wm/results") | |
| CASES = [(r, s) for r in ("manual", "engaged") for s in ("train", "val", "test")] | |
| CHANNELS = ["lm_yaw", "lm_yaw_dev", "lm_pitch_dev", "lm_roll_dev", "ear_conf_asym", "eye_open", "blink_rate_4s", "gaze_off", "gaze_off_dwell_s"] | |
| def _load(reg, sp): | |
| p = FACE / f"facehist_causal_{reg}_{sp}.npz" | |
| if not p.exists(): | |
| pytest.skip("face pack not built yet") | |
| return np.load(p, allow_pickle=True) | |
| def test_alignment_shapes_names(reg, sp): | |
| z = _load(reg, sp) | |
| s = np.load(S12 / f"targets_{reg}_{sp}.npz", allow_pickle=True) | |
| N = len(s["t"]) | |
| assert np.array_equal(z["uid"], s["uid"]) and np.allclose(z["t"], s["t"]) | |
| assert [str(c) for c in z["channels"]] == CHANNELS | |
| F = len(CHANNELS) | |
| assert z["F0_c"].shape == (N, F) and z["F0_c"].dtype == np.float32 and z["F0valid"].shape == (N,) and z["row_ok"].shape == (N,) | |
| assert z["Fh8_c"].shape == (N, 80, F) and z["Fh8_c"].dtype == np.float16 and z["Fh8valid"].shape == (N, 80) | |
| assert str(z["source"]) == "routes" | |
| thr = json.loads(str(z["thresholds"])) | |
| assert thr["gaze_off_on"] > thr["gaze_off_off"] > 0 and thr["gaze_pct_on"] == 90.0 and thr["gaze_pct_off"] == 75.0 and thr["n_train_frames_for_threshold"] > 100000 | |
| assert thr["landmarks"]["nose_tip"] == 56 and thr["landmarks"]["right_eye"] == list(range(62, 68)) and thr["landmarks"]["left_eye"] == list(range(68, 74)) | |
| def test_validity_and_history(reg, sp): | |
| z = _load(reg, sp) | |
| F0, v, ok = z["F0_c"], z["F0valid"], z["row_ok"] | |
| assert not (v & ~ok).any(), "F0valid implies row_ok" | |
| assert np.isnan(F0[~v]).all(), "invalid anchors carry NaN" | |
| fin_req = [0, 1, 2, 3, 4, 6, 7, 8] # eye_open may be NaN when neither eye has all 6 landmarks | |
| assert np.isfinite(F0[v][:, fin_req]).all() | |
| g = F0[v, 7] | |
| assert set(np.unique(g)).issubset({0.0, 1.0}) | |
| assert (F0[v, 8] >= 0).all() and (F0[v, 8] <= 30.0).all() and (F0[v & (F0[:, 7] < 0.5), 8] == 0).all(), "dwell is 0 when not gaze-off" | |
| assert (F0[v, 6] >= 0).all() | |
| eo = F0[v, 5] | |
| assert np.nanmin(eo) >= 0 and 0.8 < np.nanmedian(eo) < 1.25, "eye_open is a ratio to the route reference (EAR 0 = closed eye)" | |
| R = np.flatnonzero(ok)[:50000] | |
| Fh, Fv = z["Fh8_c"][R].astype(np.float32), z["Fh8valid"][R] | |
| both = Fv[:, -1] & v[R] | |
| assert both.any() and np.allclose(Fh[both, -1], F0[R][both], atol=2e-3, equal_nan=True), "the last history step is the anchor value" | |
| assert np.isnan(Fh[~Fv][:, fin_req]).all(), "invalid history steps are NaN" | |
| no = ~z["route_decoded"] | |
| if no.any(): | |
| assert np.isnan(z["Fh8_c"][no].astype(np.float32)).all() and not z["Fh8valid"][no].any() | |
| def test_hysteresis_reproduces_stored_gaze_off(): | |
| """re-running build_kin_causal.hysteresis on the stored |lm_yaw_dev| history with the stored thresholds reproduces the stored gaze_off state on | |
| rows whose 80 history steps are all valid and start in the off state (state initialised at the first valid step -> identical semantics).""" | |
| import sys | |
| sys.path.insert(0, str(ROOT / "data")) | |
| import build_kin_causal as BK | |
| z = _load("manual", "val") | |
| thr = json.loads(str(z["thresholds"])) | |
| Fh, Fv = z["Fh8_c"][:30000].astype(np.float32), z["Fh8valid"][:30000] | |
| full = Fv.all(1) & (Fh[:, 0, 7] < 0.5) & (np.abs(Fh[:, 0, 1]) < thr["gaze_off_off"]) | |
| idx = np.flatnonzero(full)[:400] | |
| assert len(idx) > 50 | |
| agree = [np.array_equal(BK.hysteresis(np.abs(Fh[i, :, 1]), thr["gaze_off_on"], thr["gaze_off_off"]), Fh[i, :, 7]) for i in idx] | |
| assert np.mean(agree) > 0.98, np.mean(agree) | |
Xet Storage Details
- Size:
- 4.25 kB
- Xet hash:
- 6932ee54cb540383a4fe3a066125e4ec3547c6dbc868cd84653802d9be6d0545
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.