Add Mol false-negative visual review evidence

scripts/render_detection_false_negative_review_contact_sheets.py

@@ -0,0 +1,595 @@
+#!/usr/bin/env python3
+"""Render persisted detection false negatives for explicit operator review.
+
+The script is read-only. It resolves the tile manifest recorded by the selected
+analysis run, projects missed reference geometries onto those source tiles and
+never infers a review decision or mutates application/model state.
+"""
+
+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
+
+from audit_detection_false_negative_evidence import area_bucket
+from render_detection_false_positive_review_contact_sheets import (
+    build_contact_sheet,
+    draw_geometry,
+    load_json,
+    normalize_raster,
+    require_dependencies,
+    resolve_evidence_path,
+)
+
+
+JSON_NAME = "detection_false_negative_review_summary.json"
+MARKDOWN_NAME = "detection_false_negative_review.md"
+DECISIONS_NAME = "false_negative_review_decisions.csv"
+DECISION_FIELDS = (
+    "reference_feature_id",
+    "evidence_feature_id",
+    "sample_slug",
+    "area_m2",
+    "area_bucket",
+    "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-negative 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 resolve_confined_file(raw: str, storage_root: Path, description: str) -> Path:
+    root = storage_root.expanduser().resolve()
+    candidate = Path(raw).expanduser()
+    if not candidate.is_absolute():
+        candidate = root / candidate
+    candidate = candidate.resolve()
+    try:
+        candidate.relative_to(root)
+    except ValueError as exc:
+        raise SystemExit(f"{description} is outside storage root: {candidate}") from exc
+    if not candidate.is_file():
+        raise SystemExit(f"{description} is not readable: {candidate}")
+    return candidate
+
+
+def resolve_manifest_path(
+    sample: dict[str, Any], storage_root: Path
+) -> Path:
+    raw_summary = str(
+        sample.get("copied_summary_path") or sample.get("source_summary_path") or ""
+    ).strip()
+    if not raw_summary:
+        raise SystemExit(
+            f"Portfolio sample {sample.get('sample_slug')} has no persisted run summary path"
+        )
+    summary_path = resolve_confined_file(raw_summary, storage_root, "Run summary")
+    summary = load_json(summary_path)
+    items = [item for item in summary.get("items") or [] if isinstance(item, dict)]
+    manifest_paths = {
+        str(item.get("manifest_path") or "").strip()
+        for item in items
+        if str(item.get("manifest_path") or "").strip()
+    }
+    if len(manifest_paths) != 1:
+        raise SystemExit(
+            f"Expected one persisted tile manifest for {sample.get('sample_slug')}; "
+            f"found {len(manifest_paths)}"
+        )
+    return resolve_confined_file(
+        next(iter(manifest_paths)), storage_root, "Tile manifest"
+    )
+
+
+def load_tiles(
+    manifest_path: Path,
+    storage_root: Path,
+    dependencies: dict[str, Any],
+) -> list[dict[str, Any]]:
+    manifest = load_json(manifest_path)
+    rasterio = dependencies["rasterio"]
+    box = dependencies["box"]
+    raw_paths = [
+        str(tile.get("path") or "").strip()
+        for tile in manifest.get("tiles") or []
+        if isinstance(tile, dict)
+    ] or [str(path).strip() for path in manifest.get("tile_paths") or []]
+    if not raw_paths:
+        raise SystemExit(f"Tile manifest contains no source tiles: {manifest_path}")
+
+    tiles = []
+    for raw_path in raw_paths:
+        tile_path = resolve_confined_file(raw_path, storage_root, "Source tile")
+        with rasterio.open(tile_path) as source:
+            if not source.crs:
+                raise SystemExit(f"Source tile has no CRS: {tile_path}")
+            tiles.append(
+                {
+                    "path": str(tile_path),
+                    "crs": source.crs,
+                    "geometry": box(*source.bounds),
+                }
+            )
+    return tiles
+
+
+def choose_source_tile(
+    geometry: Any,
+    tiles: list[dict[str, Any]],
+    dependencies: dict[str, Any],
+) -> str | None:
+    Transformer = dependencies["Transformer"]
+    transform_geometry = dependencies["transform"]
+    candidates: list[tuple[float, float, str]] = []
+    for tile in tiles:
+        transformer = Transformer.from_crs("EPSG:4326", tile["crs"], always_xy=True)
+        projected = transform_geometry(transformer.transform, geometry)
+        intersection = projected.intersection(tile["geometry"])
+        if intersection.is_empty:
+            continue
+        candidates.append(
+            (
+                float(intersection.area),
+                -float(projected.centroid.distance(tile["geometry"].centroid)),
+                tile["path"],
+            )
+        )
+    return max(candidates)[2] if candidates else None
+
+
+def stable_sort_key(record: dict[str, Any]) -> str:
+    identity = f"{record['sample_slug']}:{record['reference_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], list[dict[str, Any]]] = defaultdict(list)
+    for record in records:
+        grouped[(record["sample_slug"], record["area_bucket"])].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 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]]],
+    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"]
+    population: list[dict[str, Any]] = []
+    context: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    outside_tile_coverage: list[dict[str, Any]] = []
+
+    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
+        tiles = load_tiles(
+            resolve_manifest_path(sample, storage_root), storage_root, dependencies
+        )
+        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_negative_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 not in {
+                "false_negative",
+                "false_positive",
+                "match_candidate",
+                "match_reference",
+            }:
+                continue
+            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"Detection QA evidence has invalid polygon geometry: {feature.get('id')}"
+                )
+            if role != "false_negative":
+                context[slug].append(
+                    {"role": role, "geometry": feature.get("geometry")}
+                )
+                continue
+
+            false_negative_count += 1
+            reference_id = str(
+                properties.get("reference_feature_id")
+                or properties.get("source_feature_id")
+                or properties.get("feature_id")
+                or feature.get("id")
+            )
+            area_m2 = abs(float(geod.geometry_area_perimeter(geometry)[0]))
+            source_tile_path = choose_source_tile(geometry, tiles, dependencies)
+            if source_tile_path is None:
+                outside_tile_coverage.append(
+                    {
+                        "reference_feature_id": reference_id,
+                        "evidence_feature_id": str(feature.get("id") or reference_id),
+                        "sample_slug": slug,
+                        "area_m2": area_m2,
+                        "area_bucket": area_bucket(area_m2),
+                        "geometry": feature.get("geometry"),
+                        "properties": properties,
+                    }
+                )
+                continue
+            population.append(
+                {
+                    "reference_feature_id": reference_id,
+                    "evidence_feature_id": str(feature.get("id") or reference_id),
+                    "sample_slug": slug,
+                    "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,
+                    "geometry": feature.get("geometry"),
+                    "properties": properties,
+                }
+            )
+        declared = (sample.get("role_counts") or {}).get("false_negative")
+        if declared is not None and int(declared) != false_negative_count:
+            raise SystemExit(
+                f"Portfolio role count drift for {slug}: declared {declared}, "
+                f"found {false_negative_count}"
+            )
+    if not population:
+        raise SystemExit(
+            "No reviewable false-negative evidence intersects the selected tile manifests"
+        )
+    return population, context, outside_tile_coverage
+
+
+def render_card(
+    record: dict[str, Any],
+    context: list[dict[str, Any]],
+    thumb_size: int,
+    dependencies: dict[str, Any],
+    projection_cache: dict[tuple[str, str], list[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["source_tile_path"]) as source:
+        pixels = normalize_raster(source.read(), numpy)
+        transformer = Transformer.from_crs("EPSG:4326", source.crs, always_xy=True)
+        target = transform_geometry(transformer.transform, shape(record["geometry"]))
+        cache_key = (record["sample_slug"], source.crs.to_string())
+        projected_context = projection_cache.get(cache_key)
+        if projected_context is None:
+            projected_context = [
+                {
+                    "role": item["role"],
+                    "geometry": transform_geometry(
+                        transformer.transform, shape(item["geometry"])
+                    ),
+                }
+                for item in context
+            ]
+            projection_cache[cache_key] = projected_context
+        inverse = ~source.transform
+        pixel_points = [inverse * (x, y) for x, y in target.envelope.exterior.coords]
+        columns = [point[0] for point in pixel_points]
+        rows = [point[1] for point in pixel_points]
+        x_min, x_max = min(columns), max(columns)
+        y_min, y_max = min(rows), max(rows)
+        bbox_width = max(1.0, x_max - x_min)
+        bbox_height = max(1.0, y_max - 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 = (x_min + x_max) / 2
+        center_y = (y_min + 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)
+        offset_x = (thumb_size - render_width) // 2
+        offset_y = (thumb_size - render_height) // 2
+        card = Image.new(
+            "RGB", (thumb_size, thumb_size + header_height), color=(242, 245, 247)
+        )
+        card.paste(image, (offset_x, header_height + 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), record["sample_slug"], 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: missed GRB", fill=(235, 238, 241), font=font)
+        draw.text(
+            (6, 72), "blue: candidate | green: matched ref", fill=(235, 238, 241), font=font
+        )
+
+        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)
+        overlay_count = 0
+        for item in projected_context:
+            geometry = item["geometry"]
+            if geometry.is_empty or not geometry.intersects(crop_bounds):
+                continue
+            color = (
+                (39, 174, 96)
+                if item["role"] == "match_reference"
+                else (52, 152, 219)
+            )
+            draw_geometry(
+                overlay_draw,
+                geometry,
+                inverse,
+                scale,
+                offset_x,
+                offset_y,
+                crop_left,
+                crop_top,
+                color,
+            )
+            overlay_count += 1
+        draw_geometry(
+            overlay_draw,
+            target,
+            inverse,
+            scale,
+            offset_x,
+            offset_y,
+            crop_left,
+            crop_top,
+            (231, 76, 60),
+        )
+        card.paste(overlay, (0, header_height), overlay)
+        return card, overlay_count
+
+
+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-negative visual review",
+        "",
+        "> A QA false-negative is not automatically a model miss. Review imagery, the persisted GRB footprint and nearby candidate geometry.",
+        "",
+        f"- Status: `{report['status']}`",
+        f"- Population: {report['population_count']}",
+        f"- Selected for manual review: {report['selected_feature_count']}",
+        f"- Outside tile coverage (excluded): {report['outside_tile_coverage_count']}",
+        f"- AOIs: {', '.join(report['selected_sample_slugs'])}",
+        "",
+        "## Allowed decisions",
+        "",
+        "- `confirmed_model_false_negative`: imagery confirms the referenced building and no suitable detection covers it.",
+        "- `reference_gap_or_change`: the persisted reference is absent or stale in the imagery.",
+        "- `qa_alignment_mismatch`: a nearby detection exists but geometry/alignment or matching tolerance prevented a match.",
+        "- `imagery_obscured_or_uncertain`: the image does not support a confident decision.",
+        "- `unreviewed`: no operator decision has been made.",
+        "",
+        "Only explicitly confirmed model misses may inform a future positive-training review set.",
+        "",
+        "## 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("False-negative visual review")
+    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, context, outside_tile_coverage = 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
+    projection_cache: dict[tuple[str, str], list[dict[str, Any]]] = {}
+    for record in selected:
+        card, count = render_card(
+            record,
+            context.get(record["sample_slug"], []),
+            args.thumb_size,
+            dependencies,
+            projection_cache,
+        )
+        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_negative_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)
+    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),
+        "evidence_population_count": len(population) + len(outside_tile_coverage),
+        "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}),
+        "missing_manifest_count": 0,
+        "missing_tile_count": 0,
+        "outside_tile_coverage_count": len(outside_tile_coverage),
+        "outside_tile_coverage_geojson_path": (
+            "false_negatives_outside_tile_coverage.geojson"
+        ),
+        "context_overlay_feature_count": overlay_count,
+        "contact_sheets": contact_sheets,
+        "selected_features": selected,
+    }
+    (output_dir / JSON_NAME).write_text(
+        json.dumps(report, indent=2, sort_keys=True), encoding="utf-8"
+    )
+    (output_dir / "false_negatives_outside_tile_coverage.geojson").write_text(
+        json.dumps(
+            {
+                "type": "FeatureCollection",
+                "features": [
+                    {
+                        "type": "Feature",
+                        "id": record["evidence_feature_id"],
+                        "geometry": record["geometry"],
+                        "properties": {
+                            **record["properties"],
+                            "review_exclusion_reason": "outside_tile_coverage",
+                            "sample_slug": record["sample_slug"],
+                            "area_m2": record["area_m2"],
+                            "area_bucket": record["area_bucket"],
+                        },
+                    }
+                    for record in outside_tile_coverage
+                ],
+            },
+            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-negative 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())