Buckets:
| """cwm3/tests/test_aide_kin11.py — the AIDE kin11 cache (aide/features/build_aide_kin11.py) obeys the BATON kinematics pack contract: | |
| channel order / names / units are those of data/build_kin_causal.py and the causal route packs, the scale rule is x = x_norm * 1.6 / y = y_norm, | |
| the derived channels satisfy their definitions (min = fmin(L, R), dev = head_turn - ref, dist = hypot(wrist - ref)), every valid step has all 11 | |
| channels finite, and a clip recomputed from its RTMW npz with the shared functions reproduces the stored values. | |
| Run: cd cwm3 && python -m pytest tests/test_aide_kin11.py -q | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import sys | |
| from pathlib import Path | |
| import numpy as np | |
| import pytest | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| for p in (CWM3, CWM3 / "data", CWM3 / "aide" / "features"): | |
| sys.path.insert(0, str(p)) | |
| import build_kin_causal as BK # noqa: E402 | |
| import cwm3_dataset as ds # noqa: E402 | |
| import build_aide_kin11 as AK # noqa: E402 | |
| CACHE = CWM3 / "aide" / "results" / "cache" / "aide_kin11.npz" | |
| BATON_PACK = CWM3 / "results" / "targets_kin_causal_routes_v2" / "kinhist_causal_manual_test.npz" | |
| def z(): | |
| if not CACHE.exists(): | |
| pytest.skip("aide_kin11.npz not built") | |
| return np.load(CACHE, allow_pickle=True) | |
| def test_channel_contract(z): | |
| ch = [str(c) for c in z["channels"]] | |
| assert ch == BK.CHANNELS == ds.KIN_CHANNELS | |
| assert len(ch) == 11 and ch[10] == "min_hand_rest_dist" and ch[0] == "head_cx" and ch[7] == "head_turn_dev" | |
| assert str(z["units"]) == BK.UNITS | |
| assert [str(q) for q in z["ref_quantities"]] == BK.REFQ | |
| if BATON_PACK.exists(): | |
| b = np.load(BATON_PACK, allow_pickle=True) | |
| assert [str(c) for c in b["channels"]] == ch and str(b["units"]) == str(z["units"]) | |
| assert z["K"].shape[-1] == 11 and z["K"].dtype == np.float32 and z["valid"].dtype == np.bool_ | |
| for k in ("kmu", "ksd"): | |
| assert z[k].shape == (11,) and np.isfinite(z[k]).all() and (z["ksd"] > 0).all() | |
| for tok in ds.FORBIDDEN_KEYS: | |
| assert tok not in z.files | |
| def test_shapes_and_validity(z): | |
| K, V, n = z["K"], z["valid"], z["n_steps"] | |
| N, T, _ = K.shape | |
| assert N == 2898 and V.shape == (N, T) and n.shape == (N,) and (n >= 30).all() and n.max() == T | |
| assert np.isfinite(K[V]).all(), "a valid step must carry all 11 channels" | |
| assert np.isnan(K[~V]).all(), "invalid steps are NaN (pack NaN policy)" | |
| inside = np.arange(T)[None, :] < n[:, None] | |
| assert not V[~inside].any() | |
| assert np.allclose(z["t_grid"], np.arange(T) * 0.1) | |
| assert set(map(str, z["split"])) == {"train", "val", "test"} and len(z["clip_id"]) == N | |
| def test_definitions_and_scale(z): | |
| K, V, R = z["K"], z["valid"], z["ref"] | |
| k, r = K[V], R[V] | |
| f16 = lambda *cols: 2.0 ** -11 * sum(np.abs(c) for c in cols) + 1e-4 # stored in f16 (pack precision): |err| <= 2^-11 * |value| | |
| assert (np.abs(k[:, 10] - np.fmin(k[:, 8], k[:, 9])) <= f16(k[:, 8], k[:, 9], k[:, 10])).all() | |
| assert (np.abs(k[:, 7] - (k[:, 6] - r[:, 4])) <= f16(k[:, 6], k[:, 7])).all() | |
| assert (np.abs(k[:, 8] - np.hypot(k[:, 2] - r[:, 0], k[:, 3] - r[:, 1])) <= f16(k[:, 2], k[:, 3], k[:, 8])).all() | |
| assert (np.abs(k[:, 9] - np.hypot(k[:, 4] - r[:, 2], k[:, 5] - r[:, 3])) <= f16(k[:, 4], k[:, 5], k[:, 9])).all() | |
| # scale rule: x ~ [0, 1.6] (x_norm * XSCALE), y ~ [0, 1]; RTMW may place a joint slightly outside the frame (-0.1 margin); distances >= 0 | |
| for j in (0, 2, 4): | |
| assert k[:, j].min() >= -0.1 * BK.XSCALE and k[:, j].max() <= 1.1 * BK.XSCALE and np.percentile(k[:, j], 99) <= BK.XSCALE | |
| for j in (1, 3, 5): | |
| assert k[:, j].min() >= -0.1 and k[:, j].max() <= 1.1 | |
| assert (k[:, 8:] >= -1e-6).all() | |
| assert BK.XSCALE == 1.6 and BK.WHEEL_TH == 0.10 and BK.CONF_TH == 0.30 | |
| ho = z["hands_on"][V] | |
| assert np.array_equal(ho > 0.5, k[:, 10] < BK.WHEEL_TH) | |
| def test_recompute_one_clip_matches(z): | |
| mdir = AK.MOTION_DIR | |
| cid = str(z["clip_id"][0]) | |
| if not (mdir / f"{cid}.npz").exists(): | |
| pytest.skip("RTMW motion npz not available") | |
| old = json.load(open(BK.OLD_STATS)) | |
| kp = np.load(mdir / f"{cid}.npz")["kpts"].astype(np.float32) | |
| out = AK.clip_channels(kp, float(old["head_turn_on"]), float(old["head_turn_off"])) | |
| n = int(z["n_steps"][0]) | |
| assert np.array_equal(out["valid"], z["valid"][0, :n]) | |
| assert np.allclose(out["K"], z["K"][0, :n], equal_nan=True) | |
| # the raw x channel is x_norm * 1.6 of the RTMW nose / wrist (COCO body index 9 = left wrist) | |
| v = out["valid"] | |
| if v.any(): | |
| j = int(np.flatnonzero(v)[0]) | |
| assert abs(out["K"][j, 2] - kp[j, 9, 0] * BK.XSCALE) < 2e-3 and abs(out["K"][j, 3] - kp[j, 9, 1]) < 2e-3 | |
| def test_train_stats_are_train_only(z): | |
| tr = (z["split"] == "train")[:, None] & z["valid"] | |
| k = z["K"][tr].astype(np.float64) | |
| assert np.allclose(k.mean(0), z["kmu"], atol=1e-4) and np.allclose(k.std(0) + 1e-6, z["ksd"], atol=1e-4) | |
Xet Storage Details
- Size:
- 5.11 kB
- Xet hash:
- 31668c923bf8e836f6b0b726cd720d2120dfc5ee6b3a1b80c73b53976554c741
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.