Buckets:
| #!/usr/bin/env python3 | |
| """cwm3/baselines_dms/make_r0_layout.py — R0: the zero-cost FIXED gaze / eye-state rule baseline computed from our own face pack | |
| (results/targets_face_causal_routes_v2/facehist_causal_{regime}_{split}.npz; RTMW HALPE-136 landmarks, data/build_face_causal.py). | |
| No video is decoded, no model is trained, no openpilot field is read. Output = the standard warning-score layout contract | |
| (results/runs/abl_dms_lmface_rule_s0/pred_layout_{regime}_{split}.npz; uid / t byte-equal to the s12 targets pack) so that the UNCHANGED | |
| evaluators (cwm2/eval/warning_gold.py via eval/warning_head_eval.py, eval/operating_points.py) score it exactly like the CockpitWM head and the | |
| openpilot att@3s chain. Rule constants are those recorded in docs/decisions.md (2026-09-14, "DMS 基线固定规则常数"): | |
| history H = [t-8 s, t] at 10 Hz (Fh8_c, 80 steps); off(f) = gaze_off state (pack hysteresis on |lm_yaw_dev|: on 0.2121 / off 0.1033); | |
| closed(f) = eye_open < blink_th (0.6, the pack's blink threshold); dwell_off_s = seconds since the last valid NOT-off step (cap 8 s); | |
| score = clip(max(dwell_off_s / 2.0, longest_closure_s / 1.5, share_off_3s), 0, 1); undefined (NaN, w_mask False) when row_ok is False | |
| or the valid share of the 8 s window < 0.25. | |
| Columns: dms_lmface_rule0 (composite), dms_lmface_share3s, dms_lmface_dwell, dms_lmface_closure. | |
| python baselines_dms/make_r0_layout.py [--splits test val] [--out results/runs/abl_dms_lmface_rule_s0] | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import hashlib | |
| import json | |
| import sys | |
| import time | |
| from pathlib import Path | |
| import numpy as np | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| EXPLORE = CWM3.parent | |
| FACE_DIR = CWM3 / "results" / "targets_face_causal_routes_v2" | |
| S12 = EXPLORE / "s12_wm" / "results" | |
| REGIMES = ("manual", "engaged") | |
| HZ = 10.0 | |
| DWELL_CAP_S = 8.0 | |
| RULE = {"dwell_scale_s": 2.0, "closure_scale_s": 1.5, "share_window_steps": 30, "min_valid_share": 0.25, "blink_th": 0.6, "hist_s": 8.0} | |
| ARM = "abl_dms_lmface_rule" | |
| COLS = ["dms_lmface_rule0", "dms_lmface_share3s", "dms_lmface_dwell", "dms_lmface_closure"] | |
| def _dwell_since(state_true: np.ndarray, valid: np.ndarray, cap: float = DWELL_CAP_S, hz: float = HZ) -> np.ndarray: | |
| 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 _longest_run_s(state_true: np.ndarray, valid: np.ndarray, hz: float = HZ) -> np.ndarray: | |
| """longest run of consecutive steps with state_true (invalid steps break the run) -> seconds; NaN when no valid step.""" | |
| n, T = valid.shape | |
| best = np.zeros(n, np.int64) | |
| cur = np.zeros(n, np.int64) | |
| for i in range(T): | |
| on = state_true[:, i] & valid[:, i] | |
| cur = np.where(on, cur + 1, 0) | |
| best = np.maximum(best, cur) | |
| out = (best / hz).astype(np.float32) | |
| out[valid.sum(1) == 0] = np.nan | |
| return out | |
| def _share(mask_true: np.ndarray, valid: np.ndarray, k: int) -> np.ndarray: | |
| 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 rule_scores(Fh8: np.ndarray, Fv: np.ndarray, row_ok: np.ndarray, ch: list[str]) -> dict[str, np.ndarray]: | |
| ci = {c: i for i, c in enumerate(ch)} | |
| F = np.asarray(Fh8, np.float32) | |
| V = np.asarray(Fv, bool) & row_ok[:, None] | |
| g = F[:, :, ci["gaze_off"]] | |
| gv = V & np.isfinite(g) | |
| off = gv & (g > 0.5) | |
| not_off = gv & (g <= 0.5) | |
| eo = F[:, :, ci["eye_open"]] | |
| ev = V & np.isfinite(eo) | |
| closed = ev & (eo < RULE["blink_th"]) | |
| share3 = _share(off, gv, RULE["share_window_steps"]) | |
| dwell = _dwell_since(not_off, gv) # seconds in the current off-road spell (0 when currently on-road) | |
| closure = _longest_run_s(closed, ev) | |
| valid_share = V.mean(1) | |
| undefined = (~row_ok) | (valid_share < RULE["min_valid_share"]) | |
| d = np.clip(dwell / RULE["dwell_scale_s"], 0, 1) | |
| c = np.clip(closure / RULE["closure_scale_s"], 0, 1) | |
| s3 = np.clip(share3, 0, 1) | |
| with np.errstate(invalid="ignore"): | |
| comp = np.fmax(np.fmax(d, c), s3) # fmax ignores NaN components | |
| comp = np.where(np.isnan(d) & np.isnan(c) & np.isnan(s3), np.nan, comp) | |
| out = {"dms_lmface_rule0": comp, "dms_lmface_share3s": s3, "dms_lmface_dwell": d, "dms_lmface_closure": c} | |
| for k in out: | |
| out[k] = np.where(undefined, np.nan, out[k]).astype(np.float32) | |
| return out | |
| def main(argv=None) -> int: | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--splits", nargs="+", default=["test", "val"]) | |
| ap.add_argument("--out", default=str(CWM3 / "results" / "runs" / f"{ARM}_s0")) | |
| a = ap.parse_args(argv) | |
| out = Path(a.out) | |
| out.mkdir(parents=True, exist_ok=True) | |
| code_sha = hashlib.sha256(Path(__file__).read_bytes()).hexdigest() | |
| summary = {} | |
| for r in REGIMES: | |
| for s in a.splits: | |
| fp = FACE_DIR / f"facehist_causal_{r}_{s}.npz" | |
| with np.load(S12 / f"targets_{r}_{s}.npz", allow_pickle=True) as z: | |
| uid, t = z["uid"], z["t"] | |
| with np.load(fp, allow_pickle=True) as f: | |
| assert np.array_equal(f["uid"], uid) and np.array_equal(f["t"], t), f"{fp}: not aligned with the s12 pack" | |
| ch = [str(c) for c in f["channels"]] | |
| th = json.loads(str(f["thresholds"])) | |
| scores = rule_scores(f["Fh8_c"], f["Fh8valid"], f["row_ok"].astype(bool), ch) | |
| face_sha = hashlib.sha256(fp.read_bytes()).hexdigest()[:16] | |
| W = np.stack([scores[k] for k in COLS], 1).astype(np.float16) | |
| w_mask = np.isfinite(scores["dms_lmface_rule0"]) | |
| meta = {"arm": ARM, "regime": r, "seed": 0, "split": s, "score_source": "rule", "model_id": f"{ARM}:rule0", | |
| "roi": "fullframe_causal", "roi_streams": ["face"], "roi_note": "our causal driver box + HALPE face landmarks (pose_causal_routes_v2)", "driver_input_shift_s": 0.0, "code_sha256": code_sha, "dms_fields_used": [], | |
| "baseline": {"id": "R0", "kind": "fixed rule on our own HALPE face landmarks (data/build_face_causal.py)", "rule": RULE, | |
| "pack_thresholds": {k: th[k] for k in ("gaze_off_on", "gaze_off_off", "blink_th") if k in th}, | |
| "source_pack": str(fp.relative_to(CWM3)), "source_pack_sha256_16": face_sha, | |
| "missing_policy": "NaN + w_mask False when row_ok is False or valid share of the 8 s window < 0.25"}, | |
| "notes": "R0 zero-cost conventional gaze/eye-state temporal rule; comparator only, never an input to any model", "written_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())} | |
| np.savez_compressed(out / f"pred_layout_{r}_{s}.npz", meta=np.array(json.dumps(meta)), uid=uid, t=t, w_names=np.array(COLS), w=W, w_mask=w_mask) | |
| summary[f"{r}_{s}"] = {"n": int(len(uid)), "defined": int(w_mask.sum()), "defined_share": float(w_mask.mean()), | |
| "score_mean": float(np.nanmean(scores["dms_lmface_rule0"])), "score_p95": float(np.nanpercentile(scores["dms_lmface_rule0"], 95))} | |
| print(f"wrote {out / f'pred_layout_{r}_{s}.npz'}: {summary[f'{r}_{s}']}") | |
| (out / "PRED_DONE").write_text(time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + "\n") | |
| json.dump({"arm": ARM, "rule": RULE, "columns": COLS, "code_sha256": code_sha, "summary": summary}, open(out / "make_r0_layout_summary.json", "w"), indent=1) | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 7.79 kB
- Xet hash:
- 6af7098857096c1f4aaf6e50fbfb939c928d399047896b63f7adb11dccb9dac3
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.