Grib

grib

GRIB2 message index over range requests, and decoding of single messages.

A GRIB2 file is a plain concatenation of self-contained messages (one field: variable x level x step), each starting with GRIB and announcing its own length. When no inventory is published (ARPEGE open data), the index is built by reading only the first KiB of every message. The walk is inherently sequential (each length gives the next offset), so the file is split into segments walked in parallel: each segment starts at the first GRIB signature found in a window at its nominal offset, and the chain of lengths must land exactly on the next segment's start — otherwise the segment is re-walked from the previous verified message, so a false signature inside compressed data cannot corrupt the index.

Decoding needs ecCodes (pip install 'barograph-met[grib]').

GribFormatError

Bases: ValueError

Bytes that do not form a valid GRIB2 message header.

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

class GribFormatError(ValueError):
    """Bytes that do not form a valid GRIB2 message header."""

GribIndex dataclass

Messages of one GRIB2 file, looked up by (parameter, level, step).

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

@dataclass
class GribIndex:
    """Messages of one GRIB2 file, looked up by (parameter, level, step)."""

    key: str
    size: int
    messages: list[GribMessage]

    # -- construction ------------------------------------------------------------------------
    @classmethod
    def build(
        cls,
        store: ByteStore,
        key: str,
        *,
        executor: Executor | None = None,
        segments: int = 32,
    ) -> GribIndex:
        """Walk the message headers of ``key`` (in parallel segments when an executor is given)."""
        size = store.size(key)
        if executor is None or segments <= 1 or size < 16 * _WINDOW:
            return cls(key, size, _walk(store, key, 0, size))
        bounds = [size * i // segments for i in range(1, segments)]
        try:
            starts = list(executor.map(lambda b: _find_start(store, key, b, size), bounds))
            # segment ends are the next segment's start; duplicates collapse
            edges = [*sorted({0, *starts}), size]
            spans = list(itertools.pairwise(edges))
            parts = list(executor.map(lambda ab: _walk(store, key, ab[0], ab[1]), spans))
        except GribFormatError:
            # a false signature broke the chaining: fall back to the exact sequential walk
            return cls(key, size, _walk(store, key, 0, size))
        return cls(key, size, [m for part in parts for m in part])

    def __post_init__(self) -> None:
        self._by_key = {(m.param, m.level_type, m.level, m.step_h): m for m in self.messages}

    # -- queries -----------------------------------------------------------------------------
    def find(
        self, param: tuple[int, int, int], level_type: int, level: float, step_h: float
    ) -> GribMessage | None:
        """The message of ``param`` at ``level`` and ``step_h``, if any."""
        return self._by_key.get((param, level_type, level, step_h))

    def levels(
        self, param: tuple[int, int, int], level_type: int, step_h: float | None = None
    ) -> list[float]:
        """Levels of ``param``, optionally only those published at ``step_h``."""
        return sorted(
            {
                m.level
                for m in self.messages
                if m.param == param
                and m.level_type == level_type
                and (step_h is None or m.step_h == step_h)
            }
        )

    def steps(self) -> list[float]:
        """Forecast steps (hours) present in the file."""
        return sorted({m.step_h for m in self.messages})

    # -- persistence ---------------------------------------------------------------------------
    def to_json(self) -> str:
        """Serialise the index."""
        return json.dumps(
            {
                "version": _INDEX_VERSION,
                "key": self.key,
                "size": self.size,
                "messages": [asdict(m) for m in self.messages],
            }
        )

    @classmethod
    def from_json(cls, text: str) -> GribIndex:
        """Rebuild an index serialised by :meth:`to_json`."""
        d = json.loads(text)
        if d.get("version") != _INDEX_VERSION:
            raise ValueError("obsolete GRIB index version")
        return cls(d["key"], d["size"], [GribMessage(**m) for m in d["messages"]])

    def save(self, path: Path) -> None:
        """Write the index atomically to ``path``."""
        write_atomic(path, self.to_json().encode())

build(store, key, *, executor=None, segments=32) classmethod

Walk the message headers of key (in parallel segments when an executor is given).

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

@classmethod
def build(
    cls,
    store: ByteStore,
    key: str,
    *,
    executor: Executor | None = None,
    segments: int = 32,
) -> GribIndex:
    """Walk the message headers of ``key`` (in parallel segments when an executor is given)."""
    size = store.size(key)
    if executor is None or segments <= 1 or size < 16 * _WINDOW:
        return cls(key, size, _walk(store, key, 0, size))
    bounds = [size * i // segments for i in range(1, segments)]
    try:
        starts = list(executor.map(lambda b: _find_start(store, key, b, size), bounds))
        # segment ends are the next segment's start; duplicates collapse
        edges = [*sorted({0, *starts}), size]
        spans = list(itertools.pairwise(edges))
        parts = list(executor.map(lambda ab: _walk(store, key, ab[0], ab[1]), spans))
    except GribFormatError:
        # a false signature broke the chaining: fall back to the exact sequential walk
        return cls(key, size, _walk(store, key, 0, size))
    return cls(key, size, [m for part in parts for m in part])

find(param, level_type, level, step_h)

The message of param at level and step_h, if any.

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

def find(
    self, param: tuple[int, int, int], level_type: int, level: float, step_h: float
) -> GribMessage | None:
    """The message of ``param`` at ``level`` and ``step_h``, if any."""
    return self._by_key.get((param, level_type, level, step_h))

from_json(text) classmethod

Rebuild an index serialised by :meth:to_json.

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

@classmethod
def from_json(cls, text: str) -> GribIndex:
    """Rebuild an index serialised by :meth:`to_json`."""
    d = json.loads(text)
    if d.get("version") != _INDEX_VERSION:
        raise ValueError("obsolete GRIB index version")
    return cls(d["key"], d["size"], [GribMessage(**m) for m in d["messages"]])

levels(param, level_type, step_h=None)

Levels of param, optionally only those published at step_h.

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

def levels(
    self, param: tuple[int, int, int], level_type: int, step_h: float | None = None
) -> list[float]:
    """Levels of ``param``, optionally only those published at ``step_h``."""
    return sorted(
        {
            m.level
            for m in self.messages
            if m.param == param
            and m.level_type == level_type
            and (step_h is None or m.step_h == step_h)
        }
    )

save(path)

Write the index atomically to path.

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

def save(self, path: Path) -> None:
    """Write the index atomically to ``path``."""
    write_atomic(path, self.to_json().encode())

steps()

Forecast steps (hours) present in the file.

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

def steps(self) -> list[float]:
    """Forecast steps (hours) present in the file."""
    return sorted({m.step_h for m in self.messages})

to_json()

Serialise the index.

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

def to_json(self) -> str:
    """Serialise the index."""
    return json.dumps(
        {
            "version": _INDEX_VERSION,
            "key": self.key,
            "size": self.size,
            "messages": [asdict(m) for m in self.messages],
        }
    )

GribIndexStore

Message indexes of the files of one store: memory, then directory, else built.

An index is built once per process even when several readers ask for it at the same time; a saved index that is damaged, stale (the file size changed) or for another key is rebuilt.

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

class GribIndexStore:
    """Message indexes of the files of one store: memory, then ``directory``, else built.

    An index is built once per process even when several readers ask for it at
    the same time; a saved index that is damaged, stale (the file size changed)
    or for another key is rebuilt.
    """

    def __init__(self, store: ByteStore, directory: Path | None = None) -> None:
        self.store = store
        self.directory = directory
        self._indexes: Lru[str, GribIndex] = Lru(_MAX_INDEXES)
        self._building: dict[str, threading.Lock] = {}
        self._lock = threading.Lock()

    def get(self, key: str, executor: Executor | None = None) -> GribIndex:
        """Index of ``key``."""
        with self._lock:
            if key in self._indexes:
                return self._indexes[key]
            gate = self._building.setdefault(key, threading.Lock())
        with gate:
            with self._lock:
                if key in self._indexes:
                    return self._indexes[key]
            idx = self._saved(key)
            if idx is None:
                idx = GribIndex.build(self.store, key, executor=executor)
                if self.directory is not None:
                    idx.save(self._path(key))
            with self._lock:
                self._indexes[key] = idx
                self._building.pop(key, None)
        return idx

    def _path(self, key: str) -> Path:
        assert self.directory is not None
        return self.directory / f"{hashlib.sha256(key.encode()).hexdigest()[:24]}.json"

    def _saved(self, key: str) -> GribIndex | None:
        """The index saved for ``key``, if it is intact and still matches the file."""
        if self.directory is None:
            return None
        try:
            idx = GribIndex.from_json(self._path(key).read_text())
        except FileNotFoundError:
            return None
        except (ValueError, KeyError, TypeError):  # damaged (e.g. by a crash): rebuild
            return None
        if idx.key != key or idx.size != self.store.size(key):
            return None  # stale or colliding entry
        return idx

get(key, executor=None)

Index of key.

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

def get(self, key: str, executor: Executor | None = None) -> GribIndex:
    """Index of ``key``."""
    with self._lock:
        if key in self._indexes:
            return self._indexes[key]
        gate = self._building.setdefault(key, threading.Lock())
    with gate:
        with self._lock:
            if key in self._indexes:
                return self._indexes[key]
        idx = self._saved(key)
        if idx is None:
            idx = GribIndex.build(self.store, key, executor=executor)
            if self.directory is not None:
                idx.save(self._path(key))
        with self._lock:
            self._indexes[key] = idx
            self._building.pop(key, None)
    return idx

GribMessage dataclass

Where one GRIB2 message is and what it holds (variable, level, step).

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

@dataclass(frozen=True)
class GribMessage:
    """Where one GRIB2 message is and what it holds (variable, level, step)."""

    offset: int
    length: int
    discipline: int
    category: int
    number: int
    level_type: int
    level: float  # in the level type's unit (Pa for isobaric surfaces)
    step_h: float
    grid_template: int
    packing: int
    npoints: int
    start_step_h: float | None = None
    statistical_process: int | None = None

    @property
    def param(self) -> tuple[int, int, int]:
        """``(discipline, category, number)`` identifying the variable."""
        return (self.discipline, self.category, self.number)

param property

(discipline, category, number) identifying the variable.

decode_message(buf)

Values (lat, lon) of a regular lat/lon message, lat descending, lon ascending.

Longitudes are unwrapped (a 328E..42E grid gives 328..402).

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

def decode_message(
    buf: bytes,
) -> tuple[NDArray[np.float32], NDArray[np.float64], NDArray[np.float64]]:
    """Values ``(lat, lon)`` of a regular lat/lon message, lat descending, lon ascending.

    Longitudes are unwrapped (a 328E..42E grid gives 328..402).
    """
    try:
        import eccodes
    except ImportError as exc:  # pragma: no cover - depends on the environment
        raise ImportError("decoding GRIB needs ecCodes: pip install 'barograph-met[grib]'") from exc
    h = eccodes.codes_new_from_message(buf)
    try:
        if eccodes.codes_get(h, "gridType") != "regular_ll":
            raise GribFormatError("only regular lat/lon grids are supported")
        g = {
            k: eccodes.codes_get(h, k)
            for k in (
                "Ni",
                "Nj",
                "latitudeOfFirstGridPointInDegrees",
                "longitudeOfFirstGridPointInDegrees",
                "iDirectionIncrementInDegrees",
                "jDirectionIncrementInDegrees",
                "iScansNegatively",
                "jScansPositively",
                "jPointsAreConsecutive",
                "bitmapPresent",
                "missingValue",
            )
        }
        values = np.asarray(eccodes.codes_get_values(h), dtype=np.float64)
    finally:
        eccodes.codes_release(h)
    if g["jPointsAreConsecutive"]:
        raise GribFormatError("column-major scanning is not supported")
    if g["bitmapPresent"]:
        values[values == g["missingValue"]] = np.nan
    ni, nj = int(g["Ni"]), int(g["Nj"])
    field = values.reshape(nj, ni)
    di, dj = float(g["iDirectionIncrementInDegrees"]), float(g["jDirectionIncrementInDegrees"])
    lat = float(g["latitudeOfFirstGridPointInDegrees"]) + np.arange(nj) * (
        dj if g["jScansPositively"] else -dj
    )
    lon = float(g["longitudeOfFirstGridPointInDegrees"]) + np.arange(ni) * (
        -di if g["iScansNegatively"] else di
    )
    if g["jScansPositively"]:
        field, lat = field[::-1], lat[::-1]
    if g["iScansNegatively"]:
        field, lon = field[:, ::-1], lon[::-1]
    return field.astype(np.float32), lat.astype(np.float64), lon.astype(np.float64)

fetch_messages(store, key, messages, executor)

Download the given messages (one range request each).

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

def fetch_messages(
    store: ByteStore, key: str, messages: Sequence[GribMessage], executor: Executor | None
) -> list[bytes]:
    """Download the given messages (one range request each)."""

    def get(m: GribMessage) -> bytes:
        """Download one message."""
        return store.get(key, m.offset, m.offset + m.length)

    if executor is None or len(messages) <= 1:
        return [get(m) for m in messages]
    return list(executor.map(get, messages))

parse_header(prefix, offset)

Parse sections 0, 3, 4 and 5 of a GRIB2 message from its first bytes.

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

def parse_header(prefix: bytes, offset: int) -> GribMessage:
    """Parse sections 0, 3, 4 and 5 of a GRIB2 message from its first bytes."""
    length = _message_length(prefix, offset)
    info: dict[str, Any] = {"offset": offset, "length": length, "discipline": prefix[6]}
    _read_sections(prefix, offset, info)
    missing = {"category", "level_type", "packing", "grid_template"} - set(info)
    if missing:
        raise GribFormatError(
            f"header at offset {offset} lacks {sorted(missing)} (prefix too short?)"
        )
    return GribMessage(
        offset=offset,
        length=length,
        discipline=info["discipline"],
        category=info["category"],
        number=info["number"],
        level_type=info["level_type"],
        level=info["level"],
        step_h=info["step_h"],
        grid_template=info["grid_template"],
        packing=info["packing"],
        npoints=info["npoints"],
        start_step_h=info.get("start_step_h"),
        statistical_process=info.get("statistical_process"),
    )