Buckets:
| """cwm3/train/warning_features_ext.py — label-free EXTENDED anchor features for the learned warning head (roadmap §B3 items 1a / 1b, 2026-09-12). | |
| CPU, numpy only; importable (train/train_warning_head.py --extra-features dwell,mr,face). No model output, no label and no openpilot DMS field is | |
| read: the inputs are the causal driver-kinematics history pack, the CAN-derived current margins (eval/margin_rule.py) and our own HALPE face-landmark | |
| pack (data/build_face_causal.py). | |
| Blocks (each `block(regime, split, rows) -> X float32 [n, k], names`): | |
| dwell (11) from results/targets_kin_causal_routes[_v2]/kinhist_causal_{regime}_{split}.npz Kh8_c f16 [N,80,11] (t-7.9..t @ 10 Hz) + Kh8valid: | |
| hands_off_share_{1,3,8}s share of VALID steps with min_hand_rest_dist (channel 10) >= WHEEL_TH 0.10 (= NOT hands_on) in the trailing window | |
| hands_off_dwell_s seconds since the last valid hands-on step (0 when the last valid step is hands-on; DWELL_CAP 8 s when none in the window) | |
| head_away_share_{1,3,8}s share of valid steps in the head-away state = hysteresis on |head_turn_dev| (channel 7) with the stored TRAIN thresholds | |
| (cwm2/results/targets_kin/build_stats_latest.json head_turn_on 0.1941 / head_turn_off 0.1026; > 1 s gap resets) RE-RUN over | |
| the 80 history steps of the row (build_kin_causal.hysteresis semantics, vectorised over rows) | |
| head_away_dwell_s seconds since the last valid NOT-away step (cap 8 s) | |
| head_dev_max_3s max |head_turn_dev| over the trailing 3 s | |
| n_hand_leave_8s number of hands-on -> hands-off transitions between consecutive valid steps in the 8 s window | |
| hist_valid_share share of the 80 steps that are valid (Kh8valid) | |
| Every quantity is NaN when no valid step exists in its window (HGB handles NaN; LR / MLP impute 0 in the head's pipeline). | |
| mr (7) eval/margin_rule.features: mr_ttc, mr_headway, mr_lead_decel, mr_lateral, mr_badas, mr_max, mr_fired (current-margin RULE ratios). | |
| fs (22) 2026-09-17 (owner decision (6), 'TRY F2′ in the read-out'): the F2′ future-state head's PREDICTED quantities (models/future_state_head.py; | |
| results/runs/FS_<name>_s<k>/fs_{regime}_{split}.npz, row-aligned with the targets / layouts of the FINAL run of the same seed, train / val / test): | |
| fs_off_<state>_<1,2,3,5>s predicted OFF-task occupancy pi_off (hands_on -> 1 - pi; head_turned / gaze_off -> pi), fs_EDoff_<state> expected off-task | |
| seconds over 5 s, fs_remE_<state> expected remaining dwell in the CURRENT off-task state (NaN when on-task: the law is trained on off rows only), | |
| fs_leave3_<state> P(first leave of the current state within 3 s), fs_valid (causal box). PREDICTED quantities (like the gaze block), column tag `fs`. | |
| face (16) from results/targets_face_causal_routes_v2/facehist_causal_{regime}_{split}.npz (F0_c [N,F], F0valid, Fh8_c f16 [N,80,F], Fh8valid): | |
| the F = 9 current channels (fc_<channel>), fc_valid, and history summaries fc_gaze_off_share_{3,8}s, fc_yaw_dev_max_3s, fc_eye_open_min_3s, | |
| fc_blink_max_8s, fc_hist_valid_share. Channel names are the landmark-derived `lm_*` names of data/build_face_causal.py (our own RTMW | |
| HALPE-136 face points; never an openpilot DMS quantity). | |
| Monotone constraints for HistGradientBoosting (`monotone(name)`): +1 on every dwell / share / max / count feature, on the mr_* ratios and on the face | |
| gaze-off / yaw-deviation summaries; 0 on the validity shares, the signed face channels and the raw eye-open / blink channels. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import sys | |
| from pathlib import Path | |
| import numpy as np | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| EXPLORE = CWM3.parent | |
| for p in (CWM3 / "data", CWM3 / "eval"): | |
| if str(p) not in sys.path: | |
| sys.path.insert(0, str(p)) | |
| import cwm3_dataset as ds # noqa: E402 (kin_dir(): the causal route pack of the run's kin_version) | |
| import margin_rule as MR # noqa: E402 | |
| OLD_STATS = EXPLORE / "cwm2" / "results" / "targets_kin" / "build_stats_latest.json" | |
| FACE_DIR = CWM3 / "results" / "targets_face_causal_routes_v2" | |
| WHEEL_TH, GAP_RESET, DWELL_CAP, HZ = 0.10, 10, 8.0, 10.0 | |
| IDX_HEAD_DEV, IDX_MIN_DIST = 7, 10 | |
| SETS = ("dwell", "mr", "face", "gaze", "fs", "kp") | |
| SHORT = {"dwell": "dw", "mr": "mr", "face": "fc", "gaze": "gz", "fs": "fs", "kp": "kp"} | |
| # kp (5; 2026-09-18 plan A arm, pre-reg docs/decisions.md 2026-09-18T05:34:40Z): the CURRENT values of the three keypoint proxies of the kin v2p pack (K0_c columns 11-13 of | |
| # results/targets_kin_causal_routes_v2p/kin_causal_{regime}_{split}.npz: head_pitch_kp = causal deviation of the nose-below-eyes pitch from the route running median, wrist_face_kp = | |
| # min wrist-nose distance / shoulder width, hands_lap_kp = mean wrist drop below the shoulder line / shoulder width) + kp_pitch_dwell_s (PD0_c: seconds the head-down flag has held | |
| # up to t) + kp_valid (K0valid & row_ok & finite pitch). Our own HALPE-136 geometry (data/build_kin_causal_v2p.py); never an openpilot DMS quantity. Current-state block (never `is_future`). | |
| KP_NAMES = ["kp_head_pitch_kp", "kp_wrist_face_kp", "kp_hands_lap_kp", "kp_pitch_dwell_s", "kp_valid"] | |
| KP_DIR = CWM3 / "results" / "targets_kin_causal_routes_v2p" | |
| # gaze (10; 2026-09-12 gaze/robustness agent, coordinator request): the gaze_off CHAIN outputs of a U0 `*_gz*` base run (models/branches.DriverBranch gaze_head; NOT the joint | |
| # model) — gz_onset_{1,2,3,5}s = P(first 0 -> 1 within tau | z0), gz_pi_{1,2,3,5}s = P(gaze_off = 1 at t+k) (forward marginal), gz_z0 (state at t), gz_z0valid. test / val rows | |
| # from results/runs/<gaze run>/pred_layout_{regime}_{split}.npz (extra keys gaze_off_onset_p / gaze_off_pi, eval/write_pred_layout.py); TRAIN label rows from | |
| # pred_gaze_{regime}_train_labelrows.npz (eval/predict_u0_rows.py). The run dir is set by train_warning_head.py --gaze-run (GAZE_OPTS). These are PREDICTED quantities of | |
| # our own causal gaze state — never an openpilot DMS quantity; they are validation-pending until the human gaze labels exist. | |
| GAZE_OPTS = {"run_dir": None} | |
| GAZE_NAMES = ["gz_onset_1s", "gz_onset_2s", "gz_onset_3s", "gz_onset_5s", "gz_pi_1s", "gz_pi_2s", "gz_pi_3s", "gz_pi_5s", "gz_z0", "gz_z0valid"] | |
| GAZE_TAUS, GAZE_PI_STEPS = (5, 10, 15, 25), (4, 9, 14, 24) | |
| DWELL_NAMES = ["dw_hands_off_share_1s", "dw_hands_off_share_3s", "dw_hands_off_share_8s", "dw_hands_off_dwell_s", | |
| "dw_head_away_share_1s", "dw_head_away_share_3s", "dw_head_away_share_8s", "dw_head_away_dwell_s", | |
| "dw_head_dev_max_3s", "dw_n_hand_leave_8s", "dw_hist_valid_share"] | |
| FACE_HIST_NAMES = ["fc_gaze_off_share_3s", "fc_gaze_off_share_8s", "fc_yaw_dev_max_3s", "fc_eye_open_min_3s", "fc_blink_max_8s", "fc_hist_valid_share"] | |
| _MONO_ZERO_SUFFIX = ("valid_share", "fc_valid", "fc_lm_yaw", "fc_lm_yaw_dev", "fc_lm_pitch_dev", "fc_lm_roll_dev", "fc_ear_conf_asym", "fc_eye_open", "fc_blink_rate_4s", | |
| "fc_eye_open_min_3s", "fc_blink_max_8s", "gz_z0valid", "fs_valid") | |
| # fs (22; 2026-09-17): F2′ future-state pack of the SAME seed (train_warning_head.py --fs-run, default FS_dwell_s<seed>); see the module docstring. | |
| FS_OPTS = {"run_dir": None} | |
| FS_STATES = ("hands_on", "head_turned", "gaze_off") | |
| FS_HORIZONS = (("1s", 4), ("2s", 9), ("3s", 14), ("5s", 24)) # step index k-1 of pi_off[N,25,S] for +1 / +2 / +3 / +5 s | |
| FS_NAMES = [f"fs_off_{s}_{h}" for s in FS_STATES for h, _ in FS_HORIZONS] + [f"fs_EDoff_{s}" for s in FS_STATES] + [f"fs_remE_{s}" for s in FS_STATES] \ | |
| + [f"fs_leave3_{s}" for s in FS_STATES] + ["fs_valid"] | |
| def head_thresholds(): | |
| old = json.load(open(OLD_STATS)) | |
| return float(old["head_turn_on"]), float(old["head_turn_off"]) | |
| def hysteresis_rows(x, on, off, gap_reset=GAP_RESET): | |
| """x [n, T] (NaN = invalid) -> state [n, T] in {0, 1, NaN}; per-row causal hysteresis: state initialised at the first valid step of the row and after a | |
| gap of > gap_reset invalid steps (= build_kin_causal.hysteresis run along each row).""" | |
| n, T = x.shape | |
| S = np.full((n, T), np.nan, np.float32) | |
| cur = np.zeros(n, np.float32) | |
| init = np.ones(n, bool) | |
| gap = np.full(n, gap_reset + 1, np.int64) | |
| for i in range(T): | |
| v = x[:, i] | |
| fin = np.isfinite(v) | |
| reset = fin & (init | (gap > gap_reset)) | |
| cur = np.where(reset, (v >= on).astype(np.float32), np.where(fin & (v >= on), 1.0, np.where(fin & (v < off), 0.0, cur))).astype(np.float32) | |
| init = init & ~fin | |
| gap = np.where(fin, 0, gap + 1) | |
| S[:, i] = np.where(fin, cur, np.nan) | |
| return S | |
| def _share(mask_true, valid, k): | |
| """share of valid steps (last k) that are True; NaN when no valid step.""" | |
| v = valid[:, -k:].sum(1) | |
| with np.errstate(invalid="ignore", divide="ignore"): | |
| return np.where(v > 0, (mask_true[:, -k:] & valid[:, -k:]).sum(1) / np.maximum(v, 1), np.nan).astype(np.float32) | |
| def _dwell_since(state_true, valid, cap=DWELL_CAP, hz=HZ): | |
| """seconds since the last valid step where state_true holds (0 if that is the last step); cap when never; NaN when no valid step at all.""" | |
| n, T = valid.shape | |
| idx = np.where(state_true & valid, np.arange(T)[None, :], -1).max(1) | |
| out = np.where(idx >= 0, (T - 1 - idx) / hz, cap).astype(np.float32) | |
| out[valid.sum(1) == 0] = np.nan | |
| return np.minimum(out, cap).astype(np.float32) | |
| def _ffill_state(state, valid): | |
| """forward-fill the last valid binary state along the row (NaN before the first valid step).""" | |
| n, T = state.shape | |
| idx = np.where(valid, np.arange(T)[None, :], -1) | |
| idx = np.maximum.accumulate(idx, 1) | |
| out = np.take_along_axis(np.where(valid, state, 0.0), np.clip(idx, 0, T - 1), 1) | |
| out[idx < 0] = np.nan | |
| return out | |
| def dwell_features(Kh8, Kh8valid, head_on=None, head_off=None): | |
| """Kh8 [n,80,11] (f16/f32; NaN invalid), Kh8valid [n,80] -> X float32 [n, 11] (DWELL_NAMES).""" | |
| if head_on is None or head_off is None: | |
| head_on, head_off = head_thresholds() | |
| K = np.asarray(Kh8, np.float32) | |
| V = np.asarray(Kh8valid, bool) & np.isfinite(K[:, :, IDX_MIN_DIST]) & np.isfinite(K[:, :, IDX_HEAD_DEV]) | |
| md = K[:, :, IDX_MIN_DIST] | |
| hands_on = V & (md < WHEEL_TH) | |
| hands_off = V & ~(md < WHEEL_TH) | |
| dev = np.where(V, np.abs(K[:, :, IDX_HEAD_DEV]), np.nan) | |
| away = hysteresis_rows(dev, head_on, head_off) # {0, 1, NaN} | |
| away_v = np.isfinite(away) | |
| away_t = away_v & (away > 0.5) | |
| not_away = away_v & (away < 0.5) | |
| hs = _ffill_state(hands_on.astype(np.float32), V) | |
| leave = np.zeros(len(K), np.float32) | |
| prev, cur = hs[:, :-1], hs[:, 1:] | |
| leave = ((prev > 0.5) & (cur < 0.5) & V[:, 1:]).sum(1).astype(np.float32) | |
| leave[V.sum(1) == 0] = np.nan | |
| with np.errstate(all="ignore"): | |
| dmax = np.where(np.isfinite(dev[:, -30:]).any(1), np.nanmax(np.where(np.isfinite(dev[:, -30:]), dev[:, -30:], -np.inf), 1), np.nan).astype(np.float32) | |
| X = np.stack([_share(hands_off, V, 10), _share(hands_off, V, 30), _share(hands_off, V, 80), _dwell_since(hands_on, V), | |
| _share(away_t, away_v, 10), _share(away_t, away_v, 30), _share(away_t, away_v, 80), _dwell_since(not_away, away_v), | |
| dmax, leave, np.asarray(Kh8valid, bool).mean(1).astype(np.float32)], 1).astype(np.float32) | |
| assert X.shape[1] == len(DWELL_NAMES) | |
| return X | |
| def dwell_block(regime, split, rows, kin_dir=None): | |
| d = Path(kin_dir) if kin_dir else ds.kin_dir() | |
| with np.load(d / f"kinhist_causal_{regime}_{split}.npz", allow_pickle=True) as z: | |
| assert [str(c) for c in z["channels"]][:len(ds.KIN_CHANNELS)] == ds.KIN_CHANNELS # v2p packs carry 3 extra proxy columns after the 11 (2026-09-18) | |
| assert str(z["channels"][IDX_MIN_DIST]) == "min_hand_rest_dist" and str(z["channels"][IDX_HEAD_DEV]) == "head_turn_dev" | |
| rows = np.asarray(rows, np.int64) | |
| Kh8 = z["Kh8_c"][rows][:, :, :len(ds.KIN_CHANNELS)] | |
| V = z["Kh8valid"][rows] | |
| if "row_ok" in z.files: | |
| V = V & z["row_ok"][rows][:, None] | |
| return dwell_features(Kh8, V), list(DWELL_NAMES) | |
| def mr_block(regime, split, rows): | |
| X, names = MR.features(regime, split, np.asarray(rows, np.int64)) | |
| return X.astype(np.float32), names | |
| def face_block(regime, split, rows, face_dir=FACE_DIR): | |
| p = Path(face_dir) / f"facehist_causal_{regime}_{split}.npz" | |
| assert p.exists(), f"{p} missing: run data/build_face_causal.py --source routes" | |
| rows = np.asarray(rows, np.int64) | |
| with np.load(p, allow_pickle=True) as z: | |
| ch = [str(c) for c in z["channels"]] | |
| F0 = z["F0_c"][rows].astype(np.float32) | |
| F0v = z["F0valid"][rows].astype(bool) | |
| Fh = z["Fh8_c"][rows].astype(np.float32) | |
| Fv = z["Fh8valid"][rows].astype(bool) | |
| if "row_ok" in z.files: | |
| ok = z["row_ok"][rows].astype(bool) | |
| F0v &= ok | |
| Fv &= ok[:, None] | |
| F0 = np.where(F0v[:, None], F0, np.nan) | |
| ci = {c: i for i, c in enumerate(ch)} | |
| g = Fh[:, :, ci["gaze_off"]] | |
| gv = Fv & np.isfinite(g) | |
| yd = np.where(Fv, np.abs(Fh[:, :, ci["lm_yaw_dev"]]), np.nan) | |
| eo = np.where(Fv, Fh[:, :, ci["eye_open"]], np.nan) | |
| bl = np.where(Fv, Fh[:, :, ci["blink_rate_4s"]], np.nan) | |
| with np.errstate(all="ignore"): | |
| ymax = np.where(np.isfinite(yd[:, -30:]).any(1), np.nanmax(np.where(np.isfinite(yd[:, -30:]), yd[:, -30:], -np.inf), 1), np.nan) | |
| emin = np.where(np.isfinite(eo[:, -30:]).any(1), np.nanmin(np.where(np.isfinite(eo[:, -30:]), eo[:, -30:], np.inf), 1), np.nan) | |
| bmax = np.where(np.isfinite(bl).any(1), np.nanmax(np.where(np.isfinite(bl), bl, -np.inf), 1), np.nan) | |
| H = np.stack([_share(gv & (g > 0.5), gv, 30), _share(gv & (g > 0.5), gv, 80), ymax, emin, bmax, Fv.mean(1)], 1).astype(np.float32) | |
| X = np.concatenate([F0, F0v[:, None].astype(np.float32), H], 1).astype(np.float32) | |
| names = [f"fc_{c}" for c in ch] + ["fc_valid"] + list(FACE_HIST_NAMES) | |
| assert X.shape[1] == len(names) | |
| return X, names | |
| def set_gaze_run(run_dir): | |
| rd = Path(run_dir) | |
| assert (rd / "PRED_DONE").exists() and (rd / "pred_layout_manual_test.npz").exists(), f"{rd}: not a finished U0 run" | |
| GAZE_OPTS["run_dir"] = rd | |
| return rd | |
| def gaze_block(regime, split, rows): | |
| """gaze-chain outputs of the configured U0 `*_gz*` run at `rows` of (regime, split) -> X float32 [n, 10] (GAZE_NAMES); NaN where the anchor state is undefined.""" | |
| rd = GAZE_OPTS["run_dir"] | |
| assert rd is not None, "gaze block: call set_gaze_run(<U0 gaze run dir>) first (train_warning_head.py --gaze-run)" | |
| rows = np.asarray(rows, np.int64) | |
| if split == "train": | |
| p = Path(rd) / f"pred_gaze_{regime}_train_labelrows.npz" | |
| assert p.exists(), f"{p} missing: run eval/predict_u0_rows.py --run {Path(rd).name} --rows-from <joint run>/label_rows.npz" | |
| with np.load(p, allow_pickle=True) as z: | |
| prow = z["rows"].astype(np.int64) | |
| posmap = {int(r): i for i, r in enumerate(prow)} | |
| missing = [int(r) for r in rows if int(r) not in posmap] | |
| assert not missing, f"{p}: {len(missing)} requested train rows not predicted (e.g. {missing[:5]})" | |
| pos = np.array([posmap[int(r)] for r in rows], np.int64) | |
| assert tuple(int(x) for x in z["gaze_off_taus"]) == GAZE_TAUS | |
| op, pi, z0, z0v = z["gaze_off_onset_p"][pos], z["gaze_off_pi"][pos], z["gaze_off_z0"][pos], z["gaze_off_z0valid"][pos] | |
| else: | |
| p = Path(rd) / f"pred_layout_{regime}_{split}.npz" | |
| with np.load(p, allow_pickle=True) as z: | |
| assert "gaze_off_onset_p" in z.files, f"{p}: no gaze block (not a *_gz* layout)" | |
| assert tuple(int(x) for x in z["gaze_off_taus"]) == GAZE_TAUS | |
| op, pi, z0, z0v = z["gaze_off_onset_p"][rows], z["gaze_off_pi"][rows], z["gaze_off_z0"][rows], z["gaze_off_z0valid"][rows] | |
| op, pi, z0 = op.astype(np.float32), pi.astype(np.float32), z0.astype(np.float32) | |
| X = np.concatenate([op, pi[:, list(GAZE_PI_STEPS)], z0[:, None], z0v.astype(np.float32)[:, None]], 1).astype(np.float32) | |
| assert X.shape[1] == len(GAZE_NAMES) | |
| return X, list(GAZE_NAMES) | |
| def set_fs_run(run_dir): | |
| rd = Path(run_dir) | |
| assert (rd / "PRED_DONE").exists() and (rd / "fs_manual_test.npz").exists() and (rd / "fs_manual_train.npz").exists(), f"{rd}: not a finished F2′ run with train / val / test packs" | |
| FS_OPTS["run_dir"] = rd | |
| return rd | |
| def fs_block(regime, split, rows): | |
| """F2′ predicted future-state quantities at `rows` of (regime, split) -> X float32 [n, 22] (FS_NAMES); NaN where the pack row has no causal box (fs_valid 0). | |
| The train pack covers every TRAIN anchor (row-aligned with the targets file), so label rows index it directly.""" | |
| rd = FS_OPTS["run_dir"] | |
| assert rd is not None, "fs block: call set_fs_run(<results/runs/FS_<name>_s<k>>) first (train_warning_head.py --fs-run)" | |
| rows = np.asarray(rows, np.int64) | |
| p = Path(rd) / f"fs_{regime}_{split}.npz" | |
| with np.load(p, allow_pickle=True) as z: | |
| assert [str(s) for s in z["state_names"]] == list(FS_STATES), list(z["state_names"]) | |
| off = z["off_value"].astype(np.float32) | |
| valid = z["valid"][rows].astype(bool) | |
| z0 = z["z0"][rows].astype(np.float32) | |
| z0v = z["z0valid"][rows].astype(bool) | |
| ks = [k for _, k in FS_HORIZONS] | |
| pio = np.clip(z["pi_off"][rows][:, ks, :].astype(np.float32), 0.0, 1.0) # [n, 4, S] | |
| ed = z["ED_off_s"][rows].astype(np.float32) | |
| rem = z["rem_dwell_E_s"][rows].astype(np.float32) | |
| lv3 = np.clip(z["leave_within_3s"][rows].astype(np.float32), 0.0, 1.0) | |
| is_off = z0v & (np.abs(z0 - off[None, :]) < 0.5) | |
| rem = np.where(is_off, rem, np.nan) # remaining off dwell is defined for off-task rows only | |
| X = np.concatenate([pio.transpose(0, 2, 1).reshape(len(rows), -1), ed, rem, lv3], 1).astype(np.float32) | |
| X[~valid] = np.nan | |
| X = np.concatenate([X, valid[:, None].astype(np.float32)], 1).astype(np.float32) | |
| assert X.shape[1] == len(FS_NAMES) | |
| return X, list(FS_NAMES) | |
| def kp_block(regime, split, rows, kp_dir=KP_DIR): | |
| """kin v2p pack -> X float32 [n, 5] (KP_NAMES): the current keypoint-proxy values (NaN where undefined / no causal box), the pitch dwell and a validity bit.""" | |
| p = Path(kp_dir) / f"kin_causal_{regime}_{split}.npz" | |
| assert p.exists(), f"{p} missing: run data/build_kin_causal_v2p.py" | |
| rows = np.asarray(rows, np.int64) | |
| with np.load(p, allow_pickle=True) as z: | |
| ch = [str(c) for c in z["channels_k0"]] | |
| assert ch[:len(ds.KIN_CHANNELS)] == ds.KIN_CHANNELS and ch[len(ds.KIN_CHANNELS):] == ds.KIN_PROXY_CHANNELS, ch | |
| K0 = z["K0_c"][rows].astype(np.float32) | |
| ok = z["row_ok"][rows].astype(bool) & z["K0valid"][rows].astype(bool) | |
| pd_ = z["PD0_c"][rows].astype(np.float32) | |
| P = K0[:, len(ds.KIN_CHANNELS):] | |
| P[~ok] = np.nan | |
| pd_[~ok] = np.nan | |
| valid = (ok & np.isfinite(P[:, 0])).astype(np.float32) | |
| X = np.concatenate([P, pd_[:, None], valid[:, None]], 1).astype(np.float32) | |
| assert X.shape[1] == len(KP_NAMES) | |
| return X, list(KP_NAMES) | |
| BLOCKS = {"dwell": dwell_block, "mr": mr_block, "face": face_block, "gaze": gaze_block, "fs": fs_block, "kp": kp_block} | |
| def names_for(sets): | |
| out = [] | |
| for s in sets: | |
| if s == "dwell": | |
| out += list(DWELL_NAMES) | |
| elif s == "mr": | |
| out += list(MR.FEATURE_NAMES) | |
| elif s == "face": | |
| p = FACE_DIR / "facehist_causal_manual_val.npz" | |
| assert p.exists(), f"{p} missing: run data/build_face_causal.py --source routes" | |
| with np.load(p, allow_pickle=True) as z: | |
| ch = [str(c) for c in z["channels"]] | |
| out += [f"fc_{c}" for c in ch] + ["fc_valid"] + list(FACE_HIST_NAMES) | |
| elif s == "gaze": | |
| out += list(GAZE_NAMES) | |
| elif s == "fs": | |
| out += list(FS_NAMES) | |
| elif s == "kp": | |
| out += list(KP_NAMES) | |
| else: | |
| raise ValueError(s) | |
| return out | |
| def extra_block(regime, split, rows, sets): | |
| """concatenated extra feature block for the given sets (order as given) -> X [n, k] float32, names.""" | |
| Xs, names = [], [] | |
| for s in sets: | |
| X, nm = BLOCKS[s](regime, split, rows) | |
| Xs.append(X) | |
| names += nm | |
| return (np.concatenate(Xs, 1) if Xs else np.zeros((len(rows), 0), np.float32)), names | |
| def monotone(name): | |
| """HGB monotone constraint of an extended feature: +1 (more dwell / share / pressure -> more warranted) or 0.""" | |
| if name.endswith(_MONO_ZERO_SUFFIX) or name in _MONO_ZERO_SUFFIX or name == "kp_valid": | |
| return 0 | |
| if name.startswith(("dw_", "mr_")): | |
| return 1 | |
| if name.startswith(("fc_gaze_off", "fc_yaw_dev_max")): | |
| return 1 | |
| if name.startswith("gz_"): | |
| return 1 # predicted gaze-off onset / marginal / current state: more gaze-off -> more warranted | |
| if name.startswith(("fs_off_", "fs_EDoff_", "fs_remE_")): | |
| return 1 # predicted off-task occupancy / off seconds / remaining off dwell: more -> more warranted; fs_leave3_* is sign-ambiguous (0) | |
| if name in ("kp_head_pitch_kp", "kp_pitch_dwell_s", "kp_hands_lap_kp"): | |
| return 1 # kp (2026-09-18): head further down / longer head-down / wrists deeper in the lap -> more warranted | |
| if name == "kp_wrist_face_kp": | |
| return -1 # smaller wrist-nose distance (hand nearer the face) -> more warranted | |
| return 0 | |
| def tag_for(sets): | |
| return "".join(SHORT[s] for s in ("dwell", "face", "mr", "gaze", "fs", "kp") if s in sets) | |
| if __name__ == "__main__": | |
| import argparse | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--regime", default="manual") | |
| ap.add_argument("--split", default="val") | |
| ap.add_argument("--sets", default="dwell,mr") | |
| ap.add_argument("--n", type=int, default=5000) | |
| a = ap.parse_args() | |
| ds.set_kin_source("causal_routes", "v2") | |
| rows = np.arange(a.n) | |
| X, names = extra_block(a.regime, a.split, rows, a.sets.split(",")) | |
| for j, n in enumerate(names): | |
| col = X[:, j] | |
| print(f"{n:26s} mono {monotone(n):+d} finite {np.isfinite(col).mean():.3f} mean {np.nanmean(col):.4f} p90 {np.nanpercentile(col, 90) if np.isfinite(col).any() else float('nan'):.4f}") | |
Xet Storage Details
- Size:
- 22.8 kB
- Xet hash:
- f9d3e9bed93e56120aed66027ca603ee62a8826b5c9de06482a19e40344d294e
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.