Buckets:
| """cwm3/baselines_dms/rules.py — the FIXED gaze / eye-state temporal rule and the chain-style attention composition of the DMS baselines | |
| (constants written to docs/decisions.md 2026-09-14 "DMS 基线固定规则常数与 chain-style 映射" BEFORE any scoring; never tuned on results). | |
| Rule (per anchor T, history H = [T - 8 s, T] at 10 Hz = the 80 steps of summarise.window_index; causal by construction) | |
| off(f) = |dev_yaw(f)| > 20° OR |dev_pitch(f)| > 15°, dev = ((angle - ref + 180) mod 360) - 180 (CIRCULAR, summarise.circ_dev; 2026-09-14 fix for the | |
| L2CS yaw wrap at +-180°) with hysteresis (yaw on 20° / off 12°; pitch on 15° / off 9°) and a reset after > 1 s | |
| of invalid frames (summarise.hysteresis_rows); the two angle states are OR-ed where defined | |
| closed(f) = eye_closed_p(f) > 0.5 (only for baselines with an eye-closure output; a list of channels is averaged first; else closure := 0, recorded in meta) | |
| dwell_off_s seconds since the last valid NOT-off frame (= length of the current off spell; 0 when on-road now; cap 8 s) | |
| longest_closure_s longest run of consecutive valid closed frames in H (s) | |
| share_off_3s share of valid frames in the last 3 s that are off | |
| score = clip( max( dwell_off_s / 2.0 , longest_closure_s / 1.5 , share_off_3s ) , 0 , 1 ) (NaN components ignored; all-NaN -> NaN) | |
| undefined (NaN) when valid_share_8s < 0.25 (missing policy) — the layout writer turns NaN into w_mask False | |
| variants: rule0 ref = 0 (the camera axis) | |
| ruleref ref = the route's CAUSAL expanding median of the angle over MOVING valid frames <= t (summarise.causal_expanding_median, | |
| n_min 25; computed inside the route pass — no future frame, no other route) | |
| Chain-style arm (same form as the openpilot att@3s chain: cwm2/compose.py legacy_compose att_only) | |
| the legacy chain composes three per-step attention-channel marginals pA[:, k, j] (j = distracted / gaze_off / hands_off) into | |
| p_inatt_k = 1 - prod_j (1 - pA_kj) (compose.py:47, inatt_union) | |
| att_only = 1 - prod_{k < tau} (1 - p_inatt_k) (compose.py:86 without the undocumented 0.999 shrink; tau = 15 steps = 3 s) | |
| for a DMS baseline the three channels are (decisions.md): head-pose deviation state / eye-closure probability / gaze-off state — the mapping to | |
| the legacy slots is head_dev -> "distracted" (pose-based distraction), gaze_off -> "gaze_off", eye_closed -> the third slot (hands are not | |
| observable by a face model; the legacy hands_off slot is replaced by eye closure as decisions.md fixes). The baselines are CURRENT-STATE | |
| detectors (no forecast), so pA_kj = the current channel value repeated over the 25 steps ("chain-style on current state", recorded in meta): | |
| att_only = 1 - (1 - p_inatt)^tau. Missing channel (baseline has no such output) -> 0 (never inattentive on that channel); NaN in any present | |
| channel -> NaN for the row. The exact legacy function is imported from cwm2/compose.py when importable and reproduced locally otherwise; both | |
| are asserted equal in --self-test. | |
| Nothing here reads a video, a model or any openpilot field. | |
| python baselines_dms/rules.py --self-test | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import sys | |
| import warnings | |
| from pathlib import Path | |
| import numpy as np | |
| PKG = Path(__file__).resolve().parent | |
| CWM3 = PKG.parent | |
| EXPLORE = CWM3.parent | |
| if str(PKG) not in sys.path: | |
| sys.path.insert(0, str(PKG)) | |
| import summarise as SM # noqa: E402 | |
| RULE = {"hist_s": 8.0, "hz": 10, "yaw_on_deg": 20.0, "yaw_off_deg": 12.0, "pitch_on_deg": 15.0, "pitch_off_deg": 9.0, "closed_th": 0.5, "dwell_scale_s": 2.0, "closure_scale_s": 1.5, | |
| "share_window_s": 3.0, "min_valid_share": 0.25, "gap_reset_s": 1.0, "dwell_cap_s": 8.0, "ref_n_min": 25, "ref_rule0": "0 (camera axis)", | |
| "ref_ruleref": "causal expanding median over MOVING valid frames <= t of the same route (n_min 25)", "closure_without_output": "0 (baseline has no eye-closure output)"} | |
| CHAIN = {"tau_steps": 15, "n_steps": 25, "att_shrink": None, "slots": ["head_dev", "eye_closed", "gaze_off"], | |
| "legacy_slots": ["distracted", "gaze_off", "hands_off"], "mapping": {"head_dev": "distracted (pose-based)", "gaze_off": "gaze_off", "eye_closed": "third slot (hands not observable by a face model)"}, | |
| "note": "chain-style on current state: pA[:, k, j] = current channel value for every k < 25 (the baselines forecast nothing)"} | |
| def _mean_channels(Gh, ci, chs): | |
| if chs is None: | |
| return None | |
| chs = [chs] if isinstance(chs, str) else list(chs) | |
| with np.errstate(all="ignore"), warnings.catch_warnings(): | |
| warnings.simplefilter("ignore", RuntimeWarning) | |
| return np.nanmean(np.stack([Gh[:, :, ci[c]] for c in chs], -1), -1).astype(np.float32) | |
| def off_state(yaw_h, pitch_h, Vh, ref_yaw=None, ref_pitch=None, rule=RULE): | |
| """-> (off bool[n,T], state_valid bool[n,T]); off = yaw off-state OR pitch off-state (each a hysteresis on the CIRCULAR |circ_dev(angle, ref)|); a step is state-valid when either state is defined.""" | |
| parts, valids = [], [] | |
| for x, ref, on, off in ((yaw_h, ref_yaw, rule["yaw_on_deg"], rule["yaw_off_deg"]), (pitch_h, ref_pitch, rule["pitch_on_deg"], rule["pitch_off_deg"])): | |
| if x is None: | |
| continue | |
| dev = np.abs(SM.circ_dev(np.where(Vh, x, np.nan), 0.0 if ref is None else ref)) # circular deviation in degrees (wrap-safe) | |
| st = SM.hysteresis_rows(dev, on, off, int(round(rule["gap_reset_s"] * rule["hz"]))) | |
| parts.append(np.nan_to_num(st) > 0.5) | |
| valids.append(np.isfinite(st)) | |
| assert parts, "off_state needs at least one angle channel" | |
| sv = np.logical_or.reduce(valids) | |
| return np.logical_or.reduce(parts) & sv, sv | |
| def closed_state(p_h, Vh, rule=RULE): | |
| p = np.where(Vh, p_h, np.nan) | |
| ev = np.isfinite(p) | |
| return ev & (p > rule["closed_th"]), ev | |
| def rule_score(dwell_off_s, longest_closure_s, share_off_3s, valid_share_8s, rule=RULE): | |
| """score = clip(max(dwell/2.0, closure/1.5, share3s), 0, 1); NaN components ignored (fmax), all-NaN -> NaN; valid_share_8s < 0.25 -> NaN.""" | |
| d = np.clip(np.asarray(dwell_off_s, np.float32) / rule["dwell_scale_s"], 0, 1) | |
| c = np.clip(np.asarray(longest_closure_s, np.float32) / rule["closure_scale_s"], 0, 1) | |
| s = np.clip(np.asarray(share_off_3s, np.float32), 0, 1) | |
| with np.errstate(invalid="ignore"): | |
| comp = np.fmax(np.fmax(d, c), s) | |
| comp = np.where(np.isnan(d) & np.isnan(c) & np.isnan(s), np.nan, comp) | |
| undefined = ~(np.asarray(valid_share_8s, np.float32) >= rule["min_valid_share"]) | |
| return np.where(undefined, np.nan, comp).astype(np.float32) | |
| def rule_components(off, sv, closed, ev, rule=RULE): | |
| hz = rule["hz"] | |
| k3 = int(round(rule["share_window_s"] * hz)) | |
| dwell = SM._dwell_since(sv & ~off, sv, rule["dwell_cap_s"], hz) | |
| share3 = SM._share(off, sv, k3) | |
| if closed is None: | |
| closure = np.zeros(len(off), np.float32) # baseline without an eye-closure output: closure contributes 0 (documented) | |
| else: | |
| closure = SM._longest_run_s(closed, ev, hz) | |
| return {"dwell_off_s": dwell, "share_off_3s": share3, "longest_closure_s": closure} | |
| def current_state(state, sv, tol_steps=1): | |
| """the value at the last state-valid step within tol_steps of the end (S0 semantics) -> f32[n] in {0, 1, NaN}.""" | |
| n, T = sv.shape | |
| out = np.full(n, np.nan, np.float32) | |
| for back in range(tol_steps, -1, -1): | |
| j = T - 1 - back | |
| ok = sv[:, j] | |
| out[ok] = state[ok, j].astype(np.float32) | |
| return out | |
| def rule_columns(summ, spec, rule=RULE, chain=CHAIN): | |
| """summarise_route output + spec -> {rule0, ruleref, rule0_dwell_off_s, rule0_share_off_3s, ruleref_dwell_off_s, ruleref_share_off_3s, longest_closure_s, | |
| chain_att3s_rule0, chain_att3s_ruleref, chain_head_dev, chain_eye_closed, chain_gaze_off}; every column f32[n], NaN where undefined.""" | |
| out = {} | |
| rs = spec.get("rule") if spec else None | |
| if not rs or (rs.get("yaw") is None and rs.get("pitch") is None): | |
| return out | |
| Gh, Vh, Refh, ci = summ["Gh"], summ["Vh"], summ["Refh"], {c: j for j, c in enumerate(summ["channels"])} | |
| yaw = Gh[:, :, ci[rs["yaw"]]] if rs.get("yaw") else None | |
| pitch = Gh[:, :, ci[rs["pitch"]]] if rs.get("pitch") else None | |
| p_closed = _mean_channels(Gh, ci, rs.get("closed")) | |
| closed, ev = closed_state(p_closed, Vh, rule) if p_closed is not None else (None, None) | |
| vs8 = summ["valid_share_8s"] | |
| for variant in ("rule0", "ruleref"): | |
| ry = Refh.get(rs["yaw"]) if (variant == "ruleref" and rs.get("yaw")) else None | |
| rp = Refh.get(rs["pitch"]) if (variant == "ruleref" and rs.get("pitch")) else None | |
| if variant == "ruleref" and ((rs.get("yaw") and ry is None) or (rs.get("pitch") and rp is None)): | |
| continue # the angle channel was not declared with the `angle` role -> no route reference available | |
| off, sv = off_state(yaw, pitch, Vh, ry, rp, rule) | |
| comp = rule_components(off, sv, closed, ev, rule) | |
| out[variant] = rule_score(comp["dwell_off_s"], comp["longest_closure_s"], comp["share_off_3s"], vs8, rule) | |
| out[f"{variant}_dwell_off_s"] = comp["dwell_off_s"] | |
| out[f"{variant}_share_off_3s"] = comp["share_off_3s"] | |
| if variant == "rule0": | |
| out["longest_closure_s"] = comp["longest_closure_s"] | |
| # chain-style on the current states of this variant | |
| ch = spec.get("chain") or {} | |
| cur_off = current_state(off, sv) | |
| chans = {} | |
| if ch.get("head_dev"): | |
| o2, v2 = off_state(Gh[:, :, ci[ch["head_dev"][0]]], Gh[:, :, ci[ch["head_dev"][1]]] if ch["head_dev"][1] else None, Vh, | |
| Refh.get(ch["head_dev"][0]) if variant == "ruleref" else None, Refh.get(ch["head_dev"][1]) if (variant == "ruleref" and ch["head_dev"][1]) else None, rule) | |
| chans["head_dev"] = current_state(o2, v2) | |
| else: | |
| chans["head_dev"] = None | |
| pc = _mean_channels(Gh, ci, ch.get("closed")) | |
| chans["eye_closed"] = None if pc is None else current_state(np.nan_to_num(np.where(Vh, pc, np.nan), nan=0.0), Vh & np.isfinite(pc)) | |
| if ch.get("gaze_off"): | |
| o3, v3 = off_state(Gh[:, :, ci[ch["gaze_off"][0]]], Gh[:, :, ci[ch["gaze_off"][1]]] if ch["gaze_off"][1] else None, Vh, | |
| Refh.get(ch["gaze_off"][0]) if variant == "ruleref" else None, Refh.get(ch["gaze_off"][1]) if (variant == "ruleref" and ch["gaze_off"][1]) else None, rule) | |
| chans["gaze_off"] = current_state(o3, v3) | |
| else: | |
| chans["gaze_off"] = None | |
| if not ch: # no chain mapping declared: the rule's own off state is the single channel | |
| chans = {"head_dev": cur_off, "eye_closed": None, "gaze_off": None} | |
| cst = chain_style(chans, chain["tau_steps"], chain["n_steps"], chain["att_shrink"]) | |
| out[f"chain_att3s_{variant}"] = np.where(vs8 >= rule["min_valid_share"], cst["att_only"], np.nan).astype(np.float32) | |
| if variant == "rule0": | |
| for k in ("head_dev", "eye_closed", "gaze_off"): | |
| v = cst["channels"][k] | |
| out[f"chain_{k}"] = np.zeros(len(vs8), np.float32) if v is None else v # absent channel: the 0 that enters the composition (meta: absent_channels) | |
| out["chain_absent_channels"] = cst["meta"]["absent_channels"] | |
| return out | |
| def _legacy_compose(): | |
| try: | |
| sys.path.insert(0, str(EXPLORE / "cwm2")) | |
| from compose import legacy_compose | |
| return legacy_compose, "cwm2/compose.py" | |
| except Exception: | |
| def legacy_compose(pA, p_danger, tau, att_shrink=None): | |
| p_inatt = 1 - np.prod(1 - pA, axis=-1) | |
| pi = p_inatt if att_shrink is None else p_inatt * att_shrink | |
| return {"p_inatt": p_inatt, "att_only": 1 - np.prod(1 - pi[..., :tau], axis=-1)} | |
| return legacy_compose, "local replica" | |
| def chain_style(channels, tau=CHAIN["tau_steps"], n_steps=CHAIN["n_steps"], att_shrink=CHAIN["att_shrink"]): | |
| """channels {head_dev, eye_closed, gaze_off}: f32[n] in [0, 1] (state or probability) or None (absent -> 0) -> dict(att_only [n], p_inatt [n], pA [n, n_steps, 3], channels, meta). | |
| att_only = 1 - prod_{k<tau} (1 - p_inatt) with p_inatt = 1 - prod_j (1 - p_j) (cwm2/compose.py legacy_compose, p_danger None); NaN in a present channel -> NaN row.""" | |
| slots = CHAIN["slots"] | |
| present = {k: channels.get(k) for k in slots} | |
| n = len(next(v for v in present.values() if v is not None)) | |
| P = np.zeros((n, 3), np.float64) | |
| undef = np.zeros(n, bool) | |
| for j, k in enumerate(slots): | |
| v = present[k] | |
| if v is None: | |
| continue | |
| v = np.asarray(v, np.float64).reshape(-1) | |
| assert v.shape == (n,), (k, v.shape, n) | |
| undef |= ~np.isfinite(v) | |
| P[:, j] = np.clip(np.nan_to_num(v), 0, 1) | |
| pA = np.repeat(P[:, None, :], n_steps, 1) | |
| fn, src = _legacy_compose() | |
| r = fn(pA, None, tau, att_shrink) | |
| att = np.where(undef, np.nan, r["att_only"]).astype(np.float32) | |
| pin = np.where(undef, np.nan, r["p_inatt"][:, 0]).astype(np.float32) | |
| return {"att_only": att, "p_inatt": pin, "pA": pA.astype(np.float32), | |
| "channels": {k: (None if present[k] is None else np.where(undef, np.nan, P[:, j]).astype(np.float32)) for j, k in enumerate(slots)}, | |
| "meta": {**CHAIN, "tau_steps": tau, "n_steps": n_steps, "att_shrink": att_shrink, "absent_channels": [k for k in slots if present[k] is None], "legacy_source": src}} | |
| def _self_test(): | |
| rng = np.random.default_rng(0) | |
| n = 500 | |
| # closed form: att_only = 1 - (1 - p_inatt)^tau | |
| hd = (rng.random(n) < 0.3).astype(np.float32) | |
| ec = rng.random(n).astype(np.float32) | |
| ec[:5] = np.nan | |
| r = chain_style({"head_dev": hd, "eye_closed": ec, "gaze_off": None}) | |
| p_in = 1 - (1 - hd) * (1 - np.nan_to_num(ec)) | |
| exp = 1 - (1 - p_in) ** 15 | |
| fin = np.isfinite(r["att_only"]) | |
| assert np.isnan(r["att_only"][:5]).all() and np.allclose(r["att_only"][fin], exp[fin], atol=1e-6) and r["meta"]["absent_channels"] == ["gaze_off"] | |
| fn, src = _legacy_compose() | |
| r_leg = fn(r["pA"].astype(np.float64), None, 15, None)["att_only"] | |
| assert np.allclose(r_leg[fin], r["att_only"][fin], atol=1e-6), "chain_style != legacy_compose" | |
| # binary current states give a binary att_only: 0 or 1 | |
| r2 = chain_style({"head_dev": hd, "eye_closed": None, "gaze_off": None}) | |
| assert set(np.unique(r2["att_only"]).tolist()) <= {0.0, 1.0} | |
| # rule score algebra | |
| s = rule_score([0.0, 1.0, 4.0, np.nan], [0.0, 3.0, np.nan, np.nan], [0.1, np.nan, 0.5, np.nan], [1.0, 1.0, 1.0, 1.0]) | |
| assert np.allclose(s[:3], [0.1, 1.0, 1.0]) and np.isnan(s[3]) | |
| s2 = rule_score([4.0], [0.0], [0.5], [0.2]) | |
| assert np.isnan(s2[0]), "valid_share_8s < 0.25 must give NaN" | |
| # hysteresis + dwell on a synthetic off spell | |
| T = 80 | |
| yaw = np.zeros((1, T), np.float32) | |
| yaw[0, 50:] = 25.0 # off from step 50: last not-off step = 49 -> dwell (79-49)/10 = 3.0 s, share_off_3s = 30/30 | |
| off, sv = off_state(yaw, None, np.ones((1, T), bool)) | |
| comp = rule_components(off, sv, None, None) | |
| assert abs(comp["dwell_off_s"][0] - 3.0) < 1e-6 and comp["share_off_3s"][0] == 1.0 and comp["longest_closure_s"][0] == 0.0 | |
| yaw2 = yaw.copy() | |
| yaw2[0, 60:] = 15.0 # between off 12 and on 20: hysteresis holds the off state | |
| off2, _ = off_state(yaw2, None, np.ones((1, T), bool)) | |
| assert off2[0, -1] and abs(rule_components(off2, sv, None, None)["dwell_off_s"][0] - 3.0) < 1e-6 | |
| yaw3 = yaw.copy() | |
| yaw3[0, 60:] = 10.0 # below off 12: back on-road -> dwell 0 | |
| off3, sv3 = off_state(yaw3, None, np.ones((1, T), bool)) | |
| assert not off3[0, -1] and rule_components(off3, sv3, None, None)["dwell_off_s"][0] == 0.0 | |
| # end-to-end on a synthetic cache through summarise + rule_columns | |
| rc = cs_synth() | |
| spec = {"angle": {"yaw_deg": (20, 12), "pitch_deg": (15, 9)}, "eye_state": {"closed_p": {"kind": "closed_p", "th": 0.5}}, "prob": ["conf"], | |
| "rule": {"yaw": "yaw_deg", "pitch": "pitch_deg", "closed": "closed_p"}, "chain": {"head_dev": ("yaw_deg", "pitch_deg"), "closed": "closed_p", "gaze_off": None}} | |
| anchors = np.arange(10.0, 58.0, 0.2) | |
| sm = SM.summarise_route(rc, anchors, spec) | |
| cols = rule_columns(sm, sm["spec"]) | |
| for k in ("rule0", "ruleref", "chain_att3s_rule0", "chain_att3s_ruleref", "longest_closure_s", "chain_head_dev", "chain_eye_closed"): | |
| assert k in cols and cols[k].shape == (len(anchors),), k | |
| fin0 = np.isfinite(cols["rule0"]) | |
| assert fin0.mean() > 0.9 and np.all(cols["rule0"][fin0] >= 0) and np.all(cols["rule0"][fin0] <= 1) | |
| assert np.isnan(cols["rule0"][sm["valid_share_8s"] < RULE["min_valid_share"]]).all() | |
| no_ref = ~np.isfinite(sm["Refh"]["yaw_deg"]).any(1) & ~np.isfinite(sm["Refh"]["pitch_deg"]).any(1) | |
| assert np.isnan(cols["ruleref"][no_ref]).all() # ruleref undefined while the route reference is undefined (< 25 moving valid frames) | |
| print(json.dumps({"self_test": "rules", "ok": True, "legacy_source": src, "RULE": RULE, "CHAIN_slots": CHAIN["slots"], "rule_columns": sorted(cols), | |
| "rule0_defined": float(fin0.mean()), "ruleref_defined": float(np.isfinite(cols["ruleref"]).mean()), "chain_att3s_values": sorted(set(np.unique(cols["chain_att3s_rule0"][np.isfinite(cols["chain_att3s_rule0"])]).round(3).tolist()))[:5]}, indent=1)) | |
| def cs_synth(): | |
| import cache_schema as cs | |
| return cs.synthetic_route_cache(channels=("yaw_deg", "pitch_deg", "closed_p", "conf"), t1=60.0) | |
| if __name__ == "__main__": | |
| import argparse | |
| ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) | |
| ap.add_argument("--self-test", action="store_true") | |
| ap.parse_args() | |
| _self_test() | |
Xet Storage Details
- Size:
- 18.2 kB
- Xet hash:
- 0f5386aa387bb9b928e48c17a337f2de5a366ca9389eeb425d714e20069c6118
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.