Harden detection result review

2026-07-13 13:31:43 +02:00
parent 6fe1bbd946
commit 4455e242c6
12 changed files with 795 additions and 22 deletions
@@ -793,6 +793,24 @@ used only when source IDs are absent. Invalid or missing geometry fails the
 audit instead of being silently skipped. The tools do not run inference,
 create QA records, mutate model defaults or download data/models.

+Audit the false-positive review load of one persisted evidence portfolio before
+turning detections into hard-negative training input:
+
+```bash
+python scripts/audit_detection_false_positive_evidence.py \
+  --portfolio artifacts/model-review/small-building-candidate/evidence-portfolio/calibration_evidence_portfolio.json \
+  --output-dir artifacts/model-review/small-building-candidate/false-positive-audit
+```
+
+The command validates the portfolio role counts against each persisted evidence
+GeoJSON, rejects invalid/non-polygon geometry, computes WGS84 geodesic area and
+size buckets, and reports false-positive pressure per AOI, class and
+AOI-qualified source tile. It writes `detection_false_positive_audit.json`, a
+Markdown handoff and combined `false_positives.geojson` for map review. Original
+evidence properties and geometry are preserved. Confidence statistics are only
+computed when confidence is actually present in persisted evidence; missing
+coverage is reported explicitly and never inferred from the run threshold.
+
 Docker images install only the GIS runtime by default. To build a local/Tower
 image with PyTorch/Ultralytics available for the configured-YOLO preflight and
 runtime path, set:
