Buckets:
| """cwm3/tests/test_operating_points_pair.py — unit checks of the 2026-09-17 additive paired ours − comparator helpers of eval/operating_points.py | |
| (far_rate, parse_pair_spec, read_pub_thresholds, multi_verdict) and of the invariant that the flags are optional (defaults keep the historical behaviour). | |
| Run: cd cwm3 && python -m pytest tests/test_operating_points_pair.py -q | |
| """ | |
| from __future__ import annotations | |
| import importlib.util | |
| import sys | |
| import tempfile | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| import pytest | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| def OP(): | |
| spec = importlib.util.spec_from_file_location("operating_points", CWM3 / "eval" / "operating_points.py") | |
| mod = importlib.util.module_from_spec(spec) | |
| spec.loader.exec_module(mod) | |
| return mod | |
| def test_far_rate(OP): | |
| y = np.array([1, 0, 0, 0, 1]) | |
| hit = np.array([True, True, False, False, False]) | |
| assert OP.far_rate(y, hit) == pytest.approx(1 / 3) | |
| assert np.isnan(OP.far_rate(np.array([1, 1]), np.array([True, False]))) | |
| assert OP.far_rate(y, np.zeros(5, bool)) == 0.0 | |
| def test_parse_pair_spec(OP): | |
| lab, pat, col, opdir = OP.parse_pair_spec("VideoMAEv2=abl_dms_videomae2_learned_full_s{seed}:dms_videomae2_learned_sel_t15_w10_n979:results/operating_points/x_20260916T042722Z") | |
| assert lab == "VideoMAEv2" and pat.format(seed=2) == "abl_dms_videomae2_learned_full_s2" and col == "dms_videomae2_learned_sel_t15_w10_n979" | |
| assert opdir == CWM3 / "results" / "operating_points" / "x_20260916T042722Z" | |
| lab2, _, _, opdir2 = OP.parse_pair_spec("OMZ gaze=abl_dms_ov_gaze_learned_full_s{seed}:dms_ov_gaze_learned_sel_t15_w10_n979") | |
| assert lab2 == "OMZ gaze" and opdir2 is None | |
| with pytest.raises(AssertionError): | |
| OP.parse_pair_spec("bad=run_without_seed:col") | |
| def test_read_pub_thresholds(OP): | |
| with tempfile.TemporaryDirectory() as d: | |
| rows = [] | |
| for load, thm, the in (("dms_matched", 0.18, 0.19), ("low10", 0.66, 0.78)): | |
| for score in ("head", "dms"): | |
| for reg in ("manual", "engaged", "pooled"): | |
| rows.append({"w_run": "arm_s0", "seed": 0, "load": load, "score": score, "regime": reg, "thr_manual": thm if score == "head" else 0.99, "thr_engaged": the if score == "head" else 0.99, | |
| "load_target_manual": 55.5, "load_target_engaged": 49.3, "alerts_per_hour_manual": 55.6, "alerts_per_hour_engaged": 49.4}) | |
| pd.DataFrame(rows).to_csv(Path(d) / "operating_points.csv", index=False) | |
| T = OP.read_pub_thresholds(d, "arm_s0", 0) | |
| assert set(T) == {"dms_matched", "low10"} | |
| assert T["low10"]["thr"] == {"manual": 0.66, "engaged": 0.78} and T["dms_matched"]["thr"]["manual"] == 0.18 | |
| assert T["low10"]["aph"]["engaged"] == 49.4 and T["low10"]["target"]["manual"] == 55.5 | |
| assert OP.read_pub_thresholds(d, "arm_s1", 1) == {} | |
| def test_multi_verdict_rule(OP): | |
| # 3 seeds, all CIs exclude 0, same sign, |mean| > sd -> OURS HIGHER (3/3) | |
| assert OP.multi_verdict([0.2, 0.25, 0.22], [0.05, 0.1, 0.08], [0.35, 0.4, 0.36], "OURS HIGHER", "COMP HIGHER") == "OURS HIGHER (3/3)" | |
| # 2 of 3 exclude 0, same sign -> passes the >= ceil(2n/3) rule | |
| assert OP.multi_verdict([-0.2, -0.25, -0.1], [-0.35, -0.4, -0.3], [-0.05, -0.1, 0.1], "OURS HIGHER", "COMP HIGHER") == "COMP HIGHER (2/3)" | |
| # 1 of 3 -> n.s. | |
| assert OP.multi_verdict([0.2, 0.05, 0.02], [0.05, -0.1, -0.1], [0.35, 0.2, 0.15], "A", "B").startswith("n.s. (1/3") | |
| # signs differ -> n.s. even when CIs exclude 0 | |
| assert "signs differ" in OP.multi_verdict([0.2, -0.2, 0.2], [0.1, -0.3, 0.1], [0.3, -0.1, 0.3], "A", "B") | |
| # |mean| <= sd -> n.s. | |
| assert OP.multi_verdict([0.01, 0.5, 0.02], [0.001, 0.1, 0.001], [0.02, 0.9, 0.04], "A", "B").startswith("n.s.") | |
| # single seed: its CI decides | |
| assert OP.multi_verdict([0.3], [0.1], [0.5], "A", "B") == "A (1/1)" | |
| assert OP.multi_verdict([np.nan], [np.nan], [np.nan], "A", "B") == "—" | |
| def test_flags_are_optional_and_defaults_unchanged(): | |
| src = (CWM3 / "eval" / "operating_points.py").read_text() | |
| assert 'tok.startswith("low")' in src | |
| assert '"--pair-runs", nargs="*", default=[]' in src and '"--pair-surrogate", action="store_true"' in src | |
| assert 'default=str(OUT_ROOT)' in src and 'default=str(TABLE)' in src | |
| # the extra CSVs exist only when the option was used | |
| assert 'if len(rows_pub) or len(rows_pubskip):' in src and 'if len(rows_pub):' in src | |
Xet Storage Details
- Size:
- 4.53 kB
- Xet hash:
- b46935d1cefcadc2fdfa8e155903372cce0785aa91384134d88950f0bbc81ad7
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.