Add operator YOLO label QA contact sheets

scripts/README.md

@@ -420,6 +420,25 @@ median box area, small-box share and sample-specific quality warnings. Treat
 next action is usually more positive AOIs, better validation coverage or more
 unique hard negatives rather than simply extending epochs.
 
+Render visual label QA contact sheets before spending CPU on another training
+run:
+
+```bash
+docker exec -it geointel python3 /app/scripts/render_operator_yolo_label_qa_contact_sheets.py \
+  --summary-path /app/storage/operator-data/yolo-building-aoi1024-cleanpx12vis035/yolo_tile_dataset_summary.json \
+  --output-dir /app/artifacts/operator-yolo-label-qa/aoi1024-cleanpx12vis035 \
+  --max-tiles 32 \
+  --columns 4 \
+  --thumb-size 256
+```
+
+The renderer writes `operator_yolo_label_qa_summary.json`,
+`operator_yolo_label_qa_contact_sheet.md` and `contact_sheet_001.png`. It draws
+existing YOLO labels on existing tile images only; it does not run inference,
+train a model, fetch providers or create fake detections. Missing image files,
+missing label files and invalid YOLO rows are reported in the JSON/Markdown
+artifacts.
+
 Current Tower audit status:
 
 - `yolo-building-tile-expanded160`: clean baseline; no missing/invalid labels.

scripts/render_operator_yolo_label_qa_contact_sheets.py

