Buckets:
| """cwm3/eval/margin_rule.py — the current-margin RULE score of the road-pressure families, recomputed from the targets_pressure M0 fields | |
| (innovation 3 b1 interface, 2026-09-11; docs/innovations_report.md §3). CPU, numpy only; importable by the warning head. | |
| Interface (for train/train_warning_head.py or the final head): | |
| import margin_rule as MR | |
| X, names = MR.features(regime, split, rows) # float32 [n, len(MR.FEATURE_NAMES)]: mr_ttc, mr_headway, mr_lead_decel, mr_lateral, mr_badas, mr_max, mr_fired | |
| s = MR.score(M0, M0valid) # float64 [n]: the scalar rule score = max threshold-normalised margin ratio over the available families | |
| r = MR.rule_ratios(M0, M0valid) # dict family -> float64 [n] ratio (> 1 = the family's threshold is crossed at t; 0 where the signal is absent) | |
| Definitions = data/build_pressure_targets.route_pressure's S rows (the `margin rule S0` baseline of exp3_P.md) recomputed from the CURRENT margins M0 | |
| (margin_names ttc_s, headway_s, dRel_m, vRel_mps, aLead_mps2, lat_margin_m, lane_width_m, badas, lead_valid, lane_valid), so the score needs no future and no model: | |
| ttc THR.ttc / max(ttc_s, 0.1) (ttc_s is capped at ttc_none = 30 s in M0 -> 0.133 when far / no lead; 0 where invalid) | |
| headway THR.cut_in_headway_s / max(headway_s, 0.1) (the cut-in closeness proxy; headway capped at 20 s) | |
| lead_decel aLead_mps2 / THR.lead_brake_a (positive when the lead decelerates; 1 at -2 m/s^2) | |
| lateral THR.veh_half_w / max(lat_margin_m, 0.05) (lane valid only; 1 when the half vehicle width reaches the lane edge) | |
| badas badas / THR.badas_abs (the pack's own S0 uses max(badas_abs, route 95th percentile), which M0 cannot reproduce -> absolute floor) | |
| The ttc / headway / lead_decel / lateral ratios reproduce the pack's S0 columns exactly on moving anchors (tests/test_margin_rule.py); a genuine ttc >= 30 s / headway >= 20 s is | |
| indistinguishable from the sentinel in M0 and scores 0 (S0 would give <= 0.133 / 0.125 there). | |
| No openpilot DMS field is read (inputs: radar / lane / BADAS margins of targets_pressure, derived from the CAN whitelist through canspec). | |
| """ | |
| from __future__ import annotations | |
| from pathlib import Path | |
| import numpy as np | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| PRESS = CWM3 / "results" / "targets_pressure" | |
| MARGIN_NAMES = ["ttc_s", "headway_s", "dRel_m", "vRel_mps", "aLead_mps2", "lat_margin_m", "lane_width_m", "badas", "lead_valid", "lane_valid"] | |
| # thresholds = data/build_pressure_targets.THR (asserted equal in tests/test_margin_rule.py) | |
| THR = dict(ttc=4.0, cut_in_headway_s=2.5, lead_brake_a=-2.0, veh_half_w=0.9, badas_abs=0.85, ttc_none=30.0, headway_none=20.0, lat_none=2.0) | |
| FAMILIES = ("ttc", "headway", "lead_decel", "lateral", "badas") | |
| FEATURE_NAMES = [f"mr_{f}" for f in FAMILIES] + ["mr_max", "mr_fired"] | |
| _COL = {n: i for i, n in enumerate(MARGIN_NAMES)} | |
| def rule_ratios(M0, M0valid): | |
| """threshold-normalised current-margin ratios per family (0 where the signal is absent / invalid, as the pack's S0).""" | |
| M0, V = np.asarray(M0, np.float64), np.asarray(M0valid, bool) | |
| ttc, hw, al, lat, bd = (M0[:, _COL[c]] for c in ("ttc_s", "headway_s", "aLead_mps2", "lat_margin_m", "badas")) | |
| v_ttc, v_hw, v_al, v_lat, v_bd = (V[:, _COL[c]] for c in ("ttc_s", "headway_s", "aLead_mps2", "lat_margin_m", "badas")) | |
| # M0 encodes "signal absent" as a sentinel inside a VALID column (ttc_none 30 s, headway_none 20 s, lat_none 2.0 m with lane_width 0 when the lane is | |
| # not usable): those rows carry no pressure (the pack's S0 is 0 there), so the sentinels are mapped to 0 rather than to 4/30 etc. | |
| v_ttc = v_ttc & (ttc < THR["ttc_none"]) | |
| v_hw = v_hw & (hw < THR["headway_none"]) | |
| v_lat = v_lat & (M0[:, _COL["lane_width_m"]] > 0) # lane_ok = lane valid AND width inside [lane_w_min, lane_w_max] | |
| return {"ttc": np.where(v_ttc, THR["ttc"] / np.maximum(ttc, 0.1), 0.0), | |
| "headway": np.where(v_hw, THR["cut_in_headway_s"] / np.maximum(hw, 0.1), 0.0), | |
| "lead_decel": np.where(v_al, al / THR["lead_brake_a"], 0.0), | |
| "lateral": np.where(v_lat, THR["veh_half_w"] / np.maximum(lat, 0.05), 0.0), | |
| "badas": np.where(v_bd, bd / THR["badas_abs"], 0.0)} | |
| def score(M0, M0valid): | |
| """scalar rule score = max ratio over the families (> 1 = some current-margin threshold is crossed).""" | |
| r = rule_ratios(M0, M0valid) | |
| return np.max(np.stack([r[f] for f in FAMILIES], 1), 1) | |
| def features(regime, split, rows=None, pack_dir=PRESS): | |
| """[n, 7] float32 feature block for the warning head + names; rows = anchor rows of the (regime, split) targets file (None = all).""" | |
| with np.load(Path(pack_dir) / f"pressure_{regime}_{split}.npz", allow_pickle=True) as z: | |
| assert [str(m) for m in z["margin_names"]] == MARGIN_NAMES, list(z["margin_names"]) | |
| M0, V = (z["M0"], z["M0valid"]) if rows is None else (z["M0"][rows], z["M0valid"][rows]) | |
| r = rule_ratios(M0, V) | |
| mx = np.max(np.stack([r[f] for f in FAMILIES], 1), 1) | |
| X = np.stack([r[f] for f in FAMILIES] + [mx, (mx > 1.0).astype(np.float64)], 1).astype(np.float32) | |
| return X, list(FEATURE_NAMES) | |
| if __name__ == "__main__": | |
| import argparse | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--regime", default="manual") | |
| ap.add_argument("--split", default="val") | |
| a = ap.parse_args() | |
| X, names = features(a.regime, a.split) | |
| for j, n in enumerate(names): | |
| print(f"{n:14s} mean {X[:, j].mean():.4f} >1 share {(X[:, j] > 1).mean():.4f}") | |
Xet Storage Details
- Size:
- 5.69 kB
- Xet hash:
- d43401674292866c205db74dfb51f4cac9daafa1eb6c1510ef7fba9231c5079b
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.