Buckets:
| """Gate 0 — reproduce the published zero-velocity numbers on the official AIDE test split. | |
| python aide/data/verify_protocol.py | |
| Primary protocol = (protocol.DRIVER_RULE, protocol.SAMPLE_GRID, HM over 4 future transitions, | |
| PCK threshold t*max(W,H)) -- these constants were LOCKED from this script's output. PASS iff | |
| All-MPJPE 52.89±0.5, d-nMPJPE 2.40±0.03, HM-MPJPE 139.19±2, HM threshold 72.05±2, | |
| PCK@0.05 85.95±1. | |
| Sensitivity rows: grids round/floor/ceil x driver rules first / largest_bbox / max_score / | |
| iou_track x HM scored over 4 or 5 transitions (5 = incl. last-observed -> first-future) x PCK | |
| normaliser max(W,H) or DIAG. | |
| Writes results/cache/verify_protocol.json (+ run json). Never edits protocol.py. | |
| """ | |
| from __future__ import annotations | |
| import os | |
| # ops rule (docs/decisions.md 2026-09-10, host CPU saturation): cap BLAS/OMP threads before numpy import | |
| for _v in ("OMP_NUM_THREADS", "MKL_NUM_THREADS", "OPENBLAS_NUM_THREADS", "NUMEXPR_NUM_THREADS"): | |
| os.environ.setdefault(_v, "4") | |
| import json | |
| import sys | |
| import time | |
| from multiprocessing import Pool | |
| from pathlib import Path | |
| import numpy as np | |
| ROOT = Path(__file__).resolve().parents[2] | |
| sys.path.insert(0, str(ROOT)) | |
| from aide.data import protocol as P # noqa: E402 | |
| from aide.eval import metrics as M # noqa: E402 | |
| from common.runmeta import write_run_json # noqa: E402 | |
| def _parse(cid): | |
| d = json.load(open(P.AIDE_ANN_DIR / f"{cid}.json")) | |
| pl = d["pose_list"] | |
| out = {"n": len(pl)} | |
| for rule in P.DRIVER_RULES: | |
| kp, _, _, _ = P.select_driver(pl, rule) | |
| out[rule] = kp | |
| return out | |
| def zv_numbers(S10, hm_transitions=P.HM_N_TRANSITIONS, pck_norm="max_side"): | |
| """Zero-velocity: copy last observed pose. S10: [N,10,J,3].""" | |
| obs, fut = S10[:, :P.T_OBS], S10[:, P.T_OBS:] | |
| pred = np.repeat(obs[:, -1:, :, :2], P.T_PRED, axis=1) | |
| err = M.joint_errors(pred, fut) | |
| if hm_transitions == P.HM_N_TRANSITIONS: | |
| m, thr, _ = M.hm_mask(fut) | |
| else: # 5 transitions: include last observed frame | |
| m, thr, _ = M.hm_mask(S10[:, P.T_OBS - 1:], n_transitions=hm_transitions) | |
| r = { | |
| "mpjpe_all": M.mpjpe(err), "dn_mpjpe": M.dn_mpjpe(err), | |
| "mpjpe_hm": M.mpjpe(err[m]), "hm_threshold": thr, "hm_n": int(m.sum()), | |
| "pck05": M.pck(err, 0.05, pck_norm), "pck10": M.pck(err, 0.10, pck_norm), | |
| "pck05_px": P.pck_threshold_px(0.05, pck_norm), | |
| "per_h": M.per_horizon(err).tolist(), "hm_per_h": M.per_horizon(err[m]).tolist(), | |
| } | |
| r.update({f"part_{k}": v for k, v in M.per_part(err).items()}) | |
| r.update({f"hm_part_{k}": v for k, v in M.per_part(err[m]).items()}) | |
| return r | |
| def gate(r): | |
| tgt = P.PAPER["zero_velocity"] | |
| checks = {k: {"ours": r[k], "paper": tgt[k], "tol": P.GATE0_TOL[k], | |
| "pass": abs(r[k] - tgt[k]) <= P.GATE0_TOL[k]} for k in P.GATE0_TOL} | |
| return checks, all(c["pass"] for c in checks.values()) | |
| def main(): | |
| t0 = time.time() | |
| cache = np.load(P.CACHE_NPZ, allow_pickle=False) | |
| test_ids = cache["clip_id"][cache["split"] == "test"].tolist() | |
| assert len(test_ids) == P.EXPECTED_SPLIT_SIZES["test"] | |
| with Pool(12) as pool: | |
| recs = pool.map(_parse, test_ids, chunksize=8) | |
| rows = [] | |
| for rule in P.DRIVER_RULES: | |
| for grid in P.SAMPLE_GRIDS: | |
| S10 = np.stack([r[rule][P.sample_idx(r["n"], grid)] for r in recs]) | |
| for hmt in (4, 5): | |
| for pn in ("max_side", "diag"): | |
| num = zv_numbers(S10, hm_transitions=hmt, pck_norm=pn) | |
| checks, ok = gate(num) | |
| rows.append({"driver_rule": rule, "grid": grid, "hm_transitions": hmt, "pck_norm": pn, | |
| "primary": (rule == P.DRIVER_RULE and grid == P.SAMPLE_GRID | |
| and hmt == P.HM_N_TRANSITIONS and pn == "max_side"), | |
| **num, "gate": checks, "pass": ok}) | |
| # consistency: cache S10 (built with the primary rule) must equal the re-parsed primary | |
| S10c = cache["S10"][cache["split"] == "test"] | |
| S10p = np.stack([r[P.DRIVER_RULE][P.sample_idx(r["n"])] for r in recs]) | |
| assert np.allclose(S10c, S10p), "cache S10 differs from re-parsed protocol frames" | |
| S10a = np.stack([r[P.ALT_DRIVER_RULE][P.sample_idx(r["n"], P.ALT_SAMPLE_GRID)] for r in recs]) | |
| assert np.allclose(cache["S10_alt"][cache["split"] == "test"], S10a), "cache S10_alt mismatch" | |
| primary = [r for r in rows if r["primary"]][0] | |
| res = { | |
| "split": "test", "n_test": len(test_ids), "paper": P.PAPER["zero_velocity"], | |
| "tolerance": P.GATE0_TOL, "primary": primary, "PASS": primary["pass"], "rows": rows, | |
| "best_row_by_abs_err_sum": None, "wall_s": round(time.time() - t0, 1), | |
| } | |
| # which configuration reproduces best (sum of |ours - paper| / tol) | |
| def score(r): | |
| return sum(abs(r[k] - P.PAPER["zero_velocity"][k]) / P.GATE0_TOL[k] for k in P.GATE0_TOL) | |
| best = min(rows, key=score) | |
| res["best_row_by_abs_err_sum"] = {k: best[k] for k in ("driver_rule", "grid", "hm_transitions", "pck_norm", | |
| "mpjpe_all", "dn_mpjpe", "mpjpe_hm", | |
| "hm_threshold", "pck05", "pass")} | |
| res["protocol_constants"] = {"DRIVER_RULE": P.DRIVER_RULE, "SAMPLE_GRID": P.SAMPLE_GRID, | |
| "sample_idx_45": P.sample_idx(45).tolist(), "HM_N_TRANSITIONS": P.HM_N_TRANSITIONS, | |
| "PCK_NORM_px": P.PCK_NORM, "pck05_px": P.pck_threshold_px(0.05)} | |
| out = P.RESULTS / "cache" / "verify_protocol.json" | |
| out.write_text(json.dumps(res, indent=1)) | |
| hdr = f"{'rule':13s} {'grid':6s} hmT {'pckN':8s} {'All':>7s} {'d-n':>6s} {'HM':>8s} {'HMthr':>7s} {'PCK05':>7s} {'PCK10':>7s} pass" | |
| print(hdr) | |
| print(f"{'paper':13s} {'':6s} {'':3s} {'':8s} {52.89:7.2f} {2.40:6.2f} {139.19:8.2f} {72.05:7.2f} {85.95:7.2f} {'':7s}") | |
| for r in rows: | |
| flag = "*" if r["primary"] else " " | |
| print(f"{r['driver_rule']:13s} {r['grid']:6s} {r['hm_transitions']:^3d} {r['pck_norm']:8s} {r['mpjpe_all']:7.2f} " | |
| f"{r['dn_mpjpe']:6.3f} {r['mpjpe_hm']:8.2f} {r['hm_threshold']:7.2f} {r['pck05']:7.2f} " | |
| f"{r['pck10']:7.2f} {'PASS' if r['pass'] else 'FAIL'}{flag}") | |
| print(f"\nPRIMARY ({P.DRIVER_RULE}, {P.SAMPLE_GRID}, {P.HM_N_TRANSITIONS} transitions, PCK norm max(W,H)):", | |
| "PASS" if res["PASS"] else "FAIL") | |
| for k, c in primary["gate"].items(): | |
| print(f" {k:13s} ours={c['ours']:8.3f} paper={c['paper']:8.2f} tol={c['tol']:.2f} -> {'ok' if c['pass'] else 'FAIL'}") | |
| print("per-horizon ZV (px):", [round(x, 2) for x in primary["per_h"]]) | |
| print("HM per-horizon ZV (px):", [round(x, 2) for x in primary["hm_per_h"]]) | |
| print("per-part ZV:", {k: round(primary[f'part_{k}'], 2) for k in P.PART_NAMES}) | |
| print("HM per-part ZV:", {k: round(primary[f'hm_part_{k}'], 2) for k in P.PART_NAMES}) | |
| print("best row:", res["best_row_by_abs_err_sum"]) | |
| print("saved", out) | |
| write_run_json(out.parent, "verify_protocol", [__file__, P.__file__, M.__file__], [P.CACHE_NPZ], | |
| seed=None, t_start=t0, extra={"PASS": res["PASS"], "primary": { | |
| k: primary[k] for k in ("mpjpe_all", "dn_mpjpe", "mpjpe_hm", "hm_threshold", "pck05")}}, | |
| input_listing=False) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 7.45 kB
- Xet hash:
- dbd179c250e14e8ef1ca3fc7e9aa692e0890a8e39b175f0c517c5a2f548372c4
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.