Add per-sample YOLO dataset audit diagnostics

scripts/README.md

@@ -413,8 +413,9 @@ The audit reads the tile summary and YOLO label files, then writes
 `operator_yolo_dataset_quality_audit.json` and
 `operator_yolo_dataset_quality_audit.md`. It reports positive/background sample
 coverage, train/validation split coverage, repeated hard-negative pressure,
-minimum visible label ratio, missing or invalid label rows and normalized
-box-area signals. Treat
+minimum visible label ratio, missing or invalid label rows, normalized
+box-area signals and per-sample label diagnostics such as parsed label count,
+median box area, small-box share and sample-specific quality warnings. Treat
 `needs_attention` as a dataset-design warning, not as a runtime failure: the
 next action is usually more positive AOIs, better validation coverage or more
 unique hard negatives rather than simply extending epochs.

scripts/audit_operator_yolo_dataset_quality.py

@@ -105,17 +105,18 @@ def add_warning(warnings: list[dict[str, str]], code: str, message: str) -> None
     warnings.append({"code": code, "message": message})
 
 
-def summarize_labels(
-    tiles: list[dict[str, Any]],
-    summary_path: Path,
+def safe_mean(values: list[float]) -> float | None:
+    return round(statistics.fmean(values), 12) if values else None
+
+
+def safe_median(values: list[float]) -> float | None:
+    return round(float(statistics.median(values)), 12) if values else None
+
+
+def summarize_label_files(
+    label_file_paths: set[Path],
     small_box_area_threshold: float,
 ) -> dict[str, Any]:
-    label_file_paths: set[Path] = set()
-    for tile in tiles:
-        resolved = resolve_path(tile.get("label_path"), summary_path)
-        if resolved is not None:
-            label_file_paths.add(resolved)
-
     boxes: list[dict[str, float]] = []
     invalid_count = 0
     missing_file_count = 0
@@ -132,12 +133,6 @@ def summarize_labels(
     heights = [box["height"] for box in boxes]
     small_box_count = sum(1 for area in areas if area < small_box_area_threshold)
 
-    def safe_mean(values: list[float]) -> float | None:
-        return statistics.fmean(values) if values else None
-
-    def safe_median(values: list[float]) -> float | None:
-        return statistics.median(values) if values else None
-
     return {
         "label_file_count": len(label_file_paths),
         "missing_label_file_count": missing_file_count,
@@ -154,7 +149,43 @@ def summarize_labels(
     }
 
 
-def build_sample_summaries(tiles: list[dict[str, Any]]) -> list[dict[str, Any]]:
+def build_label_quality_warning_codes(
+    label_stats: dict[str, Any],
+    args: argparse.Namespace,
+) -> list[str]:
+    warning_codes: list[str] = []
+    if label_stats["missing_label_file_count"]:
+        warning_codes.append("label_files_missing")
+    if label_stats["invalid_label_count"]:
+        warning_codes.append("invalid_label_rows")
+
+    median_box_area = label_stats["median_box_area"]
+    if median_box_area is not None and median_box_area < args.min_median_box_area:
+        warning_codes.append("median_box_area_below_gate")
+    if label_stats["parsed_label_count"] and label_stats["small_box_share"] > args.max_small_box_share:
+        warning_codes.append("small_box_share_above_gate")
+    return warning_codes
+
+
+def summarize_labels(
+    tiles: list[dict[str, Any]],
+    summary_path: Path,
+    small_box_area_threshold: float,
+) -> dict[str, Any]:
+    label_file_paths: set[Path] = set()
+    for tile in tiles:
+        resolved = resolve_path(tile.get("label_path"), summary_path)
+        if resolved is not None:
+            label_file_paths.add(resolved)
+
+    return summarize_label_files(label_file_paths, small_box_area_threshold)
+
+
+def build_sample_summaries(
+    tiles: list[dict[str, Any]],
+    summary_path: Path,
+    args: argparse.Namespace,
+) -> list[dict[str, Any]]:
     samples: dict[str, dict[str, Any]] = {}
     for tile in tiles:
         slug = str(tile.get("sample_slug") or "unknown")
@@ -169,6 +200,7 @@ def build_sample_summaries(tiles: list[dict[str, Any]]) -> list[dict[str, Any]]:
                 "negative_tile_count": 0,
                 "repeated_background_negative_tile_count": 0,
                 "label_count": 0,
+                "_label_file_paths": set(),
             },
         )
         sample["splits"].add(tile.get("split") or "unknown")
@@ -182,13 +214,20 @@ def build_sample_summaries(tiles: list[dict[str, Any]]) -> list[dict[str, Any]]:
             sample["positive_tile_count"] += 1
         if tile.get("is_repeated_background_negative"):
             sample["repeated_background_negative_tile_count"] += 1
+        resolved_label_path = resolve_path(tile.get("label_path"), summary_path)
+        if resolved_label_path is not None:
+            sample["_label_file_paths"].add(resolved_label_path)
 
     result: list[dict[str, Any]] = []
     for sample in samples.values():
+        label_stats = summarize_label_files(sample.pop("_label_file_paths"), args.small_box_area_threshold)
+        quality_warnings = build_label_quality_warning_codes(label_stats, args)
         result.append(
             {
                 **sample,
                 "splits": sorted(sample["splits"]),
+                **label_stats,
+                "quality_warnings": quality_warnings,
             }
         )
     return sorted(result, key=lambda item: str(item["sample_slug"]))
@@ -196,7 +235,7 @@ def build_sample_summaries(tiles: list[dict[str, Any]]) -> list[dict[str, Any]]:
 
 def build_audit(summary: dict[str, Any], summary_path: Path, args: argparse.Namespace) -> dict[str, Any]:
     tiles = [tile for tile in summary.get("tiles", []) if isinstance(tile, dict) and tile.get("kept", True)]
-    sample_summaries = build_sample_summaries(tiles)
+    sample_summaries = build_sample_summaries(tiles, summary_path, args)
 
     split_counts = Counter(str(tile.get("split") or "unknown") for tile in tiles)
     role_counts = Counter(str(tile.get("sample_role") or "unknown") for tile in tiles)
@@ -374,11 +413,16 @@ def write_markdown(report: dict[str, Any], path: Path) -> None:
 
     lines.extend(["", "## Sample Summary", ""])
     for sample in report["sample_summaries"]:
+        quality_warnings = sample.get("quality_warnings") or []
+        warning_text = ",".join(quality_warnings) if quality_warnings else "none"
         lines.append(
             "- "
             f"{sample['sample_slug']} ({sample['sample_role']}, {','.join(sample['splits'])}): "
             f"{sample['tile_count']} tiles, {sample['positive_tile_count']} positive, "
-            f"{sample['negative_tile_count']} negative, {sample['label_count']} labels"
+            f"{sample['negative_tile_count']} negative, {sample['label_count']} labels, "
+            f"parsed labels {sample['parsed_label_count']}, "
+            f"median box area {format_optional_float(sample['median_box_area'])}, "
+            f"small-box share {sample['small_box_share']:.3f}, warnings {warning_text}"
         )
 
     path.write_text("\n".join(lines) + "\n", encoding="utf-8")