188 lines
7.2 KiB
Python
188 lines
7.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def positive_item(slug: str, *, f1: float = 0.6, coverage: bool = True) -> dict:
|
|
matches = 60
|
|
false_positives = 30
|
|
false_negatives = 50
|
|
return {
|
|
"sample_slug": slug,
|
|
"sample_display_name": f"Mol {slug}",
|
|
"municipality": "Mol",
|
|
"operational_zone": slug,
|
|
"recommended_split": "val",
|
|
"model_asset_id": "mol-model",
|
|
"tile_size": 512,
|
|
"tile_overlap": 64,
|
|
"threshold": 0.15,
|
|
"project_id": f"project-{slug}",
|
|
"area_id": f"area-{slug}",
|
|
"analysis_run_id": f"run-{slug}",
|
|
"quality_check_id": f"quality-{slug}",
|
|
"detection_count": 90,
|
|
"precision": 0.6666666667,
|
|
"recall": 0.5454545455,
|
|
"f1_score": f1,
|
|
"mean_iou": 0.65,
|
|
"matches": matches,
|
|
"false_positives": false_positives,
|
|
"false_negatives": false_negatives,
|
|
"coverage_applied": coverage,
|
|
"coverage_mode": "persisted_tile_manifest_union",
|
|
"coverage_tile_count": 9,
|
|
"candidate_raw_count": 90,
|
|
"candidate_evaluated_count": 90,
|
|
"candidate_excluded_outside_count": 0,
|
|
"candidate_clipped_boundary_count": 3,
|
|
"reference_raw_count": 112,
|
|
"reference_evaluated_count": 110,
|
|
"reference_excluded_outside_count": 2,
|
|
"reference_clipped_boundary_count": 4,
|
|
"reference_coverage_ratio": 110 / 112,
|
|
"diagnostic_only": True,
|
|
"strict_matches": matches,
|
|
"envelope_matches": 72,
|
|
"possible_box_to_footprint_mismatch_count": 12,
|
|
"envelope_precision": 0.8,
|
|
"envelope_recall": 0.65,
|
|
"envelope_f1_score": 0.717,
|
|
}
|
|
|
|
|
|
def write_inputs(tmp_path: Path, *, rejected: bool = False) -> tuple[Path, Path, Path]:
|
|
slugs = ["mol_achterbos", "mol_gompel", "mol_donk", "mol_postel"]
|
|
items = [
|
|
positive_item(slug, f1=0.05 if rejected and slug == "mol_postel" else 0.6, coverage=not rejected)
|
|
for slug in slugs
|
|
]
|
|
positive_path = tmp_path / "positive.json"
|
|
positive_path.write_text(json.dumps({"items": items}), encoding="utf-8")
|
|
background_path = tmp_path / "background.json"
|
|
background_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"items": [
|
|
{
|
|
"sample_slug": "postel_bos",
|
|
"background_category": "pure_empty_negative",
|
|
"model_asset_id": "mol-model",
|
|
"tile_size": 512,
|
|
"tile_overlap": 64,
|
|
"threshold": 0.15,
|
|
"project_id": "project-background",
|
|
"area_id": "area-background",
|
|
"analysis_run_id": "run-background",
|
|
"tile_count": 9,
|
|
"detection_count": 2 if rejected else 0,
|
|
}
|
|
]
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
manifest_path = tmp_path / "manifest.json"
|
|
manifest_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"samples": [
|
|
{
|
|
"sample_slug": slug,
|
|
"display_name": f"Mol {slug}",
|
|
"municipality": "Mol",
|
|
"operational_zone": slug,
|
|
"recommended_split": "val",
|
|
}
|
|
for slug in slugs
|
|
]
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return positive_path, background_path, manifest_path
|
|
|
|
|
|
def run_report(tmp_path: Path, *, rejected: bool = False) -> dict:
|
|
positive, background, manifest = write_inputs(tmp_path, rejected=rejected)
|
|
output_dir = tmp_path / "report"
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(ROOT / "scripts" / "build_mol_operational_benchmark_report.py"),
|
|
"--positive-summary",
|
|
str(positive),
|
|
"--background-summary",
|
|
str(background),
|
|
"--manifest-path",
|
|
str(manifest),
|
|
"--output-dir",
|
|
str(output_dir),
|
|
"--base-url",
|
|
"http://example.test",
|
|
],
|
|
cwd=ROOT,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert "Mol operational benchmark report passed" in result.stdout
|
|
assert (output_dir / "mol_operational_benchmark_report.md").is_file()
|
|
return json.loads((output_dir / "mol_operational_benchmark_report.json").read_text(encoding="utf-8"))
|
|
|
|
|
|
def test_mol_benchmark_accepts_coverage_safe_multi_zone_evidence(tmp_path: Path) -> None:
|
|
report = run_report(tmp_path)
|
|
|
|
assert report["status"] == "accepted"
|
|
assert report["recommendation"] == "retain_or_promote_candidate"
|
|
decision = report["recommended_candidate"]
|
|
assert decision["decision"] == "operationally_accepted"
|
|
assert decision["positive_sample_count"] == 4
|
|
assert decision["background_sample_count"] == 1
|
|
assert decision["total_references_raw"] == 448
|
|
assert decision["total_references_evaluated"] == 440
|
|
assert decision["total_box_to_footprint_mismatch_count"] == 48
|
|
assert decision["total_background_detections"] == 0
|
|
assert decision["failed_gates"] == []
|
|
|
|
|
|
def test_mol_benchmark_rejects_missing_coverage_zone_collapse_and_background_pressure(tmp_path: Path) -> None:
|
|
report = run_report(tmp_path, rejected=True)
|
|
|
|
assert report["status"] == "review_required"
|
|
assert report["recommended_candidate"] is None
|
|
decision = report["candidate_decisions"][0]
|
|
assert decision["decision"] == "review_required"
|
|
assert "coverage_provenance" in decision["failed_gates"]
|
|
assert "minimum_zone_f1" in decision["failed_gates"]
|
|
assert "background_false_positive_pressure" in decision["failed_gates"]
|
|
|
|
|
|
def test_mol_benchmark_is_wired_into_existing_operator_pipeline() -> None:
|
|
matrix = (ROOT / "scripts" / "run_detection_quality_matrix.sh").read_text(encoding="utf-8")
|
|
multi = (ROOT / "scripts" / "run_multi_sample_detection_quality_matrix.sh").read_text(encoding="utf-8")
|
|
runner = (ROOT / "scripts" / "run_mol_operational_validation.sh").read_text(encoding="utf-8")
|
|
readiness = (ROOT / "scripts" / "run_readiness_check.sh").read_text(encoding="utf-8")
|
|
dockerfile = (ROOT / "deploy" / "unraid" / "Dockerfile.all-in-one").read_text(encoding="utf-8")
|
|
|
|
assert 'findings.get("coverage")' in matrix
|
|
assert 'findings.get("box_to_footprint_diagnostics")' in matrix
|
|
assert '"reference_coverage_ratio"' in matrix
|
|
assert '"possible_box_to_footprint_mismatch_count"' in matrix
|
|
assert 'enriched["operational_zone"]' in multi
|
|
assert "build_mol_operational_benchmark_report.py" in runner
|
|
assert "MOL_MIN_REFERENCE_COVERAGE" in runner
|
|
assert 'MOL_MIN_REFERENCE_COVERAGE="${MOL_MIN_REFERENCE_COVERAGE:-0.90}"' in runner
|
|
assert 'default=0.90' in (ROOT / "scripts" / "build_mol_operational_benchmark_report.py").read_text(encoding="utf-8")
|
|
assert "py_compile scripts/build_mol_operational_benchmark_report.py" in readiness
|
|
assert "COPY scripts/build_mol_operational_benchmark_report.py" in dockerfile
|
|
assert "fixture_mode" not in runner
|
|
assert "manual-fixture-detector" not in runner
|