Filter labels created after dated imagery

scripts/assemble_belgium_building_corpus.py

@@ -149,6 +149,7 @@ def main() -> int:
                     else None
                 ),
                 reference_observed_at=reference.observed_at.isoformat() if reference.observed_at else None,
+                imagery_valid_to=raster.valid_to.isoformat() if raster.valid_to else None,
             )
             normalized_target.write_text(json.dumps(normalized, ensure_ascii=False), encoding="utf-8")
             audit_target.write_text(json.dumps(audit, ensure_ascii=False, indent=2), encoding="utf-8")

scripts/normalize_belgium_building_labels.py

@@ -12,7 +12,7 @@ import argparse
 import hashlib
 import json
 from collections import Counter
-from datetime import datetime
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
@@ -57,6 +57,26 @@ def _semantic_exclusion(properties: dict[str, Any]) -> str | None:
     return None
 
 
+def _source_creation_at(source_name: str, properties: dict[str, Any]) -> datetime | None:
+    raw: Any = None
+    if source_name == "grb":
+        raw = properties.get("BEGINDATUM")
+    elif source_name == "spw_picc":
+        raw = properties.get("DATE_CREAT")
+    if raw in (None, ""):
+        return None
+    try:
+        if isinstance(raw, (int, float)) or str(raw).isdigit():
+            value = float(raw)
+            if value > 10_000_000_000:
+                value /= 1000
+            return datetime.fromtimestamp(value, tz=UTC)
+        parsed = datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
+        return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC)
+    except (OSError, OverflowError, ValueError):
+        return None
+
+
 def _polygonal(geometry: Any) -> Any | None:
     if geometry.geom_type in {"Polygon", "MultiPolygon"}:
         return geometry
@@ -78,6 +98,7 @@ def normalize(
     min_label_px: float,
     imagery_observed_at: str | None,
     reference_observed_at: str | None,
+    imagery_valid_to: 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}")
@@ -89,6 +110,9 @@ def normalize(
     decisions: list[dict[str, Any]] = []
     seen: set[str] = set()
     counts: Counter[str] = Counter()
+    imagery_cutoff = (
+        datetime.fromisoformat(imagery_valid_to.replace("Z", "+00:00")) if imagery_valid_to else None
+    )
     with rasterio.open(raster_path) as raster:
         if raster.crs is None:
             raise SystemExit("Raster CRS is required")
@@ -104,6 +128,8 @@ def normalize(
                 "reason": None,
                 "geometry_repaired": False,
             }
+            source_creation_at = _source_creation_at(source_name, properties)
+            decision["source_creation_at"] = source_creation_at.isoformat() if source_creation_at else None
             try:
                 geometry = shape(feature.get("geometry"))
             except Exception:
@@ -119,6 +145,8 @@ def normalize(
                 decision["reason"] = "invalid_geometry_unrepairable"
             elif (reason := _semantic_exclusion(properties)) is not None:
                 decision["reason"] = reason
+            elif imagery_cutoff and source_creation_at and source_creation_at > imagery_cutoff:
+                decision["reason"] = "created_after_imagery_period"
             else:
                 metric = shapely_transform(transformer.transform, geometry)
                 min_x, min_y, max_x, max_y = metric.bounds
@@ -179,6 +207,7 @@ def normalize(
         "raster_path": str(raster_path),
         "min_label_px": min_label_px,
         "imagery_observed_at": imagery_observed_at,
+        "imagery_valid_to": imagery_valid_to,
         "reference_observed_at": reference_observed_at,
         "temporal_mismatch_days": temporal_mismatch_days,
         "temporal_alignment_status": temporal_alignment_status,
@@ -200,6 +229,7 @@ def main() -> int:
     parser.add_argument("--min-label-px", type=float, default=3.0)
     parser.add_argument("--imagery-observed-at")
     parser.add_argument("--reference-observed-at")
+    parser.add_argument("--imagery-valid-to")
     args = parser.parse_args()
     normalized, audit = normalize(
         reference_path=args.reference,
@@ -208,6 +238,7 @@ def main() -> int:
         min_label_px=args.min_label_px,
         imagery_observed_at=args.imagery_observed_at,
         reference_observed_at=args.reference_observed_at,
+        imagery_valid_to=args.imagery_valid_to,
     )
     args.output_reference.parent.mkdir(parents=True, exist_ok=True)
     args.output_audit.parent.mkdir(parents=True, exist_ok=True)