Public Python API

Everything exported by baro.met. The first group is what most users need; the second is for writing a source of one's own or working on cubes directly.

met

Fast weather retrieval along aircraft trajectories.

ERA5 on model or pressure levels and forecast fields from Météo-France and Open-Meteo, interpolated with explicit coverage and no silent extrapolation. Surface queries accept fixed sites or trajectories without altitude.

Enricher

Interpolate a gridded source along trajectories.

Parameters:

Name	Type	Description	Default
source	Source	Gridded source, e.g. :func:baro.met.arco_model_levels or :class:baro.met.arpege.ArpegeSource.	required
cache_dir	str | Path | None	Optional on-disk cache of every byte read (immutable data only), shared between runs, calls and processes: a later call downloads only what no earlier call has read. It is attached to source (which then keeps using it, also through other enrichers).	None
mode	Mode	"flight" (default) reads only the region and levels the points need. "global" reads whole fields on every level (FastMeteo-like), so that one cache serves any later flight.	'flight'
domain	Domain | None	Fixed domain for every cube. By default the points of each time slice are split along the source's tiling, and each group gets the box of its own points.	None
margin_deg	float	Margin around the points when the domain is derived from them.	0.5
level_margin	int	Extra levels kept above and below the bracketing levels.	1
io_workers	int	Concurrent range requests, and cubes fetched ahead of interpolation.	16
vertical	VerticalMethod	"log" or "linear" in pressure; "actual" surface pressure or "standard" (101325 Pa, reproduces a fixed altitude->level table).	'log'
on_uncovered	Literal['warn', 'raise', 'flag']	What to do with points outside the period the source covers (e.g. before the start of an archive): "warn" (default) flags them OUT_OF_COVERAGE and emits a warning with the covered period, "raise" raises :class:DataUnavailableError, "flag" only flags.	'warn'

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

class Enricher:
    """Interpolate a gridded source along trajectories.

    Parameters
    ----------
    source:
        Gridded source, e.g. :func:`baro.met.arco_model_levels` or
        :class:`baro.met.arpege.ArpegeSource`.
    cache_dir:
        Optional on-disk cache of every byte read (immutable data only), shared
        between runs, calls and processes: a later call downloads only what no
        earlier call has read. It is attached to ``source`` (which then keeps
        using it, also through other enrichers).
    mode:
        ``"flight"`` (default) reads only the region and levels the points
        need. ``"global"`` reads whole fields on every level (FastMeteo-like),
        so that one cache serves any later flight.
    domain:
        Fixed domain for every cube. By default the points of each time slice
        are split along the source's tiling, and each group gets the box of
        its own points.
    margin_deg:
        Margin around the points when the domain is derived from them.
    level_margin:
        Extra levels kept above and below the bracketing levels.
    io_workers / prefetch:
        Concurrent range requests, and cubes fetched ahead of interpolation.
    vertical / surface_pressure:
        ``"log"`` or ``"linear"`` in pressure; ``"actual"`` surface pressure or
        ``"standard"`` (101325 Pa, reproduces a fixed altitude->level table).
    on_uncovered:
        What to do with points outside the period the source covers (e.g.
        before the start of an archive): ``"warn"`` (default) flags them
        ``OUT_OF_COVERAGE`` and emits a warning with the covered period,
        ``"raise"`` raises :class:`DataUnavailableError`, ``"flag"`` only flags.
    """

    def __init__(  # noqa: PLR0913 - keyword-only configuration of the public enricher
        self,
        source: Source,
        *,
        cache_dir: str | Path | None = None,
        mode: Mode = "flight",
        domain: Domain | None = None,
        margin_deg: float = 0.5,
        level_margin: int = 1,
        io_workers: int = 16,
        prefetch: int = 2,
        vertical: VerticalMethod = "log",
        surface_pressure: SurfacePressureMode = "actual",
        on_uncovered: Literal["warn", "raise", "flag"] = "warn",
    ) -> None:
        if on_uncovered not in ("warn", "raise", "flag"):
            raise ValueError(f"unknown on_uncovered {on_uncovered!r}")
        self.on_uncovered = on_uncovered
        if mode not in ("flight", "global"):
            raise ValueError(f"unknown mode {mode!r}")
        self.source = source
        self.cache = RangeCache(cache_dir) if cache_dir is not None else None
        if self.cache is not None:
            source.use_cache(self.cache)
        self.mode = mode
        self.domain = domain if domain is not None or mode == "flight" else Domain.whole_globe()
        self.margin_deg = margin_deg
        self.level_margin = _ALL_LEVELS if mode == "global" else level_margin
        self.io_workers = io_workers
        self.prefetch = max(prefetch, 1)
        self.vertical = vertical
        self.surface_pressure = surface_pressure
        self._lock = threading.Lock()

    # -- public API ----------------------------------------------------------------------
    def enrich(
        self,
        time_utc: ArrayLike,
        lat: ArrayLike,
        lon: ArrayLike,
        *,
        altitude_ft: ArrayLike | None = None,
        pressure_pa: ArrayLike | None = None,
    ) -> EnrichResult:
        """Fields at each point. Give barometric (pressure) altitude in ft or pressure in Pa."""
        if (altitude_ft is None) == (pressure_pa is None):
            raise ValueError("give exactly one of altitude_ft or pressure_pa")
        t = np.asarray(time_utc).astype("datetime64[ns]")
        la = np.asarray(lat, dtype=np.float64)
        lo = np.asarray(lon, dtype=np.float64)
        if pressure_pa is not None:
            p = np.asarray(pressure_pa, dtype=np.float64)
        else:
            p = pressure_from_altitude_ft(np.asarray(altitude_ft, dtype=np.float64))
        if not (t.shape == la.shape == lo.shape == p.shape) or t.ndim != 1:
            raise ValueError("inputs must be 1-D arrays of equal length")
        return self._run(t, la, lo, p)

    # -- internals -----------------------------------------------------------------------
    def _run(
        self,
        t: NDArray[np.datetime64],
        lat: NDArray[np.float64],
        lon: NDArray[np.float64],
        p: NDArray[np.float64],
    ) -> EnrichResult:
        started = time.perf_counter()
        before = [s.snapshot() for s in self.source.stats]
        n = t.size
        values = {name: np.full(n, np.nan) for name in self.source.fields}
        flags = np.full(n, Flag.INVALID_INPUT, dtype=np.uint8)
        idx = self._usable(t, lat, lon, p, flags)
        step = self.source.time_step
        t_ok = t[idx]
        t0 = t_ok - (t_ok - _EPOCH) % step
        w = (t_ok - t0) / step
        la, lo, pp = lat[idx], lon[idx], p[idx]
        points = _Points(w, la, lo, pp)
        intervals, requests = self._plan(t0, w, la, lo, pp)
        counters: dict[str, Any] = {
            "cubes": len(requests),
            "fetched": 0,
            "unavailable": 0,
            "cube_bytes": 0,
        }
        cached_before = self._cache_hits()
        with (
            ThreadPoolExecutor(self.io_workers, thread_name_prefix="traj-io") as io,
            ThreadPoolExecutor(self.prefetch, thread_name_prefix="traj-cube") as cubes,
        ):
            for iv, got in self._stream(intervals, requests, io, cubes, counters):
                for sel, r0, r1 in _pairs(iv):
                    cube1 = None if r1 < 0 else got[r1]
                    v, f = self._interpolate(got[r0], cube1, points.take(sel))
                    for name in values:
                        values[name][idx[sel]] = v[name]
                    flags[idx[sel]] = f
        after = [s.snapshot() for s in self.source.stats]
        for name in ("bytes", "requests", "throttled"):
            counters[name] = int(sum(a[name] - b[name] for a, b in zip(after, before, strict=True)))
        counters["cache_bytes"] = self._cache_hits() - cached_before
        counters["seconds"] = time.perf_counter() - started
        return EnrichResult(values, flags, counters)

    def _usable(
        self,
        t: NDArray[np.datetime64],
        lat: NDArray[np.float64],
        lon: NDArray[np.float64],
        p: NDArray[np.float64],
        flags: NDArray[np.uint8],
    ) -> NDArray[np.int64]:
        """Indices of the valid, covered points; flags the uncovered ones."""
        ok = ~np.isnat(t) & np.isfinite(lat) & np.isfinite(lon) & np.isfinite(p) & (p > 0)
        uncovered = self._check_coverage(t, ok)
        flags[uncovered] = Flag.OUT_OF_COVERAGE
        return np.flatnonzero(ok & ~uncovered)

    def _cache_hits(self) -> int:
        """Bytes served by the cache so far."""
        return 0 if self.cache is None else self.cache.stats.snapshot()["hit_bytes"]

    def _check_coverage(
        self, t: NDArray[np.datetime64], ok: NDArray[np.bool_]
    ) -> NDArray[np.bool_]:
        """Points whose bracketing interval falls outside the source's coverage."""
        coverage = self.source.coverage()
        uncovered = np.zeros(t.size, dtype=bool)
        if coverage is None or not ok.any():
            return uncovered
        start = np.datetime64(coverage[0], "ns")
        end = np.datetime64(coverage[1], "ns")
        uncovered[ok] = (t[ok] < start) | (t[ok] > end)
        n = int(uncovered.sum())
        if n == 0 or self.on_uncovered == "flag":
            return uncovered
        before = int((uncovered & (t < start)).sum())
        after = n - before
        span = f"{t[uncovered].min()} .. {t[uncovered].max()}"
        message = (
            f"{n} point(s) outside the period covered by {self.source.source_id} "
            f"({coverage[0]} .. {coverage[1]} UTC): {before} before, {after} after "
            f"(their times span {span}); they are flagged OUT_OF_COVERAGE"
        )
        if self.on_uncovered == "raise":
            raise DataUnavailableError(message)
        warnings.warn(message, UncoveredPointsWarning, skip_file_prefixes=(_PACKAGE_DIR,))
        return uncovered

    def _plan(
        self,
        t0: NDArray[np.datetime64],
        w: NDArray[np.float64],
        lat: NDArray[np.float64],
        lon: NDArray[np.float64],
        p: NDArray[np.float64],
    ) -> tuple[list[_Interval], list[_CubeRequest]]:
        """Intervals in time order, and the cube requests (one per occupied-tile group)."""
        groups = _group(t0)
        keys = {start: self.source.cube_keys(start) for start in groups}
        uses: dict[CubeKey, list[NDArray[np.int64]]] = {}
        for start, members in groups.items():
            k0, k1 = keys[start]
            uses.setdefault(k0, []).append(members)
            later = members[w[members] > 0]
            if later.size:
                uses.setdefault(k1, []).append(later)
        requests: list[_CubeRequest] = []
        lookup: dict[CubeKey, tuple[NDArray[np.int64], NDArray[np.int64]]] = {}
        for key, parts in uses.items():
            sel = np.sort(np.concatenate(parts))
            tiling = None if self.domain is not None else self.source.tiling(key)
            labels = partition(lat[sel], lon[sel], self.margin_deg, tiling)
            ids = np.empty(sel.size, dtype=np.int64)
            for label in range(int(labels.max()) + 1 if sel.size else 0):
                part = sel[labels == label]
                ids[labels == label] = len(requests)
                requests.append(self._request(key, lat[part], lon[part], p[part]))
            lookup[key] = (sel, ids)
        intervals = []
        for start, members in groups.items():
            k0, k1 = keys[start]
            req1 = np.full(members.size, -1, dtype=np.int64)
            needs_next = w[members] > 0
            if needs_next.any():
                req1[needs_next] = _request_ids(lookup[k1], members[needs_next])
            intervals.append(_Interval(start, members, _request_ids(lookup[k0], members), req1))
        return intervals, requests

    def _request(
        self,
        key: CubeKey,
        lat: NDArray[np.float64],
        lon: NDArray[np.float64],
        p: NDArray[np.float64],
    ) -> _CubeRequest:
        """Request for the cube ``key`` over the region and pressures of some points."""
        domain = self.domain or Domain.from_points(lat, lon, self.margin_deg)
        return _CubeRequest(key, domain, (float(p.min()), float(p.max())))

    def _stream(
        self,
        intervals: list[_Interval],
        requests: list[_CubeRequest],
        io: ThreadPoolExecutor,
        pool: ThreadPoolExecutor,
        counters: dict[str, Any],
    ) -> Iterator[tuple[_Interval, dict[int, Cube | None]]]:
        """Intervals with their cubes; cubes are fetched ahead and freed after last use."""

        def fetch(req: int) -> Future[Cube | None]:
            """Submit the download of one cube."""
            return pool.submit(self._download, requests[req], io, counters)

        ahead = _Prefetcher(intervals, fetch, self.prefetch)
        for iv in intervals:
            yield iv, ahead.cubes(iv)
            ahead.release(iv)

    def _download(
        self, req: _CubeRequest, io: ThreadPoolExecutor, counters: dict[str, Any]
    ) -> Cube | None:
        try:
            cube = self.source.fetch(
                req.key, req.domain, req.p_range, level_margin=self.level_margin, executor=io
            )
        except DataUnavailableError:
            self._count(counters, "unavailable")
            return None
        self._count(counters, "fetched")
        self._count(counters, "cube_bytes", cube.nbytes)
        return cube

    def _count(self, counters: dict[str, Any], name: str, amount: int = 1) -> None:
        with self._lock:
            counters[name] += amount

    def _interpolate(
        self,
        cube0: Cube | None,
        cube1: Cube | None,
        pts: _Points,
    ) -> tuple[dict[str, NDArray[np.float64]], NDArray[np.uint8]]:
        n = pts.w.size
        names = list(self.source.fields)
        if cube0 is None:
            return {k: np.full(n, np.nan) for k in names}, np.full(n, Flag.MISSING_DATA, np.uint8)
        opts = SampleOptions(self.vertical, self.surface_pressure)
        v0, flags = sample(cube0, pts.lat, pts.lon, pts.p, opts)
        later = pts.w > 0
        if not later.any():
            return v0, flags
        if cube1 is None:
            v1 = {k: np.full(int(later.sum()), np.nan) for k in names}
            f1 = np.full(int(later.sum()), Flag.MISSING_DATA, np.uint8)
        else:
            nxt = pts.take(later)
            v1, f1 = sample(cube1, nxt.lat, nxt.lon, nxt.p, opts)
        flags = flags.copy()
        flags[later] |= np.where(flags[later] == 0, f1, 0).astype(np.uint8)
        out = {}
        wl = pts.w[later]
        for name in names:
            blended = v0[name].copy()
            blended[later] = (1.0 - wl) * v0[name][later] + wl * v1[name]
            blended[flags != 0] = np.nan
            out[name] = blended
        return out, flags

enrich(time_utc, lat, lon, *, altitude_ft=None, pressure_pa=None)

Fields at each point. Give barometric (pressure) altitude in ft or pressure in Pa.

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

def enrich(
    self,
    time_utc: ArrayLike,
    lat: ArrayLike,
    lon: ArrayLike,
    *,
    altitude_ft: ArrayLike | None = None,
    pressure_pa: ArrayLike | None = None,
) -> EnrichResult:
    """Fields at each point. Give barometric (pressure) altitude in ft or pressure in Pa."""
    if (altitude_ft is None) == (pressure_pa is None):
        raise ValueError("give exactly one of altitude_ft or pressure_pa")
    t = np.asarray(time_utc).astype("datetime64[ns]")
    la = np.asarray(lat, dtype=np.float64)
    lo = np.asarray(lon, dtype=np.float64)
    if pressure_pa is not None:
        p = np.asarray(pressure_pa, dtype=np.float64)
    else:
        p = pressure_from_altitude_ft(np.asarray(altitude_ft, dtype=np.float64))
    if not (t.shape == la.shape == lo.shape == p.shape) or t.ndim != 1:
        raise ValueError("inputs must be 1-D arrays of equal length")
    return self._run(t, la, lo, p)

EnrichResult dataclass

Interpolated fields (NaN where flags != 0) and run statistics.

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

@dataclass
class EnrichResult:
    """Interpolated fields (NaN where ``flags != 0``) and run statistics."""

    fields: dict[str, NDArray[np.float64]]
    flags: NDArray[np.uint8]
    stats: dict[str, Any] = field(default_factory=dict)

    def __getitem__(self, name: str) -> NDArray[np.float64]:
        return self.fields[name]

    @property
    def valid(self) -> NDArray[np.bool_]:
        """Mask of the points with a value (no flag set)."""
        return np.asarray(self.flags == Flag.OK)

valid property

Mask of the points with a value (no flag set).

WeatherSession

Reuse decoded weather across successive point queries, without extrapolation.

source supplies a fixed set of fields. Optional fields selects a subset through its select_fields capability (currently GridSource); unsupported sources must be configured with their own field selection. vertical is required: choose "linear" or "log" in pressure.

The LRU budget bounds retained array payload, including coordinates, not Python overhead or temporary cubes held by the active enrichment call. Calls on one session are serialized; its normal source I/O stays parallel. Treat a session as a snapshot: call :meth:clear to refresh decoded mutable forecasts or ERA5T. The existing on-disk byte cache retains its own rules. Do not mutate the source configuration while using the session.

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

class WeatherSession:
    """Reuse decoded weather across successive point queries, without extrapolation.

    ``source`` supplies a fixed set of fields. Optional ``fields`` selects a
    subset through its ``select_fields`` capability (currently ``GridSource``);
    unsupported sources must be configured with their own field selection.
    ``vertical`` is required: choose ``"linear"`` or ``"log"`` in pressure.

    The LRU budget bounds retained array payload, including coordinates, not
    Python overhead or temporary cubes held by the active enrichment call.
    Calls on one session are serialized; its normal source I/O stays parallel.
    Treat a session as a snapshot: call :meth:`clear` to refresh decoded mutable
    forecasts or ERA5T. The existing on-disk byte cache retains its own rules.
    Do not mutate the source configuration while using the session.
    """

    def __init__(  # noqa: PLR0913 - explicit public sampling configuration
        self,
        source: Source,
        *,
        vertical: VerticalMethod,
        fields: Sequence[str] | None = None,
        max_cache_bytes: int = 256 * 1024 * 1024,
        cache_dir: str | Path | None = None,
        domain: Domain | None = None,
        margin_deg: float = 0.5,
        level_margin: int = 1,
        io_workers: int = 16,
        prefetch: int = 2,
        surface_pressure: SurfacePressureMode = "actual",
        on_uncovered: Literal["warn", "raise", "flag"] = "raise",
    ) -> None:
        if not isinstance(max_cache_bytes, int) or max_cache_bytes < 0:
            raise ValueError("max_cache_bytes must be a nonnegative integer")
        if vertical not in ("linear", "log"):
            raise ValueError("vertical must be 'linear' or 'log'")
        if fields is not None:
            select = getattr(source, "select_fields", None)
            if select is None:
                raise TypeError("source does not support select_fields; configure its fields first")
            source = select(fields)
        self._source = _DecodedSource(source, max_cache_bytes)
        self._lock = threading.Lock()
        self._enricher = Enricher(
            self._source,
            cache_dir=cache_dir,
            domain=domain,
            margin_deg=margin_deg,
            level_margin=level_margin,
            io_workers=io_workers,
            prefetch=prefetch,
            vertical=vertical,
            surface_pressure=surface_pressure,
            on_uncovered=on_uncovered,
        )

    def sample(
        self, time_utc: ArrayLike, lat: ArrayLike, lon: ArrayLike, *, pressure_pa: ArrayLike
    ) -> EnrichResult:
        """Sample 1-D points using nanosecond time and explicit pressure in Pa.

        Missing coverage retains Enricher flags; callers must check ``valid``
        before interpreting missing meteorology as a physical disappearance.
        Time interpolation divides integer timedelta differences, never floating
        epoch hours, preserving subsecond weights at modern dates.
        """
        with self._lock:
            before = self.cache_info()
            result = self._enricher.enrich(time_utc, lat, lon, pressure_pa=pressure_pa)
            after = self.cache_info()
            result.stats.update(
                decoded_cache_hits=after["hits"] - before["hits"],
                decoded_cache_misses=after["misses"] - before["misses"],
                decoded_cache_bytes=after["bytes"],
            )
            return result

    def cache_info(self) -> dict[str, int]:
        """Retained array bytes, entry count and lifetime LRU counters."""
        with self._source.lock:
            return {
                "bytes": self._source.bytes,
                "entries": len(self._source.entries),
                "hits": self._source.hits,
                "misses": self._source.misses,
                "evictions": self._source.evictions,
                "max_bytes": self._source.budget,
            }

    def clear(self) -> None:
        """Release decoded arrays after any active call; preserve lifetime counters."""
        with self._lock, self._source.lock:
            self._source.entries.clear()
            self._source.bytes = 0

cache_info()

Retained array bytes, entry count and lifetime LRU counters.

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

def cache_info(self) -> dict[str, int]:
    """Retained array bytes, entry count and lifetime LRU counters."""
    with self._source.lock:
        return {
            "bytes": self._source.bytes,
            "entries": len(self._source.entries),
            "hits": self._source.hits,
            "misses": self._source.misses,
            "evictions": self._source.evictions,
            "max_bytes": self._source.budget,
        }

clear()

Release decoded arrays after any active call; preserve lifetime counters.

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

def clear(self) -> None:
    """Release decoded arrays after any active call; preserve lifetime counters."""
    with self._lock, self._source.lock:
        self._source.entries.clear()
        self._source.bytes = 0

sample(time_utc, lat, lon, *, pressure_pa)

Sample 1-D points using nanosecond time and explicit pressure in Pa.

Missing coverage retains Enricher flags; callers must check valid before interpreting missing meteorology as a physical disappearance. Time interpolation divides integer timedelta differences, never floating epoch hours, preserving subsecond weights at modern dates.

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

def sample(
    self, time_utc: ArrayLike, lat: ArrayLike, lon: ArrayLike, *, pressure_pa: ArrayLike
) -> EnrichResult:
    """Sample 1-D points using nanosecond time and explicit pressure in Pa.

    Missing coverage retains Enricher flags; callers must check ``valid``
    before interpreting missing meteorology as a physical disappearance.
    Time interpolation divides integer timedelta differences, never floating
    epoch hours, preserving subsecond weights at modern dates.
    """
    with self._lock:
        before = self.cache_info()
        result = self._enricher.enrich(time_utc, lat, lon, pressure_pa=pressure_pa)
        after = self.cache_info()
        result.stats.update(
            decoded_cache_hits=after["hits"] - before["hits"],
            decoded_cache_misses=after["misses"] - before["misses"],
            decoded_cache_bytes=after["bytes"],
        )
        return result

