Spaces:
Paused
Paused
| """ODD taxonomy + combinatorial coverage math (exact expectations on a | |
| hand-built cube) + run-level coverage and gap filling.""" | |
| from __future__ import annotations | |
| import math | |
| import pandas as pd | |
| import pytest | |
| from sensorflow.megaeval.cube import STAT_COLS | |
| from sensorflow.megaeval.population import DIM_NAMES, DIMENSIONS | |
| from sensorflow.safety import odd | |
| def test_taxonomy_structure(): | |
| tax = odd.taxonomy() | |
| dims = tax["dimensions"] | |
| for name in DIM_NAMES: | |
| assert dims[name]["values"] == DIMENSIONS[name] | |
| assert dims[name]["instrumented"] is True | |
| assert dims[name]["iso34503_category"] in ( | |
| "scenery", "environmental_conditions", "dynamic_elements", | |
| "sensor_configuration") | |
| # declared-but-not-instrumented dimension is honest about itself | |
| assert dims["geography"]["instrumented"] is False | |
| assert "ISO 34503" in tax["standard_basis"] | |
| def _mini_cube() -> pd.DataFrame: | |
| """Cube with data in exactly two weather cells: | |
| clear: tp=90 fn=10 (n=100), rain: tp=8 fn=2 (n=10). fog/snow empty.""" | |
| rows = [] | |
| for weather_code, tp, fn in ((0, 90, 10), (1, 8, 2)): | |
| row = {d: 0 for d in DIM_NAMES} | |
| row["weather"] = weather_code | |
| row.update({c: 0 for c in STAT_COLS}) | |
| row.update({"n": tp + fn, "tp": tp, "fn": fn, "sum_iou": 0.8 * tp}) | |
| rows.append(row) | |
| return pd.DataFrame(rows) | |
| def _wilson(k, n, z=1.959963984540054): | |
| if n == 0: | |
| return 0.0, 1.0 | |
| p = k / n | |
| denom = 1 + z ** 2 / n | |
| center = (p + z ** 2 / (2 * n)) / denom | |
| margin = z * math.sqrt(p * (1 - p) / n + z ** 2 / (4 * n ** 2)) / denom | |
| return max(0.0, center - margin), min(1.0, center + margin) | |
| def test_coverage_exact_math_on_known_population(): | |
| dim_counts = {"weather": {"clear": 700, "rain": 200, "fog": 80, "snow": 20}} | |
| cov = odd.compute_coverage(_mini_cube(), dims=["weather"], dim_counts=dim_counts, | |
| min_samples=20, max_ci_width=0.25, target_recall=0.9) | |
| cells = {c["cell"]["weather"]: c for c in cov["cells"]} | |
| assert set(cells) == {"clear", "rain", "fog", "snow"} | |
| # clear: n=100, recall 0.9, Wilson CI hand-checked, adequate | |
| clear = cells["clear"] | |
| lo, hi = _wilson(90, 100) | |
| assert clear["n"] == 100 | |
| assert clear["recall"] == pytest.approx(0.9) | |
| assert clear["wilson_ci"][0] == pytest.approx(lo, abs=2e-3) | |
| assert clear["wilson_ci"][1] == pytest.approx(hi, abs=2e-3) | |
| assert clear["adequate"] and not clear["is_gap"] | |
| assert clear["production_share"] == pytest.approx(0.7) | |
| # risk = share * max(0, target - wilson_lo) | |
| assert clear["risk"] == pytest.approx(0.7 * max(0.0, 0.9 - lo), abs=2e-3) | |
| # rain: n=10 < 20 -> insufficient samples gap | |
| rain = cells["rain"] | |
| assert rain["is_gap"] and "insufficient_samples" in rain["gap_reasons"] | |
| assert rain["recall"] == pytest.approx(0.8) | |
| # empty cells: both gap reasons, CI width 1 | |
| fog = cells["fog"] | |
| assert fog["n"] == 0 and fog["recall"] is None | |
| assert set(fog["gap_reasons"]) == {"insufficient_samples", "ci_too_wide"} | |
| assert fog["ci_width"] == pytest.approx(1.0) | |
| # summary: exactly 1 adequate of 4 cells; weighted coverage = clear share | |
| s = cov["summary"] | |
| assert s["total_cells"] == 4 | |
| assert s["adequate_cells"] == 1 and s["gap_cells"] == 3 | |
| assert s["coverage_rate"] == pytest.approx(0.25) | |
| assert s["production_weighted_coverage"] == pytest.approx(0.7) | |
| # gaps ranked by risk = share * deficit: rain(share .2) > fog(.08) > snow(.02) | |
| gap_order = [g["cell"]["weather"] for g in cov["gaps"]] | |
| assert gap_order == ["rain", "fog", "snow"] | |
| for g in cov["gaps"]: | |
| assert g["risk"] == pytest.approx( | |
| g["production_share"] * g["performance_deficit"], abs=1e-4) | |
| # every gap gets a fulfillable request | |
| assert len(cov["fill_requests"]) == 3 | |
| req = cov["fill_requests"][0] | |
| assert req["cell"]["weather"] == "rain" | |
| assert req["needed_samples"] == 10 | |
| assert req["generator"] == "sensorflow.evaluation.synthetic.generate_dataset" | |
| def test_coverage_rejects_oversized_cell_product(): | |
| with pytest.raises(ValueError): | |
| odd.compute_coverage(_mini_cube(), dims=DIM_NAMES) # way over 2000 cells | |
| def test_coverage_for_run_consistency(mega_env, safety_root): | |
| cov = odd.coverage_for_run(mega_env["store"], mega_env["good"], | |
| dims=["weather", "lighting"], include_cells=True) | |
| s = cov["summary"] | |
| assert s["total_cells"] == len(DIMENSIONS["weather"]) * len(DIMENSIONS["lighting"]) | |
| assert s["populated_cells"] + s["empty_cells"] == s["total_cells"] | |
| assert s["adequate_cells"] + s["gap_cells"] == s["total_cells"] | |
| # cell mass equals the population's object count (every object in one cell) | |
| assert sum(c["n"] for c in cov["cells"]) == mega_env["meta"]["num_objects"] | |
| # adequacy consistent with the thresholds echoed in the response | |
| th = cov["thresholds"] | |
| for c in cov["cells"]: | |
| expected = c["n"] >= th["min_samples"] and c["ci_width"] <= th["max_ci_width"] | |
| assert c["adequate"] == expected | |
| # gaps sorted by descending risk | |
| risks = [g["risk"] for g in cov["gaps"]] | |
| assert risks == sorted(risks, reverse=True) | |
| def test_fill_gap_generates_data_and_improves_cell(mega_env, fresh_safety_root, tmp_path): | |
| from sensorflow.evaluation.records import reset_store | |
| reset_store(tmp_path / "eval") | |
| store, run = mega_env["store"], mega_env["good"] | |
| cov = odd.coverage_for_run(store, run, dims=["weather", "lighting"], | |
| include_cells=True, min_samples=10_000) | |
| gap = cov["gaps"][0] # min_samples=10k guarantees gaps exist | |
| result = odd.fill_gap(store, run, gap["cell"], num_sequences=1, | |
| frames_per_sequence=10) | |
| assert result["objects_added"] > 0 | |
| assert result["cell_after"]["n"] > result["cell_before"]["n"] | |
| assert result["cell_after"]["n"] == (result["cell_before"]["n"] | |
| + result["objects_added"]) | |
| assert result["supplement"]["source"] == "synthetic" | |
| # supplement persisted and folded into subsequent coverage computations | |
| supplements = odd.load_supplements(run.run_id) | |
| assert len(supplements) == 1 | |
| cov2 = odd.coverage_for_run(store, run, dims=["weather", "lighting"], | |
| include_cells=True, min_samples=10_000) | |
| cell2 = next(c for c in cov2["cells"] if c["cell_id"] == gap["cell_id"]) | |
| assert cell2["n"] == result["cell_after"]["n"] | |
| assert cell2["synthetic_supplement"]["n_added"] == result["objects_added"] | |
| # generated frames carry the requested conditions | |
| from sensorflow.evaluation.records import get_store | |
| legacy = get_store() | |
| frames = legacy.where("frames", dataset_id=result["generated_dataset_id"]) | |
| assert frames | |
| weather = gap["cell"].get("weather") | |
| if weather: | |
| assert all(f.weather == weather for f in frames) | |
| # scenario DB registered the synthetic gap-fill scenario | |
| from sensorflow.safety.scenario_db import get_db | |
| recs = get_db().search(source="synthetic") | |
| assert any(r.scenario_type == "odd_gap_fill" for r in recs) | |
| def test_fill_gap_rejects_unknown_cell(mega_env, fresh_safety_root): | |
| with pytest.raises(ValueError): | |
| odd.fill_gap(mega_env["store"], mega_env["good"], {"weather": "hurricane"}) | |