geointel/scripts/export_operator_yolo_tile_dataset.py

"""Export operator real-data samples to a tile-level YOLO detection dataset.

This operator/runtime helper turns the prepared orthophoto + GRB reference
sample manifest into overlapping raster tiles with YOLO labels. It does not
call GeoIntel APIs, does not train automatically and does not fetch provider
data.
"""

from __future__ import annotations

import argparse
from dataclasses import dataclass
import hashlib
import json
import os
import shutil
import sys
from pathlib import Path
from typing import Any, Iterable

SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPT_DIR))

from training_dataset_eligibility import (  # noqa: E402
    TrainingEligibilityError,
    assert_frozen_manifest_training_eligible,
)
from training_release_manifest import (  # noqa: E402
    TrainingReleaseError,
    create_training_release_manifest,
)


DEFAULT_MANIFEST_PATH = Path("/app/storage/operator-data/operator_samples_manifest.json")
DEFAULT_OUTPUT_DIR = Path("/app/storage/operator-data/yolo-building-tile-dataset")
DEFAULT_CLASS_NAME = "building"
DEFAULT_REFERENCE_SOURCE = "grb"
DEFAULT_REFERENCE_LAYER = "buildings"
REFERENCE_AOI_CATEGORY = "reference_aoi"
PURE_EMPTY_BACKGROUND_CATEGORY = "pure_empty_negative"
SPARSE_BACKGROUND_CATEGORY = "sparse_building_context"
LOW_VARIANCE_NEGATIVE_SKIP_REASON = "low_visual_variance_negative"
DEFAULT_VALIDATION_SAMPLE_SLUGS = frozenset(
    {
        "turnhout",
        "retie",
        "westerlo",
        "arendonk_heide",
        "vosselaar_center",
        "grobbendonk_center",
    }
)
DEFAULT_VALIDATION_SAMPLES = ",".join(sorted(DEFAULT_VALIDATION_SAMPLE_SLUGS))
rasterio: Any = None
Window: Any = None
Transformer: Any = None
Image: Any = None


def file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


@dataclass(frozen=True)
class TileWindow:
    row_off: int
    col_off: int
    height: int
    width: int


