Read the SAR metadata from the CDSE daily parquet files

docs/index.rst

@@ -33,6 +33,16 @@ with corresponding mission paths for S1B/C/D. Each extracted SAFE needs its
 ``stop_date`` is exclusive. ``output_dir`` must be new. Replace the previous
 ``sar_cache_dir`` / ``offline`` keys with ``sar_archive_root``.
 
+To read the SAR metadata from the CDSE daily parquet files instead of opening the SAFEs,
+set ``sar_parquet_root`` in the recipe: the folder of the ``s1_YYYY-MM-DD.parquet`` files, which
+need the ``WVImagettes`` column. The footprint polygons give the imagette footprints and
+centres. A recipe needs ``sar_archive_root``, ``sar_parquet_root``, or both. With the archive
+alone the SAFEs are opened. With the parquet folder alone nothing is opened and every
+imagette counts as available. With both, the metadata comes from the parquet files and the
+archive is used only to check which imagette files exist. The manifest lists the SAFEs left
+out (no imagette list, or polygons and imagettes that disagree) and the imagettes whose file is
+not in the archive, under ``discovery``.
+
 Reference products
 ------------------
 

environment.yml

@@ -15,3 +15,4 @@ dependencies:
   - xarray=2026.9.0
   - shapely=2.2.0
   - netcdf4=1.7.4
+  - pyarrow=26.0.0

pyproject.toml

@@ -11,7 +11,7 @@ license = "MIT"
 license-files = ["LICENSE"]
 requires-python = ">=3.11"
 dependencies = ["numpy>=1.26,<3", "xarray>=2024.1", "shapely>=2",
-                "netCDF4>=1.7,<2"]
+                "netCDF4>=1.7,<2", "pyarrow>=15"]
 
 [project.scripts]
 soba_matchups = "soba_matchups.cli:main"

src/soba_matchups/config.py

@@ -38,11 +38,12 @@ class Recipe:
     stop_date: date
     sar_missions: tuple[str, ...]
     time_tolerance_minutes: float
-    sar_archive_root: Path
     output_dir: Path
     instrument: str
     platform: str
     resolution: str
+    sar_archive_root: Path | None = None
+    sar_parquet_root: Path | None = None
 
     def resolved(self):
         """Return the recipe as plain values that can be written to JSON."""
@@ -57,12 +58,16 @@ def load_recipe(path):
     with path.open("rb") as stream:
         data = tomllib.load(stream)
     # Reject unknown keys and missing required keys.
-    required = set(Recipe.__dataclass_fields__) - {"path"}
-    unknown = set(data) - required
+    optional = {"sar_archive_root", "sar_parquet_root"}
+    required = set(Recipe.__dataclass_fields__) - {"path"} - optional
+    unknown = set(data) - required - optional
     if unknown or required - set(data):
         raise ValueError(
             f"Unknown recipe keys: {sorted(unknown)}; missing keys: {sorted(required - set(data))}"
         )
+    # The SAR metadata comes from the archive, the daily parquet files, or both.
+    if not optional & set(data):
+        raise ValueError("Recipe needs sar_archive_root, sar_parquet_root, or both")
     product = (data["instrument"], data["platform"], data["resolution"])
     if any(not isinstance(v, str) for v in product) or product not in REFERENCE_PRODUCTS:
         raise ValueError("Unsupported reference instrument / platform / resolution combination")
@@ -99,12 +104,17 @@ def load_recipe(path):
     ):
         raise ValueError("sar_missions must contain distinct supported Sentinel-1 missions")
     # Make every path absolute, relative to the recipe file.
-    for key in ["reference_root", "sar_archive_root", "output_dir"]:
+    paths = ["reference_root", "output_dir"]
+    inputs = ["reference_root"]
+    for key in sorted(optional & set(data)):
+        paths.append(key)
+        inputs.append(key)
+    for key in paths:
         if not isinstance(data[key], str) or not data[key].strip():
             raise ValueError(f"{key} must be a nonempty path")
         data[key] = (path.parent / data[key]).resolve()
     # Inputs must exist, and results must never be written inside them.
-    for key in ["reference_root", "sar_archive_root"]:
+    for key in inputs:
         if not data[key].is_dir():
             raise ValueError(f"{key} directory does not exist: {data[key]}")
         if data["output_dir"].is_relative_to(data[key]):

src/soba_matchups/pipeline.py

@@ -10,6 +10,12 @@
 from soba_matchups.adapters import alti, scat, swot
 from soba_matchups.config import REFERENCE_PRODUCTS
 from soba_matchups.output import file_identity, imagette_item, publish
+from soba_matchups.sar_parquet import (
+    IMAGETTE_MAPPING,
+    TIME_SOURCE,
+    discover_parquet,
+    verify_parquet_inputs,
+)
 from soba_matchups.sar_metadata import (
     OCN_HEADER_PRECISION_SECONDS,
     decode_safe_name,
@@ -18,6 +24,22 @@
 )
 
 
