From 9d19f1baac863947528646b2bbc0de733ce04b8e Mon Sep 17 00:00:00 2001 From: Alden Keefe Sampson Date: Sun, 12 Jul 2026 23:28:33 -0400 Subject: [PATCH 01/23] Drop the std column from the point time series table when empty (#741) --- src/scripts/validation/summary.py | 35 +++++++++++++++++++++---------- 1 file changed, 24 insertions(+), 11 deletions(-) diff --git a/src/scripts/validation/summary.py b/src/scripts/validation/summary.py index 4589f92a2..3d3c87416 100644 --- a/src/scripts/validation/summary.py +++ b/src/scripts/validation/summary.py @@ -117,17 +117,30 @@ def _value_ts_table(stats: VariableStats, ctx: RunContext) -> list[str]: header += f", member={ctx.value_ts_member}" if stats.level_dim is not None: header += f", {stats.level_dim}={stats.level_value:g}" - return [ - header, - "", - "| Point | min | mean | std | max |", - "|---|---|---|---|---|", - f"| P1 | {_fmt_num(stats.value_min_p1)} | {_fmt_num(stats.value_mean_p1)} " - f"| {_fmt_num(stats.value_std_p1)} | {_fmt_num(stats.value_max_p1)} |", - f"| P2 | {_fmt_num(stats.value_min_p2)} | {_fmt_num(stats.value_mean_p2)} " - f"| {_fmt_num(stats.value_std_p2)} | {_fmt_num(stats.value_max_p2)} |", - "", - ] + # std spreads over the non-time dims (lead / member); virtual and analysis stores + # reduce to a single value per timestep, so there is no std to show — drop the column. + has_std = stats.value_std_p1 is not None or stats.value_std_p2 is not None + if has_std: + rows = [ + "| Point | min | mean | std | max |", + "|---|---|---|---|---|", + f"| P1 | {_fmt_num(stats.value_min_p1)} | {_fmt_num(stats.value_mean_p1)} " + f"| {_fmt_num(stats.value_std_p1)} | {_fmt_num(stats.value_max_p1)} |", + f"| P2 | {_fmt_num(stats.value_min_p2)} | {_fmt_num(stats.value_mean_p2)} " + f"| {_fmt_num(stats.value_std_p2)} | {_fmt_num(stats.value_max_p2)} |", + ] + else: + rows = [ + "| Point | min | mean | max |", + "|---|---|---|---|", + _stats_row( + "P1", stats.value_min_p1, stats.value_mean_p1, stats.value_max_p1 + ), + _stats_row( + "P2", stats.value_min_p2, stats.value_mean_p2, stats.value_max_p2 + ), + ] + return [header, "", *rows, ""] def _temporal_table(stats: VariableStats, ctx: RunContext) -> list[str]: From aa146c5e3ee205d1138e60d44318877fc970d781 Mon Sep 17 00:00:00 2001 From: Alden Keefe Sampson Date: Mon, 13 Jul 2026 00:05:01 -0400 Subject: [PATCH 02/23] Render per-variable plots with availability last (#742) --- src/scripts/validation/render.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/scripts/validation/render.py b/src/scripts/validation/render.py index 2bf8a2065..5a843aed7 100644 --- a/src/scripts/validation/render.py +++ b/src/scripts/validation/render.py @@ -22,11 +22,13 @@ def _extract_per_var_html(html: str) -> list[str]: return re.findall(r"

([^<]+)

", html) +# Order mirrors the per-variable text/table sections: value time series, spatial, +# temporal, then availability last. _PLOT_TYPES = ( - ("availability", "availability over append dim"), ("value_timeseries", "full-period value time series"), ("spatial", "spatial comparison"), ("temporal", "time series comparison"), + ("availability", "availability over append dim"), ) From fd6167f1571d7b9999c360d6bcae89628e93316e Mon Sep 17 00:00:00 2001 From: Alden Keefe Sampson Date: Mon, 13 Jul 2026 12:48:47 -0400 Subject: [PATCH 03/23] HRRR-virtual ops latency: parallel commit flush + smaller manifest windows (~15x) (#744) * HRRR-spatial ops latency: parallel commit flush + smaller manifest windows Operational update commits were p50 ~60s / max ~500s vs GEFS-spatial's ~5s. Profiled locally by replaying one cycle's file arrivals against a store seeded to prod manifest fill (26M refs, M=3.1k), with icechunk.testing.LatencyStorage simulating S3. Commit cost decomposes as, per changed array, one manifest read-modify-write: ~1 GET + ~1.2 PUTs serialized (max_concurrent_nodes=1) plus ~0.1s single-threaded CPU per MiB of active manifest window (decode + merge + re-serialize + zstd). An sfc file touches 142 root arrays x ~1.7 MiB windows = ~24s CPU + ~18s round trips. Neither lever alone fixes it: concurrency at 1.5 vCPU stays CPU-bound (measured 26s), smaller windows alone stay round-trip-bound. Changes (measured steady-state commits at 10ms simulated S3): - Pin icechunk fork (2.1.1 + Session.commit(max_concurrent_nodes) + O(1) manifest-info index) and pass max_concurrent_nodes=16 in commit_if_icechunk (env ICECHUNK_COMMIT_MAX_CONCURRENT_NODES). - Shrink HRRR-spatial manifest splits 2400/180/160 -> 240/45/40 along init_time (~190KiB/1.4MiB/1.6MiB full windows). Icechunk re-windows existing manifests on each array's next commit: one-time ~70s commit locally, longer on prod's full historical windows. M grows ~3.1k -> ~16.4k, covered by the fork's O(1) index. - Update pod cpu 1.5 -> 4: sfc/prs/nat commits 2.4/1.1/3.0s at 1.5 CPU vs 1.5/0.7/2.0s at 4 (vs prod today: 42/7.7/13.4s). - Rust toolchain in Dockerfile to build the git-pinned icechunk during uv sync. Not for main as-is: revert the pyproject pin + Dockerfile rust stage once the icechunk changes land upstream. GEFS-spatial needs no dataset change (its splits are already small); it gains the parallel flush automatically. Co-Authored-By: Claude Fable 5 * Moderate HRRR-spatial splits to 600/90/80 after manifest-count tradeoff review 240/45/40 minimized commit latency (1.6/0.7/2.1s) but at M ~16.4k manifests, ~10x whole-archive read amplification, and sub-200KiB manifest files. 600/90/80 keeps manifests at 0.55-3.2 MiB and M ~7.6k for sfc/prs/nat commits of 2.6/2.1/4.5s (c=16, 4 cpu, 10ms simulated S3; peak RSS 2.9G fits the pod), still ~10-15x faster than the current 42/7.7/13.4s. One-time re-window on the first commit per array: ~44s measured locally. Co-Authored-By: Claude Fable 5 * Pin icechunk fork tip: rewrite_manifests concurrency exposure Adds Repository.rewrite_manifests(max_concurrent_manifests=...) for the one-time manifest re-split (the Rust plumbing already existed; the binding hardcoded 1). Validated on the local harness: re-split of a prod-fill store in one atomic commit. Co-Authored-By: Claude Fable 5 * Trim fork pin to the minimum: max_concurrent_nodes exposure only Drop the O(1) manifest-info index from the deployed fork surface (~0.13s/commit at M~7.6k, near the noise floor at the 600/90/80 splits) to keep the prod test and the upstream PR to exactly the change that matters. The rewrite_manifests concurrency exposure is used only by the local one-time re-split wheel and is not deployed. Co-Authored-By: Claude Fable 5 * Hardcode commit manifest-write concurrency at 16 Drop the ICECHUNK_COMMIT_MAX_CONCURRENT_NODES env override; one less knob. Note in the pyproject pin comment to also remove the Dockerfile Rust stage when the pin reverts. Co-Authored-By: Claude Fable 5 --------- Co-authored-by: Claude Fable 5 --- .../manual-create-job-from-cronjob.yml | 4 ++-- deploy/Dockerfile | 10 ++++++++ docs/virtual_datasets.md | 11 ++++++--- pyproject.toml | 7 ++++++ src/reformatters/common/storage.py | 10 +++++++- .../dynamical_dataset.py | 24 +++++++++++-------- uv.lock | 19 +++------------ 7 files changed, 53 insertions(+), 32 deletions(-) diff --git a/.github/workflows/manual-create-job-from-cronjob.yml b/.github/workflows/manual-create-job-from-cronjob.yml index 1efb4db64..c6be6ca8e 100644 --- a/.github/workflows/manual-create-job-from-cronjob.yml +++ b/.github/workflows/manual-create-job-from-cronjob.yml @@ -34,10 +34,10 @@ name: 'Manual: Create Job from CronJob' - noaa-gfs-forecast-validate - noaa-hrrr-analysis-update - noaa-hrrr-analysis-validate - - noaa-hrrr-forecast-48-hour-virtual-update - - noaa-hrrr-forecast-48-hour-virtual-validate - noaa-hrrr-forecast-48-hour-update - noaa-hrrr-forecast-48-hour-validate + - noaa-hrrr-forecast-48-hour-virtual-update + - noaa-hrrr-forecast-48-hour-virtual-validate - noaa-mrms-conus-analysis-hourly-update - noaa-mrms-conus-analysis-hourly-validate - noaa-ndvi-cdr-analysis-update diff --git a/deploy/Dockerfile b/deploy/Dockerfile index c06690919..33ed6a825 100644 --- a/deploy/Dockerfile +++ b/deploy/Dockerfile @@ -20,6 +20,8 @@ apt-get install -qyy \ -o APT::Install-Suggests=false \ build-essential \ ca-certificates \ + curl \ + git \ python3-setuptools \ python3.14-dev \ build-essential \ @@ -47,6 +49,14 @@ ENV DISABLE_NUMCODECS_SSE2=1 \ ### End Build Prep +# TEST-ONLY: Rust toolchain so uv can compile the git-pinned icechunk from source +# during `uv sync`. Remove together with the pyproject git pin once the icechunk +# changes are released upstream. +RUN < None: icechunk_store.session.commit( - message=message, rebase_with=icechunk.ConflictDetector() + message=message, + rebase_with=icechunk.ConflictDetector(), + max_concurrent_nodes=_COMMIT_MAX_CONCURRENT_NODES, ) for store in replica_stores: diff --git a/src/reformatters/noaa/hrrr/forecast_48_hour_virtual/dynamical_dataset.py b/src/reformatters/noaa/hrrr/forecast_48_hour_virtual/dynamical_dataset.py index a04f899aa..6b8dce610 100644 --- a/src/reformatters/noaa/hrrr/forecast_48_hour_virtual/dynamical_dataset.py +++ b/src/reformatters/noaa/hrrr/forecast_48_hour_virtual/dynamical_dataset.py @@ -42,17 +42,21 @@ class NoaaHrrrForecast48HourVirtualDataset( _S3_LOCATION_PREFIX, icechunk.s3_store(region=_S3_BUCKET_REGION) ), ), - # Sized per group from measured ~16.4 bytes/ref so full manifests land - # comfortably under the reader budgets (single-level <= 3 MiB/var, vertical - # 5-8 MiB) while keeping the total manifest count low; see "Manifest - # splitting" in docs/virtual_datasets.md. Full-window sizes: single-level - # 2400 x 49 refs/init ~= 1.8 MiB, pressure 180 x 1911 ~= 5.4 MiB, - # model 160 x 2450 ~= 6.1 MiB. + # Sized for operational commit latency: each commit read-modify-writes the + # active window's manifest for every changed array (~0.1s CPU/MB, measured), + # so window bytes bound per-commit flush cost. Full-window sizes at ~16.4 + # bytes/ref: single-level 600 x 49 refs/init ~= 0.55 MiB, pressure 90 x 1911 + # ~= 2.8 MiB, model 80 x 2450 ~= 3.2 MiB — balancing commit latency + # (sfc/prs/nat ~2.6/2.1/4.5s measured at c=16, 4 cpu, 10ms simulated S3) + # against manifest count (M ~7.6k) and whole-archive read amplification; + # see "Manifest splitting" in docs/virtual_datasets.md. On first commit per + # array after a split-size change, icechunk re-windows that array's existing + # manifests (one-time, ~1 min measured). manifest_split=manifest_append_dim_split( split_size={ - r"^/pressure_level/": 180, - r"^/model_level/": 160, - None: 2400, + r"^/pressure_level/": 90, + r"^/model_level/": 80, + None: 600, }, dim="init_time", ), @@ -70,7 +74,7 @@ def operational_kubernetes_resources(self, image_tag: str) -> Sequence[CronJob]: pod_active_deadline=timedelta(hours=1, minutes=40), image=image_tag, dataset_id=self.dataset_id, - cpu="1.5", + cpu="4", memory="3.7G", secret_names=self.store_factory.k8s_secret_names(), ) diff --git a/uv.lock b/uv.lock index daac72838..7955e6e2d 100644 --- a/uv.lock +++ b/uv.lock @@ -549,24 +549,11 @@ wheels = [ [[package]] name = "icechunk" -version = "2.0.5" -source = { registry = "https://pypi.org/simple" } +version = "2.1.1" +source = { git = "https://github.com/aldenks/icechunk?subdirectory=icechunk-python&branch=expose-commit-max-concurrent-nodes-2.1.1#188615b19eb1feb5b18656f45b5c208877d5fc55" } dependencies = [ { name = "zarr" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/97/99/fabc9794c1f82e51b5c7c66301695b8fd920f72dee1726104dbdbd8df3e7/icechunk-2.0.5.tar.gz", hash = "sha256:50a2a44a1b561d3f2d3b5d19725c3759f300dc67225a2360fc793d894abfcab1", size = 3327412, upload-time = "2026-05-18T20:22:05.466Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/e3/47/0e29dd5248dbaddef6351e9807d61c4eacae325f3b1415a2bb0ce5b2e92e/icechunk-2.0.5-cp314-cp314-macosx_10_12_x86_64.whl", hash = "sha256:2d2369db67502118150612ad481468cc0ae1333ba5cf6084179da04591e566b2", size = 16841657, upload-time = "2026-05-18T20:22:45.594Z" }, - { url = "https://files.pythonhosted.org/packages/2f/c5/9652585ee78a0f242d2c92983a550990b6a39779d6d0a7c24ab82f293d39/icechunk-2.0.5-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:0554ef924b14f7d6369919c22f042a817f8dbc85d0b9efd793398e2d44786fa4", size = 15544742, upload-time = "2026-05-18T20:22:36.879Z" }, - { url = "https://files.pythonhosted.org/packages/59/2c/076e478b9b45616ef268bdb0473d6d0c8bfc4ad62224477c0c066b74ebe0/icechunk-2.0.5-cp314-cp314-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:43289b68bc3ca93804a8ffd96e356f509c87bfd35c3d8202382dd97febd9957d", size = 17240728, upload-time = "2026-05-18T20:22:26.306Z" }, - { url = "https://files.pythonhosted.org/packages/fa/06/8c0f3fb0df245f42d05d4b423a92766a466ad9e36711972da84196263d14/icechunk-2.0.5-cp314-cp314-manylinux_2_28_aarch64.whl", hash = "sha256:35812c0c3087688c422470610fa9f77f6edb2d5d7c3345e0610beb32c8028f96", size = 16883484, upload-time = "2026-05-18T20:22:02.374Z" }, - { url = "https://files.pythonhosted.org/packages/0f/6a/74440741ac30bb09c9d3fb74acb1332f218ae87faba24f6a81c8c575b9d2/icechunk-2.0.5-cp314-cp314-manylinux_2_28_armv7l.whl", hash = "sha256:ba30ea180b056c34fcf094b36fd7fb7cbd2521f778e5d1080ce9ab9b2f28204d", size = 16706325, upload-time = "2026-05-18T20:22:15.083Z" }, - { url = "https://files.pythonhosted.org/packages/4b/09/fe20275108f44a7ad8bbb904165feaf65f31796a8dad7baa764d86181fa9/icechunk-2.0.5-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:65075bce7674d2292344a661129fa987a579e5ab46c35862ef366b0c167e1066", size = 17099016, upload-time = "2026-05-18T20:22:56.368Z" }, - { url = "https://files.pythonhosted.org/packages/fb/f6/bd9b9d79b1a10a8382116ac018f3580f6e8021674fe53167829fba70bd7e/icechunk-2.0.5-cp314-cp314-musllinux_1_2_armv7l.whl", hash = "sha256:231f8a46de0949da8ffd0e59446342a1051e374276385ca9a529ab593e73c7ce", size = 16878446, upload-time = "2026-05-18T20:23:06.761Z" }, - { url = "https://files.pythonhosted.org/packages/c6/af/025f9c303dca742c709a8c01f809479c3d22bdf630954f08eadb6eaace5b/icechunk-2.0.5-cp314-cp314-musllinux_1_2_i686.whl", hash = "sha256:65e514a6394b1ed5f3726d7da1cc3cafae8e23c7a6b243dc793b1eb876078813", size = 16955472, upload-time = "2026-05-18T20:23:16.865Z" }, - { url = "https://files.pythonhosted.org/packages/52/b3/7429c90e9a0512f6e11443ccbc9d4ee392757b53c8604db05578457ffac9/icechunk-2.0.5-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:e1a488e9162d64e05e995aa4015cc8118c68fc54d0ed5af4616627a66a4d1d01", size = 17646740, upload-time = "2026-05-18T20:23:30.02Z" }, - { url = "https://files.pythonhosted.org/packages/25/3c/8e086299cc1a779837e65e4152d03d02b457f8974bb388082fef20894b5e/icechunk-2.0.5-cp314-cp314-win_amd64.whl", hash = "sha256:95b1beb874ad287fcb99dfea29cd5218c795b5d9bca47b8f43ef78b8e6c5b572", size = 15945027, upload-time = "2026-05-18T20:23:40.81Z" }, -] [[package]] name = "idna" @@ -1365,7 +1352,7 @@ requires-dist = [ { name = "google-crc32c", specifier = ">=1.8.0" }, { name = "gribberish", specifier = "==1.5.0" }, { name = "httpx", specifier = ">=0.28.1" }, - { name = "icechunk", specifier = "~=2.0" }, + { name = "icechunk", git = "https://github.com/aldenks/icechunk?subdirectory=icechunk-python&branch=expose-commit-max-concurrent-nodes-2.1.1" }, { name = "kubernetes", specifier = ">=36.0.1,<37" }, { name = "numba", specifier = "~=0.62" }, { name = "numcodecs", specifier = "~=0.16" }, From 396fdeb6369dcf9cd13956d0e73860fabe5dcbf2 Mon Sep 17 00:00:00 2001 From: Alden Keefe Sampson Date: Wed, 15 Jul 2026 08:14:22 -0400 Subject: [PATCH 04/23] docs: add canonical keep_mantissa_bits guidance to AGENTS.md (#745) --- AGENTS.md | 2 ++ docs/add_new_variable.md | 2 +- docs/dataset_integration_guide.md | 2 ++ 3 files changed, 5 insertions(+), 1 deletion(-) diff --git a/AGENTS.md b/AGENTS.md index 257722941..0ba477268 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -80,6 +80,8 @@ Categorical / flag variables set `flag_values` (the coded values) and `flag_mean Comment vs. review note. Put intrinsic, always-true variable facts (quirks, sentinel values, what the variable physically represents if not clear in the name/long_name) in the variable's `comment` attr so they travel with the data — these get no validation-report review note. Most common variables need no `comment` unless their interpretation is unusual. Put time-windowed characteristics of a specific archive (version-boundary behavior changes, historical low-quality windows, source outages) in the validation report's `### Review notes` (see [docs/validation.md](docs/validation.md) §3e) — these get no `comment`, since they would go stale in static template metadata as the archive grows. Each fact lives in exactly one place based on its kind. +Set each data variable's `keep_mantissa_bits` (float32 rounding for compression, or `"no-rounding"` to keep all 23) by convention: use 7 unless the variable is wind (6), a precipitation flux/rate (8), or a pressure variable with `units="Pa"` (11); match an existing equivalent variable rather than re-deriving. + ### RegionJob Base class: `src/reformatters/common/region_job.py`, commented example subclasses: `src/reformatters/example_{materialized|virtual}/region_job.py`. diff --git a/docs/add_new_variable.md b/docs/add_new_variable.md index 062a1497c..ace85c944 100644 --- a/docs/add_new_variable.md +++ b/docs/add_new_variable.md @@ -9,7 +9,7 @@ How to add a new data variable to an existing dataset. 1b. Add a new `DataVar` to your dataset’s `TemplateConfig.data_vars` (usually in `src/reformatters////template_config.py`). - **Name + externally visible attrs**: match existing naming/attrs used in this repo where possible; otherwise follow CF Conventions. Variable names generally follow the format `_`. - **Internal attrs**: derive from `gdalinfo` output on a representative source file (and GRIB index if relevant) - - **Encoding**: match existing variables, setting `keep_mantissa_bits` to 7 by default, 6 for wind variables, 8 for precipitation and 10 for pressure variables with units `pa`. + - **Encoding**: match existing variables, and follow the `keep_mantissa_bits` guidance in AGENTS.md. 1c. Regenerate the checked-in Zarr template metadata: diff --git a/docs/dataset_integration_guide.md b/docs/dataset_integration_guide.md index 24e0681a1..4f21b28af 100644 --- a/docs/dataset_integration_guide.md +++ b/docs/dataset_integration_guide.md @@ -76,6 +76,8 @@ Set `dataset_id` and `name` in `dataset_attributes` following the id/name conven Read the [chunk/shard layout tool](./chunk_shard_layout_tool.md) docs and use the tool to find chunk and shard sizes for your data variables. +Follow the `keep_mantissa_bits` guidance in AGENTS.md when setting each data variable's encoding. + Using the information in the `TemplateConfig`, `reformatters` writes the Zarr metadata for your dataset to `src/reformatters/$DATASET_PATH/templates/latest.zarr`. Run this command in your terminal to create or update the template based on the your `TemplateConfig` subclass: ```bash From a1a22d23aab283ce7fdce169fc3eacb6283ed64c Mon Sep 17 00:00:00 2001 From: Alden Keefe Sampson Date: Wed, 15 Jul 2026 10:22:08 -0400 Subject: [PATCH 05/23] Disable Karpenter consolidation for reformat/update pods (#747) Operational update workers aren't checkpointed, so a mid-run Karpenter consolidation eviction restarts the whole job from scratch, doubling wall-clock. This caused noaa-hrrr-analysis update runs to finish after their validation cronjob started, firing spurious validation alerts. Add a pod_annotations hook on Job (empty by default) and set the karpenter.sh/do-not-disrupt annotation on ReformatCronJob so Karpenter won't voluntarily evict a running update. Backfill and validation pods are unaffected. Co-authored-by: Claude Opus 4.8 (1M context) --- src/reformatters/common/kubernetes.py | 7 +++- tests/common/test_kubernetes.py | 46 ++++++++++++++++++++++++++- 2 files changed, 51 insertions(+), 2 deletions(-) diff --git a/src/reformatters/common/kubernetes.py b/src/reformatters/common/kubernetes.py index 4f5a01ecb..284878e06 100644 --- a/src/reformatters/common/kubernetes.py +++ b/src/reformatters/common/kubernetes.py @@ -36,6 +36,8 @@ class Job(pydantic.BaseModel): pod_active_deadline: timedelta = timedelta(hours=6) ttl: timedelta = timedelta(days=1) + pod_annotations: dict[str, str] = pydantic.Field(default_factory=dict) + secret_names: Sequence[str] = pydantic.Field(default_factory=list) @cached_property @@ -81,6 +83,7 @@ def as_kubernetes_object(self) -> dict[str, Any]: ] }, "template": { + "metadata": {"annotations": self.pod_annotations}, "spec": { "containers": [ { @@ -210,7 +213,7 @@ def as_kubernetes_object(self) -> dict[str, Any]: for secret_name in self.secret_names ], ], - } + }, }, "ttlSecondsAfterFinished": int(self.ttl.total_seconds()), }, @@ -250,6 +253,8 @@ class ReformatCronJob(CronJob): # Operational updates expect a single worker workers_total: int = 1 parallelism: int = 1 + # A mid-run eviction restarts the whole job, so opt out of consolidation. + pod_annotations: dict[str, str] = {"karpenter.sh/do-not-disrupt": "true"} class ValidationCronJob(CronJob): diff --git a/tests/common/test_kubernetes.py b/tests/common/test_kubernetes.py index 5b737bf9b..c76b713fd 100644 --- a/tests/common/test_kubernetes.py +++ b/tests/common/test_kubernetes.py @@ -10,6 +10,8 @@ from reformatters.common.config import Config, Env from reformatters.common.kubernetes import ( Job, + ReformatCronJob, + ValidationCronJob, _load_secret_from_kubernetes_api, load_secret, ) @@ -58,6 +60,7 @@ def test_as_kubernetes_object_comprehensive() -> None: ] }, "template": { + "metadata": {"annotations": {}}, "spec": { "containers": [ { @@ -161,7 +164,7 @@ def test_as_kubernetes_object_comprehensive() -> None: {"name": "aws-creds", "secret": {"secretName": "aws-creds"}}, {"name": "db-creds", "secret": {"secretName": "db-creds"}}, ], - } + }, }, "ttlSecondsAfterFinished": 86400, # default 24 hours } @@ -169,6 +172,47 @@ def test_as_kubernetes_object_comprehensive() -> None: assert k8s_obj["spec"] == expected_spec +def test_do_not_disrupt_annotation_only_on_reformat_jobs() -> None: + reformat = ReformatCronJob( + name="weather-data-update", + schedule="0 * * * *", + image="img:v1", + dataset_id="weather_data", + cpu="1", + memory="1Gi", + ) + validate = ValidationCronJob( + name="weather-data-validate", + schedule="0 * * * *", + image="img:v1", + dataset_id="weather_data", + cpu="1", + memory="1Gi", + ) + backfill = Job( + command=["backfill-kubernetes"], + image="img:v1", + dataset_id="weather_data", + cpu="1", + memory="1Gi", + workers_total=1, + parallelism=1, + ) + + def pod_template(obj: dict[str, Any], *, cron: bool) -> dict[str, Any]: + spec = obj["spec"]["jobTemplate"]["spec"] if cron else obj["spec"] + return spec["template"] + + def annotations(obj: dict[str, Any], *, cron: bool) -> dict[str, str]: + return pod_template(obj, cron=cron)["metadata"]["annotations"] + + assert annotations(reformat.as_kubernetes_object(), cron=True) == { + "karpenter.sh/do-not-disrupt": "true" + } + assert annotations(validate.as_kubernetes_object(), cron=True) == {} + assert annotations(backfill.as_kubernetes_object(), cron=False) == {} + + def test_kubernetes_job_name() -> None: """Test ensure that the job name is consistent across invocations""" job = Job( From 8ce68ae5f87954ef8e260aaff86fc05aa2a42849 Mon Sep 17 00:00:00 2001 From: Alden Keefe Sampson Date: Sun, 19 Jul 2026 08:48:42 -0400 Subject: [PATCH 06/23] Retry NCEI 5xx errors when listing NDVI source files (#754) --- .../noaa/ndvi_cdr/analysis/region_job.py | 12 +++++- .../noaa/ndvi_cdr/analysis/region_job_test.py | 43 +++++++++++++++++++ 2 files changed, 53 insertions(+), 2 deletions(-) diff --git a/src/reformatters/contrib/noaa/ndvi_cdr/analysis/region_job.py b/src/reformatters/contrib/noaa/ndvi_cdr/analysis/region_job.py index a8c9037df..6b77776b2 100644 --- a/src/reformatters/contrib/noaa/ndvi_cdr/analysis/region_job.py +++ b/src/reformatters/contrib/noaa/ndvi_cdr/analysis/region_job.py @@ -282,12 +282,20 @@ def _list_ncei_source_files(self, year: int) -> list[str]: """ ncei_url = f"{self.root_nc_url}/{year}/" - response = retry(lambda: requests.get(ncei_url, timeout=15)) + def get_listing() -> requests.Response: + response = requests.get(ncei_url, timeout=15) + # 404 is a deterministic outcome handled below; any other non-2xx + # (e.g. NCEI's intermittent 500s) raises here so retry() retries it. + if response.status_code != 404: + response.raise_for_status() + return response + + response = retry(get_listing) if response.status_code == 404: now = pd.Timestamp.now() if now.year == year and now.month == 1: return [] - response.raise_for_status() + response.raise_for_status() content = response.text filenames = re.findall(r"href=\"(VIIRS-Land.+nc)\"", content) diff --git a/tests/contrib/noaa/ndvi_cdr/analysis/region_job_test.py b/tests/contrib/noaa/ndvi_cdr/analysis/region_job_test.py index fd5783bd5..fa8b3afe6 100644 --- a/tests/contrib/noaa/ndvi_cdr/analysis/region_job_test.py +++ b/tests/contrib/noaa/ndvi_cdr/analysis/region_job_test.py @@ -433,6 +433,49 @@ def test_list_source_files_routing_by_year( mock_ncei.assert_not_called() +def test_list_ncei_source_files_retries_on_server_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Transient 5xx responses from NCEI are retried rather than failing the job.""" + monkeypatch.setattr("time.sleep", lambda *_: None) + + call_count = 0 + + def mock_requests_get(url: str, **kwargs: Any) -> Mock: # noqa: ANN401 + nonlocal call_count + call_count += 1 + mock_response = Mock() + if call_count < 3: + mock_response.status_code = 500 + + def raise_http_error() -> None: + raise requests.HTTPError(response=mock_response) + + mock_response.raise_for_status = raise_http_error + else: + mock_response.status_code = 200 + mock_response.raise_for_status = Mock() + mock_response.text = 'x' + return mock_response + + monkeypatch.setattr("requests.get", mock_requests_get) + + template_config = NoaaNdviCdrAnalysisTemplateConfig() + region_job = NoaaNdviCdrAnalysisRegionJob.model_construct( + tmp_store=Mock(), + template_ds=Mock(), + data_vars=template_config.data_vars, + append_dim=template_config.append_dim, + region=Mock(spec=slice), + reformat_job_name="test", + ) + + result = region_job._list_ncei_source_files(2025) + + assert call_count == 3 + assert result == ["VIIRS-Land_v001_JP113C1_NOAA-20_20250101_c20250103153010.nc"] + + @pytest.fixture def mock_404_response(monkeypatch: pytest.MonkeyPatch) -> Mock: """Fixture to mock requests.get returning a 404 response.""" From c42966b8e9c341bbd5486a5054edf4f4aa007685 Mon Sep 17 00:00:00 2001 From: Marshall Date: Sun, 19 Jul 2026 11:07:31 -0500 Subject: [PATCH 07/23] Add Dependabot version updates (#753) --- .github/dependabot.yml | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) create mode 100644 .github/dependabot.yml diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 000000000..61f607aaf --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,16 @@ +version: 2 +updates: + - package-ecosystem: "uv" + directory: "/" + schedule: + interval: "weekly" + groups: + all: + patterns: ["*"] + - package-ecosystem: "github-actions" + directory: "/" + schedule: + interval: "weekly" + groups: + all: + patterns: ["*"] From f5430ac379ca9f1e971172d739fd1cb8bfd1f3ea Mon Sep 17 00:00:00 2001 From: Alden Keefe Sampson Date: Sun, 19 Jul 2026 13:54:02 -0400 Subject: [PATCH 08/23] Fix GRIB decimal precision rounding in MRMS data (#758) * Quantize MRMS GRIB reads to the message's decimal precision GDAL's g2clib unpacks GRIB values in float32 arithmetic whose rounding error varies by architecture: on arm64, FMA contraction of refD + bdscale*X decodes packed zeros to ~4.5e-8 (mm) / -2.2e-5 (percent) instead of exact 0, which the archive then stores as 1.2391865e-11 kg m-2 s-1 and -2.2292137e-05 percent after deaccumulation and mantissa rounding. x86_64 rounds the product to float32 first and gets exact 0 from the same file. GRIB2 field values are exact multiples of 10^-D (decimal scale factor, section 5), so rounding the decoded float64 field to D decimals restores the encoded values on any architecture and makes materialized reads bit-identical to gribberish's float64 decode used by virtual datasets. Adds grib_decimal_scale_factors to parse D per field (asserting binary scale factor E is 0) and applies it in the MRMS read path. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01YVoAgQxtwjvu86nMfowU9a * Assert integer GRIB reference value; clarify quantization comment Values are exact multiples of 10^-D only when the binary scale factor E is 0 and the reference value R is an integer; previously only E was checked, so a fractional R would have silently passed and decimal rounding would corrupt values by up to half a quantum. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01YVoAgQxtwjvu86nMfowU9a * Split decimal scale lookup onto two lines for readability Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01YVoAgQxtwjvu86nMfowU9a * Document the GRIB2 byte layout behind the section 5 offsets Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01YVoAgQxtwjvu86nMfowU9a --------- Co-authored-by: Claude --- src/reformatters/common/grib.py | 65 ++++++++++++++++ .../mrms/conus_analysis_hourly/region_job.py | 13 +++- tests/common/grib_test.py | 78 +++++++++++++++++++ .../conus_analysis_hourly/region_job_test.py | 63 ++++++++++++++- 4 files changed, 215 insertions(+), 4 deletions(-) create mode 100644 src/reformatters/common/grib.py create mode 100644 tests/common/grib_test.py diff --git a/src/reformatters/common/grib.py b/src/reformatters/common/grib.py new file mode 100644 index 000000000..eb791a058 --- /dev/null +++ b/src/reformatters/common/grib.py @@ -0,0 +1,65 @@ +import struct +from pathlib import Path + + +def grib_decimal_scale_factors(path: Path) -> list[int]: + """Decimal scale factor D from each GRIB2 data representation section (section 5) + in the file, in field order (GDAL/rasterio band order). A GRIB2 field's values are + (R + X * 2^E) / 10^D; each field is asserted to have binary scale factor E == 0 and + an integer reference value R, which together guarantee its values are exact + multiples of 10^-D. + """ + # A GRIB2 file is a sequence of messages. Byte layout behind the offsets below + # (0-based from the message or section start; multi-byte integers big-endian): + # + # Section 0 (indicator, fixed 16 bytes): + # [0:4] b"GRIB" + # [8:16] total message length + # Sections 1-7 (sections 4-7 repeat per field within a message): + # [0:4] section length + # [4] section number + # Section 5 (data representation), following that 5-byte section header: + # [5:9] number of data points + # [9:11] data representation template number + # [11:15] reference value R (IEEE float32) + # [15:17] binary scale factor E (sign-and-magnitude int16) + # [17:19] decimal scale factor D (sign-and-magnitude int16) + # End section: the literal bytes b"7777" terminate each message. + # + # R/E/D sit at the same octets in all common templates (5.0 simple, 5.2/5.3 + # complex, 5.40 JPEG2000, 5.41 PNG), so no branching on template is needed. + data = path.read_bytes() + scale_factors: list[int] = [] + pos = 0 + while pos < len(data): + assert data[pos : pos + 4] == b"GRIB", ( + f"Expected GRIB message start at byte {pos} in {path}" + ) + message_end = pos + int.from_bytes(data[pos + 8 : pos + 16]) + pos += 16 + while pos < message_end and data[pos : pos + 4] != b"7777": + section_length = int.from_bytes(data[pos : pos + 4]) + if data[pos + 4] == 5: + (reference_value,) = struct.unpack(">f", data[pos + 11 : pos + 15]) + assert reference_value == round(reference_value), ( + f"Non-integer reference value {reference_value} in {path}; " + "values are not multiples of 10^-D" + ) + binary_scale = _sign_and_magnitude_int(data[pos + 15 : pos + 17]) + assert binary_scale == 0, ( + f"Binary scale factor {binary_scale} != 0 in {path}; " + "values are not multiples of 10^-D" + ) + scale_factors.append(_sign_and_magnitude_int(data[pos + 17 : pos + 19])) + pos += section_length + pos = message_end + + assert scale_factors, f"No data representation sections found in {path}" + return scale_factors + + +def _sign_and_magnitude_int(raw: bytes) -> int: + """GRIB2 signed integers use a sign bit plus magnitude, not two's complement.""" + value = int.from_bytes(raw) + sign_bit = 1 << (len(raw) * 8 - 1) + return -(value & ~sign_bit) if value & sign_bit else value diff --git a/src/reformatters/noaa/mrms/conus_analysis_hourly/region_job.py b/src/reformatters/noaa/mrms/conus_analysis_hourly/region_job.py index d15485010..6587e9b4d 100644 --- a/src/reformatters/noaa/mrms/conus_analysis_hourly/region_job.py +++ b/src/reformatters/noaa/mrms/conus_analysis_hourly/region_job.py @@ -14,6 +14,7 @@ from reformatters.common.binary_rounding import round_float32_inplace from reformatters.common.deaccumulation import deaccumulate_to_rates_inplace from reformatters.common.download import http_download_to_disk +from reformatters.common.grib import grib_decimal_scale_factors from reformatters.common.logging import get_logger from reformatters.common.materialized_region_job import MaterializedRegionJob from reformatters.common.pydantic import replace @@ -204,7 +205,17 @@ def read_data( f"Expected exactly 1 band, found {reader.count} in {coord.downloaded_path}" ) rasterio_band = 1 - result: ArrayFloat32 = reader.read(rasterio_band, out_dtype=np.float32) + raw = reader.read(rasterio_band) + + # GDAL unpacks GRIB values internally in float32 arithmetic (regardless of the + # dtype read into) whose rounding error varies by architecture (amd64 vs arm64). + # These fields' values are exact multiples of 10^-D (enforced by + # grib_decimal_scale_factors), so rounding to D decimals restores them exactly. + # Also matches gribberish's float64 decode used by virtual datasets. + decimal_scales = grib_decimal_scale_factors(coord.downloaded_path) + decimal_scale = decimal_scales[rasterio_band - 1] + np.round(raw, decimal_scale, out=raw) + result: ArrayFloat32 = raw.astype(np.float32) nodata_sentinel = data_var.internal_attrs.nodata_sentinel if nodata_sentinel is not None: diff --git a/tests/common/grib_test.py b/tests/common/grib_test.py new file mode 100644 index 000000000..d52aed548 --- /dev/null +++ b/tests/common/grib_test.py @@ -0,0 +1,78 @@ +import struct +from pathlib import Path + +import pytest + +from reformatters.common.grib import grib_decimal_scale_factors + + +def _sign_and_magnitude_bytes(value: int) -> bytes: + return ((0x8000 | -value) if value < 0 else value).to_bytes(2) + + +def _grib2_message( + fields: list[tuple[int, int]], reference_value: float = -30.0 +) -> bytes: + """Minimal GRIB2 message with a (binary_scale, decimal_scale) section 5 per field.""" + body = (21).to_bytes(4) + bytes([1]) + bytes(16) # section 1 + for binary_scale, decimal_scale in fields: + body += (9).to_bytes(4) + bytes([4]) + bytes(4) # section 4 + section5_payload = ( + (4).to_bytes(4) # number of data points + + (41).to_bytes(2) # data representation template (PNG packing) + + struct.pack(">f", reference_value) + + _sign_and_magnitude_bytes(binary_scale) + + _sign_and_magnitude_bytes(decimal_scale) + + bytes([16, 0]) # bits per value, original field type + ) + body += (5 + len(section5_payload)).to_bytes(4) + bytes([5]) + section5_payload + body += (5).to_bytes(4) + bytes([7]) # section 7 + header = b"GRIB" + bytes(2) + bytes([0, 2]) + (16 + len(body) + 4).to_bytes(8) + return header + body + b"7777" + + +def test_grib_decimal_scale_factors_single_field(tmp_path: Path) -> None: + path = tmp_path / "single.grib2" + path.write_bytes(_grib2_message([(0, 1)])) + assert grib_decimal_scale_factors(path) == [1] + + +def test_grib_decimal_scale_factors_multiple_messages_and_fields( + tmp_path: Path, +) -> None: + path = tmp_path / "multi.grib2" + path.write_bytes(_grib2_message([(0, 2), (0, 0)]) + _grib2_message([(0, -1)])) + assert grib_decimal_scale_factors(path) == [2, 0, -1] + + +def test_grib_decimal_scale_factors_rejects_nonzero_binary_scale( + tmp_path: Path, +) -> None: + path = tmp_path / "binary_scaled.grib2" + path.write_bytes(_grib2_message([(-2, 1)])) + with pytest.raises(AssertionError, match="Binary scale factor"): + grib_decimal_scale_factors(path) + + +def test_grib_decimal_scale_factors_rejects_non_integer_reference_value( + tmp_path: Path, +) -> None: + path = tmp_path / "fractional_r.grib2" + path.write_bytes(_grib2_message([(0, 1)], reference_value=-30.5)) + with pytest.raises(AssertionError, match="Non-integer reference value"): + grib_decimal_scale_factors(path) + + +def test_grib_decimal_scale_factors_rejects_non_grib(tmp_path: Path) -> None: + path = tmp_path / "not_a.grib2" + path.write_bytes(b"NOPE" + bytes(20)) + with pytest.raises(AssertionError, match="Expected GRIB message start"): + grib_decimal_scale_factors(path) + + +def test_grib_decimal_scale_factors_real_negative_decimal_scale( + tmp_path: Path, +) -> None: + path = tmp_path / "neg_d.grib2" + path.write_bytes(_grib2_message([(0, -3)])) + assert grib_decimal_scale_factors(path) == [-3] diff --git a/tests/noaa/mrms/conus_analysis_hourly/region_job_test.py b/tests/noaa/mrms/conus_analysis_hourly/region_job_test.py index 2c9a67f35..5261ab05b 100644 --- a/tests/noaa/mrms/conus_analysis_hourly/region_job_test.py +++ b/tests/noaa/mrms/conus_analysis_hourly/region_job_test.py @@ -279,7 +279,7 @@ def test_read_data_pre_v12_two_band_selects_discipline_209( reformat_job_name="test", ) - data = np.ones((2, 2), dtype=np.float32) + data = np.ones((2, 2), dtype=np.float64) class FakeReader: count = 2 @@ -296,12 +296,15 @@ def tags(self, band: int) -> dict[str, str]: 2: {"GRIB_DISCIPLINE": "209(Local)"}, }[band] - def read(self, band: int, out_dtype: np.dtype[np.float32]) -> np.ndarray: - assert out_dtype == np.float32 + def read(self, band: int) -> np.ndarray: assert band == 2 return data monkeypatch.setattr("rasterio.open", lambda *_args, **_kwargs: FakeReader()) + monkeypatch.setattr( + "reformatters.noaa.mrms.conus_analysis_hourly.region_job.grib_decimal_scale_factors", + lambda _path: [1, 1], + ) result = region_job.read_data( NoaaMrmsSourceFileCoord( @@ -318,6 +321,60 @@ def read(self, band: int, out_dtype: np.dtype[np.float32]) -> np.ndarray: np.testing.assert_array_equal(result, data) +def test_read_data_quantizes_to_grib_decimal_precision( + monkeypatch: pytest.MonkeyPatch, + template_config: NoaaMrmsConusAnalysisHourlyTemplateConfig, +) -> None: + region_job = NoaaMrmsRegionJob.model_construct( + template_ds=Mock(), + data_vars=[], + append_dim="time", + region=slice(0, 1), + reformat_job_name="test", + ) + + # Values as GDAL's float32 unpacking produces them on arm64 (FMA contraction): + # a packed zero decodes to 4.4703484e-8 and 0.2 mm to 0.2 + ~5e-8. + data = np.array( + [[4.4703484e-08, 0.20000004768371582], [-3.0, 12.3]], dtype=np.float64 + ) + + class FakeReader: + count = 1 + + def __enter__(self) -> Self: + return self + + def __exit__(self, *args: object) -> None: + return None + + def read(self, band: int) -> np.ndarray: + assert band == 1 + return data + + monkeypatch.setattr("rasterio.open", lambda *_args, **_kwargs: FakeReader()) + monkeypatch.setattr( + "reformatters.noaa.mrms.conus_analysis_hourly.region_job.grib_decimal_scale_factors", + lambda _path: [1], + ) + + result = region_job.read_data( + NoaaMrmsSourceFileCoord( + time=pd.Timestamp("2024-01-15T12:00"), + product="MultiSensor_QPE_01H_Pass2", + level="00.00", + fallback_products=(), + data_var_name="precipitation_surface", + downloaded_path=Path("fake.grib2"), + ), + next(v for v in template_config.data_vars if v.name == "precipitation_surface"), + ) + + assert result.dtype == np.float32 + expected = np.array([[0.0, 0.2], [-3.0, 12.3]], dtype=np.float32) + np.testing.assert_array_equal(result, expected) + + @pytest.mark.parametrize( ("region", "expected_processing_region"), [ From 0559432034dc75334f5e20fd8225e65a8d661d24 Mon Sep 17 00:00:00 2001 From: Alden Keefe Sampson Date: Sun, 19 Jul 2026 14:28:38 -0400 Subject: [PATCH 09/23] HRRR materialized: CF missing_value for percent_frozen_precipitation (#759) HRRR CPOFP encodes "no/undefined frozen precipitation" as a discrete -50 (verified: ~97-98% of the CONUS field where no precipitation is falling, with valid frozen-fraction values 0-100 elsewhere). The materialized HRRR datasets passed this through unmasked and undocumented, so the field mean reads ~-47 and CF-aware readers see the sentinel as data. Mirror the CF-compliant handling already used by the HRRR forecast virtual dataset: set missing_value=-50.0 with a matching -50.0 fill_value and a comment, so CF-aware readers mask it to NaN. Applies to both materialized datasets that share the common data var (forecast-48-hour and analysis); templates regenerated. Claude-Session: https://claude.ai/code/session_01Gd2qpwonTbByYkCzaNAyt9 Co-authored-by: Claude --- .../percent_frozen_precipitation_surface/zarr.json | 6 ++++-- .../noaa/hrrr/analysis/templates/latest.zarr/zarr.json | 6 ++++-- .../percent_frozen_precipitation_surface/zarr.json | 6 ++++-- .../hrrr/forecast_48_hour/templates/latest.zarr/zarr.json | 6 ++++-- src/reformatters/noaa/hrrr/template_config.py | 5 ++++- 5 files changed, 20 insertions(+), 9 deletions(-) diff --git a/src/reformatters/noaa/hrrr/analysis/templates/latest.zarr/percent_frozen_precipitation_surface/zarr.json b/src/reformatters/noaa/hrrr/analysis/templates/latest.zarr/percent_frozen_precipitation_surface/zarr.json index ee197be2e..b284a002d 100644 --- a/src/reformatters/noaa/hrrr/analysis/templates/latest.zarr/percent_frozen_precipitation_surface/zarr.json +++ b/src/reformatters/noaa/hrrr/analysis/templates/latest.zarr/percent_frozen_precipitation_surface/zarr.json @@ -21,7 +21,7 @@ "separator": "/" } }, - "fill_value": "NaN", + "fill_value": -50.0, "codecs": [ { "name": "sharding_indexed", @@ -68,9 +68,11 @@ "long_name": "Percent frozen precipitation", "short_name": "cpofp", "units": "percent", + "comment": "-50 encodes no/undefined frozen precipitation; CF-aware readers mask it to NaN.", + "missing_value": -50.0, "step_type": "instant", "coordinates": "latitude longitude spatial_ref", - "_FillValue": "AAAAAAAA+H8=" + "_FillValue": "AAAAAAAAScA=" }, "dimension_names": [ "time", diff --git a/src/reformatters/noaa/hrrr/analysis/templates/latest.zarr/zarr.json b/src/reformatters/noaa/hrrr/analysis/templates/latest.zarr/zarr.json index 27bd3558e..e64afae3b 100644 --- a/src/reformatters/noaa/hrrr/analysis/templates/latest.zarr/zarr.json +++ b/src/reformatters/noaa/hrrr/analysis/templates/latest.zarr/zarr.json @@ -931,7 +931,7 @@ "separator": "/" } }, - "fill_value": "NaN", + "fill_value": -50.0, "codecs": [ { "name": "sharding_indexed", @@ -978,9 +978,11 @@ "long_name": "Percent frozen precipitation", "short_name": "cpofp", "units": "percent", + "comment": "-50 encodes no/undefined frozen precipitation; CF-aware readers mask it to NaN.", + "missing_value": -50.0, "step_type": "instant", "coordinates": "latitude longitude spatial_ref", - "_FillValue": "AAAAAAAA+H8=" + "_FillValue": "AAAAAAAAScA=" }, "dimension_names": [ "time", diff --git a/src/reformatters/noaa/hrrr/forecast_48_hour/templates/latest.zarr/percent_frozen_precipitation_surface/zarr.json b/src/reformatters/noaa/hrrr/forecast_48_hour/templates/latest.zarr/percent_frozen_precipitation_surface/zarr.json index 973430ffa..4b2e7f90b 100644 --- a/src/reformatters/noaa/hrrr/forecast_48_hour/templates/latest.zarr/percent_frozen_precipitation_surface/zarr.json +++ b/src/reformatters/noaa/hrrr/forecast_48_hour/templates/latest.zarr/percent_frozen_precipitation_surface/zarr.json @@ -23,7 +23,7 @@ "separator": "/" } }, - "fill_value": 0.0, + "fill_value": -50.0, "codecs": [ { "name": "sharding_indexed", @@ -71,9 +71,11 @@ "long_name": "Percent frozen precipitation", "short_name": "cpofp", "units": "percent", + "comment": "-50 encodes no/undefined frozen precipitation; CF-aware readers mask it to NaN.", + "missing_value": -50.0, "step_type": "instant", "coordinates": "expected_forecast_length ingested_forecast_length latitude longitude spatial_ref valid_time", - "_FillValue": "AAAAAAAA+H8=" + "_FillValue": "AAAAAAAAScA=" }, "dimension_names": [ "init_time", diff --git a/src/reformatters/noaa/hrrr/forecast_48_hour/templates/latest.zarr/zarr.json b/src/reformatters/noaa/hrrr/forecast_48_hour/templates/latest.zarr/zarr.json index e81b87313..2aa07c99a 100644 --- a/src/reformatters/noaa/hrrr/forecast_48_hour/templates/latest.zarr/zarr.json +++ b/src/reformatters/noaa/hrrr/forecast_48_hour/templates/latest.zarr/zarr.json @@ -1192,7 +1192,7 @@ "separator": "/" } }, - "fill_value": 0.0, + "fill_value": -50.0, "codecs": [ { "name": "sharding_indexed", @@ -1240,9 +1240,11 @@ "long_name": "Percent frozen precipitation", "short_name": "cpofp", "units": "percent", + "comment": "-50 encodes no/undefined frozen precipitation; CF-aware readers mask it to NaN.", + "missing_value": -50.0, "step_type": "instant", "coordinates": "expected_forecast_length ingested_forecast_length latitude longitude spatial_ref valid_time", - "_FillValue": "AAAAAAAA+H8=" + "_FillValue": "AAAAAAAAScA=" }, "dimension_names": [ "init_time", diff --git a/src/reformatters/noaa/hrrr/template_config.py b/src/reformatters/noaa/hrrr/template_config.py index 2b60c2a60..f7758fe70 100644 --- a/src/reformatters/noaa/hrrr/template_config.py +++ b/src/reformatters/noaa/hrrr/template_config.py @@ -12,6 +12,7 @@ Encoding, StatisticsApproximate, ) +from reformatters.common.pydantic import replace from reformatters.common.template_config import TemplateConfig from reformatters.common.types import ( Array1D, @@ -349,12 +350,14 @@ def get_data_vars(self, encoding: Encoding) -> Sequence[NoaaHrrrDataVar]: ), NoaaHrrrDataVar( name="percent_frozen_precipitation_surface", - encoding=encoding, + encoding=replace(encoding, fill_value=-50.0), attrs=DataVarAttrs( short_name="cpofp", long_name="Percent frozen precipitation", units="percent", step_type="instant", + comment="-50 encodes no/undefined frozen precipitation; CF-aware readers mask it to NaN.", + missing_value=-50.0, ), internal_attrs=NoaaHrrrInternalAttrs( grib_element="CPOFP", From 404238eab7b25164d50aa116e213d8b1a0cd2c26 Mon Sep 17 00:00:00 2001 From: Alden Keefe Sampson Date: Sun, 19 Jul 2026 17:58:15 -0400 Subject: [PATCH 10/23] HRRR ops validation: exclude percent_frozen_precipitation from NaN check (#760) --- src/reformatters/noaa/hrrr/analysis/dynamical_dataset.py | 3 +++ .../noaa/hrrr/forecast_48_hour/dynamical_dataset.py | 3 +++ 2 files changed, 6 insertions(+) diff --git a/src/reformatters/noaa/hrrr/analysis/dynamical_dataset.py b/src/reformatters/noaa/hrrr/analysis/dynamical_dataset.py index 2efdcdb19..686f64181 100644 --- a/src/reformatters/noaa/hrrr/analysis/dynamical_dataset.py +++ b/src/reformatters/noaa/hrrr/analysis/dynamical_dataset.py @@ -68,5 +68,8 @@ def validators(self) -> Sequence[validation.DataValidator]: partial( validation.check_analysis_recent_nans, max_expected_delay=max_expected_delay, + # CF-masks its -50 "no precipitation" sentinel to NaN, so the field + # is legitimately all/mostly NaN wherever no precipitation is falling. + exclude_vars=("percent_frozen_precipitation_surface",), ), ) diff --git a/src/reformatters/noaa/hrrr/forecast_48_hour/dynamical_dataset.py b/src/reformatters/noaa/hrrr/forecast_48_hour/dynamical_dataset.py index e1c9474f4..098d33278 100644 --- a/src/reformatters/noaa/hrrr/forecast_48_hour/dynamical_dataset.py +++ b/src/reformatters/noaa/hrrr/forecast_48_hour/dynamical_dataset.py @@ -73,5 +73,8 @@ def validators(self) -> Sequence[validation.DataValidator]: partial( validation.check_forecast_recent_nans, additional_skip_lead_time_0_vars=HRRR_EXPECTED_HOUR_0_NAN_VARS, + # CF-masks its -50 "no precipitation" sentinel to NaN, so the field + # is legitimately all/mostly NaN wherever no precipitation is falling. + exclude_vars=("percent_frozen_precipitation_surface",), ), ) From 7216eddfaa208bac7e145712a8e301b57a731137 Mon Sep 17 00:00:00 2001 From: Alden Keefe Sampson Date: Mon, 20 Jul 2026 15:10:46 -0400 Subject: [PATCH 11/23] Add sliding window validation for recent forecast init_times (#766) * Validate recent init_times for NaNs, not just the newest check_forecast_recent_nans only ever inspected the single newest init_time (isel(init_time=[-1])), so an init_time was validated once, while it was the newest, and never again. A transient source read failure that leaves a whole forecast step NaN (e.g. all perturbed members at one lead_time, whose open-data enfo-ef file failed to download) is therefore invisible if it isn't caught in that one window. Add num_recent_init_times to check a window of recent init_times, each independently (a per-init threshold is not diluted), reporting which init_time failed. Default stays 1 (unchanged for existing datasets). Wire the ECMWF forecast datasets, which expect zero NaNs, to check the last 3. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Wvu6VqQjxWyRHZRMA4W28o * Drop explanatory comments on ECMWF recent-nans validators The rationale lives in check_forecast_recent_nans's docstring; the per-dataset repetition added no local information. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Wvu6VqQjxWyRHZRMA4W28o --------- Co-authored-by: Claude --- src/reformatters/common/validation.py | 63 +++++++++++++++---- .../aifs_ens/forecast/dynamical_dataset.py | 3 +- .../aifs_single/forecast/dynamical_dataset.py | 3 +- .../dynamical_dataset.py | 3 +- tests/common/validation_test.py | 31 +++++++++ 5 files changed, 87 insertions(+), 16 deletions(-) diff --git a/src/reformatters/common/validation.py b/src/reformatters/common/validation.py index 821c682fd..f01a7611f 100644 --- a/src/reformatters/common/validation.py +++ b/src/reformatters/common/validation.py @@ -217,6 +217,7 @@ def check_forecast_recent_nans( ds: xr.Dataset, *, init_time_offset: int = -1, + num_recent_init_times: int = 1, max_nan_fraction: float = 0.0, include_vars: Sequence[str] | Literal["all"] = "all", exclude_vars: Sequence[str] = (), @@ -225,10 +226,18 @@ def check_forecast_recent_nans( max_workers: int | None = None, ) -> ValidationResult: """ - Check the NaN fraction of a recent init_time in a forecast dataset. - - `init_time_offset` selects which init_time to check from the end - (`-1` = latest, `-2` = previous, etc.). Use `-2` for datasets whose + Check the NaN fraction of recent init_times in a forecast dataset. + + Checks `num_recent_init_times` init_times ending at `init_time_offset` + (inclusive), newest first, and fails if any of them exceeds + `max_nan_fraction`. Each init_time is checked independently, so a per-init + threshold is not diluted by a wider window. A window > 1 catches a gap that + lands in a recent-but-no-longer-newest init_time (a late source file, a + catch-up or re-backfill run): with a window of 1 each init_time is validated + only while it is the newest, then never looked at again. + + `init_time_offset` selects the newest init_time to check, counted from the + end (`-1` = latest, `-2` = previous, etc.). Use `-2` for datasets whose latest init is still being filled in (e.g. long-horizon ensembles). Default `spatial_sampling="random_points"` reads all lead_times (and any @@ -240,16 +249,44 @@ def check_forecast_recent_nans( hour 0 data). `additional_skip_lead_time_0_vars` adds extra names on top (e.g. HRRR categorical vars which are step_type=instant but have no hour 0 data). """ - sample_ds = ds.isel(init_time=[init_time_offset]) - sample_ds = _apply_spatial_sampling(sample_ds, spatial_sampling) + assert num_recent_init_times >= 1, "num_recent_init_times must be >= 1" + + newest = init_time_offset % ds.sizes["init_time"] # positive index + oldest = max(0, newest - num_recent_init_times + 1) + + results = [ + ( + index, + _check_nan_fractions( + _apply_spatial_sampling(ds.isel(init_time=[index]), spatial_sampling), + max_nan_fraction=max_nan_fraction, + include_vars=include_vars, + exclude_vars=exclude_vars, + additional_skip_lead_time_0_vars=additional_skip_lead_time_0_vars, + max_workers=max_workers or _DEFAULT_MAX_WORKERS[spatial_sampling], + ), + ) + for index in range(newest, oldest - 1, -1) + ] - return _check_nan_fractions( - sample_ds, - max_nan_fraction=max_nan_fraction, - include_vars=include_vars, - exclude_vars=exclude_vars, - additional_skip_lead_time_0_vars=additional_skip_lead_time_0_vars, - max_workers=max_workers or _DEFAULT_MAX_WORKERS[spatial_sampling], + if len(results) == 1: + return results[0][1] + + failures = [ + f"init_time={_format_coord_value(ds['init_time'].values[index])}: {result.message}" + for index, result in results + if not result.passed + ] + if failures: + return ValidationResult( + passed=False, + message=f"Excessive NaN fraction in {len(failures)} of {len(results)} " + "recent init_times:\n" + "\n".join(failures), + ) + return ValidationResult( + passed=True, + message=f"All {len(results)} recent init_times have NaN fraction " + f"<= {max_nan_fraction}", ) diff --git a/src/reformatters/ecmwf/aifs_ens/forecast/dynamical_dataset.py b/src/reformatters/ecmwf/aifs_ens/forecast/dynamical_dataset.py index 916cfae52..7fa4fc72c 100644 --- a/src/reformatters/ecmwf/aifs_ens/forecast/dynamical_dataset.py +++ b/src/reformatters/ecmwf/aifs_ens/forecast/dynamical_dataset.py @@ -1,5 +1,6 @@ from collections.abc import Sequence from datetime import timedelta +from functools import partial from reformatters.common import validation from reformatters.common.dynamical_dataset import DynamicalDataset @@ -60,5 +61,5 @@ def operational_kubernetes_resources(self, image_tag: str) -> Sequence[CronJob]: def validators(self) -> Sequence[validation.DataValidator]: return ( validation.check_forecast_current_data, - validation.check_forecast_recent_nans, + partial(validation.check_forecast_recent_nans, num_recent_init_times=3), ) diff --git a/src/reformatters/ecmwf/aifs_single/forecast/dynamical_dataset.py b/src/reformatters/ecmwf/aifs_single/forecast/dynamical_dataset.py index d15a54a40..d9c404269 100644 --- a/src/reformatters/ecmwf/aifs_single/forecast/dynamical_dataset.py +++ b/src/reformatters/ecmwf/aifs_single/forecast/dynamical_dataset.py @@ -1,5 +1,6 @@ from collections.abc import Sequence from datetime import timedelta +from functools import partial from reformatters.common import validation from reformatters.common.dynamical_dataset import DynamicalDataset @@ -57,5 +58,5 @@ def operational_kubernetes_resources(self, image_tag: str) -> Sequence[CronJob]: def validators(self) -> Sequence[validation.DataValidator]: return ( validation.check_forecast_current_data, - validation.check_forecast_recent_nans, + partial(validation.check_forecast_recent_nans, num_recent_init_times=3), ) diff --git a/src/reformatters/ecmwf/ifs_ens/forecast_15_day_0_25_degree/dynamical_dataset.py b/src/reformatters/ecmwf/ifs_ens/forecast_15_day_0_25_degree/dynamical_dataset.py index 4b12ad96f..84c5fb42b 100644 --- a/src/reformatters/ecmwf/ifs_ens/forecast_15_day_0_25_degree/dynamical_dataset.py +++ b/src/reformatters/ecmwf/ifs_ens/forecast_15_day_0_25_degree/dynamical_dataset.py @@ -1,5 +1,6 @@ from collections.abc import Sequence from datetime import timedelta +from functools import partial from reformatters.common import validation from reformatters.common.dynamical_dataset import DynamicalDataset @@ -63,5 +64,5 @@ def validators(self) -> Sequence[validation.DataValidator]: """Return a sequence of DataValidators to run on this dataset.""" return ( validation.check_forecast_current_data, - validation.check_forecast_recent_nans, + partial(validation.check_forecast_recent_nans, num_recent_init_times=3), ) diff --git a/tests/common/validation_test.py b/tests/common/validation_test.py index fd959bcfc..2bd5e5662 100644 --- a/tests/common/validation_test.py +++ b/tests/common/validation_test.py @@ -216,6 +216,37 @@ def test_check_forecast_recent_nans_init_time_offset( ).passed +def test_check_forecast_recent_nans_window_catches_older_init( + forecast_dataset: xr.Dataset, +) -> None: + """A window > 1 catches NaNs in a recent-but-not-newest init_time.""" + # NaN a whole init_time that is not the newest (index 2 of 5, i.e. offset -3) + bad_init = forecast_dataset.init_time[2] + forecast_dataset["temperature"].loc[{"init_time": bad_init}] = np.nan + + # The default window of 1 only checks the newest init_time, so it misses it. + assert validation.check_forecast_recent_nans(forecast_dataset).passed + + # A window reaching back to it catches the gap and names the offending init_time. + result = validation.check_forecast_recent_nans( + forecast_dataset, num_recent_init_times=3 + ) + assert not result.passed + assert "Excessive NaN fraction" in result.message + assert pd.Timestamp(bad_init.values).isoformat() in result.message + + +def test_check_forecast_recent_nans_window_all_clean_passes( + forecast_dataset: xr.Dataset, +) -> None: + """A clean window passes and reports how many init_times were checked.""" + result = validation.check_forecast_recent_nans( + forecast_dataset, num_recent_init_times=3 + ) + assert result.passed + assert "All 3 recent init_times" in result.message + + def test_check_analysis_current_data_passes( monkeypatch: pytest.MonkeyPatch, analysis_dataset: xr.Dataset ) -> None: From f118cea061c1069eb121ae598448acd0b8207aa9 Mon Sep 17 00:00:00 2001 From: Alden Keefe Sampson Date: Mon, 20 Jul 2026 15:56:51 -0400 Subject: [PATCH 12/23] Safer, simpler backfills: explicit overwrite API, auto append-dim end, manual GitHub action (#755) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Safer, simpler backfills: explicit overwrite API, auto append-dim end, manual GitHub action backfill-kubernetes now defaults --append-dim-end to "now" for a new store and to the store's current end for an existing one, and replaces --overwrite-existing with explicit --overwrite-chunks / --overwrite-metadata flags. Guards (assert_safe_overwrite, checked in the driver and again on worker 0) reject structural drift, never allow trimming an existing store, and only allow new arrays or extension when explicitly requested. --overwrite-metadata alone refreshes store metadata from the template in place via the shared refresh_store_metadata helper (extracted from VirtualRegionJob.refresh_metadata), creating newly added variables without launching workers. Overwrite backfills preserve job-written coordinate values (template-null coords like ingested_forecast_length are excluded from metadata copies), publish zarr3 metadata at finalize so a new variable appears only once its data is written, and suspend the dataset's update/validate cronjobs so an operational update can't move icechunk main and strand the backfill's finalize branch reset — which now raises loudly (with correct crash-retry detection) instead of silently skipping publication. New workflow_dispatch GitHub actions, generated alongside the existing manual workflows: "Manual: Backfill" exposing only the safe operations (create new store, overwrite chunks/metadata) using the deployed image, and "Manual: Suspend/Resume Dataset CronJobs" to resume updates after a backfill. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01SAB2gwVGczEvJGx4d9DnMm * Run overwrite backfills concurrently with operational updates via icechunk rebase Replaces the cronjob-suspension approach: backfills can run for days and updates have a latency SLA, so neither side may block or lose to the other. Overwrite backfills (--overwrite-chunks / --overwrite-metadata) no longer use a temp branch. Worker 0 writes refreshed template metadata straight into the live stores (skip_unchanged, store-written coordinate values preserved), then each worker's chunk-only commit to icechunk main rebases over concurrent update commits, with BasicConflictSolver(UseTheirs) so the operational update wins any chunk both wrote. A new variable appears immediately and fills in progressively, matching the previous update-creates-then-backfill flow. The reverse race is handled in finalize: an update that finds main moved past its setup snapshot (a committed temp branch cannot be rebased) now replays itself onto the moved main — copying the branch's exact data-variable chunk bytes from repo.diff plus the final template metadata onto a fresh main session and committing with UseOurs so the update wins conflicts. The replay is bounded by one update's writes, atomic to readers, and idempotent across crash retries. Fresh-store backfill divergence (nothing else should write a store being created) still fails loudly. commit_if_icechunk gains a rebase solver parameter and skips sessions with no changes; refresh_store_metadata and the overwrite setup share write_template_metadata_to_stores. The suspend/resume workflow and cronjob suspension code are removed. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01SAB2gwVGczEvJGx4d9DnMm * Mark replay commits via snapshot metadata, not commit message text The replay commit now carries the standard "Update at