@dataclass(frozen=True)
class PixelBox:
    min_col: float
    min_row: float
    max_col: float
    max_row: float


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Export operator real-data samples to a tile-level YOLO detection dataset.",
    )
    parser.add_argument(
        "--manifest-path",
        type=Path,
        default=Path(os.environ.get("OPERATOR_SAMPLE_MANIFEST_PATH", DEFAULT_MANIFEST_PATH)),
        help="operator_samples_manifest.json created by prepare_operator_real_data_samples.py.",
    )
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path(os.environ.get("OPERATOR_YOLO_TILE_DATASET_DIR", DEFAULT_OUTPUT_DIR)),
        help="Output directory for tile images, labels, dataset.yaml and summary JSON.",
    )
    parser.add_argument(
        "--class-name",
        default=os.environ.get("OPERATOR_YOLO_CLASS_NAME", DEFAULT_CLASS_NAME),
        help="Canonical single detection class written to dataset.yaml.",
    )
    parser.add_argument(
        "--reference-source",
        default=os.environ.get("OPERATOR_YOLO_REFERENCE_SOURCE", DEFAULT_REFERENCE_SOURCE),
        help="Required source_name in reference GeoJSON features.",
    )
    parser.add_argument(
        "--review-audit",
        type=Path,
        required=True,
        help="Passed corpus audit with accepted human-review evidence for this frozen source manifest.",
    )
    parser.add_argument(
        "--reference-layer",
        default=os.environ.get("OPERATOR_YOLO_REFERENCE_LAYER", DEFAULT_REFERENCE_LAYER),
        help="Required reference_layer_name in reference GeoJSON features.",
    )
    parser.add_argument("--tile-size", type=int, default=int(os.environ.get("OPERATOR_YOLO_TILE_SIZE", "256")))
    parser.add_argument("--stride", type=int, default=int(os.environ.get("OPERATOR_YOLO_TILE_STRIDE", "128")))
    parser.add_argument(
        "--samples",
        default=os.environ.get("OPERATOR_YOLO_SAMPLES", ""),
        help=(
            "Optional comma/space separated manifest sample slugs to export. "
            "An empty value keeps every manifest sample."
        ),
    )
    parser.add_argument(
        "--val-samples",
        default=os.environ.get("OPERATOR_YOLO_VAL_SAMPLES", DEFAULT_VALIDATION_SAMPLES),
        help=(
            "Comma/space separated sample slugs assigned to validation. "
            "Defaults to all documented validation samples."
        ),
    )
    parser.add_argument(
        "--allow-empty-validation",
        action="store_true",
        help=(
            "Allow a train-only export shard with no validation AOI. This is only for later "
            "composition into a dataset whose independent validation split is supplied separately."
        ),
    )
    parser.add_argument(
        "--negative-keep-ratio",
        type=float,
        default=float(os.environ.get("OPERATOR_YOLO_NEGATIVE_KEEP_RATIO", "0.35")),
        help="Deterministic ratio of empty tiles to keep as negative examples.",
    )
    parser.add_argument(
        "--min-label-px",
        type=float,
        default=float(os.environ.get("OPERATOR_YOLO_MIN_LABEL_PX", "4")),
        help="Minimum clipped box width/height in pixels before a tile label is kept.",
    )
    parser.add_argument(
        "--min-label-visible-ratio",
        type=float,
        default=float(os.environ.get("OPERATOR_YOLO_MIN_LABEL_VISIBLE_RATIO", "0")),
        help=(
            "Minimum visible share of the original object bbox required before a clipped tile label is kept. "
            "Use 0 to keep legacy edge-fragment labels."
        ),
    )
    parser.add_argument(
        "--background-negative-repeat",
        type=int,
        default=int(os.environ.get("OPERATOR_YOLO_BACKGROUND_NEGATIVE_REPEAT", "1")),
        help="Repeat kept train/background negative tiles this many times for hard-negative balancing.",
    )
    parser.add_argument(
        "--drop-low-variance-negatives",
        action=argparse.BooleanOptionalAction,
        default=env_flag("OPERATOR_YOLO_DROP_LOW_VARIANCE_NEGATIVES", default=False),
        help=(
            "Skip negative tiles whose rendered image has very low pixel variance. "
            "This is intended for no-data/blank pure-empty negatives only."
        ),
    )
    parser.add_argument(
        "--blank-range-threshold",
        type=int,
        default=int(os.environ.get("OPERATOR_YOLO_BLANK_RANGE_THRESHOLD", "3")),
        help="Max pixel value range used to classify a negative tile as visually blank/low-variance.",
    )
    parser.add_argument("--force", action="store_true", help="Remove and recreate output-dir before exporting.")
    return parser.parse_args()


def env_flag(name: str, *, default: bool) -> bool:
    raw = os.environ.get(name)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


def ensure_dependencies() -> None:
    global Image, Transformer, Window, rasterio
    try:
        import rasterio as rasterio_module
        from PIL import Image as image_module
        from pyproj import Transformer as transformer_class
        from rasterio.windows import Window as window_class
    except Exception as exc:  # pragma: no cover - runtime environment only.
        raise SystemExit(
            "export_operator_yolo_tile_dataset.py requires rasterio, pyproj and Pillow. "
            "Run it inside the GeoIntel all-in-one container or an equivalent GIS Python environment."
        ) from exc
    rasterio = rasterio_module
    Window = window_class
    Transformer = transformer_class
    Image = image_module


def split_slugs(raw: str) -> set[str]:
    return {value.strip().lower() for value in raw.replace(",", " ").split() if value.strip()}


def select_manifest_samples(
    samples: list[dict[str, Any]],
    requested_slugs: set[str],
) -> tuple[list[dict[str, Any]], list[str]]:
    manifest_slugs = {
        str(sample.get("sample_slug") or "").strip().lower()
        for sample in samples
        if str(sample.get("sample_slug") or "").strip()
    }
    if not requested_slugs:
        return samples, []
    unknown = requested_slugs - manifest_slugs
    if unknown:
        raise SystemExit(
            "YOLO sample selection references unknown samples: " + ", ".join(sorted(unknown))
        )
    selected = [
        sample
        for sample in samples
        if str(sample.get("sample_slug") or "").strip().lower() in requested_slugs
    ]
    excluded = sorted(manifest_slugs - requested_slugs)
    return selected, excluded


