feat(weather): tier-2 VIIRS products — AOD, CloudHeight, LST, SnowCover

packages/weather/src/mostlyright/weather/_fetchers/_viirs_extract.py

@@ -126,6 +126,156 @@
         tier="v1",
         valid_range_filter=False,
     ),
+    # ============ Tier-2 registrations (live samples verified 2026-07-06) ====
+    # VIIRS-JRR-AOD — Enterprise aerosol optical depth. Sample:
+    # JRR-AOD_v3r2_j01_s202407010002410 (n20, ~8-10 MB): AOD550 float32
+    # (768, 3200), _FillValue=-999.999, valid_range [-0.05, 5], units "1",
+    # no scale/offset. Archive: snpp/n20 2022+, n21 2023+. Feature: smoke/haze
+    # days measurably suppress Tmax.
+    ("VIIRS-JRR-AOD", "AOD550"): ProductVariable(
+        product_code="VIIRS-JRR-AOD",
+        variable="AOD550",
+        units_expected="1",
+        grid_shape_expected=(768, 3200),
+        cadence_s=86,
+        bounds=(-0.05, 5.0),
+        dqf_good=None,
+        dqf_variable=None,
+        is_3d_profile=False,
+        is_categorical=False,
+        description=(
+            "Aerosol Optical Depth at 550 nm — single SWATH pixel nearest the "
+            "station lat/lon, raw-as-reported float32 (smoke/haze index)"
+        ),
+        tier="v1",
+        valid_range_filter=False,
+    ),
+    # VIIRS-JRR-CloudHeight — cloud-top properties. Sample:
+    # JRR-CloudHeight_v3r2_j01_s202407010001153 (n20, ~67 MB — well under the
+    # 1 GiB VIIRS cap): CldTopHght (Meter) / CldTopTemp (Kelvin) / CldTopPres
+    # (hPa), all float32 (768, 3200), _FillValue=-999, no scale/offset.
+    # Archive: all three platforms 2023+. Feature: cold high tops = deep
+    # convection capping Tmax; low warm stratus = radiative blanket.
+    ("VIIRS-JRR-CloudHeight", "CldTopHght"): ProductVariable(
+        product_code="VIIRS-JRR-CloudHeight",
+        variable="CldTopHght",
+        units_expected="Meter",
+        grid_shape_expected=(768, 3200),
+        cadence_s=86,
+        bounds=(0.0, 25000.0),
+        dqf_good=None,
+        dqf_variable=None,
+        is_3d_profile=False,
+        is_categorical=False,
+        description=(
+            "Cloud Top Height [m] — single SWATH pixel nearest the station "
+            "lat/lon, raw-as-reported float32"
+        ),
+        tier="v1",
+        valid_range_filter=False,
+    ),
+    ("VIIRS-JRR-CloudHeight", "CldTopTemp"): ProductVariable(
+        product_code="VIIRS-JRR-CloudHeight",
+        variable="CldTopTemp",
+        units_expected="Kelvin",
+        grid_shape_expected=(768, 3200),
+        cadence_s=86,
+        bounds=(150.0, 340.0),
+        dqf_good=None,
+        dqf_variable=None,
+        is_3d_profile=False,
+        is_categorical=False,
+        description=(
+            "Cloud Top Temperature [K] — single SWATH pixel nearest the "
+            "station lat/lon, raw-as-reported float32"
+        ),
+        tier="v1",
+        valid_range_filter=False,
+    ),
+    ("VIIRS-JRR-CloudHeight", "CldTopPres"): ProductVariable(
+        product_code="VIIRS-JRR-CloudHeight",
+        variable="CldTopPres",
+        units_expected="hPa",
+        grid_shape_expected=(768, 3200),
+        cadence_s=86,
+        bounds=(50.0, 1100.0),
+        dqf_good=None,
+        dqf_variable=None,
+        is_3d_profile=False,
+        is_categorical=False,
+        description=(
+            "Cloud Top Pressure [hPa] — single SWATH pixel nearest the "
+            "station lat/lon, raw-as-reported float32"
+        ),
+        tier="v1",
+        valid_range_filter=False,
+    ),
+    # JPSSRR_LST — VIIRS land surface temperature swath (M-band grid). Sample:
+    # LST_v2r2_j01_s202412200001157 (n20, ~7 MB): LST int16 SCALED
+    # (raw*0.005 + 200 -> Kelvin), _FillValue=-32768, raw valid_range
+    # [2600, 28600] (= 213..343 K). NOTE the filename prefix is LST_v2r2, not
+    # JRR-* — the transport lists the whole day prefix so this is cosmetic.
+    # Archive starts ~Dec 2024 on ALL three platforms (young; forward value).
+    ("JPSSRR_LST", "LST"): ProductVariable(
+        product_code="JPSSRR_LST",
+        variable="LST",
+        units_expected="Kelvin",
+        grid_shape_expected=(768, 3200),
+        cadence_s=86,
+        bounds=(180.0, 360.0),
+        dqf_good=None,
+        dqf_variable=None,
+        is_3d_profile=False,
+        is_categorical=False,
+        description=(
+            "Land Surface (skin) Temperature — single SWATH pixel nearest the "
+            "station lat/lon, scaled int16 -> Kelvin. Polar-orbit LST covers "
+            "the EU/Asia stations GOES cannot see."
+        ),
+        tier="v1",
+    ),
+    # JPSSRR_SnowCover — I-BAND grid: (1536, 6400), DOUBLE the M-band
+    # resolution of every other registered VIIRS product. Sample:
+    # JRR-SnowCover_v3r2_j01_s202501150000028 (n20, ~20 MB): SnowFraction
+    # int8 %, SnowMask categorical int8, both _FillValue=-128, no
+    # scale/offset. Archive ~Dec 2024+ on all three platforms. Feature: snow
+    # albedo suppresses Tmax; snowpack enables radiative-cooling cold NLOW.
+    ("JPSSRR_SnowCover", "SnowFraction"): ProductVariable(
+        product_code="JPSSRR_SnowCover",
+        variable="SnowFraction",
+        units_expected="%",
+        grid_shape_expected=(1536, 6400),
+        cadence_s=86,
+        bounds=(0.0, 100.0),
+        dqf_good=None,
+        dqf_variable=None,
+        is_3d_profile=False,
+        is_categorical=False,
+        description=(
+            "Snow Fraction [%] — single I-band SWATH pixel nearest the "
+            "station lat/lon, raw-as-reported int8"
+        ),
+        tier="v1",
+        valid_range_filter=False,
+    ),
+    ("JPSSRR_SnowCover", "SnowMask"): ProductVariable(
+        product_code="JPSSRR_SnowCover",
+        variable="SnowMask",
+        units_expected="1",
+        grid_shape_expected=(1536, 6400),
+        cadence_s=86,
+        bounds=(0.0, 3.0),
+        dqf_good=None,
+        dqf_variable=None,
+        is_3d_profile=False,
+        is_categorical=True,
+        description=(
+            "Snow Mask (categorical) — single I-band SWATH pixel nearest the "
+            "station lat/lon, raw-as-reported int8"
+        ),
+        tier="v1",
+        valid_range_filter=False,
+    ),
 }
 
 
