#!/usr/bin/env python3 """Render persisted detection QA evidence for explicit operator review. The script is read-only. It never infers whether a QA false-positive is a model error and never mutates application persistence or source imagery. """ from __future__ import annotations import argparse import csv import hashlib import json import math from collections import defaultdict from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable JSON_NAME = "detection_false_positive_review_summary.json" MARKDOWN_NAME = "detection_false_positive_review.md" DECISIONS_NAME = "false_positive_review_decisions.csv" CONFIDENCE_BANDS = ( ("low_lt_0_30", 0.0, 0.30), ("mid_0_30_0_60", 0.30, 0.60), ("high_gte_0_60", 0.60, math.inf), ) DECISION_FIELDS = ( "candidate_feature_id", "evidence_feature_id", "sample_slug", "confidence", "area_m2", "area_bucket", "confidence_band", "analysis_run_id", "quality_check_id", "source_tile_path", "review_decision", "review_notes", ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Render persisted detection false-positive evidence for manual review." ) parser.add_argument("--portfolio", required=True, type=Path) parser.add_argument("--storage-root", default="/app/storage", type=Path) parser.add_argument("--output-dir", required=True, type=Path) parser.add_argument( "--sample-slugs", default="", help="Optional comma-separated AOI slugs. All portfolio samples are used by default.", ) parser.add_argument("--max-features", type=int, default=48) parser.add_argument("--columns", type=int, default=4) parser.add_argument("--cards-per-sheet", type=int, default=16) parser.add_argument("--thumb-size", type=int, default=256) return parser.parse_args() def load_json(path: Path) -> dict[str, Any]: if not path.is_file(): raise SystemExit(f"JSON input is not readable: {path}") payload = json.loads(path.read_text(encoding="utf-8-sig")) if not isinstance(payload, dict): raise SystemExit(f"JSON input must be an object: {path}") return payload def resolve_evidence_path(portfolio_path: Path, sample: dict[str, Any]) -> Path: raw = str(sample.get("evidence_geojson_path") or "").strip() configured = Path(raw).expanduser() sample_slug = str(sample.get("sample_slug") or "").strip().lower() candidates = [configured] if raw and not configured.is_absolute(): candidates.append(portfolio_path.parent / configured) candidates.append( portfolio_path.parent / "samples" / sample_slug / "evidence" / "calibration_evidence.geojson" ) for candidate in candidates: if candidate.is_file(): return candidate.resolve() raise SystemExit(f"Evidence GeoJSON is not readable for {sample_slug}: {raw}") def confidence_band(confidence: float) -> str: for label, minimum, maximum in CONFIDENCE_BANDS: if minimum <= confidence < maximum: return label raise SystemExit(f"Detection confidence is outside [0, 1]: {confidence}") def stable_sort_key(record: dict[str, Any]) -> str: identity = f"{record['sample_slug']}:{record['candidate_feature_id']}" return hashlib.sha256(identity.encode("utf-8")).hexdigest() def stratified_selection( records: list[dict[str, Any]], limit: int ) -> list[dict[str, Any]]: grouped: dict[tuple[str, str, str], list[dict[str, Any]]] = defaultdict(list) for record in records: grouped[ ( record["sample_slug"], record["area_bucket"], record["confidence_band"], ) ].append(record) for values in grouped.values(): values.sort(key=stable_sort_key) selected: list[dict[str, Any]] = [] keys = sorted(grouped) depth = 0 while len(selected) < limit: added = False for key in keys: values = grouped[key] if depth < len(values): selected.append(values[depth]) added = True if len(selected) == limit: break if not added: break depth += 1 return selected def require_dependencies() -> dict[str, Any]: try: import numpy import rasterio from PIL import Image, ImageDraw, ImageFont from pyproj import Geod, Transformer from shapely.geometry import Polygon, box, shape from shapely.ops import transform except ImportError as exc: raise SystemExit( "False-positive visual review requires the backend GIS/raster extras" ) from exc return { "numpy": numpy, "rasterio": rasterio, "Image": Image, "ImageDraw": ImageDraw, "ImageFont": ImageFont, "Geod": Geod, "Transformer": Transformer, "box": box, "Polygon": Polygon, "shape": shape, "transform": transform, } def resolve_source_tile(raw: str, storage_root: Path) -> Path: storage_root = storage_root.expanduser().resolve() candidate = Path(raw).expanduser() if not candidate.is_absolute(): candidate = storage_root / candidate candidate = candidate.resolve() try: candidate.relative_to(storage_root) except ValueError as exc: raise SystemExit(f"Detection source tile is outside storage root: {candidate}") from exc if not candidate.is_file(): raise SystemExit(f"Detection source tile is not readable: {candidate}") return candidate def normalize_raster(data: Any, numpy: Any) -> Any: if data.shape[0] == 1: data = numpy.repeat(data, 3, axis=0) elif data.shape[0] >= 3: data = data[:3] else: data = numpy.vstack([data, data[-1:]])[:3] if data.dtype == numpy.uint8: return numpy.moveaxis(data, 0, 2) output = numpy.zeros(data.shape, dtype=numpy.uint8) for index, band in enumerate(data): finite = band[numpy.isfinite(band)] if not finite.size: continue low, high = numpy.percentile(finite, (2, 98)) if high <= low: high = low + 1 output[index] = numpy.clip((band - low) * 255 / (high - low), 0, 255) return numpy.moveaxis(output, 0, 2) def polygon_rings(geometry: Any) -> Iterable[Any]: if geometry.geom_type == "Polygon": yield geometry.exterior yield from geometry.interiors elif geometry.geom_type == "MultiPolygon": for polygon in geometry.geoms: yield polygon.exterior yield from polygon.interiors def draw_geometry( draw: Any, geometry: Any, inverse_transform: Any, scale: float, offset_x: float, offset_y: float, crop_left: float, crop_top: float, color: tuple[int, int, int], ) -> None: for ring in polygon_rings(geometry): points = [] for x, y in ring.coords: column, row = inverse_transform * (x, y) points.append( ( (column - crop_left) * scale + offset_x, (row - crop_top) * scale + offset_y, ) ) if len(points) >= 2: draw.line(points, fill=color, width=2, joint="curve") def render_card( record: dict[str, Any], references: list[dict[str, Any]], thumb_size: int, dependencies: dict[str, Any], ) -> tuple[Any, int]: rasterio = dependencies["rasterio"] Image = dependencies["Image"] ImageDraw = dependencies["ImageDraw"] ImageFont = dependencies["ImageFont"] numpy = dependencies["numpy"] Transformer = dependencies["Transformer"] shape = dependencies["shape"] transform_geometry = dependencies["transform"] Polygon = dependencies["Polygon"] header_height = 88 with rasterio.open(record["resolved_source_tile_path"]) as source: pixels = normalize_raster(source.read(), numpy) bbox = record["bbox_json"] if ( float(bbox["x_min"]) < 0 or float(bbox["y_min"]) < 0 or float(bbox["x_max"]) > source.width or float(bbox["y_max"]) > source.height ): raise SystemExit( f"Detection bbox_json exceeds source tile dimensions: {record['candidate_feature_id']}" ) bbox_width = float(bbox["x_max"]) - float(bbox["x_min"]) bbox_height = float(bbox["y_max"]) - float(bbox["y_min"]) crop_width = min(source.width, max(128, math.ceil(bbox_width * 4))) crop_height = min(source.height, max(128, math.ceil(bbox_height * 4))) center_x = (float(bbox["x_min"]) + float(bbox["x_max"])) / 2 center_y = (float(bbox["y_min"]) + float(bbox["y_max"])) / 2 crop_left = max(0, min(source.width - crop_width, round(center_x - crop_width / 2))) crop_top = max(0, min(source.height - crop_height, round(center_y - crop_height / 2))) crop_right = crop_left + crop_width crop_bottom = crop_top + crop_height image = Image.fromarray(pixels, mode="RGB").crop( (crop_left, crop_top, crop_right, crop_bottom) ) scale = min(thumb_size / image.width, thumb_size / image.height) render_width = max(1, round(image.width * scale)) render_height = max(1, round(image.height * scale)) image = image.resize((render_width, render_height), Image.Resampling.BILINEAR) image_offset_x = (thumb_size - render_width) // 2 image_offset_y = (thumb_size - render_height) // 2 card = Image.new( "RGB", (thumb_size, thumb_size + header_height), color=(242, 245, 247) ) card.paste(image, (image_offset_x, header_height + image_offset_y)) draw = ImageDraw.Draw(card) font = ImageFont.load_default() draw.rectangle((0, 0, thumb_size, header_height), fill=(22, 29, 38)) draw.text( (6, 6), f"{record['sample_slug']} conf {record['confidence']:.2f}", fill=(255, 255, 255), font=font, ) draw.text( (6, 23), f"{record['area_bucket'].split('_', 1)[0]} | {record['area_m2']:.1f} m2", fill=(197, 215, 231), font=font, ) draw.text( (6, 40), Path(record["source_tile_path"]).name[:24], fill=(197, 215, 231), font=font, ) draw.text( (6, 56), "red: candidate", fill=(235, 238, 241), font=font ) draw.text( (6, 72), "green ref | blue miss", fill=(235, 238, 241), font=font ) candidate_box = ( (float(bbox["x_min"]) - crop_left) * scale + image_offset_x, header_height + (float(bbox["y_min"]) - crop_top) * scale + image_offset_y, (float(bbox["x_max"]) - crop_left) * scale + image_offset_x, header_height + (float(bbox["y_max"]) - crop_top) * scale + image_offset_y, ) overlay_count = 0 if source.crs: transformer = Transformer.from_crs("EPSG:4326", source.crs, always_xy=True) crop_bounds = Polygon( [ source.transform * (crop_left, crop_top), source.transform * (crop_right, crop_top), source.transform * (crop_right, crop_bottom), source.transform * (crop_left, crop_bottom), ] ) overlay = Image.new("RGBA", (thumb_size, thumb_size), (0, 0, 0, 0)) overlay_draw = ImageDraw.Draw(overlay) for reference in references: geometry = transform_geometry(transformer.transform, shape(reference["geometry"])) if geometry.is_empty or not geometry.intersects(crop_bounds): continue color = ( (39, 174, 96) if reference["role"] == "match_reference" else (52, 152, 219) ) draw_geometry( overlay_draw, geometry, ~source.transform, scale, image_offset_x, image_offset_y, crop_left, crop_top, color, ) overlay_count += 1 card.paste(overlay, (0, header_height), overlay) draw.rectangle(candidate_box, outline=(231, 76, 60), width=3) return card, overlay_count def build_contact_sheet(cards: list[Any], columns: int, output_path: Path, Image: Any) -> None: gap = 12 rows = math.ceil(len(cards) / columns) width = columns * cards[0].width + (columns + 1) * gap height = rows * cards[0].height + (rows + 1) * gap sheet = Image.new("RGB", (width, height), color=(220, 226, 231)) for index, card in enumerate(cards): column = index % columns row = index // columns sheet.paste( card, (gap + column * (card.width + gap), gap + row * (card.height + gap)), ) sheet.save(output_path) def read_population( portfolio_path: Path, selected_slugs: set[str], storage_root: Path, dependencies: dict[str, Any], ) -> tuple[list[dict[str, Any]], dict[str, list[dict[str, Any]]]]: portfolio = load_json(portfolio_path) samples = portfolio.get("samples") or [] if not isinstance(samples, list): raise SystemExit("Portfolio samples must be a list") geod = dependencies["Geod"](ellps="WGS84") shape = dependencies["shape"] from audit_detection_false_negative_evidence import area_bucket population: list[dict[str, Any]] = [] references: dict[str, list[dict[str, Any]]] = defaultdict(list) for sample in samples: if not isinstance(sample, dict): raise SystemExit("Portfolio contains a non-object sample") slug = str(sample.get("sample_slug") or "").strip().lower() if not slug or (selected_slugs and slug not in selected_slugs): continue evidence_path = resolve_evidence_path(portfolio_path, sample) evidence = load_json(evidence_path) if evidence.get("type") != "FeatureCollection": raise SystemExit(f"Evidence must be a FeatureCollection: {evidence_path}") false_positive_count = 0 for feature in evidence.get("features") or []: if not isinstance(feature, dict): raise SystemExit(f"Evidence contains a non-object feature: {evidence_path}") properties = feature.get("properties") or {} role = str(properties.get("qa_evidence_role") or "") if role in {"match_reference", "false_negative"}: reference_geometry = shape(feature.get("geometry")) if ( reference_geometry.is_empty or not reference_geometry.is_valid or reference_geometry.geom_type not in {"Polygon", "MultiPolygon"} ): raise SystemExit( f"Reference evidence has invalid polygon geometry: {feature.get('id')}" ) references[slug].append( {"role": role, "geometry": feature.get("geometry")} ) continue if role != "false_positive": continue false_positive_count += 1 missing = [ key for key in ("detection_id", "confidence", "source_tile_path", "bbox_json") if properties.get(key) in (None, "") ] if missing: raise SystemExit( f"False-positive evidence {feature.get('id')} lacks persisted detection provenance: " + ", ".join(missing) + ". Re-export the evidence portfolio with the current backend." ) geometry = shape(feature.get("geometry")) if geometry.is_empty or not geometry.is_valid or geometry.geom_type not in { "Polygon", "MultiPolygon", }: raise SystemExit(f"False-positive evidence has invalid polygon geometry: {feature.get('id')}") confidence = float(properties["confidence"]) if not 0 <= confidence <= 1: raise SystemExit(f"Detection confidence is outside [0, 1]: {confidence}") bbox = properties["bbox_json"] if not isinstance(bbox, dict) or any( key not in bbox for key in ("x_min", "y_min", "x_max", "y_max") ): raise SystemExit(f"Detection bbox_json is invalid: {feature.get('id')}") try: x_min, y_min, x_max, y_max = ( float(bbox[key]) for key in ("x_min", "y_min", "x_max", "y_max") ) except (TypeError, ValueError) as exc: raise SystemExit( f"Detection bbox_json must contain numeric values: {feature.get('id')}" ) from exc if not (x_min < x_max and y_min < y_max): raise SystemExit(f"Detection bbox_json is not ordered: {feature.get('id')}") source_tile_path = str(properties["source_tile_path"]) resolved_tile = resolve_source_tile(source_tile_path, storage_root) area_m2 = abs(float(geod.geometry_area_perimeter(geometry)[0])) candidate_id = str( properties.get("candidate_feature_id") or properties.get("detection_id") or feature.get("id") ) population.append( { "candidate_feature_id": candidate_id, "evidence_feature_id": str(feature.get("id") or candidate_id), "sample_slug": slug, "confidence": confidence, "confidence_band": confidence_band(confidence), "area_m2": area_m2, "area_bucket": area_bucket(area_m2), "analysis_run_id": properties.get("analysis_run_id"), "quality_check_id": properties.get("quality_check_id"), "source_tile_path": source_tile_path, "resolved_source_tile_path": str(resolved_tile), "bbox_json": bbox, "model_name": properties.get("model_name"), "model_version": properties.get("model_version"), "geometry": feature.get("geometry"), "properties": properties, } ) declared = (sample.get("role_counts") or {}).get("false_positive") if declared is not None and int(declared) != false_positive_count: raise SystemExit( f"Portfolio role count drift for {slug}: declared {declared}, found {false_positive_count}" ) if not population: raise SystemExit("No false-positive evidence was found for the selected samples") return population, references def write_decisions(selected: list[dict[str, Any]], output_path: Path) -> None: with output_path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=DECISION_FIELDS) writer.writeheader() for record in selected: writer.writerow( { key: record.get(key, "") for key in DECISION_FIELDS if key not in {"review_decision", "review_notes"} } | {"review_decision": "unreviewed", "review_notes": ""} ) def write_markdown(report: dict[str, Any], output_dir: Path) -> None: lines = [ "# Detection false-positive visual review", "", "> A QA false-positive is not automatically a model error. Review each selected detection against the imagery and reference context.", "", f"- Status: `{report['status']}`", f"- Population: {report['population_count']}", f"- Selected for manual review: {report['selected_feature_count']}", f"- AOIs: {', '.join(report['selected_sample_slugs'])}", "", "## Allowed decisions", "", "- `confirmed_model_false_positive`: imagery confirms that the model detection is wrong.", "- `reference_gap_or_change`: imagery supports the detection but the reference is missing or stale.", "- `qa_alignment_mismatch`: CRS, geometry or matching tolerance caused the QA result.", "- `uncertain`: available evidence is insufficient.", "- `unreviewed`: no operator decision has been made.", "", "Only `confirmed_model_false_positive` records may be exported as possible hard-negative candidates.", "", "## Contact sheets", "", ] for sheet in report["contact_sheets"]: lines.extend([f"![{sheet['path']}]({sheet['path']})", ""]) (output_dir / MARKDOWN_NAME).write_text("\n".join(lines) + "\n", encoding="utf-8") def run(args: argparse.Namespace) -> dict[str, Any]: if min(args.max_features, args.columns, args.cards_per_sheet, args.thumb_size) <= 0: raise SystemExit("Review limits, columns and thumbnail size must be positive") dependencies = require_dependencies() portfolio_path = args.portfolio.expanduser().resolve() storage_root = args.storage_root.expanduser().resolve() requested_slugs = { value.strip().lower() for value in args.sample_slugs.split(",") if value.strip() } population, references = read_population( portfolio_path, requested_slugs, storage_root, dependencies ) available_slugs = {record["sample_slug"] for record in population} missing_slugs = requested_slugs - available_slugs if missing_slugs: raise SystemExit("Selected sample slug is absent from the portfolio: " + ", ".join(sorted(missing_slugs))) selected = stratified_selection(population, min(args.max_features, len(population))) output_dir = args.output_dir.expanduser().resolve() output_dir.mkdir(parents=True, exist_ok=True) cards = [] overlay_count = 0 for record in selected: card, count = render_card( record, references.get(record["sample_slug"], []), args.thumb_size, dependencies, ) cards.append(card) overlay_count += count contact_sheets = [] for index in range(0, len(cards), args.cards_per_sheet): batch = cards[index : index + args.cards_per_sheet] path = output_dir / f"false_positive_review_{index // args.cards_per_sheet + 1:03d}.png" build_contact_sheet(batch, args.columns, path, dependencies["Image"]) contact_sheets.append({"path": path.name, "feature_count": len(batch)}) portfolio = load_json(portfolio_path) serializable_selected = [ {key: value for key, value in record.items() if key != "resolved_source_tile_path"} for record in selected ] report = { "generated_at": datetime.now(timezone.utc).isoformat(), "schema_version": 1, "status": "review_required", "portfolio_path": str(portfolio_path), "model_asset_id": portfolio.get("model_asset_id"), "model_sha256": portfolio.get("model_sha256"), "population_count": len(population), "selected_feature_count": len(selected), "selected_sample_slugs": sorted({record["sample_slug"] for record in selected}), "selected_area_buckets": sorted({record["area_bucket"] for record in selected}), "selected_confidence_bands": sorted({record["confidence_band"] for record in selected}), "missing_provenance_count": 0, "missing_tile_count": 0, "reference_overlay_feature_count": overlay_count, "contact_sheets": contact_sheets, "selected_features": serializable_selected, } (output_dir / JSON_NAME).write_text( json.dumps(report, indent=2, sort_keys=True), encoding="utf-8" ) write_decisions(selected, output_dir / DECISIONS_NAME) write_markdown(report, output_dir) return report def main() -> int: args = parse_args() report = run(args) print("Detection false-positive review required") print(f"Selected features: {report['selected_feature_count']}") print(f"Summary: {args.output_dir.expanduser().resolve() / JSON_NAME}") print(f"Decisions: {args.output_dir.expanduser().resolve() / DECISIONS_NAME}") return 0 if __name__ == "__main__": raise SystemExit(main())