Target regional training failures by context

scripts/build_failure_driven_yolo_sampling.py

@@ -36,10 +36,18 @@ def build_sampling(
     positive_repeat: int = 3,
     negative_repeat: int = 4,
     precision_positive_repeat: int = 1,
+    context_positive_repeat: int = 5,
+    context_negative_repeat: int = 6,
 ) -> tuple[list[str], dict[str, Any]]:
     if assessment.get("status") != "continue_training_loop":
         raise ValueError("Failure-driven sampling requires a failed assessment")
-    if positive_repeat < 1 or negative_repeat < 1 or precision_positive_repeat < 1:
+    if min(
+        positive_repeat,
+        negative_repeat,
+        precision_positive_repeat,
+        context_positive_repeat,
+        context_negative_repeat,
+    ) < 1:
         raise ValueError("Repeat factors must be positive")
 
     samples = {item["sample_slug"]: item for item in manifest["samples"]}
@@ -65,6 +73,23 @@ def build_sampling(
         and background["pure_empty_false_positives"]
         > gates["max_pure_empty_false_positives"]
     )
+    weak_recall_contexts: set[tuple[str, str]] = set()
+    weak_precision_contexts: set[tuple[str, str]] = set()
+    for sample_slug, metrics in evaluation.get("samples", {}).items():
+        sample = samples.get(sample_slug)
+        if not sample:
+            continue
+        key = (sample["region"], sample.get("context", "unknown"))
+        if sample["region"] in weak_recall_regions and (
+            metrics["f1"] < gates["min_region_f1"]
+            or metrics["recall"] < gates["min_region_recall"]
+        ):
+            weak_recall_contexts.add(key)
+        if (
+            sample["region"] in weak_precision_regions
+            and metrics["precision"] < gates["min_region_precision"]
+        ):
+            weak_precision_contexts.add(key)
 
     image_paths: list[str] = []
     repeat_counts: Counter[str] = Counter()
@@ -76,15 +101,24 @@ def build_sampling(
             protected_samples.add(tile["sample_slug"])
             continue
         region = sample["region"]
+        context_key = (region, sample.get("context", "unknown"))
         repeat = 1
         if tile["label_count"] > 0 and region in weak_recall_regions:
-            repeat = positive_repeat
+            repeat = (
+                context_positive_repeat
+                if context_key in weak_recall_contexts
+                else positive_repeat
+            )
         elif tile["label_count"] > 0 and region in weak_precision_regions:
             # Precision-only correction still needs positive examples to avoid
             # shifting the classifier toward background and sacrificing recall.
             repeat = precision_positive_repeat
         if tile["label_count"] == 0 and (background_failed or region in weak_precision_regions):
-            repeat = negative_repeat
+            repeat = (
+                context_negative_repeat
+                if context_key in weak_precision_contexts
+                else negative_repeat
+            )
         path = str(Path(tile["image_path"]).resolve())
         image_paths.extend([path] * repeat)
         repeat_counts[region] += repeat
@@ -99,10 +133,14 @@ def build_sampling(
         "failure_evidence_source": "test" if assessment.get("test") else "calibration",
         "weak_recall_regions": sorted(weak_recall_regions),
         "weak_precision_regions": sorted(weak_precision_regions),
+        "weak_recall_contexts": [f"{region}:{context}" for region, context in sorted(weak_recall_contexts)],
+        "weak_precision_contexts": [f"{region}:{context}" for region, context in sorted(weak_precision_contexts)],
         "background_gate_failed": background_failed,
         "positive_repeat": positive_repeat,
         "negative_repeat": negative_repeat,
         "precision_positive_repeat": precision_positive_repeat,
+        "context_positive_repeat": context_positive_repeat,
+        "context_negative_repeat": context_negative_repeat,
         "source_train_tile_count": sum(
             1
             for tile in summary["tiles"]
@@ -126,6 +164,8 @@ def main() -> int:
     parser.add_argument("--positive-repeat", type=int, default=3)
     parser.add_argument("--negative-repeat", type=int, default=4)
     parser.add_argument("--precision-positive-repeat", type=int, default=1)
+    parser.add_argument("--context-positive-repeat", type=int, default=5)
+    parser.add_argument("--context-negative-repeat", type=int, default=6)
     args = parser.parse_args()
 
     summary = json.loads(args.summary.read_text(encoding="utf-8"))
@@ -138,6 +178,8 @@ def main() -> int:
         positive_repeat=args.positive_repeat,
         negative_repeat=args.negative_repeat,
         precision_positive_repeat=args.precision_positive_repeat,
+        context_positive_repeat=args.context_positive_repeat,
+        context_negative_repeat=args.context_negative_repeat,
     )
     args.output_dir.mkdir(parents=True, exist_ok=True)
     train_list = args.output_dir / "train-failure-driven.txt"