Buckets:
| """cwm3/tests/test_newtest_sample.py — unit tests of the new-test-set sampler (eval/newtest_sample.py) on SYNTHETIC frames, plus sanity checks of the | |
| real manifest when results/newtest/manifest_v1.csv exists. | |
| Synthetic: (a) alert_rows == train_warning_head.alerts_per_hour count (per-route 60 s cooldown, (uid, t) visiting order) on random scores; | |
| (b) merge_events: every event's alerts lie within 10 s of its first alert and consecutive events of a route are > 10 s apart; the pattern is the set of systems; | |
| (c) exclusion: near_any removes exactly the anchors within 12.5 s; (d) allocation: allocate_capped respects availability and totals, stratified_draw respects the | |
| device cap, pi in (0, 1], sum of 1/pi over the draw is within the design tolerance of N (HT unbiasedness for the total), no duplicated unit, guard respected; | |
| (e) seed reproducibility: the same seed gives the identical draw, another seed a different one. | |
| Real manifest (skipped when absent): unique idx / clip_id; pi in (0, 1]; per S1 cell n <= target and sum 1/pi ~= N_cell; S1 / S2 anchors >= 12.5 s from every | |
| exclusion anchor (recomputed from the same sources); S2 anchors >= 15 s apart per route and >= 12.5 s from S1 union anchors of the manifest; S1 pattern == | |
| systems with a frame alert in [t-10, t+10]; per-device counts within the caps; the 12 calibration clips are 6 / 6 by regime and 4 per stratum. | |
| Run: cd cwm3 && python -m pytest tests/test_newtest_sample.py -q | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import sys | |
| from collections import Counter, defaultdict | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| import pytest | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| for _p in (CWM3 / "eval", CWM3 / "train", CWM3 / "common", CWM3.parent / "cwm2" / "eval", CWM3.parent / "cwm2"): | |
| sys.path.insert(0, str(_p)) | |
| import newtest_sample as NS # noqa: E402 | |
| MAN = CWM3 / "results" / "newtest" / "manifest_v1.csv" | |
| def _synthetic_routes(rng, n_routes=12, n_rows=400): | |
| uid = np.repeat(np.array([f"R{i:02d}" for i in range(n_routes)]), n_rows) | |
| t = np.tile(4.0 + 0.2 * np.arange(n_rows), n_routes) | |
| s = rng.random(len(uid)) | |
| s[rng.random(len(uid)) < 0.02] = np.nan | |
| perm = rng.permutation(len(uid)) # rows deliberately out of (uid, t) order | |
| return uid[perm], t[perm], s[perm] | |
| def test_alert_rows_matches_train_warning_head(): | |
| TW = pytest.importorskip("train_warning_head") | |
| rng = np.random.default_rng(0) | |
| uid, t, s = _synthetic_routes(rng) | |
| for thr in (0.5, 0.9, 0.99, 0.999): | |
| rows = NS.alert_rows(uid, t, s, thr) | |
| assert len(rows) == TW.alerts_per_hour(uid, t, s, thr)[0] | |
| assert NS.alerts_per_hour(uid, t, s, thr)[2] == pytest.approx(TW.alerts_per_hour(uid, t, s, thr)[2]) | |
| # cooldown: consecutive alerts of a route are >= 60 s apart and every alert row fires | |
| assert np.all(s[rows] >= thr) | |
| by = defaultdict(list) | |
| for r in rows: | |
| by[uid[r]].append(t[r]) | |
| for v in by.values(): | |
| v = np.sort(v) | |
| assert np.all(np.diff(v) >= 60.0 - 1e-9) | |
| def test_merge_events_guard_and_pattern(): | |
| alerts = [("R1", 100.0, "A", "manual"), ("R1", 104.0, "C", "manual"), ("R1", 109.9, "B", "manual"), ("R1", 110.1, "A", "manual"), | |
| ("R1", 300.0, "B", "manual"), ("R2", 50.0, "C", "engaged"), ("R2", 55.0, "C", "engaged")] | |
| ev = NS.merge_events(alerts, 10.0) | |
| assert len(ev) == 4 | |
| e1 = next(e for e in ev if e["uid"] == "R1" and e["t"] == 100.0) | |
| assert NS.pattern_of(e1["systems"]) == "A+B+C" and len(e1["alerts"]) == 3 # 110.1 is > 10 s after the first alert -> new event | |
| e2 = next(e for e in ev if e["uid"] == "R1" and e["t"] == 110.1) | |
| assert NS.pattern_of(e2["systems"]) == "A" | |
| for u in ("R1", "R2"): | |
| ts = sorted(e["t"] for e in ev if e["uid"] == u) | |
| assert all(b - a_ > 10.0 for a_, b in zip(ts, ts[1:])) | |
| for e in ev: | |
| assert all(t_ - e["t"] <= 10.0 for t_, _, _ in e["alerts"]) | |
| assert NS.pattern_of({"C", "A"}) == "A+C" | |
| def test_near_any_exclusion(): | |
| anchors = {"R1": np.array([100.0, 500.0])} | |
| assert NS.near_any(anchors, "R1", 112.5, 12.5) and NS.near_any(anchors, "R1", 87.5, 12.5) | |
| assert not NS.near_any(anchors, "R1", 112.6, 12.5) and not NS.near_any(anchors, "R1", 300.0, 12.5) | |
| assert not NS.near_any(anchors, "R9", 100.0, 12.5) | |
| def test_allocate_capped_and_largest_remainder(): | |
| assert sum(NS.largest_remainder(14, [235, 60, 5])) == 14 and NS.largest_remainder(0, [1, 2]) == [0, 0] | |
| N = {"d1": 100, "d2": 50, "d3": 10, "d4": 1} | |
| al = NS.allocate_capped(N, 20, {"d1": 3, "d2": 50, "d3": 10, "d4": 1}) | |
| assert sum(al.values()) == 20 and al["d1"] == 3 and all(al[k] <= N[k] for k in N) | |
| al2 = NS.allocate_capped(N, 500, {k: N[k] for k in N}) # more asked than available -> everything | |
| assert al2 == N | |
| def _units(rng, n_dev=8, n=600): | |
| devs = rng.choice([f"D{i}" for i in range(n_dev)], n, p=np.r_[0.5, np.full(n_dev - 1, 0.5 / (n_dev - 1))]) | |
| out = [] | |
| for i in range(n): | |
| out.append(dict(uid=f"route_{devs[i]}_{i % 7}", t=float(rng.integers(0, 4000)), dongle=str(devs[i]), regime="manual")) | |
| return out | |
| def test_stratified_draw_caps_pi_and_ht(): | |
| rng = np.random.default_rng(1) | |
| units = _units(rng) | |
| dev_used = Counter() | |
| drawn, table, cov = NS.stratified_draw(units, 40, 10, dev_used, np.random.default_rng(5)) | |
| assert len(drawn) == 40 and max(dev_used.values()) <= 10 | |
| assert all(0 < u["pi"] <= 1 for u in drawn) | |
| assert len({(u["uid"], u["t"]) for u in drawn}) == len(drawn) | |
| ht_total = sum(1 / u["pi"] for u in drawn) | |
| assert ht_total == pytest.approx(cov["N_covered"], rel=1e-9) # stratified SRS: sum 1/pi over the draw == N of the covered strata exactly | |
| assert cov["N_covered"] + cov["N_uncovered"] == len(units) and 0 < cov["coverage"] <= 1 | |
| # a second cell on the same devices: the capped device (D0 holds ~50 % of the units, 10 draws already) becomes its own budget-limited stratum with n <= remaining | |
| drawn_b, table_b, cov_b = NS.stratified_draw(units, 40, 10, dev_used, np.random.default_rng(6)) | |
| assert max(dev_used.values()) <= 10 and sum(1 / u["pi"] for u in drawn_b) == pytest.approx(cov_b["N_covered"], rel=1e-9) | |
| assert all(r["n"] <= r["avail"] for r in table_b) | |
| # with a guard: drawn anchors of a route are >= guard apart | |
| drawn2, _, _ = NS.stratified_draw(units, 30, 6, Counter(), np.random.default_rng(2), guard=15.0) | |
| by = defaultdict(list) | |
| for u in drawn2: | |
| by[u["uid"]].append(u["t"]) | |
| assert all(np.all(np.diff(np.sort(v)) >= 15.0) for v in by.values() if len(v) > 1) | |
| assert NS.exact_device_coverage(units, 40) <= 1.0 | |
| def test_seed_reproducibility(): | |
| rng = np.random.default_rng(3) | |
| units = _units(rng) | |
| a = NS.stratified_draw(units, 40, 10, Counter(), np.random.default_rng(11))[0] | |
| b = NS.stratified_draw(units, 40, 10, Counter(), np.random.default_rng(11))[0] | |
| c = NS.stratified_draw(units, 40, 10, Counter(), np.random.default_rng(12))[0] | |
| key = lambda lst: [(u["uid"], u["t"]) for u in lst] # noqa: E731 | |
| assert key(a) == key(b) and key(a) != key(c) | |
| def test_route_rows_lookup(): | |
| uid = np.array(["b", "a", "a", "b"]) | |
| t = np.array([10.0, 4.0, 4.2, 10.2]) | |
| R = NS.RouteRows(uid, t) | |
| assert R.row("a", 4.2) == 2 and R.row("b", 10.05) == 0 and R.row("a", 4.5) is None and R.row("z", 4.0) is None | |
| assert np.isnan(NS.first_in_window({"a": np.array([100.0])}, "a", 50.0)) | |
| assert NS.first_in_window({"a": np.array([95.0, 100.0])}, "a", 104.0) == 95.0 | |
| # ----------------------------------------------------------------------------------------------------------------------- real manifest | |
| def test_real_manifest_sanity(): | |
| M = pd.read_csv(MAN) | |
| assert M.idx.is_unique and M.clip_id.is_unique | |
| assert ((M.pi > 0) & (M.pi <= 1)).all() | |
| S1 = M[M.stratum == "S1"] | |
| for cell, g in S1.groupby("cell"): | |
| p = cell.split("|")[1] | |
| assert len(g) <= NS.S1_ALLOC[p] | |
| assert (1 / g.pi).sum() == pytest.approx(g.N_covered_cell.iloc[0], rel=1e-9) # HT total over the covered strata | |
| assert g.N_covered_cell.iloc[0] <= g.N_cell.iloc[0] and 0 < g.coverage_cell.iloc[0] <= 1 | |
| assert S1.groupby("dongle").size().max() <= NS.S1_DEV_CAP | |
| S2 = M[M.stratum == "S2"] | |
| assert S2.groupby("dongle").size().max() <= NS.S2_DEV_CAP and len(S2) <= 2 * NS.S2_PER_REGIME | |
| for _, g in S2.groupby("uid"): | |
| assert np.all(np.diff(np.sort(g.t_anchor.to_numpy())) >= NS.S2_GUARD_S) | |
| # S1 pattern (window rule) == systems with a frame alert in [t-10, t+10]; the cluster pattern's systems all have such an alert | |
| for r in S1.itertuples(): | |
| sys_ = {X for X in NS.SYSTEMS if np.isfinite(getattr(r, f"t_alert_{X}_frame"))} | |
| assert NS.pattern_of(sys_) == r.pattern | |
| assert set(r.pattern_cluster.split("+")) <= sys_ | |
| assert all(abs(getattr(r, f"t_alert_{X}_frame") - r.t_anchor) <= NS.ALERT_WIN_S + 1e-6 for X in sys_) | |
| # S2 anchors are >= 12.5 s from every S1 anchor of the manifest (the frame excludes all union events) | |
| s1_by = defaultdict(list) | |
| for r in S1.itertuples(): | |
| s1_by[r.uid].append(r.t_anchor) | |
| for r in S2.itertuples(): | |
| assert all(abs(r.t_anchor - x) > NS.EXCL_S for x in s1_by.get(r.uid, [])) | |
| # exclusion anchors (same sources as the sampler) | |
| try: | |
| excl, _, _ = NS.exclusion_anchors(lambda m: None) | |
| except Exception as ex: # noqa: BLE001 | |
| pytest.skip(f"exclusion sources unavailable: {ex!r}") | |
| for r in pd.concat([S1, S2]).itertuples(): | |
| assert not NS.near_any(excl, r.uid, r.t_anchor, NS.EXCL_S), (r.uid, r.t_anchor) | |
| cal = M[M.calibration == True] # noqa: E712 | |
| assert len(cal) == 8 and cal.regime.value_counts().to_dict() == {"manual": 4, "engaged": 4} | |
| assert cal.stratum.value_counts().to_dict() == {"S1": 3, "S2": 3, "S3": 2} | |
| fc = json.load(open(MAN.parent / "frame_counts_v1.json")) | |
| assert fc["S1"]["union_events"] == sum(sum(v.values()) for v in fc["S1"]["cells"].values()) | |
| thr = json.load(open(MAN.parent / "thresholds_frozen_v1.json")) | |
| assert all(not thr["thresholds"]["C"][r]["val10"]["reachable"] for r in NS.REGIMES) # the DMS chain cannot reach 10/h | |
Xet Storage Details
- Size:
- 10.6 kB
- Xet hash:
- a8db0d0d4c166e928b1113ca3c52ba3c76482dbbf902263cb47ae6418149e72b
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.