Buckets:
| """Leak guard: no predictor may read frames > index 4 of the 10 protocol frames. | |
| We feed NaNs into the future block (frames 5..9) and require predictions to be finite and | |
| bit-identical to the predictions from the clean input. Covers every analytic baseline, the | |
| trainer's obs/target split, and the SiMLPe-kp model (numpy / torch). | |
| python -m pytest aide/tests/test_no_future_leak.py -q (or run as a script) | |
| """ | |
| from __future__ import annotations | |
| import sys | |
| from pathlib import Path | |
| import numpy as np | |
| import torch | |
| sys.path.insert(0, str(Path(__file__).resolve().parents[2])) | |
| from aide.data import protocol as P # noqa: E402 | |
| from aide.eval import baselines_analytic as B # noqa: E402 | |
| from aide.models import simlpe_kp # noqa: E402 | |
| from aide.train.train_kp import TrainSampler, random_grid, split_obs_fut # noqa: E402 | |
| def _frames(n=16): | |
| if P.CACHE_NPZ.exists(): | |
| c = np.load(P.CACHE_NPZ, allow_pickle=False) | |
| return c["S10"][c["split"] == "test"][:n].astype(np.float64) | |
| rng = np.random.default_rng(1) | |
| s = np.zeros((n, P.N_SAMPLE, P.J, 3)) | |
| s[..., 0] = rng.uniform(0, P.W, s.shape[:-1]) | |
| s[..., 1] = rng.uniform(0, P.H, s.shape[:-1]) | |
| s[..., 2] = rng.uniform(0, 1, s.shape[:-1]) | |
| return s | |
| def _poison(S10): | |
| x = S10.copy() | |
| x[:, P.T_OBS:] = np.nan | |
| return x | |
| def test_analytic_baselines_do_not_read_future(): | |
| S10 = _frames() | |
| obs_clean = S10[:, :P.T_OBS] | |
| obs_poison = _poison(S10)[:, :P.T_OBS] # identical by construction ... | |
| assert np.array_equal(obs_clean, obs_poison) | |
| preds = dict(B.PREDICTORS) | |
| preds["zv_shrink_0.3"] = lambda o: B.zv_shrink(o, 0.3) | |
| for name, fn in preds.items(): | |
| a, b = fn(obs_clean), fn(obs_poison) | |
| assert np.isfinite(a).all(), name | |
| assert np.array_equal(a, b), name | |
| # predictors must refuse a 10-frame input (they only accept the observed block) | |
| try: | |
| fn(_poison(S10)) | |
| raise AssertionError(f"{name} accepted 10 frames") | |
| except AssertionError as e: | |
| assert "T_OBS" in str(e) or "accepted" not in str(e) | |
| def test_shrink_alpha_fit_uses_val_only(): | |
| # alpha fitting on a poisoned *test* block must not be possible to confuse: the fit only | |
| # takes the val array; here we check the function is deterministic and bounded. | |
| S10 = _frames(40) | |
| a1, _ = B.fit_shrink_alpha(S10, grid=np.linspace(0, 1, 11)) | |
| a2, _ = B.fit_shrink_alpha(S10, grid=np.linspace(0, 1, 11)) | |
| assert a1 == a2 and 0 <= a1 <= 1 | |
| def test_trainer_split_and_model_ignore_future(): | |
| S10 = _frames() | |
| x_clean = torch.as_tensor(S10).float() | |
| x_poison = torch.as_tensor(_poison(S10)).float() | |
| obs_c, fut_c = split_obs_fut(x_clean) | |
| obs_p, fut_p = split_obs_fut(x_poison) | |
| assert torch.equal(obs_c, obs_p) and torch.isnan(fut_p).all() and torch.isfinite(fut_c).all() | |
| torch.manual_seed(0) | |
| model = simlpe_kp.SiMLPeKP(d=64, blocks=2).eval() | |
| with torch.no_grad(): | |
| for p in model.parameters(): # non-trivial weights (out layer is zero-init) | |
| p.add_(0.01 * torch.randn_like(p)) | |
| pa, pb = model(obs_c), model(obs_p) | |
| assert torch.isfinite(pa).all() and torch.equal(pa, pb) | |
| assert pa.shape == (len(S10), P.T_PRED, P.J, 2) | |
| # the model must reject being handed all 10 frames | |
| try: | |
| model(x_poison) | |
| raise RuntimeError("model accepted 10 frames") | |
| except AssertionError: | |
| pass | |
| def test_random_grid_family_contains_protocol_and_fits_45(): | |
| rng = np.random.default_rng(0) | |
| seen_proto = False | |
| proto = P.sample_idx(P.N_FRAMES_STD).tolist() | |
| for _ in range(2000): | |
| g = random_grid(rng) | |
| assert len(g) == P.N_SAMPLE and g[0] >= 0 and g[-1] <= P.N_FRAMES_STD - 1 | |
| assert set(np.diff(g).tolist()) <= {4, 5} | |
| seen_proto |= g.tolist() == proto | |
| assert seen_proto | |
| def test_train_sampler_targets_come_after_obs(): | |
| S45 = np.zeros((4, P.N_FRAMES_STD, P.J, 3), np.float32) | |
| S45[..., 0] = np.arange(P.N_FRAMES_STD)[None, :, None] # x = frame index | |
| smp = TrainSampler(S45, True, seed=0, translate=20.0, scale=0.05) # geometric aug keeps ordering | |
| obs, fut = smp.batch(np.arange(4), torch.device("cpu")) | |
| assert (fut[:, 0, :, 0] > obs[:, -1, :, 0]).all() # every target frame is later than the last obs | |
| assert (fut[:, 1:, :, 0] > fut[:, :-1, :, 0]).all() | |
| # observation noise / joint dropout must touch the observed block only: targets stay exact frame ids | |
| smp = TrainSampler(S45, True, seed=0, noise_px=2.0, joint_drop=0.1) | |
| obs, fut = smp.batch(np.arange(4), torch.device("cpu")) | |
| # first future index >= 5 gaps x 4 = 20 (all-4 grid, start 0); the protocol grid gives 24 | |
| assert torch.equal(fut[..., 0], fut[..., 0].round()) and (fut[..., 0] >= 20).all() | |
| assert not torch.equal(obs[..., 0], obs[..., 0].round()) # noise was applied to obs | |
| assert (obs[..., 2] == 0).any() # some joints dropped (conf -> 0) | |
| if __name__ == "__main__": | |
| for k, f in list(globals().items()): | |
| if k.startswith("test_") and callable(f): | |
| f() | |
| print("ok", k) | |
Xet Storage Details
- Size:
- 5.17 kB
- Xet hash:
- 413dc13b74034553277ea34769f40b94c9791e8542a9ab5153f793489b9afe7b
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.