@@ -257,13 +407,21 @@ def parse_scan_times(s3_key: str) -> tuple[datetime, datetime]:
 # ---------------------------------------------------------------------------
 # Spatial extraction helpers (scale/offset/fill — reused GOES/AHI semantics)
 # ---------------------------------------------------------------------------
-def _apply_scale_offset(raw: np.ndarray, attrs: dict[str, Any]) -> np.ndarray:
-    """Apply ``_FillValue`` -> NaN plus ``scale_factor``/``add_offset``.
+def _apply_scale_offset(
+    raw: np.ndarray, attrs: dict[str, Any], *, valid_range_filter: bool = False
+) -> np.ndarray:
+    """Apply ``_FillValue`` -> NaN, optional ``valid_range`` mask, then scaling.
 
     JRR CloudMask is raw int8 categories with no scale/offset and
-    CloudProbability is raw float32 [0,1], so this is mostly a fill mask + float
-    cast; the scale/offset branch is kept for forward-compat with any scaled JRR
-    derived product added later.
+    CloudProbability is raw float32 [0,1], so for those this is a fill mask +
+    float cast. Scaled products (JPSSRR_LST: raw*0.005+200 -> Kelvin) ALSO
+    honor the raw-unit ``valid_range`` attr when the registry entry opts in
+    (``valid_range_filter=True``) — mirroring
+    ``_goes_extract._apply_valid_range``: CF-1.8 ``valid_range`` is in RAW
+    dtype units BEFORE scaling, and without the mask an invalid raw (e.g. LST
+    30000 > 28600) scales into a plausible-looking physical value (350 K) and
+    enters rows as clean data (Codex P2). Products whose ``valid_range`` is
+    already in physical units (AOD550) keep ``valid_range_filter=False``.
     """
     fill = attrs.get("_FillValue")
     if isinstance(fill, np.ndarray):
@@ -276,6 +434,12 @@ def _apply_scale_offset(raw: np.ndarray, attrs: dict[str, Any]) -> np.ndarray:
     if fill_mask is not None:
         arr = np.where(fill_mask, np.nan, arr)
 
+    if valid_range_filter:
+        vr = attrs.get("valid_range")
+        if vr is not None and np.size(vr) == 2:
+            lo, hi = float(np.asarray(vr).reshape(-1)[0]), float(np.asarray(vr).reshape(-1)[1])
+            arr = np.where((arr < lo) | (arr > hi), np.nan, arr)
+
     scale = float(_scalar(attrs.get("scale_factor", 1.0)))
     offset = float(_scalar(attrs.get("add_offset", 0.0)))
     if scale != 1.0 or offset != 0.0:
@@ -423,7 +587,7 @@ def _extract_from_dataset(
 
         raw_val = var.isel(Rows=pixel_row, Columns=pixel_col).values
         raw_arr = np.array([[raw_val]])
-        scaled = _apply_scale_offset(raw_arr, var_attrs)
+        scaled = _apply_scale_offset(raw_arr, var_attrs, valid_range_filter=pv.valid_range_filter)
         pixel_value_scaled = float(scaled[0, 0])
 
         out_records.append(

packages/weather/src/mostlyright/weather/_fetchers/_viirs_s3.py

@@ -67,16 +67,57 @@
 #: (the operational launch dates of each JPSS platform: Suomi-NPP 2011-10,
 #: NOAA-20 2017-11, NOAA-21 2022-11; clamped to the start of the launch year so
 #: a pre-launch day returns [] without any I/O).
-# JRR-CloudMask floors on the NODD buckets, LIVE-LISTED 2026-07-05 (parity
-# audit): the platforms flew years earlier, but the JRR product objects only
-# begin on these dates — the old platform-launch floors created phantom windows
-# (viirs-npp: ~11.5 years of guaranteed-empty listings).
+# PLATFORM floors = earliest ANY registered product exists on that bucket
+# (tier-2 fix, live-listed 2026-07-06). The 2026-07-05 parity-audit floors were
+# CloudMask's product dates masquerading as platform floors — correct while
+# CloudMask was the only product, but they would have silently clipped
+# VIIRS-JRR-AOD's 2022+ archive on snpp/n20. Product specifics now live in
+# _PRODUCT_AVAILABLE_SINCE; the effective floor is max(platform, product) via
+# _available_since_for().
 _AVAILABLE_SINCE: dict[str, date] = {
-    "viirs-npp": date(2023, 6, 20),
-    "viirs-n20": date(2023, 2, 19),
-    "viirs-n21": date(2023, 11, 13),
+    "viirs-npp": date(2022, 1, 1),
+    "viirs-n20": date(2022, 1, 1),
+    "viirs-n21": date(2023, 1, 1),
 }
 
+# Per-(satellite, product) floors, LIVE-LISTED (year-dir granularity, floored
+# conservatively EARLY — a too-early floor only wastes one empty day-listing;
+# a too-late one silently loses data):
+#   - CloudMask: exact first-object dates (parity audit 2026-07-05).
+#   - AOD: snpp/n20 year dirs from 2022, n21 from 2023.
+#   - CloudHeight: all platforms from 2023.
+#   - LST / SnowCover: first objects ~Dec 2024 (n20 verified 2024/12); floored
+#     at 2024-11-01 for margin.
+_PRODUCT_AVAILABLE_SINCE: dict[tuple[str, str], date] = {
+    ("viirs-npp", "VIIRS-JRR-CloudMask"): date(2023, 6, 20),
+    ("viirs-n20", "VIIRS-JRR-CloudMask"): date(2023, 2, 19),
+    ("viirs-n21", "VIIRS-JRR-CloudMask"): date(2023, 11, 13),
+    ("viirs-npp", "VIIRS-JRR-CloudHeight"): date(2023, 1, 1),
+    ("viirs-n20", "VIIRS-JRR-CloudHeight"): date(2023, 1, 1),
+    ("viirs-n21", "VIIRS-JRR-CloudHeight"): date(2023, 1, 1),
+    ("viirs-npp", "JPSSRR_LST"): date(2024, 11, 1),
+    ("viirs-n20", "JPSSRR_LST"): date(2024, 11, 1),
+    ("viirs-n21", "JPSSRR_LST"): date(2024, 11, 1),
+    ("viirs-npp", "JPSSRR_SnowCover"): date(2024, 11, 1),
+    ("viirs-n20", "JPSSRR_SnowCover"): date(2024, 11, 1),
+    ("viirs-n21", "JPSSRR_SnowCover"): date(2024, 11, 1),
+}
+
+
+def _available_since_for(satellite: str, product: str) -> date | None:
+    """Effective availability floor: max(platform floor, product floor).
+
+    Mirrors _goes_s3._available_since_for (PR #103). Unknown products fall
+    back to the platform floor (fail-open to cheap empty listings, never to
+    silent data loss).
+    """
+    sat = _AVAILABLE_SINCE.get(satellite)
+    prod = _PRODUCT_AVAILABLE_SINCE.get((satellite, product))
+    if sat is None:
+        return prod
+    return max(sat, prod) if prod is not None else sat
+
+
 # Conservative top-level S3 call-rate cap (shared discipline with GOES/Himawari).
 _VIIRS_S3_RATE_HZ = 20.0
 _MAX_S3_RETRIES = 3
@@ -182,7 +223,7 @@ def list_product_keys(
     returns ``[]`` without I/O). ``mirror`` is accepted for handler signature
     uniformity but VIIRS is AWS-only.
     """
-    available_since = _AVAILABLE_SINCE.get(satellite)
+    available_since = _available_since_for(satellite, product)
     if available_since is not None and day < available_since:
         return []
 

packages/weather/src/mostlyright/weather/satellite/_backfill.py

@@ -1056,7 +1056,15 @@ def _progress_key(satellite: str, product: str, station: str, year: int, month:
     only in ``product`` or ``station``. The 2i key dropped product AND station
     (``"{satellite}_{year}_{month}"``), causing silent data loss on resume.
     """
-    return f"{satellite}_{product}_{station}_{year:04d}_{month:02d}"
+    # Product codes may contain underscores (JPSSRR_LST / JPSSRR_SnowCover,
+    # tier-2 VIIRS) but "_" is this key's FIELD SEPARATOR and the validator's
+    # product segment is deliberately underscore-free — sanitize to "-" in
+    # key-space (Codex P2: an unsanitized key failed _PROGRESS_KEY_RE on the
+    # very next resume load -> ProgressCorrupt on the run's own file). Keys are
+    # opaque full-string identities (never parsed back), and build + skip-check
+    # share this function, so the transform is loss-free; no registered
+    # products differ only by "_" vs "-".
+    return f"{satellite}_{product.replace('_', '-')}_{station}_{year:04d}_{month:02d}"
 
 
 def _validate_progress(progress: dict[str, Any]) -> dict[str, str]:

packages/weather/src/mostlyright/weather/satellite/_sources.py

@@ -107,7 +107,20 @@ def _viirs_products() -> frozenset[str]:
 
         return frozenset(_KNOWN_PRODUCTS)
     except ImportError:
-        return frozenset({"VIIRS-JRR-CloudMask"})
+        # Keep in sync with _viirs_extract.PRODUCTS — the drift-guard test in
+        # test_satellite_sources.py pins this fallback equal to the heavy
+        # registry (same class as the Codex P2 on the DSRC registration: a
+        # base-install/hosted request for a registered product was rejected
+        # as unknown).
+        return frozenset(
+            {
+                "VIIRS-JRR-CloudMask",
+                "VIIRS-JRR-AOD",
+                "VIIRS-JRR-CloudHeight",
+                "JPSSRR_LST",
+                "JPSSRR_SnowCover",
+            }
+        )
 
 
 def _meteosat_products() -> frozenset[str]:

packages/weather/tests/test_satellite_backfill.py

@@ -1864,3 +1864,25 @@ def test_west_station_2024_native_gap_falls_back_to_east(self) -> None:
         sats = {s[0] for s in slices}
         assert "goes16" not in sats
         assert sats == {"goes18"}
+
+
+class TestUnderscoreProductProgressKeys:
+    """Codex P2 (tier-2 VIIRS): JPSSRR_LST/_SnowCover contain the key field
+    separator — keys must sanitize product underscores so a resume can load
+    the file it just wrote."""
+
+    def test_key_sanitizes_underscores_and_validates(self) -> None:
+        k = _backfill._progress_key("viirs-n20", "JPSSRR_LST", "KNYC", 2024, 12)
+        assert k == "viirs-n20_JPSSRR-LST_KNYC_2024_12"
+        assert _backfill._PROGRESS_KEY_RE.match(k), "sanitized key must validate"
+        # Round-trip through the validator exactly as a resume load would.
+        out = _backfill._validate_progress({"__version__": 1, k: "completed"})
+        assert out == {k: "completed"}
+
+    def test_key_identity_stable_for_skip_checks(self) -> None:
+        a = _backfill._progress_key("viirs-n20", "JPSSRR_SnowCover", "EGLC", 2025, 1)
+        b = _backfill._progress_key("viirs-n20", "JPSSRR_SnowCover", "EGLC", 2025, 1)
+        assert a == b
+        # and distinct from the sibling product
+        c = _backfill._progress_key("viirs-n20", "JPSSRR_LST", "EGLC", 2025, 1)
+        assert a != c

packages/weather/tests/test_satellite_sources.py

@@ -193,3 +193,27 @@ def test_goes_fallback_product_set_matches_heavy_registry() -> None:
             break
     assert literals, "could not locate the fallback set literal"
     assert literals == set(_KNOWN_PRODUCTS)
+
+
+def test_viirs_fallback_product_set_matches_heavy_registry() -> None:
+    """Same drift guard for the VIIRS fallback (tier-2 registrations)."""
+    import ast
+    import inspect
+    import textwrap
+
+    _viirs_extract = pytest.importorskip(
+        "mostlyright.weather._fetchers._viirs_extract",
+        reason="drift guard needs the [satellite] extra (heavy PRODUCTS registry)",
+    )
+    src = textwrap.dedent(inspect.getsource(_sources._viirs_products))
+    literals: set[str] = set()
+    for node in ast.walk(ast.parse(src)):
+        if isinstance(node, ast.Set):
+            literals = {
+                el.value
+                for el in node.elts
+                if isinstance(el, ast.Constant) and isinstance(el.value, str)
+            }
+            break
+    assert literals, "could not locate the fallback set literal"
+    assert literals == set(_viirs_extract._KNOWN_PRODUCTS)