630 lines
24 KiB
Python
630 lines
24 KiB
Python
#!/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())
|