Spaces:
Paused
Paused
File size: 5,804 Bytes
953f0c4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | """Frozen-plan bias prevention, allocation properties, paired harness and the
prediction cache."""
from __future__ import annotations
import numpy as np
import pytest
from sensorflow.seqeval import paired, planner
SAFETY = ["pedestrian|night", "cyclist|night"]
def _plan(meta, **kw):
args = dict(target_n=6000, seed=99, safety_primaries=SAFETY,
min_per_stratum=120, safety_floor=800, persist=False)
args.update(kw)
return planner.build_plan(meta, **args)
class TestFrozenPlan:
def test_plan_hash_reproducible_and_candidate_independent(self, seq_env):
"""The planner signature admits no candidate/outcome inputs; identical
(population, config, seed) must give identical plans."""
meta = seq_env["meta"]
p1 = _plan(meta)
p2 = _plan(meta)
assert p1["plan_hash"] == p2["plan_hash"]
assert p1["frozen_before_candidate_outcomes"] is True
def test_plan_hash_changes_with_seed_and_config(self, seq_env):
meta = seq_env["meta"]
assert _plan(meta)["plan_hash"] != _plan(meta, seed=100)["plan_hash"]
assert _plan(meta)["plan_hash"] != _plan(meta, target_n=7000)["plan_hash"]
def test_allocation_floors_and_weights(self, seq_env):
meta = seq_env["meta"]
plan = _plan(meta)
for s, v in plan["strata"].items():
assert v["allocated"] <= v["N"]
assert v["allocated"] >= min(120, v["N"])
# HT weight consistent with allocation
assert v["weight"] == pytest.approx(v["N"] / v["allocated"], rel=0.01)
if v["key"] in SAFETY:
assert v["safety_primary"]
assert v["allocated"] >= min(800, int(0.85 * v["N"]))
def test_whole_clusters_selected(self, seq_env):
"""Every selected container contributes ALL of its stratum objects, so
the cluster inference unit is intact."""
meta = seq_env["meta"]
plan = _plan(meta)
frame = paired.load_frame(meta)
s = next(int(k) for k, v in plan["strata"].items()
if v["key"] == "pedestrian|night")
clusters = planner.stratum_clusters(plan, s)
assert clusters, "expected non-empty stratum"
for arr in clusters[:50]:
cid = frame["container_id"][arr]
assert np.unique(cid).size == 1 # one container per cluster unit
members = np.where((frame["container_id"] == cid[0])
& (frame["class"] == 1) & (frame["lighting"] == 1))[0]
assert set(arr.tolist()) == set(members.tolist())
def test_persisted_plan_roundtrip(self, seq_env):
meta = seq_env["meta"]
plan = _plan(meta, persist=True)
loaded = planner.load_plan(plan["plan_id"])
assert loaded is not None
assert loaded["plan_hash"] == plan["plan_hash"]
s = next(iter(loaded["strata"]))
np.testing.assert_array_equal(loaded["_arrays"][f"s{s}_ids"],
plan["_arrays"][f"s{s}_ids"])
class TestPairedHarness:
def test_planted_effect_shows_up_where_planted(self, seq_env):
meta = seq_env["meta"]
frame = paired.load_frame(meta)
base = paired.SimulatedModel("harness-base")
cand = paired.SimulatedModel("harness-cand",
{"pedestrian|night": -0.05, "__global__": 0.0})
ids = np.arange(meta["num_objects"], dtype=np.int64)
oc = paired.paired_outcomes(meta, base, cand, ids)
ped_night = (frame["class"] == 1) & (frame["lighting"] == 1)
delta_in = oc["d"][ped_night].mean()
delta_out = oc["d"][~ped_night].mean()
assert delta_in < -0.03
assert abs(delta_out) < 0.005
def test_pairing_reduces_variance(self, seq_env):
"""Paired discordance must be far below what independent runs of two
models with these accuracies would produce."""
meta = seq_env["meta"]
base = paired.SimulatedModel("harness-base")
cand = paired.SimulatedModel("harness-cand2")
ids = np.arange(meta["num_objects"], dtype=np.int64)
oc = paired.paired_outcomes(meta, base, cand, ids)
p = oc["baseline"].mean()
independent_disagreement = 2 * p * (1 - p)
paired_disagreement = (oc["baseline"] != oc["candidate"]).mean()
assert paired_disagreement < 0.4 * independent_disagreement
def test_deterministic_given_fingerprints(self, seq_env):
meta = seq_env["meta"]
m = paired.SimulatedModel("det-check", {"pedestrian|night": -0.02})
ids = np.arange(0, 5000, dtype=np.int64)
a = paired.paired_outcomes(meta, paired.SimulatedModel("det-base"), m, ids)
paired.get_prediction_cache().clear_memory()
b = paired.paired_outcomes(meta, paired.SimulatedModel("det-base"), m, ids)
np.testing.assert_array_equal(a["d"], b["d"])
class TestPredictionCache:
def test_baseline_not_recomputed_across_candidates(self, seq_env):
meta = seq_env["meta"]
paired.reset_prediction_cache()
base = paired.SimulatedModel("cache-base-v1")
key = f"{paired.dataset_fingerprint(meta)}-{base.fingerprint()}"
ids = np.arange(0, 20_000, dtype=np.int64)
for i in range(3): # three successive candidate updates
cand = paired.SimulatedModel(f"cache-cand-v{i}")
paired.paired_outcomes(meta, base, cand, ids)
assert paired.COMPUTE_COUNTS[key] == 1
def test_cache_key_tracks_model_effects(self, seq_env):
meta = seq_env["meta"]
m1 = paired.SimulatedModel("same-version", {"pedestrian|night": -0.02})
m2 = paired.SimulatedModel("same-version", {"pedestrian|night": -0.03})
assert m1.fingerprint() != m2.fingerprint()
|