audit nested YOLO labels without destructive rewrites

scripts/README.md

@@ -11,6 +11,12 @@ reviews with `--tiles-per-sheet` (default `64`). This keeps large corpora
 inspectable while preserving deterministic tile selection, ordering, label
 accounting and stable `contact_sheet_001.png` naming for the first page.
 
+`audit_yolo_label_relationships.py` performs a read-only, same-class audit of
+exact duplicate, high-IoU and possible-containment pairs inside YOLO label
+files. Its output retains the flagged source tiles so it can be passed directly
+to the contact-sheet renderer. Possible nesting remains review evidence and is
+never treated as an automatic label error or rewrite instruction.
+
 Setup-, import-, demo- en maintenance-scripts voor GeoIntel.
 
 ## WALOUS source provisioning

scripts/audit_yolo_label_relationships.py

@@ -0,0 +1,273 @@
+#!/usr/bin/env python3
+"""Audit same-class relationships inside YOLO label files without rewriting labels."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+
+@dataclass(frozen=True)
+class Box:
+    class_id: int
+    center_x: float
+    center_y: float
+    width: float
+    height: float
+
+    @property
+    def coordinates(self) -> tuple[float, float, float, float]:
+        return (
+            self.center_x - self.width / 2,
+            self.center_y - self.height / 2,
+            self.center_x + self.width / 2,
+            self.center_y + self.height / 2,
+        )
+
+    @property
+    def area(self) -> float:
+        return self.width * self.height
+
+    def as_list(self) -> list[float | int]:
+        return [self.class_id, self.center_x, self.center_y, self.width, self.height]
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def resolve_path(raw_path: str, summary_path: Path) -> Path:
+    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
+    relative_candidate = summary_path.parent / raw_path
+    if relative_candidate.exists():
+        return relative_candidate
+    return candidate
+
+
+def parse_label_file(path: Path) -> list[Box]:
+    boxes: list[Box] = []
+    for line_number, line in enumerate(
+        path.read_text(encoding="utf-8").splitlines(), start=1
+    ):
+        stripped = line.strip()
+        if not stripped:
+            continue
+        parts = stripped.split()
+        if len(parts) != 5:
+            raise ValueError(f"invalid YOLO row at {path}:{line_number}")
+        class_id = int(float(parts[0]))
+        center_x, center_y, width, height = map(float, parts[1:])
+        if not (
+            0 <= center_x <= 1
+            and 0 <= center_y <= 1
+            and 0 < width <= 1
+            and 0 < height <= 1
+        ):
+            raise ValueError(f"out-of-range YOLO row at {path}:{line_number}")
+        boxes.append(Box(class_id, center_x, center_y, width, height))
+    return boxes
+
+
+def intersection_area(first: Box, second: Box) -> float:
+    first_box = first.coordinates
+    second_box = second.coordinates
+    width = max(
+        0.0, min(first_box[2], second_box[2]) - max(first_box[0], second_box[0])
+    )
+    height = max(
+        0.0, min(first_box[3], second_box[3]) - max(first_box[1], second_box[1])
+    )
+    return width * height
+
+
+def classify_pair(
+    first: Box,
+    second: Box,
+    *,
+    near_duplicate_iou: float,
+    containment_threshold: float,
+    max_nested_area_ratio: float,
+) -> tuple[str, float] | None:
+    if first.class_id != second.class_id:
+        return None
+    if first == second:
+        return "exact_duplicate", 1.0
+    intersection = intersection_area(first, second)
+    if intersection <= 0:
+        return None
+    union = first.area + second.area - intersection
+    iou = min(1.0, intersection / union)
+    if iou >= near_duplicate_iou:
+        return "near_duplicate", iou
+    minimum_area = min(first.area, second.area)
+    area_ratio = max(first.area, second.area) / minimum_area
+    containment = min(1.0, intersection / minimum_area)
+    if containment >= containment_threshold and area_ratio <= max_nested_area_ratio:
+        return "possible_nested", containment
+    return None
+
+
+def audit_boxes(
+    boxes: list[Box],
+    *,
+    near_duplicate_iou: float,
+    containment_threshold: float,
+    max_nested_area_ratio: float,
+) -> list[dict[str, Any]]:
+    relationships: list[dict[str, Any]] = []
+    for first_index, first in enumerate(boxes):
+        for second_index in range(first_index + 1, len(boxes)):
+            second = boxes[second_index]
+            classification = classify_pair(
+                first,
+                second,
+                near_duplicate_iou=near_duplicate_iou,
+                containment_threshold=containment_threshold,
+                max_nested_area_ratio=max_nested_area_ratio,
+            )
+            if classification is None:
+                continue
+            relationship, score = classification
+            relationships.append(
+                {
+                    "relationship": relationship,
+                    "score": round(score, 12),
+                    "first_index": first_index,
+                    "second_index": second_index,
+                    "first_box": first.as_list(),
+                    "second_box": second.as_list(),
+                }
+            )
+    return relationships
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--summary-path", required=True, type=Path)
+    parser.add_argument("--output", required=True, type=Path)
+    parser.add_argument("--near-duplicate-iou", type=float, default=0.90)
+    parser.add_argument("--containment-threshold", type=float, default=0.98)
+    parser.add_argument("--max-nested-area-ratio", type=float, default=4.0)
+    args = parser.parse_args()
+
+    if args.output.exists():
+        parser.error(f"output already exists: {args.output}")
+    if not 0 < args.near_duplicate_iou <= 1:
+        parser.error("--near-duplicate-iou must be in (0, 1]")
+    if not 0 < args.containment_threshold <= 1:
+        parser.error("--containment-threshold must be in (0, 1]")
+    if args.max_nested_area_ratio < 1:
+        parser.error("--max-nested-area-ratio must be at least one")
+
+    summary_path = args.summary_path.expanduser().resolve(strict=True)
+    summary = json.loads(summary_path.read_text(encoding="utf-8"))
+    tiles = summary.get("tiles")
+    if not isinstance(tiles, list):
+        raise ValueError("dataset summary must contain a tiles list")
+
+    totals = {"exact_duplicate": 0, "near_duplicate": 0, "possible_nested": 0}
+    reviewed_label_count = 0
+    flagged_tiles: list[dict[str, Any]] = []
+    renderable_tiles: list[dict[str, Any]] = []
+    for tile in tiles:
+        if not isinstance(tile, dict) or not tile.get("kept", True):
+            continue
+        label_path = resolve_path(str(tile.get("label_path") or ""), summary_path)
+        boxes = parse_label_file(label_path)
+        reviewed_label_count += len(boxes)
+        relationships = audit_boxes(
+            boxes,
+            near_duplicate_iou=args.near_duplicate_iou,
+            containment_threshold=args.containment_threshold,
+            max_nested_area_ratio=args.max_nested_area_ratio,
+        )
+        if not relationships:
+            continue
+        counts = {key: 0 for key in totals}
+        for relationship in relationships:
+            key = relationship["relationship"]
+            counts[key] += 1
+            totals[key] += 1
+        flagged_tiles.append(
+            {
+                "sample_slug": str(tile.get("sample_slug") or "unknown"),
+                "split": str(tile.get("split") or "unknown"),
+                "tile_index": int(tile.get("tile_index") or 0),
+                "label_path": str(label_path),
+                "label_count": len(boxes),
+                "relationship_counts": counts,
+                "relationships": relationships,
+            }
+        )
+        renderable_tiles.append(tile)
+
+    flagged_tiles.sort(
+        key=lambda item: (
+            -sum(item["relationship_counts"].values()),
+            item["sample_slug"],
+            item["split"],
+            item["tile_index"],
+        )
+    )
+    payload = {
+        "schema_version": 1,
+        "generated_at": datetime.now(UTC).isoformat(),
+        "status": "attention" if flagged_tiles else "ok",
+        "claim_boundary": "Read-only label relationship audit; possible nesting is not an automatic error decision.",
+        "summary_path": str(summary_path),
+        "summary_sha256": sha256_file(summary_path),
+        "output_dir": summary.get("output_dir"),
+        "class_names": summary.get("class_names", []),
+        "thresholds": {
+            "near_duplicate_iou": args.near_duplicate_iou,
+            "containment_threshold": args.containment_threshold,
+            "max_nested_area_ratio": args.max_nested_area_ratio,
+        },
+        "reviewed_tile_count": sum(
+            1 for tile in tiles if isinstance(tile, dict) and tile.get("kept", True)
+        ),
+        "reviewed_label_count": reviewed_label_count,
+        "flagged_tile_count": len(flagged_tiles),
+        "relationship_totals": totals,
+        "flagged_tiles": flagged_tiles,
+        "tiles": renderable_tiles,
+    }
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(
+        json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    )
+    print(
+        json.dumps(
+            {
+                key: payload[key]
+                for key in (
+                    "status",
+                    "reviewed_tile_count",
+                    "reviewed_label_count",
+                    "flagged_tile_count",
+                    "relationship_totals",
+                )
+            },
+            indent=2,
+        )
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/render_operator_yolo_label_qa_contact_sheets.py

@@ -26,9 +26,17 @@ 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(
+        "--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",
@@ -36,7 +44,12 @@ def parse_args() -> argparse.Namespace:
         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(
+        "--thumb-size",
+        type=int,
+        default=256,
+        help="Rendered tile thumbnail size in pixels",
+    )
     parser.add_argument(
         "--tiles-per-sheet",
         type=int,
@@ -60,14 +73,18 @@ def load_json(path: Path) -> dict[str, Any]:
     return data
 
 
-def filter_tiles_by_samples(tiles: list[dict[str, Any]], sample_slugs: list[str]) -> list[dict[str, Any]]:
+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)}")
+        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]
 
 
@@ -95,7 +112,9 @@ def resolve_path(raw_path: str | None, summary_path: Path) -> Path | None:
     return candidate
 
 
-def parse_yolo_label_file(path: Path | None) -> tuple[list[dict[str, float]], int, bool]:
+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
 
@@ -118,7 +137,12 @@ def parse_yolo_label_file(path: Path | None) -> tuple[list[dict[str, float]], in
         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):
+        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(
@@ -143,7 +167,9 @@ def tile_sort_key(tile: dict[str, Any]) -> tuple[int, str, str, int, int]:
     )
 
 
-def balanced_tiles_by_sample(tiles: list[dict[str, Any]], limit: int) -> list[dict[str, Any]]:
+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]]] = {}
@@ -174,23 +200,43 @@ def select_tiles(tiles: list[dict[str, Any]], max_tiles: int) -> list[dict[str,
     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)]
+    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)],
+        [
+            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],
+        [
+            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
+    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]
+        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)
@@ -206,17 +252,25 @@ def draw_tile_card(
     low_visual_variance: bool,
 ) -> Image.Image:
     header_height = 44
-    card = Image.new("RGB", (thumb_size, thumb_size + header_height), color=(245, 247, 250))
+    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)
+    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")]
+    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:
@@ -248,7 +302,9 @@ def image_has_low_visual_variance(image_path: Path, blank_range_threshold: int)
     return (max_value - min_value) <= blank_range_threshold
 
 
-def build_contact_sheet(cards: list[Image.Image], columns: int, output_path: Path) -> None:
+def build_contact_sheet(
+    cards: list[Image.Image], columns: int, output_path: Path
+) -> None:
     if not cards:
         return
     if columns <= 0:
@@ -272,7 +328,9 @@ def build_contact_sheet(cards: list[Image.Image], columns: int, output_path: Pat
     sheet.save(output_path)
 
 
-def build_report(summary: dict[str, Any], summary_path: Path, args: argparse.Namespace) -> tuple[dict[str, Any], list[Image.Image]]:
+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")
@@ -290,7 +348,9 @@ def build_report(summary: dict[str, Any], summary_path: Path, args: argparse.Nam
     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)
+        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:
@@ -301,7 +361,9 @@ def build_report(summary: dict[str, Any], summary_path: Path, args: argparse.Nam
         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)
+            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(
@@ -339,7 +401,9 @@ def build_report(summary: dict[str, Any], summary_path: Path, args: argparse.Nam
         raise ValueError("tiles_per_sheet must be positive")
     contact_sheets = [
         {
-            "path": CONTACT_SHEET_NAME_TEMPLATE.format(index=(start // args.tiles_per_sheet) + 1),
+            "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,
@@ -432,12 +496,16 @@ def main() -> int:
 
     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):
+    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")
+    (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")