Add fail-closed source class and SAM fallback filters

scripts/normalize_belgium_building_labels.py

@@ -148,6 +148,7 @@ def normalize(
     reference_observed_at: str | None,
     imagery_valid_to: str | None = None,
     merge_touching_roofs: bool = False,
+    allowed_source_classes: set[str] | None = None,
 ) -> tuple[dict[str, Any], dict[str, Any]]:
     if source_name not in SUPPORTED_SOURCES:
         raise SystemExit(f"Unsupported governed building source: {source_name}")
@@ -184,6 +185,7 @@ def normalize(
                 "reason": None,
                 "geometry_repaired": False,
             }
+            source_class = decision["source_class"]
             source_creation_at = _source_creation_at(source_name, properties)
             decision["source_creation_at"] = source_creation_at.isoformat() if source_creation_at else None
             try:
@@ -199,6 +201,8 @@ def normalize(
             geometry = _polygonal(geometry)
             if geometry is None or geometry.is_empty or not geometry.is_valid:
                 decision["reason"] = "invalid_geometry_unrepairable"
+            elif allowed_source_classes is not None and source_class not in allowed_source_classes:
+                decision["reason"] = "source_class_not_allowed"
             elif (reason := _semantic_exclusion(properties)) is not None:
                 decision["reason"] = reason
             elif imagery_cutoff and source_creation_at and source_creation_at > imagery_cutoff:
@@ -276,6 +280,7 @@ def normalize(
         "accepted_source_feature_count": len(accepted),
         "accepted_feature_count": len(normalized_features),
         "merge_touching_roofs": merge_touching_roofs,
+        "allowed_source_classes": sorted(allowed_source_classes) if allowed_source_classes is not None else None,
         "decision_counts": dict(sorted(counts.items())),
         "decisions": decisions,
     }
@@ -294,6 +299,12 @@ def main() -> int:
     parser.add_argument("--reference-observed-at")
     parser.add_argument("--imagery-valid-to")
     parser.add_argument("--merge-touching-roofs", action="store_true")
+    parser.add_argument(
+        "--source-class",
+        action="append",
+        dest="source_classes",
+        help="Repeatable provider-native class allowlist. No cross-provider semantic mapping is inferred.",
+    )
     args = parser.parse_args()
     normalized, audit = normalize(
         reference_path=args.reference,
@@ -304,6 +315,7 @@ def main() -> int:
         reference_observed_at=args.reference_observed_at,
         imagery_valid_to=args.imagery_valid_to,
         merge_touching_roofs=args.merge_touching_roofs,
+        allowed_source_classes=set(args.source_classes) if args.source_classes else None,
     )
     args.output_reference.parent.mkdir(parents=True, exist_ok=True)
     args.output_audit.parent.mkdir(parents=True, exist_ok=True)

scripts/refine_yolo_labels_with_sam.py

@@ -108,6 +108,7 @@ def main() -> int:
     parser.add_argument("--min-dimension-ratio", type=float, default=0.5)
     parser.add_argument("--max-dimension-ratio", type=float, default=2.0)
     parser.add_argument("--max-prompts-per-pass", type=int, default=96)
+    parser.add_argument("--fallback-policy", choices=("retain", "drop"), default="retain")
     parser.add_argument("--force", action="store_true")
     args = parser.parse_args()
     if args.output_dir.exists():
@@ -134,7 +135,7 @@ def main() -> int:
         with Image.open(source_image) as image:
             width, height = image.size
         source_boxes = read_boxes(source_label, width, height)
-        output_boxes = list(source_boxes)
+        output_boxes: list[tuple[float, float, float, float] | None] = list(source_boxes)
         if source_boxes:
             for start in range(0, len(source_boxes), args.max_prompts_per_pass):
                 source_chunk = source_boxes[start : start + args.max_prompts_per_pass]
@@ -153,6 +154,8 @@ def main() -> int:
                     if candidate_index < 0 or overlap < args.min_source_iou:
                         reason_counts["unmatched_mask"] = reason_counts.get("unmatched_mask", 0) + 1
                         fallback_count += 1
+                        if args.fallback_policy == "drop":
+                            output_boxes[start + local_index] = None
                         continue
                     candidate = candidates[candidate_index]
                     if plausible_refinement(
@@ -171,14 +174,22 @@ def main() -> int:
                     else:
                         reason_counts["geometry_gate"] = reason_counts.get("geometry_gate", 0) + 1
                         fallback_count += 1
+                        if args.fallback_policy == "drop":
+                            output_boxes[start + local_index] = None
                 del result, masks
                 if model.predictor is not None:
                     model.predictor.reset_image()
                 gc.collect()
                 torch.cuda.empty_cache()
-        target_label.write_text("\n".join(yolo_line(box, width, height) for box in output_boxes) + ("\n" if output_boxes else ""), encoding="utf-8")
+        retained_boxes = [box for box in output_boxes if box is not None]
+        target_label.write_text("\n".join(yolo_line(box, width, height) for box in retained_boxes) + ("\n" if retained_boxes else ""), encoding="utf-8")
         output_tile = dict(tile)
-        output_tile.update({"image_path": str(target_image), "label_path": str(target_label)})
+        output_tile.update({
+            "image_path": str(target_image),
+            "label_path": str(target_label),
+            "label_count": len(retained_boxes),
+            "is_negative": not retained_boxes,
+        })
         output_tiles.append(output_tile)
         print(f"{index}/{len(summary['tiles'])} {tile['sample_slug']}: {len(source_boxes)}", flush=True)
 
@@ -188,7 +199,14 @@ def main() -> int:
             "output_dir": str(args.output_dir),
             "dataset_yaml": str(args.output_dir / "dataset.yaml"),
             "tiles": output_tiles,
-            "label_semantics": "sam_visible_roof_with_official_footprint_fallback",
+            "label_semantics": (
+                "sam_visible_roof_only"
+                if args.fallback_policy == "drop"
+                else "sam_visible_roof_with_official_footprint_fallback"
+            ),
+            "label_count": refined_count if args.fallback_policy == "drop" else summary["label_count"],
+            "positive_tile_count": sum(not tile["is_negative"] for tile in output_tiles),
+            "negative_tile_count": sum(tile["is_negative"] for tile in output_tiles),
         }
     )
     summary_path = args.output_dir / "yolo_tile_dataset_summary.json"
@@ -213,8 +231,10 @@ def main() -> int:
         "min_dimension_ratio": args.min_dimension_ratio,
         "max_dimension_ratio": args.max_dimension_ratio,
         "max_prompts_per_pass": args.max_prompts_per_pass,
+        "fallback_policy": args.fallback_policy,
         "refined_label_count": refined_count,
         "fallback_label_count": fallback_count,
+        "dropped_fallback_label_count": fallback_count if args.fallback_policy == "drop" else 0,
         "fallback_reason_counts": reason_counts,
     }
     (args.output_dir / "sam-refinement.json").write_text(json.dumps(evidence, indent=2), encoding="utf-8")