def validate_validation_split(
    samples: list[dict[str, Any]],
    val_slugs: set[str],
    *,
    allow_empty: bool = False,
) -> set[str]:
    sample_slugs = {
        str(sample.get("sample_slug") or "").strip().lower()
        for sample in samples
        if str(sample.get("sample_slug") or "").strip()
    }
    if not val_slugs:
        if allow_empty:
            return set()
        raise SystemExit("YOLO validation split must include at least one sample")
    unknown = val_slugs - sample_slugs
    if unknown:
        raise SystemExit(
            "YOLO validation split references unknown samples: " + ", ".join(sorted(unknown))
        )
    recommended_holdouts = {
        str(sample.get("sample_slug") or "").strip().lower()
        for sample in samples
        if str(sample.get("recommended_split") or "").strip().lower() == "val"
    }
    missing_holdouts = recommended_holdouts - val_slugs
    if missing_holdouts:
        raise SystemExit(
            "YOLO validation split omits recommended validation holdouts: "
            + ", ".join(sorted(missing_holdouts))
        )
    if not sample_slugs - val_slugs:
        raise SystemExit("YOLO validation split leaves no training samples")
    return val_slugs


def validation_sample_coverage(
    tiles: list[dict[str, Any]],
    val_slugs: set[str],
) -> dict[str, list[str]]:
    retained = {
        str(tile.get("sample_slug") or "").strip().lower()
        for tile in tiles
        if tile.get("kept", True) and str(tile.get("split") or "") == "val"
    }
    return {
        "retained_validation_sample_slugs": sorted(retained),
        "empty_validation_sample_slugs": sorted(val_slugs - retained),
    }


def edge_starts(length: int, tile_size: int, stride: int) -> list[int]:
    if tile_size <= 0:
        raise ValueError("tile_size must be positive")
    if stride <= 0:
        raise ValueError("stride must be positive")
    if length <= tile_size:
        return [0]
    starts = list(range(0, max(length - tile_size, 0) + 1, stride))
    final_start = length - tile_size
    if starts[-1] != final_start:
        starts.append(final_start)
    return starts


def iter_tile_windows(width: int, height: int, tile_size: int, stride: int) -> Iterable[TileWindow]:
    for row_off in edge_starts(height, tile_size, stride):
        for col_off in edge_starts(width, tile_size, stride):
            yield TileWindow(row_off=row_off, col_off=col_off, height=min(tile_size, height), width=min(tile_size, width))


def keep_negative_tile(sample_slug: str, tile_index: int, negative_keep_ratio: float) -> bool:
    if negative_keep_ratio <= 0:
        return False
    if negative_keep_ratio >= 1:
        return True
    digest = hashlib.sha256(f"{sample_slug}:{tile_index}".encode("utf-8")).hexdigest()
    bucket = int(digest[:8], 16) / 0xFFFFFFFF
    return bucket < negative_keep_ratio


def background_negative_repeat_count(
    *,
    is_negative: bool,
    sample_role: str,
    split: str,
    background_negative_repeat: int,
) -> int:
    if not is_negative:
        return 1
    if split != "train":
        return 1
    if sample_role != "background_candidate":
        return 1
    return max(1, background_negative_repeat)


def background_category_for_sample(sample: dict[str, Any]) -> str:
    explicit_category = str(sample.get("background_category") or "").strip()
    if explicit_category:
        return explicit_category
    if str(sample.get("sample_role") or "reference") != "background_candidate":
        return REFERENCE_AOI_CATEGORY
    reference_feature_count = int(sample.get("reference_feature_count") or 0)
    if reference_feature_count <= 0:
        return PURE_EMPTY_BACKGROUND_CATEGORY
    return SPARSE_BACKGROUND_CATEGORY


def resolve_manifest_path(raw: str, manifest_path: Path) -> Path:
    path = Path(raw)
    if path.exists():
        return path
    if raw.startswith("/app/"):
        relative = Path(raw.removeprefix("/app/"))
        candidates = [
            Path.cwd() / relative,
            manifest_path.resolve().parent.parent.parent / relative,
        ]
        for candidate in candidates:
            if candidate.exists():
                return candidate
    return path


