Accept NOAA-omitted freezing fields in RRFS/REFS decode validation

docs/virtual_datasets.md

@@ -138,7 +138,7 @@ The final update pod validates each operational update. A dataset lists every ch
 Two virtual checks ship in `validation.py`; both are bounded so the whole validation stays well under a couple of minutes even on a large ensemble dataset:
 
 - **`CheckVirtualManifestCompleteness`** — the virtual analog of the materialized `CheckExpectedShards`. Re-runs the operational filter (`region_job.source_file_coords()` + `filter_already_present`) over the recent window and checks, per append-dim position, the fraction of expected source files present against `min_present_fraction` (a tuple indexed newest-first; older positions held to its last value). Positions past the store's append-dim extent are skipped, since the window runs ahead of what the source has published: a wholly un-ingested recent stretch is `CheckCurrentData`'s job, while an interior gap still fails here. `(1.0,)` requires every remaining append-dim position fully ingested; `(0.5, 1.0)` lets the newest be half-published (e.g. GEFS 35-day's slow long lead times) while older ones must be complete. The last tier must be 1.0 (asserted) since it holds for every older append-dim position; loosen only the leading tiers, which cover positions the source may still be filling in. Fails — never silently passes — if the window is shorter than the threshold tiers. Reusing the dataset's own coord generation means structural absences (e.g. hour-0 accumulated vars) are already excluded — no false positives. Cheap: one ref-existence probe per file, no decode. `include_vars` / `exclude_vars` narrow the check to the source files carrying those variables, so variables publishing on different schedules can be checked separately, each held to a whole 1.0 from the position where its own files are expected; the split must fall on source-file boundaries (asserted), since presence is probed through each file's representative variable.
-- **`CheckVirtualDecodeHealth`** — where completeness checks references *exist*, this checks the ones that exist actually decode. It keeps only the source files present in the manifest (`filter_already_present`) — so a not-yet-published ref is never mistaken for a decode failure — and decodes a bounded sample (first + last + interior lead times across every ensemble member). `positions` (an int, default 1) targets that many of the newest positions *with data*, so a broken newest reference is caught at the very next validation rather than a cycle later; use more than 1 when the newest position does not carry every variable. `positions="all"` covers the whole window, optionally capped to `max_positions` evenly spaced positions for an offline sweep. A variable fails if any sampled chunk errors or all of its sampled chunks decode entirely NaN; the check fails — never silently passes — when no references are present. `allow_all_nan_vars` explicitly names exceptional variables whose sampled chunks may legitimately all be NaN; they still fail on decode errors or missing sampled references. The full-archive NaN/coverage scan stays offline (one full-message decode per chunk across all leads × members is hours).
+- **`CheckVirtualDecodeHealth`** — where completeness checks references *exist*, this checks the ones that exist actually decode. It keeps only the source files present in the manifest (`filter_already_present`) — so a not-yet-published ref is never mistaken for a decode failure — and decodes a bounded sample (first + last + interior lead times across every ensemble member). `positions` (an int, default 1) targets that many of the newest positions *with data*, so a broken newest reference is caught at the very next validation rather than a cycle later; use more than 1 when the newest position does not carry every variable. `positions="all"` covers the whole window, optionally capped to `max_positions` evenly spaced positions for an offline sweep. A variable fails if any sampled chunk errors or all of its sampled chunks decode entirely NaN; the check fails — never silently passes — when no references are present. `allow_all_nan_vars` explicitly names exceptional variables whose sampled chunks may legitimately all be NaN; they still fail on decode errors or missing sampled references. `allow_missing_vars` additionally permits all-NaN samples caused by absent references for explicitly named source omissions; present all-NaN references and decode errors still fail. The full-archive NaN/coverage scan stays offline (one full-message decode per chunk across all leads × members is hours).
 
 Tuning (completeness's `min_present_fraction`; decode health's `positions`, `sampled_leads`, `max_workers`) is set where the check is listed in `validators()` — one place. The materialized `CheckExpectedShards` does not carry over — a virtual store holds chunk references, not shards. `CheckReplicaMatchesPrimary` does apply: any dataset declaring a replica gets it, and on a virtual pair it reads both manifests' references back to catch a replica whose references diverge from the primary's.
 

src/reformatters/common/validation.py

@@ -1055,7 +1055,8 @@ class CheckVirtualDecodeHealth(Validator):
 
     `allow_all_nan_vars` explicitly names variables that may legitimately carry their
     missing state across every sampled chunk. They still fail on decode errors or missing
-    sampled chunk references.
+    sampled chunk references. `allow_missing_vars` also permits absent references for
+    named source-omitted variables; present all-NaN references and decode errors still fail.
     """
 
     positions: int | Literal["all"] = 1
@@ -1064,6 +1065,7 @@ class CheckVirtualDecodeHealth(Validator):
     max_positions: int | None = None
     max_workers: int = 32
     allow_all_nan_vars: Sequence[str] = ()
+    allow_missing_vars: Sequence[str] = ()
     # Offline opt-in. Given (var_path, out_loc), returns presence by vertical label
     # (None for root variables). A variable with no reference at a position is skipped
     # (not decoded, not a failure) -- reference existence is the availability check's
@@ -1091,12 +1093,7 @@ def check(self, context: ValidationContext) -> ValidationResult:
         store = context.store
         assert isinstance(store, IcechunkStore)
         ds = context.ds
-        unknown = sorted(set(self.allow_all_nan_vars) - set(context.known_var_paths()))
-        if unknown:
-            raise ValueError(
-                f"{type(self).__name__} names unknown allow_all_nan_vars {unknown}. "
-                f"Known variable paths: {sorted(context.known_var_paths())}"
-            )
+        self._validate_allowlists(context.known_var_paths())
         append_dim = region_job.append_dim
         candidates = region_job.source_file_coords()
         if not candidates:
@@ -1137,8 +1134,7 @@ def check(self, context: ValidationContext) -> ValidationResult:
                     )
                     if error is not None and var_path not in first_error:
                         first_error[var_path] = error
-                if self.reference_presence is not None:
-                    no_reference_vars |= skipped
+                no_reference_vars |= skipped
 
         problems = (
             ["No sampled variable had a reference at the selected positions"]
@@ -1154,29 +1150,40 @@ def check(self, context: ValidationContext) -> ValidationResult:
             ):
                 problems.append(f"{var_path}: every sampled chunk decoded entirely NaN")
 
+        missing_summary = (
+            f" ({len(no_reference_vars)} variable(s) had no reference at sampled "
+            f"positions: {', '.join(sorted(no_reference_vars))})"
+            if no_reference_vars
+            else ""
+        )
         target_label = ", ".join(str(p) for p in sorted(targets))
         if problems:
             return ValidationResult(
                 passed=False,
                 checked_count=decoded_refs,
                 message=f"Decode health failures at {append_dim}={target_label}:\n"
-                + "\n".join(f"- {p}" for p in problems),
+                + "\n".join(f"- {p}" for p in problems)
+                + missing_summary,
             )
         message = (
             f"Decoded {len(to_decode)} present source files across "
             f"{len(min_nan_fraction)} variables at {append_dim}={target_label} "
             "— all readable"
         )
-        if self.reference_presence is not None and no_reference_vars:
-            message += (
-                f" ({len(no_reference_vars)} variable(s) had no reference at sampled "
-                "positions — reference existence is reported by the "
-                "availability/manifest check)"
-            )
+        message += missing_summary
         return ValidationResult(
             passed=True, message=message, checked_count=decoded_refs
         )
 
+    def _validate_allowlists(self, known_var_paths: Sequence[str]) -> None:
+        for field in ("allow_all_nan_vars", "allow_missing_vars"):
+            unknown = sorted(set(getattr(self, field)) - set(known_var_paths))
+            if unknown:
+                raise ValueError(
+                    f"{type(self).__name__} names unknown {field} {unknown}. "
+                    f"Known variable paths: {sorted(known_var_paths)}"
+                )
+
     def _select_targets(self, present_positions: Sequence[Any]) -> set[Any]:
         if isinstance(self.positions, int):
             return set(present_positions[-self.positions :])
@@ -1235,6 +1242,15 @@ def _decode_coord(
                 continue
 
             nan_fraction = float(np.isnan(values).mean())
+            if (
+                nan_fraction >= 1.0
+                and var.path in self.allow_missing_vars
+                and not self._sampled_refs_present(
+                    da, loc, var, region_job, store, group, require_all=False
+                )
+            ):
+                skipped.add(var.path)
+                continue
             error = None
             if (
                 nan_fraction >= 1.0
@@ -1255,6 +1271,8 @@ def _sampled_refs_present(
         region_job: VirtualRegionJob[Any, Any],
         store: IcechunkStore,
         group: zarr.Group,
+        *,
+        require_all: bool = True,
     ) -> bool:
         locations: list[dict[Dim, CoordinateValue]] = [dict(out_loc)]
         spatial_dims = {"y", "x", "latitude", "longitude"}
@@ -1278,7 +1296,8 @@ def _sampled_refs_present(
             assert index is not None
             encoded = metadata.chunk_key_encoding.encode_chunk_key(index)
             keys.append(f"{array.path}/{encoded}")
-        return all(zarr.core.sync.sync(store.exists(key)) for key in keys)
+        present = (zarr.core.sync.sync(store.exists(key)) for key in keys)
+        return all(present) if require_all else any(present)
 
     def _sample_leads(
         self, coords: Sequence[SourceFileCoord]

src/reformatters/noaa/refs/forecast_3_hourly_virtual/dynamical_dataset.py

@@ -1,10 +1,31 @@
+from typing import ClassVar
+
 from reformatters.noaa.refs.dynamical_dataset import NoaaRefsDataset
 
 from .region_job import NoaaRefsForecast3HourlyVirtualRegionJob
 from .template_config import NoaaRefsForecast3HourlyVirtualTemplateConfig
 
 
 class NoaaRefsForecast3HourlyVirtualDataset(NoaaRefsDataset):
+    # NOAA omits these messages starting with the 2026-10-15 12Z cycle.
+    allow_missing_vars: ClassVar[tuple[str, ...]] = (
+        "neighborhood_probability_of_freezing_rain_3_hour_surface_above_0p254",
+        "neighborhood_probability_of_freezing_rain_3_hour_surface_above_2p54",
+        "neighborhood_probability_of_freezing_rain_3_hour_surface_above_6p35",
+        "neighborhood_probability_of_freezing_rain_6_hour_surface_above_0p254",
+        "neighborhood_probability_of_freezing_rain_6_hour_surface_above_12p7",
+        "neighborhood_probability_of_freezing_rain_6_hour_surface_above_2p54",
+        "neighborhood_probability_of_freezing_rain_6_hour_surface_above_6p35",
+        "neighborhood_probability_of_freezing_rain_12_hour_surface_above_0p254",
+        "neighborhood_probability_of_freezing_rain_12_hour_surface_above_12p7",
+        "neighborhood_probability_of_freezing_rain_12_hour_surface_above_2p54",
+        "neighborhood_probability_of_freezing_rain_12_hour_surface_above_6p35",
+        "neighborhood_probability_of_freezing_rain_24_hour_surface_above_0p254",
+        "neighborhood_probability_of_freezing_rain_24_hour_surface_above_12p7",
+        "neighborhood_probability_of_freezing_rain_24_hour_surface_above_2p54",
+        "neighborhood_probability_of_freezing_rain_24_hour_surface_above_25p4",
+        "neighborhood_probability_of_freezing_rain_24_hour_surface_above_6p35",
+    )
     template_config: NoaaRefsForecast3HourlyVirtualTemplateConfig = (
         NoaaRefsForecast3HourlyVirtualTemplateConfig()
     )

src/reformatters/noaa/refs/forecast_hourly_virtual/dynamical_dataset.py

@@ -1,10 +1,18 @@
+from typing import ClassVar
+
 from reformatters.noaa.refs.dynamical_dataset import NoaaRefsDataset
 
 from .region_job import NoaaRefsForecastHourlyVirtualRegionJob
 from .template_config import NoaaRefsForecastHourlyVirtualTemplateConfig
 
 
 class NoaaRefsForecastHourlyVirtualDataset(NoaaRefsDataset):
+    # NOAA omits these messages starting with the 2026-10-15 12Z cycle.
+    allow_missing_vars: ClassVar[tuple[str, ...]] = (
+        "neighborhood_probability_of_freezing_rain_1_hour_surface_above_0p254",
+        "neighborhood_probability_of_freezing_rain_1_hour_surface_above_2p54",
+        "neighborhood_probability_of_freezing_rain_1_hour_surface_above_6p35",
+    )
     template_config: NoaaRefsForecastHourlyVirtualTemplateConfig = (
         NoaaRefsForecastHourlyVirtualTemplateConfig()
     )

src/reformatters/noaa/rrfs/dynamical_dataset.py

@@ -30,6 +30,7 @@ class NoaaRrfsDataset(
     update_job_deadline: timedelta | None = None
     min_present_fraction: ClassVar[tuple[float, ...]] = (1.0,)
     allow_all_nan_vars: ClassVar[tuple[str, ...]] = ()
+    allow_missing_vars: ClassVar[tuple[str, ...]] = ()
     virtual_poll_deadline_grace: ClassVar[timedelta] = timedelta(minutes=5)
     icechunk_virtual_config: IcechunkVirtualConfig = Field(
         default_factory=lambda: IcechunkVirtualConfig(
@@ -85,6 +86,8 @@ def validators(self) -> Sequence[validation.Validator]:
                 min_present_fraction=self.min_present_fraction
             ),
             validation.CheckVirtualDecodeHealth(
-                max_workers=2, allow_all_nan_vars=self.allow_all_nan_vars
+                max_workers=2,
+                allow_all_nan_vars=self.allow_all_nan_vars,
+                allow_missing_vars=self.allow_missing_vars,
             ),
         )

src/reformatters/noaa/rrfs/forecast_18_hour_virtual/dynamical_dataset.py

@@ -15,6 +15,13 @@ class NoaaRrfsForecast18HourVirtualDataset(NoaaRrfsDataset):
     update_deadline: timedelta = timedelta(minutes=60)
     # Snow density is all NaN when the air 300 m above ground is 5 °C or warmer everywhere.
     allow_all_nan_vars: ClassVar[tuple[str, ...]] = ("new_snow_density_surface",)
+    # NOAA omits these messages starting with the 2026-10-15 12Z cycle.
+    allow_missing_vars: ClassVar[tuple[str, ...]] = (
+        "frozen_precipitation_run_total_surface",
+        "freezing_rain_run_total_surface",
+        "frozen_precipitation_surface",
+        "freezing_rain_surface",
+    )
     template_config: NoaaRrfsForecast18HourVirtualTemplateConfig = (
         NoaaRrfsForecast18HourVirtualTemplateConfig()
     )

src/reformatters/noaa/rrfs/forecast_84_hour_virtual/dynamical_dataset.py

@@ -15,6 +15,13 @@ class NoaaRrfsForecast84HourVirtualDataset(NoaaRrfsDataset):
     update_deadline: timedelta = timedelta(minutes=135)
     # Snow density is all NaN when the air 300 m above ground is 5 °C or warmer everywhere.
     allow_all_nan_vars: ClassVar[tuple[str, ...]] = ("new_snow_density_surface",)
+    # NOAA omits these messages starting with the 2026-10-15 12Z cycle.
+    allow_missing_vars: ClassVar[tuple[str, ...]] = (
+        "frozen_precipitation_run_total_surface",
+        "freezing_rain_run_total_surface",
+        "frozen_precipitation_surface",
+        "freezing_rain_surface",
+    )
     template_config: NoaaRrfsForecast84HourVirtualTemplateConfig = (
         NoaaRrfsForecast84HourVirtualTemplateConfig()
     )

src/reformatters/noaa/rrfs/forecast_sub_hourly_virtual/dynamical_dataset.py

@@ -17,6 +17,11 @@ class NoaaRrfsForecastSubHourlyVirtualDataset(NoaaRrfsDataset):
     virtual_poll_deadline_grace: ClassVar[timedelta] = timedelta(seconds=60)
     # 00Z and 12Z source publication can continue past the poll deadline.
     min_present_fraction: ClassVar[tuple[float, ...]] = (6 / 18, 1.0)
+    # NOAA omits these messages starting with the 2026-10-15 12Z cycle.
+    allow_missing_vars: ClassVar[tuple[str, ...]] = (
+        "frozen_precipitation_run_total_surface",
+        "freezing_rain_run_total_surface",
+    )
     template_config: NoaaRrfsForecastSubHourlyVirtualTemplateConfig = (
         NoaaRrfsForecastSubHourlyVirtualTemplateConfig()
     )

src/reformatters/noaa/rrfs_ens/forecast_virtual/dynamical_dataset.py

@@ -1,4 +1,5 @@
 from datetime import timedelta
+from typing import ClassVar
 
 from pydantic import NonNegativeInt
 
@@ -12,6 +13,11 @@ class NoaaRrfsEnsForecastVirtualDataset(NoaaRrfsDataset):
     # Files publish near init +81-212 min; polling starts before the first lead.
     update_offset_minutes: NonNegativeInt = 75
     update_deadline: timedelta = timedelta(minutes=160)
+    # NOAA omits these messages starting with the 2026-10-15 12Z cycle.
+    allow_missing_vars: ClassVar[tuple[str, ...]] = (
+        "freezing_rain_run_total_surface",
+        "freezing_rain_surface",
+    )
     template_config: NoaaRrfsEnsForecastVirtualTemplateConfig = (
         NoaaRrfsEnsForecastVirtualTemplateConfig()
     )