#!/usr/bin/env python3 from __future__ import annotations import argparse import json import statistics from collections import Counter from pathlib import Path from typing import Any JSON_NAME = "operator_yolo_dataset_quality_audit.json" MARKDOWN_NAME = "operator_yolo_dataset_quality_audit.md" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Audit an operator YOLO tile dataset summary for sample coverage, " "negative balance and label-quality risks." ) ) 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 and Markdown reports") parser.add_argument("--min-positive-samples", type=int, default=6) parser.add_argument("--min-val-positive-samples", type=int, default=2) parser.add_argument("--max-repeated-negative-share", type=float, default=0.65) parser.add_argument("--min-median-box-area", type=float, default=0.001) parser.add_argument("--small-box-area-threshold", type=float, default=0.0005) parser.add_argument("--max-small-box-share", type=float, default=0.5) 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]: if path is None or not path.exists(): return [], 0 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( { "center_x": center_x, "center_y": center_y, "width": width, "height": height, "area": width * height, } ) return boxes, invalid_count def add_warning(warnings: list[dict[str, str]], code: str, message: str) -> None: warnings.append({"code": code, "message": message}) 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 safe_percentile(values: list[float], fraction: float) -> float | None: if not values: return None ordered = sorted(values) return round(float(ordered[round((len(ordered) - 1) * fraction)]), 12) def summarize_label_files( label_file_paths: set[Path], small_box_area_threshold: float, ) -> dict[str, Any]: boxes: list[dict[str, float]] = [] invalid_count = 0 missing_file_count = 0 for label_file in sorted(label_file_paths): if not label_file.exists(): missing_file_count += 1 continue parsed_boxes, invalid = parse_yolo_label_file(label_file) boxes.extend(parsed_boxes) invalid_count += invalid areas = [box["area"] for box in boxes] widths = [box["width"] for box in boxes] heights = [box["height"] for box in boxes] aspect_ratios = [max(box["width"] / box["height"], box["height"] / box["width"]) for box in boxes] small_box_count = sum(1 for area in areas if area < small_box_area_threshold) return { "label_file_count": len(label_file_paths), "missing_label_file_count": missing_file_count, "parsed_label_count": len(boxes), "invalid_label_count": invalid_count, "min_box_area": min(areas) if areas else None, "median_box_area": safe_median(areas), "mean_box_area": safe_mean(areas), "median_box_width": safe_median(widths), "median_box_height": safe_median(heights), "box_area_p99": safe_percentile(areas, 0.99), "box_area_max": max(areas) if areas else None, "aspect_ratio_median": safe_median(aspect_ratios), "aspect_ratio_p99": safe_percentile(aspect_ratios, 0.99), "aspect_ratio_max": max(aspect_ratios) if aspect_ratios else None, "small_box_area_threshold": small_box_area_threshold, "small_box_count": small_box_count, "small_box_share": small_box_count / len(areas) if areas else 0.0, } 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") sample = samples.setdefault( slug, { "sample_slug": slug, "sample_role": tile.get("sample_role") or "unknown", "splits": set(), "tile_count": 0, "positive_tile_count": 0, "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") sample["tile_count"] += 1 label_count = int(tile.get("label_count") or 0) sample["label_count"] += label_count is_negative = bool(tile.get("is_negative")) or label_count == 0 if is_negative: sample["negative_tile_count"] += 1 else: 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"])) 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, 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) negative_tiles = [ tile for tile in tiles if bool(tile.get("is_negative")) or int(tile.get("label_count") or 0) == 0 ] positive_tiles = [tile for tile in tiles if tile not in negative_tiles] train_negative_tiles = [tile for tile in negative_tiles if tile.get("split") == "train"] low_variance_positive_tiles = [tile for tile in positive_tiles if tile.get("low_visual_variance")] val_positive_samples = { str(tile.get("sample_slug")) for tile in positive_tiles if tile.get("split") == "val" and tile.get("sample_slug") } positive_samples = { sample["sample_slug"] for sample in sample_summaries if int(sample["positive_tile_count"]) > 0 } background_samples = { sample["sample_slug"] for sample in sample_summaries if sample["sample_role"] == "background_candidate" } repeated_negative_count = sum(1 for tile in negative_tiles if tile.get("is_repeated_background_negative")) repeated_negative_share = repeated_negative_count / len(negative_tiles) if negative_tiles else 0.0 label_stats = summarize_labels(tiles, summary_path, args.small_box_area_threshold) warnings: list[dict[str, str]] = [] if len(positive_samples) < args.min_positive_samples: add_warning( warnings, "positive_sample_count_below_gate", f"Only {len(positive_samples)} positive samples; gate is {args.min_positive_samples}.", ) if len(val_positive_samples) < args.min_val_positive_samples: add_warning( warnings, "val_positive_sample_count_below_gate", f"Only {len(val_positive_samples)} validation positive samples; gate is {args.min_val_positive_samples}.", ) if repeated_negative_share > args.max_repeated_negative_share: add_warning( warnings, "repeated_background_negative_share_above_gate", ( f"Repeated background negatives are {repeated_negative_share:.3f} of negative tiles; " f"gate is {args.max_repeated_negative_share:.3f}." ), ) if low_variance_positive_tiles: add_warning( warnings, "positive_tiles_have_low_visual_variance", ( f"{len(low_variance_positive_tiles)} positive tiles are visually blank/low-variance; " "the imagery source does not support their labels." ), ) if label_stats["missing_label_file_count"]: add_warning( warnings, "label_files_missing", f"{label_stats['missing_label_file_count']} referenced label files could not be read.", ) if label_stats["invalid_label_count"]: add_warning( warnings, "invalid_label_rows", f"{label_stats['invalid_label_count']} YOLO label rows are malformed or out of range.", ) median_box_area = label_stats["median_box_area"] if median_box_area is not None and median_box_area < args.min_median_box_area: add_warning( warnings, "median_box_area_below_gate", f"Median normalized box area {median_box_area:.6f} is below gate {args.min_median_box_area:.6f}.", ) if label_stats["small_box_share"] > args.max_small_box_share: add_warning( warnings, "small_box_share_above_gate", ( f"Small boxes are {label_stats['small_box_share']:.3f} of parsed labels; " f"gate is {args.max_small_box_share:.3f}." ), ) recommendations = build_recommendations(warnings) return { "status": "needs_attention" if warnings else "ok", "summary_path": str(summary_path), "dataset_yaml": summary.get("dataset_yaml"), "output_dir": summary.get("output_dir"), "class_names": summary.get("class_names", []), "tile_size": summary.get("tile_size"), "stride": summary.get("stride"), "negative_keep_ratio": summary.get("negative_keep_ratio"), "background_negative_repeat": summary.get("background_negative_repeat"), "min_label_px": summary.get("min_label_px"), "min_label_visible_ratio": summary.get("min_label_visible_ratio"), "tile_count": len(tiles), "positive_tile_count": len(positive_tiles), "negative_tile_count": len(negative_tiles), "train_tile_count": split_counts.get("train", 0), "val_tile_count": split_counts.get("val", 0), "train_negative_tile_count": len(train_negative_tiles), "low_variance_positive_tile_count": len(low_variance_positive_tiles), "repeated_background_negative_tile_count": repeated_negative_count, "repeated_background_negative_share_of_negatives": repeated_negative_share, "positive_tile_share": len(positive_tiles) / len(tiles) if tiles else 0.0, "negative_tile_share": len(negative_tiles) / len(tiles) if tiles else 0.0, "label_count": sum(int(tile.get("label_count") or 0) for tile in tiles), "sample_count": len(sample_summaries), "positive_sample_count": len(positive_samples), "background_sample_count": len(background_samples), "split_counts": dict(sorted(split_counts.items())), "tile_role_counts": dict(sorted(role_counts.items())), "label_stats": label_stats, "sample_summaries": sample_summaries, "warnings": warnings, "recommendations": recommendations, } def build_recommendations(warnings: list[dict[str, str]]) -> list[str]: codes = {warning["code"] for warning in warnings} recommendations: list[str] = [] if "positive_sample_count_below_gate" in codes or "val_positive_sample_count_below_gate" in codes: recommendations.append( "Add more labeled positive AOIs before extending training duration or increasing model size." ) if "repeated_background_negative_share_above_gate" in codes: recommendations.append( "Reduce background repeat pressure or add more unique hard-negative AOIs to avoid overfitting." ) if "median_box_area_below_gate" in codes or "small_box_share_above_gate" in codes: recommendations.append( "Inspect clipped building labels visually; very small boxes may indicate tile size or label clipping issues." ) if "label_files_missing" in codes or "invalid_label_rows" in codes: recommendations.append("Regenerate the YOLO tile dataset and review exporter path/label integrity.") if "positive_tiles_have_low_visual_variance" in codes: recommendations.append( "Reject the raster product for affected AOIs or replace it with an officially complete imagery edition before training." ) if not recommendations: recommendations.append("Dataset audit passed the configured gates; continue with benchmarked training.") return recommendations def write_markdown(report: dict[str, Any], path: Path) -> None: warnings = report["warnings"] recommendations = report["recommendations"] label_stats = report["label_stats"] lines = [ "# Operator YOLO Dataset Quality Audit", "", f"- Status: `{report['status']}`", f"- Summary: `{report['summary_path']}`", f"- Tiles: {report['tile_count']} ({report['positive_tile_count']} positive, {report['negative_tile_count']} negative)", f"- Samples: {report['sample_count']} ({report['positive_sample_count']} positive, {report['background_sample_count']} background)", f"- Repeated background negative share: {report['repeated_background_negative_share_of_negatives']:.3f}", f"- Minimum visible label ratio: {format_optional_float(report.get('min_label_visible_ratio'))}", f"- Blank/low-variance positive tiles: {report['low_variance_positive_tile_count']}", "", "## Label Quality", "", f"- Parsed labels: {label_stats['parsed_label_count']}", f"- Invalid label rows: {label_stats['invalid_label_count']}", f"- Missing label files: {label_stats['missing_label_file_count']}", f"- Median normalized box area: {format_optional_float(label_stats['median_box_area'])}", f"- Small-box share: {label_stats['small_box_share']:.3f}", "", "## Warnings", "", ] if warnings: for warning in warnings: lines.append(f"- `{warning['code']}`: {warning['message']}") else: lines.append("- None") lines.extend(["", "## Recommendations", ""]) for recommendation in recommendations: lines.append(f"- {recommendation}") 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"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") def format_optional_float(value: float | None) -> str: return "n/a" if value is None else f"{value:.6f}" 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 = build_audit(summary, summary_path, args) json_path = output_dir / JSON_NAME markdown_path = output_dir / MARKDOWN_NAME json_path.write_text(json.dumps(report, indent=2, sort_keys=True), encoding="utf-8") write_markdown(report, markdown_path) print("Operator YOLO dataset quality audit passed") print(f"Status: {report['status']}") print(f"JSON: {json_path}") print(f"Markdown: {markdown_path}") return 0 if __name__ == "__main__": raise SystemExit(main())