def iter_geometry_coords(geometry: dict[str, Any]) -> Iterable[tuple[float, float]]:
    geometry_type = geometry.get("type")
    coordinates = geometry.get("coordinates")
    if geometry_type == "Polygon":
        for ring in coordinates or []:
            for point in ring:
                if len(point) >= 2:
                    yield float(point[0]), float(point[1])
    elif geometry_type == "MultiPolygon":
        for polygon in coordinates or []:
            for ring in polygon:
                for point in ring:
                    if len(point) >= 2:
                        yield float(point[0]), float(point[1])


def load_reference_pixel_boxes(
    reference_path: Path,
    dataset: Any,
    min_label_px: float,
    *,
    reference_source: str,
    reference_layer: str,
) -> list[PixelBox]:
    reference = json.loads(reference_path.read_text(encoding="utf-8-sig"))
    features = reference.get("features") or []
    transformer = Transformer.from_crs("EPSG:4326", dataset.crs, always_xy=True)
    boxes: list[PixelBox] = []

    for feature in features:
        properties = feature.get("properties") or {}
        if str(properties.get("source_name") or "").strip().lower() != reference_source:
            continue
        if str(properties.get("reference_layer_name") or "").strip().lower() != reference_layer:
            continue
        coords = list(iter_geometry_coords(feature.get("geometry") or {}))
        if not coords:
            continue
        xs, ys = zip(*(transformer.transform(lon, lat) for lon, lat in coords), strict=False)
        rows_cols = [dataset.index(x, y) for x, y in zip(xs, ys, strict=False)]
        rows = [row for row, _ in rows_cols]
        cols = [col for _, col in rows_cols]
        min_col = max(0, min(cols))
        max_col = min(dataset.width - 1, max(cols))
        min_row = max(0, min(rows))
        max_row = min(dataset.height - 1, max(rows))
        if max_col - min_col < min_label_px or max_row - min_row < min_label_px:
            continue
        boxes.append(PixelBox(min_col=min_col, min_row=min_row, max_col=max_col, max_row=max_row))
    return boxes


def labels_for_tile(tile_window: TileWindow, boxes: list[PixelBox], min_label_px: float, min_visible_ratio: float = 0.0) -> list[str]:
    labels: list[str] = []
    tile_min_col = tile_window.col_off
    tile_min_row = tile_window.row_off
    tile_max_col = tile_window.col_off + tile_window.width
    tile_max_row = tile_window.row_off + tile_window.height

    for box in boxes:
        min_col = max(box.min_col, tile_min_col)
        max_col = min(box.max_col, tile_max_col)
        min_row = max(box.min_row, tile_min_row)
        max_row = min(box.max_row, tile_max_row)
        box_width = max_col - min_col
        box_height = max_row - min_row
        if box_width < min_label_px or box_height < min_label_px:
            continue
        original_area = max((box.max_col - box.min_col) * (box.max_row - box.min_row), 0.0)
        visible_area = box_width * box_height
        visible_ratio = visible_area / original_area if original_area > 0 else 0.0
        if min_visible_ratio > 0 and visible_ratio < min_visible_ratio:
            continue
        local_min_col = min_col - tile_min_col
        local_max_col = max_col - tile_min_col
        local_min_row = min_row - tile_min_row
        local_max_row = max_row - tile_min_row
        x_center = (local_min_col + local_max_col) / 2.0 / tile_window.width
        y_center = (local_min_row + local_max_row) / 2.0 / tile_window.height
        norm_width = box_width / tile_window.width
        norm_height = box_height / tile_window.height
        labels.append(f"0 {x_center:.8f} {y_center:.8f} {norm_width:.8f} {norm_height:.8f}")
    return labels


