Buckets:
| #!/usr/bin/env python3 | |
| """tools/review_tool/hon10hz_engaged_hr.py — read-only diagnostic: reconstruct the 10 Hz hands_on series exactly as | |
| data/build_kin_causal.py main_routes() does (raw_from_kpts -> clip_series with the stored route_refs -> f16 round -> < WHEEL_TH) | |
| for the 34 engaged hand_return review items (results/review_tool/items.json, task B) and print, per item, hands_on at | |
| t_event + dt for dt = -1.0..+1.0 (0.1 s), the run lengths of the OFF run ending at t_anchor and the ON run starting at the | |
| switch, which wrist carries the min distance, and whether the other wrist was NaN. Writes results/tables/hr_engaged_34_hon10hz.csv.""" | |
| import json, sys | |
| from pathlib import Path | |
| import numpy as np | |
| CWM3 = Path(__file__).resolve().parents[2] | |
| sys.path.insert(0, str(CWM3 / "data")) | |
| import build_kin_causal as BK # noqa: E402 | |
| KIN = CWM3 / "results/targets_kin_causal_routes_v2" | |
| P2 = CWM3 / "results/pose_causal_routes_v2"; P1 = CWM3 / "results/pose_causal_routes" | |
| items = json.load(open(CWM3 / "results/review_tool/items.json"))["items"] | |
| sel = sorted([i for i in items if i["task"] == "B" and i["meta"]["event"] == "hand_return" and i["meta"]["regime"] == "engaged"], key=lambda i: i["id"]) | |
| refs = np.load(KIN / "route_refs.npz", allow_pickle=True) | |
| kin = np.load(KIN / "kin_causal_engaged_test.npz", allow_pickle=True) | |
| kuid, kt = kin["uid"], kin["t"] | |
| DTS = np.round(np.arange(-1.0, 1.0001, 0.1), 1) | |
| rows = []; hdr = "id,uid,t_anchor,t_event,pose_dir,E0_h0,EM0_h0,S0,S_h0,hon_t+0.1,off_run_frames_ending_at_t,on_run_frames_from_switch,first_on_dt_vs_anchor,min_wrist_at_t,min_wrist_at_te,other_wrist_nan_at_t,other_wrist_nan_at_te,V_t,V_te,d_t,d_te," + ",".join(f"hon{d:+.1f}" for d in DTS) | |
| print(hdr) | |
| for it in sel: | |
| m = it["meta"]; u = m["uid"]; ta = float(m["t_anchor"]); te = float(m["t_event"]) | |
| pp = P2 / f"{u}.npz"; src = "v2" | |
| if not pp.exists(): pp = P1 / f"{u}.npz"; src = "v1" | |
| z = np.load(pp, allow_pickle=True); t = z["t"].astype(np.float64) | |
| F = np.isfinite(z["box"][:, 0]) & np.isfinite(z["kpts"][:, 0, 2].astype(np.float32)) | |
| raw = BK.raw_from_kpts(z["kpts"], F); ref = refs[f"ref__{u}"] | |
| assert len(ref) == len(t) | |
| G5 = BK.clip_series(raw, ref); V = raw["V"] & np.isfinite(G5[:, [0, 6, 7, 10]]).all(1) | |
| G = G5.astype(np.float16).astype(np.float32) | |
| k = np.round(t * 10).astype(np.int64); k0, k1 = int(k.min()), int(k.max()); M = k1 - k0 + 1 | |
| GG = np.full((M, BK.D), np.nan, np.float32); VV = np.zeros(M, bool); GG[k - k0] = G; VV[k - k0] = V | |
| hon = np.where(np.isnan(GG[:, 10]), np.nan, (GG[:, 10] < BK.WHEEL_TH).astype(np.float32)) | |
| ka = int(round(ta * 10)) - k0; ke = int(round(te * 10)) - k0 | |
| # cross-check with the stored targets | |
| r = np.flatnonzero((kuid == u) & np.isclose(kt, ta, atol=1e-6)); assert len(r) == 1, (u, ta, len(r)); r = r[0] | |
| E0 = int(kin["E_c"][r, 0, 0]); EM0 = bool(kin["EM_c"][r, 0, 0]); S0 = kin["S0_c"][r, 0]; Sh0 = kin["S_c"][r, 0, 0] | |
| assert np.isclose(S0, hon[ka], equal_nan=True) and np.isclose(Sh0, hon[ke], equal_nan=True), (u, S0, hon[ka], Sh0, hon[ke]) | |
| def hv(i): return hon[i] if 0 <= i < M else np.nan | |
| # off run ending at ka (consecutive frames with hon==0 up to and including ka) | |
| off = 0; i = ka | |
| while i >= 0 and hon[i] == 0.0: off += 1; i -= 1 | |
| # first on frame after ka | |
| fo = ka + 1 | |
| while fo < M and not (hon[fo] == 1.0): fo += 1 | |
| on = 0; i = fo | |
| while i < M and hon[i] == 1.0: on += 1; i += 1 | |
| def wr(i): | |
| if not (0 <= i < M): return "na", "na", np.nan | |
| L, R = GG[i, 8], GG[i, 9] | |
| if np.isnan(L) and np.isnan(R): return "none", "both", np.nan | |
| if np.isnan(L): return "R", "L_nan", R | |
| if np.isnan(R): return "L", "R_nan", L | |
| return ("L" if L <= R else "R"), "no", min(L, R) | |
| wt, ot, d_t = wr(ka); wte, ote, d_te = wr(ke) | |
| hs = [hv(ke + int(round(d * 10))) for d in DTS] | |
| row = [it["id"], u, f"{ta:.1f}", f"{te:.1f}", src, E0, EM0, S0, Sh0, hv(ka + 1), off, on, f"{(fo - ka) / 10:+.1f}", wt, wte, ot, ote, bool(VV[ka]) if 0 <= ka < M else None, bool(VV[ke]) if 0 <= ke < M else None, f"{d_t:.3f}", f"{d_te:.3f}"] + [("" if np.isnan(h) else int(h)) for h in hs] | |
| rows.append(row); print(",".join(str(x) for x in row)) | |
| out = CWM3 / "results/tables/hr_engaged_34_hon10hz.csv" | |
| out.write_text(hdr + "\n" + "\n".join(",".join(str(x) for x in r) for r in rows) + "\n") | |
| # summary | |
| import collections | |
| off_runs = [r[10] for r in rows]; on_runs = [r[11] for r in rows]; h01 = [r[9] for r in rows] | |
| print("\nSUMMARY n=", len(rows)) | |
| print("pose source:", collections.Counter(r[4] for r in rows)) | |
| print("E_c[h=0]==1 & EM all:", all(r[5] == 1 and r[6] for r in rows)) | |
| print("hon(t+0.1) counts (1=on,0=off,nan):", collections.Counter("nan" if (isinstance(h, float) and np.isnan(h)) else int(h) for h in h01)) | |
| print("OFF run ending at anchor (frames @10Hz): dist", sorted(collections.Counter(off_runs).items()), " median", np.median(off_runs), " n==1:", sum(o == 1 for o in off_runs), " n<=2:", sum(o <= 2 for o in off_runs), " n>=5:", sum(o >= 5 for o in off_runs)) | |
| print("ON run from switch (frames @10Hz): dist", sorted(collections.Counter(on_runs).items()), " median", np.median(on_runs), " n==1:", sum(o == 1 for o in on_runs), " n<=2:", sum(o <= 2 for o in on_runs), " n>=5:", sum(o >= 5 for o in on_runs)) | |
| print("first_on dt vs anchor:", collections.Counter(r[12] for r in rows)) | |
| print("other wrist NaN at t:", collections.Counter(r[15] for r in rows), " at te:", collections.Counter(r[16] for r in rows)) | |
| print("min wrist at t:", collections.Counter(r[13] for r in rows), " at te:", collections.Counter(r[14] for r in rows), " wrist switches L<->R between t and te:", sum(r[13] != r[14] for r in rows)) | |
| print("V(t) true:", sum(bool(r[17]) for r in rows), " V(te) true:", sum(bool(r[18]) for r in rows)) | |
| d_t = np.array([float(r[19]) for r in rows]); d_te = np.array([float(r[20]) for r in rows]) | |
| print("min_hand_rest_dist at t: median %.3f, IQR %.3f-%.3f, n in [0.10,0.12): %d, n>=0.20: %d" % (np.nanmedian(d_t), np.nanpercentile(d_t, 25), np.nanpercentile(d_t, 75), int(((d_t >= 0.10) & (d_t < 0.12)).sum()), int((d_t >= 0.20).sum()))) | |
| print("min_hand_rest_dist at te: median %.3f, IQR %.3f-%.3f, n in (0.08,0.10): %d" % (np.nanmedian(d_te), np.nanpercentile(d_te, 25), np.nanpercentile(d_te, 75), int(((d_te > 0.08) & (d_te < 0.10)).sum()))) | |
| # hon pattern over the 21 pass-3 frames: fraction ON at dt<0 (should be off if a clean return) and dt>=0 | |
| H = np.array([[np.nan if r[21 + j] == "" else float(r[21 + j]) for j in range(len(DTS))] for r in rows]) | |
| neg = DTS < -1e-9; pos = DTS > 1e-9 | |
| print("mean hands_on over items at dt<0 frames: %.3f (nan frac %.3f); at dt>0: %.3f (nan frac %.3f)" % (np.nanmean(H[:, neg]), np.isnan(H[:, neg]).mean(), np.nanmean(H[:, pos]), np.isnan(H[:, pos]).mean())) | |
| print("items with >=1 ON frame in dt in [-1.0,-0.3]:", int((np.nan_to_num(H[:, DTS <= -0.3 + 1e-9], nan=0) == 1).any(1).sum()), " items with all-ON dt in [0,+1.0]:", int((H[:, DTS >= -1e-9] == 1).all(1).sum())) | |
| print("per-dt ON fraction:", " ".join(f"{d:+.1f}:{np.nanmean(H[:, j]):.2f}" for j, d in enumerate(DTS))) | |
| print("wrote", out) | |
Xet Storage Details
- Size:
- 7.12 kB
- Xet hash:
- 31eae19bc48131fc29c6ffc78136e4652d646f161900ce1c1457a20220825628
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.