Add NOAA MRMS CONUS hourly precipitation analysis dataset

[aldenks]

Summary

This PR adds support for the NOAA Multi-Radar Multi-Sensor (MRMS) CONUS hourly precipitation analysis dataset. The implementation includes data ingestion, reformatting, and operational update capabilities for multiple precipitation-related variables from MRMS.

Towards #461, closes #473

Key Changes

• Template Configuration (template_config.py): Defines the dataset structure with 4 data variables (precipitation_surface, precipitation_pass_1_surface, precipitation_radar_only_surface, and categorical_precipitation_type_surface) covering the Continental US at 0.01° resolution with hourly frequency from October 2014 onwards.

• Region Job Implementation (region_job.py): Implements data processing logic including:

  • Source file coordinate generation with support for multiple MRMS product versions
  • Handling of MRMS v12.0 transition (October 2020) with fallback to pre-v12 products (GaugeCorr_QPE_01H) for historical data
  • Download support from three sources: AWS S3, Iowa Mesonet archive, and NCEP
  • GRIB2 decompression and rasterio-based data extraction
  • Deaccumulation of hourly precipitation accumulations to rates
  • Processing region buffering for proper deaccumulation without gaps

• Dataset Class (dynamical_dataset.py): Provides the main dataset interface with:

  • Operational update scheduling (every 3 hours via Kubernetes CronJob)
  • Data validation pipeline
  • Support for both backfill and incremental updates

• Zarr Templates: Complete Zarr v3 metadata templates for all coordinates and data variables with optimized chunking and compression (zstd + blosc).

• Tests: Comprehensive test coverage including:

  • Source file coordinate URL generation for different sources and time periods
  • Product version selection logic (v12 vs pre-v12)
  • Processing region buffering behavior
  • End-to-end backfill and operational update workflows
  • Data validation

Notable Implementation Details

• Version-aware product selection: Automatically selects appropriate MRMS product names based on data timestamp, with pre-v12 products only available via Iowa Mesonet archive
• Pass 1 availability: Pass 1 precipitation data only available from October 2020 onwards; earlier requests are skipped
• Deaccumulation handling: First timestep in output is NaN due to deaccumulation requiring a prior timestep
• Spatial reference: Uses IAU 1965 spheroid (native to MRMS) with sub-pixel difference from WGS84 at 0.01° resolution

https://claude.ai/code/session_01JD7KMBFUaoUjEYcNa6VhtK

---

TODOs

• make sure integration tests aren't too slow / don't OOM

small run to

• check actual compressed size and update chunks/shards if needed
• check resource requirements and adjust

[aldenks]

change all the not np.all(np.isnan(... checks in all tests in this PR to assert that the values are all finite. the only nans should be at the very first timestep of the entire dataset.

[aldenks]

assert "spatial_ref" in common_key and "crs_wkt" in common_keys so we know its not empty

[aldenks]

we have download_file and read_data tests in region_job_test.py. remove this test and move just the download + crs/spatial coords test part of it to template_config_test.py

[aldenks]

assert len == 1

[aldenks]

remove default value

[aldenks]

add mrms_fallback_products_pre_v12 and mrms_fallback_products. these should be empty tuples for all variables in the template config except precipitation_surface. add a new fallback_products: tuple[str, ...] on the source file coord and fill it in here.

for precipitation surface:
mrms_fallback_products = ("MultiSensor_QPE_01H_Pass1", "RadarOnly_QPE_01H")
mrms_fallback_products_pre_v12 = ("RadarOnly_QPE_01H",)

[aldenks]

update this logic to the follwoing after making the changes to add the fallback product attributes the mrms data var internal attrs and mrms source file coord.

from reformatters.common.pydantic import replace

is_pre_v12 = coord.time < MRMS_V12_START
is_recent = coord.time > (pd.Timestamp.now() - pd.Timedelta(hours=12))

sources = 
   match
       case is_pre_v12 -> [iowa]
       case not is_pre_v12 and not is_recent -> [s3]
       case is_recent -> [s3, nomads]
 
products = [coord.product, *coord.fallback_products]

last_exception: Exception | None = None
for product in products:
  for source in sources:
     try:
        return self._download_from_source(replace(coord, product=product), source=source)
     except FileNotFoundError as e:
        last_exception = e
        continue

raise last_exception

[aldenks]

rather than expect a specific order do this:

if reader.count == 2 and coord.time < MRMS_V12_START: find the band with grib discipline 209 and use that. assert that a band with 209 exists if we are in the count == 2 and pre v12 case. set a rasterio_band = n variable in both if/else branches and then use that in the reader.read call.

[aldenks]

remove the default value and explicitly set this in all data variables in this template

[aldenks]

Suggested change

	comment="Average precipitation rate over the previous hour. Derived from MultiSensor_QPE_01H_Pass2 from October 2020, GaugeCorr_QPE_01H before. Units equivalent to mm/s.",
	comment="Average precipitation rate over the previous hour. Derived from MultiSensor_QPE_01H_Pass2 from October 2020, GaugeCorr_QPE_01H before. If primary product is unavailable, falls back to MultiSensor_QPE_01H_Pass1 and then RadarOnly_QPE_01H. Units equivalent to mm/s.",