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()