Buckets:
| """cwm3/data/build_kin_causal_v2p.py — kin version v2p = the v2 causal route-level driver kinematics + THREE keypoint proxies as extra INPUT columns | |
| (plan A arm; owner decision docs/decisions.md row 510; pre-registration row 2026-09-18T05:34:40Z; diagnosis results/tables/pitch_wristface_diag.md). | |
| Step 1 data/build_kin_causal.main_routes UNCHANGED (argv: --source routes --pose-dir results/pose_causal_routes_v2 --out-dir <v2p dir>) -> the 11-channel | |
| kin_causal_* / kinhist_causal_* files, then ASSERTED bit-identical to results/targets_kin_causal_routes_v2 on every shared array (targets, states, | |
| events, validity, row_ok) — the v2p pack changes NOTHING a loss or an evaluation reads. | |
| Step 2 per route (same pose file rule: results/pose_causal_routes_v2/<uid>.npz else results/pose_causal_routes/<uid>.npz) the per-frame proxies of | |
| eval/pitch_wristface_diag.py (frame_proxies / causal_smooth / causal_median_mad: x scaled W/H, y in H units, conf >= 0.3, causal, no DMS field): | |
| col 11 head_pitch_kp 0.5 s trailing mean of (NT_y - eye_mid_y)/IOD MINUS the causal expanding per-route median over MOVING valid frames <= t | |
| (N_MIN 25; NaN while undefined) = the diagnosis' head_pitch_dev; larger = head pitched further DOWN than the running norm | |
| col 12 wrist_face_kp 0.5 s trailing mean of min over wrists |wrist - nose| / shoulder width (smaller = hand nearer the face) | |
| col 13 hands_lap_kp 0.5 s trailing mean over body-valid frames of the mean over both wrists of (wrist_y - shoulder_mid_y) / shoulder width | |
| (positive = wrists below the shoulder line; redefinition of the uninformative hands_lap_headdown_kp, a fixed geometric quantity) | |
| per-anchor extras (read-out only): HD0_c head-down flag at t (head_pitch_kp > K_MAD * MAD_t; NaN undefined), PD0_c pitch dwell = seconds the flag has | |
| held continuously up to t over valid frames (gap > GAP_RESET frames resets; cap PD_CAP s; 0 when off; NaN undefined). | |
| The kinhist files get Kh_c [N,40,14] / Kh8_c [N,80,14] (channels = 14 names); the kin files get K0_c [N,14] (+ channels_k0 14 names, proxy_channels) | |
| + HD0_c + PD0_c; K_c stays [N,25,11] (`channels` 11 = the K TARGET contract of data/cwm3_dataset.py); Kvalid / K0valid / Khvalid / Kh8valid keep the | |
| 11-channel rule (a NaN proxy on a valid step is standardised to the mean by train/batch.std_kh's nan_to_num, like any other NaN channel). | |
| Outputs results/targets_kin_causal_routes_v2p/ + build_stats_v2p_<utc>.json + run json. Nothing under results/targets_kin_causal_routes_v2 is touched. | |
| python data/build_kin_causal_v2p.py [--out-dir results/targets_kin_causal_routes_v2p] [--v2-dir results/targets_kin_causal_routes_v2] [--skip-step1] | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import sys | |
| import time | |
| from collections import defaultdict | |
| from pathlib import Path | |
| import numpy as np | |
| ROOT = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(ROOT / "data")) | |
| sys.path.insert(0, str(ROOT / "eval")) | |
| sys.path.insert(0, str(ROOT / "common")) | |
| import build_kin_causal as BK # noqa: E402 (main_routes, POSE_R, CHANNELS, GAP_RESET) | |
| import pitch_wristface_diag as PW # noqa: E402 (frame_proxies, causal_smooth, causal_median_mad, densify, K_MAD, N_MIN) | |
| from runmeta import write_run_json # noqa: E402 | |
| POSE_V2 = ROOT / "results" / "pose_causal_routes_v2" | |
| POSE_V1 = ROOT / "results" / "pose_causal_routes" | |
| V2_DIR = ROOT / "results" / "targets_kin_causal_routes_v2" | |
| OUT_DEFAULT = ROOT / "results" / "targets_kin_causal_routes_v2p" | |
| PROXY_CHANNELS = ["head_pitch_kp", "wrist_face_kp", "hands_lap_kp"] | |
| CHANNELS_14 = list(BK.CHANNELS) + PROXY_CHANNELS | |
| HIST, HIST8 = 40, 80 | |
| PD_CAP = 30.0 # s, pitch-dwell cap | |
| GAP_RESET = BK.GAP_RESET # 10 frames (> 1 s of undefined frames resets the dwell) | |
| SHARED_KIN = ("uid", "dongle", "t", "row_ok", "route_decoded", "K_c", "Kvalid", "Kframe", "K0valid", "S_c", "S0_c", "E_c", "EM_c", "E0_c", "channels", "event_names", "state_names", "units", "source") | |
| SHARED_HIST = ("uid", "t", "row_ok", "Khvalid", "Kh8valid", "units", "source") | |
| def pose_file(uid): | |
| for d in (POSE_V2, POSE_V1): | |
| p = d / f"{uid}.npz" | |
| if p.exists(): | |
| return p | |
| return None | |
| def lap_depth(kpts, W, H): | |
| """per-frame mean over both wrists of (wrist_y - shoulder_mid_y) / shoulder width (H-unit isotropic geometry of frame_proxies); NaN where the body is invalid.""" | |
| k = np.asarray(kpts, np.float32) | |
| xy = k[:, :, :2].copy(); cf = np.nan_to_num(k[:, :, 2], nan=0.0) | |
| xy[:, :, 0] *= float(W) / float(H) | |
| ok = np.isfinite(xy).all(-1) & (cf >= PW.CONF_TH) | |
| sw = np.linalg.norm(xy[:, PW.LSHO] - xy[:, PW.RSHO], axis=1) | |
| body_valid = ok[:, [PW.NOSE, PW.LSHO, PW.RSHO, PW.LWRI, PW.RWRI]].all(1) & np.isfinite(sw) & (sw > 1e-3) | |
| with np.errstate(invalid="ignore", divide="ignore"): | |
| sho_mid_y = (xy[:, PW.LSHO, 1] + xy[:, PW.RSHO, 1]) / 2 | |
| d = 0.5 * ((xy[:, PW.LWRI, 1] - sho_mid_y) + (xy[:, PW.RWRI, 1] - sho_mid_y)) / sw | |
| return np.where(body_valid, d, np.nan).astype(np.float32) | |
| def pitch_dwell(headdown, valid, hz=10.0, cap=PD_CAP, gap_reset=GAP_RESET): | |
| """headdown [T] bool, valid [T] bool (pitch deviation defined) -> dwell [T] f32: seconds the head-down flag has held continuously up to t over valid frames | |
| (0 on a valid not-down frame; an undefined frame keeps the running value for <= gap_reset frames, longer gaps reset; NaN on undefined frames).""" | |
| T = len(headdown); out = np.full(T, np.nan, np.float32); cur = 0.0; gap = gap_reset + 1 | |
| for i in range(T): | |
| if not valid[i]: | |
| gap += 1 | |
| if gap > gap_reset: | |
| cur = 0.0 | |
| continue | |
| if headdown[i]: | |
| cur = min(cur + 1.0 / hz, cap) | |
| else: | |
| cur = 0.0 | |
| gap = 0 | |
| out[i] = cur | |
| return out | |
| def route_proxy_series(z): | |
| """pose npz -> dict on the dense 0.1 s tick grid (tick = round(10 t)): P [T,3] proxies (NaN undefined), HD [T] flag (NaN undefined), PD [T] dwell, k0 = 0.""" | |
| W, H = [int(v) for v in z["native_wh"]] | |
| kp, moving, _ = PW.densify(z["kpts"], z["moving"].astype(bool), z["t"].astype(np.float64)) | |
| S = PW.route_series(kp, moving, W, H) # pitch_dev (dev from the causal median), wrist_face (smoothed), headdown, mad | |
| lap = PW.causal_smooth(lap_depth(kp, W, H)) | |
| P = np.stack([S["pitch_dev"], S["wrist_face"], lap], 1).astype(np.float32) | |
| dev_ok = np.isfinite(S["pitch_dev"]) & np.isfinite(S["mad"]) | |
| HD = np.where(dev_ok, S["headdown"].astype(np.float32), np.nan).astype(np.float32) | |
| PD = pitch_dwell(S["headdown"], dev_ok) | |
| return dict(P=P, HD=HD, PD=PD) | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--out-dir", default=str(OUT_DEFAULT)) | |
| ap.add_argument("--v2-dir", default=str(V2_DIR)) | |
| ap.add_argument("--regimes", default="manual,engaged") | |
| ap.add_argument("--splits", default="train,val,test") | |
| ap.add_argument("--skip-step1", action="store_true", help="re-use the 11-channel files already present in --out-dir (idempotent re-run of step 2)") | |
| a = ap.parse_args(); t_start = time.time() | |
| out = Path(a.out_dir); out.mkdir(parents=True, exist_ok=True); v2 = Path(a.v2_dir) | |
| regimes, splits = a.regimes.split(","), a.splits.split(",") | |
| # ---------------------------------------------------------------- step 1: the unchanged v2 builder into the v2p dir | |
| if not a.skip_step1: | |
| argv0 = sys.argv | |
| sys.argv = ["build_kin_causal.py", "--source", "routes", "--pose-dir", str(POSE_V2), "--out-dir", str(out), "--regimes", a.regimes, "--splits", a.splits] | |
| try: | |
| BK.main_routes() | |
| finally: | |
| sys.argv = argv0 | |
| # bit-identity of every shared array against v2 | |
| ident = {} | |
| for r in regimes: | |
| for s in splits: | |
| for kind, keys in (("kin", SHARED_KIN), ("kinhist", SHARED_HIST)): | |
| A = np.load(out / f"{kind}_causal_{r}_{s}.npz", allow_pickle=True); B = np.load(v2 / f"{kind}_causal_{r}_{s}.npz", allow_pickle=True) | |
| for k in keys: | |
| if k not in B.files: | |
| continue | |
| x, y = A[k], B[k] | |
| same = x.shape == y.shape and (np.array_equal(x, y, equal_nan=True) if x.dtype.kind in "fc" else np.array_equal(x, y)) | |
| ident[f"{kind}_{r}_{s}:{k}"] = bool(same) | |
| assert same, f"v2p step-1 output differs from v2: {kind}_{r}_{s}:{k}" | |
| # K0_c (11) and Kh_c / Kh8_c (11) of step 1 must equal v2 too (they are widened below) | |
| for k in (("K0_c",) if kind == "kin" else ("Kh_c", "Kh8_c")): | |
| same = np.array_equal(A[k], B[k], equal_nan=True); ident[f"{kind}_{r}_{s}:{k}"] = bool(same); assert same, f"{kind}_{r}_{s}:{k}" | |
| print(f"step 1: {len(ident)} shared arrays bit-identical to {v2.name}", flush=True) | |
| # ---------------------------------------------------------------- step 2: proxies per route, gathered on the anchor grid | |
| uids = sorted(set().union(*[set(np.unique(np.load(out / f"kin_causal_{r}_{s}.npz", allow_pickle=True)["uid"]).tolist()) for r in regimes for s in splits])) | |
| series = {}; n_no_pose = 0; t_last = time.time(); cov = defaultdict(int) | |
| for i, u in enumerate(uids): | |
| pf = pose_file(u) | |
| if pf is None: | |
| n_no_pose += 1; continue | |
| z = np.load(pf, allow_pickle=True) | |
| series[u] = route_proxy_series(z) | |
| cov["frames"] += len(series[u]["PD"]); cov["pitch_defined"] += int(np.isfinite(series[u]["P"][:, 0]).sum()); cov["wf_defined"] += int(np.isfinite(series[u]["P"][:, 1]).sum()) | |
| cov["lap_defined"] += int(np.isfinite(series[u]["P"][:, 2]).sum()); cov["headdown"] += int(np.nan_to_num(series[u]["HD"]).sum()) | |
| if time.time() - t_last > 60: | |
| t_last = time.time(); print(f" routes {i + 1}/{len(uids)} {time.time() - t_start:.0f}s", flush=True) | |
| print(f"step 2: proxy series for {len(series)}/{len(uids)} routes ({n_no_pose} without pose) in {time.time() - t_start:.0f}s; defined shares pitch {cov['pitch_defined'] / max(cov['frames'], 1):.4f} " | |
| f"wrist_face {cov['wf_defined'] / max(cov['frames'], 1):.4f} lap {cov['lap_defined'] / max(cov['frames'], 1):.4f}; head-down flag share {cov['headdown'] / max(cov['pitch_defined'], 1):.4f}", flush=True) | |
| offs_h, offs_h8 = np.arange(-(HIST - 1), 1), np.arange(-(HIST8 - 1), 1) | |
| stats = dict(kin_version="v2p", base="v2 (build_kin_causal.main_routes, pose dir results/pose_causal_routes_v2 else results/pose_causal_routes)", proxy_channels=PROXY_CHANNELS, channels_k0=CHANNELS_14, | |
| proxy_defs={"head_pitch_kp": "0.5 s trailing mean of (NT_y - eye_mid_y)/IOD minus the causal expanding per-route median over moving valid frames <= t (N_MIN 25); larger = head further DOWN", | |
| "wrist_face_kp": "0.5 s trailing mean of min over wrists |wrist - nose| / shoulder width", "hands_lap_kp": "0.5 s trailing mean of the mean wrist drop below the shoulder line / shoulder width (positive = below)", | |
| "HD0_c": f"head-down flag at t: head_pitch_kp > {PW.K_MAD} * MAD_t (causal running median of |pitch - median_t|)", "PD0_c": f"pitch dwell s: seconds the head-down flag has held continuously up to t (gap > {GAP_RESET} frames resets; cap {PD_CAP} s)"}, | |
| params=dict(CONF_TH=PW.CONF_TH, IOD_MIN_PX=PW.IOD_MIN_PX, SMOOTH=PW.SMOOTH, N_MIN=PW.N_MIN, K_MAD=PW.K_MAD, PD_CAP=PD_CAP, GAP_RESET=GAP_RESET), | |
| n_routes=len(uids), n_routes_with_pose=len(series), frame_coverage={k: int(v) for k, v in cov.items()}, shared_arrays_identical_to_v2=all(ident.values()), n_shared_arrays_checked=len(ident), splits={}) | |
| for r in regimes: | |
| for s in splits: | |
| kin_p, hist_p = out / f"kin_causal_{r}_{s}.npz", out / f"kinhist_causal_{r}_{s}.npz" | |
| K = dict(np.load(kin_p, allow_pickle=True)); Hh = dict(np.load(hist_p, allow_pickle=True)) | |
| uid, t = K["uid"], K["t"]; N = len(t) | |
| assert np.array_equal(Hh["uid"], uid) and np.allclose(Hh["t"], t) | |
| assert K["K0_c"].shape[1] == len(BK.CHANNELS) and Hh["Kh_c"].shape[2] == len(BK.CHANNELS), "step-2 input must be the 11-channel files" | |
| Kh = np.full((N, HIST, 3), np.nan, np.float16); Kh8 = np.full((N, HIST8, 3), np.nan, np.float16); K0 = np.full((N, 3), np.nan, np.float32) | |
| HD0 = np.full(N, np.nan, np.float32); PD0 = np.full(N, np.nan, np.float32) | |
| order = np.argsort(uid, kind="stable"); bounds = np.flatnonzero(np.r_[1, uid[order][1:] != uid[order][:-1], 1]) | |
| for a_, b_ in zip(bounds[:-1], bounds[1:]): | |
| rows = order[a_:b_]; u = uid[rows[0]] | |
| if u not in series: | |
| continue | |
| rs = series[u]; P, T = rs["P"], len(rs["PD"]); ka = np.round(t[rows] * 10).astype(np.int64) | |
| def gather(offs): | |
| idx = ka[:, None] + offs[None, :]; ok = (idx >= 0) & (idx < T) | |
| return np.where(ok[..., None], P[np.clip(idx, 0, T - 1)], np.nan) | |
| Kh[rows] = gather(offs_h).astype(np.float16); Kh8[rows] = gather(offs_h8).astype(np.float16) | |
| ok0 = (ka >= 0) & (ka < T); kc = np.clip(ka, 0, T - 1) | |
| K0[rows] = np.where(ok0[:, None], P[kc], np.nan); HD0[rows] = np.where(ok0, rs["HD"][kc], np.nan); PD0[rows] = np.where(ok0, rs["PD"][kc], np.nan) | |
| # rows without a causal box / outside the route series: the v2 rule (row_ok False -> the loader NaNs the whole row anyway); keep proxies as gathered | |
| Kh_c14 = np.concatenate([Hh["Kh_c"], Kh], 2); Kh8_c14 = np.concatenate([Hh["Kh8_c"], Kh8], 2); K0_c14 = np.concatenate([K["K0_c"], K0], 1) | |
| Hh.update(Kh_c=Kh_c14, Kh8_c=Kh8_c14, channels=np.array(CHANNELS_14), kin_version=np.array("v2p"), proxy_channels=np.array(PROXY_CHANNELS)) | |
| K.update(K0_c=K0_c14, channels_k0=np.array(CHANNELS_14), proxy_channels=np.array(PROXY_CHANNELS), HD0_c=HD0, PD0_c=PD0, kin_version=np.array("v2p")) | |
| np.savez_compressed(kin_p, **K); np.savez_compressed(hist_p, **Hh) | |
| ok = K["row_ok"].astype(bool) & K["K0valid"].astype(bool) | |
| st = dict(N=int(N), n_row_ok=int(K["row_ok"].sum()), proxy_defined_given_valid={c: float(np.isfinite(K0[ok, j]).mean()) if ok.any() else None for j, c in enumerate(PROXY_CHANNELS)}, | |
| headdown_share_given_defined=float(np.nanmean(HD0[ok])) if np.isfinite(HD0[ok]).any() else None, pitch_dwell_ge1s_share=float(np.nanmean(PD0[ok] >= 1.0)) if np.isfinite(PD0[ok]).any() else None, | |
| Kh_proxy_nan_frac_row_ok={c: float(np.isnan(Kh[K["row_ok"].astype(bool)][:, :, j].astype(np.float32)).mean()) for j, c in enumerate(PROXY_CHANNELS)}, | |
| proxy_mean_given_valid={c: float(np.nanmean(K0[ok, j])) for j, c in enumerate(PROXY_CHANNELS)}, proxy_sd_given_valid={c: float(np.nanstd(K0[ok, j])) for j, c in enumerate(PROXY_CHANNELS)}) | |
| stats["splits"][f"{r}_{s}"] = st | |
| print(f"{r}/{s}: N={N} row_ok={st['n_row_ok']} proxy defined|valid {st['proxy_defined_given_valid']} head-down share {st['headdown_share_given_defined']} dwell>=1s {st['pitch_dwell_ge1s_share']}", flush=True) | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| (out / f"build_stats_v2p_{stamp}.json").write_text(json.dumps(stats, indent=1)); (out / "build_stats_v2p_latest.json").write_text(json.dumps(stats, indent=1)) | |
| write_run_json(out, "build_kin_causal_v2p", [Path(__file__), Path(BK.__file__), Path(PW.__file__)], [v2 / f"kin_causal_{r}_{s}.npz" for r in regimes for s in splits], None, t_start, | |
| extra=dict(summary={k: v for k, v in stats.items() if k != "splits"}, stats_file=f"build_stats_v2p_{stamp}.json"), input_listing=True) | |
| print("KIN_CAUSAL_V2P_DONE", flush=True) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 16.1 kB
- Xet hash:
- bd456c665ce53f6e67bdcb0dfe5a9011b37b72974619e16ddfaeba985ce5e879
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.