Add hard-negative balanced YOLO tile export

scripts/export_operator_yolo_tile_dataset.py

@@ -78,6 +78,12 @@ def parse_args() -> argparse.Namespace:
         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(
+        "--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("--force", action="store_true", help="Remove and recreate output-dir before exporting.")
     return parser.parse_args()
 
@@ -134,6 +140,22 @@ def keep_negative_tile(sample_slug: str, tile_index: int, negative_keep_ratio: f
     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 resolve_manifest_path(raw: str, manifest_path: Path) -> Path:
     path = Path(raw)
     if path.exists():
@@ -274,8 +296,10 @@ def export_sample_tiles(
     stride: int,
     negative_keep_ratio: float,
     min_label_px: float,
+    background_negative_repeat: int,
 ) -> list[dict[str, Any]]:
     sample_slug = str(sample["sample_slug"])
+    sample_role = str(sample.get("sample_role") or "reference")
     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)
@@ -302,31 +326,43 @@ def export_sample_tiles(
                     }
                 )
                 continue
-            tile_name = f"{sample_slug}_{tile_index:04d}_r{tile_window.row_off}_c{tile_window.col_off}"
-            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_from_raster_window(dataset, tile_window)).save(image_path)
-            label_path.write_text("\n".join(labels) + ("\n" if labels else ""), encoding="utf-8")
-            exported.append(
-                {
-                    "sample_slug": sample_slug,
-                    "split": split,
-                    "tile_index": tile_index,
-                    "kept": True,
-                    "image_path": str(image_path),
-                    "label_path": str(label_path),
-                    "label_count": len(labels),
-                    "is_negative": is_negative,
-                    "window": {
-                        "row_off": tile_window.row_off,
-                        "col_off": tile_window.col_off,
-                        "height": tile_window.height,
-                        "width": tile_window.width,
-                    },
-                }
+            repeats = background_negative_repeat_count(
+                is_negative=is_negative,
+                sample_role=sample_role,
+                split=split,
+                background_negative_repeat=background_negative_repeat,
             )
+            image_array = image_array_from_raster_window(dataset, tile_window)
+            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,
+                        "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,
+                        "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
 
 
@@ -355,6 +391,7 @@ def main() -> int:
                 stride=args.stride,
                 negative_keep_ratio=args.negative_keep_ratio,
                 min_label_px=args.min_label_px,
+                background_negative_repeat=args.background_negative_repeat,
             )
         )
 
@@ -375,6 +412,7 @@ def main() -> int:
         "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,
         "source_sample_count": len(samples),
         "tile_count": len(kept_tiles),