#!/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_TEMPLATE = "contact_sheet_{index:03d}.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( "--sample-slug", action="append", default=[], help="Render only this sample slug; repeat to select multiple samples.", ) 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( "--tiles-per-sheet", type=int, default=64, help="Maximum rendered cards per contact-sheet page. Default: 64.", ) 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 filter_tiles_by_samples( tiles: list[dict[str, Any]], sample_slugs: list[str] ) -> list[dict[str, Any]]: requested = {slug.strip() for slug in sample_slugs if slug.strip()} if not requested: return tiles available = {str(tile.get("sample_slug") or "") for tile in tiles} missing = sorted(requested - available) if missing: raise ValueError( f"Requested sample slugs absent from summary: {', '.join(missing)}" ) return [tile for tile in tiles if str(tile.get("sample_slug") or "") in requested] 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 balanced_tiles_by_sample( tiles: list[dict[str, Any]], limit: int ) -> list[dict[str, Any]]: if limit <= 0 or not tiles: return [] grouped: dict[str, list[dict[str, Any]]] = {} for tile in tiles: grouped.setdefault(str(tile.get("sample_slug") or "unknown"), []).append(tile) for sample_tiles in grouped.values(): sample_tiles.sort(key=tile_sort_key) sample_order = sorted(grouped, key=lambda slug: tile_sort_key(grouped[slug][0])) selected: list[dict[str, Any]] = [] depth = 0 while len(selected) < limit: added = False for slug in sample_order: sample_tiles = grouped[slug] if depth < len(sample_tiles): selected.append(sample_tiles[depth]) added = True if len(selected) == limit: break if not added: break depth += 1 return selected 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 = balanced_tiles_by_sample(positives, max_tiles - negative_slots) selected.extend(balanced_tiles_by_sample(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, relationship_highlights: dict[int, str], ) -> 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") if relationship_highlights: subtitle_parts.append(f"highlighted:{len(relationship_highlights)}") 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) highlight_styles = { "exact_duplicate": ((0, 220, 255), 5), "near_duplicate": ((255, 55, 55), 5), "possible_nested": ((255, 0, 220), 5), } for box_index, box in enumerate(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) if right < left or bottom < top: continue relationship = relationship_highlights.get(box_index) color, stroke_width = highlight_styles.get(relationship, ((255, 214, 10), 3)) draw.rectangle((left, top, right, bottom), outline=color, width=stroke_width) if relationship: draw.rectangle( (left, top, min(right, left + 30), min(bottom, top + 12)), fill=(20, 31, 44), ) draw.text( (left + 2, top + 1), f"#{box_index}", fill=color, font=font, ) return card def build_relationship_highlights( summary: dict[str, Any], ) -> dict[str, dict[int, str]]: """Map audited label rows to their highest-priority visual warning.""" priorities = { "possible_nested": 1, "near_duplicate": 2, "exact_duplicate": 3, } highlights: dict[str, dict[int, str]] = {} flagged_tiles = summary.get("flagged_tiles") or [] if not isinstance(flagged_tiles, list): return highlights for tile in flagged_tiles: if not isinstance(tile, dict) or not tile.get("label_path"): continue tile_highlights = highlights.setdefault(str(tile["label_path"]), {}) relationships = tile.get("relationships") or [] if not isinstance(relationships, list): continue for relationship in relationships: if not isinstance(relationship, dict): continue relationship_type = str(relationship.get("relationship") or "") if relationship_type not in priorities: continue for field in ("first_index", "second_index"): index = relationship.get(field) if not isinstance(index, int) or index < 0: continue existing = tile_highlights.get(index) if ( existing is None or priorities[relationship_type] > priorities[existing] ): tile_highlights[index] = relationship_type return highlights 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") filtered_tiles = filter_tiles_by_samples(tiles, args.sample_slug) selected_tiles = select_tiles(filtered_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 relationship_highlight_count = 0 relationship_highlights = build_relationship_highlights(summary) 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 ) tile_relationship_highlights = relationship_highlights.get(str(label_path), {}) relationship_highlight_count += len(tile_relationship_highlights) 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, relationship_highlights=tile_relationship_highlights, ) ) 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, "relationship_highlight_count": len(tile_relationship_highlights), "rendered": rendered, } ) if args.tiles_per_sheet <= 0: raise ValueError("tiles_per_sheet must be positive") contact_sheets = [ { "path": CONTACT_SHEET_NAME_TEMPLATE.format( index=(start // args.tiles_per_sheet) + 1 ), "tile_count": len(rendered_cards[start : start + args.tiles_per_sheet]), "columns": args.columns, "thumb_size": args.thumb_size, } for start in range(0, len(rendered_cards), args.tiles_per_sheet) ] 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, "requested_sample_slugs": sorted(set(args.sample_slug)), "columns": args.columns, "thumb_size": args.thumb_size, "tiles_per_sheet": args.tiles_per_sheet, "selected_tile_count": len(selected_tiles), "selected_sample_count": len( {str(tile.get("sample_slug") or "unknown") for tile in selected_tiles} ), "selected_sample_slugs": sorted( {str(tile.get("sample_slug") or "unknown") for tile in 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, "relationship_highlight_count": relationship_highlight_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"- selected source samples: {report['selected_sample_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']}", f"- relationship-highlighted labels: {report['relationship_highlight_count']}", "", "## Contact Sheets", "", ] if report["contact_sheets"]: for sheet in report["contact_sheets"]: lines.append(f"- `{sheet['path']}` ({sheet['tile_count']} tiles)") lines.append("") 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) for page_index, start in enumerate( range(0, len(cards), args.tiles_per_sheet), start=1 ): page_cards = cards[start : start + args.tiles_per_sheet] page_name = CONTACT_SHEET_NAME_TEMPLATE.format(index=page_index) build_contact_sheet(page_cards, args.columns, output_dir / page_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())