Immediate grib

immediate_grib

Strict normalization of deterministic immediate-forecast surface responses.

ImmediateField dataclass

One validated field with ascending axes and explicit accumulation bounds.

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

@dataclass(frozen=True)
class ImmediateField:
    """One validated field with ascending axes and explicit accumulation bounds."""

    values: NDArray[np.float32]
    lat: NDArray[np.float64]
    lon: NDArray[np.float64]
    interval_start: np.datetime64 | None
    interval_end: np.datetime64

decode_surface_response(buf, *, field, run, valid_time, interval_seconds)

Decode a single field; cumulative precipitation still needs differencing.

Precipitation is water-equivalent millimetres (1 kg/m² = 1 mm). Statistical probabilities, percentiles and multi-range products are rejected, including probabilistic PIAF products. Gusts must be a scalar speed interval maximum.

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

def decode_surface_response(
    buf: bytes,
    *,
    field: str,
    run: np.datetime64,
    valid_time: np.datetime64,
    interval_seconds: int,
) -> ImmediateField:
    """Decode a single field; cumulative precipitation still needs differencing.

    Precipitation is water-equivalent millimetres (1 kg/m² = 1 mm). Statistical
    probabilities, percentiles and multi-range products are rejected, including
    probabilistic PIAF products. Gusts must be a scalar speed interval maximum.
    """
    if field not in _PARAMS:
        raise ValueError(f"unsupported surface field: {field}")
    if interval_seconds <= 0:
        raise ValueError("interval_seconds must be positive")
    run, valid_time = np.datetime64(run, "ns"), np.datetime64(valid_time, "ns")
    msg = _semantics(buf, field, run)
    start = _bounds(msg, field, run, valid_time, interval_seconds)
    values, lat, lon = decode_message(buf)
    return ImmediateField(values[::-1], lat[::-1], lon, start, valid_time)

normalize_precipitation(current, *, interval_seconds, previous=None)

Return a native interval amount, subtracting a compatible cumulative field.

Negative counter differences below numerical tolerance are missing locally. A previous field is required only when the current interval is cumulative.

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

def normalize_precipitation(
    current: ImmediateField,
    *,
    interval_seconds: int,
    previous: ImmediateField | None = None,
) -> ImmediateField:
    """Return a native interval amount, subtracting a compatible cumulative field.

    Negative counter differences below numerical tolerance are missing locally.
    A previous field is required only when the current interval is cumulative.
    """
    if interval_seconds <= 0 or current.interval_start is None:
        raise ValueError("expected positive interval and statistical field")
    start = current.interval_end - np.timedelta64(interval_seconds, "s")
    values = current.values.copy()
    if current.interval_start != start:
        if previous is None:
            raise GribFormatError("cumulative precipitation requires preceding counter")
        if previous.interval_start != current.interval_start or previous.interval_end != start:
            raise GribFormatError("precipitation counters have incompatible intervals")
        if not np.array_equal(current.lat, previous.lat) or not np.array_equal(
            current.lon, previous.lon
        ):
            raise GribFormatError("precipitation counters have different grids")
        values -= previous.values
    values = np.where(values < _NEGATIVE_TOLERANCE, np.nan, np.maximum(values, 0)).astype(
        np.float32
    )
    return ImmediateField(values, current.lat, current.lon, start, current.interval_end)