"""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 DEFAULT_MANIFEST_PATH = Path("/app/storage/operator-data/operator_samples_manifest.json") DEFAULT_OUTPUT_DIR = Path("/app/storage/operator-data/yolo-building-tile-dataset") 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 @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("--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( "--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]) -> 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: 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) -> 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 properties.get("source_name") != "grb": continue if properties.get("reference_layer_name") != "buildings": 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) -> Path: yaml_path = output_dir / "dataset.yaml" yaml_path.write_text( "\n".join( [ f"path: {output_dir}", "train: images/train", "val: images/val", "names:", " 0: building", "", ] ), 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, ) -> list[dict[str, Any]]: sample_slug = str(sample["sample_slug"]) sample_role = str(sample.get("sample_role") or "reference") 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) 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() 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 = json.loads(args.manifest_path.read_text(encoding="utf-8-sig")) 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)) 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, ) ) 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 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) 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), "output_dir": str(args.output_dir), "class_names": ["building"], "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_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), **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())