@@ -0,0 +1,360 @@
+#!/usr/bin/env python3
+"""Render visual QA contact sheets for exported operator YOLO tile datasets.
+
+This helper is operator tooling only. It reads existing tile images and YOLO
+label files, then writes visual evidence artifacts. It does not train, infer,
+fetch provider data or mutate application persistence.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+from pathlib import Path
+from typing import Any
+
+from PIL import Image, ImageDraw, ImageFont
+
+
+JSON_NAME = "operator_yolo_label_qa_summary.json"
+MARKDOWN_NAME = "operator_yolo_label_qa_contact_sheet.md"
+CONTACT_SHEET_NAME = "contact_sheet_001.png"
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Render YOLO tile label overlays into deterministic operator QA contact sheets.",
+    )
+    parser.add_argument("--summary-path", required=True, help="Path to yolo_tile_dataset_summary.json")
+    parser.add_argument("--output-dir", required=True, help="Directory for JSON, Markdown and PNG artifacts")
+    parser.add_argument("--max-tiles", type=int, default=24, help="Maximum selected tiles to render")
+    parser.add_argument("--columns", type=int, default=4, help="Contact-sheet columns")
+    parser.add_argument("--thumb-size", type=int, default=256, help="Rendered tile thumbnail size in pixels")
+    return parser.parse_args()
+
+
+def load_json(path: Path) -> dict[str, Any]:
+    with path.open("r", encoding="utf-8") as handle:
+        data = json.load(handle)
+    if not isinstance(data, dict):
+        raise ValueError(f"Expected JSON object in {path}")
+    return data
+
+
+def resolve_path(raw_path: str | None, summary_path: Path) -> Path | None:
+    if not raw_path:
+        return None
+
+    candidate = Path(raw_path)
+    if candidate.exists():
+        return candidate
+
+    if candidate.is_absolute() and raw_path.startswith("/app/"):
+        local_candidate = Path.cwd() / raw_path.removeprefix("/app/")
+        if local_candidate.exists():
+            return local_candidate
+
+        summary_parent_candidate = summary_path.parent / raw_path.removeprefix("/app/")
+        if summary_parent_candidate.exists():
+            return summary_parent_candidate
+
+    relative_candidate = summary_path.parent / raw_path
+    if relative_candidate.exists():
+        return relative_candidate
+
+    return candidate
+
+
+def parse_yolo_label_file(path: Path | None) -> tuple[list[dict[str, float]], int, bool]:
+    if path is None or not path.exists():
+        return [], 0, True
+
+    boxes: list[dict[str, float]] = []
+    invalid_count = 0
+    for line in path.read_text(encoding="utf-8").splitlines():
+        stripped = line.strip()
+        if not stripped:
+            continue
+        parts = stripped.split()
+        if len(parts) != 5:
+            invalid_count += 1
+            continue
+        try:
+            class_id = int(float(parts[0]))
+            center_x = float(parts[1])
+            center_y = float(parts[2])
+            width = float(parts[3])
+            height = float(parts[4])
+        except ValueError:
+            invalid_count += 1
+            continue
+        if not (0 <= center_x <= 1 and 0 <= center_y <= 1 and 0 < width <= 1 and 0 < height <= 1):
+            invalid_count += 1
+            continue
+        boxes.append(
+            {
+                "class_id": float(class_id),
+                "center_x": center_x,
+                "center_y": center_y,
+                "width": width,
+                "height": height,
+            }
+        )
+    return boxes, invalid_count, False
+
+
+def tile_sort_key(tile: dict[str, Any]) -> tuple[int, str, str, int, int]:
+    return (
+        -int(tile.get("label_count") or 0),
+        str(tile.get("sample_slug") or ""),
+        str(tile.get("split") or ""),
+        int(tile.get("tile_index") or 0),
+        int(tile.get("repeat_index") or 0),
+    )
+
+
+def select_tiles(tiles: list[dict[str, Any]], max_tiles: int) -> list[dict[str, Any]]:
+    if max_tiles <= 0:
+        raise ValueError("max_tiles must be positive")
+
+    kept_tiles = [tile for tile in tiles if isinstance(tile, dict) and tile.get("kept", True)]
+    positives = sorted(
+        [tile for tile in kept_tiles if not (bool(tile.get("is_negative")) or int(tile.get("label_count") or 0) == 0)],
+        key=tile_sort_key,
+    )
+    negatives = sorted(
+        [tile for tile in kept_tiles if bool(tile.get("is_negative")) or int(tile.get("label_count") or 0) == 0],
+        key=tile_sort_key,
+    )
+
+    negative_slots = min(len(negatives), max(1, max_tiles // 5)) if negatives and max_tiles > 1 else 0
+    selected = positives[: max_tiles - negative_slots] + negatives[:negative_slots]
+
+    if len(selected) < max_tiles:
+        selected_ids = {id(tile) for tile in selected}
+        remainder = [tile for tile in sorted(kept_tiles, key=tile_sort_key) if id(tile) not in selected_ids]
+        selected.extend(remainder[: max_tiles - len(selected)])
+
+    return sorted(selected[:max_tiles], key=tile_sort_key)
+
+
+def draw_tile_card(
+    image_path: Path,
+    boxes: list[dict[str, float]],
+    tile: dict[str, Any],
+    thumb_size: int,
+    invalid_label_count: int,
+    missing_label_file: bool,
+) -> Image.Image:
+    header_height = 44
+    card = Image.new("RGB", (thumb_size, thumb_size + header_height), color=(245, 247, 250))
+    image = Image.open(image_path).convert("RGB").resize((thumb_size, thumb_size))
+    card.paste(image, (0, header_height))
+
+    draw = ImageDraw.Draw(card)
+    draw.rectangle((0, 0, thumb_size - 1, header_height - 1), fill=(20, 31, 44))
+    draw.rectangle((0, header_height, thumb_size - 1, thumb_size + header_height - 1), outline=(20, 31, 44), width=1)
+
+    font = ImageFont.load_default()
+    title = f"{tile.get('sample_slug', 'unknown')} / {tile.get('split', 'unknown')} / labels {tile.get('label_count', 0)}"
+    subtitle_parts = [str(tile.get("background_category") or tile.get("sample_role") or "unknown")]
+    if missing_label_file:
+        subtitle_parts.append("missing-label-file")
+    if invalid_label_count:
+        subtitle_parts.append(f"invalid:{invalid_label_count}")
+    draw.text((6, 6), title[:44], fill=(255, 255, 255), font=font)
+    draw.text((6, 24), " | ".join(subtitle_parts)[:52], fill=(191, 219, 254), font=font)
+
+    for box in boxes:
+        x_center = box["center_x"] * thumb_size
+        y_center = box["center_y"] * thumb_size + header_height
+        width = box["width"] * thumb_size
+        height = box["height"] * thumb_size
+        left = max(0, x_center - width / 2)
+        top = max(header_height, y_center - height / 2)
+        right = min(thumb_size - 1, x_center + width / 2)
+        bottom = min(thumb_size + header_height - 1, y_center + height / 2)
+        draw.rectangle((left, top, right, bottom), outline=(255, 214, 10), width=3)
+
+    return card
+
+
+def build_contact_sheet(cards: list[Image.Image], columns: int, output_path: Path) -> None:
+    if not cards:
+        return
+    if columns <= 0:
+        raise ValueError("columns must be positive")
+
+    gap = 12
+    cell_width = max(card.width for card in cards)
+    cell_height = max(card.height for card in cards)
+    rows = math.ceil(len(cards) / columns)
+    sheet_width = columns * cell_width + (columns + 1) * gap
+    sheet_height = rows * cell_height + (rows + 1) * gap
+    sheet = Image.new("RGB", (sheet_width, sheet_height), color=(226, 232, 240))
+
+    for index, card in enumerate(cards):
+        row = index // columns
+        column = index % columns
+        x = gap + column * (cell_width + gap)
+        y = gap + row * (cell_height + gap)
+        sheet.paste(card, (x, y))
+
+    sheet.save(output_path)
+
+
+def build_report(summary: dict[str, Any], summary_path: Path, args: argparse.Namespace) -> tuple[dict[str, Any], list[Image.Image]]:
+    tiles = summary.get("tiles") or []
+    if not isinstance(tiles, list):
+        raise ValueError("Expected summary tiles to be a list")
+
+    selected_tiles = select_tiles(tiles, args.max_tiles)
+    rendered_cards: list[Image.Image] = []
+    selected_report_tiles: list[dict[str, Any]] = []
+    missing_image_count = 0
+    missing_label_file_count = 0
+    invalid_label_count = 0
+    valid_label_count = 0
+
+    for tile in selected_tiles:
+        image_path = resolve_path(tile.get("image_path"), summary_path)
+        label_path = resolve_path(tile.get("label_path"), summary_path)
+        boxes, tile_invalid_count, missing_label_file = parse_yolo_label_file(label_path)
+        invalid_label_count += tile_invalid_count
+        valid_label_count += len(boxes)
+        if missing_label_file:
+            missing_label_file_count += 1
+
+        rendered = False
+        if image_path is None or not image_path.exists():
+            missing_image_count += 1
+        else:
+            rendered_cards.append(
+                draw_tile_card(
+                    image_path=image_path,
+                    boxes=boxes,
+                    tile=tile,
+                    thumb_size=args.thumb_size,
+                    invalid_label_count=tile_invalid_count,
+                    missing_label_file=missing_label_file,
+                )
+            )
+            rendered = True
+
+        selected_report_tiles.append(
+            {
+                "sample_slug": str(tile.get("sample_slug") or "unknown"),
+                "sample_role": str(tile.get("sample_role") or "unknown"),
+                "background_category": str(tile.get("background_category") or ""),
+                "split": str(tile.get("split") or "unknown"),
+                "tile_index": int(tile.get("tile_index") or 0),
+                "image_path": str(image_path) if image_path is not None else None,
+                "label_path": str(label_path) if label_path is not None else None,
+                "label_count": int(tile.get("label_count") or 0),
+                "valid_label_count": len(boxes),
+                "invalid_label_count": tile_invalid_count,
+                "missing_label_file": missing_label_file,
+                "rendered": rendered,
+            }
+        )
+
+    contact_sheets = []
+    if rendered_cards:
+        contact_sheets.append(
+            {
+                "path": CONTACT_SHEET_NAME,
+                "tile_count": len(rendered_cards),
+                "columns": args.columns,
+                "thumb_size": args.thumb_size,
+            }
+        )
+
+    return (
+        {
+            "status": "ok" if rendered_cards else "no_renderable_tiles",
+            "summary_path": str(summary_path),
+            "dataset_output_dir": summary.get("output_dir"),
+            "class_names": summary.get("class_names", []),
+            "max_tiles": args.max_tiles,
+            "columns": args.columns,
+            "thumb_size": args.thumb_size,
+            "selected_tile_count": len(selected_tiles),
+            "rendered_tile_count": len(rendered_cards),
+            "missing_image_count": missing_image_count,
+            "missing_label_file_count": missing_label_file_count,
+            "invalid_label_count": invalid_label_count,
+            "valid_label_count": valid_label_count,
+            "contact_sheets": contact_sheets,
+            "selected_tiles": selected_report_tiles,
+        },
+        rendered_cards,
+    )
+
+
+def write_markdown(report: dict[str, Any], output_dir: Path) -> None:
+    lines = [
+        "# Operator YOLO Label QA Contact Sheets",
+        "",
+        f"- status: `{report['status']}`",
+        f"- selected tiles: {report['selected_tile_count']}",
+        f"- rendered tiles: {report['rendered_tile_count']}",
+        f"- missing images: {report['missing_image_count']}",
+        f"- missing label files: {report['missing_label_file_count']}",
+        f"- invalid label rows: {report['invalid_label_count']}",
+        f"- valid labels rendered: {report['valid_label_count']}",
+        "",
+        "## Contact Sheets",
+        "",
+    ]
+    if report["contact_sheets"]:
+        for sheet in report["contact_sheets"]:
+            lines.append(f"- `{sheet['path']}` ({sheet['tile_count']} tiles)")
+            lines.append(f"")
+            lines.append(f"![{sheet['path']}]({sheet['path']})")
+            lines.append("")
+    else:
+        lines.append("- None")
+
+    lines.extend(["", "## Selected Tiles", ""])
+    for tile in report["selected_tiles"]:
+        flags = []
+        if tile["missing_label_file"]:
+            flags.append("missing-label-file")
+        if tile["invalid_label_count"]:
+            flags.append(f"invalid:{tile['invalid_label_count']}")
+        flag_text = ", ".join(flags) if flags else "ok"
+        lines.append(
+            "- "
+            f"{tile['sample_slug']} ({tile['split']}, {tile['background_category'] or tile['sample_role']}): "
+            f"{tile['label_count']} expected labels, {tile['valid_label_count']} rendered labels, {flag_text}"
+        )
+
+    (output_dir / MARKDOWN_NAME).write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def main() -> int:
+    args = parse_args()
+    summary_path = Path(args.summary_path).resolve()
+    output_dir = Path(args.output_dir).resolve()
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    summary = load_json(summary_path)
+    report, cards = build_report(summary, summary_path, args)
+    if cards:
+        build_contact_sheet(cards, args.columns, output_dir / CONTACT_SHEET_NAME)
+
+    (output_dir / JSON_NAME).write_text(json.dumps(report, indent=2, sort_keys=True), encoding="utf-8")
+    write_markdown(report, output_dir)
+
+    print("Operator YOLO label QA contact sheets rendered")
+    print(f"Status: {report['status']}")
+    print(f"JSON: {output_dir / JSON_NAME}")
+    print(f"Markdown: {output_dir / MARKDOWN_NAME}")
+    for sheet in report["contact_sheets"]:
+        print(f"Contact sheet: {output_dir / sheet['path']}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())