218 lines
7.6 KiB
Python
218 lines
7.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Audit individual YOLO rows for geometric training risks without mutation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
try:
|
|
from scripts.audit_yolo_label_relationships import (
|
|
Box,
|
|
parse_label_file,
|
|
resolve_path,
|
|
sha256_file,
|
|
)
|
|
except ModuleNotFoundError: # Standalone operator-tool copy beside the auditor.
|
|
from audit_yolo_label_relationships import (
|
|
Box,
|
|
parse_label_file,
|
|
resolve_path,
|
|
sha256_file,
|
|
)
|
|
|
|
|
|
def classify_box(
|
|
box: Box,
|
|
*,
|
|
tile_size: int,
|
|
min_dimension_pixels: float,
|
|
extreme_aspect_ratio: float,
|
|
edge_tolerance_pixels: float,
|
|
) -> list[dict[str, Any]]:
|
|
width_px = box.width * tile_size
|
|
height_px = box.height * tile_size
|
|
aspect_ratio = max(width_px / height_px, height_px / width_px)
|
|
left, top, right, bottom = box.coordinates
|
|
edge_sides = [
|
|
side
|
|
for side, distance in (
|
|
("left", left * tile_size),
|
|
("top", top * tile_size),
|
|
("right", (1 - right) * tile_size),
|
|
("bottom", (1 - bottom) * tile_size),
|
|
)
|
|
if distance <= edge_tolerance_pixels
|
|
]
|
|
metrics = {
|
|
"width_px": round(width_px, 6),
|
|
"height_px": round(height_px, 6),
|
|
"area_px2": round(width_px * height_px, 6),
|
|
"aspect_ratio": round(aspect_ratio, 6),
|
|
}
|
|
outliers: list[dict[str, Any]] = []
|
|
if min(width_px, height_px) < min_dimension_pixels:
|
|
outliers.append({"category": "small_dimension", **metrics})
|
|
if aspect_ratio >= extreme_aspect_ratio:
|
|
outliers.append({"category": "extreme_aspect_ratio", **metrics})
|
|
if edge_sides:
|
|
outliers.append({"category": "tile_edge", "edge_sides": edge_sides, **metrics})
|
|
return outliers
|
|
|
|
|
|
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("--tile-size", type=int)
|
|
parser.add_argument("--min-dimension-pixels", type=float, default=4.0)
|
|
parser.add_argument("--extreme-aspect-ratio", type=float, default=8.0)
|
|
parser.add_argument("--edge-tolerance-pixels", type=float, default=0.5)
|
|
parser.add_argument(
|
|
"--category",
|
|
action="append",
|
|
choices=("small_dimension", "extreme_aspect_ratio", "tile_edge"),
|
|
default=[],
|
|
help="Limit output to one risk category; repeat to select multiple.",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
if args.output.exists():
|
|
parser.error(f"output already exists: {args.output}")
|
|
if args.min_dimension_pixels <= 0:
|
|
parser.error("--min-dimension-pixels must be positive")
|
|
if args.extreme_aspect_ratio < 1:
|
|
parser.error("--extreme-aspect-ratio must be at least one")
|
|
if args.edge_tolerance_pixels < 0:
|
|
parser.error("--edge-tolerance-pixels must be non-negative")
|
|
|
|
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")
|
|
tile_size = args.tile_size or summary.get("tile_size")
|
|
if not isinstance(tile_size, int) or tile_size <= 0:
|
|
raise ValueError("a positive integer tile size is required")
|
|
|
|
selected_categories = set(
|
|
args.category or ("small_dimension", "extreme_aspect_ratio", "tile_edge")
|
|
)
|
|
category_totals = {
|
|
"small_dimension": 0,
|
|
"extreme_aspect_ratio": 0,
|
|
"tile_edge": 0,
|
|
}
|
|
reviewed_label_count = 0
|
|
unique_flagged_rows: set[tuple[str, int]] = set()
|
|
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)
|
|
outliers: list[dict[str, Any]] = []
|
|
for index, box in enumerate(boxes):
|
|
classifications = classify_box(
|
|
box,
|
|
tile_size=tile_size,
|
|
min_dimension_pixels=args.min_dimension_pixels,
|
|
extreme_aspect_ratio=args.extreme_aspect_ratio,
|
|
edge_tolerance_pixels=args.edge_tolerance_pixels,
|
|
)
|
|
for classification in classifications:
|
|
if classification["category"] not in selected_categories:
|
|
continue
|
|
category_totals[classification["category"]] += 1
|
|
unique_flagged_rows.add((str(label_path), index))
|
|
outliers.append(
|
|
{
|
|
"index": index,
|
|
"box": box.as_list(),
|
|
**classification,
|
|
}
|
|
)
|
|
if not outliers:
|
|
continue
|
|
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),
|
|
"outlier_count": len(outliers),
|
|
"outliers": outliers,
|
|
}
|
|
)
|
|
renderable_tiles.append(tile)
|
|
|
|
flagged_tiles.sort(
|
|
key=lambda item: (
|
|
-item["outlier_count"],
|
|
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 geometric risk triage; a flagged label is not an automatic "
|
|
"ground-truth error or rewrite instruction."
|
|
),
|
|
"summary_path": str(summary_path),
|
|
"summary_sha256": sha256_file(summary_path),
|
|
"output_dir": summary.get("output_dir"),
|
|
"class_names": summary.get("class_names", []),
|
|
"tile_size": tile_size,
|
|
"thresholds": {
|
|
"min_dimension_pixels": args.min_dimension_pixels,
|
|
"extreme_aspect_ratio": args.extreme_aspect_ratio,
|
|
"edge_tolerance_pixels": args.edge_tolerance_pixels,
|
|
},
|
|
"selected_categories": sorted(selected_categories),
|
|
"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),
|
|
"unique_flagged_label_count": len(unique_flagged_rows),
|
|
"category_totals": category_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",
|
|
"unique_flagged_label_count",
|
|
"category_totals",
|
|
)
|
|
},
|
|
indent=2,
|
|
)
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|