Buckets:
| """pack_v2 alignment contract (cwm3/data/build_pack_v2.py): (a) shift test — a synthetic route whose container fps != 20 is re-indexed | |
| to frame/20 exactly and its drifting labels are undone; (b) on the materialised packs: row order == s12, fps == 20 routes are | |
| byte-identical to v1 under the v1 basis, and the alignment report meets the acceptance (|lag| <= 0.25 s for >= 99.5 %).""" | |
| import json, sys | |
| from pathlib import Path | |
| import numpy as np, pytest | |
| ROOT = Path(__file__).resolve().parents[1]; sys.path.insert(0, str(ROOT / "data")) | |
| OUT = ROOT / "results" / "pack_v2"; S12 = Path("/home/USER/s12_wm/results") | |
| def test_shift_reindex_synthetic(): | |
| import build_pack_v2 as B | |
| n_frames, fps = 120_000, 19.998063631942284 # a real road-camera avg fps | |
| dur = n_frames / fps; ts = np.arange(2.0, dur, 0.5) # s5 label axis (container fps) | |
| fe = np.round(ts * fps).astype(np.int64); tt = fe / 20.0 # v2: true frame/20 time of the clip END | |
| drift = ts - tt # v1 lag | |
| assert drift[-1] > 0.5 and abs(drift[-1] - ts[-1] * (1 - fps / 20.0)) < 0.05 and np.all(np.diff(tt) > 0) # drift grows ~ t·(1−fps/20) | |
| # pooled window: anchors at t use clips with true END in [t-2, t] -> the last clip's true time is <= t always (causal), and the | |
| # v1 rule on the label axis would pick clips whose true time is EARLIER by `drift` (stale) — never later than t | |
| F = np.random.default_rng(0).standard_normal((len(ts), 8)).astype(np.float16) | |
| t = np.array([1000.0, 3000.0, 5900.0]); P, ok, last = B.pool_rows(tt, F, t) | |
| assert ok.all() and np.all(tt[last] <= t + 1e-6) and np.all(t - tt[last] < 0.5 + 1e-6) # v2: causal, at most one slot stale | |
| hi1 = np.searchsorted(ts, t + 1e-6, side="right") - 1 | |
| assert np.all(tt[hi1] <= tt[last] + 1e-9), "v1 (label axis) never picks later content than v2" | |
| assert (t - tt[hi1])[-1] > 0.5, "late in the route the v1 content is more than one slot stale (drift ~0.57 s at 5900 s)" | |
| # fps == 20 -> identity | |
| tt20 = np.round(ts * 20.0) / 20.0; assert np.allclose(tt20, ts) | |
| def test_pack_layout_and_identity(stream, tag): | |
| p = OUT / f"{tag}_manual_val.npz" | |
| if not p.exists(): pytest.skip("pack_v2 not built") | |
| z = np.load(p, allow_pickle=True); s = np.load(S12 / "targets_manual_val.npz", allow_pickle=True) | |
| assert z["Z"].shape == (len(s["t"]), 256) and z["Z"].dtype == np.float16 and z["row_ok"].shape == (len(s["t"]),) and int(z["version"]) == 2 | |
| assert np.all(z["Z"][~z["row_ok"]] == 0) | |
| rep = json.load(open(OUT / "alignment.json")) | |
| idn = rep["streams"]["manual_val"]["identical_to_v1_with_v1_basis"][stream]["frac_identical"] | |
| fps_eq = rep["fps"]["fps_eq_20"]["front" if stream == "Vf" else "cabin"] | |
| assert idn is not None and (idn > 0 or fps_eq == 0), "fps == 20 routes must be byte-identical to v1 under the v1 basis" | |
| def test_alignment_acceptance(): | |
| p = OUT / "alignment.json" | |
| if not p.exists(): pytest.skip("pack_v2 not built") | |
| rep = json.load(open(p)) | |
| for k, v in rep["streams"].items(): | |
| if k.startswith("pca_"): continue | |
| for cam in ("front", "cabin"): | |
| assert v["lag_v2"][cam]["le_0p25"] >= 0.995 and v["stale_v2"][cam]["future_content_frac"] == 0.0 and v["stale_v2"][cam]["min"] >= -1e-6 | |
| if rep.get("seqcheck"): | |
| sq = rep["seqcheck"]; assert sq["v2_resolvable"]["le_0p25_anchor_weighted"] >= 0.995, "sequence-level v2 alignment below acceptance" | |
| assert sq["v2_resolvable"]["ge_1s"] < sq["v1_resolvable"]["ge_1s"] or sq["v1_resolvable"]["ge_1s"] == 0.0 | |
| if rep.get("empirical"): # per-anchor argmax is only exact on fps == 20 routes; here it must at least beat v1's audit numbers | |
| for k, v in rep["empirical"].items(): | |
| if v.get("v1_audit") and v["v1_audit"].get("frac_abs_lag_ge_1s") is not None: assert v["v2"]["ge_1s"] < v["v1_audit"]["frac_abs_lag_ge_1s"], k | |
| def test_future_layout(): | |
| p = OUT / "future_manual_val.npz" | |
| if not p.exists(): pytest.skip("pack_v2 future not built") | |
| z = np.load(p, allow_pickle=True); s = np.load(S12 / "targets_manual_val.npz", allow_pickle=True); v1 = np.load(S12 / "future_manual_val.npz") | |
| assert z["F"].shape == v1["F"].shape == (len(s["t"]), 5, 256) and z["ok"].shape == v1["ok"].shape and z["F"].dtype == np.float16 | |
| assert np.all(z["F"][~z["ok"]] == 0) | |
Xet Storage Details
- Size:
- 4.49 kB
- Xet hash:
- 3b62ed7742864936c685f64743b0b769b094163ce503144ca95ee5429e601fe1
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.