def image_array_from_raster_window(dataset: Any, tile_window: TileWindow) -> Any:
    import numpy as np

    data = dataset.read(window=Window(tile_window.col_off, tile_window.row_off, tile_window.width, tile_window.height))
    if data.shape[0] == 1:
        rgb = np.repeat(data[:1], 3, axis=0)
    else:
        rgb = data[:3]
    rgb = np.moveaxis(rgb, 0, -1)
    if rgb.dtype != np.uint8:
        rgb_min = float(np.nanmin(rgb))
        rgb_max = float(np.nanmax(rgb))
        if rgb_max > rgb_min:
            rgb = ((rgb - rgb_min) / (rgb_max - rgb_min) * 255.0).clip(0, 255).astype("uint8")
        else:
            rgb = np.zeros(rgb.shape, dtype="uint8")
    return rgb


def image_array_has_low_visual_variance(image_array: Any, blank_range_threshold: int) -> bool:
    import numpy as np

    array = np.asarray(image_array)
    if array.size == 0:
        return True
    max_value = float(np.nanmax(array))
    min_value = float(np.nanmin(array))
    return (max_value - min_value) <= blank_range_threshold


def ensure_yolo_directories(output_dir: Path) -> None:
    for relative_path in ("images/train", "labels/train", "images/val", "labels/val"):
        (output_dir / relative_path).mkdir(parents=True, exist_ok=True)


def write_dataset_yaml(output_dir: Path, class_name: str) -> Path:
    yaml_path = output_dir / "dataset.yaml"
    yaml_path.write_text(
        "\n".join(
            [
                f"path: {output_dir}",
                "train: images/train",
                "val: images/val",
                "names:",
                f"  0: {class_name}",
                "",
            ]
        ),
        encoding="utf-8",
    )
    return yaml_path


def export_sample_tiles(
    sample: dict[str, Any],
    manifest_path: Path,
    output_dir: Path,
    val_slugs: set[str],
    tile_size: int,
    stride: int,
    negative_keep_ratio: float,
    min_label_px: float,
    min_label_visible_ratio: float,
    background_negative_repeat: int,
    drop_low_variance_negatives: bool,
    blank_range_threshold: int,
    reference_source: str,
    reference_layer: str,
) -> list[dict[str, Any]]:
    sample_slug = str(sample["sample_slug"])
    sample_role = str(sample.get("sample_role") or "reference")
    sample_reference_source = str(sample.get("reference_source") or reference_source).strip().lower()
    sample_reference_layer = str(sample.get("reference_layer") or reference_layer).strip().lower()
    background_category = background_category_for_sample(sample)
    recommended_split = str(sample.get("recommended_split") or "")
    split = "val" if sample_slug.lower() in val_slugs else "train"
    raster_path = resolve_manifest_path(str(sample["raster_path"]), manifest_path)
    reference_path = resolve_manifest_path(str(sample["reference_path"]), manifest_path)
    if not raster_path.exists():
        raise SystemExit(f"Raster path is not readable for sample {sample_slug}: {raster_path}")
    if not reference_path.exists():
        raise SystemExit(f"Reference path is not readable for sample {sample_slug}: {reference_path}")

    exported: list[dict[str, Any]] = []
    with rasterio.open(raster_path) as dataset:
        boxes = load_reference_pixel_boxes(
            reference_path,
            dataset,
            min_label_px=min_label_px,
            reference_source=sample_reference_source,
            reference_layer=sample_reference_layer,
        )
        for tile_index, tile_window in enumerate(iter_tile_windows(dataset.width, dataset.height, tile_size, stride)):
            labels = labels_for_tile(
                tile_window,
                boxes,
                min_label_px=min_label_px,
                min_visible_ratio=min_label_visible_ratio,
            )
            is_negative = not labels
            if is_negative and not keep_negative_tile(sample_slug, tile_index, negative_keep_ratio):
                exported.append(
                    {
                        "sample_slug": sample_slug,
                        "sample_role": sample_role,
                        "background_category": background_category,
                        "recommended_split": recommended_split,
                        "split": split,
                        "tile_index": tile_index,
                        "kept": False,
                        "label_count": 0,
                        "is_negative": True,
                        "skip_reason": "negative_keep_ratio",
                    }
                )
                continue
            image_array = image_array_from_raster_window(dataset, tile_window)
            low_visual_variance = image_array_has_low_visual_variance(
                image_array,
                blank_range_threshold=blank_range_threshold,
            )
            if is_negative and drop_low_variance_negatives and low_visual_variance:
                exported.append(
                    {
                        "sample_slug": sample_slug,
                        "sample_role": sample_role,
                        "background_category": background_category,
                        "recommended_split": recommended_split,
                        "split": split,
                        "tile_index": tile_index,
                        "kept": False,
                        "label_count": 0,
                        "is_negative": True,
                        "low_visual_variance": True,
                        "skip_reason": LOW_VARIANCE_NEGATIVE_SKIP_REASON,
                        "window": {
                            "row_off": tile_window.row_off,
                            "col_off": tile_window.col_off,
                            "height": tile_window.height,
                            "width": tile_window.width,
                        },
                    }
                )
                continue
            repeats = background_negative_repeat_count(
                is_negative=is_negative,
                sample_role=sample_role,
                split=split,
                background_negative_repeat=background_negative_repeat,
            )
            for repeat_index in range(repeats):
                repeat_suffix = f"_hn{repeat_index + 1:02d}" if repeats > 1 else ""
                tile_name = f"{sample_slug}_{tile_index:04d}_r{tile_window.row_off}_c{tile_window.col_off}{repeat_suffix}"
                image_path = output_dir / "images" / split / f"{tile_name}.png"
                label_path = output_dir / "labels" / split / f"{tile_name}.txt"
                image_path.parent.mkdir(parents=True, exist_ok=True)
                label_path.parent.mkdir(parents=True, exist_ok=True)
                Image.fromarray(image_array).save(image_path)
                label_path.write_text("\n".join(labels) + ("\n" if labels else ""), encoding="utf-8")
                exported.append(
                    {
                        "sample_slug": sample_slug,
                        "sample_role": sample_role,
                        "background_category": background_category,
                        "recommended_split": recommended_split,
                        "split": split,
                        "tile_index": tile_index,
                        "repeat_index": repeat_index,
                        "kept": True,
                        "image_path": str(image_path),
                        "label_path": str(label_path),
                        "label_count": len(labels),
                        "is_negative": is_negative,
                        "low_visual_variance": low_visual_variance,
                        "is_repeated_background_negative": repeats > 1,
                        "window": {
                            "row_off": tile_window.row_off,
                            "col_off": tile_window.col_off,
                            "height": tile_window.height,
                            "width": tile_window.width,
                        },
                    }
                )
    return exported


