Buckets:
| """cwm3/eval/pitch_wristface_diag.py — ZERO-TRAINING separability diagnosis of keypoint-derived head-pitch / wrist-face / hands-lap-head-down | |
| proxies on phone / object / gaze-away episodes vs matched normal windows on the TEST routes (plan A diagnosis; pre-registration row 511 of | |
| docs/decisions.md, 2026-09-18; owner decision row 510). Labels of the TEST routes are used ONLY to measure separability — nothing is fitted. | |
| Inputs (label-free, our own causal HALPE-136 pose at 10 Hz): results/pose_causal_routes_v2/<uid>.npz else results/pose_causal_routes/<uid>.npz | |
| (the kin-v2 directory rule). Nothing here reads an openpilot DMS quantity; proxy names carry the `_kp` suffix (our own keypoint geometry). | |
| Proxies per frame (causal): | |
| head_pitch_kp (NT_y - eye_mid_y) / IOD, NT = HALPE 56, eye centres = mean of the >= 5 valid landmarks of 62-67 / 68-73, IOD >= 10 px native; | |
| image y points down -> larger = nose lower relative to the eyes = head pitched DOWN; trailing 0.5 s mean over valid frames | |
| head_pitch_dev head_pitch_kp - causal expanding per-route median over MOVING valid frames <= t (N_MIN 25); MAD_t = causal running median of | |
| |head_pitch_kp - median_t|; head-down flag = head_pitch_dev > K_MAD * MAD_t | |
| wrist_face_kp min over wrists (9 / 10) of |wrist - nose(0)| / shoulder width (|5 - 6|), trailing 0.5 s mean | |
| hands_lap_headdown_kp both wrists with wrist_y - shoulder_mid_y > 1.0 shoulder width AND head-down flag | |
| Window scores (window = [t, t + 10 s]): primary = max head_pitch_dev, max(-wrist_face_kp), share of valid frames flagged hands_lap_headdown_kp; | |
| secondary = window means; combination = mean of the pooled ranks of the three primary scores (nothing fitted). | |
| Metrics: AUROC pooled / per regime with 95 % percentile CI from 2,000 device-cluster (dongle) bootstrap resamples; face / body / box coverage; | |
| share of positives where the existing yaw proxy gaze_off (face pack F0_c[:, 7]) never flags but the head-down flag holds >= 10 valid frames. | |
| RULE (pre-registered): separable iff best single primary proxy OR the rank combination reaches pooled AUROC >= 0.80 with CI-LB >= 0.70 AND face coverage >= 0.8. | |
| python eval/pitch_wristface_diag.py [--out results/tables/pitch_wristface_diag] [--n-boot 2000] | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import bisect | |
| import json | |
| import re | |
| import sys | |
| import time | |
| from collections import defaultdict | |
| from pathlib import Path | |
| import numpy as np | |
| ROOT = Path(__file__).resolve().parents[1] | |
| EXPLORE = ROOT.parent | |
| POSE_V2 = ROOT / "results" / "pose_causal_routes_v2" | |
| POSE_V1 = ROOT / "results" / "pose_causal_routes" | |
| KIN_V2 = ROOT / "results" / "targets_kin_causal_routes_v2" | |
| KIN_V3 = ROOT / "results" / "targets_kin_causal_routes_v3" | |
| GAZE = ROOT / "results" / "targets_gaze_causal_routes_v2" | |
| FACE = ROOT / "results" / "targets_face_causal_routes_v2" | |
| REVIEW = ROOT / "results" / "review_tool" | |
| NEWTEST = EXPLORE / "cwm2" / "annotation_human" / "labels_newtest.jsonl" | |
| CLIPS_NEWTEST = EXPLORE / "cwm2" / "annotation_human" / "clips_newtest.json" | |
| CONF_TH = 0.30 | |
| NT, RE, LE = 56, list(range(62, 68)), list(range(68, 74)) | |
| NOSE, LSHO, RSHO, LWRI, RWRI = 0, 5, 6, 9, 10 | |
| IOD_MIN_PX = 10.0 | |
| SMOOTH = 5 # frames (0.5 s at 10 Hz) | |
| N_MIN = 25 # moving valid frames before the causal median / MAD are defined | |
| K_MAD = 3.0 | |
| LAP_SW = 1.0 # wrist below the shoulder line by > 1 shoulder width | |
| WIN_S = 10.0 | |
| CLEAR_S = 60.0 | |
| MERGE_S = 10.0 | |
| NEG_PER_POS = 3 | |
| MOVING_MIN = 0.8 | |
| HEADDOWN_MIN_FRAMES = 10 | |
| REGIMES = ("manual", "engaged") | |
| NOTE_RE = re.compile(r"phone|item", re.I) | |
| NEWTEST_OBJ = {"phone", "other_object", "food_or_drink"} | |
| NEWTEST_CODES = {"sustained_gaze_away_with_demand", "object_interaction_while_moving", "hands_off_with_demand"} | |
| # ----------------------------------------------------------------------------- per-frame proxies (pure functions; unit-tested) | |
| def frame_proxies(kpts, W, H): | |
| """kpts f16/f32 [T,136,3] normalised (x/W, y/H, conf; NaN where no box) -> dict of per-frame raw quantities on the H-unit scale. | |
| Returns head_pitch (raw, NaN invalid), wrist_face (NaN invalid), lap (bool, both wrists below the shoulder line), face_valid, body_valid, box.""" | |
| 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) | |
| box = np.isfinite(k[:, NOSE, :2]).all(-1) | np.isfinite(k[:, NT, :2]).all(-1) | np.isfinite(k[:, LSHO, :2]).all(-1) | |
| ok = np.isfinite(xy).all(-1) & (cf >= CONF_TH) | |
| # --- face: eye centres, IOD, pitch | |
| def centre(idx): | |
| m = ok[:, idx]; w = m.astype(np.float32)[:, :, None] | |
| with np.errstate(invalid="ignore", divide="ignore"): | |
| c = (np.nan_to_num(xy[:, idx]) * w).sum(1) / np.maximum(w.sum(1), 1e-6) | |
| return c, m.sum(1) | |
| re_c, n_re = centre(RE); le_c, n_le = centre(LE) | |
| d = le_c - re_c; iod = np.hypot(d[:, 0], d[:, 1]) | |
| eyes_ok = (n_re >= 5) & (n_le >= 5) & (iod * float(H) >= IOD_MIN_PX) | |
| mid = (re_c + le_c) / 2 | |
| with np.errstate(invalid="ignore", divide="ignore"): | |
| pitch = (xy[:, NT, 1] - mid[:, 1]) / iod | |
| face_valid = ok[:, NT] & eyes_ok & np.isfinite(pitch) | |
| pitch = np.where(face_valid, pitch, np.nan).astype(np.float32) | |
| # --- body: wrist-face distance / shoulder width, lap flag | |
| sw = np.linalg.norm(xy[:, LSHO] - xy[:, RSHO], axis=1) | |
| body_valid = ok[:, [NOSE, LSHO, RSHO, LWRI, RWRI]].all(1) & np.isfinite(sw) & (sw > 1e-3) | |
| with np.errstate(invalid="ignore", divide="ignore"): | |
| dl = np.linalg.norm(xy[:, LWRI] - xy[:, NOSE], axis=1) / sw | |
| dr = np.linalg.norm(xy[:, RWRI] - xy[:, NOSE], axis=1) / sw | |
| wf = np.minimum(dl, dr) | |
| sho_mid_y = (xy[:, LSHO, 1] + xy[:, RSHO, 1]) / 2 | |
| lap = ((xy[:, LWRI, 1] - sho_mid_y) > LAP_SW * sw) & ((xy[:, RWRI, 1] - sho_mid_y) > LAP_SW * sw) | |
| wf = np.where(body_valid, wf, np.nan).astype(np.float32) | |
| lap = lap & body_valid | |
| return dict(head_pitch=pitch, wrist_face=wf, lap=lap, face_valid=face_valid, body_valid=body_valid, box=box) | |
| def causal_smooth(x, n=SMOOTH): | |
| """trailing mean over the valid (finite) values among the last n frames (incl. t); NaN when none. Causal by construction.""" | |
| x = np.asarray(x, np.float32); T = len(x); out = np.full(T, np.nan, np.float32) | |
| v = np.isfinite(x); xz = np.where(v, x, 0.0).astype(np.float64) | |
| cs = np.concatenate([[0.0], np.cumsum(xz)]); cn = np.concatenate([[0], np.cumsum(v.astype(np.int64))]) | |
| idx = np.arange(T); lo = np.maximum(idx - n + 1, 0) | |
| s = cs[idx + 1] - cs[lo]; c = cn[idx + 1] - cn[lo] | |
| with np.errstate(invalid="ignore", divide="ignore"): | |
| out = np.where(c > 0, s / np.maximum(c, 1), np.nan).astype(np.float32) | |
| return out | |
| def causal_median_mad(x, moving, n_min=N_MIN): | |
| """causal expanding median over MOVING finite values <= t (defined once >= n_min values), and the causal running median of |x - median_t| | |
| over the same frames (MAD_t). Returns (med[T], mad[T]) with NaN while undefined. Only values <= t enter either statistic.""" | |
| x = np.asarray(x, np.float64); T = len(x) | |
| med = np.full(T, np.nan); mad = np.full(T, np.nan) | |
| arr, res = [], []; last_m = np.nan; last_r = np.nan | |
| for t in range(T): | |
| v = x[t] | |
| if moving[t] and np.isfinite(v): | |
| bisect.insort(arr, float(v)); n = len(arr) | |
| if n >= n_min: | |
| last_m = arr[n // 2] if n % 2 else 0.5 * (arr[n // 2 - 1] + arr[n // 2]) | |
| bisect.insort(res, abs(float(v) - last_m)); m = len(res) | |
| if m >= n_min: | |
| last_r = res[m // 2] if m % 2 else 0.5 * (res[m // 2 - 1] + res[m // 2]) | |
| med[t] = last_m; mad[t] = last_r | |
| return med.astype(np.float32), mad.astype(np.float32) | |
| def densify(kpts, moving, t): | |
| """pose caches hold the decoded spans only (monotone ticks with gaps) -> dense arrays indexed by tick = round(10 t); absent frames NaN / not moving.""" | |
| tk = np.round(np.asarray(t, np.float64) * 10).astype(np.int64) | |
| assert (np.diff(tk) > 0).all() and tk[0] >= 0, "pose ticks must be strictly increasing" | |
| T = int(tk[-1]) + 1 | |
| kd = np.full((T,) + tuple(kpts.shape[1:]), np.nan, np.float32); kd[tk] = np.asarray(kpts, np.float32) | |
| md = np.zeros(T, bool); md[tk] = moving | |
| return kd, md, int(len(tk)) | |
| def route_series(kpts, moving, W, H): | |
| """full causal per-frame proxy series of one route (10 Hz).""" | |
| p = frame_proxies(kpts, W, H) | |
| pitch_s = causal_smooth(p["head_pitch"]); wf_s = causal_smooth(p["wrist_face"]) | |
| med, mad = causal_median_mad(pitch_s, moving) | |
| dev = pitch_s - med | |
| with np.errstate(invalid="ignore"): | |
| headdown = np.isfinite(dev) & np.isfinite(mad) & (dev > K_MAD * np.maximum(mad, 1e-6)) | |
| lap_hd = p["lap"] & headdown | |
| return dict(pitch=pitch_s, pitch_dev=dev.astype(np.float32), mad=mad, wrist_face=wf_s, headdown=headdown, lap=p["lap"], lap_headdown=lap_hd, | |
| face_valid=p["face_valid"], body_valid=p["body_valid"], box=p["box"]) | |
| def window_scores(S, i0, i1): | |
| """window frames [i0, i1) of the route series -> dict of scores + coverage.""" | |
| T = len(S["pitch_dev"]); sl = slice(min(max(i0, 0), T), min(max(i1, 0), T)); n = sl.stop - sl.start | |
| out = dict(n_frames=int(n)) | |
| if n == 0: | |
| for k in ("pitch_dev_max", "pitch_dev_mean", "neg_wrist_face_max", "neg_wrist_face_mean", "lap_headdown_share", "lap_headdown_any", | |
| "headdown_frames", "cov_face", "cov_body", "cov_box"): | |
| out[k] = float("nan") | |
| return out | |
| dev = S["pitch_dev"][sl]; wf = S["wrist_face"][sl]; fv = S["face_valid"][sl]; bv = S["body_valid"][sl] | |
| with np.errstate(invalid="ignore"): | |
| out["pitch_dev_max"] = float(np.nanmax(dev)) if np.isfinite(dev).any() else float("nan") | |
| out["pitch_dev_mean"] = float(np.nanmean(dev)) if np.isfinite(dev).any() else float("nan") | |
| out["neg_wrist_face_max"] = float(np.nanmax(-wf)) if np.isfinite(wf).any() else float("nan") | |
| out["neg_wrist_face_mean"] = float(np.nanmean(-wf)) if np.isfinite(wf).any() else float("nan") | |
| nb = int(bv.sum()); nl = int(S["lap_headdown"][sl].sum()) | |
| out["lap_headdown_share"] = (nl / nb) if nb > 0 else float("nan") | |
| out["lap_headdown_any"] = float(nl > 0) if nb > 0 else float("nan") | |
| out["headdown_frames"] = int(S["headdown"][sl].sum()) | |
| out["cov_face"] = float(fv.mean()); out["cov_body"] = float(bv.mean()); out["cov_box"] = float(S["box"][sl].mean()) | |
| return out | |
| # ----------------------------------------------------------------------------- statistics | |
| def auroc(pos, neg): | |
| """Mann-Whitney AUROC (ties = 0.5); NaN if either class is empty.""" | |
| pos = np.asarray(pos, np.float64); neg = np.asarray(neg, np.float64) | |
| if len(pos) == 0 or len(neg) == 0: return float("nan") | |
| gt = (pos[:, None] > neg[None, :]).sum(); eq = (pos[:, None] == neg[None, :]).sum() | |
| return float((gt + 0.5 * eq) / (len(pos) * len(neg))) | |
| def rankdata(a): | |
| """average ranks (1-based) of a 1-D array; NaN ranks lowest (all NaN tie below every finite value).""" | |
| a = np.asarray(a, np.float64); b = np.where(np.isfinite(a), a, -np.inf) | |
| order = np.argsort(b, kind="mergesort"); ranks = np.empty(len(b)); sb = b[order] | |
| i = 0 | |
| while i < len(sb): | |
| j = i | |
| while j + 1 < len(sb) and sb[j + 1] == sb[i]: j += 1 | |
| ranks[order[i:j + 1]] = 0.5 * (i + j) + 1.0; i = j + 1 | |
| return ranks | |
| def cluster_boot_auroc(score, y, cluster, rng, n_boot, drop_nan=True): | |
| """point AUROC and percentile 95 % CI over cluster (device) bootstrap; windows with NaN score are dropped when drop_nan, else ranked lowest.""" | |
| score = np.asarray(score, np.float64); y = np.asarray(y, bool); cluster = np.asarray(cluster) | |
| if drop_nan: | |
| keep = np.isfinite(score); score, y, cluster = score[keep], y[keep], cluster[keep] | |
| else: | |
| score = np.where(np.isfinite(score), score, -np.inf) | |
| point = auroc(score[y], score[~y]) | |
| cl = np.unique(cluster); idx_by = {c: np.flatnonzero(cluster == c) for c in cl} | |
| vals = [] | |
| for _ in range(n_boot): | |
| pick = rng.choice(cl, size=len(cl), replace=True) | |
| ii = np.concatenate([idx_by[c] for c in pick]) if len(pick) else np.array([], int) | |
| if len(ii) == 0 or y[ii].all() or (~y[ii]).all(): continue | |
| vals.append(auroc(score[ii][y[ii]], score[ii][~y[ii]])) | |
| lo, hi = (float(np.percentile(vals, 2.5)), float(np.percentile(vals, 97.5))) if vals else (float("nan"), float("nan")) | |
| return dict(auroc=point, ci=[lo, hi], n_pos=int(y.sum()), n_neg=int((~y).sum()), n_boot_eff=len(vals), n_clusters=int(len(cl))) | |
| # ----------------------------------------------------------------------------- episodes / exclusions | |
| def pose_file(uid): | |
| for d in (POSE_V2, POSE_V1): | |
| p = d / f"{uid}.npz" | |
| if p.exists(): return p | |
| return None | |
| def load_items(): | |
| """every reviewed item (id -> meta) across items.json / items_gold_engaged.json / items_relabel_*.json (labels irrelevant here: exclusion zones).""" | |
| out = {} | |
| for f in sorted(REVIEW.glob("items*.json")): | |
| d = json.load(open(f)); items = d["items"] if isinstance(d, dict) else d | |
| for it in items: | |
| out.setdefault(it["id"], it["meta"] | {"_file": f.name}) | |
| return out | |
| def item_time(meta): | |
| for k in ("t_event", "t_anchor", "t0"): | |
| if meta.get(k) is not None: return float(meta[k]) | |
| return None | |
| def build_positives(test_uids): | |
| """owner-noted phone / item items (labels_latest.json notes) + newtest object / gaze-away / hands-off prefix rows -> merged episodes.""" | |
| items = load_items(); labels = json.load(open(REVIEW / "labels_latest.json")) | |
| raw = [] | |
| for k, v in labels.items(): | |
| note = v.get("note") or "" | |
| if not NOTE_RE.search(note): continue | |
| m = items.get(k) | |
| if m is None: print(f"WARNING noted item {k} not found in any items file"); continue | |
| t = item_time(m); tag = "phone" if re.search("phone", note, re.I) else "object" | |
| raw.append(dict(src=f"review:{k}", uid=m["uid"], dongle=m["uid"].split("__")[1], regime=m["regime"], t=t, tag=tag, note=note)) | |
| for line in open(NEWTEST): | |
| r = json.loads(line) | |
| f = r.get("fields") or {} | |
| if r.get("stage") != "prefix" or not f: continue | |
| codes = set(f.get("reason_codes") or []) | |
| obj = f.get("object_interaction") in NEWTEST_OBJ; lap = f.get("view_region") == "lap_or_phone_area"; cod = bool(codes & NEWTEST_CODES) | |
| if not (obj or lap or cod): continue | |
| if f.get("object_interaction") == "phone" or lap: tag = "phone" | |
| elif obj or "object_interaction_while_moving" in codes: tag = "object" | |
| elif "sustained_gaze_away_with_demand" in codes: tag = "gaze_away" | |
| else: tag = "hands_off" | |
| raw.append(dict(src=f"newtest:{r['idx']}", uid=r["uid"], dongle=r["uid"].split("__")[1], regime=r["regime"], t=float(r["t0"]) + float(r["anchor_s"]), | |
| tag=tag, note=f"{f.get('object_interaction')}/{f.get('view_region')}/{sorted(codes)}")) | |
| raw = [r for r in raw if r["uid"] in test_uids] | |
| raw.sort(key=lambda r: (r["uid"], r["regime"], r["t"])) | |
| eps = [] | |
| for r in raw: | |
| if eps and eps[-1]["uid"] == r["uid"] and eps[-1]["regime"] == r["regime"] and r["t"] - eps[-1]["t_last"] <= MERGE_S: | |
| e = eps[-1]; e["members"].append(r["src"]); e["t_last"] = r["t"]; e["tags"] = sorted(set(e["tags"]) | {r["tag"]}) | |
| else: | |
| eps.append(dict(uid=r["uid"], dongle=r["dongle"], regime=r["regime"], t=r["t"], t_last=r["t"], members=[r["src"]], tags=[r["tag"]])) | |
| for i, e in enumerate(eps): e["id"] = f"E{i:03d}"; e.pop("t_last") | |
| return eps, raw | |
| def anchor_tables(): | |
| """per (regime): uid -> sorted anchor ticks (row_ok) of the kin-v2 test grid; automatic events from kin v2 / v3 S0_c and gaze S0_g transitions; | |
| face-pack gaze_off at anchors.""" | |
| grid, events, gaze_at = {r: defaultdict(list) for r in REGIMES}, defaultdict(list), defaultdict(dict) | |
| for r in REGIMES: | |
| z = np.load(KIN_V2 / f"kin_causal_{r}_test.npz", allow_pickle=True) | |
| uid, t, ok = z["uid"], z["t"], z["row_ok"] | |
| for u in np.unique(uid): | |
| m = (uid == u) & ok | |
| grid[r][str(u)] = np.unique(np.round(t[m] * 10).astype(np.int64)) | |
| for src, key in ((KIN_V2, "S0_c"), (KIN_V3, "S0_c")): | |
| zz = np.load(src / f"kin_causal_{r}_test.npz", allow_pickle=True) | |
| assert (zz["uid"] == uid).all() and np.allclose(zz["t"], t) | |
| S = zz[key] | |
| for u in np.unique(uid): | |
| m = np.flatnonzero(uid == u); o = m[np.argsort(t[m])]; tt = t[o]; ss = S[o] | |
| for c in range(ss.shape[1]): | |
| a, b = ss[:-1, c], ss[1:, c] | |
| ch = np.isfinite(a) & np.isfinite(b) & (a != b) & (np.diff(tt) <= 1.0 + 1e-6) | |
| events[str(u)].extend(tt[1:][ch].tolist()) | |
| zg = np.load(GAZE / f"gaze_{r}_test.npz", allow_pickle=True) | |
| assert (zg["uid"] == uid).all() and np.allclose(zg["t"], t) | |
| Sg = zg["S0_g"] | |
| for u in np.unique(uid): | |
| m = np.flatnonzero(uid == u); o = m[np.argsort(t[m])]; tt = t[o]; ss = Sg[o] | |
| ch = np.isfinite(ss[:-1]) & np.isfinite(ss[1:]) & (ss[:-1] != ss[1:]) & (np.diff(tt) <= 1.0 + 1e-6) | |
| events[str(u)].extend(tt[1:][ch].tolist()) | |
| zf = np.load(FACE / f"facehist_causal_{r}_test.npz", allow_pickle=True) | |
| assert (zf["uid"] == uid).all() and np.allclose(zf["t"], t) | |
| ch_names = list(zf["channels"]); gi = ch_names.index("gaze_off") | |
| g = zf["F0_c"][:, gi] | |
| for u in np.unique(uid): | |
| m = np.flatnonzero(uid == u) | |
| gaze_at[str(u)].update({int(round(float(t[i]) * 10)): float(g[i]) for i in m}) | |
| return grid, events, gaze_at | |
| def labelled_intervals(uid, items, clips): | |
| iv = [] | |
| for m in items.values(): | |
| if m.get("uid") != uid: continue | |
| t0 = item_time(m) | |
| if t0 is None: continue | |
| t1 = float(m["t1"]) if m.get("t1") is not None else t0 | |
| iv.append((t0, t1)) | |
| for c in clips: | |
| if c.get("uid") == uid: iv.append((float(c["t0"]), float(c["t1"]))) | |
| return iv | |
| def interval_dist(a0, a1, b0, b1): | |
| return max(0.0, max(a0, b0) - min(a1, b1)) | |
| def draw_negatives(uid, regime, pos_windows, grid_ticks, excl, moving, rng, n_want, excl_auto=(), clear_auto=CLEAR_S): | |
| """candidate starts on the 1 s grid that are row_ok anchors of the regime, moving share >= MOVING_MIN, >= CLEAR_S from every labelled exclusion interval | |
| and positive window and >= clear_auto from every automatic driver event (primary: clear_auto = CLEAR_S); draw up to n_want with pairwise separation >= WIN_S.""" | |
| T = len(moving); cands = [] | |
| ex = np.array(excl, np.float64).reshape(-1, 2); ea = np.array(excl_auto, np.float64).reshape(-1, 2); pw = np.array(pos_windows, np.float64).reshape(-1, 2) | |
| def clear(t0, t1, iv, c): | |
| return len(iv) == 0 or (np.maximum(0.0, np.maximum(t0, iv[:, 0]) - np.minimum(t1, iv[:, 1])) >= c).all() | |
| for tick in grid_ticks: | |
| if tick % 10: continue | |
| i0, i1 = int(tick), int(tick + WIN_S * 10) | |
| if i1 > T: continue | |
| if moving[i0:i1].mean() < MOVING_MIN: continue | |
| t0, t1 = tick / 10.0, tick / 10.0 + WIN_S | |
| if not (clear(t0, t1, ex, CLEAR_S) and clear(t0, t1, pw, CLEAR_S) and clear(t0, t1, ea, clear_auto)): continue | |
| cands.append(t0) | |
| cands = np.array(sorted(cands)); chosen = [] | |
| order = rng.permutation(len(cands)) | |
| for j in order: | |
| c = cands[j] | |
| if all(abs(c - x) >= WIN_S for x in chosen): chosen.append(float(c)) | |
| if len(chosen) >= n_want: break | |
| return sorted(chosen), int(len(cands)) | |
| # ----------------------------------------------------------------------------- main | |
| def fmt(x, nd=3): | |
| return "nan" if x is None or (isinstance(x, float) and not np.isfinite(x)) else f"{x:.{nd}f}" | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--out", default=str(ROOT / "results" / "tables" / "pitch_wristface_diag")) | |
| ap.add_argument("--n-boot", type=int, default=2000) | |
| ap.add_argument("--seed", type=int, default=0) | |
| a = ap.parse_args(); t_start = time.time(); stamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) | |
| test_uids = set() | |
| for r in REGIMES: | |
| test_uids |= set(np.unique(np.load(KIN_V2 / f"kin_causal_{r}_test.npz", allow_pickle=True)["uid"]).tolist()) | |
| eps, raw = build_positives(test_uids) | |
| print(f"positives: {len(raw)} raw items -> {len(eps)} episodes on {len(set(e['uid'] for e in eps))} routes / {len(set(e['dongle'] for e in eps))} devices") | |
| grid, events, gaze_at = anchor_tables() | |
| items = load_items(); clips = json.load(open(CLIPS_NEWTEST))["clips"] | |
| rng = np.random.default_rng(a.seed) | |
| rows = []; route_stats = {}; SER, MOV, EXC = {}, {}, {} | |
| for uid in sorted(set(e["uid"] for e in eps)): | |
| pf = pose_file(uid); assert pf is not None, uid | |
| z = np.load(pf, allow_pickle=True) | |
| W, H = [int(v) for v in z["native_wh"]]; tt = z["t"].astype(np.float64) | |
| kp, moving, n_present = densify(z["kpts"], z["moving"].astype(bool), tt) # the caches hold decoded spans only -> dense 0.1 s tick grid | |
| t0 = time.time(); S = route_series(kp, moving, W, H) | |
| excl_lab = labelled_intervals(uid, items, clips); excl_auto = [(e, e) for e in events.get(uid, [])] | |
| excl = excl_lab + excl_auto | |
| SER[uid] = S; MOV[uid] = moving; EXC[uid] = (excl_lab, excl_auto) | |
| route_stats[uid] = dict(pose_dir=pf.parent.name, n_frames=int(len(moving)), n_present=int(n_present), n_excl=len(excl), n_excl_labelled=len(excl_lab), n_excl_auto=len(excl_auto), face_valid_share=float(S["face_valid"].mean()), | |
| body_valid_share=float(S["body_valid"].mean()), moving_share=float(moving.mean()), sec=round(time.time() - t0, 1)) | |
| my = [e for e in eps if e["uid"] == uid] | |
| pos_windows = [(e["t"], e["t"] + WIN_S) for e in my] | |
| for e in my: | |
| i0 = int(round(e["t"] * 10)); i1 = i0 + int(WIN_S * 10) | |
| sc = window_scores(S, i0, i1) | |
| g = [gaze_at[uid].get(k, np.nan) for k in range(i0, i1)]; g = np.array([v for v in g if np.isfinite(v)]) | |
| gmax = float(g.max()) if len(g) else float("nan") | |
| rows.append(dict(kind="pos", ep=e["id"], uid=uid, dongle=e["dongle"], regime=e["regime"], t=e["t"], tags=e["tags"], members=e["members"], | |
| gaze_off_max=gmax, n_gaze_anchors=int(len(g)), moving_share=float(moving[i0:i1].mean()) if i1 <= len(moving) else float("nan"), **sc)) | |
| negs, n_c = draw_negatives(uid, e["regime"], pos_windows, grid[e["regime"]].get(uid, []), excl_lab, moving, rng, NEG_PER_POS, excl_auto=excl_auto, clear_auto=CLEAR_S) | |
| for tn in negs: | |
| j0 = int(round(tn * 10)); j1 = j0 + int(WIN_S * 10); scn = window_scores(S, j0, j1) | |
| g = [gaze_at[uid].get(k, np.nan) for k in range(j0, j1)]; g = np.array([v for v in g if np.isfinite(v)]) | |
| rows.append(dict(kind="neg", ep=e["id"], uid=uid, dongle=e["dongle"], regime=e["regime"], t=tn, tags=[], members=[], gaze_off_max=float(g.max()) if len(g) else float("nan"), | |
| n_gaze_anchors=int(len(g)), moving_share=float(moving[j0:j1].mean()), n_candidates=n_c, **scn)) | |
| print(f" {e['id']} {e['regime']:8s} {uid[:48]:48s} t={e['t']:8.1f} tags={','.join(e['tags']):10s} negs={len(negs)}/{n_c} cov_face={sc['cov_face']:.2f} " | |
| f"pitch_dev_max={fmt(sc['pitch_dev_max'])} negwf_max={fmt(sc['neg_wrist_face_max'])} lap_hd={fmt(sc['lap_headdown_share'])}") | |
| # de-duplicate negatives shared by episodes on the same route (same start) — keep the first | |
| seen = set(); dedup = [] | |
| for r in rows: | |
| key = (r["kind"], r["uid"], r["regime"], round(r["t"], 1)) | |
| if r["kind"] == "neg" and key in seen: continue | |
| seen.add(key); dedup.append(r) | |
| rows = dedup | |
| y = np.array([r["kind"] == "pos" for r in rows]); reg = np.array([r["regime"] for r in rows]); dong = np.array([r["dongle"] for r in rows]) | |
| SCORES = {"head_pitch_kp (max dev)": "pitch_dev_max", "wrist_face_kp (max closeness)": "neg_wrist_face_max", "hands_lap_headdown_kp (share)": "lap_headdown_share", | |
| "head_pitch_kp (mean dev)": "pitch_dev_mean", "wrist_face_kp (mean closeness)": "neg_wrist_face_mean", "hands_lap_headdown_kp (any)": "lap_headdown_any"} | |
| arr = {k: np.array([r[v] if r[v] is not None else np.nan for r in rows], np.float64) for k, v in SCORES.items()} | |
| prim = ["head_pitch_kp (max dev)", "wrist_face_kp (max closeness)", "hands_lap_headdown_kp (share)"] | |
| arr["combination (mean pooled rank of the 3 primary)"] = np.mean([rankdata(arr[k]) for k in prim], axis=0) | |
| res = {} | |
| for name, s in arr.items(): | |
| res[name] = {} | |
| for scope in ("pooled",) + REGIMES: | |
| m = np.ones(len(rows), bool) if scope == "pooled" else (reg == scope) | |
| if m.sum() == 0 or y[m].sum() == 0 or (~y[m]).sum() == 0: res[name][scope] = dict(auroc=float("nan"), ci=[float("nan")] * 2, n_pos=int(y[m].sum()), n_neg=int((~y[m]).sum())); continue | |
| res[name][scope] = cluster_boot_auroc(s[m], y[m], dong[m], np.random.default_rng(a.seed + 1), a.n_boot, drop_nan=True) | |
| res[name][scope]["nan_lowest"] = cluster_boot_auroc(s[m], y[m], dong[m], np.random.default_rng(a.seed + 2), a.n_boot, drop_nan=False)["auroc"] | |
| res[name][scope]["n_nan_scores"] = int((~np.isfinite(s[m])).sum()) | |
| # per class tag (descriptive): positives of that tag vs all negatives | |
| tags_all = sorted(set(t for r in rows for t in r["tags"])) | |
| per_tag = {} | |
| for tg in tags_all: | |
| m = (~y) | np.array([tg in r["tags"] for r in rows]) | |
| per_tag[tg] = {k: dict(auroc=auroc(arr[k][m & y & np.isfinite(arr[k])], arr[k][m & ~y & np.isfinite(arr[k])]), n_pos=int((m & y).sum())) for k in prim + ["combination (mean pooled rank of the 3 primary)"]} | |
| # coverage | |
| cov = {k: float(np.nanmean([r[k] for r in rows])) for k in ("cov_face", "cov_body", "cov_box")} | |
| cov_pos = {k: float(np.nanmean([r[k] for r in rows if r["kind"] == "pos"])) for k in ("cov_face", "cov_body", "cov_box")} | |
| cov_neg = {k: float(np.nanmean([r[k] for r in rows if r["kind"] == "neg"])) for k in ("cov_face", "cov_body", "cov_box")} | |
| cov_face_ge05 = float(np.mean([r["cov_face"] >= 0.5 for r in rows])) | |
| # gaze_off = 0 but head-down | |
| pos_rows = [r for r in rows if r["kind"] == "pos"] | |
| hd = [r["headdown_frames"] >= HEADDOWN_MIN_FRAMES for r in pos_rows] | |
| g0 = [np.isfinite(r["gaze_off_max"]) and r["gaze_off_max"] == 0 for r in pos_rows] | |
| gnan = [not np.isfinite(r["gaze_off_max"]) for r in pos_rows] | |
| share_g0_hd = float(np.mean([a_ and b_ for a_, b_ in zip(g0, hd)])) if pos_rows else float("nan") | |
| share_hd = float(np.mean(hd)) if pos_rows else float("nan"); share_g0 = float(np.mean(g0)) if pos_rows else float("nan") | |
| share_hd_neg = float(np.mean([r["headdown_frames"] >= HEADDOWN_MIN_FRAMES for r in rows if r["kind"] == "neg"])) if (~y).any() else float("nan") | |
| # ---------------------------------------------------------------- sensitivity (POST-HOC secondary analyses; the verdict is read from the primary above) | |
| def evaluate(sub_rows, seed_off=10): | |
| yy = np.array([r["kind"] == "pos" for r in sub_rows]); dd = np.array([r["dongle"] for r in sub_rows]) | |
| A = {k: np.array([r[v] if r[v] is not None else np.nan for r in sub_rows], np.float64) for k, v in SCORES.items() if k in prim} | |
| A["combination (mean pooled rank of the 3 primary)"] = np.mean([rankdata(A[k]) for k in prim], axis=0) | |
| out = {} | |
| for name, sc in A.items(): | |
| if yy.sum() == 0 or (~yy).sum() == 0: out[name] = dict(auroc=float("nan"), ci=[float("nan")] * 2, n_pos=int(yy.sum()), n_neg=int((~yy).sum())); continue | |
| out[name] = cluster_boot_auroc(sc, yy, dd, np.random.default_rng(a.seed + seed_off), a.n_boot, drop_nan=True) | |
| out["_n_routes"] = int(len(set(r["uid"] for r in sub_rows))); out["_n_devices_neg"] = int(len(set(r["dongle"] for r in sub_rows if r["kind"] == "neg"))) | |
| return out | |
| sens = {} | |
| eps_with_neg = set(r["ep"] for r in rows if r["kind"] == "neg") | |
| sens["S1 route-matched subset (episodes with >= 1 own negative) vs their negatives"] = evaluate([r for r in rows if r["ep"] in eps_with_neg or r["kind"] == "neg"]) | |
| paired = {} | |
| for k in prim: | |
| v = SCORES[k]; wins_all, wins_med, n = 0, 0, 0 | |
| for r in pos_rows: | |
| own = [x[v] for x in rows if x["kind"] == "neg" and x["ep"] == r["ep"] and x[v] is not None and np.isfinite(x[v])] | |
| if not own or r[v] is None or not np.isfinite(r[v]): continue | |
| n += 1; wins_all += r[v] > max(own); wins_med += r[v] > float(np.median(own)) | |
| paired[k] = dict(n_episodes=n, share_pos_above_all_own_negs=(wins_all / n if n else float("nan")), share_pos_above_median_own_neg=(wins_med / n if n else float("nan"))) | |
| sens["S1 paired within-episode wins"] = paired | |
| sens["S3 positives with moving share >= 0.8 vs the primary negatives"] = evaluate([r for r in rows if r["kind"] == "neg" or (np.isfinite(r["moving_share"]) and r["moving_share"] >= MOVING_MIN)], 11) | |
| for ca in (30.0, 15.0): | |
| rng2 = np.random.default_rng(a.seed + int(ca)); rows2 = [r for r in rows if r["kind"] == "pos"]; ncand = 0 | |
| for uid in sorted(SER): | |
| my = [e for e in eps if e["uid"] == uid]; pos_windows = [(e["t"], e["t"] + WIN_S) for e in my]; excl_lab, excl_auto = EXC[uid] | |
| for e in my: | |
| negs, n_c = draw_negatives(uid, e["regime"], pos_windows, grid[e["regime"]].get(uid, []), excl_lab, MOV[uid], rng2, NEG_PER_POS, excl_auto=excl_auto, clear_auto=ca) | |
| ncand += n_c | |
| for tn in negs: | |
| j0 = int(round(tn * 10)); rows2.append(dict(kind="neg", ep=e["id"], uid=uid, dongle=e["dongle"], regime=e["regime"], t=tn, tags=[], members=[], | |
| moving_share=float(MOV[uid][j0:j0 + int(WIN_S * 10)].mean()), **window_scores(SER[uid], j0, j0 + int(WIN_S * 10)))) | |
| seen2 = set(); ded2 = [] | |
| for r in rows2: | |
| key = (r["kind"], r["uid"], r["regime"], round(r["t"], 1)) | |
| if r["kind"] == "neg" and key in seen2: continue | |
| seen2.add(key); ded2.append(r) | |
| sens[f"S2 automatic-event clearance {ca:.0f} s (labelled items / positives still {CLEAR_S:.0f} s)"] = evaluate(ded2, 12 + int(ca)) | {"_n_pos": int(sum(r["kind"] == "pos" for r in ded2)), "_n_neg": int(sum(r["kind"] == "neg" for r in ded2)), | |
| "_episodes_with_neg": int(len(set(r["ep"] for r in ded2 if r["kind"] == "neg")))} | |
| # rule | |
| best_name, best = None, None | |
| for name in prim + ["combination (mean pooled rank of the 3 primary)"]: | |
| p = res[name]["pooled"] | |
| if best is None or (np.nan_to_num(p["auroc"], nan=-1) > np.nan_to_num(best["auroc"], nan=-1)): best_name, best = name, p | |
| separable = bool(np.isfinite(best["auroc"]) and best["auroc"] >= 0.80 and best["ci"][0] >= 0.70 and cov["cov_face"] >= 0.8) | |
| n_pos, n_neg = int(y.sum()), int((~y).sum()) | |
| out = Path(a.out); out.parent.mkdir(parents=True, exist_ok=True) | |
| J = dict(stamp=stamp, script="eval/pitch_wristface_diag.py", prereg_row=511, rule="separable iff best primary proxy or the rank combination reaches pooled AUROC >= 0.80 with CI-LB >= 0.70 AND face coverage >= 0.8", | |
| params=dict(CONF_TH=CONF_TH, NT=NT, RE=RE, LE=LE, IOD_MIN_PX=IOD_MIN_PX, SMOOTH=SMOOTH, N_MIN=N_MIN, K_MAD=K_MAD, LAP_SW=LAP_SW, WIN_S=WIN_S, CLEAR_S=CLEAR_S, MERGE_S=MERGE_S, | |
| NEG_PER_POS=NEG_PER_POS, MOVING_MIN=MOVING_MIN, HEADDOWN_MIN_FRAMES=HEADDOWN_MIN_FRAMES, n_boot=a.n_boot, seed=a.seed), | |
| n_raw_items=len(raw), n_episodes=len(eps), n_pos=n_pos, n_neg=n_neg, n_routes=len(route_stats), n_devices=int(len(set(dong))), | |
| n_pos_by_regime={r: int((y & (reg == r)).sum()) for r in REGIMES}, n_neg_by_regime={r: int((~y & (reg == r)).sum()) for r in REGIMES}, | |
| auroc=res, per_tag=per_tag, coverage=dict(all=cov, pos=cov_pos, neg=cov_neg, share_windows_face_cov_ge_0_5=cov_face_ge05), | |
| gaze_off_vs_headdown=dict(share_pos_gaze_off_never_and_headdown=share_g0_hd, share_pos_headdown=share_hd, share_pos_gaze_off_never=share_g0, | |
| share_pos_gaze_unknown=float(np.mean(gnan)) if pos_rows else float("nan"), share_neg_headdown=share_hd_neg), | |
| best=dict(name=best_name, **best), separable=separable, verdict="A" if separable else "B", sensitivity_post_hoc=sens, raw_items=raw, episodes=eps, windows=rows, route_stats=route_stats, | |
| sec=round(time.time() - t_start, 1)) | |
| json.dump(J, open(out.with_suffix(".json"), "w"), indent=1, default=lambda o: o.item() if hasattr(o, "item") else str(o)) | |
| # ------------------------------------------------------------------ markdown | |
| L = [f"# Plan A diagnosis — keypoint head-pitch / wrist-face / hands-lap-head-down proxies on phone / object / gaze-away episodes (TEST routes)", "", | |
| f"Generated {stamp} by `eval/pitch_wristface_diag.py` (pre-registration row 511 of docs/decisions.md; owner decision row 510). ZERO training; labels on TEST routes used only to measure separability.", | |
| f"Inputs: our own causal HALPE-136 pose at 10 Hz (`results/pose_causal_routes_v2` else `results/pose_causal_routes`); no DMS quantity is read. Proxy names carry `_kp`.", "", | |
| f"**PRE-SPECIFIED RULE:** separable iff the best single primary proxy OR the rank combination reaches pooled AUROC >= 0.80 with device-cluster bootstrap CI lower bound >= 0.70 AND face coverage >= 0.8.", "", | |
| f"**VERDICT: {'SEPARABLE -> plan A' if separable else 'NOT SEPARABLE -> plan B'}** — best = {best_name}: pooled AUROC {fmt(best['auroc'])} [{fmt(best['ci'][0])}, {fmt(best['ci'][1])}], face coverage {fmt(cov['cov_face'])} " | |
| f"(n_pos {n_pos}, n_neg {n_neg}, {J['n_devices']} devices, {J['n_routes']} routes).", "", | |
| "## Windows", "", | |
| f"- positives: {len(raw)} raw labelled items -> {len(eps)} merged episodes (same uid + regime within {MERGE_S:.0f} s), window [t, t+{WIN_S:.0f} s]; by regime {J['n_pos_by_regime']}; tags {dict((tg, sum(tg in r['tags'] for r in pos_rows)) for tg in tags_all)}", | |
| f"- negatives: {n_neg} windows (target {NEG_PER_POS} per positive; same route + regime, moving share >= {MOVING_MIN}, >= {CLEAR_S:.0f} s from every labelled item / automatic driver event (kin v2 + v3 S0 changes, gaze_off changes) / positive window; seed {a.seed}); by regime {J['n_neg_by_regime']}", | |
| f"- windows with NaN score are DROPPED from the primary AUROC (n reported per cell) — a reporting choice not fixed in the pre-registration; the `NaN-lowest` column ranks them below every finite score.", "", | |
| "## AUROC (positives vs matched negatives; 95 % percentile CI over device-cluster bootstrap, n_boot {})".format(a.n_boot), "", | |
| "| proxy / window score | scope | AUROC | 95 % CI | n_pos | n_neg | n NaN dropped | AUROC NaN-lowest |", "|---|---|---|---|---|---|---|---|"] | |
| for name in prim + ["combination (mean pooled rank of the 3 primary)"] + [k for k in SCORES if k not in prim]: | |
| for scope in ("pooled",) + REGIMES: | |
| p = res[name][scope] | |
| L.append(f"| {name} | {scope} | {fmt(p['auroc'])} | [{fmt(p['ci'][0])}, {fmt(p['ci'][1])}] | {p['n_pos']} | {p['n_neg']} | {p.get('n_nan_scores', '')} | {fmt(p.get('nan_lowest', float('nan')))} |") | |
| L += ["", "## Per class tag (descriptive; positives of the tag vs all negatives, NaN dropped)", "", "| tag | n_pos | " + " | ".join(prim + ["combination"]) + " |", "|---|---|" + "---|" * 4] | |
| for tg in tags_all: | |
| L.append(f"| {tg} | {per_tag[tg][prim[0]]['n_pos']} | " + " | ".join(fmt(per_tag[tg][k]['auroc']) for k in prim + ["combination (mean pooled rank of the 3 primary)"]) + " |") | |
| L += ["", "## Coverage (share of window frames; mean over windows)", "", "| set | face proxy valid (nose + both eyes, IOD >= 10 px) | body valid (nose, shoulders, wrists) | driver box present |", "|---|---|---|---|", | |
| f"| all windows | {fmt(cov['cov_face'])} | {fmt(cov['cov_body'])} | {fmt(cov['cov_box'])} |", f"| positives | {fmt(cov_pos['cov_face'])} | {fmt(cov_pos['cov_body'])} | {fmt(cov_pos['cov_box'])} |", | |
| f"| negatives | {fmt(cov_neg['cov_face'])} | {fmt(cov_neg['cov_body'])} | {fmt(cov_neg['cov_box'])} |", "", f"share of windows with face coverage >= 0.5: {fmt(cov_face_ge05)}", "", | |
| "## Sensitivity (POST-HOC secondary analyses, not in the pre-registration; the verdict above is read from the primary table only)", "", | |
| f"Motivation: under the strict {CLEAR_S:.0f} s clearance from every automatic driver event only {len(eps_with_neg)} of {n_pos} episodes ({J['n_devices']} devices with positives, {len(set(r['dongle'] for r in rows if r['kind']=='neg'))} with negatives) obtain negatives on their own route, so the primary AUROC partly compares across routes.", "", | |
| "| analysis | n_pos | n_neg | routes | devices w/ neg | " + " | ".join(prim + ["combination"]) + " |", "|---|---|---|---|---|" + "---|" * 4] | |
| for name, o in sens.items(): | |
| if name.startswith("S1 paired"): continue | |
| cells = [f"{fmt(o[k]['auroc'])} [{fmt(o[k]['ci'][0])}, {fmt(o[k]['ci'][1])}]" for k in prim + ["combination (mean pooled rank of the 3 primary)"]] | |
| L.append(f"| {name} | {o[prim[0]]['n_pos']} | {o[prim[0]]['n_neg']} | {o['_n_routes']} | {o['_n_devices_neg']} | " + " | ".join(cells) + " |") | |
| L += ["", "Within-episode paired comparison (positive window score vs the negatives of the SAME episode's route + regime):", "", "| proxy | episodes | share pos > ALL own negs | share pos > median own neg |", "|---|---|---|---|"] | |
| for k, o in paired.items(): | |
| L.append(f"| {k} | {o['n_episodes']} | {fmt(o['share_pos_above_all_own_negs'])} | {fmt(o['share_pos_above_median_own_neg'])} |") | |
| L += ["", | |
| "## Existing yaw proxy vs head-down", "", | |
| f"- share of positive windows where the face-pack `gaze_off` (yaw hysteresis) never flags inside the window AND the head-down flag (head_pitch_dev > {K_MAD:.0f} MAD) holds on >= {HEADDOWN_MIN_FRAMES} valid frames: **{fmt(share_g0_hd)}**", | |
| f"- share of positives with the head-down flag >= {HEADDOWN_MIN_FRAMES} frames: {fmt(share_hd)}; with gaze_off never flagging: {fmt(share_g0)}; gaze_off unknown (no anchors): {fmt(J['gaze_off_vs_headdown']['share_pos_gaze_unknown'])}; head-down flag on NEGATIVES: {fmt(share_hd_neg)}", "", | |
| "## Episodes (positives)", "", "| id | regime | route (uid, truncated) | t (s) | tags | members | negs | cov_face | pitch_dev_max | −wrist_face max | lap_headdown share | gaze_off max | head-down frames |", "|---|---|---|---|---|---|---|---|---|---|---|---|---|"] | |
| for r in pos_rows: | |
| nn = sum(1 for x in rows if x["kind"] == "neg" and x["ep"] == r["ep"]) | |
| L.append(f"| {r['ep']} | {r['regime']} | {r['uid'][:58]} | {r['t']:.1f} | {','.join(r['tags'])} | {','.join(r['members'])} | {nn} | {fmt(r['cov_face'], 2)} | {fmt(r['pitch_dev_max'])} | {fmt(r['neg_wrist_face_max'])} | {fmt(r['lap_headdown_share'])} | {fmt(r['gaze_off_max'], 0)} | {r['headdown_frames']} |") | |
| L += ["", "## Routes", "", "| uid | pose dir | ticks (dense) | frames present | face valid share (route, dense) | body valid share | moving share | exclusion intervals |", "|---|---|---|---|---|---|---|---|"] | |
| for u, s in route_stats.items(): | |
| L.append(f"| {u[:70]} | {s['pose_dir']} | {s['n_frames']} | {s['n_present']} | {fmt(s['face_valid_share'])} | {fmt(s['body_valid_share'])} | {fmt(s['moving_share'])} | {s['n_excl']} |") | |
| L += ["", f"Wall time {J['sec']} s. JSON: `{out.with_suffix('.json').relative_to(ROOT)}`."] | |
| open(out.with_suffix(".md"), "w").write("\n".join(L) + "\n") | |
| print(f"\nVERDICT separable={separable} best={best_name} AUROC={fmt(best['auroc'])} CI=[{fmt(best['ci'][0])},{fmt(best['ci'][1])}] cov_face={fmt(cov['cov_face'])} n_pos={n_pos} n_neg={n_neg}") | |
| print(f"wrote {out.with_suffix('.md')} + .json in {J['sec']} s") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 40.6 kB
- Xet hash:
- 499918f505dd064bbed0621ffbb3526d0e5777340563717a9c53df90ed1fdebd
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.