WeatherWindowProvider

Fetch just the native weather windows requested by a forward model.

Fields are fixed at construction. The default includes CoCiP's optical depth; when absent from the source it is computed from the complete upper cirrus column by :func:materialize_weather, requiring optional pycontrails.

Returned containers are borrowed snapshots: their arrays are read-only; callers must not replace coordinates, fields or metadata. Engine exports make independent arrays. The byte budget counts retained array payload, excluding Python overhead, active downloads, exports and caller references. Calls are serialized. Use clear() to refresh decoded mutable data; the underlying immutable byte cache keeps its existing policy.

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

class WeatherWindowProvider:
    """Fetch just the native weather windows requested by a forward model.

    Fields are fixed at construction. The default includes CoCiP's optical
    depth; when absent from the source it is computed from the complete upper
    cirrus column by :func:`materialize_weather`, requiring optional pycontrails.

    Returned containers are borrowed snapshots: their arrays are read-only;
    callers must not replace coordinates, fields or metadata. Engine exports
    make independent arrays. The byte budget counts retained array payload,
    excluding Python overhead, active downloads, exports and caller references.
    Calls are serialized. Use ``clear()`` to refresh decoded mutable data;
    the underlying immutable byte cache keeps its existing policy.
    """

    def __init__(  # noqa: PLR0913 - explicit provider retrieval configuration
        self,
        source: GridSource,
        radiation: ArcoRadiationSource,
        *,
        fields: Sequence[str] = _DEFAULT_FIELDS,
        max_cache_bytes: int = 256 * 1024 * 1024,
        cache_dir: str | Path | None = None,
        margin_deg: float = 0.0,
        pressure_margin_pa: float = 0.0,
        prefetch: np.timedelta64 = _NO_PREFETCH,
    ) -> None:
        self._fields, self._derive_tau = _native_fields(source, fields)
        duration = _retrieval_options(max_cache_bytes, margin_deg, pressure_margin_pa, prefetch)
        # Materialization takes its own field views, keeping dependencies such
        # as ciwc available only for the separate upper-column calculation.
        self.source, self.radiation = source, radiation
        self.max_cache_bytes = max_cache_bytes
        self.margin_deg, self.pressure_margin_pa = margin_deg, pressure_margin_pa
        self.prefetch = duration
        self._cache: OrderedDict[int, tuple[_Window, int]] = OrderedDict()
        self._lock = threading.RLock()
        self._bytes = self._hits = self._misses = self._evictions = self._next = 0
        if cache_dir is not None:
            cache = RangeCache(cache_dir)
            source.use_cache(cache)
            radiation.use_cache(cache)

    def _expanded(self, request: WeatherRequest) -> WeatherRequest:
        d, m = request.domain, self.margin_deg
        width = d.lon_max - d.lon_min + 2 * m
        centre = (d.lon_min + d.lon_max) / 2
        lon_min, lon_max = (
            (d.lon_min - m, d.lon_max + m) if width < FULL_TURN else (centre - 180, centre + 180)
        )
        domain = Domain(max(-90, d.lat_min - m), min(90, d.lat_max + m), lon_min, lon_max)
        # Reject the actual demand before clipping speculative margins.
        native = self.source.vertical.a
        if not native[0] <= request.pressure_pa[0] <= request.pressure_pa[1] <= native[-1]:
            raise ValueError("weather demand exceeds source pressure coverage")
        lo = max(float(native[0]), request.pressure_pa[0] - self.pressure_margin_pa)
        hi = min(float(native[-1]), request.pressure_pa[1] + self.pressure_margin_pa)
        end_ns = int(request.end.astype(np.int64)) + int(self.prefetch.astype(np.int64))
        for bounds in (self.source.coverage(), self.radiation.coverage()):
            if bounds is not None:
                if not bounds[0] <= request.start <= request.end <= bounds[1]:
                    raise ValueError("weather demand exceeds source time coverage")
                end_ns = min(end_ns, int(np.datetime64(bounds[1], "ns").astype(np.int64)))
        if end_ns > np.iinfo(np.int64).max:
            raise ValueError("prefetch exceeds datetime64 nanosecond range")
        return WeatherRequest(domain, request.start, np.datetime64(end_ns, "ns"), (lo, hi))

    def get(self, request: WeatherRequest) -> _Window:
        """Return native met/radiation grids fully covering ``request``.

        Cache hits use actual coordinate coverage, including native bracketing
        neighbours. Misses use the usual targeted readers; no preadvection or
        whole archive retrieval is performed. Failures are never cached.
        """
        with self._lock:
            hit = next(
                (k for k in reversed(self._cache) if _contains(self._cache[k][0], request)), None
            )
            if hit is not None:
                self._cache.move_to_end(hit)
                self._hits += 1
                return self._cache[hit][0]
            self._misses += 1
            expanded = self._expanded(request)
            window = (
                materialize_weather(
                    self.source, expanded, fields=self._fields, tau_cirrus=self._derive_tau
                ),
                materialize_radiation(self.radiation, expanded),
            )
            if not _contains(window, request):
                raise ValueError("materialized weather does not cover the complete demand")
            size = _freeze(window)
            if size <= self.max_cache_bytes:
                while self._cache and self._bytes + size > self.max_cache_bytes:
                    self._bytes -= self._cache.popitem(last=False)[1][1]
                    self._evictions += 1
                self._cache[self._next] = (window, size)
                self._next += 1
                self._bytes += size
            return window

    def for_diffcocip(
        self, demand: WeatherDemand
    ) -> tuple[dict[str, NDArray[np.float64]], dict[str, NDArray[np.float64]]]:
        """Driver callback: lon/lat degrees, pressure hPa, seconds since 2000.

        Downloads run on the host outside JAX compilation/autodifferentiation.
        The driver remains responsible for complete evolution/lower-probe
        demands and for retrying a step when new support is required.
        """
        bounds = demand.bounds
        lon, lat, level, time = (bounds[k] for k in ("lon", "lat", "level", "time_s"))
        request = WeatherRequest(
            Domain(lat[0], lat[1], lon[0], lon[1]),
            _utc_ns(time[0], upper=False),
            _utc_ns(time[1], upper=True),
            (level[0] * 100, level[1] * 100),
        )
        weather, radiation = self.get(request)
        return weather.to_diffcocip(), radiation.to_diffcocip()

    def cache_info(self) -> dict[str, int]:
        """Retained payload, entries and lifetime hit/miss/eviction counters."""
        with self._lock:
            return {
                "bytes": self._bytes,
                "entries": len(self._cache),
                "hits": self._hits,
                "misses": self._misses,
                "evictions": self._evictions,
                "max_bytes": self.max_cache_bytes,
            }

    def clear(self) -> None:
        """Release decoded windows, retaining byte caches and lifetime counters."""
        with self._lock:
            self._cache.clear()
            self._bytes = 0

cache_info()

Retained payload, entries and lifetime hit/miss/eviction counters.

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

def cache_info(self) -> dict[str, int]:
    """Retained payload, entries and lifetime hit/miss/eviction counters."""
    with self._lock:
        return {
            "bytes": self._bytes,
            "entries": len(self._cache),
            "hits": self._hits,
            "misses": self._misses,
            "evictions": self._evictions,
            "max_bytes": self.max_cache_bytes,
        }

clear()

Release decoded windows, retaining byte caches and lifetime counters.

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

def clear(self) -> None:
    """Release decoded windows, retaining byte caches and lifetime counters."""
    with self._lock:
        self._cache.clear()
        self._bytes = 0

for_diffcocip(demand)

Driver callback: lon/lat degrees, pressure hPa, seconds since 2000.

Downloads run on the host outside JAX compilation/autodifferentiation. The driver remains responsible for complete evolution/lower-probe demands and for retrying a step when new support is required.

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

def for_diffcocip(
    self, demand: WeatherDemand
) -> tuple[dict[str, NDArray[np.float64]], dict[str, NDArray[np.float64]]]:
    """Driver callback: lon/lat degrees, pressure hPa, seconds since 2000.

    Downloads run on the host outside JAX compilation/autodifferentiation.
    The driver remains responsible for complete evolution/lower-probe
    demands and for retrying a step when new support is required.
    """
    bounds = demand.bounds
    lon, lat, level, time = (bounds[k] for k in ("lon", "lat", "level", "time_s"))
    request = WeatherRequest(
        Domain(lat[0], lat[1], lon[0], lon[1]),
        _utc_ns(time[0], upper=False),
        _utc_ns(time[1], upper=True),
        (level[0] * 100, level[1] * 100),
    )
    weather, radiation = self.get(request)
    return weather.to_diffcocip(), radiation.to_diffcocip()

get(request)

Return native met/radiation grids fully covering request.

Cache hits use actual coordinate coverage, including native bracketing neighbours. Misses use the usual targeted readers; no preadvection or whole archive retrieval is performed. Failures are never cached.

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

def get(self, request: WeatherRequest) -> _Window:
    """Return native met/radiation grids fully covering ``request``.

    Cache hits use actual coordinate coverage, including native bracketing
    neighbours. Misses use the usual targeted readers; no preadvection or
    whole archive retrieval is performed. Failures are never cached.
    """
    with self._lock:
        hit = next(
            (k for k in reversed(self._cache) if _contains(self._cache[k][0], request)), None
        )
        if hit is not None:
            self._cache.move_to_end(hit)
            self._hits += 1
            return self._cache[hit][0]
        self._misses += 1
        expanded = self._expanded(request)
        window = (
            materialize_weather(
                self.source, expanded, fields=self._fields, tau_cirrus=self._derive_tau
            ),
            materialize_radiation(self.radiation, expanded),
        )
        if not _contains(window, request):
            raise ValueError("materialized weather does not cover the complete demand")
        size = _freeze(window)
        if size <= self.max_cache_bytes:
            while self._cache and self._bytes + size > self.max_cache_bytes:
                self._bytes -= self._cache.popitem(last=False)[1][1]
                self._evictions += 1
            self._cache[self._next] = (window, size)
            self._next += 1
            self._bytes += size
        return window

ArcoRadiationSource

Hourly ERA5 flux averages, represented at interval midpoints in W/m².

ARCO stores one-hour energies (J/m²) at interval ends. This source divides them by 3600 and moves the interpolation coordinate back 30 minutes, exactly once. It preserves ERA5's signed net-flux convention. These are top-of-atmosphere fields, not surface irradiance. Only requested fields and occupied latitude blocks are read through the existing range reader.

grid_source permits local mirrors/tests; it must use the ERA5 ARCO time and spatial axes. The default opens the public pressure-level store.

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

class ArcoRadiationSource:
    """Hourly ERA5 flux averages, represented at interval midpoints in W/m².

    ARCO stores one-hour energies (J/m²) at interval ends. This source divides
    them by 3600 and moves the interpolation coordinate back 30 minutes,
    exactly once. It preserves ERA5's signed net-flux convention. These are
    top-of-atmosphere fields, not surface irradiance. Only requested fields
    and occupied latitude blocks are read through the existing range reader.

    ``grid_source`` permits local mirrors/tests; it must use the ERA5 ARCO
    time and spatial axes. The default opens the public pressure-level store.
    """

    time_step = np.timedelta64(1, "h")
    time_origin = np.datetime64("1970-01-01T00:30", "ns")

    def __init__(
        self,
        *,
        fields: Sequence[str] = _FIELDS,
        base_url: str = ARCO_BASE_URL,
        grid_source: GridSource | None = None,
    ) -> None:
        names = tuple(fields)
        if not names or len(set(names)) != len(names) or set(names) - set(_FIELDS):
            raise ValueError("select unique ERA5 top net solar/thermal radiation fields")
        self.grid = grid_source or arco_pressure_levels(
            base_url=base_url, fields={name: name for name in names}
        )
        if any(self.grid.fields.get(name) != name for name in names):
            raise ValueError("grid_source must be configured with the requested radiation arrays")
        self.grid = self.grid.select_fields(names)
        # TOA is a radiative boundary, not a finite height above ground.
        self.variables = {name: SurfaceVariable("W m-2", float("nan")) for name in names}
        self._scale = {name: _flux_scale(self.grid, name) for name in names}

    @property
    def stats(self) -> list[TransferStats]:
        """Underlying transfer counters, including byte-cache hits."""
        return self.grid.stats

    def coverage(self) -> tuple[np.datetime64, np.datetime64]:
        """Effective interpolation coverage after the ERA5 midpoint shift."""
        bounds = self.grid.coverage()
        if bounds is None:
            raise DataUnavailableError("ERA5 radiation coverage is unavailable")
        return bounds[0] - _HALF_HOUR, bounds[1] - _HALF_HOUR

    def cube_keys(self, t0: np.datetime64) -> tuple[CubeKey, CubeKey]:
        """Midpoint samples bracketing the actual query time."""
        end = np.datetime64(t0 + _HALF_HOUR, "h")
        return CubeKey(end - _HALF_HOUR), CubeKey(end + _HALF_HOUR)

    def tiling(self, key: CubeKey) -> Tiling | None:
        """Block-sized latitude bands shared with the range reader."""
        return self.grid.tiling(CubeKey(key.valid + _HALF_HOUR))

    def use_cache(self, cache: RangeCache) -> None:
        """Cache only immutable ERA5 archive ranges."""
        self.grid.use_cache(cache)

    def fetch(
        self, key: CubeKey, domain: Domain, *, fields: Sequence[str] | None = None
    ) -> SurfaceCube:
        """Read a midpoint field without downloading unrelated pressure variables."""
        names = tuple(self.variables if fields is None else fields)
        if set(names) - self.variables.keys():
            raise ValueError("requested radiation field is not configured")
        end = key.valid + _HALF_HOUR
        if end != np.datetime64(end, "h") or key.run is not None:
            raise ValueError("ERA5 radiation keys must be hourly interval midpoints, without run")
        try:
            index = self.grid.time_index(end)
        except KeyError as exc:
            raise DataUnavailableError(f"ERA5 radiation: {end} outside archive") from exc
        grid = GridIndex.locate(domain, self.grid.lat, self.grid.lon)
        values = {
            name: (
                self.grid.store.read_rows(name, [(index,)], grid.rows)[0][:, grid.cols][::-1]
                * self._scale[name]
            ).astype(np.float32)
            for name in names
        }
        return SurfaceCube(
            key, grid.lat[::-1], grid.lon, values, {n: self.variables[n] for n in names}
        )

stats property

Underlying transfer counters, including byte-cache hits.

coverage()

Effective interpolation coverage after the ERA5 midpoint shift.

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

def coverage(self) -> tuple[np.datetime64, np.datetime64]:
    """Effective interpolation coverage after the ERA5 midpoint shift."""
    bounds = self.grid.coverage()
    if bounds is None:
        raise DataUnavailableError("ERA5 radiation coverage is unavailable")
    return bounds[0] - _HALF_HOUR, bounds[1] - _HALF_HOUR

cube_keys(t0)

Midpoint samples bracketing the actual query time.

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

def cube_keys(self, t0: np.datetime64) -> tuple[CubeKey, CubeKey]:
    """Midpoint samples bracketing the actual query time."""
    end = np.datetime64(t0 + _HALF_HOUR, "h")
    return CubeKey(end - _HALF_HOUR), CubeKey(end + _HALF_HOUR)

fetch(key, domain, *, fields=None)

Read a midpoint field without downloading unrelated pressure variables.

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

def fetch(
    self, key: CubeKey, domain: Domain, *, fields: Sequence[str] | None = None
) -> SurfaceCube:
    """Read a midpoint field without downloading unrelated pressure variables."""
    names = tuple(self.variables if fields is None else fields)
    if set(names) - self.variables.keys():
        raise ValueError("requested radiation field is not configured")
    end = key.valid + _HALF_HOUR
    if end != np.datetime64(end, "h") or key.run is not None:
        raise ValueError("ERA5 radiation keys must be hourly interval midpoints, without run")
    try:
        index = self.grid.time_index(end)
    except KeyError as exc:
        raise DataUnavailableError(f"ERA5 radiation: {end} outside archive") from exc
    grid = GridIndex.locate(domain, self.grid.lat, self.grid.lon)
    values = {
        name: (
            self.grid.store.read_rows(name, [(index,)], grid.rows)[0][:, grid.cols][::-1]
            * self._scale[name]
        ).astype(np.float32)
        for name in names
    }
    return SurfaceCube(
        key, grid.lat[::-1], grid.lon, values, {n: self.variables[n] for n in names}
    )

tiling(key)

Block-sized latitude bands shared with the range reader.

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

def tiling(self, key: CubeKey) -> Tiling | None:
    """Block-sized latitude bands shared with the range reader."""
    return self.grid.tiling(CubeKey(key.valid + _HALF_HOUR))

use_cache(cache)

Cache only immutable ERA5 archive ranges.

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

def use_cache(self, cache: RangeCache) -> None:
    """Cache only immutable ERA5 archive ranges."""
    self.grid.use_cache(cache)

EvolutionResult dataclass

Forward result; age_limit_reached is conservative, never extinction.

Diagnostics contain only executed step ends; trailing extinct steps from the dense scan are omitted. energy_forcing_j sums the same hard EF.

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

@dataclass(frozen=True)
class EvolutionResult:
    """Forward result; ``age_limit_reached`` is conservative, never extinction.

    Diagnostics contain only executed step ends; trailing extinct steps from
    the dense scan are omitted. ``energy_forcing_j`` sums the same hard EF.
    """

    energy_forcing_j: float
    stop_reason: str
    formation: Mapping[str, Any]
    final_state: Mapping[str, Any]
    diagnostics: Mapping[str, Any]
    times: npt.NDArray[np.datetime64]
    weather_requests: int

ArpegeSource

ARPEGE isobaric fields.

Parameters:

