285 lines
10 KiB
Python
285 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def _polygon(x: float, y: float, size: float) -> dict:
|
|
return {
|
|
"type": "Polygon",
|
|
"coordinates": [
|
|
[
|
|
[x, y],
|
|
[x + size, y],
|
|
[x + size, y + size],
|
|
[x, y + size],
|
|
[x, y],
|
|
]
|
|
],
|
|
}
|
|
|
|
|
|
def _evidence_feature(role: str, source_id: str, geometry: dict) -> dict:
|
|
return {
|
|
"type": "Feature",
|
|
"id": f"{role}:{source_id}",
|
|
"properties": {
|
|
"qa_evidence_role": role,
|
|
"source_feature_id": source_id,
|
|
"reference_feature_id": source_id,
|
|
},
|
|
"geometry": geometry,
|
|
}
|
|
|
|
|
|
def test_fixed_threshold_portfolio_inputs_select_one_comparable_run_per_aoi(tmp_path: Path) -> None:
|
|
script = ROOT / "scripts" / "build_fixed_threshold_evidence_portfolio_inputs.py"
|
|
assert script.exists()
|
|
|
|
sample_summaries = []
|
|
for slug in ("geel", "mol"):
|
|
summary_path = tmp_path / f"{slug}-quality-summary.json"
|
|
items = [
|
|
{
|
|
"project_id": f"project-{slug}-low",
|
|
"quality_check_id": f"qc-{slug}-low",
|
|
"model_asset_id": "model-a",
|
|
"threshold": 0.05,
|
|
"quality_score": 0.2,
|
|
"f1_score": 0.2,
|
|
},
|
|
{
|
|
"project_id": f"project-{slug}-fixed",
|
|
"quality_check_id": f"qc-{slug}-fixed",
|
|
"model_asset_id": "model-a",
|
|
"threshold": 0.15,
|
|
"quality_score": 0.3,
|
|
"f1_score": 0.3,
|
|
},
|
|
]
|
|
summary_path.write_text(json.dumps({"items": items}), encoding="utf-8")
|
|
sample_summaries.append(
|
|
{"sample_slug": slug, "summary_path": str(summary_path)}
|
|
)
|
|
|
|
multi_summary_path = tmp_path / "multi-sample.json"
|
|
multi_summary_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"sample_count": 2,
|
|
"sample_summaries": sample_summaries,
|
|
"items": [],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
output_dir = tmp_path / "fixed-inputs"
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(script),
|
|
"--multi-sample-summary",
|
|
str(multi_summary_path),
|
|
"--threshold",
|
|
"0.15",
|
|
"--model-asset-id",
|
|
"model-a",
|
|
"--model-sha256",
|
|
"abc123",
|
|
"--output-dir",
|
|
str(output_dir),
|
|
],
|
|
cwd=ROOT,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
manifest_path = output_dir / "calibration-evidence-portfolio-manifest.json"
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
assert manifest["model_asset_id"] == "model-a"
|
|
assert manifest["model_sha256"] == "abc123"
|
|
assert manifest["fixed_threshold"] == 0.15
|
|
assert [sample["sample_slug"] for sample in manifest["samples"]] == ["geel", "mol"]
|
|
for sample in manifest["samples"]:
|
|
filtered = json.loads(Path(sample["summary_path"]).read_text(encoding="utf-8"))
|
|
assert len(filtered["items"]) == 1
|
|
assert filtered["items"][0]["threshold"] == 0.15
|
|
assert filtered["best_by_score"] == filtered["items"][0]
|
|
assert str(manifest_path) in result.stdout
|
|
|
|
|
|
def test_false_negative_audit_finds_persistent_reference_misses(tmp_path: Path) -> None:
|
|
script = ROOT / "scripts" / "audit_detection_false_negative_evidence.py"
|
|
readiness = (ROOT / "scripts" / "run_readiness_check.sh").read_text(encoding="utf-8")
|
|
assert script.exists()
|
|
assert "py_compile scripts/build_fixed_threshold_evidence_portfolio_inputs.py" in readiness
|
|
assert "py_compile scripts/audit_detection_false_negative_evidence.py" in readiness
|
|
|
|
portfolio_paths = []
|
|
for label, features in (
|
|
(
|
|
"active",
|
|
[
|
|
_evidence_feature("false_negative", "persistent-small", _polygon(5.0, 51.2, 0.0001)),
|
|
_evidence_feature("false_negative", "recovered-large", _polygon(5.001, 51.2, 0.0003)),
|
|
_evidence_feature("match_reference", "matched", _polygon(5.002, 51.2, 0.0002)),
|
|
],
|
|
),
|
|
(
|
|
"candidate",
|
|
[
|
|
_evidence_feature("false_negative", "persistent-small", _polygon(5.0, 51.2, 0.0001)),
|
|
_evidence_feature("match_reference", "recovered-large", _polygon(5.001, 51.2, 0.0003)),
|
|
_evidence_feature("match_reference", "matched", _polygon(5.002, 51.2, 0.0002)),
|
|
],
|
|
),
|
|
):
|
|
portfolio_dir = tmp_path / label
|
|
evidence_dir = portfolio_dir / "samples" / "geel" / "evidence"
|
|
evidence_dir.mkdir(parents=True)
|
|
evidence_path = evidence_dir / "calibration_evidence.geojson"
|
|
evidence_path.write_text(
|
|
json.dumps({"type": "FeatureCollection", "features": features}),
|
|
encoding="utf-8",
|
|
)
|
|
portfolio_path = portfolio_dir / "calibration_evidence_portfolio.json"
|
|
portfolio_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"model_asset_id": f"model-{label}",
|
|
"samples": [
|
|
{
|
|
"sample_slug": "geel",
|
|
"evidence_geojson_path": str(evidence_path),
|
|
}
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
portfolio_paths.append((label, portfolio_path))
|
|
|
|
output_dir = tmp_path / "audit"
|
|
subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(script),
|
|
"--portfolio",
|
|
f"active={portfolio_paths[0][1]}",
|
|
"--portfolio",
|
|
f"candidate={portfolio_paths[1][1]}",
|
|
"--output-dir",
|
|
str(output_dir),
|
|
],
|
|
cwd=ROOT,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
report = json.loads(
|
|
(output_dir / "detection_false_negative_audit.json").read_text(encoding="utf-8")
|
|
)
|
|
assert report["portfolio_count"] == 2
|
|
sample = report["samples"][0]
|
|
assert sample["sample_slug"] == "geel"
|
|
assert sample["persistent_false_negative_count"] == 1
|
|
assert sample["persistent_reference_ids"] == ["source:persistent-small"]
|
|
active = next(item for item in sample["portfolios"] if item["label"] == "active")
|
|
candidate = next(item for item in sample["portfolios"] if item["label"] == "candidate")
|
|
assert active["false_negative_count"] == 2
|
|
assert active["matched_reference_count"] == 1
|
|
assert active["false_negative_rate"] == 2 / 3
|
|
assert candidate["false_negative_count"] == 1
|
|
assert candidate["false_negative_rate"] == 1 / 3
|
|
assert active["false_negative_area_m2"]["median"] > 0
|
|
assert sample["persistent_false_negative_area_m2"]["count"] == 1
|
|
assert sample["persistent_false_negative_area_m2"]["median"] > 0
|
|
assert sum(
|
|
bucket["count"] for bucket in sample["persistent_area_buckets"].values()
|
|
) == 1
|
|
assert sum(
|
|
bucket["share"] for bucket in sample["persistent_area_buckets"].values()
|
|
) == 1.0
|
|
persistent_evidence = json.loads(
|
|
(output_dir / "persistent_false_negatives.geojson").read_text(encoding="utf-8")
|
|
)
|
|
assert persistent_evidence["type"] == "FeatureCollection"
|
|
assert len(persistent_evidence["features"]) == 1
|
|
persistent_feature = persistent_evidence["features"][0]
|
|
assert persistent_feature["properties"]["qa_evidence_role"] == "persistent_false_negative"
|
|
assert persistent_feature["properties"]["sample_slug"] == "geel"
|
|
assert persistent_feature["properties"]["persistent_reference_id"] == "source:persistent-small"
|
|
assert persistent_feature["properties"]["area_m2"] > 0
|
|
assert persistent_feature["properties"]["area_bucket"] in sample["persistent_area_buckets"]
|
|
assert report["persistent_evidence_geojson_path"] == str(
|
|
output_dir / "persistent_false_negatives.geojson"
|
|
)
|
|
assert report["recommendations"]
|
|
assert (output_dir / "detection_false_negative_audit.md").is_file()
|
|
|
|
|
|
def test_false_negative_audit_rejects_mismatched_reference_populations(tmp_path: Path) -> None:
|
|
script = ROOT / "scripts" / "audit_detection_false_negative_evidence.py"
|
|
portfolio_args = []
|
|
for label, source_ids in (("active", ("one", "two")), ("candidate", ("one",))):
|
|
portfolio_dir = tmp_path / label
|
|
evidence_dir = portfolio_dir / "samples" / "geel" / "evidence"
|
|
evidence_dir.mkdir(parents=True)
|
|
evidence_path = evidence_dir / "calibration_evidence.geojson"
|
|
evidence_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"type": "FeatureCollection",
|
|
"features": [
|
|
_evidence_feature(
|
|
"false_negative",
|
|
source_id,
|
|
_polygon(5.0 + index * 0.001, 51.2, 0.0001),
|
|
)
|
|
for index, source_id in enumerate(source_ids)
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
portfolio_path = portfolio_dir / "calibration_evidence_portfolio.json"
|
|
portfolio_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"model_asset_id": f"model-{label}",
|
|
"samples": [
|
|
{
|
|
"sample_slug": "geel",
|
|
"evidence_geojson_path": str(evidence_path),
|
|
}
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
portfolio_args.extend(("--portfolio", f"{label}={portfolio_path}"))
|
|
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(script),
|
|
*portfolio_args,
|
|
"--output-dir",
|
|
str(tmp_path / "audit"),
|
|
],
|
|
cwd=ROOT,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
assert result.returncode != 0
|
|
assert "different reference populations" in result.stderr
|