Files
geointel/scripts/render_operator_yolo_label_qa_contact_sheets.py
T
Codex 5688feec35
GeoIntel CI / docs-smoke (push) Has been cancelled
GeoIntel CI / contract-smoke (push) Has been cancelled
Report blank YOLO label QA tiles
2026-07-11 22:05:13 +02:00

388 lines
15 KiB
Python

#!/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")
parser.add_argument(
"--blank-range-threshold",
type=int,
default=3,
help="Mark rendered images with grayscale max-min range at or below this value as low variance.",
)
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,
low_visual_variance: 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}")
if low_visual_variance:
subtitle_parts.append("low-variance")
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 image_has_low_visual_variance(image_path: Path, blank_range_threshold: int) -> bool:
image = Image.open(image_path).convert("L")
min_value, max_value = image.getextrema()
return (max_value - min_value) <= blank_range_threshold
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
low_visual_variance_tile_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
low_visual_variance = False
if image_path is None or not image_path.exists():
missing_image_count += 1
else:
low_visual_variance = image_has_low_visual_variance(image_path, args.blank_range_threshold)
if low_visual_variance:
low_visual_variance_tile_count += 1
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,
low_visual_variance=low_visual_variance,
)
)
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,
"low_visual_variance": low_visual_variance,
"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,
"low_visual_variance_tile_count": low_visual_variance_tile_count,
"blank_range_threshold": args.blank_range_threshold,
"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"- low-variance rendered tiles: {report['low_visual_variance_tile_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']}")
if tile["low_visual_variance"]:
flags.append("low-variance")
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())