Name	Type	Description	Default
resolution	Resolution	"01" (Europe) or "025" (global).	'01'
fields	dict[str, str] | None	Output name -> ARPEGE short name (default u, v, t).	None
levels	Sequence[int] | None	Isobaric levels (hPa) allowed; default: all published levels.	None
run	RunPolicy | datetime64 | datetime	"latest": most recent run (00/06/12/18 UTC) not after the interval, falling back to older runs if a file is not published; "daily": the 00 UTC run of the same day (FastMeteo's behaviour); or a fixed run time. Both cubes bracketing an interval always come from the same run.	'latest'
base_url	str	Where files live (HTTP base URL, local mirror directory, or a store).	ARPEGE_BASE_URL
index_dir	str | Path | None	Directory caching the per-file message indexes (strongly recommended; defaults to the enricher's cache when it has one).	None

Notes

The files of a published run never change: with a cache (:meth:use_cache) every GRIB message read is kept on disk and never downloaded again.

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

class ArpegeSource:
    """ARPEGE isobaric fields.

    Parameters
    ----------
    resolution:
        ``"01"`` (Europe) or ``"025"`` (global).
    fields:
        Output name -> ARPEGE short name (default ``u``, ``v``, ``t``).
    levels:
        Isobaric levels (hPa) allowed; default: all published levels.
    run:
        ``"latest"``: most recent run (00/06/12/18 UTC) not after the interval,
        falling back to older runs if a file is not published; ``"daily"``:
        the 00 UTC run of the same day (FastMeteo's behaviour); or a fixed run
        time. Both cubes bracketing an interval always come from the same run.
    base_url / store:
        Where files live (HTTP base URL, local mirror directory, or a store).
    index_dir:
        Directory caching the per-file message indexes (strongly recommended;
        defaults to the enricher's cache when it has one).

    Notes
    -----
    The files of a published run never change: with a cache (:meth:`use_cache`)
    every GRIB message read is kept on disk and never downloaded again.
    """

    def __init__(  # noqa: PLR0913 - keyword-only configuration of a public source
        self,
        *,
        resolution: Resolution = "01",
        package: str = "IP1",
        fields: dict[str, str] | None = None,
        levels: Sequence[int] | None = None,
        run: RunPolicy | np.datetime64 | datetime = "latest",
        run_hours: Sequence[int] = (0, 6, 12, 18),
        max_run_fallback: int = 4,
        base_url: str = ARPEGE_BASE_URL,
        store: ByteStore | None = None,
        index_dir: str | Path | None = None,
        listing_url: str | None = ARPEGE_LISTING_URL,
    ) -> None:
        if resolution not in _LEAD_RANGES:
            raise ValueError(f"unknown resolution {resolution!r}")
        self.resolution = resolution
        self.package = package
        self.fields = _checked_fields(fields)
        self.levels = None if levels is None else sorted(int(x) for x in levels)
        self.run = run if isinstance(run, str) else np.datetime64(run, "h")
        self.run_hours = sorted(run_hours)
        self.max_run_fallback = max_run_fallback
        self.time_step = _HOUR  # hourly for isobaric levels >= 100 hPa on both grids
        is_local = "://" not in base_url
        self.store = store or (FileStore(base_url) if is_local else HttpStore(base_url))
        self._messages: ByteStore = self.store  # message reads (through the cache, if any)
        self.source_id = f"arpege-{resolution}-{package}"
        self.indexes = GribIndexStore(
            self.store,
            None if index_dir is None else Path(index_dir).expanduser() / self.source_id,
        )
        self.listing_url = listing_url
        self._local_root = Path(base_url) if is_local else None
        self._coverage: tuple[float, tuple[np.datetime64, np.datetime64] | None] | None = None
        self._exists: Lru[str, tuple[bool, float]] = Lru(4096)  # published?, checked at
        self._lock = threading.Lock()

    @property
    def stats(self) -> list[TransferStats]:
        """Transfer counters of the underlying store."""
        return [self.store.stats]

    def tiling(self, key: CubeKey) -> Tiling | None:
        """``None``: a GRIB message holds a whole field, read in one piece."""
        return None

    def use_cache(self, cache: RangeCache) -> None:
        """Keep every message read in ``cache``; message indexes go there too by default."""
        self._messages = CachedStore(
            self.store, cache, namespace_for(self.store.location), lambda _key: True
        )
        if self.indexes.directory is None:
            self.indexes.directory = cache.root / "grib-index" / self.source_id

    # -- coverage ------------------------------------------------------------------------------
    def published_runs(self) -> list[np.datetime64]:
        """Runs currently published (rolling ~14-day window), oldest first."""
        if self._local_root is not None:
            names = [p.name for p in self._local_root.iterdir() if p.is_dir()]
        elif self.listing_url is not None:
            listed = list_s3(self.listing_url, "pnt/", dirs=True, stats=self.store.stats)
            names = [name.removeprefix("pnt/").rstrip("/") for name in listed]
        else:
            return []
        runs = sorted({np.datetime64(m[1], "h") for n in names if (m := _RUN_PREFIX.fullmatch(n))})
        return [r for r in runs if int(str(r)[11:13]) in self.run_hours]

    def coverage(self, max_age_s: float = 600.0) -> tuple[np.datetime64, np.datetime64] | None:
        """From the oldest published run to the last lead time of the newest one."""
        now = time.monotonic()
        if self._coverage is None or now - self._coverage[0] > max_age_s:
            try:
                runs = self.published_runs()
            except httpx.HTTPError:
                runs = []
            cov = (runs[0], runs[-1] + np.timedelta64(_MAX_LEAD_H, "h")) if runs else None
            self._coverage = (now, cov)
        return self._coverage[1]

    def serves(self, t0: np.datetime64) -> bool:
        """True if both ends of the interval starting at ``t0`` are published."""
        t0h = np.datetime64(t0, "h")
        for run in self._candidate_runs(t0h):
            leads = [int((t - run) / _HOUR) for t in (t0h, t0h + self.time_step)]
            try:
                if all(self._published(self.file_key(run, lead)) for lead in leads):
                    return True
            except DataUnavailableError:
                continue
        return False

    # -- files and runs ------------------------------------------------------------------------
    def file_key(self, run: np.datetime64, lead_h: int) -> str:
        """Path of the GRIB2 file holding ``lead_h`` hours of ``run``."""
        for a, b in _LEAD_RANGES[self.resolution]:
            if a <= lead_h <= b:
                break
        else:
            raise DataUnavailableError(f"lead time {lead_h} h beyond the published range")
        stamp = str(np.datetime64(run, "h")) + ":00:00Z"
        res, pkg = self.resolution, self.package
        name = f"arpege__{res}__{pkg}__{a:03d}H{b:03d}H__{stamp}.grib2"
        return f"{stamp}/arpege/{res}/{pkg}/{name}"

    def _published(self, key: str) -> bool:
        """Whether ``key`` exists; a "not yet" is re-checked after a while (runs keep coming)."""
        with self._lock:
            known = self._exists.get(key)
        if known is not None and (known[0] or time.monotonic() - known[1] < _RECHECK_S):
            return known[0]
        try:
            self.store.size(key)
            ok = True
        except MissingChunkError:
            ok = False
        with self._lock:
            self._exists[key] = (ok, time.monotonic())
        return ok

    def _candidate_runs(self, t0: np.datetime64) -> list[np.datetime64]:
        if not isinstance(self.run, str):
            return [self.run]
        day = np.datetime64(t0, "D")
        if self.run == "daily":
            return [np.datetime64(day, "h")]
        runs = []
        for d in (day, day - np.timedelta64(1, "D")):
            runs += [
                np.datetime64(d, "h") + np.timedelta64(h, "h") for h in reversed(self.run_hours)
            ]
        runs = [r for r in runs if r <= t0]
        return runs[: self.max_run_fallback + 1]

    def run_for(self, t0: np.datetime64) -> np.datetime64:
        """Run whose files hold both ends of the interval starting at ``t0``."""
        t0h = np.datetime64(t0, "h")
        candidates = self._candidate_runs(t0h)
        for run in candidates:
            leads = [int((t - run) / _HOUR) for t in (t0h, t0h + self.time_step)]
            try:
                if all(self._published(self.file_key(run, lead)) for lead in leads):
                    return run
            except DataUnavailableError:
                continue
        return candidates[0]  # nothing published: fetch will report the data as unavailable

    def cube_keys(self, t0: np.datetime64) -> tuple[CubeKey, CubeKey]:
        """Cubes bracketing ``[t0, t0 + 1 h)``, both from the run chosen by :meth:`run_for`."""
        t0h = np.datetime64(t0, "h")
        run = self.run_for(t0h)
        return CubeKey(t0h, run), CubeKey(t0h + self.time_step, run)

    # -- cubes ---------------------------------------------------------------------------------
    def fetch(
        self,
        key: CubeKey,
        domain: Domain,
        p_range: tuple[float, float],
        *,
        level_margin: int = 1,
        executor: Executor | None = None,
    ) -> Cube:
        """One validity time of one run over ``domain``, levels bracketing ``p_range`` (Pa)."""
        if key.run is None:
            raise ValueError("ARPEGE cubes need a run")
        lead_h = int((key.valid - key.run) / _HOUR)
        file = self.file_key(key.run, lead_h)
        try:
            idx = self.indexes.get(file, executor)
        except MissingChunkError as exc:
            raise DataUnavailableError(f"{self.source_id}: {file} not published") from exc
        vertical = self._vertical(idx, float(lead_h))
        positions = vertical.select_bracketing(
            p_range[0],
            p_range[1],
            STANDARD_SURFACE_PRESSURE,
            STANDARD_SURFACE_PRESSURE,
            level_margin,
        )
        wanted = []
        for name in self.fields.values():
            for hpa in vertical.ids[positions]:
                msg = idx.find(ARPEGE_PARAMS[name], _ISOBARIC, float(hpa) * 100.0, float(lead_h))
                if msg is None:
                    raise DataUnavailableError(f"{file}: no {name} at {hpa} hPa, step {lead_h} h")
                wanted.append(msg)
        buffers = fetch_messages(self._messages, file, wanted, executor)
        decoded = [decode_message(b) for b in buffers]
        _, lat, lon = decoded[0]
        grid = GridIndex.locate(domain, lat, lon)
        nlev = len(positions)
        fields = {}
        for i, out_name in enumerate(self.fields):
            layers = [decoded[i * nlev + k][0][grid.rows][:, grid.cols] for k in range(nlev)]
            fields[out_name] = np.stack(layers).astype(np.float32)
        return Cube(
            source=self.source_id,
            hour=key.valid,
            lat=grid.lat,
            lon=grid.lon,
            vertical=vertical.subset(positions),
            fields=fields,
            run=key.run,
        )

    def _vertical(self, idx: GribIndex, step_h: float) -> VerticalCoordinate:
        """Isobaric levels published for every field at this step.

        Stratospheric levels (1-70 hPa) of the 0.25 deg grid, and every level
        beyond 48 h, are only 3-hourly: they are simply absent at other steps.
        """
        common: set[int] | None = None
        for name in self.fields.values():
            lv = {round(x / 100.0) for x in idx.levels(ARPEGE_PARAMS[name], _ISOBARIC, step_h)}
            common = lv if common is None else common & lv
        levels = sorted(common or set())
        if self.levels is not None:
            levels = [x for x in levels if x in set(self.levels)]
        if len(levels) < MIN_LEVELS:
            raise DataUnavailableError(f"{idx.key}: fewer than 2 common isobaric levels")
        return pressure_levels(levels)

stats property

Transfer counters of the underlying store.

coverage(max_age_s=600.0)

From the oldest published run to the last lead time of the newest one.

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

def coverage(self, max_age_s: float = 600.0) -> tuple[np.datetime64, np.datetime64] | None:
    """From the oldest published run to the last lead time of the newest one."""
    now = time.monotonic()
    if self._coverage is None or now - self._coverage[0] > max_age_s:
        try:
            runs = self.published_runs()
        except httpx.HTTPError:
            runs = []
        cov = (runs[0], runs[-1] + np.timedelta64(_MAX_LEAD_H, "h")) if runs else None
        self._coverage = (now, cov)
    return self._coverage[1]

cube_keys(t0)

Cubes bracketing [t0, t0 + 1 h), both from the run chosen by :meth:run_for.

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

def cube_keys(self, t0: np.datetime64) -> tuple[CubeKey, CubeKey]:
    """Cubes bracketing ``[t0, t0 + 1 h)``, both from the run chosen by :meth:`run_for`."""
    t0h = np.datetime64(t0, "h")
    run = self.run_for(t0h)
    return CubeKey(t0h, run), CubeKey(t0h + self.time_step, run)

fetch(key, domain, p_range, *, level_margin=1, executor=None)

One validity time of one run over domain, levels bracketing p_range (Pa).

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

def fetch(
    self,
    key: CubeKey,
    domain: Domain,
    p_range: tuple[float, float],
    *,
    level_margin: int = 1,
    executor: Executor | None = None,
) -> Cube:
    """One validity time of one run over ``domain``, levels bracketing ``p_range`` (Pa)."""
    if key.run is None:
        raise ValueError("ARPEGE cubes need a run")
    lead_h = int((key.valid - key.run) / _HOUR)
    file = self.file_key(key.run, lead_h)
    try:
        idx = self.indexes.get(file, executor)
    except MissingChunkError as exc:
        raise DataUnavailableError(f"{self.source_id}: {file} not published") from exc
    vertical = self._vertical(idx, float(lead_h))
    positions = vertical.select_bracketing(
        p_range[0],
        p_range[1],
        STANDARD_SURFACE_PRESSURE,
        STANDARD_SURFACE_PRESSURE,
        level_margin,
    )
    wanted = []
    for name in self.fields.values():
        for hpa in vertical.ids[positions]:
            msg = idx.find(ARPEGE_PARAMS[name], _ISOBARIC, float(hpa) * 100.0, float(lead_h))
            if msg is None:
                raise DataUnavailableError(f"{file}: no {name} at {hpa} hPa, step {lead_h} h")
            wanted.append(msg)
    buffers = fetch_messages(self._messages, file, wanted, executor)
    decoded = [decode_message(b) for b in buffers]
    _, lat, lon = decoded[0]
    grid = GridIndex.locate(domain, lat, lon)
    nlev = len(positions)
    fields = {}
    for i, out_name in enumerate(self.fields):
        layers = [decoded[i * nlev + k][0][grid.rows][:, grid.cols] for k in range(nlev)]
        fields[out_name] = np.stack(layers).astype(np.float32)
    return Cube(
        source=self.source_id,
        hour=key.valid,
        lat=grid.lat,
        lon=grid.lon,
        vertical=vertical.subset(positions),
        fields=fields,
        run=key.run,
    )

file_key(run, lead_h)

Path of the GRIB2 file holding lead_h hours of run.

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

def file_key(self, run: np.datetime64, lead_h: int) -> str:
    """Path of the GRIB2 file holding ``lead_h`` hours of ``run``."""
    for a, b in _LEAD_RANGES[self.resolution]:
        if a <= lead_h <= b:
            break
    else:
        raise DataUnavailableError(f"lead time {lead_h} h beyond the published range")
    stamp = str(np.datetime64(run, "h")) + ":00:00Z"
    res, pkg = self.resolution, self.package
    name = f"arpege__{res}__{pkg}__{a:03d}H{b:03d}H__{stamp}.grib2"
    return f"{stamp}/arpege/{res}/{pkg}/{name}"

published_runs()

Runs currently published (rolling ~14-day window), oldest first.

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

def published_runs(self) -> list[np.datetime64]:
    """Runs currently published (rolling ~14-day window), oldest first."""
    if self._local_root is not None:
        names = [p.name for p in self._local_root.iterdir() if p.is_dir()]
    elif self.listing_url is not None:
        listed = list_s3(self.listing_url, "pnt/", dirs=True, stats=self.store.stats)
        names = [name.removeprefix("pnt/").rstrip("/") for name in listed]
    else:
        return []
    runs = sorted({np.datetime64(m[1], "h") for n in names if (m := _RUN_PREFIX.fullmatch(n))})
    return [r for r in runs if int(str(r)[11:13]) in self.run_hours]

run_for(t0)

Run whose files hold both ends of the interval starting at t0.

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

def run_for(self, t0: np.datetime64) -> np.datetime64:
    """Run whose files hold both ends of the interval starting at ``t0``."""
    t0h = np.datetime64(t0, "h")
    candidates = self._candidate_runs(t0h)
    for run in candidates:
        leads = [int((t - run) / _HOUR) for t in (t0h, t0h + self.time_step)]
        try:
            if all(self._published(self.file_key(run, lead)) for lead in leads):
                return run
        except DataUnavailableError:
            continue
    return candidates[0]  # nothing published: fetch will report the data as unavailable

serves(t0)

True if both ends of the interval starting at t0 are published.

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

def serves(self, t0: np.datetime64) -> bool:
    """True if both ends of the interval starting at ``t0`` are published."""
    t0h = np.datetime64(t0, "h")
    for run in self._candidate_runs(t0h):
        leads = [int((t - run) / _HOUR) for t in (t0h, t0h + self.time_step)]
        try:
            if all(self._published(self.file_key(run, lead)) for lead in leads):
                return True
        except DataUnavailableError:
            continue
    return False

tiling(key)

None: a GRIB message holds a whole field, read in one piece.

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

def tiling(self, key: CubeKey) -> Tiling | None:
    """``None``: a GRIB message holds a whole field, read in one piece."""
    return None

use_cache(cache)

Keep every message read in cache; message indexes go there too by default.

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

def use_cache(self, cache: RangeCache) -> None:
    """Keep every message read in ``cache``; message indexes go there too by default."""
    self._messages = CachedStore(
        self.store, cache, namespace_for(self.store.location), lambda _key: True
    )
    if self.indexes.directory is None:
        self.indexes.directory = cache.root / "grib-index" / self.source_id

OpenMeteoSource

Bases: _ArchiveSource

Isobaric fields of one Open-Meteo archived model.

This is a stitched forecast archive, not an ERA5-like reanalysis. Original run identities and forecast lead times are not retained (CubeKey.run and CubeKey.lead are None). ecmwf_ifs025 provides 3-hourly data; the enricher interpolates linearly between native valid times.

Parameters:

Name	Type	Description	Default
model	str	Open-Meteo model folder, e.g. "meteofrance_arpege_europe" (0.1 deg, from 2023-12-01) or "meteofrance_arpege_world025" (0.25 deg, from 2023-12-22).	'meteofrance_arpege_europe'
fields	dict[str, str] | None	Output name -> key of :data:OPENMETEO_VARIABLES (default u, v, t).	None
levels	Sequence[int] | None	Isobaric levels (hPa) allowed; default: all archived levels.	None
base_url	str	The public bucket, or a local mirror directory with the same layout.	OPENMETEO_BASE_URL
tile_deg	float	Size of the geographic tiles read and cached (degrees).	5.0
cache_mb	float	Memory budget of the tile LRU.	1024.0
merge_adjacent_tiles	bool	Group touching tiles into one request (default True). Set False for sparse trajectories on fine grids: each occupied tile gets its own domain and pressure range, with the enricher's interpolation margin.	True

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

class OpenMeteoSource(_ArchiveSource):
    """Isobaric fields of one Open-Meteo archived model.

    This is a stitched forecast archive, not an ERA5-like reanalysis. Original
    run identities and forecast lead times are not retained (``CubeKey.run``
    and ``CubeKey.lead`` are ``None``). ``ecmwf_ifs025`` provides 3-hourly data;
    the enricher interpolates linearly between native valid times.

    Parameters
    ----------
    model:
        Open-Meteo model folder, e.g. ``"meteofrance_arpege_europe"`` (0.1 deg,
        from 2023-12-01) or ``"meteofrance_arpege_world025"`` (0.25 deg, from
        2023-12-22).
    fields:
        Output name -> key of :data:`OPENMETEO_VARIABLES` (default u, v, t).
    levels:
        Isobaric levels (hPa) allowed; default: all archived levels.
    base_url:
        The public bucket, or a local mirror directory with the same layout.
    tile_deg:
        Size of the geographic tiles read and cached (degrees).
    cache_mb:
        Memory budget of the tile LRU.
    merge_adjacent_tiles:
        Group touching tiles into one request (default True). Set False for
        sparse trajectories on fine grids: each occupied tile gets its own
        domain and pressure range, with the enricher's interpolation margin.
    """

    def __init__(  # noqa: PLR0913 - keyword-only configuration of a public source
        self,
        model: str = "meteofrance_arpege_europe",
        *,
        fields: dict[str, str] | None = None,
        levels: Sequence[int] | None = None,
        base_url: str = OPENMETEO_BASE_URL,
        tile_deg: float = 5.0,
        cache_mb: float = 1024.0,
        merge_adjacent_tiles: bool = True,
    ) -> None:
        self.model = model
        self.merge_adjacent_tiles = merge_adjacent_tiles
        self.fields = dict(fields or {"u": "u", "v": "v", "t": "t"})
        unknown = set(self.fields.values()) - set(OPENMETEO_VARIABLES)
        if unknown:
            raise ValueError(f"unknown variables {sorted(unknown)}")
        self.source_id = f"openmeteo-{model}"
        self._stats = TransferStats()
        variables = [OPENMETEO_VARIABLES[name][0] for name in self.fields.values()]
        self.catalog = ArchiveCatalog(base_url, model, variables, self._stats)
        self.time_step = self.catalog.time_step
        self.levels = self.catalog.common_levels(levels)
        self.reader = TileReader(self.catalog, self._stats, tile_deg, cache_mb)
        self.reader.final = self._final_path
        self._probe = (variables[0], self.levels[-1])  # a file that always exists

    def fetch(
        self,
        key: CubeKey,
        domain: Domain,
        p_range: tuple[float, float],
        *,
        level_margin: int = 1,
        executor: Executor | None = None,
    ) -> Cube:
        """One native time step over ``domain`` with the levels bracketing ``p_range`` (Pa)."""
        start, end = self.coverage()
        if not start <= key.valid <= end:
            raise DataUnavailableError(f"{self.source_id}: {key.valid} outside {start}..{end}")
        index = self.catalog.time_index(key.valid)
        if key.valid != np.datetime64(index * self.catalog.step_seconds, "s"):
            raise ValueError(f"{key.valid} is not aligned to {self.time_step}")
        chunk, t_in = divmod(index, self.catalog.chunk_steps)
        vertical = pressure_levels(self.levels)
        positions = vertical.select_bracketing(
            p_range[0],
            p_range[1],
            STANDARD_SURFACE_PRESSURE,
            STANDARD_SURFACE_PRESSURE,
            level_margin,
        )
        nlat, _ = self.reader.ensure_shape(*self._probe)
        lat, lon = self.reader.grid()
        grid = GridIndex.locate(domain, lat, lon)
        rows = nlat - 1 - np.arange(grid.rows.start, grid.rows.stop)  # archive rows, lat ascending
        hpas = [int(h) for h in vertical.ids[positions]]
        wanted = [(OPENMETEO_VARIABLES[n][0], h) for n in self.fields.values() for h in hpas]
        blocks = self.reader.blocks(wanted, chunk, rows, grid.cols)
        tp = self.reader.tile_points()
        fields: dict[str, NDArray[np.float32]] = {}
        for out_name, name in self.fields.items():
            variable, offset = OPENMETEO_VARIABLES[name]
            layers = [
                gather_tiles(blocks[(variable, h)], rows, grid.cols, tp, t_in) + offset
                for h in hpas
            ]
            fields[out_name] = np.stack(layers).astype(np.float32)
        return Cube(
            source=self.source_id,
            hour=key.valid,
            lat=grid.lat,
            lon=grid.lon,
            vertical=vertical.subset(positions),
            fields=fields,
        )

fetch(key, domain, p_range, *, level_margin=1, executor=None)

One native time step over domain with the levels bracketing p_range (Pa).

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

def fetch(
    self,
    key: CubeKey,
    domain: Domain,
    p_range: tuple[float, float],
    *,
    level_margin: int = 1,
    executor: Executor | None = None,
) -> Cube:
    """One native time step over ``domain`` with the levels bracketing ``p_range`` (Pa)."""
    start, end = self.coverage()
    if not start <= key.valid <= end:
        raise DataUnavailableError(f"{self.source_id}: {key.valid} outside {start}..{end}")
    index = self.catalog.time_index(key.valid)
    if key.valid != np.datetime64(index * self.catalog.step_seconds, "s"):
        raise ValueError(f"{key.valid} is not aligned to {self.time_step}")
    chunk, t_in = divmod(index, self.catalog.chunk_steps)
    vertical = pressure_levels(self.levels)
    positions = vertical.select_bracketing(
        p_range[0],
        p_range[1],
        STANDARD_SURFACE_PRESSURE,
        STANDARD_SURFACE_PRESSURE,
        level_margin,
    )
    nlat, _ = self.reader.ensure_shape(*self._probe)
    lat, lon = self.reader.grid()
    grid = GridIndex.locate(domain, lat, lon)
    rows = nlat - 1 - np.arange(grid.rows.start, grid.rows.stop)  # archive rows, lat ascending
    hpas = [int(h) for h in vertical.ids[positions]]
    wanted = [(OPENMETEO_VARIABLES[n][0], h) for n in self.fields.values() for h in hpas]
    blocks = self.reader.blocks(wanted, chunk, rows, grid.cols)
    tp = self.reader.tile_points()
    fields: dict[str, NDArray[np.float32]] = {}
    for out_name, name in self.fields.items():
        variable, offset = OPENMETEO_VARIABLES[name]
        layers = [
            gather_tiles(blocks[(variable, h)], rows, grid.cols, tp, t_in) + offset
            for h in hpas
        ]
        fields[out_name] = np.stack(layers).astype(np.float32)
    return Cube(
        source=self.source_id,
        hour=key.valid,
        lat=grid.lat,
        lon=grid.lon,
        vertical=vertical.subset(positions),
        fields=fields,
    )

OpenMeteoSurfaceSource

Bases: _ArchiveSource

Native surface fields of an Open-Meteo archived model.

Parameters:

Name	Type	Description	Default
model	str	Native folder, default meteofrance_arome_france0025 (regional 2.5 km).	'meteofrance_arome_france0025'
fields	dict[str, str] | None	Output name -> key in OPENMETEO_SURFACE_VARIABLES. Default u10/v10/t2m, precipitation, rain and gusts. Only these variables/dependencies are read.	None
base_url	str	Public bucket or local mirror. Archive forecasts extend to metadata coverage.	OPENMETEO_BASE_URL
tile_deg	float	Geographic tile edge in degrees (default 0.25 for sparse surface queries).	0.25
merge_adjacent_tiles	bool	Merge touching occupied tiles into one request (default False, preserving sparse point reads). Tile caches still reuse neighbouring interpolation cells.	False
cache_mb	float	Tile LRU memory budget in megabytes.	1024.0

Notes

Rain is max(precipitation - snowfall_water_equivalent, 0) at each native grid point. Both archived amounts are already totals over the preceding native period; never difference them again. Gusts are the preceding-period maximum. Fetch returns native fields; SurfaceEnricher preserves these periods. Heights are above model terrain; there is no terrain/bias correction. Run/lead times are unavailable for this stitched archive.

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

class OpenMeteoSurfaceSource(_ArchiveSource):
    """Native surface fields of an Open-Meteo archived model.

    Parameters
    ----------
    model:
        Native folder, default ``meteofrance_arome_france0025`` (regional 2.5 km).
    fields:
        Output name -> key in OPENMETEO_SURFACE_VARIABLES. Default u10/v10/t2m,
        precipitation, rain and gusts. Only these variables/dependencies are read.
    base_url:
        Public bucket or local mirror. Archive forecasts extend to metadata coverage.
    tile_deg:
        Geographic tile edge in degrees (default 0.25 for sparse surface queries).
    merge_adjacent_tiles:
        Merge touching occupied tiles into one request (default False, preserving
        sparse point reads). Tile caches still reuse neighbouring interpolation cells.
    cache_mb:
        Tile LRU memory budget in megabytes.

    Notes
    -----
    Rain is ``max(precipitation - snowfall_water_equivalent, 0)`` at each native
    grid point. Both archived amounts are already totals over the preceding
    native period; never difference them again. Gusts are the preceding-period
    maximum. Fetch returns native fields; SurfaceEnricher preserves these periods.
    Heights are above model terrain; there is no terrain/bias correction.
    Run/lead times are unavailable for this stitched archive.
    """

    def __init__(  # noqa: PLR0913 - keyword-only archive configuration
        self,
        model: str = "meteofrance_arome_france0025",
        *,
        fields: dict[str, str] | None = None,
        base_url: str = OPENMETEO_BASE_URL,
        tile_deg: float = 0.25,
        merge_adjacent_tiles: bool = False,
        cache_mb: float = 1024.0,
    ) -> None:
        self.model = model
        self.merge_adjacent_tiles = merge_adjacent_tiles
        self.fields = _field_mapping(fields)
        _validate_storage(tile_deg, cache_mb)
        self.variables = {
            out: OPENMETEO_SURFACE_VARIABLES[name][2] for out, name in self.fields.items()
        }
        self._native = _native_variables(self.fields)
        self.source_id = f"openmeteo-surface-{model}"
        self._stats = TransferStats()
        self.catalog = ArchiveCatalog(base_url, model, self._native, self._stats)
        # Folder availability, not pressure levels, defines surface support.
        available = self.catalog.available_variables
        missing = set(self._native) - set(available)
        if missing:
            raise ValueError(f"{model}: surface variables not archived: {sorted(missing)}")
        self.time_step = self.catalog.time_step
        self.reader = TileReader(self.catalog, self._stats, tile_deg, cache_mb)
        self.reader.final = self._final_path
        self._probe = (self._native[0], 0)  # level zero selects the surface folder

    def coverage(self) -> tuple[np.datetime64, np.datetime64]:
        """Common native endpoint coverage (oldest chunks may be partly empty)."""
        first = max(self.catalog.locate_first_chunk(v, 0) for v in self._native)
        return np.datetime64(first * self.catalog.chunk_seconds, "s"), self.catalog.end

    def fetch(
        self, key: CubeKey, domain: Domain, *, fields: Sequence[str] | None = None
    ) -> SurfaceCube:
        """Read one native timestamp over the requested domain, without pressure."""
        selected = self.fields if fields is None else {n: self.fields[n] for n in fields}
        if not selected:
            raise ValueError("at least one surface field is required")
        native_names = _native_variables(selected)
        index = self._validate_key(key)
        chunk, t_in = divmod(index, self.catalog.chunk_steps)
        nlat, _ = self.reader.ensure_shape(*self._probe)
        lat, lon = self.reader.grid()
        grid = GridIndex.locate(domain, lat, lon)
        rows = nlat - 1 - np.arange(grid.rows.start, grid.rows.stop)
        blocks = self.reader.blocks([(v, 0) for v in native_names], chunk, rows, grid.cols)
        tp = self.reader.tile_points()
        native = {v: gather_tiles(blocks[(v, 0)], rows, grid.cols, tp, t_in) for v in native_names}
        values = _output_values(native, selected)
        return SurfaceCube(
            key,
            grid.lat[::-1].copy(),
            grid.lon,
            {n: a[::-1].copy() for n, a in values.items()},
            {n: self.variables[n] for n in selected},
        )

    def _validate_key(self, key: CubeKey) -> int:
        start, end = self.coverage()
        if not start <= key.valid <= end:
            raise DataUnavailableError(f"{self.source_id}: {key.valid} outside {start}..{end}")
        index = self.catalog.time_index(key.valid)
        if key.valid != np.datetime64(index * self.catalog.step_seconds, "s"):
            raise ValueError(f"{key.valid} is not aligned to {self.time_step}")
        if key.run is not None:
            raise ValueError("stitched archives do not retain forecast runs")
        return index

    def enrich(
        self, time: ArrayLike, lat: ArrayLike, lon: ArrayLike, *, cache: RangeCache | None = None
    ) -> SurfaceResult:
        """Convenience query; use SurfaceEnricher for a persistent cache directory."""
        if cache is not None:
            self.use_cache(cache)
        return SurfaceEnricher(self).enrich(time, lat, lon)

coverage()

Common native endpoint coverage (oldest chunks may be partly empty).

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

def coverage(self) -> tuple[np.datetime64, np.datetime64]:
    """Common native endpoint coverage (oldest chunks may be partly empty)."""
    first = max(self.catalog.locate_first_chunk(v, 0) for v in self._native)
    return np.datetime64(first * self.catalog.chunk_seconds, "s"), self.catalog.end

enrich(time, lat, lon, *, cache=None)

Convenience query; use SurfaceEnricher for a persistent cache directory.

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

def enrich(
    self, time: ArrayLike, lat: ArrayLike, lon: ArrayLike, *, cache: RangeCache | None = None
) -> SurfaceResult:
    """Convenience query; use SurfaceEnricher for a persistent cache directory."""
    if cache is not None:
        self.use_cache(cache)
    return SurfaceEnricher(self).enrich(time, lat, lon)

fetch(key, domain, *, fields=None)

Read one native timestamp over the requested domain, without pressure.

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

def fetch(
    self, key: CubeKey, domain: Domain, *, fields: Sequence[str] | None = None
) -> SurfaceCube:
    """Read one native timestamp over the requested domain, without pressure."""
    selected = self.fields if fields is None else {n: self.fields[n] for n in fields}
    if not selected:
        raise ValueError("at least one surface field is required")
    native_names = _native_variables(selected)
    index = self._validate_key(key)
    chunk, t_in = divmod(index, self.catalog.chunk_steps)
    nlat, _ = self.reader.ensure_shape(*self._probe)
    lat, lon = self.reader.grid()
    grid = GridIndex.locate(domain, lat, lon)
    rows = nlat - 1 - np.arange(grid.rows.start, grid.rows.stop)
    blocks = self.reader.blocks([(v, 0) for v in native_names], chunk, rows, grid.cols)
    tp = self.reader.tile_points()
    native = {v: gather_tiles(blocks[(v, 0)], rows, grid.cols, tp, t_in) for v in native_names}
    values = _output_values(native, selected)
    return SurfaceCube(
        key,
        grid.lat[::-1].copy(),
        grid.lon,
        {n: a[::-1].copy() for n, a in values.items()},
        {n: self.variables[n] for n in selected},
    )

AromeSurfaceSource

Hourly surface fields from one explicit AROME initialization (0..51 h).

fields maps output names to u10, v10, t2m, precipitation or gusts. Precipitation is a preceding-hour amount, derived from cumulative totals within the same run. Gusts retain the reported preceding-hour maximum. Only the requested GRIB messages are downloaded; their immutable ranges and per-file indexes use the shared range cache. No altitude is required. published_runs describes files visible now, never historical availability.

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

class AromeSurfaceSource:
    """Hourly surface fields from one explicit AROME initialization (0..51 h).

    ``fields`` maps output names to u10, v10, t2m, precipitation or gusts.
    Precipitation is a preceding-hour amount, derived from cumulative totals
    within the same run. Gusts retain the reported preceding-hour maximum.
    Only the requested GRIB messages are downloaded; their immutable ranges
    and per-file indexes use the shared range cache. No altitude is required.
    ``published_runs`` describes files visible now, never historical availability.
    """

    def __init__(
        self,
        *,
        run: np.datetime64 | datetime,
        fields: dict[str, str] | None = None,
        base_url: str = "https://files.data.gouv.fr/meteofrance-pnt/pnt",
        index_dir: str | Path | None = None,
    ) -> None:
        stamp = np.datetime64(run, "ns")
        hour = stamp.astype("datetime64[h]")
        if np.isnat(stamp) or stamp != hour or int(str(hour)[11:13]) % 3:
            raise ValueError("AROME run must be an exact UTC three-hour initialization")
        self._forecast = ArpegeSource(
            run=hour, run_hours=range(0, 24, 3), base_url=base_url, index_dir=index_dir
        )
        self.run = hour
        self.observed_at: np.datetime64 | None = None
        self.time_step = np.timedelta64(1, "h")
        self.indexes = self._forecast.indexes
        self.fields = _checked_fields(fields)
        self.variables = {out: _VARIABLES[name] for out, name in self.fields.items()}
        self.source_id = "arome-0025-surface"
        self._forecast.source_id = self.source_id
        self._reader = _NativeFields(self._forecast, self.run, self.file_key)
        if index_dir is not None:
            self.indexes.directory = Path(index_dir).expanduser() / self.source_id

    def __enter__(self) -> AromeSurfaceSource:
        return self

    def __exit__(self, *exc: object) -> None:
        self.close()

    @classmethod
    def latest(cls, valid_time: np.datetime64, **kwargs: Any) -> AromeSurfaceSource:
        """Newest run whose required files are available now for ``valid_time``.

        ``observed_at`` records when publication was checked. It is not a
        historical publication timestamp and cannot establish a past cutoff.
        """
        if np.isnat(valid_time):
            raise ValueError("valid time cannot be NaT")
        probe = cls(run=np.datetime64("2000-01-01T00"), **kwargs)
        try:
            run = probe._latest_available_run(valid_time)
            result = cls(run=run, **kwargs)
            result.observed_at = np.datetime64("now", "s")
            return result
        finally:
            probe.close()

    def _latest_available_run(self, valid_time: np.datetime64) -> np.datetime64:
        """Check the required native files of candidate initializations."""
        for run in reversed(self.published_runs()):
            leads = _required_leads(valid_time, run, self.fields)
            if leads and all(self._forecast._published(self.file_key(run, lead)) for lead in leads):
                return run
        raise DataUnavailableError("no currently published AROME run covers this valid time")

    @property
    def stats(self) -> list[TransferStats]:
        return self._forecast.stats

    def close(self) -> None:
        close = getattr(self._forecast.store, "close", None)
        if close is not None:
            close()

    def tiling(self, key: CubeKey) -> Tiling | None:
        return None

    def use_cache(self, cache: RangeCache) -> None:
        self._forecast.use_cache(cache)

    def published_runs(self) -> list[np.datetime64]:
        return self._forecast.published_runs()

    def coverage(self, max_age_s: float = 600) -> tuple[np.datetime64, np.datetime64]:
        return self.run, self.run + np.timedelta64(51, "h")

    def cube_keys(self, t0: np.datetime64) -> tuple[CubeKey, CubeKey]:
        hour = np.datetime64(t0, "h")
        return CubeKey(hour, self.run), CubeKey(hour + self.time_step, self.run)

    def file_key(self, run: np.datetime64, lead_h: int) -> str:
        for start, end in _RANGES:
            if start <= lead_h <= end:
                stamp = str(np.datetime64(run, "h")) + ":00:00Z"
                name = f"arome__0025__SP1__{start:02d}H{end:02d}H__{stamp}.grib2"
                return f"{stamp}/arome/0025/SP1/{name}"
        raise DataUnavailableError(f"AROME lead {lead_h} h outside 0..51")

    def fetch(
        self, key: CubeKey, domain: Domain, *, fields: Sequence[str] | None = None
    ) -> SurfaceCube:
        lead = _forecast_lead(key, self.run)
        self.file_key(self.run, lead)
        selected = self.fields if fields is None else {out: self.fields[out] for out in fields}
        if not selected:
            raise ValueError("at least one surface field is required")
        arrays = {name: self._reader.field(name, lead) for name in set(selected.values())}
        _, lat, lon = next(iter(arrays.values()))
        for _, field_lat, field_lon in arrays.values():
            _require_same_grid((lat, lon), (field_lat, field_lon))
        # The shared GRIB decoder emits descending latitude.
        grid = GridIndex.locate(domain, lat, lon)
        cube_fields = {
            out: arrays[name][0][grid.rows][:, grid.cols][::-1].copy()
            for out, name in selected.items()
        }
        return SurfaceCube(
            key,
            grid.lat[::-1].copy(),
            grid.lon,
            cube_fields,
            {out: self.variables[out] for out in selected},
        )

latest(valid_time, **kwargs) classmethod

Newest run whose required files are available now for valid_time.

observed_at records when publication was checked. It is not a historical publication timestamp and cannot establish a past cutoff.

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

@classmethod
def latest(cls, valid_time: np.datetime64, **kwargs: Any) -> AromeSurfaceSource:
    """Newest run whose required files are available now for ``valid_time``.

    ``observed_at`` records when publication was checked. It is not a
    historical publication timestamp and cannot establish a past cutoff.
    """
    if np.isnat(valid_time):
        raise ValueError("valid time cannot be NaT")
    probe = cls(run=np.datetime64("2000-01-01T00"), **kwargs)
    try:
        run = probe._latest_available_run(valid_time)
        result = cls(run=run, **kwargs)
        result.observed_at = np.datetime64("now", "s")
        return result
    finally:
        probe.close()

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

SurfaceEnricher

Retrieve surface weather at fixed sites or trajectories in one batch.

Points sharing a native interval are grouped by occupied geographic tiles; only requested fields and their interpolation neighbours are read. Supply cache_dir for persistent byte-range caching. Inputs are UTC datetime64 (Unix nanoseconds are accepted), latitude and longitude arrays of equal length. No altitude or pressure argument is needed.

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

class SurfaceEnricher:
    """Retrieve surface weather at fixed sites or trajectories in one batch.

    Points sharing a native interval are grouped by occupied geographic tiles;
    only requested fields and their interpolation neighbours are read. Supply
    ``cache_dir`` for persistent byte-range caching. Inputs are UTC datetime64
    (Unix nanoseconds are accepted), latitude and longitude arrays of equal
    length. No altitude or pressure argument is needed.
    """

    def __init__(
        self,
        source: SurfaceSource,
        *,
        cache_dir: str | Path | None = None,
        wind_components: tuple[str, str] | None = ("u10", "v10"),
    ) -> None:
        self.source = source
        self.wind_components = wind_components
        if cache_dir is not None:
            source.use_cache(RangeCache(cache_dir))

    def enrich(self, time: ArrayLike, lat: ArrayLike, lon: ArrayLike) -> SurfaceResult:
        """Interpolate instants; preserve the enclosing native totals and maxima."""
        times, la, lo = _inputs(time, lat, lon)
        result = self._empty(times)
        valid = ~np.isnat(times) & np.isfinite(la) & np.isfinite(lo) & (abs(la) <= LAT_LIMIT)
        result.flags[~valid] = Flag.INVALID_INPUT
        if times.size == 0:
            self._wind(result, times)
            result.stats = {"bytes": 0}
            return result
        start, end = self.source.coverage()
        covered = (times >= start) & (times <= end)
        result.flags[valid & ~covered] = Flag.OUT_OF_COVERAGE
        points = np.flatnonzero(valid & covered)
        step_ns = int(self.source.time_step / np.timedelta64(1, "ns"))
        # Some interval averages are interpolated at their midpoints rather
        # than on epoch-aligned hours (ERA5 radiation, for example).
        origin = np.datetime64(getattr(self.source, "time_origin", "1970-01-01"), "ns")
        origin_ns = int(origin.astype(np.int64))
        groups = (times[points].astype(np.int64) - origin_ns) // step_ns
        before = sum(s.bytes for s in self.source.stats)
        for group in np.unique(groups):
            selected = points[groups == group]
            t0 = np.datetime64(int(group) * step_ns + origin_ns, "ns")
            self._interval(result, (times, la, lo), selected, t0)
        self._wind(result, times)
        result.stats = {"bytes": sum(s.bytes for s in self.source.stats) - before}
        return result

    def _interval(
        self,
        result: SurfaceResult,
        coordinates: _Coordinates,
        selected: NDArray[np.int64],
        t0: np.datetime64,
    ) -> None:
        times, la, lo = coordinates
        try:
            keys = self.source.cube_keys(t0)
            tiling = self.source.tiling(keys[0])
        except DataUnavailableError:
            result.flags[selected] = Flag.MISSING_DATA
            return
        if keys[0].run != keys[1].run:
            raise ValueError("surface interpolation requires both endpoints from the same run")
        labels = partition(la[selected], lo[selected], 0.0, tiling)
        for label in np.unique(labels):
            sel = selected[labels == label]
            exact = times[sel] == keys[0].valid
            for mask in (exact, ~exact):
                if mask.any():
                    self._group(result, coordinates, sel[mask], keys)

    def _wind(self, result: SurfaceResult, times: NDArray[np.datetime64]) -> None:
        if self.wind_components is None or not all(
            n in result.fields for n in self.wind_components
        ):
            return
        u, v = (result.fields[n] for n in self.wind_components)
        speed = np.hypot(u, v)
        direction = np.mod(np.degrees(np.arctan2(-u, -v)), 360.0)
        direction[speed == 0] = np.nan  # wind-from direction is undefined in calm air
        for name, value, unit in (
            ("wind_speed", speed, "m/s"),
            ("wind_direction", direction, "degree"),
        ):
            if name in result.fields:
                raise ValueError(f"derived field {name} conflicts with a source field")
            result.fields[name] = value
            result.variables[name] = SurfaceVariable(
                unit, result.variables[self.wind_components[0]].height_m
            )
            result.interval_start[name] = times.copy()
            result.interval_end[name] = times.copy()

    def _empty(self, times: NDArray[np.datetime64]) -> SurfaceResult:
        variables = dict(self.source.variables)
        n = len(times)
        return SurfaceResult(
            {name: np.full(n, np.nan) for name in variables},
            np.zeros(n, dtype=np.uint8),
            variables,
            {name: np.full(n, np.datetime64("NaT", "ns")) for name in variables},
            {name: np.full(n, np.datetime64("NaT", "ns")) for name in variables},
            np.full(n, np.datetime64("NaT", "ns")),
            np.full(n, np.timedelta64("NaT", "ns")),
        )

    def _group(
        self,
        result: SurfaceResult,
        coordinates: _Coordinates,
        sel: NDArray[np.int64],
        keys: tuple[CubeKey, CubeKey],
    ) -> None:
        times, lat, lon = coordinates
        domain = Domain.from_points(lat[sel], lon[sel], margin=0.0)
        w = np.asarray((times[sel] - keys[0].valid) / (keys[1].valid - keys[0].valid), dtype=float)
        try:
            sampled = self._samples(keys, domain, (lat[sel], lon[sel]), w)
        except DataUnavailableError:
            result.flags[sel] |= np.uint8(Flag.MISSING_DATA)
            return
        for name, variable in result.variables.items():
            if variable.temporal == "instant":
                value, flags = _instant(sampled, name, w)
                begin = finish = times[sel]
            else:
                value, flags = _period(sampled, name, w)
                finish = np.where(w == 0, keys[0].valid, keys[1].valid).astype("datetime64[ns]")
                begin = finish - self.source.time_step
            result.fields[name][sel] = value
            result.flags[sel] |= flags
            result.interval_start[name][sel], result.interval_end[name][sel] = begin, finish
        key = keys[0]
        if key.run is not None:
            result.run[sel] = key.run
            result.lead[sel] = times[sel] - key.run
        for value in result.fields.values():
            value[sel[result.flags[sel] != 0]] = np.nan

    def _samples(
        self,
        keys: tuple[CubeKey, CubeKey],
        domain: Domain,
        positions: tuple[NDArray[np.float64], NDArray[np.float64]],
        weight: NDArray[np.float64],
    ) -> _Samples:
        """Fetch only endpoints and fields that contribute to this point group."""
        variables = self.source.variables
        exact = np.all(weight == 0)
        left = tuple(n for n, v in variables.items() if exact or v.temporal == "instant")
        endpoints = {}
        if left:
            endpoints[0] = left
        if np.any(weight != 0):
            endpoints[1] = tuple(variables)
        sampled = {}
        for which, names in endpoints.items():
            cube = self.source.fetch(keys[which], domain, fields=names)
            sampled[which] = cube.sample(*positions, fields=names)
        return sampled

enrich(time, lat, lon)

Interpolate instants; preserve the enclosing native totals and maxima.

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

def enrich(self, time: ArrayLike, lat: ArrayLike, lon: ArrayLike) -> SurfaceResult:
    """Interpolate instants; preserve the enclosing native totals and maxima."""
    times, la, lo = _inputs(time, lat, lon)
    result = self._empty(times)
    valid = ~np.isnat(times) & np.isfinite(la) & np.isfinite(lo) & (abs(la) <= LAT_LIMIT)
    result.flags[~valid] = Flag.INVALID_INPUT
    if times.size == 0:
        self._wind(result, times)
        result.stats = {"bytes": 0}
        return result
    start, end = self.source.coverage()
    covered = (times >= start) & (times <= end)
    result.flags[valid & ~covered] = Flag.OUT_OF_COVERAGE
    points = np.flatnonzero(valid & covered)
    step_ns = int(self.source.time_step / np.timedelta64(1, "ns"))
    # Some interval averages are interpolated at their midpoints rather
    # than on epoch-aligned hours (ERA5 radiation, for example).
    origin = np.datetime64(getattr(self.source, "time_origin", "1970-01-01"), "ns")
    origin_ns = int(origin.astype(np.int64))
    groups = (times[points].astype(np.int64) - origin_ns) // step_ns
    before = sum(s.bytes for s in self.source.stats)
    for group in np.unique(groups):
        selected = points[groups == group]
        t0 = np.datetime64(int(group) * step_ns + origin_ns, "ns")
        self._interval(result, (times, la, lo), selected, t0)
    self._wind(result, times)
    result.stats = {"bytes": sum(s.bytes for s in self.source.stats) - before}
    return result

SurfaceResult dataclass

Surface query values and native support/provenance for every output.

Period fields retain full native interval totals/maxima even at fractional query times. interval_start and interval_end are per-field arrays; instant fields have both equal to query time. run is NaT and lead NaT where a stitched archive has discarded the original run identity.

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

@dataclass
class SurfaceResult:
    """Surface query values and native support/provenance for every output.

    Period fields retain full native interval totals/maxima even at fractional
    query times. ``interval_start`` and ``interval_end`` are per-field arrays;
    instant fields have both equal to query time. ``run`` is NaT and ``lead``
    NaT where a stitched archive has discarded the original run identity.
    """

    fields: dict[str, NDArray[np.float64]]
    flags: NDArray[np.uint8]
    variables: dict[str, SurfaceVariable]
    interval_start: dict[str, NDArray[np.datetime64]]
    interval_end: dict[str, NDArray[np.datetime64]]
    run: NDArray[np.datetime64]
    lead: NDArray[np.timedelta64]
    stats: dict[str, Any] = field(default_factory=dict)

    def __getitem__(self, name: str) -> NDArray[np.float64]:
        return self.fields[name]

    @property
    def valid(self) -> NDArray[np.bool_]:
        """Points with every requested field available."""
        return np.asarray(self.flags == Flag.OK)

valid property

Points with every requested field available.

SourceChain

Sources of the same fields, tried in order for each interval.

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

class SourceChain:
    """Sources of the same fields, tried in order for each interval."""

    def __init__(self, *sources: Source) -> None:
        if not sources:
            raise ValueError("a chain needs at least one source")
        first = sources[0]
        for src in sources[1:]:
            if list(src.fields) != list(first.fields):
                raise ValueError("chained sources must provide the same output fields")
            if src.time_step != first.time_step:
                raise ValueError("chained sources must share their time step")
        ids = [s.source_id for s in sources]
        if len(set(ids)) != len(ids):
            raise ValueError("chained sources must have distinct ids")
        self.sources = sources
        self.fields = dict(first.fields)
        self.time_step = first.time_step
        self.source_id = "+".join(ids)
        self._by_id = {s.source_id: s for s in sources}

    @property
    def stats(self) -> list[TransferStats]:
        """Transfer counters of every chained source."""
        return [st for s in self.sources for st in s.stats]

    def coverage(self) -> tuple[np.datetime64, np.datetime64] | None:
        """Union of the chained sources' periods."""
        spans = [c for s in self.sources if (c := s.coverage()) is not None]
        if not spans:
            return None
        return min(c[0] for c in spans), max(c[1] for c in spans)

    def _serves(self, source: Source, t0: np.datetime64) -> bool:
        cov = source.coverage()
        t1 = t0 + self.time_step
        if cov is not None and not (cov[0] <= t0 and t1 <= cov[1]):
            return False
        check = getattr(source, "serves", None)
        return bool(check(t0)) if callable(check) else True

    def cube_keys(self, t0: np.datetime64) -> tuple[CubeKey, CubeKey]:
        """Keys of the first source serving ``[t0, t0 + step)``, tagged with its id."""
        t0h = np.datetime64(t0, "h")
        chosen = next((s for s in self.sources if self._serves(s, t0h)), self.sources[-1])
        k0, k1 = chosen.cube_keys(t0h)
        return replace(k0, source=chosen.source_id), replace(k1, source=chosen.source_id)

    def tiling(self, key: CubeKey) -> Tiling | None:
        """Tiling of the source the key was routed to."""
        routed = self._by_id.get(key.source or "")
        return None if routed is None else routed.tiling(key)

    def use_cache(self, cache: RangeCache) -> None:
        """Share ``cache`` with every chained source."""
        for source in self.sources:
            source.use_cache(cache)

    def fetch(
        self,
        key: CubeKey,
        domain: Domain,
        p_range: tuple[float, float],
        *,
        level_margin: int = 1,
        executor: Executor | None = None,
    ) -> Cube:
        """Fetch from the source the key was routed to."""
        if key.source is None or key.source not in self._by_id:
            raise DataUnavailableError(f"cube key {key} is not routed to a chained source")
        return self._by_id[key.source].fetch(
            key, domain, p_range, level_margin=level_margin, executor=executor
        )

stats property

Transfer counters of every chained source.

coverage()

Union of the chained sources' periods.

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

def coverage(self) -> tuple[np.datetime64, np.datetime64] | None:
    """Union of the chained sources' periods."""
    spans = [c for s in self.sources if (c := s.coverage()) is not None]
    if not spans:
        return None
    return min(c[0] for c in spans), max(c[1] for c in spans)

cube_keys(t0)

Keys of the first source serving [t0, t0 + step), tagged with its id.

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

def cube_keys(self, t0: np.datetime64) -> tuple[CubeKey, CubeKey]:
    """Keys of the first source serving ``[t0, t0 + step)``, tagged with its id."""
    t0h = np.datetime64(t0, "h")
    chosen = next((s for s in self.sources if self._serves(s, t0h)), self.sources[-1])
    k0, k1 = chosen.cube_keys(t0h)
    return replace(k0, source=chosen.source_id), replace(k1, source=chosen.source_id)

fetch(key, domain, p_range, *, level_margin=1, executor=None)

Fetch from the source the key was routed to.

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

def fetch(
    self,
    key: CubeKey,
    domain: Domain,
    p_range: tuple[float, float],
    *,
    level_margin: int = 1,
    executor: Executor | None = None,
) -> Cube:
    """Fetch from the source the key was routed to."""
    if key.source is None or key.source not in self._by_id:
        raise DataUnavailableError(f"cube key {key} is not routed to a chained source")
    return self._by_id[key.source].fetch(
        key, domain, p_range, level_margin=level_margin, executor=executor
    )

tiling(key)

Tiling of the source the key was routed to.

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

def tiling(self, key: CubeKey) -> Tiling | None:
    """Tiling of the source the key was routed to."""
    routed = self._by_id.get(key.source or "")
    return None if routed is None else routed.tiling(key)

use_cache(cache)

Share cache with every chained source.

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

def use_cache(self, cache: RangeCache) -> None:
    """Share ``cache`` with every chained source."""
    for source in self.sources:
        source.use_cache(cache)

Domain dataclass

Latitude/longitude box; longitudes may be expressed beyond 180 (no wrap inside).

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

@dataclass(frozen=True)
class Domain:
    """Latitude/longitude box; longitudes may be expressed beyond 180 (no wrap inside)."""

    lat_min: float
    lat_max: float
    lon_min: float
    lon_max: float

    def __post_init__(self) -> None:
        if not (-LAT_LIMIT <= self.lat_min <= self.lat_max <= LAT_LIMIT):
            raise ValueError(f"invalid latitude range {self.lat_min}..{self.lat_max}")
        if not (0.0 <= self.lon_max - self.lon_min <= FULL_TURN):
            raise ValueError(f"invalid longitude range {self.lon_min}..{self.lon_max}")

    @classmethod
    def from_points(cls, lat: ArrayLike, lon: ArrayLike, margin: float = 0.5) -> Domain:
        """Smallest box around the points (shortest longitude arc), plus ``margin`` degrees."""
        la = np.asarray(lat, dtype=np.float64)
        lo = np.asarray(lon, dtype=np.float64)
        if la.size == 0:
            raise ValueError("no points")
        best: tuple[float, float] | None = None
        for base in (-180.0, 0.0):  # [-180, 180) and [0, 360) framings
            framed = base + np.mod(lo - base, 360.0)
            span = (float(framed.min()), float(framed.max()))
            if best is None or span[1] - span[0] < best[1] - best[0]:
                best = span
        assert best is not None
        lon_min, lon_max = best[0] - margin, best[1] + margin
        if lon_max - lon_min > FULL_TURN:
            lon_min, lon_max = -180.0, 180.0
        return cls(
            max(float(la.min()) - margin, -90.0),
            min(float(la.max()) + margin, 90.0),
            lon_min,
            lon_max,
        )

    def to_json(self) -> dict[str, float]:
        """Bounds as a JSON-friendly dict."""
        return {k: getattr(self, k) for k in ("lat_min", "lat_max", "lon_min", "lon_max")}

    @classmethod
    def whole_globe(cls) -> Domain:
        """The domain covering the entire globe."""
        return cls(-90.0, 90.0, 0.0, 360.0)

from_points(lat, lon, margin=0.5) classmethod

Smallest box around the points (shortest longitude arc), plus margin degrees.

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

@classmethod
def from_points(cls, lat: ArrayLike, lon: ArrayLike, margin: float = 0.5) -> Domain:
    """Smallest box around the points (shortest longitude arc), plus ``margin`` degrees."""
    la = np.asarray(lat, dtype=np.float64)
    lo = np.asarray(lon, dtype=np.float64)
    if la.size == 0:
        raise ValueError("no points")
    best: tuple[float, float] | None = None
    for base in (-180.0, 0.0):  # [-180, 180) and [0, 360) framings
        framed = base + np.mod(lo - base, 360.0)
        span = (float(framed.min()), float(framed.max()))
        if best is None or span[1] - span[0] < best[1] - best[0]:
            best = span
    assert best is not None
    lon_min, lon_max = best[0] - margin, best[1] + margin
    if lon_max - lon_min > FULL_TURN:
        lon_min, lon_max = -180.0, 180.0
    return cls(
        max(float(la.min()) - margin, -90.0),
        min(float(la.max()) + margin, 90.0),
        lon_min,
        lon_max,
    )

to_json()

Bounds as a JSON-friendly dict.

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

def to_json(self) -> dict[str, float]:
    """Bounds as a JSON-friendly dict."""
    return {k: getattr(self, k) for k in ("lat_min", "lat_max", "lon_min", "lon_max")}

whole_globe() classmethod

The domain covering the entire globe.

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

@classmethod
def whole_globe(cls) -> Domain:
    """The domain covering the entire globe."""
    return cls(-90.0, 90.0, 0.0, 360.0)

Flag

Bases: IntFlag

Bit flags describing why a value is missing.

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

class Flag(IntFlag):
    """Bit flags describing why a value is missing."""

    OK = 0
    INVALID_INPUT = 1  # non-finite time/position/pressure
    OUTSIDE_DOMAIN = 2  # outside the cube's horizontal grid
    ABOVE_TOP = 4  # above the highest retrieved level
    BELOW_BOTTOM = 8  # below the lowest retrieved level (near/below the surface)
    MISSING_DATA = 16  # NaN in the source data or cube not published
    OUT_OF_COVERAGE = 32  # time outside the period the source covers

DataUnavailableError

Bases: LookupError

The source has no data for this key (not published, expired, out of range).

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

class DataUnavailableError(LookupError):
    """The source has no data for this key (not published, expired, out of range)."""

UncoveredPointsWarning

Bases: UserWarning

Some points fall outside the period the source covers.

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

class UncoveredPointsWarning(UserWarning):
    """Some points fall outside the period the source covers."""

SampleOptions dataclass

How to interpolate on the vertical.

vertical: "log" (linear in ln p, default) or "linear" in p. surface_pressure: "actual" (column pressures from the cube's surface pressure) or "standard" (101325 Pa everywhere, i.e. a fixed altitude->level table).

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

@dataclass(frozen=True)
class SampleOptions:
    """How to interpolate on the vertical.

    ``vertical``: ``"log"`` (linear in ln p, default) or ``"linear"`` in p.
    ``surface_pressure``: ``"actual"`` (column pressures from the cube's
    surface pressure) or ``"standard"`` (101325 Pa everywhere, i.e. a fixed
    altitude->level table).
    """

    vertical: VerticalMethod = "log"
    surface_pressure: SurfacePressureMode = "actual"

materialize_weather(source, request, *, fields=('u', 'v', 't', 'q', 'omega'), tau_cirrus=False, units=None, executor=None)

Read a pressure-level window, extending only cirrus dependencies upward.

source must map the requested aliases to native variables. Deriving tau_cirrus additionally requires t and ciwc, the complete native pressure-level axis, and optional pycontrails. Its height convention is pycontrails' ISA approximation unless geopotential is configured. Hybrid-level conversion is deliberately not implicit.

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

def materialize_weather(  # noqa: PLR0913 - explicit retrieval/preprocessing options
    source: GridSource,
    request: WeatherRequest,
    *,
    fields: Sequence[str] = ("u", "v", "t", "q", "omega"),
    tau_cirrus: bool = False,
    units: Mapping[str, str] | None = None,
    executor: Executor | None = None,
) -> WeatherCube:
    """Read a pressure-level window, extending only cirrus dependencies upward.

    ``source`` must map the requested aliases to native variables. Deriving
    ``tau_cirrus`` additionally requires ``t`` and ``ciwc``, the complete native
    pressure-level axis, and optional pycontrails. Its height convention is
    pycontrails' ISA approximation unless ``geopotential`` is configured.
    Hybrid-level conversion is deliberately not implicit.
    """
    names = tuple(fields)
    _validate_window(source, request, names)
    if tau_cirrus and "tau_cirrus" in names:
        raise ValueError("choose supplied or derived tau_cirrus, not both")
    if tau_cirrus:
        _validate_tau(source)
    metadata = dict(_UNITS if units is None else units)
    _validate_units(source, names, metadata)
    selected = source.select_fields(names)
    times = _times(
        request,
        getattr(source, "time_origin", np.datetime64("1970-01-01")),
        source.time_step,
        source.coverage(),
    )
    with ThreadPoolExecutor(max_workers=8) if executor is None else nullcontext(executor) as pool:
        first, values = _read_weather_times(selected, request, times, pool)
    if tau_cirrus:
        values["tau_cirrus"] = _derive_tau(source, request, first.vertical.ids, times)
        metadata["tau_cirrus"] = "1"
    return WeatherCube(
        lon=first.lon,
        lat=first.lat[::-1],
        level_hpa=first.vertical.ids,
        time_ns=times.astype(np.int64),
        fields=values,
        units={name: metadata[name] for name in values},
        provenance=_provenance(source, tau_cirrus),
    )

materialize_radiation(source, request, *, fields=None)

Read only bracketing ERA5 midpoint fluxes on the requested domain.

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

def materialize_radiation(
    source: ArcoRadiationSource, request: WeatherRequest, *, fields: Sequence[str] | None = None
) -> RadiationCube:
    """Read only bracketing ERA5 midpoint fluxes on the requested domain."""
    times = _times(request, source.time_origin, source.time_step, source.coverage())
    names = tuple(source.variables if fields is None else fields)
    if not names or len(set(names)) != len(names) or set(names) - source.variables.keys():
        raise ValueError("select unique configured radiation fields")
    cubes = [source.fetch(CubeKey(t), request.domain, fields=names) for t in times]
    first = cubes[0]
    _check_domain(request.domain, first.lat, first.lon)
    return RadiationCube(
        lon=first.lon,
        lat=first.lat,
        time_ns=times.astype(np.int64),
        fields={n: np.stack([c.fields[n].T for c in cubes], axis=-1) for n in first.fields},
        units=dict.fromkeys(first.fields, "W m-2"),
        time_semantics="interval_midpoint",
        provenance={
            "source": "ARCO-ERA5",
            "provider": "ECMWF",
            "dataset": "ERA5",
            "product": "reanalysis",
        },
    )

run_diffcocip(flight, provider, params, *, max_age_s)

Evolve one flight with demand-driven weather, stopping at extinction/cap.

provider supplies native-grid dicts in diffcocip's documented units. It may use Barograph caches and return larger coverage than requested. Requests include lower meteorological probes and wake downwash. Trial steps outside coverage are discarded and replayed with complete weather. The input flight follows diffcocip's dict contract (altitude in metres). This host driver does not support autodifferentiation or smoothing. Provider responses must describe one consistent meteorological dataset: extending a window must not change values at previously covered nodes. Longitudes use [-180, 180]; seam crossings can request a wide domain. max_age_s is mandatory and overrides the corresponding params value.

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

def run_diffcocip(
    flight: dict[str, Any],
    provider: Callable[[WeatherDemand], Weather],
    params: Mapping[str, Any],
    *,
    max_age_s: float,
) -> EvolutionResult:
    """Evolve one flight with demand-driven weather, stopping at extinction/cap.

    ``provider`` supplies native-grid dicts in diffcocip's documented units.
    It may use Barograph caches and return larger coverage than requested.
    Requests include lower meteorological probes and wake downwash. Trial
    steps outside coverage are discarded and replayed with complete weather.
    The input flight follows diffcocip's dict contract (altitude in metres).
    This host driver does not support autodifferentiation or smoothing.
    Provider responses must describe one consistent meteorological dataset:
    extending a window must not change values at previously covered nodes.
    Longitudes use [-180, 180]; seam crossings can request a wide domain.
    ``max_age_s`` is mandatory and overrides the corresponding params value.
    """
    options = _validate(flight, params, max_age_s)
    weather = _Coverage(provider)
    formed, state, times = _initialize(flight, options, weather)
    state, records, executed, cap, previous = _evolve(state, times, options, weather)
    cap = cap or _limit_reached(state, previous)
    diagnostics = (
        {key: np.stack([record[key] for record in records]) for key in records[0]}
        if records
        else {}
    )
    return EvolutionResult(
        float(np.sum(diagnostics.get("ef", 0))),
        "age_limit_reached" if cap else "extinct",
        formed,
        state,
        diagnostics,
        np.asarray(executed, dtype="datetime64[ns]"),
        weather.requests,
    )

arco_model_levels(*, levels=None, base_url=ARCO_BASE_URL, fields=None)

ARCO-ERA5 L137 model levels (0.25 deg, hourly) + surface pressure from the PL store.

levels restricts the hybrid levels used (default: all 137).

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

def arco_model_levels(
    *,
    levels: ArrayLike | None = None,
    base_url: str = ARCO_BASE_URL,
    fields: dict[str, str] | None = None,
) -> GridSource:
    """ARCO-ERA5 L137 model levels (0.25 deg, hourly) + surface pressure from the PL store.

    ``levels`` restricts the hybrid levels used (default: all 137).
    """
    ml = _open(f"{base_url}/{ARCO_MODEL_LEVEL}")
    pl = _open(f"{base_url}/{ARCO_PRESSURE_LEVEL}")
    vertical = model_levels(levels)
    return GridSource(
        _source_id("arco-era5-ml", model_levels().ids, vertical.ids),
        ml,
        vertical,
        fields=fields,
        level_array="hybrid",
        surface_pressure=SurfacePressure(pl),
    )

arco_pressure_levels(*, levels=None, base_url=ARCO_BASE_URL, fields=None)

ARCO-ERA5 pressure levels (0.25 deg, hourly); levels in hPa (default: all 37).

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

def arco_pressure_levels(
    *,
    levels: ArrayLike | None = None,
    base_url: str = ARCO_BASE_URL,
    fields: dict[str, str] | None = None,
) -> GridSource:
    """ARCO-ERA5 pressure levels (0.25 deg, hourly); ``levels`` in hPa (default: all 37)."""
    pl = _open(f"{base_url}/{ARCO_PRESSURE_LEVEL}")
    all_ids = np.asarray(pl.read_1d("level"), dtype=np.int64)
    vertical = pressure_levels(all_ids if levels is None else levels)
    return GridSource(
        _source_id("arco-era5-pl", all_ids, vertical.ids),
        pl,
        vertical,
        fields=fields,
        level_array="level",
    )

track_components(u, v, track_deg)

Along-track (tailwind > 0) and cross-track (from left to right > 0) wind.

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

def track_components(
    u: ArrayLike, v: ArrayLike, track_deg: ArrayLike
) -> tuple[NDArray[np.float64], NDArray[np.float64]]:
    """Along-track (tailwind > 0) and cross-track (from left to right > 0) wind."""
    chi = np.deg2rad(np.asarray(track_deg, dtype=np.float64))
    ua = np.asarray(u, dtype=np.float64)
    va = np.asarray(v, dtype=np.float64)
    return ua * np.sin(chi) + va * np.cos(chi), ua * np.cos(chi) - va * np.sin(chi)

pressure_from_altitude_ft(h_ft)

Static pressure (Pa) for a pressure altitude in feet.

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

def pressure_from_altitude_ft(h_ft: ArrayLike) -> NDArray[np.float64]:
    """Static pressure (Pa) for a pressure altitude in feet."""
    return pressure_from_altitude(np.asarray(h_ft, dtype=np.float64) * FT)

set_host_limit(host, limit)

Change the cap for host (takes effect for hosts not contacted yet).

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

def set_host_limit(host: str, limit: int) -> None:
    """Change the cap for ``host`` (takes effect for hosts not contacted yet)."""
    if limit < 1:
        raise ValueError("limit must be >= 1")
    with _lock:
        _overrides[host.lower()] = limit
        _semaphores.pop(host.lower(), None)
        _async_semaphores.pop(host.lower(), None)

Building blocks

met

Fast weather retrieval along aircraft trajectories.

ERA5 on model or pressure levels and forecast fields from Météo-France and Open-Meteo, interpolated with explicit coverage and no silent extrapolation. Surface queries accept fixed sites or trajectories without altitude.

Source

Bases: Protocol

What the enricher needs from a gridded source.

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

class Source(Protocol):
    """What the enricher needs from a gridded source."""

    source_id: str
    fields: dict[str, str]  # output name -> source-specific variable identifier
    time_step: np.timedelta64

    @property
    def stats(self) -> list[TransferStats]:
        """Transfer counters of the source's stores."""
        ...

    def cube_keys(self, t0: np.datetime64) -> tuple[CubeKey, CubeKey]:
        """Cubes bracketing the interval ``[t0, t0 + time_step)``."""
        ...

    def coverage(self) -> tuple[np.datetime64, np.datetime64] | None:
        """First and last valid times the source can serve (``None``: unknown)."""
        ...

    def tiling(self, key: CubeKey) -> Tiling | None:
        """Horizontal tiling of the cube ``key`` in storage (``None``: read as one piece)."""
        ...

    def use_cache(self, cache: RangeCache) -> None:
        """Serve the immutable bytes this source reads through ``cache`` from now on."""
        ...

    def fetch(
        self,
        key: CubeKey,
        domain: Domain,
        p_range: tuple[float, float],
        *,
        level_margin: int = 1,
        executor: Executor | None = None,
    ) -> Cube:
        """Read one cube; raise :class:`DataUnavailableError` if it does not exist."""
        ...

stats property

Transfer counters of the source's stores.

coverage()

First and last valid times the source can serve (None: unknown).

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

def coverage(self) -> tuple[np.datetime64, np.datetime64] | None:
    """First and last valid times the source can serve (``None``: unknown)."""
    ...

cube_keys(t0)

Cubes bracketing the interval [t0, t0 + time_step).

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

def cube_keys(self, t0: np.datetime64) -> tuple[CubeKey, CubeKey]:
    """Cubes bracketing the interval ``[t0, t0 + time_step)``."""
    ...

fetch(key, domain, p_range, *, level_margin=1, executor=None)

Read one cube; raise :class:DataUnavailableError if it does not exist.

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

def fetch(
    self,
    key: CubeKey,
    domain: Domain,
    p_range: tuple[float, float],
    *,
    level_margin: int = 1,
    executor: Executor | None = None,
) -> Cube:
    """Read one cube; raise :class:`DataUnavailableError` if it does not exist."""
    ...

tiling(key)

Horizontal tiling of the cube key in storage (None: read as one piece).

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

def tiling(self, key: CubeKey) -> Tiling | None:
    """Horizontal tiling of the cube ``key`` in storage (``None``: read as one piece)."""
    ...

use_cache(cache)

Serve the immutable bytes this source reads through cache from now on.

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

def use_cache(self, cache: RangeCache) -> None:
    """Serve the immutable bytes this source reads through ``cache`` from now on."""
    ...

SurfaceSource

Bases: Protocol

Native gridded surface-source contract, separate from pressure fields.

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

class SurfaceSource(Protocol):
    """Native gridded surface-source contract, separate from pressure fields."""

    time_step: np.timedelta64
    variables: dict[str, SurfaceVariable]

    @property
    def stats(self) -> list[TransferStats]: ...
    def coverage(self) -> tuple[np.datetime64, np.datetime64]: ...
    def cube_keys(self, t0: np.datetime64) -> tuple[CubeKey, CubeKey]: ...
    def tiling(self, key: CubeKey) -> Tiling | None: ...
    def use_cache(self, cache: RangeCache) -> None: ...
    def fetch(
        self, key: CubeKey, domain: Domain, *, fields: Sequence[str] | None = None
    ) -> SurfaceCube: ...

SurfaceCube dataclass

Native valid-time surface fields on an ascending regular lat/lon grid.

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

@dataclass(frozen=True)
class SurfaceCube:
    """Native valid-time surface fields on an ascending regular lat/lon grid."""

    key: CubeKey
    lat: NDArray[np.float64]
    lon: NDArray[np.float64]
    fields: dict[str, NDArray[np.float32]]
    variables: dict[str, SurfaceVariable]

    def __post_init__(self) -> None:
        for axis in (self.lat, self.lon):
            if axis.ndim != 1 or len(axis) < _MIN_AXIS_POINTS or not np.all(np.isfinite(axis)):
                raise ValueError("surface axes must have at least two finite coordinates")
            increments = np.diff(axis)
            if increments[0] <= 0 or not np.allclose(increments, increments[0]):
                raise ValueError("surface axes must be ascending and regularly spaced")
        if set(self.fields) != set(self.variables):
            raise ValueError("surface field names and metadata must agree")
        if any(a.shape != (len(self.lat), len(self.lon)) for a in self.fields.values()):
            raise ValueError("surface fields must have shape (latitude, longitude)")

    def sample(
        self, lat: ArrayLike, lon: ArrayLike, *, fields: tuple[str, ...] | None = None
    ) -> tuple[dict[str, NDArray[np.float64]], NDArray[np.uint8]]:
        """Bilinearly interpolate the grid without horizontal extrapolation."""
        la, lo = np.broadcast_arrays(np.asarray(lat, dtype=float), np.asarray(lon, dtype=float))
        la, lo = la.ravel(), lo.ravel()
        invalid = ~np.isfinite(la) | ~np.isfinite(lo)
        la = np.where(invalid, self.lat[0], la)
        lo = np.where(invalid, self.lon[0], lo)
        fi = (la - self.lat[0]) / (self.lat[1] - self.lat[0])
        centre = (self.lon[0] + self.lon[-1]) / 2
        framed = lo + 360.0 * np.round((centre - lo) / 360.0)
        fj = (framed - self.lon[0]) / (self.lon[1] - self.lon[0])
        outside = (
            (fi < -_GRID_EPS)
            | (fi > len(self.lat) - 1 + _GRID_EPS)
            | (fj < -_GRID_EPS)
            | (fj > len(self.lon) - 1 + _GRID_EPS)
        )
        fi = np.clip(np.nan_to_num(fi), 0, len(self.lat) - 1)
        fj = np.clip(np.nan_to_num(fj), 0, len(self.lon) - 1)
        i = np.minimum(fi.astype(np.int64), len(self.lat) - 2)
        j = np.minimum(fj.astype(np.int64), len(self.lon) - 2)
        wi, wj = fi - i, fj - j
        flags = np.zeros(len(la), dtype=np.uint8)
        flags[outside] = Flag.OUTSIDE_DOMAIN
        flags[invalid] = Flag.INVALID_INPUT
        values = {}
        for name in self.fields if fields is None else fields:
            grid = self.fields[name]
            value = np.zeros(len(la))
            for di, dj, weight in (
                (0, 0, (1 - wi) * (1 - wj)),
                (1, 0, wi * (1 - wj)),
                (0, 1, (1 - wi) * wj),
                (1, 1, wi * wj),
            ):
                corner = grid[i + di, j + dj]
                # Missing unused corners must not invalidate exact grid points.
                value += np.where(weight == 0, 0, weight * corner)
            flags[(~np.isfinite(value)) & (flags == 0)] = Flag.MISSING_DATA
            values[name] = value
        for value in values.values():
            value[flags != 0] = np.nan
        return values, flags

sample(lat, lon, *, fields=None)

Bilinearly interpolate the grid without horizontal extrapolation.

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

def sample(
    self, lat: ArrayLike, lon: ArrayLike, *, fields: tuple[str, ...] | None = None
) -> tuple[dict[str, NDArray[np.float64]], NDArray[np.uint8]]:
    """Bilinearly interpolate the grid without horizontal extrapolation."""
    la, lo = np.broadcast_arrays(np.asarray(lat, dtype=float), np.asarray(lon, dtype=float))
    la, lo = la.ravel(), lo.ravel()
    invalid = ~np.isfinite(la) | ~np.isfinite(lo)
    la = np.where(invalid, self.lat[0], la)
    lo = np.where(invalid, self.lon[0], lo)
    fi = (la - self.lat[0]) / (self.lat[1] - self.lat[0])
    centre = (self.lon[0] + self.lon[-1]) / 2
    framed = lo + 360.0 * np.round((centre - lo) / 360.0)
    fj = (framed - self.lon[0]) / (self.lon[1] - self.lon[0])
    outside = (
        (fi < -_GRID_EPS)
        | (fi > len(self.lat) - 1 + _GRID_EPS)
        | (fj < -_GRID_EPS)
        | (fj > len(self.lon) - 1 + _GRID_EPS)
    )
    fi = np.clip(np.nan_to_num(fi), 0, len(self.lat) - 1)
    fj = np.clip(np.nan_to_num(fj), 0, len(self.lon) - 1)
    i = np.minimum(fi.astype(np.int64), len(self.lat) - 2)
    j = np.minimum(fj.astype(np.int64), len(self.lon) - 2)
    wi, wj = fi - i, fj - j
    flags = np.zeros(len(la), dtype=np.uint8)
    flags[outside] = Flag.OUTSIDE_DOMAIN
    flags[invalid] = Flag.INVALID_INPUT
    values = {}
    for name in self.fields if fields is None else fields:
        grid = self.fields[name]
        value = np.zeros(len(la))
        for di, dj, weight in (
            (0, 0, (1 - wi) * (1 - wj)),
            (1, 0, wi * (1 - wj)),
            (0, 1, (1 - wi) * wj),
            (1, 1, wi * wj),
        ):
            corner = grid[i + di, j + dj]
            # Missing unused corners must not invalidate exact grid points.
            value += np.where(weight == 0, 0, weight * corner)
        flags[(~np.isfinite(value)) & (flags == 0)] = Flag.MISSING_DATA
        values[name] = value
    for value in values.values():
        value[flags != 0] = np.nan
    return values, flags

SurfaceVariable dataclass

Unit, height above ground and native temporal meaning of one field.

height_m is NaN when no finite height applies, as for a top-of-atmosphere radiative boundary.

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

@dataclass(frozen=True)
class SurfaceVariable:
    """Unit, height above ground and native temporal meaning of one field.

    ``height_m`` is NaN when no finite height applies, as for a
    top-of-atmosphere radiative boundary.
    """

    unit: str
    height_m: float
    temporal: Literal["instant", "sum", "max"] = "instant"

CubeKey dataclass

Validity time of a cube and, for forecasts, the run it belongs to.

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

@dataclass(frozen=True, order=True)
class CubeKey:
    """Validity time of a cube and, for forecasts, the run it belongs to."""

    valid: np.datetime64
    run: np.datetime64 | None = None
    source: str | None = None  # set by composite sources to route the fetch

    @property
    def lead(self) -> np.timedelta64 | None:
        """Forecast lead time, or ``None`` for analyses and stitched forecast archives."""
        return None if self.run is None else self.valid - self.run

    def __str__(self) -> str:
        return str(self.valid) if self.run is None else f"{self.valid} (run {self.run})"

lead property

Forecast lead time, or None for analyses and stitched forecast archives.

Cube dataclass

One hour of gridded fields on a regional grid, levels top to bottom.

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

@dataclass
class Cube:
    """One hour of gridded fields on a regional grid, levels top to bottom."""

    source: str
    hour: np.datetime64
    lat: NDArray[np.float64]  # descending
    lon: NDArray[np.float64]  # unwrapped, ascending
    vertical: VerticalCoordinate
    fields: dict[str, NDArray[np.float32]]  # (level, lat, lon)
    ps: NDArray[np.float32] | None = None  # surface pressure (Pa), (lat, lon)
    run: np.datetime64 | None = None  # forecast base time (None for analyses)

    def __post_init__(self) -> None:
        shape = (len(self.vertical), len(self.lat), len(self.lon))
        for name, arr in self.fields.items():
            if arr.shape != shape:
                raise ValueError(f"field {name} has shape {arr.shape}, expected {shape}")
        if self.ps is not None and self.ps.shape != shape[1:]:
            raise ValueError("surface pressure grid does not match the fields")
        if len(self.lat) < _MIN_GRID_POINTS or len(self.lon) < _MIN_GRID_POINTS:
            raise ValueError("a cube needs at least 2x2 grid points")

    @property
    def nbytes(self) -> int:
        """Memory held by the fields and surface pressure."""
        total = sum(a.nbytes for a in self.fields.values())
        return total + (0 if self.ps is None else self.ps.nbytes)

nbytes property

Memory held by the fields and surface pressure.

WeatherCube dataclass

Bases: RadiationCube

Pressure-level grid with dimensions (lon, lat, level_hpa, time).

A cube may contain wind only; CoCiP exports require u, v, t, q, omega, tau_cirrus. Optical depth must be supplied after a correct full-column calculation; it cannot safely be derived from a cropped pressure slab.

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

@dataclass(kw_only=True)
class WeatherCube(RadiationCube):
    """Pressure-level grid with dimensions ``(lon, lat, level_hpa, time)``.

    A cube may contain wind only; CoCiP exports require ``u, v, t, q, omega,
    tau_cirrus``. Optical depth must be supplied after a correct full-column
    calculation; it cannot safely be derived from a cropped pressure slab.
    """

    level_hpa: npt.ArrayLike
    time_semantics: str = "valid"

    def __post_init__(self) -> None:
        """Validate the pressure axis and inherited grid contract."""
        self.level_hpa = _axis(self.level_hpa, "level_hpa")
        if np.any(np.asarray(self.level_hpa) <= 0):
            raise ValueError("level_hpa must be positive")
        if self.time_semantics != "valid":
            raise ValueError("Pressure-level weather must use valid timestamps")
        super().__post_init__()

    @property
    def specs(self) -> Mapping[str, tuple[str, str, str]]:
        """Pressure-level fields and units."""
        return MET_FIELDS

    @property
    def shape(self) -> tuple[int, ...]:
        """Required array shape in export axis order."""
        return (
            np.size(self.lon),
            np.size(self.lat),
            np.size(self.level_hpa),
            np.size(self.time_ns),
        )

    @property
    def coordinates(self) -> dict[str, npt.NDArray[Any]]:
        """Engine coordinates, including pressure in hPa."""
        return {
            "longitude": np.asarray(self.lon),
            "latitude": np.asarray(self.lat),
            "level": np.asarray(self.level_hpa),
            "time": np.asarray(self.time_ns).astype("datetime64[ns]"),
        }

    def _require_complete(self) -> None:
        missing = COCIP_FIELDS - self.fields.keys()
        if missing:
            raise ValueError(f"Missing required CoCiP fields: {sorted(missing)}")

coordinates property

Engine coordinates, including pressure in hPa.

shape property

Required array shape in export axis order.

specs property

Pressure-level fields and units.

__post_init__()

Validate the pressure axis and inherited grid contract.

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

def __post_init__(self) -> None:
    """Validate the pressure axis and inherited grid contract."""
    self.level_hpa = _axis(self.level_hpa, "level_hpa")
    if np.any(np.asarray(self.level_hpa) <= 0):
        raise ValueError("level_hpa must be positive")
    if self.time_semantics != "valid":
        raise ValueError("Pressure-level weather must use valid timestamps")
    super().__post_init__()

RadiationCube dataclass

TOA flux grid with explicitly resolved timestamps.

Arrays have dimensions (lon, lat, time). time_semantics must be valid (instantaneous flux) or interval_midpoint (mean flux assigned to its averaging-interval midpoint). Interval-end or accumulated radiation must be converted upstream. No engine export shifts these timestamps.

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

@dataclass(kw_only=True)
class RadiationCube:
    """TOA flux grid with explicitly resolved timestamps.

    Arrays have dimensions ``(lon, lat, time)``. ``time_semantics`` must be
    ``valid`` (instantaneous flux) or ``interval_midpoint`` (mean flux assigned
    to its averaging-interval midpoint). Interval-end or accumulated radiation
    must be converted upstream. No engine export shifts these timestamps.
    """

    lon: npt.ArrayLike
    lat: npt.ArrayLike
    time_ns: npt.ArrayLike
    fields: Mapping[str, npt.ArrayLike]
    units: Mapping[str, str]
    provenance: Mapping[str, str]
    time_semantics: str

    def __post_init__(self) -> None:
        """Validate axes, shape, units, and timestamp semantics."""
        self.lon = _axis(self.lon, "longitude")
        self.lat = _axis(self.lat, "latitude")
        if np.any(np.abs(self.lat) > _MAX_LATITUDE):
            raise ValueError("latitude must be in [-90, 90]")
        self.time_ns = _times(self.time_ns)
        if self.time_semantics not in {"valid", "interval_midpoint"}:
            raise ValueError("Resolve radiation times to valid or interval_midpoint first")
        self.fields = _fields(self.fields, self.units, self.shape, self.specs)
        self.units = {k: self.specs[k][1] for k in self.fields}
        self.provenance = dict(self.provenance)

    @property
    def specs(self) -> Mapping[str, tuple[str, str, str]]:
        """Recognized variables, canonical and engine units."""
        return RAD_FIELDS

    @property
    def shape(self) -> tuple[int, ...]:
        """Required array shape in export axis order."""
        return (np.size(self.lon), np.size(self.lat), np.size(self.time_ns))

    @property
    def coordinates(self) -> dict[str, npt.NDArray[Any]]:
        """Coordinates in native engine dimension order."""
        return {
            "longitude": np.asarray(self.lon),
            "latitude": np.asarray(self.lat),
            "time": np.asarray(self.time_ns).astype("datetime64[ns]"),
        }

    @property
    def coverage(self) -> dict[str, tuple[Any, Any]]:
        """Inclusive coordinate bounds, without implying extrapolation support."""
        return {k: (v[0], v[-1]) for k, v in self.coordinates.items()}

    def _require_complete(self) -> None:
        missing = set(RAD_FIELDS) - self.fields.keys()
        if missing:
            raise ValueError(f"Missing required radiation fields: {sorted(missing)}")

    def to_diffcocip(self) -> dict[str, Array]:
        """Export the NumPy dict accepted by diffcocip, without importing JAX.

        Every interpolation axis needs at least two points. The engine's time
        bridge uses seconds since 2000; grids before this epoch are rejected.
        Metadata stays available on this container, not in the engine dict.
        """
        self._require_complete()
        if min(self.shape) < _MIN_INTERPOLATION_POINTS:
            raise ValueError("diffcocip requires at least two points on every axis")
        time_ns = np.asarray(self.time_ns)
        if np.any(time_ns < _EPOCH_NS):
            raise ValueError("diffcocip requires timestamps on or after 2000-01-01")
        relative = time_ns - _EPOCH_NS
        out = {
            "lon": np.asarray(self.lon).copy(),
            "lat": np.asarray(self.lat).copy(),
            "time_s": relative / 1e9,
            "days_since_2000": relative / _DAY_NS,
            "hours_of_day": np.remainder(time_ns, _DAY_NS) / 3_600_000_000_000,
        }
        if "level" in self.coordinates:
            out["level"] = self.coordinates["level"].copy()
        out.update({self.specs[k][0]: np.asarray(v).copy() for k, v in self.fields.items()})
        return out

    def to_pycontrails(self, *, require_cocip: bool = True) -> Any:
        """Export an optional pycontrails MetDataset with resolved time metadata.

        Import pycontrails and xarray only on this call. Radiation's time
        coordinate is marked as resolved, preventing a second engine shift.
        Set ``require_cocip=False`` to export partial fields for preprocessing.
        """
        if require_cocip:
            self._require_complete()
        try:
            xr = importlib.import_module("xarray")
            pc = importlib.import_module("pycontrails")
        except ImportError as exc:
            raise ImportError("Install pycontrails and xarray to export MetDataset") from exc
        coords = self.coordinates
        dims = tuple(coords)
        data = {
            self.specs[k][0]: (dims, np.asarray(v).copy(), {"units": self.specs[k][2]})
            for k, v in self.fields.items()
        }
        ds = xr.Dataset(data, coords=coords, attrs=dict(self.provenance))
        if "level" not in coords:
            ds = ds.expand_dims(level=[-1])
            ds["time"].attrs["shift_radiation_time"] = "already resolved by caller"
        return pc.MetDataset(ds)

coordinates property

Coordinates in native engine dimension order.

coverage property

Inclusive coordinate bounds, without implying extrapolation support.

shape property

Required array shape in export axis order.

specs property

Recognized variables, canonical and engine units.

__post_init__()

Validate axes, shape, units, and timestamp semantics.

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

def __post_init__(self) -> None:
    """Validate axes, shape, units, and timestamp semantics."""
    self.lon = _axis(self.lon, "longitude")
    self.lat = _axis(self.lat, "latitude")
    if np.any(np.abs(self.lat) > _MAX_LATITUDE):
        raise ValueError("latitude must be in [-90, 90]")
    self.time_ns = _times(self.time_ns)
    if self.time_semantics not in {"valid", "interval_midpoint"}:
        raise ValueError("Resolve radiation times to valid or interval_midpoint first")
    self.fields = _fields(self.fields, self.units, self.shape, self.specs)
    self.units = {k: self.specs[k][1] for k in self.fields}
    self.provenance = dict(self.provenance)

to_diffcocip()

Export the NumPy dict accepted by diffcocip, without importing JAX.

Every interpolation axis needs at least two points. The engine's time bridge uses seconds since 2000; grids before this epoch are rejected. Metadata stays available on this container, not in the engine dict.

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

def to_diffcocip(self) -> dict[str, Array]:
    """Export the NumPy dict accepted by diffcocip, without importing JAX.

    Every interpolation axis needs at least two points. The engine's time
    bridge uses seconds since 2000; grids before this epoch are rejected.
    Metadata stays available on this container, not in the engine dict.
    """
    self._require_complete()
    if min(self.shape) < _MIN_INTERPOLATION_POINTS:
        raise ValueError("diffcocip requires at least two points on every axis")
    time_ns = np.asarray(self.time_ns)
    if np.any(time_ns < _EPOCH_NS):
        raise ValueError("diffcocip requires timestamps on or after 2000-01-01")
    relative = time_ns - _EPOCH_NS
    out = {
        "lon": np.asarray(self.lon).copy(),
        "lat": np.asarray(self.lat).copy(),
        "time_s": relative / 1e9,
        "days_since_2000": relative / _DAY_NS,
        "hours_of_day": np.remainder(time_ns, _DAY_NS) / 3_600_000_000_000,
    }
    if "level" in self.coordinates:
        out["level"] = self.coordinates["level"].copy()
    out.update({self.specs[k][0]: np.asarray(v).copy() for k, v in self.fields.items()})
    return out

to_pycontrails(*, require_cocip=True)

Export an optional pycontrails MetDataset with resolved time metadata.

Import pycontrails and xarray only on this call. Radiation's time coordinate is marked as resolved, preventing a second engine shift. Set require_cocip=False to export partial fields for preprocessing.

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

def to_pycontrails(self, *, require_cocip: bool = True) -> Any:
    """Export an optional pycontrails MetDataset with resolved time metadata.

    Import pycontrails and xarray only on this call. Radiation's time
    coordinate is marked as resolved, preventing a second engine shift.
    Set ``require_cocip=False`` to export partial fields for preprocessing.
    """
    if require_cocip:
        self._require_complete()
    try:
        xr = importlib.import_module("xarray")
        pc = importlib.import_module("pycontrails")
    except ImportError as exc:
        raise ImportError("Install pycontrails and xarray to export MetDataset") from exc
    coords = self.coordinates
    dims = tuple(coords)
    data = {
        self.specs[k][0]: (dims, np.asarray(v).copy(), {"units": self.specs[k][2]})
        for k, v in self.fields.items()
    }
    ds = xr.Dataset(data, coords=coords, attrs=dict(self.provenance))
    if "level" not in coords:
        ds = ds.expand_dims(level=[-1])
        ds["time"].attrs["shift_radiation_time"] = "already resolved by caller"
    return pc.MetDataset(ds)

WeatherRequest dataclass

Region, UTC time bounds and pressure range (Pa) requiring interpolation.

Includes neighbouring native times and pressure levels, without imposing a fixed integration horizon. Callers supply margins for lower probes, sedimentation and evolution; a window is not a bound on future motion.

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

@dataclass(frozen=True)
class WeatherRequest:
    """Region, UTC time bounds and pressure range (Pa) requiring interpolation.

    Includes neighbouring native times and pressure levels, without imposing
    a fixed integration horizon. Callers supply margins for lower probes,
    sedimentation and evolution; a window is not a bound on future motion.
    """

    domain: Domain
    start: np.datetime64
    end: np.datetime64
    pressure_pa: tuple[float, float]

    def __post_init__(self) -> None:
        start, end = np.datetime64(self.start, "ns"), np.datetime64(self.end, "ns")
        if np.isnat(start) or np.isnat(end) or end < start:
            raise ValueError("weather request needs ordered finite UTC time bounds")
        lo, hi = self.pressure_pa
        if not np.isfinite([lo, hi]).all() or not 0 < lo <= hi:
            raise ValueError("weather request needs ordered positive pressures in Pa")
        object.__setattr__(self, "start", start)
        object.__setattr__(self, "end", end)

WeatherDemand dataclass

Exact probe coordinates to cover, including interpolation neighbours.

points columns are longitude degrees, latitude degrees, pressure hPa, and seconds since 2000-01-01. Radiation ignores the pressure column. A provider returns complete diffcocip met/radiation dictionaries covering all probes in this request. Radiation values must be W/m² at midpoints. Spatial/time bounds are requests, never permission to extrapolate.

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

@dataclass(frozen=True)
class WeatherDemand:
    """Exact probe coordinates to cover, including interpolation neighbours.

    ``points`` columns are longitude degrees, latitude degrees, pressure hPa,
    and seconds since 2000-01-01. Radiation ignores the pressure column.
    A provider returns complete diffcocip met/radiation dictionaries covering
    all probes in this request. Radiation values must be W/m² at midpoints.
    Spatial/time bounds are requests, never permission to extrapolate.
    """

    points: Array

    @property
    def bounds(self) -> dict[str, tuple[float, float]]:
        """Minimum/maximum requested coordinates, in diffcocip units."""
        return {
            axis: (float(self.points[:, i].min()), float(self.points[:, i].max()))
            for i, axis in enumerate(_AXES)
        }

bounds property

Minimum/maximum requested coordinates, in diffcocip units.

PycontrailsWeather

Supply lazy native pressure grids and optional resolved radiation grids.

datasets creates pycontrails MetDataset objects with no field reads. Source constructors may already have read coordinate metadata. Fields are read only when the consumer evaluates an indexed array. The retained-cache budget excludes source buffers, tau-column temporaries, xarray/consumer caches and caller-held arrays; it is not a process memory limit.

Metadata comes from source attributes or explicit provenance; unknown metadata stays unknown. Radiation times are already resolved by its source. Use native CoCiP with interpolation_bounds_error=True and compute_tau_cirrus_in_model_init=False to preserve strict coverage and lazy cirrus reads. Missing/nonfinite weather raises, never means extinction. Only final source time chunks qualify for bridge retention; recent ERA5T is always materialized anew. Do not mutate source configuration in use.

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

class PycontrailsWeather:
    """Supply lazy native pressure grids and optional resolved radiation grids.

    ``datasets`` creates pycontrails MetDataset objects with no field reads.
    Source constructors may already have read coordinate metadata. Fields are
    read only when the consumer evaluates an indexed array. The retained-cache
    budget excludes source buffers, tau-column temporaries, xarray/consumer
    caches and caller-held arrays; it is not a process memory limit.

    Metadata comes from source attributes or explicit ``provenance``; unknown
    metadata stays unknown. Radiation times are already resolved by its source.
    Use native CoCiP with ``interpolation_bounds_error=True`` and
    ``compute_tau_cirrus_in_model_init=False`` to preserve strict coverage and
    lazy cirrus reads. Missing/nonfinite weather raises, never means extinction.
    Only final source time chunks qualify for bridge retention; recent ERA5T
    is always materialized anew. Do not mutate source configuration in use.
    """

    def __init__(  # noqa: PLR0913 - public optional retrieval configuration
        self,
        source: GridSource,
        radiation: ArcoRadiationSource | None = None,
        *,
        fields: Sequence[str] = _DEFAULT_FIELDS,
        max_cache_bytes: int = 256 * 1024**2,
        cache_dir: str | Path | None = None,
        provenance: Mapping[str, str] | None = None,
    ) -> None:
        names = _fields(source, fields)
        if (
            not isinstance(max_cache_bytes, int)
            or isinstance(max_cache_bytes, bool)
            or max_cache_bytes < 0
        ):
            raise ValueError("max_cache_bytes must be a nonnegative integer")
        self.source, self.radiation, self.fields = source, radiation, names
        self.max_cache_bytes = max_cache_bytes
        self.provenance = dict(provenance or {})
        self._cache: OrderedDict[Any, NDArray[np.float64]] = OrderedDict()
        self._lock = threading.RLock()
        self._bytes = self._hits = self._misses = self._evictions = 0
        self._seconds = 0.0
        self._field_reads: dict[str, int] = {}
        if cache_dir is not None:
            cache = RangeCache(cache_dir)
            source.use_cache(cache)
            if radiation is not None:
                radiation.use_cache(cache)

    def _retention_allowed(self, field: str, axes: Axes, radiation: bool) -> bool:
        source = self.radiation.grid if radiation and self.radiation is not None else self.source
        # Reserve a complete time chunk beyond the selected interval. This is
        # deliberately conservative for midpoint radiation and tau dependencies.
        names = ("t", "ciwc") if field == "tau_cirrus" and field not in source.fields else (field,)
        arrays = tuple(source.fields[name] for name in names)
        hours = max(source.store.array(name).chunks[0] for name in arrays)
        latest = axes[-1].max() + np.timedelta64(hours + 1, "h")
        return bool(latest < np.datetime64("now", "ns") - source.final_after)

    def _read(self, field: str, axes: Axes, radiation: bool = False) -> NDArray[np.float64]:
        key = (field, radiation, tuple((a.dtype.str, a.tobytes()) for a in axes))
        with self._lock:
            eligible = self.max_cache_bytes > 0 and self._retention_allowed(field, axes, radiation)
            if eligible and key in self._cache:
                self._cache.move_to_end(key)
                self._hits += 1
                return self._cache[key]
            self._misses += 1
            started = perf_counter()
            out = self._materialize(field, axes, radiation)
            if not np.isfinite(out).all():
                raise DataUnavailableError("Nonfinite materialized meteorology")
            out.setflags(write=False)
            self._seconds += perf_counter() - started
            self._field_reads[field] = self._field_reads.get(field, 0) + 1
            if eligible and out.nbytes <= self.max_cache_bytes:
                while self._cache and self._bytes + out.nbytes > self.max_cache_bytes:
                    self._bytes -= self._cache.popitem(last=False)[1].nbytes
                    self._evictions += 1
                self._cache[key] = out
                self._bytes += out.nbytes
            return out

    def _materialize(self, field: str, axes: Axes, radiation: bool) -> NDArray[np.float64]:
        lon, lat, level, times = axes
        request = WeatherRequest(
            Domain(float(lat.min()), float(lat.max()), float(lon.min()), float(lon.max())),
            times.min(),
            times.max(),
            (1.0, 1.0) if radiation else (float(level.min() * 100), float(level.max() * 100)),
        )
        if radiation:
            if self.radiation is None:
                raise ValueError("radiation source is not configured")
            cube = materialize_radiation(self.radiation, request, fields=(field,))
            native: Axes = (
                np.asarray(cube.lon),
                np.asarray(cube.lat),
                np.asarray(cube.time_ns).astype("datetime64[ns]"),
            )
            return np.array(
                _select(native, (lon, lat, times), cube.fields[field])[:, :, None, :], copy=True
            )
        tau = field == "tau_cirrus" and field not in self.source.fields
        weather = materialize_weather(
            self.source, request, fields=("t",) if tau else (field,), tau_cirrus=tau
        )
        native = (
            np.asarray(weather.lon),
            np.asarray(weather.lat),
            np.asarray(weather.level_hpa),
            np.asarray(weather.time_ns).astype("datetime64[ns]"),
        )
        return _select(native, axes, weather.fields[field])

    def _dataset(self, start: np.datetime64, end: np.datetime64, radiation: bool) -> Any:
        source: Any = self.radiation if radiation else self.source
        grid = source.grid if radiation else source
        request = WeatherRequest(Domain.whole_globe(), start, end, (1.0, 1.0))
        origin = getattr(source, "time_origin", np.datetime64("1970-01-01"))
        times = _times(request, origin, source.time_step, source.coverage())
        coords = {
            "longitude": _longitude(grid.lon),
            "latitude": np.sort(np.asarray(grid.lat, dtype=np.float64)),
            "level": np.array([-1.0])
            if radiation
            else np.asarray(self.source.vertical.ids, dtype=np.float64),
            "time": times,
        }
        specs = (
            {name: RAD_FIELDS[name] for name in source.variables}
            if radiation
            else {name: MET_FIELDS[name] for name in self.fields}
        )
        variables = {
            spec[0]: lazy_variable(coords, partial(self._read, name, radiation=radiation), spec[2])
            for name, spec in specs.items()
        }
        xr = import_module("xarray")
        ds = xr.Dataset(variables, coords=coords, attrs=_attributes(grid, self.provenance))
        if radiation:
            ds.time.attrs["shift_radiation_time"] = "already resolved by source"
        return import_module("pycontrails").MetDataset(ds, wrap_longitude=False)

    def datasets(self, start: np.datetime64, end: np.datetime64) -> tuple[Any, Any | None]:
        """Construct metadata-only lazy datasets covering the requested UTC interval.

        Times are bracketed with each source's own cadence and origin. Spatial
        and pressure axes remain native; a global longitude axis is closed at
        its periodic seam. An omitted radiation source returns ``None``.
        """
        return self._dataset(start, end, False), self._dataset(
            start, end, True
        ) if self.radiation is not None else None

    def cache_info(self) -> dict[str, Any]:
        """Bounded retention and cumulative successful field materialization counters."""
        with self._lock:
            return {
                "bytes": self._bytes,
                "entries": len(self._cache),
                "hits": self._hits,
                "misses": self._misses,
                "evictions": self._evictions,
                "max_bytes": self.max_cache_bytes,
                "materialization_seconds": self._seconds,
                "field_reads": dict(self._field_reads),
            }

    def clear(self) -> None:
        """Drop bridge-owned arrays; consumer/xarray caches may retain their copies."""
        with self._lock:
            self._cache.clear()
            self._bytes = 0

cache_info()

Bounded retention and cumulative successful field materialization counters.

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

def cache_info(self) -> dict[str, Any]:
    """Bounded retention and cumulative successful field materialization counters."""
    with self._lock:
        return {
            "bytes": self._bytes,
            "entries": len(self._cache),
            "hits": self._hits,
            "misses": self._misses,
            "evictions": self._evictions,
            "max_bytes": self.max_cache_bytes,
            "materialization_seconds": self._seconds,
            "field_reads": dict(self._field_reads),
        }

clear()

Drop bridge-owned arrays; consumer/xarray caches may retain their copies.

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

def clear(self) -> None:
    """Drop bridge-owned arrays; consumer/xarray caches may retain their copies."""
    with self._lock:
        self._cache.clear()
        self._bytes = 0

datasets(start, end)

Construct metadata-only lazy datasets covering the requested UTC interval.

Times are bracketed with each source's own cadence and origin. Spatial and pressure axes remain native; a global longitude axis is closed at its periodic seam. An omitted radiation source returns None.

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

def datasets(self, start: np.datetime64, end: np.datetime64) -> tuple[Any, Any | None]:
    """Construct metadata-only lazy datasets covering the requested UTC interval.

    Times are bracketed with each source's own cadence and origin. Spatial
    and pressure axes remain native; a global longitude axis is closed at
    its periodic seam. An omitted radiation source returns ``None``.
    """
    return self._dataset(start, end, False), self._dataset(
        start, end, True
    ) if self.radiation is not None else None

GridSource

An hourly, regular lat/lon zarr store with a level dimension.

vertical lists all levels of the store's level axis, in the store's order (top to bottom). Cubes carry only the levels needed to bracket the requested pressures, read with block-level range requests. With a cache (:meth:use_cache), the field chunks of hours older than final_after are kept on disk; metadata and recent (ERA5T) hours are always re-read.

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

class GridSource:
    """An hourly, regular lat/lon zarr store with a level dimension.

    ``vertical`` lists all levels of the store's level axis, in the store's
    order (top to bottom). Cubes carry only the levels needed to bracket the
    requested pressures, read with block-level range requests. With a cache
    (:meth:`use_cache`), the field chunks of hours older than ``final_after``
    are kept on disk; metadata and recent (ERA5T) hours are always re-read.
    """

    def __init__(  # noqa: PLR0913 - keyword-only configuration of a public source
        self,
        source_id: str,
        store: ZarrStore,
        vertical: VerticalCoordinate,
        *,
        fields: dict[str, str] | None = None,
        level_array: str,
        surface_pressure: SurfacePressure | None = None,
        final_after: np.timedelta64 = FINAL_AFTER,
    ) -> None:
        self.source_id = source_id
        self.final_after = final_after
        self.store = store
        self.vertical = vertical
        self.fields = dict(fields or DEFAULT_FIELDS)
        self._cache_fields = set(self.fields.values())
        self.surface_pressure = surface_pressure
        if vertical.needs_surface_pressure and surface_pressure is None:
            raise ValueError("hybrid levels need a surface pressure array")
        store_levels = np.asarray(store.read_1d(level_array), dtype=np.int64)
        missing = np.setdiff1d(vertical.ids, store_levels)
        if missing.size:
            raise ValueError(f"{source_id}: levels {missing.tolist()} not in the store")
        # position on the store's level axis of each level of ``vertical``
        self._store_index = np.searchsorted(store_levels, vertical.ids)
        if not np.array_equal(store_levels[self._store_index], vertical.ids):
            raise ValueError(f"{source_id}: store level axis must be sorted top to bottom")
        self.lat = store.read_1d("latitude").astype(np.float64)
        self.lon = store.read_1d("longitude").astype(np.float64)
        self._epoch = _hour_epoch(store.array("time").attrs.get("units", ""))
        self._t0 = int(store.read_1d("time", 0, 1)[0])
        self._checked_time_chunks: set[int] = set()
        self._lock = threading.Lock()
        self._tiling: Tiling | None = None
        self._tiling_known = False

    def select_fields(self, names: Sequence[str]) -> GridSource:
        """Independent field view sharing stores and byte caches with this source.

        Only listed output fields are fetched. Cache eligibility retains all
        original fields, so attaching a cache through a view cannot narrow the
        parent's cache policy. Configure shared stores before concurrent use.
        """
        if not names or len(set(names)) != len(names):
            raise ValueError("fields must be a nonempty sequence of unique names")
        unknown = set(names) - self.fields.keys()
        if unknown:
            raise ValueError(f"unknown fields: {sorted(unknown)}")
        view = copy(self)
        view.fields = {name: self.fields[name] for name in names}
        view._tiling = None
        view._tiling_known = False
        return view

    @property
    def stats(self) -> list[TransferStats]:
        """Transfer counters of every underlying byte store."""
        stores = [self.store.store]
        if self.surface_pressure is not None and self.surface_pressure.store is not self.store:
            stores.append(self.surface_pressure.store.store)
        return [s.stats for s in stores]

    def time_index(self, hour: np.datetime64) -> int:
        """Index of ``hour`` on the store's time axis, verified against the time array.

        The axis is assumed hourly and contiguous; each time chunk is checked
        once (a few KiB) before its indices are trusted.
        """
        hours = int((np.datetime64(hour, "h") - self._epoch) / np.timedelta64(1, "h"))
        idx = hours - self._t0
        meta = self.store.array("time")
        if not 0 <= idx < meta.shape[0]:
            raise KeyError(f"{hour} outside the store's time axis")
        size = meta.chunks[0]
        chunk = idx // size
        with self._lock:
            known = chunk in self._checked_time_chunks
        if not known:
            start, stop = chunk * size, min((chunk + 1) * size, meta.shape[0])
            values = self.store.read_1d("time", start, stop)
            if not np.array_equal(values, np.arange(start, stop) + self._t0):
                raise RuntimeError(f"{self.source_id}: time axis is not hourly-contiguous")
            with self._lock:
                self._checked_time_chunks.add(chunk)
        return idx

    time_step = np.timedelta64(1, "h")

    def coverage(self) -> tuple[np.datetime64, np.datetime64] | None:
        """Published period, from the store's ``valid_time_start``/``valid_time_stop``.

        For ARCO-ERA5 the model-level store lags ERA5T (about three months).
        """
        attrs = self.store.attrs
        start, stop = attrs.get("valid_time_start"), attrs.get("valid_time_stop")
        if not start or not stop:
            return None
        begin = np.datetime64(start, "h")
        end = (
            np.datetime64(stop, "D") + np.timedelta64(23, "h")
            if len(stop) <= len("YYYY-MM-DD")
            else np.datetime64(stop, "h")
        )
        return begin, np.datetime64(end, "h")

    def tiling(self, key: CubeKey) -> Tiling | None:
        """Latitude bands of about one Blosc block (every longitude), the same for every hour.

        Read once from the header of one fixed chunk (the first published hour),
        so every call and process probes the same, cacheable, few KiB.
        """
        if self._tiling_known:
            return self._tiling
        coverage = self.coverage()
        try:
            t = 0 if coverage is None else self.time_index(coverage[0])
        except KeyError:
            t = 0
        name = next(iter(self.fields.values()))
        rows = self.store.block_rows(name, [t, 0][: self._lead_dims(name)])
        if rows is not None and len(self.lat) > 1:
            step = rows * abs(float(self.lat[0] - self.lat[1]))
            bands = int(np.ceil(len(self.lat) / rows))
            self._tiling = Tiling(lat0=float(self.lat[0]) - bands * step, lat_step=step)
        self._tiling_known = True
        return self._tiling

    def _lead_dims(self, name: str) -> int:
        """Number of leading (time, level) dimensions of array ``name``."""
        return len(self.store.array(name).shape) - 2

    def configure_decoded_cache(self, max_bytes: int) -> None:
        """Bound decoded Blosc reuse to final field chunks, using ``final_after``.

        Field views share this reader and its budget. Recent ERA5T chunks never
        qualify. Surface pressure in a separate reader has its own budget.
        """
        if not isinstance(max_bytes, int) or isinstance(max_bytes, bool) or max_bytes < 0:
            raise ValueError("max_bytes must be a nonnegative integer")
        if not max_bytes:
            self.store.configure_decoded_cache(0, immutable=lambda key: False)
            return
        arrays = set(self._cache_fields)
        if self.surface_pressure is not None and self.surface_pressure.store is self.store:
            arrays.add(self.surface_pressure.array)
        self.store.configure_decoded_cache(
            max_bytes, immutable=_final_keys(self.store, arrays, self.final_after)
        )

    def use_cache(self, cache: RangeCache) -> None:
        """Keep the final hours' field chunks in ``cache`` (see the class docstring)."""
        arrays = {*self._cache_fields, "time"}  # past time chunks never change either
        stores = [(self.store, arrays)]
        if self.surface_pressure is not None:
            sp = self.surface_pressure
            if sp.store is self.store:
                arrays.add(sp.array)
            else:
                stores.append((sp.store, {sp.array}))
        for zarr, names in stores:
            inner = zarr.store.inner if isinstance(zarr.store, CachedStore) else zarr.store
            final = _final_keys(zarr, names, self.final_after)
            zarr.store = CachedStore(inner, cache, namespace_for(inner.location), final)

    def cube_keys(self, t0: np.datetime64) -> tuple[CubeKey, CubeKey]:
        """Analyses at ``t0`` and one hour later (no run: reanalysis)."""
        h = np.datetime64(t0, "h")
        return CubeKey(h), CubeKey(h + self.time_step)

    def fetch(
        self,
        key: CubeKey,
        domain: Domain,
        p_range: tuple[float, float],
        *,
        level_margin: int = 1,
        executor: Executor | None = None,
    ) -> Cube:
        """Read one hour over ``domain`` with the levels bracketing ``p_range`` (Pa)."""
        hour = key.valid
        try:
            t = self.time_index(hour)
        except KeyError as exc:
            raise DataUnavailableError(f"{self.source_id}: {hour} not in the store") from exc
        grid = GridIndex.locate(domain, self.lat, self.lon)
        ps: NDArray[np.float32] | None = None
        ps_lo = ps_hi = STANDARD_SURFACE_PRESSURE
        if self.surface_pressure is not None:
            sp = self.surface_pressure
            ps = sp.store.read_rows(sp.array, [(t,)], grid.rows, executor=executor)[0][:, grid.cols]
            if np.isfinite(ps).any():
                ps_lo, ps_hi = float(np.nanmin(ps)), float(np.nanmax(ps))
        positions = self.vertical.select_bracketing(
            p_range[0], p_range[1], ps_lo, ps_hi, level_margin
        )
        lead = [(t, int(k)) for k in self._store_index[positions]]
        fields = {
            name: self.store.read_rows(array, lead, grid.rows, executor=executor)[:, :, grid.cols]
            for name, array in self.fields.items()
        }
        return Cube(
            source=self.source_id,
            hour=np.datetime64(hour, "h"),
            lat=grid.lat,
            lon=grid.lon,
            vertical=self.vertical.subset(positions),
            fields=fields,
            ps=ps,
        )

stats property

Transfer counters of every underlying byte store.

configure_decoded_cache(max_bytes)

Bound decoded Blosc reuse to final field chunks, using final_after.

Field views share this reader and its budget. Recent ERA5T chunks never qualify. Surface pressure in a separate reader has its own budget.

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

def configure_decoded_cache(self, max_bytes: int) -> None:
    """Bound decoded Blosc reuse to final field chunks, using ``final_after``.

    Field views share this reader and its budget. Recent ERA5T chunks never
    qualify. Surface pressure in a separate reader has its own budget.
    """
    if not isinstance(max_bytes, int) or isinstance(max_bytes, bool) or max_bytes < 0:
        raise ValueError("max_bytes must be a nonnegative integer")
    if not max_bytes:
        self.store.configure_decoded_cache(0, immutable=lambda key: False)
        return
    arrays = set(self._cache_fields)
    if self.surface_pressure is not None and self.surface_pressure.store is self.store:
        arrays.add(self.surface_pressure.array)
    self.store.configure_decoded_cache(
        max_bytes, immutable=_final_keys(self.store, arrays, self.final_after)
    )

coverage()

Published period, from the store's valid_time_start/valid_time_stop.

For ARCO-ERA5 the model-level store lags ERA5T (about three months).

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

def coverage(self) -> tuple[np.datetime64, np.datetime64] | None:
    """Published period, from the store's ``valid_time_start``/``valid_time_stop``.

    For ARCO-ERA5 the model-level store lags ERA5T (about three months).
    """
    attrs = self.store.attrs
    start, stop = attrs.get("valid_time_start"), attrs.get("valid_time_stop")
    if not start or not stop:
        return None
    begin = np.datetime64(start, "h")
    end = (
        np.datetime64(stop, "D") + np.timedelta64(23, "h")
        if len(stop) <= len("YYYY-MM-DD")
        else np.datetime64(stop, "h")
    )
    return begin, np.datetime64(end, "h")

cube_keys(t0)

Analyses at t0 and one hour later (no run: reanalysis).

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

def cube_keys(self, t0: np.datetime64) -> tuple[CubeKey, CubeKey]:
    """Analyses at ``t0`` and one hour later (no run: reanalysis)."""
    h = np.datetime64(t0, "h")
    return CubeKey(h), CubeKey(h + self.time_step)

fetch(key, domain, p_range, *, level_margin=1, executor=None)

Read one hour over domain with the levels bracketing p_range (Pa).

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

def fetch(
    self,
    key: CubeKey,
    domain: Domain,
    p_range: tuple[float, float],
    *,
    level_margin: int = 1,
    executor: Executor | None = None,
) -> Cube:
    """Read one hour over ``domain`` with the levels bracketing ``p_range`` (Pa)."""
    hour = key.valid
    try:
        t = self.time_index(hour)
    except KeyError as exc:
        raise DataUnavailableError(f"{self.source_id}: {hour} not in the store") from exc
    grid = GridIndex.locate(domain, self.lat, self.lon)
    ps: NDArray[np.float32] | None = None
    ps_lo = ps_hi = STANDARD_SURFACE_PRESSURE
    if self.surface_pressure is not None:
        sp = self.surface_pressure
        ps = sp.store.read_rows(sp.array, [(t,)], grid.rows, executor=executor)[0][:, grid.cols]
        if np.isfinite(ps).any():
            ps_lo, ps_hi = float(np.nanmin(ps)), float(np.nanmax(ps))
    positions = self.vertical.select_bracketing(
        p_range[0], p_range[1], ps_lo, ps_hi, level_margin
    )
    lead = [(t, int(k)) for k in self._store_index[positions]]
    fields = {
        name: self.store.read_rows(array, lead, grid.rows, executor=executor)[:, :, grid.cols]
        for name, array in self.fields.items()
    }
    return Cube(
        source=self.source_id,
        hour=np.datetime64(hour, "h"),
        lat=grid.lat,
        lon=grid.lon,
        vertical=self.vertical.subset(positions),
        fields=fields,
        ps=ps,
    )

select_fields(names)

Independent field view sharing stores and byte caches with this source.

Only listed output fields are fetched. Cache eligibility retains all original fields, so attaching a cache through a view cannot narrow the parent's cache policy. Configure shared stores before concurrent use.

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

def select_fields(self, names: Sequence[str]) -> GridSource:
    """Independent field view sharing stores and byte caches with this source.

    Only listed output fields are fetched. Cache eligibility retains all
    original fields, so attaching a cache through a view cannot narrow the
    parent's cache policy. Configure shared stores before concurrent use.
    """
    if not names or len(set(names)) != len(names):
        raise ValueError("fields must be a nonempty sequence of unique names")
    unknown = set(names) - self.fields.keys()
    if unknown:
        raise ValueError(f"unknown fields: {sorted(unknown)}")
    view = copy(self)
    view.fields = {name: self.fields[name] for name in names}
    view._tiling = None
    view._tiling_known = False
    return view

tiling(key)

Latitude bands of about one Blosc block (every longitude), the same for every hour.

Read once from the header of one fixed chunk (the first published hour), so every call and process probes the same, cacheable, few KiB.

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

def tiling(self, key: CubeKey) -> Tiling | None:
    """Latitude bands of about one Blosc block (every longitude), the same for every hour.

    Read once from the header of one fixed chunk (the first published hour),
    so every call and process probes the same, cacheable, few KiB.
    """
    if self._tiling_known:
        return self._tiling
    coverage = self.coverage()
    try:
        t = 0 if coverage is None else self.time_index(coverage[0])
    except KeyError:
        t = 0
    name = next(iter(self.fields.values()))
    rows = self.store.block_rows(name, [t, 0][: self._lead_dims(name)])
    if rows is not None and len(self.lat) > 1:
        step = rows * abs(float(self.lat[0] - self.lat[1]))
        bands = int(np.ceil(len(self.lat) / rows))
        self._tiling = Tiling(lat0=float(self.lat[0]) - bands * step, lat_step=step)
    self._tiling_known = True
    return self._tiling

time_index(hour)

Index of hour on the store's time axis, verified against the time array.

The axis is assumed hourly and contiguous; each time chunk is checked once (a few KiB) before its indices are trusted.

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

def time_index(self, hour: np.datetime64) -> int:
    """Index of ``hour`` on the store's time axis, verified against the time array.

    The axis is assumed hourly and contiguous; each time chunk is checked
    once (a few KiB) before its indices are trusted.
    """
    hours = int((np.datetime64(hour, "h") - self._epoch) / np.timedelta64(1, "h"))
    idx = hours - self._t0
    meta = self.store.array("time")
    if not 0 <= idx < meta.shape[0]:
        raise KeyError(f"{hour} outside the store's time axis")
    size = meta.chunks[0]
    chunk = idx // size
    with self._lock:
        known = chunk in self._checked_time_chunks
    if not known:
        start, stop = chunk * size, min((chunk + 1) * size, meta.shape[0])
        values = self.store.read_1d("time", start, stop)
        if not np.array_equal(values, np.arange(start, stop) + self._t0):
            raise RuntimeError(f"{self.source_id}: time axis is not hourly-contiguous")
        with self._lock:
            self._checked_time_chunks.add(chunk)
    return idx

use_cache(cache)

Keep the final hours' field chunks in cache (see the class docstring).

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

def use_cache(self, cache: RangeCache) -> None:
    """Keep the final hours' field chunks in ``cache`` (see the class docstring)."""
    arrays = {*self._cache_fields, "time"}  # past time chunks never change either
    stores = [(self.store, arrays)]
    if self.surface_pressure is not None:
        sp = self.surface_pressure
        if sp.store is self.store:
            arrays.add(sp.array)
        else:
            stores.append((sp.store, {sp.array}))
    for zarr, names in stores:
        inner = zarr.store.inner if isinstance(zarr.store, CachedStore) else zarr.store
        final = _final_keys(zarr, names, self.final_after)
        zarr.store = CachedStore(inner, cache, namespace_for(inner.location), final)

Tiling dataclass

Regular tiling of a grid: tile (i, j) starts at lat0 + i * lat_step.

lon_step=None means tiles span every longitude (ARCO-ERA5 bands). merge_adjacent=False groups points by their own tile, keeping long connected trajectories from requesting large mostly empty rectangles.

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

@dataclass(frozen=True)
class Tiling:
    """Regular tiling of a grid: tile ``(i, j)`` starts at ``lat0 + i * lat_step``.

    ``lon_step=None`` means tiles span every longitude (ARCO-ERA5 bands).
    ``merge_adjacent=False`` groups points by their own tile, keeping long
    connected trajectories from requesting large mostly empty rectangles.
    """

    lat0: float
    lat_step: float
    lon0: float = -180.0
    lon_step: float | None = None
    merge_adjacent: bool = True

    def __post_init__(self) -> None:
        if self.lat_step <= 0 or (self.lon_step is not None and self.lon_step <= 0):
            raise ValueError("tile steps must be positive")

    @property
    def n_cols(self) -> int:
        """Number of tile columns around the globe."""
        return 1 if self.lon_step is None else int(np.ceil(360.0 / self.lon_step - 1e-9))

n_cols property

Number of tile columns around the globe.

RangeCache

Immutable byte-range cache rooted at root (see the module docstring).

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

class RangeCache:
    """Immutable byte-range cache rooted at ``root`` (see the module docstring)."""

    def __init__(self, root: str | Path) -> None:
        self.root = Path(root).expanduser()
        self.stats = CacheStats()

    # -- layout ----------------------------------------------------------------------------
    def object_dir(self, namespace: str, key: str) -> Path:
        """Folder of one object of one namespace."""
        digest = hashlib.sha256(key.encode()).hexdigest()
        return self.root / namespace / digest[:2] / digest[2:24]

    # -- reads -----------------------------------------------------------------------------
    def get(self, namespace: str, key: str, start: int, stop: int, fetch: Fetch) -> bytes:
        """Bytes ``[start, stop)`` of ``key``, fetching only the missing sub-ranges.

        ``fetch(a, b)`` downloads ``[a, b)``; a shorter answer marks the end of
        the object, and the result is then truncated there like a range read.
        """
        folder = self.object_dir(namespace, key)
        data = self._cached(folder, start, stop)
        if data is not None:
            return data
        windows = range(start // _WINDOW, (max(stop, start + 1) - 1) // _WINDOW + 1)
        with _locked(folder, [f"w{w}" for w in windows]):
            data = self._cached(folder, start, stop)  # fetched meanwhile by another reader
            if data is not None:
                return data
            return self._fill(folder, key, start, stop, fetch)

    def _cached(self, folder: Path, start: int, stop: int) -> bytes | None:
        """Bytes ``[start, stop)`` if they are all on disk, else ``None``."""
        size = _read_size(folder)
        if size is not None:
            stop = min(stop, size)
        if stop <= start:
            return b""
        pieces = cover(_extents(folder), start, stop)
        if any(path is None for _, _, path in pieces):
            return None
        self.stats.add(hit=stop - start)
        return b"".join(_read_slice(p, a, b) for a, b, p in pieces if p is not None)

    def _fill(self, folder: Path, key: str, start: int, stop: int, fetch: Fetch) -> bytes:
        """Serve ``[start, stop)`` from disk and ``fetch``, keeping what was fetched."""
        size = _read_size(folder)
        if size is not None:
            stop = min(stop, size)
        pieces: list[bytes] = []
        hit = 0
        for a, b, path in cover(_extents(folder), start, stop):
            if path is not None:
                pieces.append(_read_slice(path, a, b))
                hit += b - a
                continue
            data = fetch(a, b)
            if len(data) > b - a:
                raise ValueError(f"{key}: asked for {b - a} bytes at {a}, got {len(data)}")
            self._store(folder, key, a, data)
            pieces.append(data)
            if len(data) < b - a:  # the object ends inside this gap
                write_atomic(folder / "size", str(a + len(data)).encode())
                break
        self.stats.add(hit=hit)
        return b"".join(pieces)

    def lookup(self, namespace: str, key: str, start: int, stop: int) -> bytes | None:
        """Bytes ``[start, stop)`` of ``key`` if they are all on disk, else ``None``.

        For callers that deduplicate their own concurrent fetches (the
        Open-Meteo page reader); they keep what they fetched with :meth:`put`.
        """
        return self._cached(self.object_dir(namespace, key), start, stop)

    def put(self, namespace: str, key: str, start: int, data: bytes) -> None:
        """Keep ``data``, the bytes of ``key`` from ``start`` on."""
        self._store(self.object_dir(namespace, key), key, start, data)

    def get_full(self, namespace: str, key: str, fetch_all: Callable[[], bytes]) -> bytes:
        """The whole object ``key``, fetched once."""
        folder = self.object_dir(namespace, key)
        data = self._whole(folder)
        if data is not None:
            return data
        with _locked(folder, ["full"]):
            data = self._whole(folder)
            if data is not None:
                return data
            data = fetch_all()
            self._store(folder, key, 0, data)
            write_atomic(folder / "size", str(len(data)).encode())
            return data

    def _whole(self, folder: Path) -> bytes | None:
        """The whole object if its size is known and all its bytes are on disk."""
        size = _read_size(folder)
        return None if size is None else self._cached(folder, 0, size)

    # -- writes ------------------------------------------------------------------------------
    def _store(self, folder: Path, key: str, start: int, data: bytes) -> None:
        if not data:
            return
        folder.mkdir(parents=True, exist_ok=True)
        label = folder / "key.txt"
        if not label.exists():
            write_atomic(label, key.encode())
        write_atomic(folder / f"{start}-{start + len(data)}.bin", data)
        self.stats.add(stored=len(data))

get(namespace, key, start, stop, fetch)

Bytes [start, stop) of key, fetching only the missing sub-ranges.

fetch(a, b) downloads [a, b); a shorter answer marks the end of the object, and the result is then truncated there like a range read.

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

def get(self, namespace: str, key: str, start: int, stop: int, fetch: Fetch) -> bytes:
    """Bytes ``[start, stop)`` of ``key``, fetching only the missing sub-ranges.

    ``fetch(a, b)`` downloads ``[a, b)``; a shorter answer marks the end of
    the object, and the result is then truncated there like a range read.
    """
    folder = self.object_dir(namespace, key)
    data = self._cached(folder, start, stop)
    if data is not None:
        return data
    windows = range(start // _WINDOW, (max(stop, start + 1) - 1) // _WINDOW + 1)
    with _locked(folder, [f"w{w}" for w in windows]):
        data = self._cached(folder, start, stop)  # fetched meanwhile by another reader
        if data is not None:
            return data
        return self._fill(folder, key, start, stop, fetch)

get_full(namespace, key, fetch_all)

The whole object key, fetched once.

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

def get_full(self, namespace: str, key: str, fetch_all: Callable[[], bytes]) -> bytes:
    """The whole object ``key``, fetched once."""
    folder = self.object_dir(namespace, key)
    data = self._whole(folder)
    if data is not None:
        return data
    with _locked(folder, ["full"]):
        data = self._whole(folder)
        if data is not None:
            return data
        data = fetch_all()
        self._store(folder, key, 0, data)
        write_atomic(folder / "size", str(len(data)).encode())
        return data

lookup(namespace, key, start, stop)

Bytes [start, stop) of key if they are all on disk, else None.

For callers that deduplicate their own concurrent fetches (the Open-Meteo page reader); they keep what they fetched with :meth:put.

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

def lookup(self, namespace: str, key: str, start: int, stop: int) -> bytes | None:
    """Bytes ``[start, stop)`` of ``key`` if they are all on disk, else ``None``.

    For callers that deduplicate their own concurrent fetches (the
    Open-Meteo page reader); they keep what they fetched with :meth:`put`.
    """
    return self._cached(self.object_dir(namespace, key), start, stop)

object_dir(namespace, key)

Folder of one object of one namespace.

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

def object_dir(self, namespace: str, key: str) -> Path:
    """Folder of one object of one namespace."""
    digest = hashlib.sha256(key.encode()).hexdigest()
    return self.root / namespace / digest[:2] / digest[2:24]

put(namespace, key, start, data)

Keep data, the bytes of key from start on.

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

def put(self, namespace: str, key: str, start: int, data: bytes) -> None:
    """Keep ``data``, the bytes of ``key`` from ``start`` on."""
    self._store(self.object_dir(namespace, key), key, start, data)

VerticalCoordinate dataclass

Full-level pressures A + B * ps for a set of native level identifiers.

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

@dataclass(frozen=True)
class VerticalCoordinate:
    """Full-level pressures ``A + B * ps`` for a set of native level identifiers."""

    kind: str  # "model" or "pressure"
    ids: NDArray[np.int64]  # native identifiers (hybrid index 1..137 or hPa)
    a: NDArray[np.float64]  # Pa
    b: NDArray[np.float64]  # dimensionless

    def __post_init__(self) -> None:
        if not (len(self.ids) == len(self.a) == len(self.b)):
            raise ValueError("ids, a and b must have the same length")

    def __len__(self) -> int:
        return len(self.ids)

    @property
    def needs_surface_pressure(self) -> bool:
        """True for hybrid levels (pressures depend on the surface pressure)."""
        return bool(np.any(self.b != 0.0))

    def pressure(self, ps: ArrayLike) -> NDArray[np.float64]:
        """Full-level pressure (Pa); ``ps`` broadcasts on a trailing level axis."""
        ps_arr = np.asarray(ps, dtype=np.float64)[..., None]
        return self.a + self.b * ps_arr

    def subset(self, positions: ArrayLike) -> VerticalCoordinate:
        """The levels at ``positions``."""
        pos = np.asarray(positions, dtype=np.int64)
        return VerticalCoordinate(self.kind, self.ids[pos], self.a[pos], self.b[pos])

    def select_bracketing(
        self, p_min: float, p_max: float, ps_min: float, ps_max: float, margin: int = 1
    ) -> NDArray[np.int64]:
        """Positions of the levels needed to bracket pressures in ``[p_min, p_max]``.

        The level pressures move with ``ps``: a level is kept if, for some
        ``ps`` in ``[ps_min, ps_max]``, it can be the level just above or just
        below a target pressure. ``margin`` extra levels are added on each side.
        """
        lo = self.pressure(ps_min)  # smallest pressure of each level (B >= 0)
        hi = self.pressure(ps_max)
        # first level that can lie at/above p_min, last that can lie at/below p_max
        top = int(np.searchsorted(hi, p_min, side="right")) - 1
        bottom = int(np.searchsorted(lo, p_max, side="left"))
        top = max(top - margin, 0)
        bottom = min(bottom + margin, len(self) - 1)
        return np.arange(top, bottom + 1, dtype=np.int64)

needs_surface_pressure property

True for hybrid levels (pressures depend on the surface pressure).

pressure(ps)

Full-level pressure (Pa); ps broadcasts on a trailing level axis.

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

def pressure(self, ps: ArrayLike) -> NDArray[np.float64]:
    """Full-level pressure (Pa); ``ps`` broadcasts on a trailing level axis."""
    ps_arr = np.asarray(ps, dtype=np.float64)[..., None]
    return self.a + self.b * ps_arr

select_bracketing(p_min, p_max, ps_min, ps_max, margin=1)

Positions of the levels needed to bracket pressures in [p_min, p_max].

The level pressures move with ps: a level is kept if, for some ps in [ps_min, ps_max], it can be the level just above or just below a target pressure. margin extra levels are added on each side.

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

def select_bracketing(
    self, p_min: float, p_max: float, ps_min: float, ps_max: float, margin: int = 1
) -> NDArray[np.int64]:
    """Positions of the levels needed to bracket pressures in ``[p_min, p_max]``.

    The level pressures move with ``ps``: a level is kept if, for some
    ``ps`` in ``[ps_min, ps_max]``, it can be the level just above or just
    below a target pressure. ``margin`` extra levels are added on each side.
    """
    lo = self.pressure(ps_min)  # smallest pressure of each level (B >= 0)
    hi = self.pressure(ps_max)
    # first level that can lie at/above p_min, last that can lie at/below p_max
    top = int(np.searchsorted(hi, p_min, side="right")) - 1
    bottom = int(np.searchsorted(lo, p_max, side="left"))
    top = max(top - margin, 0)
    bottom = min(bottom + margin, len(self) - 1)
    return np.arange(top, bottom + 1, dtype=np.int64)

subset(positions)

The levels at positions.

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

def subset(self, positions: ArrayLike) -> VerticalCoordinate:
    """The levels at ``positions``."""
    pos = np.asarray(positions, dtype=np.int64)
    return VerticalCoordinate(self.kind, self.ids[pos], self.a[pos], self.b[pos])

model_levels(ids=None)

ERA5 L137 full levels (default: all 137, hybrid index 1 = model top).

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

def model_levels(ids: ArrayLike | None = None) -> VerticalCoordinate:
    """ERA5 L137 full levels (default: all 137, hybrid index 1 = model top)."""
    a_half, b_half = _l137()
    all_ids = np.arange(1, 138, dtype=np.int64)
    a_full = 0.5 * (a_half[:-1] + a_half[1:])
    b_full = 0.5 * (b_half[:-1] + b_half[1:])
    coord = VerticalCoordinate("model", all_ids, a_full, b_full)
    if ids is None:
        return coord
    sel = np.asarray(ids, dtype=np.int64)
    if np.any((sel < 1) | (sel > N_MODEL_LEVELS)) or np.any(np.diff(sel) <= 0):
        raise ValueError("model level ids must be strictly increasing within 1..137")
    return coord.subset(sel - 1)

pressure_levels(hpa)

Isobaric levels given in hPa (must be strictly increasing).

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

def pressure_levels(hpa: ArrayLike) -> VerticalCoordinate:
    """Isobaric levels given in hPa (must be strictly increasing)."""
    lev = np.asarray(hpa, dtype=np.int64)
    if np.any(np.diff(lev) <= 0):
        raise ValueError("pressure levels must be strictly increasing (top to bottom)")
    return VerticalCoordinate("pressure", lev, lev.astype(np.float64) * 100.0, np.zeros(len(lev)))

sample(cube, lat, lon, pressure, options=None)

Interpolate every cube field at the given positions (pressure in Pa).

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

def sample(
    cube: Cube,
    lat: ArrayLike,
    lon: ArrayLike,
    pressure: ArrayLike,
    options: SampleOptions | None = None,
) -> SampleResult:
    """Interpolate every cube field at the given positions (pressure in Pa)."""
    la = np.asarray(lat, dtype=np.float64)
    lo = np.asarray(lon, dtype=np.float64)
    p = np.asarray(pressure, dtype=np.float64)
    n = la.size
    values = {name: np.full(n, np.nan) for name in cube.fields}
    flags = np.zeros(n, dtype=np.uint8)
    if n == 0:
        return values, flags
    if len(cube.vertical) < MIN_LEVELS:
        flags[:] = Flag.ABOVE_TOP | Flag.BELOW_BOTTOM
        return values, flags
    opts = options or SampleOptions()
    for start in range(0, n, _BATCH):
        sl = slice(start, min(start + _BATCH, n))
        v, f = _sample_batch(cube, la[sl], lo[sl], p[sl], opts)
        for name in values:
            values[name][sl] = v[name]
        flags[sl] = f
    return values, flags