def main() -> int:
    args = parse_args()
    class_name = args.class_name.strip().lower()
    reference_source = args.reference_source.strip().lower()
    reference_layer = args.reference_layer.strip().lower()
    for label, value in (
        ("class-name", class_name),
        ("reference-source", reference_source),
        ("reference-layer", reference_layer),
    ):
        if not value or any(character not in "abcdefghijklmnopqrstuvwxyz0123456789_-" for character in value):
            raise SystemExit(f"YOLO {label} must be a non-empty canonical slug")
    manifest = json.loads(args.manifest_path.read_text(encoding="utf-8-sig"))
    try:
        assert_frozen_manifest_training_eligible(args.manifest_path, verify_live=True)
    except TrainingEligibilityError as exc:
        raise SystemExit(str(exc)) from exc
    ensure_dependencies()
    if args.force and args.output_dir.exists():
        shutil.rmtree(args.output_dir)
    args.output_dir.mkdir(parents=True, exist_ok=True)
    ensure_yolo_directories(args.output_dir)

    manifest_samples = manifest.get("samples") or []
    if not manifest_samples:
        raise SystemExit("Operator sample manifest contains no samples")
    samples, excluded_sample_slugs = select_manifest_samples(
        manifest_samples,
        split_slugs(args.samples),
    )
    val_slugs = validate_validation_split(
        samples,
        split_slugs(args.val_samples),
        allow_empty=args.allow_empty_validation,
    )
    exported_tiles: list[dict[str, Any]] = []
    for sample in samples:
        exported_tiles.extend(
            export_sample_tiles(
                sample=sample,
                manifest_path=args.manifest_path,
                output_dir=args.output_dir,
                val_slugs=val_slugs,
                tile_size=args.tile_size,
                stride=args.stride,
                negative_keep_ratio=args.negative_keep_ratio,
                min_label_px=args.min_label_px,
                min_label_visible_ratio=args.min_label_visible_ratio,
                background_negative_repeat=args.background_negative_repeat,
                drop_low_variance_negatives=args.drop_low_variance_negatives,
                blank_range_threshold=args.blank_range_threshold,
                reference_source=reference_source,
                reference_layer=reference_layer,
            )
        )

    kept_tiles = [tile for tile in exported_tiles if tile["kept"]]
    if not any(tile["split"] == "train" for tile in kept_tiles):
        raise SystemExit("YOLO tile dataset export produced no training tiles")
    if not args.allow_empty_validation and not any(tile["split"] == "val" for tile in kept_tiles):
        raise SystemExit("YOLO tile dataset export produced no validation tiles")
    dataset_yaml = write_dataset_yaml(args.output_dir, class_name)
    try:
        release_paths = create_training_release_manifest(
            train_yaml=dataset_yaml,
            corpus_manifest=args.manifest_path,
            review_audit_path=args.review_audit,
        )
    except TrainingReleaseError as exc:
        raise SystemExit(str(exc)) from exc
    positive_tiles = [tile for tile in kept_tiles if not tile["is_negative"]]
    negative_tiles = [tile for tile in kept_tiles if tile["is_negative"]]
    skipped_negative_tiles = [tile for tile in exported_tiles if not tile["kept"] and tile["is_negative"]]
    skipped_low_variance_negative_tiles = [
        tile
        for tile in skipped_negative_tiles
        if tile.get("skip_reason") == LOW_VARIANCE_NEGATIVE_SKIP_REASON
    ]
    validation_coverage = validation_sample_coverage(kept_tiles, val_slugs)
    summary = {
        "status": "ok",
        "dataset_yaml": str(dataset_yaml),
        "training_release_manifest": str(release_paths["release_manifest"]),
        "training_release_manifest_sha256": file_sha256(release_paths["release_manifest"]),
        "training_release_freeze": str(release_paths["release_freeze"]),
        "training_asset_manifest": str(release_paths["asset_manifest"]),
        "output_dir": str(args.output_dir),
        "class_names": [class_name],
        "reference_source": reference_source,
        "reference_layer": reference_layer,
        "tile_size": args.tile_size,
        "stride": args.stride,
        "negative_keep_ratio": args.negative_keep_ratio,
        "background_negative_repeat": args.background_negative_repeat,
        "min_label_px": args.min_label_px,
        "min_label_visible_ratio": args.min_label_visible_ratio,
        "drop_low_variance_negatives": args.drop_low_variance_negatives,
        "blank_range_threshold": args.blank_range_threshold,
        "source_manifest_sample_count": len(manifest_samples),
        "source_manifest": str(args.manifest_path),
        "source_manifest_sha256": file_sha256(args.manifest_path),
        "source_sample_count": len(samples),
        "selected_sample_slugs": sorted(
            str(sample.get("sample_slug") or "").strip().lower() for sample in samples
        ),
        "excluded_sample_slugs": excluded_sample_slugs,
        "validation_sample_slugs": sorted(val_slugs),
        "train_only_shard": args.allow_empty_validation,
        **validation_coverage,
        "tile_count": len(kept_tiles),
        "positive_tile_count": len(positive_tiles),
        "negative_tile_count": len(negative_tiles),
        "skipped_negative_tile_count": len(skipped_negative_tiles),
        "skipped_low_variance_negative_tile_count": len(skipped_low_variance_negative_tiles),
        "label_count": sum(tile["label_count"] for tile in kept_tiles),
        "train_tile_count": sum(1 for tile in kept_tiles if tile["split"] == "train"),
        "val_tile_count": sum(1 for tile in kept_tiles if tile["split"] == "val"),
        "tiles": kept_tiles,
    }
    summary_path = args.output_dir / "yolo_tile_dataset_summary.json"
    summary_path.write_text(json.dumps(summary, indent=2, sort_keys=True), encoding="utf-8")
    print(json.dumps(summary, indent=2, sort_keys=True))
    return 0


if __name__ == "__main__":
    sys.exit(main())