+# What each SAR source can and cannot do, copied into the manifest of every run.
+ARCHIVE_LIMITATIONS = [
+    "Requires extracted L2 OCN SAFEs with manifest.safe and every listed measurement file.",
+    "Missing/ambiguous native frame associations fail without suffix fallback.",
+    "Reads OSW geolocation and UTC attributes, not wind/spectral arrays; no network.",
+    "Archive inventory only; missing mission/day directories are listed in discovery.",
+    "SAR file identities use size/mtime; manifests and references use SHA256.",
+]
+PARQUET_LIMITATIONS = [
+    "SAR metadata comes from the daily parquet files; no SAFE is opened, nothing is fetched.",
+    "Imagette centres are polygon centroids; polygon k is paired with imagette number k.",
+    "SAFEs with no imagette list, or with polygons and imagettes that disagree, are left out.",
+    "Imagettes whose file is not in the archive are matched and listed as unavailable.",
+    "Parquet files and references use SHA256.",
+]
+
 # One adapter per instrument: it finds the files, reads them and names the point fields.
 ADAPTERS = {"ascat": scat, "hscat": scat, "karin": swot, "alti": alti}
 
@@ -75,7 +97,17 @@ def iso(t):
     # Search window: reference coverage widened by the time tolerance and header rounding.
     start = iso(min(ref["coverage_start"] for ref in refs) - delta - precision)
     end = iso(max(ref["coverage_end"] for ref in refs) + delta + precision)
-    products, discovery = discover_local(start, end, recipe.sar_missions, recipe.sar_archive_root)
+    from_parquet = recipe.sar_parquet_root is not None
+    if from_parquet:
+        products, discovery = discover_parquet(
+            start, end, recipe.sar_missions, recipe.sar_parquet_root, recipe.sar_archive_root
+        )
+        verify_inputs = verify_parquet_inputs
+    else:
+        products, discovery = discover_local(
+            start, end, recipe.sar_missions, recipe.sar_archive_root
+        )
+        verify_inputs = verify_local_inputs
     items, diagnostics = [], []
     seen_points = set()
     imagettes_processed = duplicate_points = safes_without_candidates = 0
@@ -189,7 +221,7 @@ def number(x, integer=False):
                     imagette_item(product, imagette, fields, groups, recipe.time_tolerance_minutes)
                 )
     # Confirm no input changed while matching, before anything is published.
-    verify_local_inputs(discovery)
+    verify_inputs(discovery)
     for ref in refs:
         if file_identity(ref["path"]) != ref["identity"]:
             raise ValueError(f"Input changed during processing: {ref['path']}")
@@ -227,6 +259,9 @@ def number(x, integer=False):
         "safe_candidate_cell_rows": candidate_cells,
         "imagettes_processed": imagettes_processed,
         "imagettes_skipped": sum(len(p["SkippedImagettes"]) for p in products),
+        "safes_without_imagettes": len(discovery.get("safes_without_imagettes", [])),
+        "safes_skipped": len(discovery.get("safes_skipped", [])),
+        "imagettes_unavailable": len(discovery.get("imagettes_unavailable", [])),
         "stac_items": len(items),
         "matchup_imagette_cell_rows": len(seen_points),
         "duplicate_associations_discarded": duplicate_points,
@@ -235,12 +270,18 @@ def number(x, integer=False):
         "source_code": [file_identity(p) for p in sorted(Path(__file__).parent.rglob("*.py"))],
         "rules": {
             "policy": adapter.POLICY,
-            "imagette_mapping": "local_ocn_geolocation_to_manifest_frame",
+            "imagette_mapping": (
+                IMAGETTE_MAPPING if from_parquet else "local_ocn_geolocation_to_manifest_frame"
+            ),
             "space": getattr(
                 adapter, "SPACE", "strict individual imagette footprint point containment"
             ),
             "time": "strict imagette start - tolerance < cell time < imagette end + tolerance",
-            "time_source": "OCN firstMeasurementTime / lastMeasurementTime UTC attributes",
+            "time_source": (
+                TIME_SOURCE
+                if from_parquet
+                else "OCN firstMeasurementTime / lastMeasurementTime UTC attributes"
+            ),
             "candidate_padding_seconds": OCN_HEADER_PRECISION_SECONDS,
             "header_precision": "whole-second NC headers use outward-rounded manifest bounds",
             "time_tolerance_minutes": recipe.time_tolerance_minutes,
@@ -252,11 +293,7 @@ def number(x, integer=False):
             "ref_lat": "degrees_north",
         },
         "limitations": [
-            "Requires extracted L2 OCN SAFEs with manifest.safe and all listed measurement files.",
-            "Missing/ambiguous native frame associations fail without suffix fallback.",
-            "Reads OSW geolocation and UTC attributes, not wind/spectral arrays; no network.",
-            "Archive inventory only; missing mission/day directories are listed in discovery.",
-            "SAR file identities use size/mtime; manifests and references use SHA256.",
+            *(PARQUET_LIMITATIONS if from_parquet else ARCHIVE_LIMITATIONS),
             "sarmatchup:* properties are project-specific, not a registered STAC extension.",
             "No sea-percent, polar, quality or ice filter.",
             "Tolerance is explicitly configured, not scientifically calibrated.",