Buckets:
| """Layout contract of results/targets_kin_causal_routes/{kin,kinhist}_causal_{regime}_{split}.npz (build_kin_causal.py --source routes).""" | |
| from pathlib import Path | |
| import numpy as np, pytest | |
| import os | |
| ROOT = Path(__file__).resolve().parents[1]; KIN = Path(os.environ.get("KIN_ROUTES_DIR", ROOT / "results" / "targets_kin_causal_routes")) | |
| 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 = KIN / f"kin_causal_{reg}_{sp}.npz" | |
| if not p.exists(): pytest.skip("route targets not built yet") | |
| return np.load(p, allow_pickle=True), np.load(KIN / f"kinhist_causal_{reg}_{sp}.npz", allow_pickle=True) | |
| def test_alignment_shapes(reg, sp): | |
| k, h = _load(reg, sp); s = np.load(S12 / f"targets_{reg}_{sp}.npz", allow_pickle=True); N = len(s["t"]) | |
| assert np.array_equal(k["uid"], s["uid"]) and np.allclose(k["t"], s["t"]) and np.array_equal(h["uid"], s["uid"]) | |
| assert k["K_c"].shape == (N, 25, 11) and h["Kh_c"].shape == (N, 40, 11) and h["Kh8_c"].shape == (N, 80, 11) and h["Kh8valid"].shape == (N, 80) | |
| assert k["S_c"].shape == (N, 25, 2) and k["E_c"].shape == (N, 25, 3) and k["route_decoded"].shape == (N,) and str(k["source"]) == "routes" | |
| def test_nan_policy_and_rules(reg, sp): | |
| k, h = _load(reg, sp); R = np.flatnonzero(k["row_ok"]); no = ~k["route_decoded"] | |
| assert np.isnan(k["K_c"][no].astype(np.float32)).all() and np.isnan(h["Kh8_c"][no].astype(np.float32)).all() and not k["Kvalid"][no].any() | |
| if not len(R): pytest.skip("no coverage") | |
| K = k["K_c"][R].astype(np.float32); V = k["Kvalid"][R]; assert np.isfinite(K[V][:, [0, 2, 4, 6, 7, 10]]).all() | |
| S = k["S_c"][R]; E, EM = k["E_c"][R], k["EM_c"][R]; assert not (E.astype(bool) & ~EM).any() | |
| hon, md = S[:, :, 0], K[:, :, 10]; m = np.isfinite(hon) & np.isfinite(md); assert np.array_equal(hon[m] > 0.5, md[m] < 0.10) | |
| mc = np.load(S12 / f"targets_{reg}_{sp}.npz", allow_pickle=True)["M"].astype(bool)[R]; assert not (V & ~mc).any() and not (EM & ~mc[:, :, None]).any() | |
| # history at 10 Hz: the last history step equals K0 where both valid | |
| Kh = h["Kh_c"][R].astype(np.float32); K0 = k["K0_c"][R]; both = h["Khvalid"][R][:, -1] & k["K0valid"][R] | |
| assert np.allclose(Kh[both, -1], K0[both], atol=2e-3, equal_nan=True) | |
| Kh8 = h["Kh8_c"][R].astype(np.float32); assert np.allclose(Kh8[:, -40:], Kh, atol=2e-3, equal_nan=True), "Kh_c is the last 4 s of Kh8_c" | |
Xet Storage Details
- Size:
- 2.54 kB
- Xet hash:
- 7bae708d820264db1a1b3df8ca7119cdc900f3ece5b75fbf643e683710cd4b07
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.