NWP — API reference

Open numerical-weather-prediction data source subpackage — earthlens.nwp. Background and usage are covered under the other pages in this section (Introduction, Usage, Catalog & install); this page is the rendered API.

earthlens.nwp

NWP backend — open numerical-weather-prediction forecasts.

One subpackage over the open NWP buckets (NOAA NODD, ECMWF Open Data, DWD Open Data, with Météo-France / ECCC as follow-ons). Unlike the observation-time backends, NWP is indexed by a forecast time axis (cycle_datetime_utc, forecast_step_hours): start / end select the cycle date range and a steps= / horizon= kwarg picks the lead times. The request is variables = {model_key: [param, ...]} and the output is one bbox-cropped COG per (cycle, step).

Herbie owns the GRIB2 .idx byte-range subsetting for the NOAA / ECMWF models; earthlens contributes a thin per-centre adapter plus direct modules (DWD HTTPS .bz2) for what Herbie does not cover. The [nwp] extra pulls herbie-data + ecmwf-opendata; both SDKs (and the cfgrib / eccodes stack Herbie's import chain needs) are imported lazily, so this package imports without the extra installed.

Public surface (re-exported from this package):

• :class:NWP — the backend; instantiate with a date range, a bbox, and a {model_key: [param, ...]} mapping, then call :meth:NWP.download.
• :class:Catalog — pydantic-backed loader for the bundled nwp_data_catalog.yaml.
• :class:NWPModel — one curated model row (provider, cycles, backend, mirrors, band → selector map).
• :data:CATALOG_PATH — absolute path to the bundled catalog YAML; monkey-patchable to redirect the loader.

Catalog

Bases: AbstractCatalog

Model catalog for the NWP backend.

Reads the bundled nwp_data_catalog.yaml (shipped as package data) and exposes its datasets: block as a typed dict[str, NWPModel]. Instantiate with no arguments (Catalog()) — :func:model_post_init parses the YAML and populates :attr:datasets in one pass.

Attributes:

Name	Type	Description
datasets	dict[str, NWPModel]	Structural map keyed by the model key; each value is an :class:NWPModel.

Examples:

• Load the bundled catalog and check which models are present:

  >>> from earthlens.nwp import Catalog
  >>> cat = Catalog()
  >>> "gfs" in cat and "icon-eu" in cat and "aifs" in cat
  True
  

• Resolve one model and read its download backend:

  >>> from earthlens.nwp import Catalog
  >>> Catalog().get_model("icon-global").backend
  'direct-https'
  

Source code in libs/providers/atmosphere/src/earthlens/nwp/catalog.py

class Catalog(AbstractCatalog):
    """Model catalog for the NWP backend.

    Reads the bundled `nwp_data_catalog.yaml` (shipped as package data)
    and exposes its `datasets:` block as a typed `dict[str, NWPModel]`.
    Instantiate with no arguments (`Catalog()`) —
    :func:`model_post_init` parses the YAML and populates
    :attr:`datasets` in one pass.

    Attributes:
        datasets: Structural map keyed by the model key; each value is
            an :class:`NWPModel`.

    Examples:
        - Load the bundled catalog and check which models are present:
            ```python
            >>> from earthlens.nwp import Catalog
            >>> cat = Catalog()
            >>> "gfs" in cat and "icon-eu" in cat and "aifs" in cat
            True

            ```
        - Resolve one model and read its download backend:
            ```python
            >>> from earthlens.nwp import Catalog
            >>> Catalog().get_model("icon-global").backend
            'direct-https'

            ```
    """

    _catalog_kind: str = "NWP catalog"

    datasets: dict[str, NWPModel] = Field(default_factory=dict)

    @classmethod
    def _autoload(cls) -> dict[str, Any]:
        """Read the bundled catalog from disk.

        Returns:
            dict[str, Any]: The `datasets` read from
                the bundled catalog.
        """
        return {"datasets": _load_catalog_data(CATALOG_PATH)}

    def get_catalog(self) -> dict[str, NWPModel]:
        """Return the structural per-model map (satisfies the base contract)."""
        return self.datasets

    def get_model(self, model_key: str) -> NWPModel:
        """Resolve a model key to its :class:`NWPModel` row.

        Args:
            model_key: A curated model key (e.g. `"gfs"`,
                `"icon-global"`).

        Returns:
            NWPModel: The resolved row.

        Raises:
            ValueError: When `model_key` is unknown (with a
                did-you-mean hint from the base class).

        Examples:
            - Resolve a known model:
                ```python
                >>> from earthlens.nwp import Catalog
                >>> Catalog().get_model("gfs").horizon_h
                384

                ```
            - A typo raises with a did-you-mean hint:
                ```python
                >>> from earthlens.nwp import Catalog
                >>> Catalog().get_model("gffs")  # doctest: +ELLIPSIS
                Traceback (most recent call last):
                    ...
                ValueError: 'gffs' is not in the NWP catalog. Known datasets: [...]. Did you mean 'gfs'?

                ```
        """
        return cast("NWPModel", self.get_dataset(model_key))

    def resolve(self, model_key: str) -> NWPModel:
        """Alias for :meth:`get_model` (matches the other backends' surface)."""
        return self.get_model(model_key)

get_catalog()

Return the structural per-model map (satisfies the base contract).

Source code in libs/providers/atmosphere/src/earthlens/nwp/catalog.py

def get_catalog(self) -> dict[str, NWPModel]:
    """Return the structural per-model map (satisfies the base contract)."""
    return self.datasets

get_model(model_key)

Resolve a model key to its :class:NWPModel row.

Parameters:

Name	Type	Description	Default
model_key	str	A curated model key (e.g. "gfs", "icon-global").	required

Returns:

Name	Type	Description
NWPModel	NWPModel	The resolved row.

Raises:

Type	Description
ValueError	When model_key is unknown (with a did-you-mean hint from the base class).

Examples:

• Resolve a known model:

  >>> from earthlens.nwp import Catalog
  >>> Catalog().get_model("gfs").horizon_h
  384
  

• A typo raises with a did-you-mean hint:

  >>> from earthlens.nwp import Catalog
  >>> Catalog().get_model("gffs")  # doctest: +ELLIPSIS
  Traceback (most recent call last):
      ...
  ValueError: 'gffs' is not in the NWP catalog. Known datasets: [...]. Did you mean 'gfs'?
  

Source code in libs/providers/atmosphere/src/earthlens/nwp/catalog.py

def get_model(self, model_key: str) -> NWPModel:
    """Resolve a model key to its :class:`NWPModel` row.

    Args:
        model_key: A curated model key (e.g. `"gfs"`,
            `"icon-global"`).

    Returns:
        NWPModel: The resolved row.

    Raises:
        ValueError: When `model_key` is unknown (with a
            did-you-mean hint from the base class).

    Examples:
        - Resolve a known model:
            ```python
            >>> from earthlens.nwp import Catalog
            >>> Catalog().get_model("gfs").horizon_h
            384

            ```
        - A typo raises with a did-you-mean hint:
            ```python
            >>> from earthlens.nwp import Catalog
            >>> Catalog().get_model("gffs")  # doctest: +ELLIPSIS
            Traceback (most recent call last):
                ...
            ValueError: 'gffs' is not in the NWP catalog. Known datasets: [...]. Did you mean 'gfs'?

            ```
    """
    return cast("NWPModel", self.get_dataset(model_key))

resolve(model_key)

Alias for :meth:get_model (matches the other backends' surface).

Source code in libs/providers/atmosphere/src/earthlens/nwp/catalog.py

def resolve(self, model_key: str) -> NWPModel:
    """Alias for :meth:`get_model` (matches the other backends' surface)."""
    return self.get_model(model_key)

NWP

Bases: AbstractDataSource

Open numerical-weather-prediction backend (forecast time axis).

Resolves each requested model key against the bundled catalog, dispatches its download to the matching centre module, and yields one bbox-cropped COG per (cycle, step). Open buckets only — no authentication.

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	Fixed "raster"; every model yields gridded output, so the facade always forwards aggregate=.

Source code in libs/providers/atmosphere/src/earthlens/nwp/backend.py

class NWP(AbstractDataSource):
    """Open numerical-weather-prediction backend (forecast time axis).

    Resolves each requested model key against the bundled catalog,
    dispatches its download to the matching centre module, and yields
    one bbox-cropped COG per `(cycle, step)`. Open buckets only — no
    authentication.

    Attributes:
        OUTPUT_KIND: Fixed `"raster"`; every model yields gridded
            output, so the facade always forwards `aggregate=`.
    """

    OUTPUT_KIND: OutputKind = "raster"

    #: Wires the temporal reducer (ARC-1).
    SUPPORTS_AGGREGATE = True

    #: Clips to the exact polygon when `aoi=` carries one, not just its bbox.
    SUPPORTS_POLYGON_AOI = True

    def __init__(
        self,
        start: str,
        end: str,
        variables: dict[str, list[str]],
        lat_lim: list[float],
        lon_lim: list[float],
        temporal_resolution: str = "6hourly",
        path: Path | str | None = None,
        fmt: str = "%Y-%m-%d",
        *,
        mirror: str = "auto",
        steps: list[int] | None = None,
        horizon: int | None = None,
        members: list[str] | None = None,
        mode: str = "subset",
        catalog: Catalog | None = None,
    ):
        """Initialise an NWP backend instance.

        Resolves every requested model key against the catalog
        **before** the parent constructor runs, because the parent
        calls :meth:`_initialize` first and `self.vars` is not yet set
        there.

        Args:
            start: Inclusive start of the cycle-date range (parsed with
                `fmt`).
            end: Inclusive end of the cycle-date range.
            variables: Mapping from model key to a list of parameter
                names, e.g. `{"gfs": ["temperature_2m"]}`.
            lat_lim: `[lat_min, lat_max]` in degrees.
            lon_lim: `[lon_min, lon_max]` in degrees.
            temporal_resolution: Advisory label only — **ignored** by
                NWP. The real cadence is per-model (`cycles_utc` for
                cycles, `step_cadence_h` for steps), so this argument
                does not affect the request. Accepted for parity with
                the other backends and the facade (whose default is
                `"daily"`); defaults to `"6hourly"` here.
            path: Output directory. Created by the parent class.
            fmt: `strptime` format for `start` / `end`.
            mirror: Cloud-mirror key (`"auto"` lets the centre choose).
            steps: Explicit forecast lead times in hours. Defaults to
                `[0]` (the analysis step) when neither `steps` nor
                `horizon` is given.
            horizon: Maximum forecast lead time in hours; expands to a
                step list per model cadence (resolved in `C3`).
            members: Ensemble member ids to fetch (e.g. GEFS `["mean",
                "1", "2"]`, ENS `["control", "10"]`). Defaults to the
                model's first listed member when omitted; ignored for
                deterministic models. One COG is written per
                `(cycle, step, member)`.
            mode: How much of each GRIB2 to download — `"subset"` (the
                default) fetches only the requested bands via the
                `.idx` byte-range index where the model has one
                (`idx: true`), else the whole field; `"whole"` forces
                a full-file download even for `.idx`-capable models,
                then crops. `"whole"` only changes behaviour for the
                NOAA / Herbie centre — the other centres are already
                whole-per-variable, so `mode` is a no-op there.
                `"zarr"` is rejected (no `nwp` catalog row carries a
                `zarr_url`).
            catalog: Optional pre-built :class:`Catalog` (tests inject
                a faked one); defaults to the bundled catalog.

        Raises:
            ValueError: When `variables` is empty, `mode` is not
                `"subset"` / `"whole"`, a model key is unknown, or a
                model declares an unknown `backend:`.
        """
        if not variables:
            raise ValueError(
                "NWP requires a non-empty `variables` mapping of "
                "{model_key: [param, ...]}."
            )
        if mode not in _VALID_MODES:
            if mode == "zarr":
                raise ValueError(
                    "mode='zarr' is not supported: no NWP catalog row carries a "
                    "`zarr_url`, and Zarr sources (NWM, hrrrzarr) are separate "
                    "backends. Use mode='subset' (default) or mode='whole'."
                )
            raise ValueError(
                f"mode must be one of {sorted(_VALID_MODES)}; got {mode!r}."
            )
        self._mode = mode
        self._mirror = mirror
        #: Per-batch context `_fetch` sets up for `_fetch_one`: the crop box
        #: and the two pyramids entry points, imported once per download.
        self._crop_bbox: list[float] = []
        self._open_grib: Any = None
        self._write_cog: Any = None
        self._steps_arg = steps
        self._horizon_arg = horizon
        self._members_arg = members
        self._catalog = catalog if catalog is not None else Catalog()
        self._requests: list[tuple[str, NWPModel, list[str]]] = self._resolve_models(
            variables
        )
        # Centre instances are cached per backend so a multi-cycle fetch
        # reuses one Herbie / ecmwf-opendata adapter rather than rebuilding it.
        self._centres: dict[str, _NWPCentre] = {}

        super().__init__(
            start=start,
            end=end,
            variables=variables,
            temporal_resolution=temporal_resolution,
            lat_lim=lat_lim,
            lon_lim=lon_lim,
            fmt=fmt,
            path=path,
        )
        self._warn_retention()

    def _resolve_models(
        self, variables: dict[str, list[str]]
    ) -> list[tuple[str, NWPModel, list[str]]]:
        """Resolve every requested model key to a catalog row + params.

        Args:
            variables: The `{model_key: [param, ...]}` request.

        Returns:
            list[tuple[str, NWPModel, list[str]]]: One `(key, model,
                params)` triple per request key, in request order.

        Raises:
            ValueError: When a key is unknown (the catalog's
                did-you-mean is surfaced), a model declares an unknown
                `backend:`, or a requested param is not in the model's
                band map.
        """
        resolved: list[tuple[str, NWPModel, list[str]]] = []
        for model_key, params in variables.items():
            model = self._catalog.get_model(model_key)
            if model.backend not in KNOWN_BACKENDS:
                raise ValueError(
                    f"model {model_key!r} declares unknown backend "
                    f"{model.backend!r}; known: {sorted(KNOWN_BACKENDS)}."
                )
            unknown = [p for p in params if p not in model.bands]
            if unknown:
                raise ValueError(
                    f"model {model_key!r} has no band(s) {unknown}; "
                    f"known params: {sorted(model.bands)}."
                )
            resolved.append((model_key, model, list(params)))
        return resolved

    def _warn_retention(self) -> None:
        """Emit `RetentionWarning` for any request older than a model's window.

        Iterates `self._requests` once per construction; a model row with
        `retention_days = None` is treated as archival and is silent. The
        cutoff is computed in naive UTC against `self.time.start_date` so
        the comparison matches the catalog's `start` / `end` parsing.

        The warning message renders both `start` and `cutoff` to the hour
        (`timespec='hours'`) rather than to the day, so a same-day
        sub-window failure reads as "older than 2026-06-16T14:00" rather
        than the ambiguous "older than 2026-06-16".

        Stacklevel attribution: 3 frames is correct for a direct
        `NWP(...)` call (1=this method, 2=`__init__`, 3=caller). The
        :class:`~earthlens.core.EarthLens` facade adds one frame, so a
        facade-route warning is attributed to `earthlens.py`; users
        wanting a precise call-site should filter on
        `category=RetentionWarning` rather than module.
        """
        cutoff_base = dt.datetime.now(dt.UTC).replace(tzinfo=None)
        start = self.time.start_date
        for model_key, model, _params in self._requests:
            window = model.retention_days
            if window is None:
                continue
            cutoff = cutoff_base - dt.timedelta(days=window)
            if start < cutoff:
                warnings.warn(
                    f"{model_key!r} retains ~{window} day(s); requested "
                    f"start {start.isoformat(timespec='hours')} is older than "
                    f"the retention cutoff at {cutoff.isoformat(timespec='hours')} "
                    "UTC — expect empty results.",
                    RetentionWarning,
                    stacklevel=3,
                )

    def _check_input_dates(
        self,
        start: str,
        end: str,
        temporal_resolution: str,
        fmt: str,
    ) -> TemporalExtent:
        """Parse the cycle-date range into a :class:`TemporalExtent`.

        For NWP the `dates` index is the requested **cycle date**
        range; the per-cycle / per-step expansion happens in
        :meth:`_search` (`C3`).

        Args:
            start: Inclusive start of the cycle-date range.
            end: Inclusive end of the cycle-date range.
            temporal_resolution: Advisory cadence label.
            fmt: `strptime` format tried first for a string `start` /
                `end`; a non-matching string falls back to an ISO-8601
                parse, and a `datetime` / `date` ignores it.

        Returns:
            TemporalExtent: Frozen model with parsed bounds.

        Raises:
            ValueError: If `start` parses to a date later than `end`.
        """
        start_dt = to_datetime(start, fmt)
        end_dt = to_datetime(end, fmt)
        dates = date_windows(start_dt, end_dt, "D")
        return TemporalExtent(
            start_date=start_dt,
            end_date=end_dt,
            resolution="D",
            dates=dates,
        )

    def _steps_for(self, model: NWPModel) -> list[int]:
        """Resolve the forecast lead times to fetch for one model (`G1`).

        Precedence: an explicit `steps=` list wins; otherwise `horizon=`
        expands from `0` to the horizon on the model's `step_cadence_h`
        (e.g. every 3 h for GFS), so it does not request hourly steps a
        coarse model never publishes (`M2`); otherwise the default is
        `[0]` (the analysis step), keeping the MVP bounded. A step the
        model still doesn't carry is handled by the `errors` fetch
        policy (`M1`), not here.

        Args:
            model: The resolved catalog row (bounds the request via
                `horizon_h`, and sets the `horizon=` cadence via
                `step_cadence_h`).

        Returns:
            list[int]: Sorted, de-duplicated lead times in hours.

        Raises:
            ValueError: When a requested step exceeds the model's
                `horizon_h`.
        """
        if self._steps_arg is not None:
            steps = sorted({int(s) for s in self._steps_arg})
        elif self._horizon_arg is not None:
            steps = list(
                range(0, int(self._horizon_arg) + 1, max(model.step_cadence_h, 1))
            )
        else:
            steps = [0]
        too_far = [s for s in steps if s > model.horizon_h]
        if too_far:
            raise ValueError(
                f"step(s) {too_far} exceed the {model.horizon_h} h horizon "
                f"of the requested model."
            )
        return steps

    def _members_for(self, model: NWPModel) -> list[str | None]:
        """Resolve the ensemble members to fetch for one model.

        A deterministic model (no `members`) has a single `[None]` axis.
        For an ensemble model, an explicit `members=` list wins (each
        validated against the model's members); otherwise the default is
        the model's first listed member (e.g. the mean/control), keeping
        a plain ensemble request bounded.

        Args:
            model: The resolved catalog row.

        Returns:
            list[str | None]: The member ids to fetch (`[None]` for a
                deterministic model).

        Raises:
            ValueError: When a requested member is not one of the
                model's members.
        """
        if not model.members:
            return [None]
        if self._members_arg is not None:
            unknown = [m for m in self._members_arg if m not in model.members]
            if unknown:
                raise ValueError(
                    f"members {unknown} are not in the model's members {model.members}."
                )
            return list(self._members_arg)
        return [model.members[0]]

    def _centre_for(self, backend: str) -> _NWPCentre:
        """Return the cached :class:`_NWPCentre` for a catalog `backend:`.

        Args:
            backend: The model's `backend:` value (e.g. `"herbie"`).

        Returns:
            _NWPCentre: A centre bound to the output directory; one
                instance per backend, reused across cycles.
        """
        if backend not in self._centres:
            self._centres[backend] = resolve_centre(backend, self.root_dir)
        # Reflect the current download(progress_bar=) onto the centre so a
        # progress-aware SDK (Herbie) can honour it (L4).
        self._centres[backend].show_progress = getattr(self, "_show_progress", True)
        # Give server-side-subsetting centres (the Météo-France WCS API) the
        # request bbox; others ignore it (the backend crops their full field).
        self._centres[backend].bbox = (
            self.space.west,
            self.space.south,
            self.space.east,
            self.space.north,
        )
        return self._centres[backend]

    def _search(self) -> list[RemoteProduct]:
        """Expand the request into one product per `(model, cycle, step)`.

        Walks the cycle grid (`G1`): for each requested model, every
        cycle in the `start`/`end` date range (per the model's
        `cycles_utc`) crossed with every requested forecast step.

        Returns:
            list[RemoteProduct]: One product per `(model, cycle, step)`,
                each carrying the model row, cycle, step, and requested
                params in `metadata` so `_fetch` needs no re-query.
        """
        products: list[RemoteProduct] = []
        for model_key, model, params in self._requests:
            cycles = enumerate_cycles(
                self.time.start_date, self.time.end_date, model.cycles_utc
            )
            for cycle in cycles:
                for step in self._steps_for(model):
                    for member in self._members_for(model):
                        suffix = f".m{member}" if member is not None else ""
                        products.append(
                            RemoteProduct(
                                id=f"{model_key}.{cycle:%Y%m%d%H}.f{step:03d}{suffix}",
                                metadata={
                                    "model_key": model_key,
                                    "model": model,
                                    "cycle": cycle,
                                    "step": step,
                                    "member": member,
                                    "params": params,
                                },
                            )
                        )
        return products

    def _fetch(self, products: list[RemoteProduct]) -> list[Path]:
        """Fetch each product's GRIB2, crop to the bbox, write a COG (`G4`).

        Per product: the matching centre downloads the variable-subset
        GRIB2 (the >99 % bandwidth win — Herbie `.idx` or DWD's
        per-variable files), then `pyramids.grib.open_grib` reads it,
        the result is cropped to the request bbox, and written as a COG.
        Global models on a 0–360° longitude grid are normalised to
        −180..180 first when the bbox reaches into negative longitudes,
        so an Americas crop lands correctly.

        A single `(cycle, step)` can legitimately be unavailable — the
        latest cycle may not be published yet, or a model may not carry
        a step on every cycle (`M2`/`M4`). The `errors` policy (set by
        :meth:`download`, default `"warn"`) governs that: `"warn"` logs
        the miss and keeps the COGs already produced, `"skip"` drops it
        silently, and `"raise"` aborts the whole fetch.

        Args:
            products: The products from :meth:`_search`.

        Returns:
            list[Path]: One cropped COG path per successfully fetched
                product, in order. Shorter than `products` when some
                were skipped under `errors` in `{"warn", "skip"}`.
        """
        from earthlens.nwp._eccodes import ensure_eccodes

        ensure_eccodes()

        from pyramids.dataset.cog import write_cog
        from pyramids.grib import open_grib

        # The crop box and the two pyramids entry points are the same for
        # every item, so they ride on the instance for the batch instead of
        # widening `_fetch_one` past the base hook's one-argument shape.
        self._crop_bbox = [
            self.space.west,
            self.space.south,
            self.space.east,
            self.space.north,
        ]
        self._open_grib = open_grib
        self._write_cog = write_cog
        try:
            out, _failed = self._run_items(
                products,
                self._fetch_one,
                errors=getattr(self, "_errors", "warn"),
                label="forecast step",
                describe=lambda product: str(product.id),
            )
        finally:
            # Clear the batch context so a later stray `_fetch_one` fails
            # loudly instead of silently reusing the previous download's crop
            # box and pyramids handles.
            self._crop_bbox = []
            self._open_grib = None
            self._write_cog = None
        return out

    def _fetch_one(self, product: RemoteProduct) -> Path:
        """Fetch + crop + write the COG for one product (no error handling).

        Reads the batch context :meth:`_fetch` set up — the crop box and the
        two pyramids entry points it imported once — from the instance.

        Args:
            product: One product from :meth:`_search`.

        Returns:
            pathlib.Path: The written COG path.
        """
        if not self._crop_bbox or self._open_grib is None or self._write_cog is None:
            raise RuntimeError(
                "NWP._fetch_one was called outside a download: the per-batch "
                "crop box and pyramids handles are only set up by _fetch(). "
                "Checked before the download so a stray call costs nothing."
            )
        meta = product.metadata
        centre = self._centre_for(meta["model"].backend)
        grib_path = centre.fetch_one(
            meta["model"],
            meta["cycle"],
            meta["step"],
            meta["params"],
            self._mirror,
            meta.get("member"),
            whole=self._mode == "whole",
        )
        dataset = self._open_grib(str(grib_path))
        dataset = self._normalise_longitude(dataset)
        # touch=False crops to the bbox *extent*; touch=True takes pyramids'
        # cutline path, which masks the field but keeps the full grid extent
        # (and historically crashed on the GRIB driver's EPSG:9122 CRS — fixed
        # in pyramids 0.24.1, pyramids#403 / PY-1). We want the bbox window.
        cropped = crop_to_aoi(dataset, self.space, bbox=self._crop_bbox, touch=False)
        target = self.root_dir / cog_name(
            meta["model_key"], meta["cycle"], meta["step"], meta.get("member")
        )
        self._write_cog(cropped, str(target))
        return target

    def _normalise_longitude(self, dataset):
        """Shift a 0–360° global grid to −180..180 when the bbox needs it.

        `pyramids` `wrap_longitude` only applies to a whole-globe
        0–360 raster (it raises otherwise). A regional model (HRRR) or a
        bbox entirely in the eastern hemisphere needs no shift, so this
        is a no-op unless the request bbox reaches a negative longitude.

        This handles the 0–360 ↔ −180..180 convention only. A bbox that
        *crosses* the antimeridian would need `longitude_min >
        longitude_max`, which the `SpatialExtent` value object forbids (it
        requires `longitude_min <= longitude_max`). pyramids' `crop` itself
        gained antimeridian-crossing support in 0.41, so the residual
        limitation is earthlens's own `SpatialExtent`, not the GIS backend;
        until that is relaxed, split such an AOI into two requests.

        Args:
            dataset: The freshly opened GRIB2 `Dataset`.

        Returns:
            The same `Dataset`, or a longitude-shifted copy.
        """
        if self.space.west >= 0:
            return dataset
        try:
            return dataset.wrap_longitude()
        except ValueError:
            # Not a 0–360 global raster (e.g. a regional model already in
            # −180..180); the bbox is already in the dataset's CRS.
            return dataset

    def download(
        self,
        progress_bar: bool = True,
        aggregate: AggregationConfig | None = None,
        errors: str = "warn",
    ) -> list[Path]:
        """Fetch the requested forecasts as bbox-cropped COGs.

        Args:
            progress_bar: Whether the centres show per-download progress
                (threaded into Herbie's `verbose`).
            aggregate: Optional
                :class:`earthlens.aggregate.AggregationConfig`; reduces
                the `(cycle, step)` COG stack (`C6`).
            errors: How to treat a `(cycle, step)` that fails to fetch or
                crop (an unpublished cycle, a step the model does not
                carry):

                * `"warn"` (default) — log the miss and return the COGs
                  that did succeed.
                * `"skip"` — drop the miss silently.
                * `"raise"` — abort the whole download on the first miss.

        Returns:
            list[Path]: One cropped COG per successfully fetched
                `(cycle, step)`, or — when `aggregate` is set — the
                per-window reduced rasters.

        Raises:
            ValueError: If `errors` is not one of
                `{"raise", "warn", "skip"}`.
        """
        self._show_progress = progress_bar
        # Shared validator: accepts the canonical raise/warn/ignore and keeps
        # nwp's original "skip" working as an alias for "ignore".
        self._errors = self.check_errors_policy(errors)
        paths = self._api_via_search_fetch()
        if aggregate is not None:
            return self._aggregate(paths, aggregate)
        return paths

    def _aggregate(self, paths: list[Path], config: AggregationConfig) -> list[Path]:
        """Reduce the `(cycle, step)` COG stack into per-window COGs (`C6`).

        Labels each COG by the window its **valid time** (`cycle + step`)
        falls in, then reduces the co-registered stack with `config.op`
        via `DatasetCollection.groupby(labels).<op>()` — the COG analog
        of the NetCDF reducer the observation-time backends use. One COG
        is written per window.

        Aggregation requires a **single model**: different models have
        different native grids and cannot be co-registered into one
        stack, so a multi-model request is rejected here rather than
        silently mixing grids.

        Accumulated bands (the `*_acc` convention, e.g.
        `precipitation_acc`) are reduced like any other, but a warning is
        logged because summing/averaging an accumulation across steps
        mixes its step-dependent windows and can mislead (`M3`).

        Args:
            paths: The per-`(cycle, step)` COGs from :meth:`_fetch`.
            config: The aggregation request (`freq` window, `op`
                reducer, `out_dir`, `skipna`).

        Returns:
            list[Path]: The per-window reduced COG paths.

        Raises:
            ValueError: When the request names more than one model.
        """
        if not paths:
            return []
        model_keys = {key for key, _, _ in self._requests}
        if len(model_keys) > 1:
            raise ValueError(
                "aggregate= over an NWP request needs a single model; "
                f"got {sorted(model_keys)}. Different models have different "
                "native grids and cannot be co-registered into one stack — "
                "issue one request per model."
            )
        # The only model row that is not a regular lat/lon raster is ICON
        # global on its native icosahedral grid (DWD `icon_global_icosahedral_…`
        # URL pattern). The shared COG stack reducer assumes co-registered
        # rasters, so refuse the aggregation explicitly here rather than letting
        # pyramids silently mis-grid an unstructured layout (the C4 / M12
        # icosahedral guard).
        only_key, only_model, _ = self._requests[0]
        # `grid_kind` is the declarative source of truth for whether a row
        # is co-registerable into a DatasetCollection stack. The catalog
        # tags every icosahedral DWD ICON row (icon-global, icon-d2,
        # icon-eps, icon-eu-eps, icon-d2-eps); every other row defaults to
        # `"regular-latlon"`.
        if only_model.grid_kind == "icosahedral":
            raise NotImplementedError(
                f"NWP aggregate: {only_key!r} is on an icosahedral grid "
                "(not a regular lat/lon raster); aggregation is not "
                "supported. Request a griddable model (ICON-EU, or a "
                "regridded global feed) instead."
            )
        from pyramids.dataset import Dataset, DatasetCollection
        from pyramids.dataset.cog import write_cog

        op = "mean" if config.op == "auto" else config.op
        # Accumulated fields (precipitation_acc / APCP / tp) carry a running
        # total over a step-dependent window; reducing them across steps by
        # valid time mixes accumulation intervals and can mislead. Warn rather
        # than silently produce wrong totals (M3) — de-accumulation is a future
        # enhancement.
        accumulated = [p for p in self._requests[0][2] if p.endswith("_acc")]
        if accumulated:
            logger.warning(
                f"NWP aggregate: {accumulated} are accumulated field(s); "
                f"reducing them by valid time with op={op!r} mixes accumulation "
                "windows and may give misleading totals. Prefer the per-(cycle, "
                "step) COGs, or de-accumulate before aggregating."
            )
        out_dir = (
            Path(config.out_dir) if config.out_dir is not None else Path(self.root_dir)
        )
        out_dir.mkdir(parents=True, exist_ok=True)

        dated = sorted((parse_cog_valid_time(p), str(p)) for p in paths)
        times = [t for t, _ in dated]
        files = [f for _, f in dated]
        labels = window_labels(times, config.freq)
        collection = DatasetCollection.from_files(files)
        reduced = getattr(collection.groupby(labels), op)(skipna=config.skipna)
        reference = Dataset.read_file(files[0])
        geo, epsg = reference.geotransform, reference.epsg
        model_key = model_keys.pop()
        written: list[Path] = []
        for label, array in reduced.items():
            target = out_dir / f"{model_key}_{op}_{config.freq}_{label}.tif"
            write_cog(
                Dataset.create_from_array(arr=array, geo=geo, epsg=epsg), str(target)
            )
            written.append(target)
        return written

__init__(start, end, variables, lat_lim, lon_lim, temporal_resolution='6hourly', path=None, fmt='%Y-%m-%d', *, mirror='auto', steps=None, horizon=None, members=None, mode='subset', catalog=None)

Initialise an NWP backend instance.

Resolves every requested model key against the catalog before the parent constructor runs, because the parent calls :meth:_initialize first and self.vars is not yet set there.

Parameters:

Name	Type	Description	Default
start	str	Inclusive start of the cycle-date range (parsed with fmt).	required
end	str	Inclusive end of the cycle-date range.	required
variables	dict[str, list[str]]	Mapping from model key to a list of parameter names, e.g. {"gfs": ["temperature_2m"]}.	required
lat_lim	list[float]	[lat_min, lat_max] in degrees.	required
lon_lim	list[float]	[lon_min, lon_max] in degrees.	required
temporal_resolution	str	Advisory label only — ignored by NWP. The real cadence is per-model (cycles_utc for cycles, step_cadence_h for steps), so this argument does not affect the request. Accepted for parity with the other backends and the facade (whose default is "daily"); defaults to "6hourly" here.	'6hourly'
path	Path | str | None	Output directory. Created by the parent class.	None
fmt	str	strptime format for start / end.	'%Y-%m-%d'
mirror	str	Cloud-mirror key ("auto" lets the centre choose).	'auto'
steps	list[int] | None	Explicit forecast lead times in hours. Defaults to [0] (the analysis step) when neither steps nor horizon is given.	None
horizon	int | None	Maximum forecast lead time in hours; expands to a step list per model cadence (resolved in C3).	None
members	list[str] | None	Ensemble member ids to fetch (e.g. GEFS ["mean", "1", "2"], ENS ["control", "10"]). Defaults to the model's first listed member when omitted; ignored for deterministic models. One COG is written per (cycle, step, member).	None
mode	str	How much of each GRIB2 to download — "subset" (the default) fetches only the requested bands via the .idx byte-range index where the model has one (idx: true), else the whole field; "whole" forces a full-file download even for .idx-capable models, then crops. "whole" only changes behaviour for the NOAA / Herbie centre — the other centres are already whole-per-variable, so mode is a no-op there. "zarr" is rejected (no nwp catalog row carries a zarr_url).	'subset'
catalog	Catalog | None	Optional pre-built :class:Catalog (tests inject a faked one); defaults to the bundled catalog.	None

Raises:

Type	Description
ValueError	When variables is empty, mode is not "subset" / "whole", a model key is unknown, or a model declares an unknown backend:.

Source code in libs/providers/atmosphere/src/earthlens/nwp/backend.py

def __init__(
    self,
    start: str,
    end: str,
    variables: dict[str, list[str]],
    lat_lim: list[float],
    lon_lim: list[float],
    temporal_resolution: str = "6hourly",
    path: Path | str | None = None,
    fmt: str = "%Y-%m-%d",
    *,
    mirror: str = "auto",
    steps: list[int] | None = None,
    horizon: int | None = None,
    members: list[str] | None = None,
    mode: str = "subset",
    catalog: Catalog | None = None,
):
    """Initialise an NWP backend instance.

    Resolves every requested model key against the catalog
    **before** the parent constructor runs, because the parent
    calls :meth:`_initialize` first and `self.vars` is not yet set
    there.

    Args:
        start: Inclusive start of the cycle-date range (parsed with
            `fmt`).
        end: Inclusive end of the cycle-date range.
        variables: Mapping from model key to a list of parameter
            names, e.g. `{"gfs": ["temperature_2m"]}`.
        lat_lim: `[lat_min, lat_max]` in degrees.
        lon_lim: `[lon_min, lon_max]` in degrees.
        temporal_resolution: Advisory label only — **ignored** by
            NWP. The real cadence is per-model (`cycles_utc` for
            cycles, `step_cadence_h` for steps), so this argument
            does not affect the request. Accepted for parity with
            the other backends and the facade (whose default is
            `"daily"`); defaults to `"6hourly"` here.
        path: Output directory. Created by the parent class.
        fmt: `strptime` format for `start` / `end`.
        mirror: Cloud-mirror key (`"auto"` lets the centre choose).
        steps: Explicit forecast lead times in hours. Defaults to
            `[0]` (the analysis step) when neither `steps` nor
            `horizon` is given.
        horizon: Maximum forecast lead time in hours; expands to a
            step list per model cadence (resolved in `C3`).
        members: Ensemble member ids to fetch (e.g. GEFS `["mean",
            "1", "2"]`, ENS `["control", "10"]`). Defaults to the
            model's first listed member when omitted; ignored for
            deterministic models. One COG is written per
            `(cycle, step, member)`.
        mode: How much of each GRIB2 to download — `"subset"` (the
            default) fetches only the requested bands via the
            `.idx` byte-range index where the model has one
            (`idx: true`), else the whole field; `"whole"` forces
            a full-file download even for `.idx`-capable models,
            then crops. `"whole"` only changes behaviour for the
            NOAA / Herbie centre — the other centres are already
            whole-per-variable, so `mode` is a no-op there.
            `"zarr"` is rejected (no `nwp` catalog row carries a
            `zarr_url`).
        catalog: Optional pre-built :class:`Catalog` (tests inject
            a faked one); defaults to the bundled catalog.

    Raises:
        ValueError: When `variables` is empty, `mode` is not
            `"subset"` / `"whole"`, a model key is unknown, or a
            model declares an unknown `backend:`.
    """
    if not variables:
        raise ValueError(
            "NWP requires a non-empty `variables` mapping of "
            "{model_key: [param, ...]}."
        )
    if mode not in _VALID_MODES:
        if mode == "zarr":
            raise ValueError(
                "mode='zarr' is not supported: no NWP catalog row carries a "
                "`zarr_url`, and Zarr sources (NWM, hrrrzarr) are separate "
                "backends. Use mode='subset' (default) or mode='whole'."
            )
        raise ValueError(
            f"mode must be one of {sorted(_VALID_MODES)}; got {mode!r}."
        )
    self._mode = mode
    self._mirror = mirror
    #: Per-batch context `_fetch` sets up for `_fetch_one`: the crop box
    #: and the two pyramids entry points, imported once per download.
    self._crop_bbox: list[float] = []
    self._open_grib: Any = None
    self._write_cog: Any = None
    self._steps_arg = steps
    self._horizon_arg = horizon
    self._members_arg = members
    self._catalog = catalog if catalog is not None else Catalog()
    self._requests: list[tuple[str, NWPModel, list[str]]] = self._resolve_models(
        variables
    )
    # Centre instances are cached per backend so a multi-cycle fetch
    # reuses one Herbie / ecmwf-opendata adapter rather than rebuilding it.
    self._centres: dict[str, _NWPCentre] = {}

    super().__init__(
        start=start,
        end=end,
        variables=variables,
        temporal_resolution=temporal_resolution,
        lat_lim=lat_lim,
        lon_lim=lon_lim,
        fmt=fmt,
        path=path,
    )
    self._warn_retention()

download(progress_bar=True, aggregate=None, errors='warn')

Fetch the requested forecasts as bbox-cropped COGs.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Whether the centres show per-download progress (threaded into Herbie's verbose).	True
aggregate	AggregationConfig | None	Optional :class:earthlens.aggregate.AggregationConfig; reduces the (cycle, step) COG stack (C6).	None
errors	str	How to treat a (cycle, step) that fails to fetch or crop (an unpublished cycle, a step the model does not carry): "warn" (default) — log the miss and return the COGs that did succeed. "skip" — drop the miss silently. "raise" — abort the whole download on the first miss.	'warn'

Returns:

Type	Description
list[Path]	list[Path]: One cropped COG per successfully fetched (cycle, step), or — when aggregate is set — the per-window reduced rasters.

Raises:

Type	Description
ValueError	If errors is not one of {"raise", "warn", "skip"}.

Source code in libs/providers/atmosphere/src/earthlens/nwp/backend.py

def download(
    self,
    progress_bar: bool = True,
    aggregate: AggregationConfig | None = None,
    errors: str = "warn",
) -> list[Path]:
    """Fetch the requested forecasts as bbox-cropped COGs.

    Args:
        progress_bar: Whether the centres show per-download progress
            (threaded into Herbie's `verbose`).
        aggregate: Optional
            :class:`earthlens.aggregate.AggregationConfig`; reduces
            the `(cycle, step)` COG stack (`C6`).
        errors: How to treat a `(cycle, step)` that fails to fetch or
            crop (an unpublished cycle, a step the model does not
            carry):

            * `"warn"` (default) — log the miss and return the COGs
              that did succeed.
            * `"skip"` — drop the miss silently.
            * `"raise"` — abort the whole download on the first miss.

    Returns:
        list[Path]: One cropped COG per successfully fetched
            `(cycle, step)`, or — when `aggregate` is set — the
            per-window reduced rasters.

    Raises:
        ValueError: If `errors` is not one of
            `{"raise", "warn", "skip"}`.
    """
    self._show_progress = progress_bar
    # Shared validator: accepts the canonical raise/warn/ignore and keeps
    # nwp's original "skip" working as an alias for "ignore".
    self._errors = self.check_errors_policy(errors)
    paths = self._api_via_search_fetch()
    if aggregate is not None:
        return self._aggregate(paths, aggregate)
    return paths

NWPModel

Bases: BaseModel

One curated NWP model row.

Mirrors a single datasets.<key>: block in nwp_data_catalog.yaml. The model key itself is the parent key in :attr:Catalog.datasets and is not stored on the row.

Attributes:

Name	Type	Description
provider	str	Provider slug (e.g. "noaa-nodd", "dwd-opendata", "ecmwf-opendata", "meteofrance").
model_family	str	Herbie model family token (e.g. "gfs", "hrrr", "ifs"); empty for direct-centre models.
resolution	str	Native horizontal resolution, advisory (e.g. "0.25deg", "13km").
cycles_utc	list[int]	The model's daily run hours, in [0, 23] (e.g. [0, 6, 12, 18]).
horizon_h	int	Maximum forecast lead time in hours.
cadence_h	int | None	Spacing of the run hours in cycles_utc, advisory.
step_cadence_h	int	Spacing between published forecast steps, in hours, used to expand a horizon= request (e.g. 3 for a model whose steps are f000/f003/f006/…). Approximate — a model with a mixed grid (hourly then 3-hourly) is coarsened to a single cadence; the errors="warn" fetch policy skips any expanded step the model doesn't actually carry. Defaults to 1 (hourly).
product	str | None	Herbie product token, required for some models (e.g. HRRR "wrfsfcf"); None otherwise.
format	str	On-disk format the fetch produces ("grib2").
idx	bool	Whether the source exposes a .idx byte-range index (NOAA / ECMWF yes; DWD no — its files are already per-variable).
backend	BackendLiteral	Which download path handles this model (see :data:BackendLiteral).
mirrors	list[str]	Ordered list of cloud-mirror keys the model is served from (e.g. ["aws", "google", "azure"]).
url_template	str | None	For direct-* backends, a str.format template with {cycle} / {date} / {step} / {var} / {var_lc} fields. None for SDK-backed models.
bands	dict[str, str]	Map from earthlens parameter name to the centre's selector — a Herbie search regex (":TMP:2 m above ground:") for SDK models, or a provider var token ("T_2M") for direct ones.
members	list[str]	Ensemble member ids, when the model is an ensemble (empty for deterministic models). The first entry is the default representative fetched when no members= is given (e.g. "mean" for GEFS, "control" for ENS). Each centre maps an id to its SDK's member selector (Herbie member= for GEFS; type=pf + number= for ECMWF ENS).
request_options	dict[str, Any]	Free-form per-centre extras the adapter splats into its request. ECMWF Open Data uses ecmwf_model / stream / type (e.g. {"ecmwf_model": "aifs-single"} for AIFS, {"stream": "enfo", "type": "cf"} for the ENS control); an unsigned-S3 centre uses bucket / key_template / region. Empty for the simple deterministic models.
license	str | None	SPDX-style licence identifier the provider publishes the data under, surfaced as catalog metadata so downstream users and redistribution honour it. Never inferred from the URL — populated row-by-row in nwp_data_catalog.yaml from the provider's stated terms ("PD-US-GOV" for NOAA NODD; "CC-BY-4.0" for ECMWF Open Data and DWD; "Etalab-2.0" for Météo-France; "OGL-Canada-2.0" for ECCC). None only for an ad-hoc row that has not been curated yet.
retention_days	int | None	How long the provider keeps a cycle online before it rolls off the live endpoint. None means archival or unspecified (the backend stays silent); a positive integer is the rolling-window length in days, and NWP.__init__ emits a :class:RetentionWarning when the requested start is older than now - retention_days (the #1 confusing failure for short-retention providers — DWD keeps roughly one day, MF fourteen). Never inferred — populated per row from the provider's stated retention policy.
grid_kind	Literal['regular-latlon', 'icosahedral']	The model's native horizontal grid type. Defaults to "regular-latlon" (every NOAA / ECMWF / DWD-ICON-EU / ECCC row); "icosahedral" flags an unstructured DWD ICON grid (icon-global, icon-d2, icon-eps, icon-eu-eps, icon-d2-eps). NWP._aggregate checks this field — not the URL — to refuse aggregation on a grid pyramids.dataset.DatasetCollection cannot co-register.
title	str | None	Short human-readable label (e.g. "NOAA GFS (Global Forecast System)"). Surfaced as the title column in the federated earthlens datasets where / search / list CLI output, so an nwp row reads the same as its gee / s3 / radar siblings instead of a blank cell. None for an ad-hoc row.
description	str | None	One-sentence summary of the model — provider, domain, resolution, and forecast horizon. Backs the CLI's title fallback and gives datasets search free-text a richer field to match against. None for an ad-hoc row.

Examples:

• Build a minimal Herbie-backed row and read its selector:

  >>> from earthlens.nwp import NWPModel
  >>> row = NWPModel(
  ...     provider="noaa-nodd",
  ...     backend="herbie",
  ...     cycles_utc=[0, 12],
  ...     bands={"temperature_2m": ":TMP:2 m above ground:"},
  ... )
  >>> row.backend
  'herbie'
  >>> row.bands["temperature_2m"]
  ':TMP:2 m above ground:'
  

• Optional fields fall back to documented defaults:

  >>> from earthlens.nwp import NWPModel
  >>> row = NWPModel(provider="dwd-opendata")
  >>> row.format, row.idx, row.cycles_utc
  ('grib2', True, [])
  

Source code in libs/providers/atmosphere/src/earthlens/nwp/catalog.py

class NWPModel(BaseModel):
    """One curated NWP model row.

    Mirrors a single `datasets.<key>:` block in
    `nwp_data_catalog.yaml`. The model key itself is the parent key in
    :attr:`Catalog.datasets` and is not stored on the row.

    Attributes:
        provider: Provider slug (e.g. `"noaa-nodd"`, `"dwd-opendata"`,
            `"ecmwf-opendata"`, `"meteofrance"`).
        model_family: Herbie model family token (e.g. `"gfs"`,
            `"hrrr"`, `"ifs"`); empty for direct-centre models.
        resolution: Native horizontal resolution, advisory (e.g.
            `"0.25deg"`, `"13km"`).
        cycles_utc: The model's daily run hours, in `[0, 23]` (e.g.
            `[0, 6, 12, 18]`).
        horizon_h: Maximum forecast lead time in hours.
        cadence_h: Spacing of the run hours in `cycles_utc`, advisory.
        step_cadence_h: Spacing between published forecast steps, in
            hours, used to expand a `horizon=` request (e.g. `3` for a
            model whose steps are f000/f003/f006/…). Approximate — a
            model with a mixed grid (hourly then 3-hourly) is coarsened
            to a single cadence; the `errors="warn"` fetch policy skips
            any expanded step the model doesn't actually carry. Defaults
            to `1` (hourly).
        product: Herbie product token, required for some models (e.g.
            HRRR `"wrfsfcf"`); `None` otherwise.
        format: On-disk format the fetch produces (`"grib2"`).
        idx: Whether the source exposes a `.idx` byte-range index (NOAA
            / ECMWF yes; DWD no — its files are already per-variable).
        backend: Which download path handles this model (see
            :data:`BackendLiteral`).
        mirrors: Ordered list of cloud-mirror keys the model is served
            from (e.g. `["aws", "google", "azure"]`).
        url_template: For `direct-*` backends, a `str.format` template
            with `{cycle}` / `{date}` / `{step}` / `{var}` / `{var_lc}`
            fields. `None` for SDK-backed models.
        bands: Map from earthlens parameter name to the centre's
            selector — a Herbie `search` regex (`":TMP:2 m above
            ground:"`) for SDK models, or a provider var token
            (`"T_2M"`) for direct ones.
        members: Ensemble member ids, when the model is an ensemble
            (empty for deterministic models). The first entry is the
            default representative fetched when no `members=` is given
            (e.g. `"mean"` for GEFS, `"control"` for ENS). Each centre
            maps an id to its SDK's member selector (Herbie `member=`
            for GEFS; `type=pf` + `number=` for ECMWF ENS).
        request_options: Free-form per-centre extras the adapter splats
            into its request. ECMWF Open Data uses `ecmwf_model` /
            `stream` / `type` (e.g. `{"ecmwf_model": "aifs-single"}` for
            AIFS, `{"stream": "enfo", "type": "cf"}` for the ENS control);
            an unsigned-S3 centre uses `bucket` / `key_template` /
            `region`. Empty for the simple deterministic models.
        license: SPDX-style licence identifier the provider publishes the
            data under, surfaced as catalog metadata so downstream users
            and redistribution honour it. Never inferred from the URL —
            populated row-by-row in `nwp_data_catalog.yaml` from the
            provider's stated terms (`"PD-US-GOV"` for NOAA NODD;
            `"CC-BY-4.0"` for ECMWF Open Data and DWD; `"Etalab-2.0"`
            for Météo-France; `"OGL-Canada-2.0"` for ECCC). `None` only
            for an ad-hoc row that has not been curated yet.
        retention_days: How long the provider keeps a cycle online before
            it rolls off the live endpoint. `None` means archival or
            unspecified (the backend stays silent); a positive integer is
            the rolling-window length in days, and `NWP.__init__` emits a
            :class:`RetentionWarning` when the requested `start` is older
            than `now - retention_days` (the #1 confusing failure for
            short-retention providers — DWD keeps roughly one day, MF
            fourteen). Never inferred — populated per row from the
            provider's stated retention policy.
        grid_kind: The model's native horizontal grid type. Defaults to
            `"regular-latlon"` (every NOAA / ECMWF / DWD-ICON-EU / ECCC
            row); `"icosahedral"` flags an unstructured DWD ICON grid
            (icon-global, icon-d2, icon-eps, icon-eu-eps, icon-d2-eps).
            `NWP._aggregate` checks this field — not the URL — to refuse
            aggregation on a grid `pyramids.dataset.DatasetCollection`
            cannot co-register.
        title: Short human-readable label (e.g. `"NOAA GFS (Global
            Forecast System)"`). Surfaced as the `title` column in the
            federated `earthlens datasets where / search / list` CLI
            output, so an `nwp` row reads the same as its `gee` / `s3` /
            `radar` siblings instead of a blank cell. `None` for an
            ad-hoc row.
        description: One-sentence summary of the model — provider,
            domain, resolution, and forecast horizon. Backs the CLI's
            title fallback and gives `datasets search` free-text a
            richer field to match against. `None` for an ad-hoc row.

    Examples:
        - Build a minimal Herbie-backed row and read its selector:
            ```python
            >>> from earthlens.nwp import NWPModel
            >>> row = NWPModel(
            ...     provider="noaa-nodd",
            ...     backend="herbie",
            ...     cycles_utc=[0, 12],
            ...     bands={"temperature_2m": ":TMP:2 m above ground:"},
            ... )
            >>> row.backend
            'herbie'
            >>> row.bands["temperature_2m"]
            ':TMP:2 m above ground:'

            ```
        - Optional fields fall back to documented defaults:
            ```python
            >>> from earthlens.nwp import NWPModel
            >>> row = NWPModel(provider="dwd-opendata")
            >>> row.format, row.idx, row.cycles_utc
            ('grib2', True, [])

            ```
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    provider: str
    model_family: str = ""
    resolution: str = ""
    cycles_utc: list[int] = Field(default_factory=list)
    horizon_h: int = 0
    cadence_h: int | None = None
    step_cadence_h: int = 1
    product: str | None = None
    format: str = "grib2"
    idx: bool = True
    backend: BackendLiteral = "herbie"
    mirrors: list[str] = Field(default_factory=list)
    url_template: str | None = None
    bands: dict[str, str] = Field(default_factory=dict)
    request_options: dict[str, Any] = Field(default_factory=dict)
    members: list[str] = Field(default_factory=list)
    license: str | None = None
    retention_days: int | None = None
    grid_kind: Literal["regular-latlon", "icosahedral"] = "regular-latlon"
    title: str | None = None
    description: str | None = None

RetentionWarning

Bases: UserWarning

A request asks for a cycle the provider has already rolled off.

Emitted by NWP.__init__ when the resolved model row carries a retention_days and the request start is older than now - retention_days. Subclasses UserWarning so it surfaces by default and can be promoted to an error with warnings.simplefilter ('error', RetentionWarning).

Source code in libs/providers/atmosphere/src/earthlens/nwp/_warnings.py

class RetentionWarning(UserWarning):
    """A request asks for a cycle the provider has already rolled off.

    Emitted by `NWP.__init__` when the resolved model row carries a
    `retention_days` and the request `start` is older than
    `now - retention_days`. Subclasses `UserWarning` so it surfaces by
    default and can be promoted to an error with `warnings.simplefilter
    ('error', RetentionWarning)`.
    """

earthlens.nwp.backend

Backend that fetches open NWP forecasts as bbox-cropped COGs.

NWP(AbstractDataSource) is one backend over the open numerical-weather-prediction buckets — NOAA NODD (GFS / GEFS / HRRR / …), ECMWF Open Data (IFS), DWD Open Data (ICON), with Météo-France / ECCC as follow-ons. It differs from the observation-time backends in its forecast time axis: data is indexed by (cycle_datetime_utc, forecast_step_hours), not a single valid time. start / end select the cycle date range; a steps= / horizon= kwarg picks the forecast lead times; one COG is produced per (cycle, step).

The request shape is variables = {model_key: [param, ...]} (mirrors the GEE / STAC backends). Each param resolves through the catalog to the centre's selector — a Herbie search regex or a DWD variable token. The download path per model is the catalog backend: value, dispatched to a sibling :mod:earthlens.nwp.centres module.

OUTPUT_KIND is fixed "raster": every centre yields a GRIB2 file that the shared pipeline reads with pyramids.grib.open_grib, crops to the request bbox, and writes as a COG (C3); aggregate= reduces the (cycle, step) stack (C6).

NWP

Bases: AbstractDataSource

Open numerical-weather-prediction backend (forecast time axis).

Resolves each requested model key against the bundled catalog, dispatches its download to the matching centre module, and yields one bbox-cropped COG per (cycle, step). Open buckets only — no authentication.

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	Fixed "raster"; every model yields gridded output, so the facade always forwards aggregate=.

Source code in libs/providers/atmosphere/src/earthlens/nwp/backend.py

class NWP(AbstractDataSource):
    """Open numerical-weather-prediction backend (forecast time axis).

    Resolves each requested model key against the bundled catalog,
    dispatches its download to the matching centre module, and yields
    one bbox-cropped COG per `(cycle, step)`. Open buckets only — no
    authentication.

    Attributes:
        OUTPUT_KIND: Fixed `"raster"`; every model yields gridded
            output, so the facade always forwards `aggregate=`.
    """

    OUTPUT_KIND: OutputKind = "raster"

    #: Wires the temporal reducer (ARC-1).
    SUPPORTS_AGGREGATE = True

    #: Clips to the exact polygon when `aoi=` carries one, not just its bbox.
    SUPPORTS_POLYGON_AOI = True

    def __init__(
        self,
        start: str,
        end: str,
        variables: dict[str, list[str]],
        lat_lim: list[float],
        lon_lim: list[float],
        temporal_resolution: str = "6hourly",
        path: Path | str | None = None,
        fmt: str = "%Y-%m-%d",
        *,
        mirror: str = "auto",
        steps: list[int] | None = None,
        horizon: int | None = None,
        members: list[str] | None = None,
        mode: str = "subset",
        catalog: Catalog | None = None,
    ):
        """Initialise an NWP backend instance.

        Resolves every requested model key against the catalog
        **before** the parent constructor runs, because the parent
        calls :meth:`_initialize` first and `self.vars` is not yet set
        there.

        Args:
            start: Inclusive start of the cycle-date range (parsed with
                `fmt`).
            end: Inclusive end of the cycle-date range.
            variables: Mapping from model key to a list of parameter
                names, e.g. `{"gfs": ["temperature_2m"]}`.
            lat_lim: `[lat_min, lat_max]` in degrees.
            lon_lim: `[lon_min, lon_max]` in degrees.
            temporal_resolution: Advisory label only — **ignored** by
                NWP. The real cadence is per-model (`cycles_utc` for
                cycles, `step_cadence_h` for steps), so this argument
                does not affect the request. Accepted for parity with
                the other backends and the facade (whose default is
                `"daily"`); defaults to `"6hourly"` here.
            path: Output directory. Created by the parent class.
            fmt: `strptime` format for `start` / `end`.
            mirror: Cloud-mirror key (`"auto"` lets the centre choose).
            steps: Explicit forecast lead times in hours. Defaults to
                `[0]` (the analysis step) when neither `steps` nor
                `horizon` is given.
            horizon: Maximum forecast lead time in hours; expands to a
                step list per model cadence (resolved in `C3`).
            members: Ensemble member ids to fetch (e.g. GEFS `["mean",
                "1", "2"]`, ENS `["control", "10"]`). Defaults to the
                model's first listed member when omitted; ignored for
                deterministic models. One COG is written per
                `(cycle, step, member)`.
            mode: How much of each GRIB2 to download — `"subset"` (the
                default) fetches only the requested bands via the
                `.idx` byte-range index where the model has one
                (`idx: true`), else the whole field; `"whole"` forces
                a full-file download even for `.idx`-capable models,
                then crops. `"whole"` only changes behaviour for the
                NOAA / Herbie centre — the other centres are already
                whole-per-variable, so `mode` is a no-op there.
                `"zarr"` is rejected (no `nwp` catalog row carries a
                `zarr_url`).
            catalog: Optional pre-built :class:`Catalog` (tests inject
                a faked one); defaults to the bundled catalog.

        Raises:
            ValueError: When `variables` is empty, `mode` is not
                `"subset"` / `"whole"`, a model key is unknown, or a
                model declares an unknown `backend:`.
        """
        if not variables:
            raise ValueError(
                "NWP requires a non-empty `variables` mapping of "
                "{model_key: [param, ...]}."
            )
        if mode not in _VALID_MODES:
            if mode == "zarr":
                raise ValueError(
                    "mode='zarr' is not supported: no NWP catalog row carries a "
                    "`zarr_url`, and Zarr sources (NWM, hrrrzarr) are separate "
                    "backends. Use mode='subset' (default) or mode='whole'."
                )
            raise ValueError(
                f"mode must be one of {sorted(_VALID_MODES)}; got {mode!r}."
            )
        self._mode = mode
        self._mirror = mirror
        #: Per-batch context `_fetch` sets up for `_fetch_one`: the crop box
        #: and the two pyramids entry points, imported once per download.
        self._crop_bbox: list[float] = []
        self._open_grib: Any = None
        self._write_cog: Any = None
        self._steps_arg = steps
        self._horizon_arg = horizon
        self._members_arg = members
        self._catalog = catalog if catalog is not None else Catalog()
        self._requests: list[tuple[str, NWPModel, list[str]]] = self._resolve_models(
            variables
        )
        # Centre instances are cached per backend so a multi-cycle fetch
        # reuses one Herbie / ecmwf-opendata adapter rather than rebuilding it.
        self._centres: dict[str, _NWPCentre] = {}

        super().__init__(
            start=start,
            end=end,
            variables=variables,
            temporal_resolution=temporal_resolution,
            lat_lim=lat_lim,
            lon_lim=lon_lim,
            fmt=fmt,
            path=path,
        )
        self._warn_retention()

    def _resolve_models(
        self, variables: dict[str, list[str]]
    ) -> list[tuple[str, NWPModel, list[str]]]:
        """Resolve every requested model key to a catalog row + params.

        Args:
            variables: The `{model_key: [param, ...]}` request.

        Returns:
            list[tuple[str, NWPModel, list[str]]]: One `(key, model,
                params)` triple per request key, in request order.

        Raises:
            ValueError: When a key is unknown (the catalog's
                did-you-mean is surfaced), a model declares an unknown
                `backend:`, or a requested param is not in the model's
                band map.
        """
        resolved: list[tuple[str, NWPModel, list[str]]] = []
        for model_key, params in variables.items():
            model = self._catalog.get_model(model_key)
            if model.backend not in KNOWN_BACKENDS:
                raise ValueError(
                    f"model {model_key!r} declares unknown backend "
                    f"{model.backend!r}; known: {sorted(KNOWN_BACKENDS)}."
                )
            unknown = [p for p in params if p not in model.bands]
            if unknown:
                raise ValueError(
                    f"model {model_key!r} has no band(s) {unknown}; "
                    f"known params: {sorted(model.bands)}."
                )
            resolved.append((model_key, model, list(params)))
        return resolved

    def _warn_retention(self) -> None:
        """Emit `RetentionWarning` for any request older than a model's window.

        Iterates `self._requests` once per construction; a model row with
        `retention_days = None` is treated as archival and is silent. The
        cutoff is computed in naive UTC against `self.time.start_date` so
        the comparison matches the catalog's `start` / `end` parsing.

        The warning message renders both `start` and `cutoff` to the hour
        (`timespec='hours'`) rather than to the day, so a same-day
        sub-window failure reads as "older than 2026-06-16T14:00" rather
        than the ambiguous "older than 2026-06-16".

        Stacklevel attribution: 3 frames is correct for a direct
        `NWP(...)` call (1=this method, 2=`__init__`, 3=caller). The
        :class:`~earthlens.core.EarthLens` facade adds one frame, so a
        facade-route warning is attributed to `earthlens.py`; users
        wanting a precise call-site should filter on
        `category=RetentionWarning` rather than module.
        """
        cutoff_base = dt.datetime.now(dt.UTC).replace(tzinfo=None)
        start = self.time.start_date
        for model_key, model, _params in self._requests:
            window = model.retention_days
            if window is None:
                continue
            cutoff = cutoff_base - dt.timedelta(days=window)
            if start < cutoff:
                warnings.warn(
                    f"{model_key!r} retains ~{window} day(s); requested "
                    f"start {start.isoformat(timespec='hours')} is older than "
                    f"the retention cutoff at {cutoff.isoformat(timespec='hours')} "
                    "UTC — expect empty results.",
                    RetentionWarning,
                    stacklevel=3,
                )

    def _check_input_dates(
        self,
        start: str,
        end: str,
        temporal_resolution: str,
        fmt: str,
    ) -> TemporalExtent:
        """Parse the cycle-date range into a :class:`TemporalExtent`.

        For NWP the `dates` index is the requested **cycle date**
        range; the per-cycle / per-step expansion happens in
        :meth:`_search` (`C3`).

        Args:
            start: Inclusive start of the cycle-date range.
            end: Inclusive end of the cycle-date range.
            temporal_resolution: Advisory cadence label.
            fmt: `strptime` format tried first for a string `start` /
                `end`; a non-matching string falls back to an ISO-8601
                parse, and a `datetime` / `date` ignores it.

        Returns:
            TemporalExtent: Frozen model with parsed bounds.

        Raises:
            ValueError: If `start` parses to a date later than `end`.
        """
        start_dt = to_datetime(start, fmt)
        end_dt = to_datetime(end, fmt)
        dates = date_windows(start_dt, end_dt, "D")
        return TemporalExtent(
            start_date=start_dt,
            end_date=end_dt,
            resolution="D",
            dates=dates,
        )

    def _steps_for(self, model: NWPModel) -> list[int]:
        """Resolve the forecast lead times to fetch for one model (`G1`).

        Precedence: an explicit `steps=` list wins; otherwise `horizon=`
        expands from `0` to the horizon on the model's `step_cadence_h`
        (e.g. every 3 h for GFS), so it does not request hourly steps a
        coarse model never publishes (`M2`); otherwise the default is
        `[0]` (the analysis step), keeping the MVP bounded. A step the
        model still doesn't carry is handled by the `errors` fetch
        policy (`M1`), not here.

        Args:
            model: The resolved catalog row (bounds the request via
                `horizon_h`, and sets the `horizon=` cadence via
                `step_cadence_h`).

        Returns:
            list[int]: Sorted, de-duplicated lead times in hours.

        Raises:
            ValueError: When a requested step exceeds the model's
                `horizon_h`.
        """
        if self._steps_arg is not None:
            steps = sorted({int(s) for s in self._steps_arg})
        elif self._horizon_arg is not None:
            steps = list(
                range(0, int(self._horizon_arg) + 1, max(model.step_cadence_h, 1))
            )
        else:
            steps = [0]
        too_far = [s for s in steps if s > model.horizon_h]
        if too_far:
            raise ValueError(
                f"step(s) {too_far} exceed the {model.horizon_h} h horizon "
                f"of the requested model."
            )
        return steps

    def _members_for(self, model: NWPModel) -> list[str | None]:
        """Resolve the ensemble members to fetch for one model.

        A deterministic model (no `members`) has a single `[None]` axis.
        For an ensemble model, an explicit `members=` list wins (each
        validated against the model's members); otherwise the default is
        the model's first listed member (e.g. the mean/control), keeping
        a plain ensemble request bounded.

        Args:
            model: The resolved catalog row.

        Returns:
            list[str | None]: The member ids to fetch (`[None]` for a
                deterministic model).

        Raises:
            ValueError: When a requested member is not one of the
                model's members.
        """
        if not model.members:
            return [None]
        if self._members_arg is not None:
            unknown = [m for m in self._members_arg if m not in model.members]
            if unknown:
                raise ValueError(
                    f"members {unknown} are not in the model's members {model.members}."
                )
            return list(self._members_arg)
        return [model.members[0]]

    def _centre_for(self, backend: str) -> _NWPCentre:
        """Return the cached :class:`_NWPCentre` for a catalog `backend:`.

        Args:
            backend: The model's `backend:` value (e.g. `"herbie"`).

        Returns:
            _NWPCentre: A centre bound to the output directory; one
                instance per backend, reused across cycles.
        """
        if backend not in self._centres:
            self._centres[backend] = resolve_centre(backend, self.root_dir)
        # Reflect the current download(progress_bar=) onto the centre so a
        # progress-aware SDK (Herbie) can honour it (L4).
        self._centres[backend].show_progress = getattr(self, "_show_progress", True)
        # Give server-side-subsetting centres (the Météo-France WCS API) the
        # request bbox; others ignore it (the backend crops their full field).
        self._centres[backend].bbox = (
            self.space.west,
            self.space.south,
            self.space.east,
            self.space.north,
        )
        return self._centres[backend]

    def _search(self) -> list[RemoteProduct]:
        """Expand the request into one product per `(model, cycle, step)`.

        Walks the cycle grid (`G1`): for each requested model, every
        cycle in the `start`/`end` date range (per the model's
        `cycles_utc`) crossed with every requested forecast step.

        Returns:
            list[RemoteProduct]: One product per `(model, cycle, step)`,
                each carrying the model row, cycle, step, and requested
                params in `metadata` so `_fetch` needs no re-query.
        """
        products: list[RemoteProduct] = []
        for model_key, model, params in self._requests:
            cycles = enumerate_cycles(
                self.time.start_date, self.time.end_date, model.cycles_utc
            )
            for cycle in cycles:
                for step in self._steps_for(model):
                    for member in self._members_for(model):
                        suffix = f".m{member}" if member is not None else ""
                        products.append(
                            RemoteProduct(
                                id=f"{model_key}.{cycle:%Y%m%d%H}.f{step:03d}{suffix}",
                                metadata={
                                    "model_key": model_key,
                                    "model": model,
                                    "cycle": cycle,
                                    "step": step,
                                    "member": member,
                                    "params": params,
                                },
                            )
                        )
        return products

    def _fetch(self, products: list[RemoteProduct]) -> list[Path]:
        """Fetch each product's GRIB2, crop to the bbox, write a COG (`G4`).

        Per product: the matching centre downloads the variable-subset
        GRIB2 (the >99 % bandwidth win — Herbie `.idx` or DWD's
        per-variable files), then `pyramids.grib.open_grib` reads it,
        the result is cropped to the request bbox, and written as a COG.
        Global models on a 0–360° longitude grid are normalised to
        −180..180 first when the bbox reaches into negative longitudes,
        so an Americas crop lands correctly.

        A single `(cycle, step)` can legitimately be unavailable — the
        latest cycle may not be published yet, or a model may not carry
        a step on every cycle (`M2`/`M4`). The `errors` policy (set by
        :meth:`download`, default `"warn"`) governs that: `"warn"` logs
        the miss and keeps the COGs already produced, `"skip"` drops it
        silently, and `"raise"` aborts the whole fetch.

        Args:
            products: The products from :meth:`_search`.

        Returns:
            list[Path]: One cropped COG path per successfully fetched
                product, in order. Shorter than `products` when some
                were skipped under `errors` in `{"warn", "skip"}`.
        """
        from earthlens.nwp._eccodes import ensure_eccodes

        ensure_eccodes()

        from pyramids.dataset.cog import write_cog
        from pyramids.grib import open_grib

        # The crop box and the two pyramids entry points are the same for
        # every item, so they ride on the instance for the batch instead of
        # widening `_fetch_one` past the base hook's one-argument shape.
        self._crop_bbox = [
            self.space.west,
            self.space.south,
            self.space.east,
            self.space.north,
        ]
        self._open_grib = open_grib
        self._write_cog = write_cog
        try:
            out, _failed = self._run_items(
                products,
                self._fetch_one,
                errors=getattr(self, "_errors", "warn"),
                label="forecast step",
                describe=lambda product: str(product.id),
            )
        finally:
            # Clear the batch context so a later stray `_fetch_one` fails
            # loudly instead of silently reusing the previous download's crop
            # box and pyramids handles.
            self._crop_bbox = []
            self._open_grib = None
            self._write_cog = None
        return out

    def _fetch_one(self, product: RemoteProduct) -> Path:
        """Fetch + crop + write the COG for one product (no error handling).

        Reads the batch context :meth:`_fetch` set up — the crop box and the
        two pyramids entry points it imported once — from the instance.

        Args:
            product: One product from :meth:`_search`.

        Returns:
            pathlib.Path: The written COG path.
        """
        if not self._crop_bbox or self._open_grib is None or self._write_cog is None:
            raise RuntimeError(
                "NWP._fetch_one was called outside a download: the per-batch "
                "crop box and pyramids handles are only set up by _fetch(). "
                "Checked before the download so a stray call costs nothing."
            )
        meta = product.metadata
        centre = self._centre_for(meta["model"].backend)
        grib_path = centre.fetch_one(
            meta["model"],
            meta["cycle"],
            meta["step"],
            meta["params"],
            self._mirror,
            meta.get("member"),
            whole=self._mode == "whole",
        )
        dataset = self._open_grib(str(grib_path))
        dataset = self._normalise_longitude(dataset)
        # touch=False crops to the bbox *extent*; touch=True takes pyramids'
        # cutline path, which masks the field but keeps the full grid extent
        # (and historically crashed on the GRIB driver's EPSG:9122 CRS — fixed
        # in pyramids 0.24.1, pyramids#403 / PY-1). We want the bbox window.
        cropped = crop_to_aoi(dataset, self.space, bbox=self._crop_bbox, touch=False)
        target = self.root_dir / cog_name(
            meta["model_key"], meta["cycle"], meta["step"], meta.get("member")
        )
        self._write_cog(cropped, str(target))
        return target

    def _normalise_longitude(self, dataset):
        """Shift a 0–360° global grid to −180..180 when the bbox needs it.

        `pyramids` `wrap_longitude` only applies to a whole-globe
        0–360 raster (it raises otherwise). A regional model (HRRR) or a
        bbox entirely in the eastern hemisphere needs no shift, so this
        is a no-op unless the request bbox reaches a negative longitude.

        This handles the 0–360 ↔ −180..180 convention only. A bbox that
        *crosses* the antimeridian would need `longitude_min >
        longitude_max`, which the `SpatialExtent` value object forbids (it
        requires `longitude_min <= longitude_max`). pyramids' `crop` itself
        gained antimeridian-crossing support in 0.41, so the residual
        limitation is earthlens's own `SpatialExtent`, not the GIS backend;
        until that is relaxed, split such an AOI into two requests.

        Args:
            dataset: The freshly opened GRIB2 `Dataset`.

        Returns:
            The same `Dataset`, or a longitude-shifted copy.
        """
        if self.space.west >= 0:
            return dataset
        try:
            return dataset.wrap_longitude()
        except ValueError:
            # Not a 0–360 global raster (e.g. a regional model already in
            # −180..180); the bbox is already in the dataset's CRS.
            return dataset

    def download(
        self,
        progress_bar: bool = True,
        aggregate: AggregationConfig | None = None,
        errors: str = "warn",
    ) -> list[Path]:
        """Fetch the requested forecasts as bbox-cropped COGs.

        Args:
            progress_bar: Whether the centres show per-download progress
                (threaded into Herbie's `verbose`).
            aggregate: Optional
                :class:`earthlens.aggregate.AggregationConfig`; reduces
                the `(cycle, step)` COG stack (`C6`).
            errors: How to treat a `(cycle, step)` that fails to fetch or
                crop (an unpublished cycle, a step the model does not
                carry):

                * `"warn"` (default) — log the miss and return the COGs
                  that did succeed.
                * `"skip"` — drop the miss silently.
                * `"raise"` — abort the whole download on the first miss.

        Returns:
            list[Path]: One cropped COG per successfully fetched
                `(cycle, step)`, or — when `aggregate` is set — the
                per-window reduced rasters.

        Raises:
            ValueError: If `errors` is not one of
                `{"raise", "warn", "skip"}`.
        """
        self._show_progress = progress_bar
        # Shared validator: accepts the canonical raise/warn/ignore and keeps
        # nwp's original "skip" working as an alias for "ignore".
        self._errors = self.check_errors_policy(errors)
        paths = self._api_via_search_fetch()
        if aggregate is not None:
            return self._aggregate(paths, aggregate)
        return paths

    def _aggregate(self, paths: list[Path], config: AggregationConfig) -> list[Path]:
        """Reduce the `(cycle, step)` COG stack into per-window COGs (`C6`).

        Labels each COG by the window its **valid time** (`cycle + step`)
        falls in, then reduces the co-registered stack with `config.op`
        via `DatasetCollection.groupby(labels).<op>()` — the COG analog
        of the NetCDF reducer the observation-time backends use. One COG
        is written per window.

        Aggregation requires a **single model**: different models have
        different native grids and cannot be co-registered into one
        stack, so a multi-model request is rejected here rather than
        silently mixing grids.

        Accumulated bands (the `*_acc` convention, e.g.
        `precipitation_acc`) are reduced like any other, but a warning is
        logged because summing/averaging an accumulation across steps
        mixes its step-dependent windows and can mislead (`M3`).

        Args:
            paths: The per-`(cycle, step)` COGs from :meth:`_fetch`.
            config: The aggregation request (`freq` window, `op`
                reducer, `out_dir`, `skipna`).

        Returns:
            list[Path]: The per-window reduced COG paths.

        Raises:
            ValueError: When the request names more than one model.
        """
        if not paths:
            return []
        model_keys = {key for key, _, _ in self._requests}
        if len(model_keys) > 1:
            raise ValueError(
                "aggregate= over an NWP request needs a single model; "
                f"got {sorted(model_keys)}. Different models have different "
                "native grids and cannot be co-registered into one stack — "
                "issue one request per model."
            )
        # The only model row that is not a regular lat/lon raster is ICON
        # global on its native icosahedral grid (DWD `icon_global_icosahedral_…`
        # URL pattern). The shared COG stack reducer assumes co-registered
        # rasters, so refuse the aggregation explicitly here rather than letting
        # pyramids silently mis-grid an unstructured layout (the C4 / M12
        # icosahedral guard).
        only_key, only_model, _ = self._requests[0]
        # `grid_kind` is the declarative source of truth for whether a row
        # is co-registerable into a DatasetCollection stack. The catalog
        # tags every icosahedral DWD ICON row (icon-global, icon-d2,
        # icon-eps, icon-eu-eps, icon-d2-eps); every other row defaults to
        # `"regular-latlon"`.
        if only_model.grid_kind == "icosahedral":
            raise NotImplementedError(
                f"NWP aggregate: {only_key!r} is on an icosahedral grid "
                "(not a regular lat/lon raster); aggregation is not "
                "supported. Request a griddable model (ICON-EU, or a "
                "regridded global feed) instead."
            )
        from pyramids.dataset import Dataset, DatasetCollection
        from pyramids.dataset.cog import write_cog

        op = "mean" if config.op == "auto" else config.op
        # Accumulated fields (precipitation_acc / APCP / tp) carry a running
        # total over a step-dependent window; reducing them across steps by
        # valid time mixes accumulation intervals and can mislead. Warn rather
        # than silently produce wrong totals (M3) — de-accumulation is a future
        # enhancement.
        accumulated = [p for p in self._requests[0][2] if p.endswith("_acc")]
        if accumulated:
            logger.warning(
                f"NWP aggregate: {accumulated} are accumulated field(s); "
                f"reducing them by valid time with op={op!r} mixes accumulation "
                "windows and may give misleading totals. Prefer the per-(cycle, "
                "step) COGs, or de-accumulate before aggregating."
            )
        out_dir = (
            Path(config.out_dir) if config.out_dir is not None else Path(self.root_dir)
        )
        out_dir.mkdir(parents=True, exist_ok=True)

        dated = sorted((parse_cog_valid_time(p), str(p)) for p in paths)
        times = [t for t, _ in dated]
        files = [f for _, f in dated]
        labels = window_labels(times, config.freq)
        collection = DatasetCollection.from_files(files)
        reduced = getattr(collection.groupby(labels), op)(skipna=config.skipna)
        reference = Dataset.read_file(files[0])
        geo, epsg = reference.geotransform, reference.epsg
        model_key = model_keys.pop()
        written: list[Path] = []
        for label, array in reduced.items():
            target = out_dir / f"{model_key}_{op}_{config.freq}_{label}.tif"
            write_cog(
                Dataset.create_from_array(arr=array, geo=geo, epsg=epsg), str(target)
            )
            written.append(target)
        return written

__init__(start, end, variables, lat_lim, lon_lim, temporal_resolution='6hourly', path=None, fmt='%Y-%m-%d', *, mirror='auto', steps=None, horizon=None, members=None, mode='subset', catalog=None)

Initialise an NWP backend instance.

Resolves every requested model key against the catalog before the parent constructor runs, because the parent calls :meth:_initialize first and self.vars is not yet set there.

Parameters:

Name	Type	Description	Default
start	str	Inclusive start of the cycle-date range (parsed with fmt).	required
end	str	Inclusive end of the cycle-date range.	required
variables	dict[str, list[str]]	Mapping from model key to a list of parameter names, e.g. {"gfs": ["temperature_2m"]}.	required
lat_lim	list[float]	[lat_min, lat_max] in degrees.	required
lon_lim	list[float]	[lon_min, lon_max] in degrees.	required
temporal_resolution	str	Advisory label only — ignored by NWP. The real cadence is per-model (cycles_utc for cycles, step_cadence_h for steps), so this argument does not affect the request. Accepted for parity with the other backends and the facade (whose default is "daily"); defaults to "6hourly" here.	'6hourly'
path	Path | str | None	Output directory. Created by the parent class.	None
fmt	str	strptime format for start / end.	'%Y-%m-%d'
mirror	str	Cloud-mirror key ("auto" lets the centre choose).	'auto'
steps	list[int] | None	Explicit forecast lead times in hours. Defaults to [0] (the analysis step) when neither steps nor horizon is given.	None
horizon	int | None	Maximum forecast lead time in hours; expands to a step list per model cadence (resolved in C3).	None
members	list[str] | None	Ensemble member ids to fetch (e.g. GEFS ["mean", "1", "2"], ENS ["control", "10"]). Defaults to the model's first listed member when omitted; ignored for deterministic models. One COG is written per (cycle, step, member).	None
mode	str	How much of each GRIB2 to download — "subset" (the default) fetches only the requested bands via the .idx byte-range index where the model has one (idx: true), else the whole field; "whole" forces a full-file download even for .idx-capable models, then crops. "whole" only changes behaviour for the NOAA / Herbie centre — the other centres are already whole-per-variable, so mode is a no-op there. "zarr" is rejected (no nwp catalog row carries a zarr_url).	'subset'
catalog	Catalog | None	Optional pre-built :class:Catalog (tests inject a faked one); defaults to the bundled catalog.	None

Raises:

Type	Description
ValueError	When variables is empty, mode is not "subset" / "whole", a model key is unknown, or a model declares an unknown backend:.

Source code in libs/providers/atmosphere/src/earthlens/nwp/backend.py

def __init__(
    self,
    start: str,
    end: str,
    variables: dict[str, list[str]],
    lat_lim: list[float],
    lon_lim: list[float],
    temporal_resolution: str = "6hourly",
    path: Path | str | None = None,
    fmt: str = "%Y-%m-%d",
    *,
    mirror: str = "auto",
    steps: list[int] | None = None,
    horizon: int | None = None,
    members: list[str] | None = None,
    mode: str = "subset",
    catalog: Catalog | None = None,
):
    """Initialise an NWP backend instance.

    Resolves every requested model key against the catalog
    **before** the parent constructor runs, because the parent
    calls :meth:`_initialize` first and `self.vars` is not yet set
    there.

    Args:
        start: Inclusive start of the cycle-date range (parsed with
            `fmt`).
        end: Inclusive end of the cycle-date range.
        variables: Mapping from model key to a list of parameter
            names, e.g. `{"gfs": ["temperature_2m"]}`.
        lat_lim: `[lat_min, lat_max]` in degrees.
        lon_lim: `[lon_min, lon_max]` in degrees.
        temporal_resolution: Advisory label only — **ignored** by
            NWP. The real cadence is per-model (`cycles_utc` for
            cycles, `step_cadence_h` for steps), so this argument
            does not affect the request. Accepted for parity with
            the other backends and the facade (whose default is
            `"daily"`); defaults to `"6hourly"` here.
        path: Output directory. Created by the parent class.
        fmt: `strptime` format for `start` / `end`.
        mirror: Cloud-mirror key (`"auto"` lets the centre choose).
        steps: Explicit forecast lead times in hours. Defaults to
            `[0]` (the analysis step) when neither `steps` nor
            `horizon` is given.
        horizon: Maximum forecast lead time in hours; expands to a
            step list per model cadence (resolved in `C3`).
        members: Ensemble member ids to fetch (e.g. GEFS `["mean",
            "1", "2"]`, ENS `["control", "10"]`). Defaults to the
            model's first listed member when omitted; ignored for
            deterministic models. One COG is written per
            `(cycle, step, member)`.
        mode: How much of each GRIB2 to download — `"subset"` (the
            default) fetches only the requested bands via the
            `.idx` byte-range index where the model has one
            (`idx: true`), else the whole field; `"whole"` forces
            a full-file download even for `.idx`-capable models,
            then crops. `"whole"` only changes behaviour for the
            NOAA / Herbie centre — the other centres are already
            whole-per-variable, so `mode` is a no-op there.
            `"zarr"` is rejected (no `nwp` catalog row carries a
            `zarr_url`).
        catalog: Optional pre-built :class:`Catalog` (tests inject
            a faked one); defaults to the bundled catalog.

    Raises:
        ValueError: When `variables` is empty, `mode` is not
            `"subset"` / `"whole"`, a model key is unknown, or a
            model declares an unknown `backend:`.
    """
    if not variables:
        raise ValueError(
            "NWP requires a non-empty `variables` mapping of "
            "{model_key: [param, ...]}."
        )
    if mode not in _VALID_MODES:
        if mode == "zarr":
            raise ValueError(
                "mode='zarr' is not supported: no NWP catalog row carries a "
                "`zarr_url`, and Zarr sources (NWM, hrrrzarr) are separate "
                "backends. Use mode='subset' (default) or mode='whole'."
            )
        raise ValueError(
            f"mode must be one of {sorted(_VALID_MODES)}; got {mode!r}."
        )
    self._mode = mode
    self._mirror = mirror
    #: Per-batch context `_fetch` sets up for `_fetch_one`: the crop box
    #: and the two pyramids entry points, imported once per download.
    self._crop_bbox: list[float] = []
    self._open_grib: Any = None
    self._write_cog: Any = None
    self._steps_arg = steps
    self._horizon_arg = horizon
    self._members_arg = members
    self._catalog = catalog if catalog is not None else Catalog()
    self._requests: list[tuple[str, NWPModel, list[str]]] = self._resolve_models(
        variables
    )
    # Centre instances are cached per backend so a multi-cycle fetch
    # reuses one Herbie / ecmwf-opendata adapter rather than rebuilding it.
    self._centres: dict[str, _NWPCentre] = {}

    super().__init__(
        start=start,
        end=end,
        variables=variables,
        temporal_resolution=temporal_resolution,
        lat_lim=lat_lim,
        lon_lim=lon_lim,
        fmt=fmt,
        path=path,
    )
    self._warn_retention()

download(progress_bar=True, aggregate=None, errors='warn')

Fetch the requested forecasts as bbox-cropped COGs.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Whether the centres show per-download progress (threaded into Herbie's verbose).	True
aggregate	AggregationConfig | None	Optional :class:earthlens.aggregate.AggregationConfig; reduces the (cycle, step) COG stack (C6).	None
errors	str	How to treat a (cycle, step) that fails to fetch or crop (an unpublished cycle, a step the model does not carry): "warn" (default) — log the miss and return the COGs that did succeed. "skip" — drop the miss silently. "raise" — abort the whole download on the first miss.	'warn'

Returns:

Type	Description
list[Path]	list[Path]: One cropped COG per successfully fetched (cycle, step), or — when aggregate is set — the per-window reduced rasters.

Raises:

Type	Description
ValueError	If errors is not one of {"raise", "warn", "skip"}.

Source code in libs/providers/atmosphere/src/earthlens/nwp/backend.py

def download(
    self,
    progress_bar: bool = True,
    aggregate: AggregationConfig | None = None,
    errors: str = "warn",
) -> list[Path]:
    """Fetch the requested forecasts as bbox-cropped COGs.

    Args:
        progress_bar: Whether the centres show per-download progress
            (threaded into Herbie's `verbose`).
        aggregate: Optional
            :class:`earthlens.aggregate.AggregationConfig`; reduces
            the `(cycle, step)` COG stack (`C6`).
        errors: How to treat a `(cycle, step)` that fails to fetch or
            crop (an unpublished cycle, a step the model does not
            carry):

            * `"warn"` (default) — log the miss and return the COGs
              that did succeed.
            * `"skip"` — drop the miss silently.
            * `"raise"` — abort the whole download on the first miss.

    Returns:
        list[Path]: One cropped COG per successfully fetched
            `(cycle, step)`, or — when `aggregate` is set — the
            per-window reduced rasters.

    Raises:
        ValueError: If `errors` is not one of
            `{"raise", "warn", "skip"}`.
    """
    self._show_progress = progress_bar
    # Shared validator: accepts the canonical raise/warn/ignore and keeps
    # nwp's original "skip" working as an alias for "ignore".
    self._errors = self.check_errors_policy(errors)
    paths = self._api_via_search_fetch()
    if aggregate is not None:
        return self._aggregate(paths, aggregate)
    return paths

earthlens.nwp.catalog

Dataset-catalog loader for the NWP backend.

Hosts :class:Catalog, the pydantic-backed reader for the bundled nwp_data_catalog.yaml. Mirrors the shape of :mod:earthlens.ecmwf.catalog and :mod:earthlens.gee.catalog: a single YAML file with a top-level datasets: block keyed by model key (gfs, gefs, hrrr, ifs-hres, icon-global, …). Each block parses into an :class:NWPModel carrying the provider, the forecast cadence (cycles_utc / horizon_h), the download backend (herbie / ecmwf-opendata / direct-https / direct-boto3), the cloud mirrors, the direct-centre url_template, and the param → selector band map.

A model key resolves to an :class:NWPModel via :meth:Catalog.get_model / :meth:Catalog.resolve / Catalog()["..."], each with a did-you-mean hint on a miss (inherited from :class:earthlens.base.AbstractCatalog). The path to the bundled YAML lives at :data:CATALOG_PATH; tests can monkey-patch that module attribute to redirect the loader at a temporary file.

Catalog

Bases: AbstractCatalog

Model catalog for the NWP backend.

Reads the bundled nwp_data_catalog.yaml (shipped as package data) and exposes its datasets: block as a typed dict[str, NWPModel]. Instantiate with no arguments (Catalog()) — :func:model_post_init parses the YAML and populates :attr:datasets in one pass.

Attributes:

Name	Type	Description
datasets	dict[str, NWPModel]	Structural map keyed by the model key; each value is an :class:NWPModel.

Examples:

• Load the bundled catalog and check which models are present:

  >>> from earthlens.nwp import Catalog
  >>> cat = Catalog()
  >>> "gfs" in cat and "icon-eu" in cat and "aifs" in cat
  True
  

• Resolve one model and read its download backend:

  >>> from earthlens.nwp import Catalog
  >>> Catalog().get_model("icon-global").backend
  'direct-https'
  

Source code in libs/providers/atmosphere/src/earthlens/nwp/catalog.py

class Catalog(AbstractCatalog):
    """Model catalog for the NWP backend.

    Reads the bundled `nwp_data_catalog.yaml` (shipped as package data)
    and exposes its `datasets:` block as a typed `dict[str, NWPModel]`.
    Instantiate with no arguments (`Catalog()`) —
    :func:`model_post_init` parses the YAML and populates
    :attr:`datasets` in one pass.

    Attributes:
        datasets: Structural map keyed by the model key; each value is
            an :class:`NWPModel`.

    Examples:
        - Load the bundled catalog and check which models are present:
            ```python
            >>> from earthlens.nwp import Catalog
            >>> cat = Catalog()
            >>> "gfs" in cat and "icon-eu" in cat and "aifs" in cat
            True

            ```
        - Resolve one model and read its download backend:
            ```python
            >>> from earthlens.nwp import Catalog
            >>> Catalog().get_model("icon-global").backend
            'direct-https'

            ```
    """

    _catalog_kind: str = "NWP catalog"

    datasets: dict[str, NWPModel] = Field(default_factory=dict)

    @classmethod
    def _autoload(cls) -> dict[str, Any]:
        """Read the bundled catalog from disk.

        Returns:
            dict[str, Any]: The `datasets` read from
                the bundled catalog.
        """
        return {"datasets": _load_catalog_data(CATALOG_PATH)}

    def get_catalog(self) -> dict[str, NWPModel]:
        """Return the structural per-model map (satisfies the base contract)."""
        return self.datasets

    def get_model(self, model_key: str) -> NWPModel:
        """Resolve a model key to its :class:`NWPModel` row.

        Args:
            model_key: A curated model key (e.g. `"gfs"`,
                `"icon-global"`).

        Returns:
            NWPModel: The resolved row.

        Raises:
            ValueError: When `model_key` is unknown (with a
                did-you-mean hint from the base class).

        Examples:
            - Resolve a known model:
                ```python
                >>> from earthlens.nwp import Catalog
                >>> Catalog().get_model("gfs").horizon_h
                384

                ```
            - A typo raises with a did-you-mean hint:
                ```python
                >>> from earthlens.nwp import Catalog
                >>> Catalog().get_model("gffs")  # doctest: +ELLIPSIS
                Traceback (most recent call last):
                    ...
                ValueError: 'gffs' is not in the NWP catalog. Known datasets: [...]. Did you mean 'gfs'?

                ```
        """
        return cast("NWPModel", self.get_dataset(model_key))

    def resolve(self, model_key: str) -> NWPModel:
        """Alias for :meth:`get_model` (matches the other backends' surface)."""
        return self.get_model(model_key)

get_catalog()

Return the structural per-model map (satisfies the base contract).

Source code in libs/providers/atmosphere/src/earthlens/nwp/catalog.py

def get_catalog(self) -> dict[str, NWPModel]:
    """Return the structural per-model map (satisfies the base contract)."""
    return self.datasets

get_model(model_key)

Resolve a model key to its :class:NWPModel row.

Parameters:

Name	Type	Description	Default
model_key	str	A curated model key (e.g. "gfs", "icon-global").	required

Returns:

Name	Type	Description
NWPModel	NWPModel	The resolved row.

Raises:

Type	Description
ValueError	When model_key is unknown (with a did-you-mean hint from the base class).

Examples:

• Resolve a known model:

  >>> from earthlens.nwp import Catalog
  >>> Catalog().get_model("gfs").horizon_h
  384
  

• A typo raises with a did-you-mean hint:

  >>> from earthlens.nwp import Catalog
  >>> Catalog().get_model("gffs")  # doctest: +ELLIPSIS
  Traceback (most recent call last):
      ...
  ValueError: 'gffs' is not in the NWP catalog. Known datasets: [...]. Did you mean 'gfs'?
  

Source code in libs/providers/atmosphere/src/earthlens/nwp/catalog.py

def get_model(self, model_key: str) -> NWPModel:
    """Resolve a model key to its :class:`NWPModel` row.

    Args:
        model_key: A curated model key (e.g. `"gfs"`,
            `"icon-global"`).

    Returns:
        NWPModel: The resolved row.

    Raises:
        ValueError: When `model_key` is unknown (with a
            did-you-mean hint from the base class).

    Examples:
        - Resolve a known model:
            ```python
            >>> from earthlens.nwp import Catalog
            >>> Catalog().get_model("gfs").horizon_h
            384

            ```
        - A typo raises with a did-you-mean hint:
            ```python
            >>> from earthlens.nwp import Catalog
            >>> Catalog().get_model("gffs")  # doctest: +ELLIPSIS
            Traceback (most recent call last):
                ...
            ValueError: 'gffs' is not in the NWP catalog. Known datasets: [...]. Did you mean 'gfs'?

            ```
    """
    return cast("NWPModel", self.get_dataset(model_key))

resolve(model_key)

Alias for :meth:get_model (matches the other backends' surface).

Source code in libs/providers/atmosphere/src/earthlens/nwp/catalog.py

def resolve(self, model_key: str) -> NWPModel:
    """Alias for :meth:`get_model` (matches the other backends' surface)."""
    return self.get_model(model_key)

NWPModel

Bases: BaseModel

One curated NWP model row.

Mirrors a single datasets.<key>: block in nwp_data_catalog.yaml. The model key itself is the parent key in :attr:Catalog.datasets and is not stored on the row.

Attributes:

Name	Type	Description
provider	str	Provider slug (e.g. "noaa-nodd", "dwd-opendata", "ecmwf-opendata", "meteofrance").
model_family	str	Herbie model family token (e.g. "gfs", "hrrr", "ifs"); empty for direct-centre models.
resolution	str	Native horizontal resolution, advisory (e.g. "0.25deg", "13km").
cycles_utc	list[int]	The model's daily run hours, in [0, 23] (e.g. [0, 6, 12, 18]).
horizon_h	int	Maximum forecast lead time in hours.
cadence_h	int | None	Spacing of the run hours in cycles_utc, advisory.
step_cadence_h	int	Spacing between published forecast steps, in hours, used to expand a horizon= request (e.g. 3 for a model whose steps are f000/f003/f006/…). Approximate — a model with a mixed grid (hourly then 3-hourly) is coarsened to a single cadence; the errors="warn" fetch policy skips any expanded step the model doesn't actually carry. Defaults to 1 (hourly).
product	str | None	Herbie product token, required for some models (e.g. HRRR "wrfsfcf"); None otherwise.
format	str	On-disk format the fetch produces ("grib2").
idx	bool	Whether the source exposes a .idx byte-range index (NOAA / ECMWF yes; DWD no — its files are already per-variable).
backend	BackendLiteral	Which download path handles this model (see :data:BackendLiteral).
mirrors	list[str]	Ordered list of cloud-mirror keys the model is served from (e.g. ["aws", "google", "azure"]).
url_template	str | None	For direct-* backends, a str.format template with {cycle} / {date} / {step} / {var} / {var_lc} fields. None for SDK-backed models.
bands	dict[str, str]	Map from earthlens parameter name to the centre's selector — a Herbie search regex (":TMP:2 m above ground:") for SDK models, or a provider var token ("T_2M") for direct ones.
members	list[str]	Ensemble member ids, when the model is an ensemble (empty for deterministic models). The first entry is the default representative fetched when no members= is given (e.g. "mean" for GEFS, "control" for ENS). Each centre maps an id to its SDK's member selector (Herbie member= for GEFS; type=pf + number= for ECMWF ENS).
request_options	dict[str, Any]	Free-form per-centre extras the adapter splats into its request. ECMWF Open Data uses ecmwf_model / stream / type (e.g. {"ecmwf_model": "aifs-single"} for AIFS, {"stream": "enfo", "type": "cf"} for the ENS control); an unsigned-S3 centre uses bucket / key_template / region. Empty for the simple deterministic models.
license	str | None	SPDX-style licence identifier the provider publishes the data under, surfaced as catalog metadata so downstream users and redistribution honour it. Never inferred from the URL — populated row-by-row in nwp_data_catalog.yaml from the provider's stated terms ("PD-US-GOV" for NOAA NODD; "CC-BY-4.0" for ECMWF Open Data and DWD; "Etalab-2.0" for Météo-France; "OGL-Canada-2.0" for ECCC). None only for an ad-hoc row that has not been curated yet.
retention_days	int | None	How long the provider keeps a cycle online before it rolls off the live endpoint. None means archival or unspecified (the backend stays silent); a positive integer is the rolling-window length in days, and NWP.__init__ emits a :class:RetentionWarning when the requested start is older than now - retention_days (the #1 confusing failure for short-retention providers — DWD keeps roughly one day, MF fourteen). Never inferred — populated per row from the provider's stated retention policy.
grid_kind	Literal['regular-latlon', 'icosahedral']	The model's native horizontal grid type. Defaults to "regular-latlon" (every NOAA / ECMWF / DWD-ICON-EU / ECCC row); "icosahedral" flags an unstructured DWD ICON grid (icon-global, icon-d2, icon-eps, icon-eu-eps, icon-d2-eps). NWP._aggregate checks this field — not the URL — to refuse aggregation on a grid pyramids.dataset.DatasetCollection cannot co-register.
title	str | None	Short human-readable label (e.g. "NOAA GFS (Global Forecast System)"). Surfaced as the title column in the federated earthlens datasets where / search / list CLI output, so an nwp row reads the same as its gee / s3 / radar siblings instead of a blank cell. None for an ad-hoc row.
description	str | None	One-sentence summary of the model — provider, domain, resolution, and forecast horizon. Backs the CLI's title fallback and gives datasets search free-text a richer field to match against. None for an ad-hoc row.

Examples:

• Build a minimal Herbie-backed row and read its selector:

  >>> from earthlens.nwp import NWPModel
  >>> row = NWPModel(
  ...     provider="noaa-nodd",
  ...     backend="herbie",
  ...     cycles_utc=[0, 12],
  ...     bands={"temperature_2m": ":TMP:2 m above ground:"},
  ... )
  >>> row.backend
  'herbie'
  >>> row.bands["temperature_2m"]
  ':TMP:2 m above ground:'
  

• Optional fields fall back to documented defaults:

  >>> from earthlens.nwp import NWPModel
  >>> row = NWPModel(provider="dwd-opendata")
  >>> row.format, row.idx, row.cycles_utc
  ('grib2', True, [])
  

Source code in libs/providers/atmosphere/src/earthlens/nwp/catalog.py

class NWPModel(BaseModel):
    """One curated NWP model row.

    Mirrors a single `datasets.<key>:` block in
    `nwp_data_catalog.yaml`. The model key itself is the parent key in
    :attr:`Catalog.datasets` and is not stored on the row.

    Attributes:
        provider: Provider slug (e.g. `"noaa-nodd"`, `"dwd-opendata"`,
            `"ecmwf-opendata"`, `"meteofrance"`).
        model_family: Herbie model family token (e.g. `"gfs"`,
            `"hrrr"`, `"ifs"`); empty for direct-centre models.
        resolution: Native horizontal resolution, advisory (e.g.
            `"0.25deg"`, `"13km"`).
        cycles_utc: The model's daily run hours, in `[0, 23]` (e.g.
            `[0, 6, 12, 18]`).
        horizon_h: Maximum forecast lead time in hours.
        cadence_h: Spacing of the run hours in `cycles_utc`, advisory.
        step_cadence_h: Spacing between published forecast steps, in
            hours, used to expand a `horizon=` request (e.g. `3` for a
            model whose steps are f000/f003/f006/…). Approximate — a
            model with a mixed grid (hourly then 3-hourly) is coarsened
            to a single cadence; the `errors="warn"` fetch policy skips
            any expanded step the model doesn't actually carry. Defaults
            to `1` (hourly).
        product: Herbie product token, required for some models (e.g.
            HRRR `"wrfsfcf"`); `None` otherwise.
        format: On-disk format the fetch produces (`"grib2"`).
        idx: Whether the source exposes a `.idx` byte-range index (NOAA
            / ECMWF yes; DWD no — its files are already per-variable).
        backend: Which download path handles this model (see
            :data:`BackendLiteral`).
        mirrors: Ordered list of cloud-mirror keys the model is served
            from (e.g. `["aws", "google", "azure"]`).
        url_template: For `direct-*` backends, a `str.format` template
            with `{cycle}` / `{date}` / `{step}` / `{var}` / `{var_lc}`
            fields. `None` for SDK-backed models.
        bands: Map from earthlens parameter name to the centre's
            selector — a Herbie `search` regex (`":TMP:2 m above
            ground:"`) for SDK models, or a provider var token
            (`"T_2M"`) for direct ones.
        members: Ensemble member ids, when the model is an ensemble
            (empty for deterministic models). The first entry is the
            default representative fetched when no `members=` is given
            (e.g. `"mean"` for GEFS, `"control"` for ENS). Each centre
            maps an id to its SDK's member selector (Herbie `member=`
            for GEFS; `type=pf` + `number=` for ECMWF ENS).
        request_options: Free-form per-centre extras the adapter splats
            into its request. ECMWF Open Data uses `ecmwf_model` /
            `stream` / `type` (e.g. `{"ecmwf_model": "aifs-single"}` for
            AIFS, `{"stream": "enfo", "type": "cf"}` for the ENS control);
            an unsigned-S3 centre uses `bucket` / `key_template` /
            `region`. Empty for the simple deterministic models.
        license: SPDX-style licence identifier the provider publishes the
            data under, surfaced as catalog metadata so downstream users
            and redistribution honour it. Never inferred from the URL —
            populated row-by-row in `nwp_data_catalog.yaml` from the
            provider's stated terms (`"PD-US-GOV"` for NOAA NODD;
            `"CC-BY-4.0"` for ECMWF Open Data and DWD; `"Etalab-2.0"`
            for Météo-France; `"OGL-Canada-2.0"` for ECCC). `None` only
            for an ad-hoc row that has not been curated yet.
        retention_days: How long the provider keeps a cycle online before
            it rolls off the live endpoint. `None` means archival or
            unspecified (the backend stays silent); a positive integer is
            the rolling-window length in days, and `NWP.__init__` emits a
            :class:`RetentionWarning` when the requested `start` is older
            than `now - retention_days` (the #1 confusing failure for
            short-retention providers — DWD keeps roughly one day, MF
            fourteen). Never inferred — populated per row from the
            provider's stated retention policy.
        grid_kind: The model's native horizontal grid type. Defaults to
            `"regular-latlon"` (every NOAA / ECMWF / DWD-ICON-EU / ECCC
            row); `"icosahedral"` flags an unstructured DWD ICON grid
            (icon-global, icon-d2, icon-eps, icon-eu-eps, icon-d2-eps).
            `NWP._aggregate` checks this field — not the URL — to refuse
            aggregation on a grid `pyramids.dataset.DatasetCollection`
            cannot co-register.
        title: Short human-readable label (e.g. `"NOAA GFS (Global
            Forecast System)"`). Surfaced as the `title` column in the
            federated `earthlens datasets where / search / list` CLI
            output, so an `nwp` row reads the same as its `gee` / `s3` /
            `radar` siblings instead of a blank cell. `None` for an
            ad-hoc row.
        description: One-sentence summary of the model — provider,
            domain, resolution, and forecast horizon. Backs the CLI's
            title fallback and gives `datasets search` free-text a
            richer field to match against. `None` for an ad-hoc row.

    Examples:
        - Build a minimal Herbie-backed row and read its selector:
            ```python
            >>> from earthlens.nwp import NWPModel
            >>> row = NWPModel(
            ...     provider="noaa-nodd",
            ...     backend="herbie",
            ...     cycles_utc=[0, 12],
            ...     bands={"temperature_2m": ":TMP:2 m above ground:"},
            ... )
            >>> row.backend
            'herbie'
            >>> row.bands["temperature_2m"]
            ':TMP:2 m above ground:'

            ```
        - Optional fields fall back to documented defaults:
            ```python
            >>> from earthlens.nwp import NWPModel
            >>> row = NWPModel(provider="dwd-opendata")
            >>> row.format, row.idx, row.cycles_utc
            ('grib2', True, [])

            ```
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    provider: str
    model_family: str = ""
    resolution: str = ""
    cycles_utc: list[int] = Field(default_factory=list)
    horizon_h: int = 0
    cadence_h: int | None = None
    step_cadence_h: int = 1
    product: str | None = None
    format: str = "grib2"
    idx: bool = True
    backend: BackendLiteral = "herbie"
    mirrors: list[str] = Field(default_factory=list)
    url_template: str | None = None
    bands: dict[str, str] = Field(default_factory=dict)
    request_options: dict[str, Any] = Field(default_factory=dict)
    members: list[str] = Field(default_factory=list)
    license: str | None = None
    retention_days: int | None = None
    grid_kind: Literal["regular-latlon", "icosahedral"] = "regular-latlon"
    title: str | None = None
    description: str | None = None

clear_catalog_cache()

Empty the module-level catalog parse cache.

Useful in tests that rewrite the catalog on disk and want to force a re-parse. Production callers do not need this — the cache key includes the file's st_mtime_ns, so any real mutation invalidates the entry on its own.

Source code in libs/providers/atmosphere/src/earthlens/nwp/catalog.py

def clear_catalog_cache() -> None:
    """Empty the module-level catalog parse cache.

    Useful in tests that rewrite the catalog on disk and want to force
    a re-parse. Production callers do not need this — the cache key
    includes the file's `st_mtime_ns`, so any real mutation invalidates
    the entry on its own.
    """
    _CATALOG_CACHE.clear()

earthlens.nwp.centres.base

Centre-dispatch base for the NWP backend.

Each numerical-weather-prediction centre (NOAA NODD, ECMWF Open Data, DWD Open Data, …) has its own download protocol — Herbie's .idx byte-range subsetting, ecmwf-opendata's Client.retrieve, or a plain per-variable HTTPS .bz2 fetch. The NWP backend owns the provider-agnostic half (the cycle-grid walk, the GRIB2→cropped-COG pipeline); the per-centre half — "given a model, cycle, step, the requested params, and a mirror, put a GRIB2 file on disk" — lives behind the :class:_NWPCentre interface implemented by the sibling centres/*.py modules.

:func:resolve_centre maps a model's catalog backend: value to the concrete centre class, importing it lazily so the optional SDK for a centre you do not use never has to be installed.

resolve_centre(backend, save_dir)

Construct the :class:_NWPCentre for a catalog backend: value.

Parameters:

Name	Type	Description	Default
backend	str	The model's backend: value (e.g. "herbie", "direct-https").	required
save_dir	Path | str	Directory raw GRIB2 downloads are written to.	required

Returns:

Name	Type	Description
_NWPCentre	_NWPCentre	A centre instance bound to save_dir.

Raises:

Type	Description
ValueError	When backend has no registered centre.
ImportError	When the centre module is registered but its optional SDK is not installed (re-raised with a hint).

Source code in libs/providers/atmosphere/src/earthlens/nwp/centres/base.py

def resolve_centre(backend: str, save_dir: Path | str) -> _NWPCentre:
    """Construct the :class:`_NWPCentre` for a catalog `backend:` value.

    Args:
        backend: The model's `backend:` value (e.g. `"herbie"`,
            `"direct-https"`).
        save_dir: Directory raw GRIB2 downloads are written to.

    Returns:
        _NWPCentre: A centre instance bound to `save_dir`.

    Raises:
        ValueError: When `backend` has no registered centre.
        ImportError: When the centre module is registered but its
            optional SDK is not installed (re-raised with a hint).
    """
    try:
        module_name, class_name = CENTRE_REGISTRY[backend]
    except KeyError:
        raise ValueError(
            f"no NWP centre registered for backend {backend!r}; "
            f"known backends: {sorted(CENTRE_REGISTRY)}."
        ) from None
    import importlib

    module = importlib.import_module(module_name)
    centre_cls = getattr(module, class_name)
    return cast("_NWPCentre", centre_cls(save_dir))

earthlens.nwp.centres.noaa

NOAA NODD centre — GRIB2 subset fetch via Herbie.

Herbie owns the .idx byte-range subsetting that cuts >99 % of the download volume for the NOAA models (GFS / GEFS / HRRR / RAP / NAM / …). :class:NOAACentre is the thin adapter: it maps the requested earthlens params to a single Herbie search regex, builds the mirror-priority list from the mirror= kwarg, and returns the local path of the variable-subset GRIB2 that Herbie wrote.

Herbie is imported lazily inside :meth:NOAACentre.fetch_one (never at module import) for two reasons: its import chain pulls cfgrib / eccodes (the [nwp] extra + the eccodes binary), and its package __init__ prints a Unicode banner that crashes a cp1252 Windows console — :func:_import_herbie captures that banner and rewrites a missing-dependency import into a friendly earthlens[nwp] hint.

NOAACentre

Bases: _NWPCentre

Herbie-backed fetcher for the NOAA NODD models.

Source code in libs/providers/atmosphere/src/earthlens/nwp/centres/noaa.py

class NOAACentre(_NWPCentre):
    """Herbie-backed fetcher for the NOAA NODD models."""

    def fetch_one(
        self,
        model: NWPModel,
        cycle: dt.datetime,
        step: int,
        params: list[str],
        mirror: str,
        member: str | None = None,
        *,
        whole: bool = False,
    ) -> Path:
        """Download the GRIB2 for one `(cycle, step[, member])`.

        In the default `subset` path, joins the requested params' Herbie
        `search` regexes with `|` into a single `.idx` selector, runs
        `Herbie(...).download(search)`, and returns the path Herbie
        wrote. When `whole` is set, calls `Herbie(...).download(None)`
        instead — Herbie's contract for a full-file download (the whole
        GRIB2 is cropped downstream exactly like a subset).

        Args:
            model: The resolved catalog row.
            cycle: The forecast cycle datetime (UTC).
            step: The forecast lead time in hours.
            params: The requested earthlens parameter names.
            mirror: The selected cloud-mirror key.
            member: Ensemble member id (e.g. GEFS `"mean"` or `"5"`),
                forwarded to Herbie's `member=` — numeric ids become an
                `int`. `None` for a deterministic model.
            whole: When `True`, download the full file (no `.idx`
                byte-range subset) via `download(None)`; when `False`
                (default), subset to the requested bands.

        Returns:
            pathlib.Path: The local GRIB2 file — a variable subset by
                default, or the full field when `whole` is set.
        """
        herbie_cls = _import_herbie()
        search = "|".join(model.bands[p] for p in params)
        kwargs: dict[str, Any] = {
            "model": model.model_family,
            "fxx": step,
            "priority": _priority(mirror, model),
            "save_dir": str(self.save_dir),
            "verbose": self.show_progress,
        }
        if model.product is not None:
            kwargs["product"] = model.product
        if member is not None:
            kwargs["member"] = int(member) if member.isdigit() else member
        # request_options carries any extra Herbie constructor kwargs a model
        # needs — e.g. `domain` for HiResW / HREF. Splat last so the catalog
        # row can override a default if it ever needs to.
        kwargs.update(model.request_options)
        handle = herbie_cls(cycle, **kwargs)
        # `search=None` is Herbie's full-file download; a regex subsets via .idx.
        return Path(handle.download(None if whole else search))

fetch_one(model, cycle, step, params, mirror, member=None, *, whole=False)

Download the GRIB2 for one (cycle, step[, member]).

In the default subset path, joins the requested params' Herbie search regexes with | into a single .idx selector, runs Herbie(...).download(search), and returns the path Herbie wrote. When whole is set, calls Herbie(...).download(None) instead — Herbie's contract for a full-file download (the whole GRIB2 is cropped downstream exactly like a subset).

Parameters:

Name	Type	Description	Default
model	NWPModel	The resolved catalog row.	required
cycle	datetime	The forecast cycle datetime (UTC).	required
step	int	The forecast lead time in hours.	required
params	list[str]	The requested earthlens parameter names.	required
mirror	str	The selected cloud-mirror key.	required
member	str | None	Ensemble member id (e.g. GEFS "mean" or "5"), forwarded to Herbie's member= — numeric ids become an int. None for a deterministic model.	None
whole	bool	When True, download the full file (no .idx byte-range subset) via download(None); when False (default), subset to the requested bands.	False

Returns:

Type	Description
Path	pathlib.Path: The local GRIB2 file — a variable subset by default, or the full field when whole is set.

Source code in libs/providers/atmosphere/src/earthlens/nwp/centres/noaa.py

def fetch_one(
    self,
    model: NWPModel,
    cycle: dt.datetime,
    step: int,
    params: list[str],
    mirror: str,
    member: str | None = None,
    *,
    whole: bool = False,
) -> Path:
    """Download the GRIB2 for one `(cycle, step[, member])`.

    In the default `subset` path, joins the requested params' Herbie
    `search` regexes with `|` into a single `.idx` selector, runs
    `Herbie(...).download(search)`, and returns the path Herbie
    wrote. When `whole` is set, calls `Herbie(...).download(None)`
    instead — Herbie's contract for a full-file download (the whole
    GRIB2 is cropped downstream exactly like a subset).

    Args:
        model: The resolved catalog row.
        cycle: The forecast cycle datetime (UTC).
        step: The forecast lead time in hours.
        params: The requested earthlens parameter names.
        mirror: The selected cloud-mirror key.
        member: Ensemble member id (e.g. GEFS `"mean"` or `"5"`),
            forwarded to Herbie's `member=` — numeric ids become an
            `int`. `None` for a deterministic model.
        whole: When `True`, download the full file (no `.idx`
            byte-range subset) via `download(None)`; when `False`
            (default), subset to the requested bands.

    Returns:
        pathlib.Path: The local GRIB2 file — a variable subset by
            default, or the full field when `whole` is set.
    """
    herbie_cls = _import_herbie()
    search = "|".join(model.bands[p] for p in params)
    kwargs: dict[str, Any] = {
        "model": model.model_family,
        "fxx": step,
        "priority": _priority(mirror, model),
        "save_dir": str(self.save_dir),
        "verbose": self.show_progress,
    }
    if model.product is not None:
        kwargs["product"] = model.product
    if member is not None:
        kwargs["member"] = int(member) if member.isdigit() else member
    # request_options carries any extra Herbie constructor kwargs a model
    # needs — e.g. `domain` for HiResW / HREF. Splat last so the catalog
    # row can override a default if it ever needs to.
    kwargs.update(model.request_options)
    handle = herbie_cls(cycle, **kwargs)
    # `search=None` is Herbie's full-file download; a regex subsets via .idx.
    return Path(handle.download(None if whole else search))

earthlens.nwp.centres.ecmwf

ECMWF Open Data centre — IFS GRIB2 fetch via ecmwf-opendata.

ECMWF publishes IFS HRES / ENS / AIFS forecasts as open data (CC-BY-4.0, no auth) and ships the ecmwf-opendata client, which does its own index-based parameter subsetting. :class:ECMWFCentre maps the requested earthlens params to the client's param tokens ("2t", "tp", …), selects the mirror source, and returns the local GRIB2 the client wrote.

ecmwf-opendata is imported lazily inside :meth:ECMWFCentre.fetch_one so the package imports without the [nwp] extra.

ECMWFCentre

Bases: _NWPCentre

ecmwf-opendata-backed fetcher for the IFS models.

Source code in libs/providers/atmosphere/src/earthlens/nwp/centres/ecmwf.py

class ECMWFCentre(_NWPCentre):
    """`ecmwf-opendata`-backed fetcher for the IFS models."""

    def fetch_one(
        self,
        model: NWPModel,
        cycle: dt.datetime,
        step: int,
        params: list[str],
        mirror: str,
        member: str | None = None,
        *,
        whole: bool = False,
    ) -> Path:
        """Retrieve the param-subset GRIB2 for one `(cycle, step[, member])`.

        `whole` is accepted for interface parity but ignored: `ecmwf-opendata`
        is param-addressed (`Client.retrieve(param=...)`), so there is no
        whole-file request to force — every retrieve is already the
        requested subset.

        Args:
            model: The resolved catalog row.
            cycle: The forecast cycle datetime (UTC).
            step: The forecast lead time in hours.
            params: The requested earthlens parameter names.
            mirror: The selected cloud-mirror key.
            member: ENS member id — a numeric id selects `type=pf` +
                `number=<id>`; `"control"` (or `None`) keeps the
                row's configured type (`cf` for the ENS control).
            whole: Ignored (see above); accepted for `NWP(mode=)` parity.

        Returns:
            pathlib.Path: The local param-subset GRIB2 file.
        """
        client_cls = _import_client()
        opts = model.request_options
        # `ecmwf_model` picks the deterministic IFS (`ifs`, default) vs AIFS
        # (`aifs-single`); `stream` / `type` select the ENS control forecast
        # (`{"stream": "enfo", "type": "cf"}`). All are no-ops for IFS HRES.
        client = client_cls(
            source=_source_for(mirror, model),
            model=opts.get("ecmwf_model", "ifs"),
        )
        target = self.save_dir / grib_name(
            model.model_family or "ifs", cycle, step, member
        )
        ens_type = opts.get("type", "fc")
        if member is not None and member.isdigit():
            # A perturbed ENS member: type=pf + number=<member>.
            ens_type = "pf"
        base: dict[str, Any] = {
            "date": cycle.strftime("%Y-%m-%d"),
            "time": cycle.hour,
            "step": step,
            "type": ens_type,
        }
        if opts.get("stream"):
            base["stream"] = opts["stream"]
        if member is not None and member.isdigit():
            base["number"] = int(member)
        # ecmwf-opendata needs one retrieve per level type / level (a single
        # request can't mix sfc + pl, nor different pressure levels), so group
        # the band tokens and concatenate the per-group GRIBs into one file.
        groups = _group_params([model.bands[p] for p in params])
        tmp = target.with_name(target.name + ".part")
        try:
            with open(tmp, "wb") as handle:
                for index, (level, tokens) in enumerate(groups):
                    part = self.save_dir / f"{target.name}.g{index}"
                    request = {**base, "param": tokens, "target": str(part)}
                    if level is not None:
                        request["levtype"] = "pl"
                        request["levelist"] = level
                    client.retrieve(**request)
                    handle.write(Path(part).read_bytes())
                    part.unlink(missing_ok=True)
            tmp.replace(target)
        except BaseException:
            tmp.unlink(missing_ok=True)
            raise
        return target

fetch_one(model, cycle, step, params, mirror, member=None, *, whole=False)

Retrieve the param-subset GRIB2 for one (cycle, step[, member]).

whole is accepted for interface parity but ignored: ecmwf-opendata is param-addressed (Client.retrieve(param=...)), so there is no whole-file request to force — every retrieve is already the requested subset.

Parameters:

Name	Type	Description	Default
model	NWPModel	The resolved catalog row.	required
cycle	datetime	The forecast cycle datetime (UTC).	required
step	int	The forecast lead time in hours.	required
params	list[str]	The requested earthlens parameter names.	required
mirror	str	The selected cloud-mirror key.	required
member	str | None	ENS member id — a numeric id selects type=pf + number=<id>; "control" (or None) keeps the row's configured type (cf for the ENS control).	None
whole	bool	Ignored (see above); accepted for NWP(mode=) parity.	False

Returns:

Type	Description
Path	pathlib.Path: The local param-subset GRIB2 file.

Source code in libs/providers/atmosphere/src/earthlens/nwp/centres/ecmwf.py

def fetch_one(
    self,
    model: NWPModel,
    cycle: dt.datetime,
    step: int,
    params: list[str],
    mirror: str,
    member: str | None = None,
    *,
    whole: bool = False,
) -> Path:
    """Retrieve the param-subset GRIB2 for one `(cycle, step[, member])`.

    `whole` is accepted for interface parity but ignored: `ecmwf-opendata`
    is param-addressed (`Client.retrieve(param=...)`), so there is no
    whole-file request to force — every retrieve is already the
    requested subset.

    Args:
        model: The resolved catalog row.
        cycle: The forecast cycle datetime (UTC).
        step: The forecast lead time in hours.
        params: The requested earthlens parameter names.
        mirror: The selected cloud-mirror key.
        member: ENS member id — a numeric id selects `type=pf` +
            `number=<id>`; `"control"` (or `None`) keeps the
            row's configured type (`cf` for the ENS control).
        whole: Ignored (see above); accepted for `NWP(mode=)` parity.

    Returns:
        pathlib.Path: The local param-subset GRIB2 file.
    """
    client_cls = _import_client()
    opts = model.request_options
    # `ecmwf_model` picks the deterministic IFS (`ifs`, default) vs AIFS
    # (`aifs-single`); `stream` / `type` select the ENS control forecast
    # (`{"stream": "enfo", "type": "cf"}`). All are no-ops for IFS HRES.
    client = client_cls(
        source=_source_for(mirror, model),
        model=opts.get("ecmwf_model", "ifs"),
    )
    target = self.save_dir / grib_name(
        model.model_family or "ifs", cycle, step, member
    )
    ens_type = opts.get("type", "fc")
    if member is not None and member.isdigit():
        # A perturbed ENS member: type=pf + number=<member>.
        ens_type = "pf"
    base: dict[str, Any] = {
        "date": cycle.strftime("%Y-%m-%d"),
        "time": cycle.hour,
        "step": step,
        "type": ens_type,
    }
    if opts.get("stream"):
        base["stream"] = opts["stream"]
    if member is not None and member.isdigit():
        base["number"] = int(member)
    # ecmwf-opendata needs one retrieve per level type / level (a single
    # request can't mix sfc + pl, nor different pressure levels), so group
    # the band tokens and concatenate the per-group GRIBs into one file.
    groups = _group_params([model.bands[p] for p in params])
    tmp = target.with_name(target.name + ".part")
    try:
        with open(tmp, "wb") as handle:
            for index, (level, tokens) in enumerate(groups):
                part = self.save_dir / f"{target.name}.g{index}"
                request = {**base, "param": tokens, "target": str(part)}
                if level is not None:
                    request["levtype"] = "pl"
                    request["levelist"] = level
                client.retrieve(**request)
                handle.write(Path(part).read_bytes())
                part.unlink(missing_ok=True)
        tmp.replace(target)
    except BaseException:
        tmp.unlink(missing_ok=True)
        raise
    return target

earthlens.nwp.centres.dwd

DWD Open Data centre — ICON GRIB2 fetch over plain HTTPS.

DWD publishes ICON forecasts as per-variable, bz2-compressed GRIB2 files over plain HTTPS (no .idx, no SDK): one file per (cycle, step, variable). :class:DWDCentre builds each variable's URL from the catalog url_template, downloads and decompresses it in-flight, and concatenates the decompressed GRIB messages into a single .grib2 — valid because GRIB is a stream of self-describing messages, so pyramids.grib.open_grib sees every requested band.

Grid caveat. DWD's native ICON-global files are on an icosahedral grid (icon_global_icosahedral_…), which is not a regular lat/lon raster and will not crop meaningfully through the shared _fetch pipeline. For a croppable COG the catalog should point at a regular-lat/lon ICON product (e.g. ICON-EU, or a regridded global feed). The download path here is correct regardless of grid; only the downstream crop assumes a regular raster.

DWDCentre

Bases: _NWPCentre

Direct-HTTPS fetcher for the DWD ICON models.

Source code in libs/providers/atmosphere/src/earthlens/nwp/centres/dwd.py

class DWDCentre(_NWPCentre):
    """Direct-HTTPS fetcher for the DWD ICON models."""

    def fetch_one(
        self,
        model: NWPModel,
        cycle: dt.datetime,
        step: int,
        params: list[str],
        mirror: str,
        member: str | None = None,
        *,
        whole: bool = False,
    ) -> Path:
        """Download + decompress one `.bz2` per variable into one GRIB2.

        Each variable is streamed and fed through an incremental
        `bz2.BZ2Decompressor`, so neither the compressed body nor the
        decompressed result is ever held whole in memory — a global ICON
        band runs to hundreds of megabytes on each side. The decompressed
        messages are appended to a single `.part` that is renamed only once
        every variable has succeeded.

        `member` and `whole` are accepted for interface parity but ignored
        — the ICON rows here are deterministic (ICON-EPS is a separate
        model), and DWD already serves one whole `.bz2` per variable, so
        there is no byte-range subset for `whole` to override.

        Args:
            model: The resolved catalog row (carries `url_template` and
                the param -> DWD variable-token band map).
            cycle: The forecast cycle datetime (UTC).
            step: The forecast lead time in hours.
            params: The requested earthlens parameter names.
            mirror: Ignored — DWD serves from a single origin host
                (kept for interface parity with the other centres).
            member: Ignored (see above).
            whole: Ignored — already whole-per-variable.

        Returns:
            pathlib.Path: One local `.grib2` holding every requested
                band's decompressed messages.

        Raises:
            ValueError: When the model has no `url_template` (not a
                direct-HTTPS model).
            requests.HTTPError: When any variable's download fails — the
                partial file is removed first, so no truncated `.grib2`
                is left for a later `open_grib` to misread.
        """
        from earthlens.base.http import HttpClient

        client = HttpClient()
        out = self.save_dir / grib_name(model.model_family, cycle, step)
        # Stream into a sibling .part and atomically rename on full success, so
        # a failure partway through (variable 2 of N) never leaves a truncated
        # .grib2 at `out` (L1).
        tmp = out.with_name(out.name + ".part")
        try:
            with open(tmp, "wb") as handle:
                for param in params:
                    band_offset = handle.tell()
                    url = self._band_url(model, param, cycle, step)
                    response = client.get(url, timeout=_HTTP_TIMEOUT, stream=True)
                    # Decompress incrementally: a global ICON band is a
                    # multi-hundred-MB .bz2, and `bz2.decompress(resp.content)`
                    # would hold both the whole compressed body and the whole
                    # decompressed result in memory at once.
                    try:
                        written = _decompress_stream(
                            response.iter_content(chunk_size=_CHUNK_SIZE), handle, url
                        )
                    finally:
                        response.close()
                    if not written:
                        # Every requested band must contribute messages. An
                        # empty body would otherwise drop that band from the
                        # concatenated `.grib2` with nothing to show for it —
                        # and under `errors="warn"` the whole step is skipped
                        # with only a log line.
                        raise ValueError(
                            f"{redact_url(url)} returned an empty body for band "
                            f"{param!r}: the decompressed GRIB2 would be missing "
                            "that band entirely. Retry the download."
                        )
                    # `_starts_with_grib` opens its own handle, so the buffered
                    # writes have to reach disk before it can see them.
                    handle.flush()
                    if not _starts_with_grib(tmp, band_offset):
                        raise ValueError(
                            f"{redact_url(url)} did not return GRIB2 for band "
                            f"{param!r}: the decompressed body starts "
                            f"{_head_at(tmp, band_offset)!r}, not {_GRIB_MAGIC!r}. "
                            "The cycle or step is probably not published yet."
                        )
            tmp.replace(out)
        except BaseException:
            tmp.unlink(missing_ok=True)
            raise
        return out

    @staticmethod
    def _band_url(model: NWPModel, param: str, cycle: dt.datetime, step: int) -> str:
        """Build the DWD URL for one band — single-level or pressure-level.

        A surface band token is a bare DWD variable (`"T_2M"`) and uses the
        model's `url_template`. A pressure-level token uses the
        `VAR@level` convention (`"T@850"`) and the `pl_url_template` in
        `request_options`, which additionally takes a `{level}` field.

        Args:
            model: The resolved catalog row.
            param: The requested earthlens parameter name.
            cycle: The forecast cycle datetime (UTC).
            step: The forecast lead time in hours.

        Returns:
            str: The fully-formatted `.grib2.bz2` URL.

        Raises:
            ValueError: When a pressure-level band is requested but the
                row has no `pl_url_template`, or the row has no
                single-level `url_template` for a surface band.
        """
        token = model.bands[param]
        if "@" in token:
            var, level = token.split("@", 1)
            template = model.request_options.get("pl_url_template")
            if not template:
                raise ValueError(
                    f"model {model.model_family!r} has no 'pl_url_template' in "
                    f"request_options for pressure-level band {param!r}."
                )
            return cast(
                "str",
                template.format(
                    cycle=cycle,
                    date=cycle,
                    step=step,
                    level=level,
                    var=var,
                    var_lc=var.lower(),
                ),
            )
        if not model.url_template:
            raise ValueError(
                f"model with backend {model.backend!r} has no url_template; "
                "a direct-HTTPS centre needs one."
            )
        return model.url_template.format(
            cycle=cycle, date=cycle, step=step, var=token, var_lc=token.lower()
        )

fetch_one(model, cycle, step, params, mirror, member=None, *, whole=False)

Download + decompress one .bz2 per variable into one GRIB2.

Each variable is streamed and fed through an incremental bz2.BZ2Decompressor, so neither the compressed body nor the decompressed result is ever held whole in memory — a global ICON band runs to hundreds of megabytes on each side. The decompressed messages are appended to a single .part that is renamed only once every variable has succeeded.

member and whole are accepted for interface parity but ignored — the ICON rows here are deterministic (ICON-EPS is a separate model), and DWD already serves one whole .bz2 per variable, so there is no byte-range subset for whole to override.

Parameters:

Name	Type	Description	Default
model	NWPModel	The resolved catalog row (carries url_template and the param -> DWD variable-token band map).	required
cycle	datetime	The forecast cycle datetime (UTC).	required
step	int	The forecast lead time in hours.	required
params	list[str]	The requested earthlens parameter names.	required
mirror	str	Ignored — DWD serves from a single origin host (kept for interface parity with the other centres).	required
member	str | None	Ignored (see above).	None
whole	bool	Ignored — already whole-per-variable.	False

Returns:

Type	Description
Path	pathlib.Path: One local .grib2 holding every requested band's decompressed messages.

Raises:

Type	Description
ValueError	When the model has no url_template (not a direct-HTTPS model).
HTTPError	When any variable's download fails — the partial file is removed first, so no truncated .grib2 is left for a later open_grib to misread.

Source code in libs/providers/atmosphere/src/earthlens/nwp/centres/dwd.py

def fetch_one(
    self,
    model: NWPModel,
    cycle: dt.datetime,
    step: int,
    params: list[str],
    mirror: str,
    member: str | None = None,
    *,
    whole: bool = False,
) -> Path:
    """Download + decompress one `.bz2` per variable into one GRIB2.

    Each variable is streamed and fed through an incremental
    `bz2.BZ2Decompressor`, so neither the compressed body nor the
    decompressed result is ever held whole in memory — a global ICON
    band runs to hundreds of megabytes on each side. The decompressed
    messages are appended to a single `.part` that is renamed only once
    every variable has succeeded.

    `member` and `whole` are accepted for interface parity but ignored
    — the ICON rows here are deterministic (ICON-EPS is a separate
    model), and DWD already serves one whole `.bz2` per variable, so
    there is no byte-range subset for `whole` to override.

    Args:
        model: The resolved catalog row (carries `url_template` and
            the param -> DWD variable-token band map).
        cycle: The forecast cycle datetime (UTC).
        step: The forecast lead time in hours.
        params: The requested earthlens parameter names.
        mirror: Ignored — DWD serves from a single origin host
            (kept for interface parity with the other centres).
        member: Ignored (see above).
        whole: Ignored — already whole-per-variable.

    Returns:
        pathlib.Path: One local `.grib2` holding every requested
            band's decompressed messages.

    Raises:
        ValueError: When the model has no `url_template` (not a
            direct-HTTPS model).
        requests.HTTPError: When any variable's download fails — the
            partial file is removed first, so no truncated `.grib2`
            is left for a later `open_grib` to misread.
    """
    from earthlens.base.http import HttpClient

    client = HttpClient()
    out = self.save_dir / grib_name(model.model_family, cycle, step)
    # Stream into a sibling .part and atomically rename on full success, so
    # a failure partway through (variable 2 of N) never leaves a truncated
    # .grib2 at `out` (L1).
    tmp = out.with_name(out.name + ".part")
    try:
        with open(tmp, "wb") as handle:
            for param in params:
                band_offset = handle.tell()
                url = self._band_url(model, param, cycle, step)
                response = client.get(url, timeout=_HTTP_TIMEOUT, stream=True)
                # Decompress incrementally: a global ICON band is a
                # multi-hundred-MB .bz2, and `bz2.decompress(resp.content)`
                # would hold both the whole compressed body and the whole
                # decompressed result in memory at once.
                try:
                    written = _decompress_stream(
                        response.iter_content(chunk_size=_CHUNK_SIZE), handle, url
                    )
                finally:
                    response.close()
                if not written:
                    # Every requested band must contribute messages. An
                    # empty body would otherwise drop that band from the
                    # concatenated `.grib2` with nothing to show for it —
                    # and under `errors="warn"` the whole step is skipped
                    # with only a log line.
                    raise ValueError(
                        f"{redact_url(url)} returned an empty body for band "
                        f"{param!r}: the decompressed GRIB2 would be missing "
                        "that band entirely. Retry the download."
                    )
                # `_starts_with_grib` opens its own handle, so the buffered
                # writes have to reach disk before it can see them.
                handle.flush()
                if not _starts_with_grib(tmp, band_offset):
                    raise ValueError(
                        f"{redact_url(url)} did not return GRIB2 for band "
                        f"{param!r}: the decompressed body starts "
                        f"{_head_at(tmp, band_offset)!r}, not {_GRIB_MAGIC!r}. "
                        "The cycle or step is probably not published yet."
                    )
        tmp.replace(out)
    except BaseException:
        tmp.unlink(missing_ok=True)
        raise
    return out