Add operator YOLO dataset quality audit

CHANGELOG.md

@@ -1409,3 +1409,7 @@ Added:
 - Prevented multi-model portfolios from overwriting runs that share the same threshold.
 - Added regression coverage proving same-threshold runs are preserved in the assembled portfolio.
 - No API contracts, migrations, model downloads, provider fetching or AI inference behavior changed.
+# Unreleased
+
+- Added `scripts/audit_operator_yolo_dataset_quality.py`, an operator-only YOLO tile dataset quality audit that produces JSON and Markdown reports for sample coverage, split coverage, repeated hard-negative pressure and label-size integrity before further training runs.
+- Added pytest coverage and readiness syntax checking for the new operator YOLO dataset audit script.

backend/tests/test_sprint146_operator_yolo_dataset_quality_audit.py

@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def test_operator_yolo_dataset_quality_audit_reports_dataset_risks(tmp_path: Path) -> None:
+    script_path = ROOT / "scripts" / "audit_operator_yolo_dataset_quality.py"
+    assert script_path.exists()
+
+    dataset_dir = tmp_path / "yolo-dataset"
+    labels_train = dataset_dir / "labels" / "train"
+    labels_val = dataset_dir / "labels" / "val"
+    labels_train.mkdir(parents=True)
+    labels_val.mkdir(parents=True)
+
+    (labels_train / "geel_000.txt").write_text(
+        "0 0.500000 0.500000 0.010000 0.010000\n"
+        "0 0.250000 0.250000 0.100000 0.100000\n",
+        encoding="utf-8",
+    )
+    (labels_train / "postel_bos_000.txt").write_text("", encoding="utf-8")
+    (labels_train / "postel_bos_000_hn01.txt").write_text("", encoding="utf-8")
+    (labels_val / "turnhout_000.txt").write_text(
+        "0 0.600000 0.600000 0.080000 0.080000\n",
+        encoding="utf-8",
+    )
+
+    summary_path = dataset_dir / "yolo_tile_dataset_summary.json"
+    summary_path.write_text(
+        json.dumps(
+            {
+                "status": "ready",
+                "dataset_yaml": str(dataset_dir / "dataset.yaml"),
+                "output_dir": str(dataset_dir),
+                "class_names": ["building"],
+                "tile_size": 160,
+                "stride": 80,
+                "negative_keep_ratio": 1.0,
+                "background_negative_repeat": 2,
+                "min_label_px": 2,
+                "source_sample_count": 3,
+                "tile_count": 4,
+                "positive_tile_count": 2,
+                "negative_tile_count": 2,
+                "skipped_negative_tile_count": 0,
+                "label_count": 3,
+                "train_tile_count": 3,
+                "val_tile_count": 1,
+                "tiles": [
+                    {
+                        "sample_slug": "geel",
+                        "sample_role": "reference",
+                        "split": "train",
+                        "tile_index": 0,
+                        "repeat_index": 0,
+                        "kept": True,
+                        "label_path": str(labels_train / "geel_000.txt"),
+                        "label_count": 2,
+                        "is_negative": False,
+                        "is_repeated_background_negative": False,
+                    },
+                    {
+                        "sample_slug": "postel_bos",
+                        "sample_role": "background_candidate",
+                        "split": "train",
+                        "tile_index": 1,
+                        "repeat_index": 0,
+                        "kept": True,
+                        "label_path": str(labels_train / "postel_bos_000.txt"),
+                        "label_count": 0,
+                        "is_negative": True,
+                        "is_repeated_background_negative": False,
+                    },
+                    {
+                        "sample_slug": "postel_bos",
+                        "sample_role": "background_candidate",
+                        "split": "train",
+                        "tile_index": 1,
+                        "repeat_index": 1,
+                        "kept": True,
+                        "label_path": str(labels_train / "postel_bos_000_hn01.txt"),
+                        "label_count": 0,
+                        "is_negative": True,
+                        "is_repeated_background_negative": True,
+                    },
+                    {
+                        "sample_slug": "turnhout",
+                        "sample_role": "reference",
+                        "split": "val",
+                        "tile_index": 2,
+                        "repeat_index": 0,
+                        "kept": True,
+                        "label_path": str(labels_val / "turnhout_000.txt"),
+                        "label_count": 1,
+                        "is_negative": False,
+                        "is_repeated_background_negative": False,
+                    },
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    output_dir = tmp_path / "audit"
+    result = subprocess.run(
+        [
+            sys.executable,
+            str(script_path),
+            "--summary-path",
+            str(summary_path),
+            "--output-dir",
+            str(output_dir),
+            "--min-positive-samples",
+            "3",
+            "--min-val-positive-samples",
+            "2",
+            "--max-repeated-negative-share",
+            "0.25",
+            "--min-median-box-area",
+            "0.02",
+            "--max-small-box-share",
+            "0.25",
+        ],
+        cwd=ROOT,
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+
+    assert "Operator YOLO dataset quality audit passed" in result.stdout
+
+    report = json.loads(
+        (output_dir / "operator_yolo_dataset_quality_audit.json").read_text(encoding="utf-8")
+    )
+    assert report["status"] == "needs_attention"
+    assert report["sample_count"] == 3
+    assert report["positive_sample_count"] == 2
+    assert report["background_sample_count"] == 1
+    assert report["train_negative_tile_count"] == 2
+    assert report["repeated_background_negative_tile_count"] == 1
+    assert report["label_stats"]["parsed_label_count"] == 3
+    assert report["label_stats"]["invalid_label_count"] == 0
+
+    warning_codes = {warning["code"] for warning in report["warnings"]}
+    assert "positive_sample_count_below_gate" in warning_codes
+    assert "val_positive_sample_count_below_gate" in warning_codes
+    assert "repeated_background_negative_share_above_gate" in warning_codes
+    assert "median_box_area_below_gate" in warning_codes
+    assert "small_box_share_above_gate" in warning_codes
+
+    markdown = (output_dir / "operator_yolo_dataset_quality_audit.md").read_text(encoding="utf-8")
+    assert "Operator YOLO Dataset Quality Audit" in markdown
+    assert "Label Quality" in markdown
+    assert "positive_sample_count_below_gate" in markdown

docs/CODEX_EXECUTION_LOG.md

@@ -5643,3 +5643,24 @@ Tested:
 
 Open:
 - The expanded160e50 model is consistently best on the current positive AOI portfolio, but hard-negative/background AOI evidence still prevents blind default promotion.
+# Sprint 146 - Operator YOLO dataset quality audit
+
+## What changed
+
+- Added `scripts/audit_operator_yolo_dataset_quality.py` to inspect generated operator YOLO tile datasets before further model training.
+- Added a focused pytest that creates a synthetic tile summary and YOLO label files, then verifies JSON/Markdown audit output and warning gates.
+- Added the audit script to the readiness syntax gate.
+- Documented the operator audit command in `scripts/README.md`.
+
+## What was tested
+
+- `python -m pytest backend\tests\test_sprint146_operator_yolo_dataset_quality_audit.py -q`
+
+## Known limitations
+
+- The audit is evidence tooling only. It does not modify datasets, train models, fetch external data or change active YOLO configuration.
+- The report flags likely dataset risks, but final promotion decisions must still come from persisted detection QA/QC matrices and hard-negative benchmarks.
+
+## Next recommended pass
+
+- Run the audit against the existing Tower tile datasets and use the results to decide the next training-data expansion pass.

docs/TODO.md

@@ -436,3 +436,9 @@ This file now starts with the current implementation status. Older preparation/b
 - [ ] Add more AOIs after the tile-level baseline so the next local model attempt is not limited to Geel/Mol/Turnhout.
 - [ ] Add negative/background AOIs so the next tile dataset is not all positive tiles.
 - [ ] Improve positive training coverage/label quality before the next higher-capacity model attempt; simply extending the same hardneg r8 run is not enough.
+# Sprint 146 - Operator YOLO dataset quality audit
+
+- [x] Add a dataset/label-quality audit for generated operator YOLO tile datasets.
+- [x] Report sample coverage, validation coverage, repeated hard-negative pressure and YOLO label area integrity.
+- [x] Wire the audit script into the readiness syntax gate.
+- [ ] Use live audit output to decide whether the next model pass needs more positive AOIs, label cleanup or unique hard negatives.

scripts/README.md

@@ -328,6 +328,23 @@ negative tiles, and records `yolo_tile_dataset_summary.json` with
 It remains operator tooling only: no provider fetch, no API mutation and no
 automatic model training.
 
+Audit the generated tile dataset before spending another long training run:
+
+```bash
+python scripts/audit_operator_yolo_dataset_quality.py \
+  --summary-path /mnt/user/appdata/geointel/storage/operator-data/yolo-building-tile-hardneg160r8/yolo_tile_dataset_summary.json \
+  --output-dir /mnt/user/appdata/geointel/artifacts/operator-yolo-dataset-audit/hardneg160r8
+```
+
+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,
+missing or invalid label rows and normalized box-area signals. 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.
+
 For hard-negative-balanced experiments, repeat only train-split negative tiles
 from samples marked `sample_role=background_candidate`:
 

scripts/audit_operator_yolo_dataset_quality.py

@@ -0,0 +1,411 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+import statistics
+from collections import Counter, defaultdict
+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 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)
+
+    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]
+    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,
+        "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),
+        "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_sample_summaries(tiles: list[dict[str, Any]]) -> 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,
+            },
+        )
+        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
+
+    result: list[dict[str, Any]] = []
+    for sample in samples.values():
+        result.append(
+            {
+                **sample,
+                "splits": sorted(sample["splits"]),
+            }
+        )
+    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)
+
+    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"]
+    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 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"),
+        "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),
+        "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 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}",
+        "",
+        "## 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"]:
+        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"
+        )
+
+    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())

scripts/run_readiness_check.sh

@@ -44,6 +44,7 @@ ${PYTHON_BIN} -m py_compile backend/scripts/yolo_preflight.py
 ${PYTHON_BIN} -m py_compile scripts/prepare_operator_real_data_samples.py
 ${PYTHON_BIN} -m py_compile scripts/export_operator_yolo_dataset.py
 ${PYTHON_BIN} -m py_compile scripts/export_operator_yolo_tile_dataset.py
+${PYTHON_BIN} -m py_compile scripts/audit_operator_yolo_dataset_quality.py
 ${PYTHON_BIN} -m py_compile scripts/build_detection_model_promotion_report.py
 ${PYTHON_BIN} -m py_compile scripts/cleanup_demo_artifacts.py
 ${PYTHON_BIN} -m py_compile backend/scripts/cleanup_demo_artifacts.py