Buckets:
| """cwm3/tests/test_margin_rule.py — eval/margin_rule.py: thresholds equal the pressure builder's, synthetic ratios, and the pack's S0 columns are reproduced | |
| from M0 on moving anchors within the caps (ttc / headway / lead_decel / lateral).""" | |
| from pathlib import Path | |
| import sys | |
| import numpy as np | |
| import pytest | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| for p in (CWM3 / "eval", CWM3 / "data"): | |
| sys.path.insert(0, str(p)) | |
| import margin_rule as MR # noqa: E402 | |
| def test_thresholds_match_pressure_builder(): | |
| import build_pressure_targets as BP | |
| for k, v in MR.THR.items(): | |
| assert BP.THR[k] == v, (k, BP.THR[k], v) | |
| assert MR.MARGIN_NAMES == BP.MARGIN_NAMES | |
| def test_synthetic_ratios_and_score(): | |
| M0 = np.array([[2.0, 5.0, 40.0, -3.0, -4.0, 0.3, 3.5, 0.425, 1, 1], # ttc 2 s -> 2.0; headway 5 -> 0.5; aLead -4 -> 2.0; lat 0.3 -> 3.0; badas -> 0.5 | |
| [30.0, 20.0, 200.0, 0.0, 0.0, 2.0, 0.0, 0.0, 0, 0]], np.float32) # everything absent | |
| V = np.array([[1] * 10, [0] * 10], bool) | |
| r = MR.rule_ratios(M0, V) | |
| assert np.isclose(r["ttc"][0], 2.0) and np.isclose(r["headway"][0], 0.5) and np.isclose(r["lead_decel"][0], 2.0) and np.isclose(r["lateral"][0], 3.0) and np.isclose(r["badas"][0], 0.5) | |
| assert all(r[f][1] == 0.0 for f in MR.FAMILIES) | |
| s = MR.score(M0, V) | |
| assert np.isclose(s[0], 3.0) and s[1] == 0.0 | |
| assert MR.FEATURE_NAMES == ["mr_ttc", "mr_headway", "mr_lead_decel", "mr_lateral", "mr_badas", "mr_max", "mr_fired"] | |
| # sentinels inside a valid column (lane width 0 = lane not usable, ttc 30 = none, headway 20 = none) score 0 | |
| M1 = np.array([[30.0, 20.0, 40.0, -3.0, -4.0, 2.0, 0.0, 0.425, 1, 1]], np.float32) | |
| r1 = MR.rule_ratios(M1, np.ones((1, 10), bool)) | |
| assert r1["ttc"][0] == 0.0 and r1["headway"][0] == 0.0 and r1["lateral"][0] == 0.0 and np.isclose(r1["lead_decel"][0], 2.0) | |
| def test_reproduces_pack_S0_on_moving_anchors(regime): | |
| p = MR.PRESS / f"pressure_{regime}_val.npz" | |
| if not p.exists(): | |
| pytest.skip("pressure pack missing") | |
| with np.load(p, allow_pickle=True) as z: | |
| n = min(200_000, len(z["t"])) | |
| M0, V, S0, mv = z["M0"][:n], z["M0valid"][:n], z["S0"][:n].astype(np.float64), z["Moving0"][:n] | |
| r = MR.rule_ratios(M0, V) | |
| # every moving anchor: ttc (S0[:, 0]) / headway (S0[:, 2]) / lead_decel (S0[:, 3]) / lateral (S0[:, 4]) ratios equal the pack's S0, except the | |
| # sentinel-ambiguous rows (ttc == 30 s / headway == 20 s exactly) where S0 may hold <= 0.133 / 0.125 for a genuine far lead | |
| amb_ttc, amb_hw = M0[:, 0] >= MR.THR["ttc_none"] - 1e-3, M0[:, 1] >= MR.THR["headway_none"] - 1e-3 | |
| for fam, scol, amb in (("ttc", 0, amb_ttc), ("headway", 2, amb_hw), ("lead_decel", 3, np.zeros(n, bool)), ("lateral", 4, np.zeros(n, bool))): | |
| d = np.abs(r[fam] - S0[:, scol]) | |
| assert (mv & ~amb).sum() > 1000 and d[mv & ~amb].max() < 1e-3, (fam, d[mv & ~amb].max()) | |
| assert (d[mv & amb] <= 0.134).all(), fam | |
| X, names = MR.features(regime, "val", rows=np.arange(n)) | |
| assert X.shape == (n, len(names)) and np.isfinite(X).all() | |
Xet Storage Details
- Size:
- 3.2 kB
- Xet hash:
- baae35ff4ad86ecf77e1263a4751d5eec33df517117a214a91bc443c44fd918a
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.