@@ -0,0 +1,361 @@
+from __future__ import annotations
+
+import argparse
+import json
+import math
+from collections import Counter
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from audit_detection_false_negative_evidence import (
+    AREA_BUCKETS,
+    area_bucket,
+    area_stats,
+    load_json,
+    resolve_evidence_path,
+)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Audit persisted GeoIntel false-positive QA evidence for one model portfolio."
+    )
+    parser.add_argument(
+        "--portfolio",
+        required=True,
+        type=Path,
+        help="Path to calibration_evidence_portfolio.json.",
+    )
+    parser.add_argument("--output-dir", required=True, type=Path)
+    return parser.parse_args()
+
+
+def _confidence(properties: dict[str, Any], feature_id: Any) -> float | None:
+    for key in ("candidate_confidence", "confidence"):
+        value = properties.get(key)
+        if value is None:
+            continue
+        if isinstance(value, bool) or not isinstance(value, (int, float)):
+            raise SystemExit(f"Evidence feature {feature_id} has non-numeric {key}")
+        numeric = float(value)
+        if not math.isfinite(numeric) or not 0.0 <= numeric <= 1.0:
+            raise SystemExit(f"Evidence feature {feature_id} has invalid {key}: {value}")
+        return numeric
+    return None
+
+
+def _source_tile(properties: dict[str, Any]) -> str:
+    source_tile_path = properties.get("source_tile_path")
+    if source_tile_path not in (None, ""):
+        return str(source_tile_path)
+    tile_index = properties.get("tile_index")
+    if tile_index not in (None, ""):
+        return str(tile_index)
+    return "unknown"
+
+
+def _feature_class(properties: dict[str, Any]) -> str:
+    value = properties.get("feature_class") or properties.get("class_name")
+    return str(value) if value not in (None, "") else "unknown"
+
+
+def _bucket_summary(areas: list[float]) -> dict[str, dict[str, float | int]]:
+    counts = Counter(area_bucket(area) for area in areas)
+    total = len(areas)
+    return {
+        label: {
+            "count": counts[label],
+            "share": counts[label] / total if total else 0.0,
+        }
+        for label, _, _ in AREA_BUCKETS
+    }
+
+
+def audit_sample(
+    payload: dict[str, Any],
+    *,
+    sample_slug: str,
+    geod: Any,
+    shape: Any,
+) -> tuple[dict[str, Any], list[dict[str, Any]], list[float], list[float]]:
+    if payload.get("type") != "FeatureCollection":
+        raise SystemExit(f"Evidence for {sample_slug} must be a FeatureCollection")
+    features = payload.get("features")
+    if not isinstance(features, list):
+        raise SystemExit(f"Evidence for {sample_slug} must contain a features list")
+
+    false_positive_features: list[dict[str, Any]] = []
+    areas: list[float] = []
+    confidences: list[float] = []
+    role_counts: Counter[str] = Counter()
+    source_tile_counts: Counter[str] = Counter()
+    class_counts: Counter[str] = Counter()
+
+    for feature in features:
+        if not isinstance(feature, dict):
+            raise SystemExit(f"Evidence for {sample_slug} contains a non-object feature")
+        properties = feature.get("properties") or {}
+        if not isinstance(properties, dict):
+            raise SystemExit(f"Evidence feature {feature.get('id')} has invalid properties")
+        role = str(properties.get("qa_evidence_role") or "")
+        role_counts[role] += 1
+        if role != "false_positive":
+            continue
+
+        geometry_payload = feature.get("geometry")
+        if not isinstance(geometry_payload, dict):
+            raise SystemExit(f"Evidence feature {feature.get('id')} has no geometry")
+        geometry = shape(geometry_payload)
+        if geometry.is_empty or not geometry.is_valid:
+            raise SystemExit(f"Evidence feature {feature.get('id')} has invalid geometry")
+        if geometry.geom_type not in {"Polygon", "MultiPolygon"}:
+            raise SystemExit(
+                f"Evidence feature {feature.get('id')} must be Polygon or MultiPolygon"
+            )
+
+        area_m2 = abs(float(geod.geometry_area_perimeter(geometry)[0]))
+        confidence = _confidence(properties, feature.get("id"))
+        source_tile = _source_tile(properties)
+        feature_class = _feature_class(properties)
+        augmented_properties = dict(properties)
+        augmented_properties.update(
+            {
+                "sample_slug": sample_slug,
+                "area_m2": area_m2,
+                "area_bucket": area_bucket(area_m2),
+                "review_source_tile": source_tile,
+                "review_confidence_available": confidence is not None,
+            }
+        )
+        false_positive_features.append(
+            {
+                "type": "Feature",
+                "id": f"{sample_slug}:{feature.get('id')}",
+                "properties": augmented_properties,
+                "geometry": geometry_payload,
+            }
+        )
+        areas.append(area_m2)
+        if confidence is not None:
+            confidences.append(confidence)
+        source_tile_counts[source_tile] += 1
+        class_counts[feature_class] += 1
+
+    false_positive_count = role_counts["false_positive"]
+    matched_candidate_count = role_counts["match_candidate"]
+    candidate_count = false_positive_count + matched_candidate_count
+    report = {
+        "sample_slug": sample_slug,
+        "false_positive_count": false_positive_count,
+        "matched_candidate_count": matched_candidate_count,
+        "candidate_count": candidate_count,
+        "false_positive_rate": false_positive_count / candidate_count if candidate_count else None,
+        "false_positive_area_m2": area_stats(areas),
+        "area_buckets": _bucket_summary(areas),
+        "confidence_coverage_count": len(confidences),
+        "confidence": area_stats(confidences),
+        "source_tile_counts": dict(sorted(source_tile_counts.items())),
+        "class_counts": dict(sorted(class_counts.items())),
+    }
+    return report, false_positive_features, areas, confidences
+
+
+def _merge_counts(rows: list[dict[str, Any]], key: str) -> dict[str, int]:
+    counts: Counter[str] = Counter()
+    for row in rows:
+        counts.update(row[key])
+    return dict(sorted(counts.items()))
+
+
+def _merge_source_tile_counts(rows: list[dict[str, Any]]) -> dict[str, int]:
+    counts: Counter[str] = Counter()
+    for row in rows:
+        counts.update(
+            {
+                f"{row['sample_slug']}:{source_tile}": count
+                for source_tile, count in row["source_tile_counts"].items()
+            }
+        )
+    return dict(sorted(counts.items()))
+
+
+def build_recommendations(report: dict[str, Any]) -> list[str]:
+    samples = report["samples"]
+    recommendations: list[str] = []
+    ranked = sorted(
+        samples,
+        key=lambda sample: (
+            sample["false_positive_count"],
+            sample["false_positive_rate"] or 0.0,
+        ),
+        reverse=True,
+    )
+    if ranked:
+        recommendations.append(
+            "Review the highest false-positive volumes first: "
+            + ", ".join(
+                f"{sample['sample_slug']} ({sample['false_positive_count']})"
+                for sample in ranked[:3]
+            )
+            + "."
+        )
+
+    area_buckets = report["area_buckets"]
+    small_count = sum(
+        area_buckets[key]["count"]
+        for key in ("tiny_lt_25_m2", "small_25_100_m2")
+    )
+    if report["false_positive_count"]:
+        small_share = small_count / report["false_positive_count"]
+        recommendations.append(
+            f"Inspect tiny/small candidate geometry first when sampling hard negatives; it represents {small_share:.1%} of persisted false positives."
+        )
+
+    if report["confidence_coverage_count"] == 0:
+        recommendations.append(
+            "Persisted QA evidence has no per-detection confidence values; use the recorded run threshold and geometry/tile evidence without inventing confidence bands."
+        )
+    elif report["confidence_coverage_count"] < report["false_positive_count"]:
+        recommendations.append(
+            "Confidence coverage is partial; do not infer a portfolio-wide confidence distribution from the covered subset."
+        )
+
+    tile_counts = report["source_tile_counts"]
+    if tile_counts:
+        top_tile, top_count = max(tile_counts.items(), key=lambda item: item[1])
+        recommendations.append(
+            f"Start spatial review with source tile {top_tile} ({top_count} false positives), then confirm examples visually before adding any hard-negative labels."
+        )
+    return recommendations
+
+
+def run_audit(portfolio_path: Path, output_dir: Path) -> tuple[Path, Path]:
+    try:
+        from pyproj import Geod
+        from shapely.geometry import shape
+    except ImportError as exc:
+        raise SystemExit(
+            "False-positive GIS audit requires the existing GeoIntel GIS extras (pyproj and shapely)."
+        ) from exc
+
+    portfolio_path = portfolio_path.expanduser().resolve()
+    portfolio = load_json(portfolio_path)
+    samples = portfolio.get("samples")
+    if not isinstance(samples, list) or not samples:
+        raise SystemExit(f"Evidence portfolio has no samples: {portfolio_path}")
+
+    geod = Geod(ellps="WGS84")
+    sample_reports: list[dict[str, Any]] = []
+    review_features: list[dict[str, Any]] = []
+    all_areas: list[float] = []
+    all_confidences: list[float] = []
+    seen_slugs: set[str] = set()
+
+    for sample in samples:
+        if not isinstance(sample, dict):
+            raise SystemExit("Evidence portfolio contains a non-object sample")
+        sample_slug = str(sample.get("sample_slug") or "").strip().lower()
+        if not sample_slug or sample_slug in seen_slugs:
+            raise SystemExit(f"Evidence portfolio has an invalid or duplicate sample_slug: {sample_slug}")
+        seen_slugs.add(sample_slug)
+        evidence_path = resolve_evidence_path(portfolio_path, sample)
+        sample_report, sample_features, areas, confidences = audit_sample(
+            load_json(evidence_path),
+            sample_slug=sample_slug,
+            geod=geod,
+            shape=shape,
+        )
+        declared = (sample.get("role_counts") or {}).get("false_positive")
+        if declared is not None and declared != sample_report["false_positive_count"]:
+            raise SystemExit(
+                f"Sample {sample_slug} declares {declared} false positives but evidence contains {sample_report['false_positive_count']}"
+            )
+        sample_report["evidence_geojson_path"] = str(evidence_path)
+        sample_reports.append(sample_report)
+        review_features.extend(sample_features)
+        all_areas.extend(areas)
+        all_confidences.extend(confidences)
+
+    false_positive_count = sum(row["false_positive_count"] for row in sample_reports)
+    candidate_count = sum(row["candidate_count"] for row in sample_reports)
+    output_dir = output_dir.expanduser().resolve()
+    output_dir.mkdir(parents=True, exist_ok=True)
+    geojson_path = output_dir / "false_positives.geojson"
+    geojson_path.write_text(
+        json.dumps(
+            {"type": "FeatureCollection", "features": review_features},
+            indent=2,
+            sort_keys=True,
+        ),
+        encoding="utf-8",
+    )
+
+    report = {
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "schema_version": 1,
+        "portfolio_path": str(portfolio_path),
+        "model_asset_id": portfolio.get("model_asset_id"),
+        "model_sha256": portfolio.get("model_sha256"),
+        "input_crs": "EPSG:4326",
+        "area_method": "WGS84 geodesic area via pyproj.Geod",
+        "sample_count": len(sample_reports),
+        "false_positive_count": false_positive_count,
+        "candidate_count": candidate_count,
+        "false_positive_rate": false_positive_count / candidate_count if candidate_count else None,
+        "false_positive_area_m2": area_stats(all_areas),
+        "area_buckets": _bucket_summary(all_areas),
+        "confidence_coverage_count": len(all_confidences),
+        "confidence": area_stats(all_confidences),
+        "source_tile_counts": _merge_source_tile_counts(sample_reports),
+        "class_counts": _merge_counts(sample_reports, "class_counts"),
+        "samples": sample_reports,
+        "false_positive_geojson_path": str(geojson_path),
+    }
+    report["recommendations"] = build_recommendations(report)
+
+    json_path = output_dir / "detection_false_positive_audit.json"
+    json_path.write_text(json.dumps(report, indent=2, sort_keys=True), encoding="utf-8")
+
+    lines = [
+        "# Detection false-positive evidence audit",
+        "",
+        f"- Generated: {report['generated_at']}",
+        f"- Model asset: {report['model_asset_id'] or 'not recorded'}",
+        f"- Model SHA256: {report['model_sha256'] or 'not recorded'}",
+        f"- Persisted false positives: {false_positive_count}",
+        f"- Candidate detections represented: {candidate_count}",
+        f"- Confidence coverage: {len(all_confidences)}/{false_positive_count}",
+        f"- Area method: {report['area_method']}",
+        "",
+        "## AOI review pressure",
+        "",
+        "| AOI | False positives | Candidate detections | FP rate | Median area m2 | Confidence coverage |",
+        "|---|---:|---:|---:|---:|---:|",
+    ]
+    for sample in sample_reports:
+        rate = sample["false_positive_rate"]
+        rate_text = f"{rate:.3f}" if rate is not None else "n/a"
+        median_area = sample["false_positive_area_m2"]["median"]
+        median_text = f"{median_area:.1f}" if median_area is not None else "n/a"
+        lines.append(
+            f"| {sample['sample_slug']} | {sample['false_positive_count']} | "
+            f"{sample['candidate_count']} | {rate_text} | {median_text} | "
+            f"{sample['confidence_coverage_count']}/{sample['false_positive_count']} |"
+        )
+    lines.extend(["", "## Recommended review actions", ""])
+    lines.extend(f"- {item}" for item in report["recommendations"])
+    markdown_path = output_dir / "detection_false_positive_audit.md"
+    markdown_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+    return json_path, markdown_path
+
+
+def main() -> None:
+    args = parse_args()
+    json_path, markdown_path = run_audit(args.portfolio, args.output_dir)
+    print(f"False-positive audit JSON: {json_path}")
+    print(f"False-positive audit Markdown: {markdown_path}")
+
+
+if __name__ == "__main__":
+    main()
@@ -49,6 +49,7 @@ ${PYTHON_BIN} -m py_compile scripts/build_detection_model_promotion_report.py
 ${PYTHON_BIN} -m py_compile scripts/build_background_corpus_split_report.py
 ${PYTHON_BIN} -m py_compile scripts/build_fixed_threshold_evidence_portfolio_inputs.py
 ${PYTHON_BIN} -m py_compile scripts/audit_detection_false_negative_evidence.py
+${PYTHON_BIN} -m py_compile scripts/audit_detection_false_positive_evidence.py
 ${PYTHON_BIN} -m py_compile scripts/activate_promoted_yolo_candidate.py
 ${PYTHON_BIN} -m py_compile scripts/cleanup_demo_artifacts.py
 ${PYTHON_BIN} -m py_compile backend/scripts/cleanup_demo_artifacts.py