Buckets:
| """cwm3/data/build_kp_residual_store.py — LANE F1 (R2 WP4 batch 2; pre-registration docs/decisions.md 2026-09-17T19:33:12Z): the keypoint store and anchor | |
| index of the keypoint RESIDUAL forecaster (models/kp_forecaster.py, train/train_kp_residual.py). | |
| python data/build_kp_residual_store.py [--out results/kp_residual] | |
| Source (label-free, causal, our own extraction): the route-level causal RTMW / HALPE-136 pose at 10 Hz, results/pose_causal_routes_v2/<uid>.npz first (the 74 | |
| co-occupant routes re-extracted with a population driver-side prior), else results/pose_causal_routes/<uid>.npz — exactly the directory rule of the kin-v2 targets | |
| (build_stats_latest.json: pose_dir / pose_fallback_dir). Only the 11 UPPER-BODY joints are kept (HALPE 0-10 = nose, eyes, ears, shoulders, elbows, wrists; x/W, | |
| y/H, conf; valid = stored mask = finite ∧ conf >= 0.3). | |
| Outputs (new files only): | |
| store.npz kp f16[F,11,3] (x, y, conf; NaN where no box), valid bool[F,11], tick i64[F] (round(10 t)), route i32[F], lut i32[L] (global frame index per | |
| route-local tick or -1: the frame grid of a route has gaps between its extraction spans), route_names, route_tick0 i64[R], route_lut_off i64[R], | |
| route_lut_len i64[R], joints, source_dir per route | |
| anchors_{regime}_{split}.npz uid, t, dongle, mode u8, route i32, lut_pos i64 (position of the anchor tick in `lut`; -1 = the route has no pose frames), S f32[N,3,2] | |
| (hands_on / head_turned at +1 / +2 / +3 s = S_c steps 4 / 9 / 14 of results/targets_kin_causal_routes_v2, NaN censored), S0 f32[N,2], kin_row_ok bool | |
| build_kp_residual_store_<utc>.json (+ _latest) — counts, coverage, run record (common/runmeta.write_run_json) | |
| Row order = the s12 anchor grid (s12_wm/results/targets_{regime}_{split}.npz; uid / t asserted against the kin-v2 file). Nothing here reads a CAN column, a | |
| label or an openpilot DMS field; the future frames are NOT materialised — the trainer gathers observation and target frames from `lut` (tests/test_kp_forecaster.py | |
| asserts that the observation offsets are all <= 0). | |
| `--holdout` (2026-09-18, candidate (iii) hold-out confirmation, pre-registration docs/decisions.md 2026-09-18T03:41:14Z): the SAME builder on the BATON-holdout grid | |
| (split `holdout` only) with the hold-out pose extraction results/holdout/pose_causal_routes (fallback = the benchmark results/pose_causal_routes, the kin-v2 hold-out | |
| build's rule) and the hold-out kin-v2 targets -> results/holdout/kp_residual/ (new files only; the benchmark store is untouched; defaults unchanged — | |
| tests/test_holdout_kpf_overrides.py). | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import os | |
| import sys | |
| import time | |
| from pathlib import Path | |
| import numpy as np | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| EXPLORE = CWM3.parent | |
| sys.path.insert(0, str(CWM3 / "common")) | |
| from runmeta import write_run_json # noqa: E402 | |
| S12R = EXPLORE / "s12_wm" / "results" | |
| POSE_V2 = CWM3 / "results" / "pose_causal_routes_v2" | |
| POSE_V1 = CWM3 / "results" / "pose_causal_routes" | |
| KIN_V2 = CWM3 / "results" / "targets_kin_causal_routes_v2" | |
| OUT = CWM3 / "results" / "kp_residual" | |
| JOINTS = ["nose", "l_eye", "r_eye", "l_ear", "r_ear", "l_shoulder", "r_shoulder", "l_elbow", "r_elbow", "l_wrist", "r_wrist"] # HALPE-136 / COCO body 0-10 | |
| NJ = len(JOINTS) | |
| STATE_STEPS = (4, 9, 14) # S_c steps (+0.2 s grid) = +1 / +2 / +3 s | |
| REGIMES, SPLITS = ("manual", "engaged"), ("train", "val", "test") | |
| MODE = {"manual": 0, "engaged": 1} | |
| HOLDOUT_ROOT = Path(os.environ.get("CWM3_HOLDOUT_ROOT", str(CWM3 / "results" / "holdout"))) # --holdout root (build_road_dynamics_targets.py --holdout convention) | |
| def holdout_override(root=None): | |
| """--holdout (candidate (iii), pre-registration 2026-09-18T03:41:14Z): point the module at the BATON-holdout artefacts — pose dir results/holdout/pose_causal_routes | |
| (driver-side prior = population median, runbook §2; fallback = the benchmark results/pose_causal_routes exactly as the kin-v2 hold-out build), kin-v2 dir | |
| results/holdout/targets_kin_causal_routes_v2, split `holdout` only, output results/holdout/kp_residual. Returns the previous values (tests restore them). | |
| Nothing else changes: same joints, same tick grid, same anchor / lut construction, same S / S0 fields.""" | |
| global POSE_V2, POSE_V1, KIN_V2, OUT, SPLITS | |
| root = Path(root) if root else HOLDOUT_ROOT | |
| prev = {"POSE_V2": POSE_V2, "POSE_V1": POSE_V1, "KIN_V2": KIN_V2, "OUT": OUT, "SPLITS": SPLITS} | |
| POSE_V2 = root / "pose_causal_routes" | |
| POSE_V1 = CWM3 / "results" / "pose_causal_routes" | |
| KIN_V2 = root / "targets_kin_causal_routes_v2" | |
| OUT = root / "kp_residual" | |
| SPLITS = ("holdout",) | |
| return prev | |
| def restore(prev): | |
| global POSE_V2, POSE_V1, KIN_V2, OUT, SPLITS | |
| POSE_V2, POSE_V1, KIN_V2, OUT, SPLITS = prev["POSE_V2"], prev["POSE_V1"], prev["KIN_V2"], prev["OUT"], prev["SPLITS"] | |
| def pose_file(uid): | |
| p2, p1 = POSE_V2 / f"{uid}.npz", POSE_V1 / f"{uid}.npz" | |
| return p2 if p2.exists() else (p1 if p1.exists() else None) | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--out", default="", help="default results/kp_residual (results/holdout/kp_residual with --holdout)") | |
| ap.add_argument("--holdout", action="store_true", help="candidate (iii) hold-out confirmation: the same builder on the BATON-holdout grid / pose / kin-v2 artefacts (holdout_override)") | |
| a = ap.parse_args() | |
| if a.holdout: | |
| holdout_override() | |
| out = Path(a.out) if a.out else OUT | |
| out.mkdir(parents=True, exist_ok=True) | |
| t_start = time.time() | |
| # ---------------------------------------------------------------- routes of the s12 grid | |
| grids = {} | |
| routes = [] | |
| for r in REGIMES: | |
| for s in SPLITS: | |
| z = np.load(S12R / f"targets_{r}_{s}.npz", allow_pickle=True) | |
| grids[(r, s)] = {"uid": z["uid"].astype(str), "t": z["t"].astype(np.float64), "dongle": z["dongle"].astype(str)} | |
| routes += list(np.unique(grids[(r, s)]["uid"])) | |
| routes = sorted(set(routes)) | |
| # ---------------------------------------------------------------- pose store | |
| kp, valid, tick, route_of = [], [], [], [] | |
| lut_parts, tick0, lut_off, lut_len, src = [], [], [], [], [] | |
| off_frames, off_lut = 0, 0 | |
| inputs = [] | |
| n_frames_total = 0 | |
| for ri, uid in enumerate(routes): | |
| p = pose_file(uid) | |
| if p is None: | |
| tick0.append(0); lut_off.append(off_lut); lut_len.append(0); src.append("") | |
| continue | |
| inputs.append(p) | |
| with np.load(p, allow_pickle=True) as z: | |
| t = z["t"].astype(np.float64) | |
| k = z["kpts"][:, :NJ, :].astype(np.float16) | |
| m = z["mask"][:, :NJ].astype(bool) | |
| tk = np.round(t * 10.0).astype(np.int64) | |
| order = np.argsort(tk, kind="mergesort") | |
| tk, k, m = tk[order], k[order], m[order] | |
| keep = np.concatenate([[True], np.diff(tk) > 0]) # duplicate ticks (none expected) -> first kept | |
| tk, k, m = tk[keep], k[keep], m[keep] | |
| n = len(tk) | |
| t0 = int(tk[0]) | |
| L = int(tk[-1] - t0 + 1) | |
| lut = np.full(L, -1, np.int32) | |
| lut[tk - t0] = np.arange(n, dtype=np.int32) + off_frames | |
| kp.append(k); valid.append(m); tick.append(tk); route_of.append(np.full(n, ri, np.int32)) | |
| lut_parts.append(lut); tick0.append(t0); lut_off.append(off_lut); lut_len.append(L); src.append(p.parent.name) | |
| off_frames += n; off_lut += L; n_frames_total += n | |
| kp = np.concatenate(kp); valid = np.concatenate(valid); tick = np.concatenate(tick); route_of = np.concatenate(route_of) | |
| lut = np.concatenate(lut_parts) | |
| tick0, lut_off, lut_len = np.asarray(tick0, np.int64), np.asarray(lut_off, np.int64), np.asarray(lut_len, np.int64) | |
| np.savez(out / "store.npz", kp=kp, valid=valid, tick=tick, route=route_of, lut=lut, route_names=np.asarray(routes), route_tick0=tick0, | |
| route_lut_off=lut_off, route_lut_len=lut_len, joints=np.asarray(JOINTS), source_dir=np.asarray(src), | |
| note="HALPE 0-10 upper body; x/W, y/H, conf; valid = finite & conf >= 0.3; lut = global frame index per route tick (0.1 s) or -1") | |
| print(f"store: {len(routes)} routes, {n_frames_total:,} frames, lut {len(lut):,}, kp {kp.nbytes / 2**20:.0f} MiB; routes without pose: {int((lut_len == 0).sum())}") | |
| # ---------------------------------------------------------------- anchors | |
| ridx = {u: i for i, u in enumerate(routes)} | |
| stats = {"routes": len(routes), "frames": int(n_frames_total), "routes_without_pose": int((lut_len == 0).sum()), "splits": {}, | |
| "holdout": bool(a.holdout), "pose_dir": str(POSE_V2), "pose_fallback_dir": str(POSE_V1), "kin_dir": str(KIN_V2), "splits_built": list(SPLITS), | |
| "routes_from_pose_dir": int(sum(1 for x in src if x == POSE_V2.name)), "routes_from_fallback": int(sum(1 for x in src if x and x != POSE_V2.name))} | |
| for (r, s), g in grids.items(): | |
| N = len(g["uid"]) | |
| kz = np.load(KIN_V2 / f"kin_causal_{r}_{s}.npz", allow_pickle=True) | |
| assert np.array_equal(kz["uid"].astype(str), g["uid"]) and np.allclose(kz["t"], g["t"]), f"kin v2 {r}/{s}: uid/t witness" | |
| S = kz["S_c"][:, list(STATE_STEPS), :].astype(np.float32) | |
| S0 = kz["S0_c"].astype(np.float32) | |
| row_ok = kz["row_ok"].astype(bool) | |
| rt = np.asarray([ridx[u] for u in g["uid"]], np.int32) | |
| atick = np.round(g["t"] * 10.0).astype(np.int64) | |
| rel = atick - tick0[rt] | |
| inside = (lut_len[rt] > 0) & (rel >= 0) & (rel < lut_len[rt]) | |
| pos = np.where(inside, lut_off[rt] + rel, -1) | |
| has_frame = inside & (lut[np.where(inside, pos, 0)] >= 0) | |
| pos = np.where(has_frame, pos, -1).astype(np.int64) | |
| # uid is NOT stored (uid == route_names[route]; the 2.6 M strings were 1.4 GB) — readers rebuild it from `route` | |
| np.savez(out / f"anchors_{r}_{s}.npz", t=g["t"], dongle=g["dongle"], mode=np.full(N, MODE[r], np.uint8), route=rt, lut_pos=pos, | |
| S=S, S0=S0, kin_row_ok=row_ok, state_names=np.asarray(["hands_on", "head_turned"]), state_horizons_s=np.asarray([1.0, 2.0, 3.0])) | |
| stats["splits"][f"{r}_{s}"] = {"N": int(N), "anchor_frame_present": float(has_frame.mean()), "kin_row_ok": float(row_ok.mean()), | |
| "S_finite_3s": float(np.isfinite(S[:, 2, :]).all(1).mean())} | |
| print(f"anchors {r}/{s}: N={N:,} frame present {has_frame.mean():.4f} kin_row_ok {row_ok.mean():.4f}") | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| (out / f"build_kp_residual_store_{stamp}.json").write_text(json.dumps(stats, indent=1)) | |
| (out / "build_kp_residual_store_latest.json").write_text(json.dumps(stats, indent=1)) | |
| write_run_json(out, "build_kp_residual_store", [Path(__file__)], inputs + [KIN_V2 / f"kin_causal_{r}_{s}.npz" for r in REGIMES for s in SPLITS], | |
| seed=0, t_start=t_start, extra={"stats": stats, "joints": JOINTS, "dms_fields_used": []}, input_listing=False) | |
| print("DONE", time.time() - t_start) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 11.2 kB
- Xet hash:
- 374d191e1b857a5d7600091b1cd5575a8623254f521f5a47d0112f5e17642587
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.