Buckets:
| """Layout contract of results/targets_gaze_causal_routes_v2/gaze_{regime}_{split}.npz (data/build_gaze_targets.py; gaze/robustness agent 2026-09-12): | |
| s12 / kin_causal alignment, NaN policy, onset masks inside the s12 M, and exact agreement of the anchor state with the face pack's gaze_off channel.""" | |
| from pathlib import Path | |
| import json | |
| import numpy as np | |
| import pytest | |
| ROOT = Path(__file__).resolve().parents[1] | |
| GAZE = ROOT / "results" / "targets_gaze_causal_routes_v2" | |
| FACE = ROOT / "results" / "targets_face_causal_routes_v2" | |
| KIN = ROOT / "results" / "targets_kin_causal_routes_v2" | |
| S12 = Path("/home/USER/s12_wm/results") | |
| CASES = [(r, s) for r in ("manual", "engaged") for s in ("train", "val", "test")] | |
| def _load(reg, sp): | |
| p = GAZE / f"gaze_{reg}_{sp}.npz" | |
| if not p.exists(): | |
| pytest.skip("gaze targets not built yet") | |
| return np.load(p, allow_pickle=True) | |
| def test_alignment_shapes(reg, sp): | |
| g = _load(reg, sp) | |
| s = np.load(S12 / f"targets_{reg}_{sp}.npz", allow_pickle=True) | |
| k = np.load(KIN / f"kin_causal_{reg}_{sp}.npz", allow_pickle=True) | |
| N = len(s["t"]) | |
| assert np.array_equal(g["uid"], s["uid"]) and np.allclose(g["t"], s["t"]) and np.array_equal(g["uid"], k["uid"]) | |
| assert g["S_g"].shape == (N, 25) and g["S0_g"].shape == (N,) and g["E_g"].shape == (N, 25, 2) and g["EM_g"].shape == (N, 25, 2) and g["E0_g"].shape == (N, 2) | |
| assert [str(x) for x in g["state_names"]] == ["gaze_off"] and [str(x) for x in g["event_names"]] == ["gaze_off_on", "gaze_off_off"] and str(g["source"]) == "routes" | |
| thr = json.loads(str(g["thresholds"])) | |
| assert thr["gaze_off_on"] > thr["gaze_off_off"] > 0 | |
| def test_nan_policy_masks_and_face_agreement(reg, sp): | |
| g = _load(reg, sp) | |
| Mc = np.load(S12 / f"targets_{reg}_{sp}.npz", allow_pickle=True)["M"].astype(bool) | |
| S, S0, E, EM = g["S_g"], g["S0_g"], g["E_g"], g["EM_g"] | |
| fin = np.isfinite(S) | |
| assert set(np.unique(S[fin]).tolist()) <= {0.0, 1.0} and set(np.unique(S0[np.isfinite(S0)]).tolist()) <= {0.0, 1.0} | |
| assert not (fin & ~Mc).any(), "future states must be censored outside the s12 M" | |
| assert not (EM & ~Mc[:, :, None]).any() and not (E.astype(bool) & ~EM).any() | |
| zp = np.concatenate([S0[:, None], S[:, :-1]], 1) | |
| assert np.array_equal(EM[:, :, 0], np.isfinite(zp) & fin) and np.array_equal(EM[:, :, 1], EM[:, :, 0]) | |
| assert np.array_equal(E[:, :, 0].astype(bool), EM[:, :, 0] & (zp < 0.5) & (S > 0.5)) | |
| assert np.array_equal(E[:, :, 1].astype(bool), EM[:, :, 1] & (zp > 0.5) & (S < 0.5)) | |
| fz = np.load(FACE / f"facehist_causal_{reg}_{sp}.npz", allow_pickle=True) | |
| F0, F0v = fz["F0_c"].astype(np.float32), fz["F0valid"].astype(bool) | |
| both = np.isfinite(S0) & F0v & np.isfinite(F0[:, 7]) | |
| assert both.sum() > 0 and np.array_equal(S0[both], F0[both, 7]), "anchor gaze_off state must equal the face pack channel" | |
| assert np.array_equal(g["row_ok"].astype(bool), fz["row_ok"].astype(bool)) | |
| absdev = g["absdev0"] | |
| assert np.allclose(absdev[F0v], np.abs(F0[F0v, 1]), equal_nan=True) | |
Xet Storage Details
- Size:
- 3.12 kB
- Xet hash:
- c6da33106f7c8880eb5f36fe914e2526bbc167f002e4ff1d14ffedf6700d2a55
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.