Immediate

immediate

Targeted, run-preserving AROME-PI and PIAF surface forecasts over WCS.

AromePISource

Bases: _ImmediateSource

AROME-PI surface forecasts every 15 minutes, preserving one hourly run.

resolution selects the native 0.025° or 0.01° grid. The finer grid does not provide mean u/v wind; its defaults are t2m, precipitation and gusts. The 0.025° defaults also include u10/v10; rain is optional on either grid. token may be a callable for credential renewal. API keys use the default auth='apikey'; OAuth access tokens require auth='bearer'.

Source code in packages/barograph-met/src/baro/met/immediate.py

class AromePISource(_ImmediateSource):
    """AROME-PI surface forecasts every 15 minutes, preserving one hourly run.

    ``resolution`` selects the native 0.025° or 0.01° grid. The finer grid does
    not provide mean u/v wind; its defaults are t2m, precipitation and gusts.
    The 0.025° defaults also include u10/v10; rain is optional on either grid.
    ``token`` may be a callable for credential renewal. API keys use the default
    ``auth='apikey'``; OAuth access tokens require ``auth='bearer'``.
    """

    def __init__(  # noqa: PLR0913 - explicit source and transport configuration
        self,
        *,
        run: np.datetime64 | datetime,
        token: str | Callable[[], str],
        resolution: Literal["0.025", "0.01"] = "0.025",
        fields: dict[str, str] | None = None,
        tile_deg: float = 0.25,
        auth: Literal["apikey", "bearer"] = "apikey",
        client: httpx.Client | None = None,
    ) -> None:
        if resolution not in {"0.025", "0.01"}:
            raise ValueError("AROME-PI resolution must be 0.025 or 0.01 degrees")
        grid = "0025" if resolution == "0.025" else "001"
        defaults: tuple[str, ...] = ("t2m", "precipitation", "gusts")
        if resolution == "0.025":
            defaults = ("u10", "v10", *defaults)
        model = _Model(
            _AROME_URL + f"MF-NWP-HIGHRES-AROMEPI-{grid}-FRANCE-WCS", 900, 3600, 21600, defaults
        )
        # Validate user configuration before allocating an HTTP client.
        _stamp(run, model.run_seconds)
        transport = WCSClient(model.endpoint, token, auth=auth, client=client)
        try:
            super().__init__(model, run, transport, fields=fields, tile_deg=tile_deg)
        except BaseException:
            transport.close()
            raise

PiafSource

Bases: _ImmediateSource

Deterministic PIAF precipitation, native 5-minute amounts up to 195 minutes.

Both run cadence and output step are five minutes. Despite its WCS name containing RATE, the supported product contains interval amounts in mm, verified against the GRIB parameter, units and statistical interval.

Source code in packages/barograph-met/src/baro/met/immediate.py

class PiafSource(_ImmediateSource):
    """Deterministic PIAF precipitation, native 5-minute amounts up to 195 minutes.

    Both run cadence and output step are five minutes. Despite its WCS name
    containing RATE, the supported product contains interval amounts in mm,
    verified against the GRIB parameter, units and statistical interval.
    """

    def __init__(  # noqa: PLR0913 - explicit source and transport configuration
        self,
        *,
        run: np.datetime64 | datetime,
        token: str | Callable[[], str],
        fields: dict[str, str] | None = None,
        tile_deg: float = 0.25,
        auth: Literal["apikey", "bearer"] = "apikey",
        client: httpx.Client | None = None,
    ) -> None:
        model = _Model(_PIAF_URL, 300, 300, 11700, ("precipitation",))
        _stamp(run, model.run_seconds)
        transport = WCSClient(model.endpoint, token, auth=auth, client=client)
        try:
            super().__init__(model, run, transport, fields=fields, tile_deg=tile_deg)
        except BaseException:
            transport.close()
            raise