Spaces:
Paused
Paused
File size: 5,052 Bytes
a2bbd94 | 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 | """Cube correctness: cube aggregates must EXACTLY match brute-force computation
over the raw per-object records, and re-running with the same lineage must be
bit-identical (deterministic seeds)."""
from __future__ import annotations
import os
import numpy as np
import pandas as pd
import pytest
from sensorflow.megaeval import cube as cube_mod
from sensorflow.megaeval import population as pop_mod
from sensorflow.megaeval.population import DIM_NAMES, DIMENSIONS
from sensorflow.megaeval.runs import run_dir
def _brute_force(env, group_dims):
"""Recompute tp/fp/fn/n per group directly from raw per-object outputs."""
run = env["good"]
frames = []
d = run_dir(run.run_id)
for p in range(run.partitions_total):
pop_cols = pop_mod.load_partition(run.population_id, p)
with np.load(os.path.join(d, f"objects-part-{p:04d}.npz")) as z:
df = pd.DataFrame({
**{k: pop_cols[k] for k in DIM_NAMES},
"tp": z["tp"].astype(np.int64),
"detected": z["detected"].astype(bool),
})
df["fn"] = 1 - df["tp"]
df["fp"] = (df["detected"] & (df["tp"] == 0)).astype(np.int64)
df["n"] = 1
frames.append(df[group_dims + ["n", "tp", "fp", "fn"]])
with np.load(os.path.join(d, f"fp-part-{p:04d}.npz")) as z:
if z["container_id"].size:
fpdf = pd.DataFrame({k: z[k] for k in DIM_NAMES})
fpdf["n"] = 0
fpdf["tp"] = 0
fpdf["fp"] = 1
fpdf["fn"] = 0
frames.append(fpdf[group_dims + ["n", "tp", "fp", "fn"]])
full = pd.concat(frames, ignore_index=True)
return full.groupby(group_dims, as_index=False).sum()
@pytest.mark.parametrize("group_dims", [["class"], ["class", "lighting"], ["scenario"]])
def test_cube_matches_brute_force_exactly(mega_env, group_dims):
run = mega_env["good"]
cube = mega_env["store"].artifacts(run.run_id)["cube"]
rows, _ = cube_mod.aggregate(cube, None, group_dims,
["n", "tp", "fp", "fn", "precision", "recall"], 2000)
brute = _brute_force(mega_env, group_dims)
assert len(rows) == len(brute)
for _, b in brute.iterrows():
key = {d: DIMENSIONS[d][int(b[d])] for d in group_dims}
match = [r for r in rows if all(r[d] == key[d] for d in group_dims)]
assert len(match) == 1, f"missing cube row for {key}"
r = match[0]
assert r["n"] == int(b["n"])
assert r["tp"] == int(b["tp"])
assert r["fp"] == int(b["fp"])
assert r["fn"] == int(b["fn"])
# derived ratios match brute force (API rounds ratios to 6 decimals)
if b["tp"] + b["fp"] > 0:
assert r["precision"] == pytest.approx(b["tp"] / (b["tp"] + b["fp"]), abs=1e-6)
if b["tp"] + b["fn"] > 0:
assert r["recall"] == pytest.approx(b["tp"] / (b["tp"] + b["fn"]), abs=1e-6)
def test_rollup_consistency(mega_env):
"""Any grouped rollup must sum to the global totals (no double counting)."""
run = mega_env["good"]
cube = mega_env["store"].artifacts(run.run_id)["cube"]
total_rows, _ = cube_mod.aggregate(cube, None, None, ["n", "tp", "fp", "fn"], 1)
total = total_rows[0]
for dim in ("class", "weather", "occlusion", "distance_band"):
rows, _ = cube_mod.aggregate(cube, None, [dim], ["n", "tp", "fp", "fn"], 2000)
for stat in ("n", "tp", "fp", "fn"):
assert sum(r[stat] for r in rows) == total[stat], f"{dim}/{stat} rollup mismatch"
def test_filtered_aggregate_equals_manual_subset(mega_env):
run = mega_env["good"]
cube = mega_env["store"].artifacts(run.run_id)["cube"]
filt = {"class": ["pedestrian"], "lighting": ["night"]}
rows, _ = cube_mod.aggregate(cube, filt, None, ["n", "tp", "fn", "recall"], 1)
sub = cube[(cube["class"] == DIMENSIONS["class"].index("pedestrian"))
& (cube["lighting"] == DIMENSIONS["lighting"].index("night"))]
assert rows[0]["n"] == int(sub["n"].sum())
assert rows[0]["tp"] == int(sub["tp"].sum())
assert rows[0]["recall"] == pytest.approx(sub["tp"].sum() / max(sub["tp"].sum() + sub["fn"].sum(), 1), abs=1e-6)
def test_reproducible_run_same_lineage(mega_env):
"""Same population + model_version (same derived seed) -> identical cube."""
store, meta = mega_env["store"], mega_env["meta"]
r1 = store.create_run(population_id=meta["population_id"],
model_version="model-repro", worker_delay_s=0.0)
store.execute_sync(r1)
r2 = store.create_run(population_id=meta["population_id"],
model_version="model-repro", worker_delay_s=0.0)
store.execute_sync(r2)
assert r1.seed == r2.seed
c1 = mega_env["store"].artifacts(r1.run_id)["cube"].sort_values(DIM_NAMES).reset_index(drop=True)
c2 = mega_env["store"].artifacts(r2.run_id)["cube"].sort_values(DIM_NAMES).reset_index(drop=True)
pd.testing.assert_frame_equal(c1, c2)
assert r1.headline == {**r2.headline}
|