Buckets:
| """cwm3/tests/test_nocan_readout.py — the `--no-can` column rule of train/train_warning_head.py (pre-registration docs/decisions.md 2026-09-18T03:34:28Z). | |
| (a) the pre-registered column counts on the paper read-out's table (demand P + dwell + face): 43 current-state and 93 predicted-future columns KEPT; | |
| (b) every kept current-state column is one of the documented pose / face / vision quantities; every dropped column matches the documented CAN rule | |
| (vehicle anchor states, P0 / margins, R anchor, V_valid, vEgo, mode; vehicle onset3s / pi_resp / R forecasts, CAN-seeded pi_P); | |
| (c) the base table without extras keeps only the pose / vision columns (no dw_ / fc_ needed), the legacy D block is dropped entirely; | |
| (d) default off: nocan_keep is all-True only through the flag (NOCAN_OPTS default False) and the argparse flag exists; --no-can + --per-regime is rejected; | |
| (e) no forbidden openpilot DMS identifier in the edited file. | |
| Run: cd cwm3 && python -m pytest tests/test_nocan_readout.py -q | |
| """ | |
| from __future__ import annotations | |
| import os | |
| import re | |
| import sys | |
| from pathlib import Path | |
| os.environ.setdefault("OMP_NUM_THREADS", "4") | |
| import numpy as np # noqa: E402 | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| for p in (CWM3 / "train", CWM3 / "tests"): | |
| sys.path.insert(0, str(p)) | |
| import train_warning_head as TW # noqa: E402 | |
| from test_no_dms_inputs import FORBIDDEN # noqa: E402 | |
| KEEP_CUR_EXACT = {"z0_hands_on", "z0_head_turned", "K0_valid", "vf_ok", "tok_all16"} | |
| KEEP_CUR_PREFIX = ("K0_", "dw_", "fc_") | |
| KEEP_FUT_PREFIX = ("pi_hands_", "pi_head_", "K_1s_", "K_3s_", "K_5s_", "onset3s_P_") | |
| KEEP_FUT_EXACT = {"onset3s_hand_return", "onset3s_hand_leave", "onset3s_head_turn_onset", "onset3s_head_return"} | |
| def _names(extra=("dwell", "face")): | |
| return TW.feature_names("P", extra=extra) | |
| def test_counts_paper_table(): | |
| names = _names() | |
| keep = TW.nocan_keep(names) | |
| cur = TW.feature_mask(names, "current") & keep | |
| fut = TW.feature_mask(names, "future") & keep | |
| assert int(cur.sum()) == 43, [n for n, k in zip(names, cur) if k] | |
| assert int(fut.sum()) == 93, [n for n, k in zip(names, fut) if k] | |
| assert int((TW.feature_mask(names, "both") & keep).sum()) == 136 | |
| def test_kept_and_dropped_follow_the_documented_rule(): | |
| names = _names() | |
| for n in names: | |
| kept = not TW.is_can_derived(n) | |
| if TW.is_future(n): | |
| expect = n.startswith(KEEP_FUT_PREFIX) or n in KEEP_FUT_EXACT | |
| else: | |
| expect = n in KEEP_CUR_EXACT or n.startswith(KEEP_CUR_PREFIX) | |
| assert kept == expect, (n, kept, expect) | |
| dropped = [n for n in names if TW.is_can_derived(n)] | |
| for must in ("z0_gas", "z0_brake", "z0_steer_phase", "z0_resp", "z0_R_hard", "V_valid", "vEgo", "mode", "onset3s_brake_on", "pi_resp_0", "margin_ttc_s", "margin_valid_ttc_s", | |
| "P0_ttc", "pi_P_ttc_1s", "onset3s_R_hard", "pi_R_hard_1s"): | |
| assert must in dropped, must | |
| for must in ("z0_hands_on", "K0_valid", "vf_ok", "tok_all16", "dw_hands_off_dwell_s", "fc_valid", "onset3s_hand_leave", "onset3s_P_ttc", "pi_hands_0", "K_3s_min_hand_rest_dist"): | |
| assert must in names and must not in dropped, must | |
| def test_base_table_and_legacy_D_block(): | |
| names = TW.feature_names("both", extra=()) | |
| keep = TW.nocan_keep(names) | |
| kept = [n for n, k in zip(names, keep) if k] | |
| assert not any(n.startswith(("onset3s_D_", "S0_", "D0valid_")) for n in kept) | |
| assert not any(n.startswith("z0_") and n not in ("z0_hands_on", "z0_head_turned") for n in kept), "legacy D anchor states z0_<family> must be dropped" | |
| assert all(n.startswith(("z0_hands_on", "z0_head_turned", "K0_", "pi_hands_", "pi_head_", "K_1s_", "K_3s_", "K_5s_", "onset3s_P_", "onset3s_hand", "onset3s_head", "vf_ok", "tok_all16")) for n in kept) | |
| def test_flag_wiring_default_off(): | |
| assert TW.NOCAN_OPTS == {"on": False} | |
| src = (CWM3 / "train" / "train_warning_head.py").read_text() | |
| assert '"--no-can"' in src and "keep_nocan" in src and 'assert not (a.no_can and a.per_regime)' in src | |
| names = _names() | |
| assert TW.nocan_keep(names).sum() < len(names) and TW.feature_mask(names, "both").all() # the rule bites only when applied; the default masks are unchanged | |
| def test_no_forbidden_tokens(): | |
| pat = re.compile("|".join(re.escape(t) for t in FORBIDDEN)) | |
| for f in ("train/train_warning_head.py", "tests/test_nocan_readout.py"): | |
| assert not pat.search((CWM3 / f).read_text()), f | |
| def test_reference_scores_undefined_everywhere_do_not_crash(): | |
| """(f) 2026-09-18: with the vehicle branch a missing modality the run's own W_direct reference is NaN on every gold row -> NaN cells, no crash.""" | |
| y = np.array([], int) | |
| au, ci = TW.boot_auroc(y, np.array([]), np.array([]), 10, 0) | |
| assert np.isnan(au) and np.isnan(ci[0]) and np.isnan(ci[1]) | |
| au, ci = TW.boot_auroc(np.zeros(5, int), np.arange(5.0), np.array(list("aabbc")), 10, 0) | |
| assert np.isnan(au) | |
| assert np.isnan(TW.safe_ap(np.array([], int), np.array([]))) | |
| assert np.isfinite(TW.safe_ap(np.array([0, 1, 0, 1]), np.array([0.1, 0.9, 0.4, 0.6]))) | |
| assert np.isnan(TW.threshold_at_recall(np.array([], int), np.array([]), 0.5)) | |
Xet Storage Details
- Size:
- 5.29 kB
- Xet hash:
- 5ca0c154a02a1e5015b4af10d6257d4a52c86fb872fe65962a58170b7745bf07
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.