Store

store

Minimal zarr-v2 reader that fetches only the bytes a regional read needs.

The ARCO-ERA5 0.25 deg stores keep a global latitude x longitude field in each chunk ([1, n_levels, 721, 1440]), so a small region never makes a chunk smaller. Chunks are Blosc frames split into independently compressed blocks of ~512 KiB, i.e. bands of ~90 latitude rows of a single level. This reader fetches the frame header, then only the block runs covering the requested (level, latitude band) slabs, with HTTP range requests.

ArrayMeta dataclass

Metadata of one zarr-v2 array.

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

@dataclass(frozen=True)
class ArrayMeta:
    """Metadata of one zarr-v2 array."""

    name: str
    shape: tuple[int, ...]
    chunks: tuple[int, ...]
    dtype: np.dtype[Any]
    compressor: dict[str, Any] | None
    fill_value: float | None
    separator: str
    dims: tuple[str, ...]
    attrs: dict[str, Any]

    def chunk_key(self, index: Sequence[int]) -> str:
        """Store key of the chunk at ``index``."""
        return f"{self.name}/" + self.separator.join(str(i) for i in index)

    @property
    def fill(self) -> float:
        """Fill value as a float (NaN when unset)."""
        return float("nan") if self.fill_value is None else float(self.fill_value)

fill property

Fill value as a float (NaN when unset).

chunk_key(index)

Store key of the chunk at index.

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

def chunk_key(self, index: Sequence[int]) -> str:
    """Store key of the chunk at ``index``."""
    return f"{self.name}/" + self.separator.join(str(i) for i in index)

ByteStore

Bases: Protocol

Key/value byte access with optional [start, stop) ranges.

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

class ByteStore(Protocol):
    """Key/value byte access with optional ``[start, stop)`` ranges."""

    stats: TransferStats
    location: str  # where the objects live (URL or directory), identifies the store

    def get(self, key: str, start: int | None = None, stop: int | None = None) -> bytes:
        """Object bytes, or the ``[start, stop)`` range of them."""
        ...

    def size(self, key: str) -> int:
        """Object size in bytes; raises :class:`MissingChunkError` if absent."""
        ...

get(key, start=None, stop=None)

Object bytes, or the [start, stop) range of them.

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

def get(self, key: str, start: int | None = None, stop: int | None = None) -> bytes:
    """Object bytes, or the ``[start, stop)`` range of them."""
    ...

size(key)

Object size in bytes; raises :class:MissingChunkError if absent.

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

def size(self, key: str) -> int:
    """Object size in bytes; raises :class:`MissingChunkError` if absent."""
    ...

FileStore

Local directory store (tests, mirrored stores).

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

class FileStore:
    """Local directory store (tests, mirrored stores)."""

    def __init__(self, root: str | Path) -> None:
        self.root = Path(root)
        self.location = str(self.root.resolve())
        self.stats = TransferStats()

    def get(self, key: str, start: int | None = None, stop: int | None = None) -> bytes:
        """Object bytes, or the ``[start, stop)`` range of them."""
        path = self.root / key
        t0 = time.perf_counter()
        try:
            with path.open("rb") as fh:
                if start is not None:
                    fh.seek(start)
                    data = fh.read(None if stop is None else stop - start)
                else:
                    data = fh.read()
        except FileNotFoundError as exc:
            raise MissingChunkError(key) from exc
        self.stats.add(len(data), time.perf_counter() - t0)
        return data

    def size(self, key: str) -> int:
        """Object size in bytes."""
        try:
            return (self.root / key).stat().st_size
        except FileNotFoundError as exc:
            raise MissingChunkError(key) from exc

get(key, start=None, stop=None)

Object bytes, or the [start, stop) range of them.

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

def get(self, key: str, start: int | None = None, stop: int | None = None) -> bytes:
    """Object bytes, or the ``[start, stop)`` range of them."""
    path = self.root / key
    t0 = time.perf_counter()
    try:
        with path.open("rb") as fh:
            if start is not None:
                fh.seek(start)
                data = fh.read(None if stop is None else stop - start)
            else:
                data = fh.read()
    except FileNotFoundError as exc:
        raise MissingChunkError(key) from exc
    self.stats.add(len(data), time.perf_counter() - t0)
    return data

size(key)

Object size in bytes.

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

def size(self, key: str) -> int:
    """Object size in bytes."""
    try:
        return (self.root / key).stat().st_size
    except FileNotFoundError as exc:
        raise MissingChunkError(key) from exc

HttpStore

Anonymous HTTP(S) access (honours HTTPS_PROXY), with retries.

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

class HttpStore:
    """Anonymous HTTP(S) access (honours ``HTTPS_PROXY``), with retries."""

    def __init__(  # noqa: PLR0913 - keyword-only HTTP client settings
        self,
        base_url: str,
        *,
        client: httpx.Client | None = None,
        retries: int = 4,
        backoff: float = 0.5,
        timeout: float = 120.0,
        max_connections: int = 64,
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.location = self.base_url
        limits = httpx.Limits(
            max_connections=max_connections, max_keepalive_connections=max_connections
        )
        self.client = client or httpx.Client(
            timeout=timeout,
            follow_redirects=True,
            limits=limits,
            headers={"User-Agent": USER_AGENT},
        )
        self.retries = retries
        self.backoff = backoff
        self.stats = TransferStats()
        self._base = self.base_url  # becomes the redirect target once one is observed

    def _url(self, key: str) -> str:
        return f"{self._base}/{key}"

    def _learn_redirect(self, resp: httpx.Response, key: str) -> None:
        """Remember a host redirect (e.g. data.gouv.fr -> object storage).

        Only a redirect that preserves the object path and carries no query
        (signed URLs expire) is learnt, so later requests go straight to the
        target instead of paying one extra round trip. If the learnt host later
        refuses a request, the original base URL is used again (see ``_request``).
        """
        if not resp.history or resp.url.query:
            return
        final = unquote(str(resp.url))  # targets may percent-encode ':'
        if final.endswith("/" + key):
            self._base = final[: -len(key) - 1]

    def get(self, key: str, start: int | None = None, stop: int | None = None) -> bytes:
        """Object bytes, or the ``[start, stop)`` range of them."""
        headers = {}
        if start is not None:
            headers["Range"] = f"bytes={start}-" + ("" if stop is None else str(stop - 1))
        t0 = time.perf_counter()
        resp = self._request("GET", key, headers)
        if start is not None and resp.status_code != HTTPStatus.PARTIAL_CONTENT:
            raise RuntimeError(f"server ignored the range request for {key}")
        self.stats.add(len(resp.content), time.perf_counter() - t0)
        return resp.content

    def size(self, key: str) -> int:
        """Object size in bytes (HEAD request)."""
        return int(self._request("HEAD", key, {}).headers["content-length"])

    def _request(self, method: str, key: str, headers: dict[str, str]) -> httpx.Response:
        """One request with retries: transport errors and 429/5xx back off, 404 raises."""
        for attempt in range(self.retries + 1):
            url = self._url(key)
            try:
                with slot(url):
                    resp = self.client.request(method, url, headers=headers)
            except httpx.TransportError:
                if attempt == self.retries:
                    raise
                time.sleep(self.backoff * 2**attempt)
                continue
            if self._base != self.base_url and resp.status_code in _FORGET_REDIRECT:
                self._base = self.base_url  # a learnt (temporary) redirect went stale
                continue
            if resp.status_code == HTTPStatus.NOT_FOUND:
                raise MissingChunkError(key)
            if resp.status_code in _RETRY_STATUS and attempt < self.retries:
                self._pushed_back(resp, url, attempt)
                continue
            resp.raise_for_status()
            self._learn_redirect(resp, key)
            return resp
        raise RuntimeError("unreachable")  # pragma: no cover

    def _pushed_back(self, resp: httpx.Response, url: str, attempt: int) -> None:
        """Record a 429/5xx, honour ``Retry-After`` when given, then back off."""
        self.stats.throttle()
        _log.warning(
            "HTTP %s from %s (attempt %d), backing off", resp.status_code, url, attempt + 1
        )
        retry_after = resp.headers.get("retry-after", "")
        delay = float(retry_after) if retry_after.isdigit() else self.backoff * 2**attempt
        time.sleep(min(delay, 60.0))

get(key, start=None, stop=None)

Object bytes, or the [start, stop) range of them.

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

def get(self, key: str, start: int | None = None, stop: int | None = None) -> bytes:
    """Object bytes, or the ``[start, stop)`` range of them."""
    headers = {}
    if start is not None:
        headers["Range"] = f"bytes={start}-" + ("" if stop is None else str(stop - 1))
    t0 = time.perf_counter()
    resp = self._request("GET", key, headers)
    if start is not None and resp.status_code != HTTPStatus.PARTIAL_CONTENT:
        raise RuntimeError(f"server ignored the range request for {key}")
    self.stats.add(len(resp.content), time.perf_counter() - t0)
    return resp.content

size(key)

Object size in bytes (HEAD request).

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

def size(self, key: str) -> int:
    """Object size in bytes (HEAD request)."""
    return int(self._request("HEAD", key, {}).headers["content-length"])

MissingChunkError

Bases: KeyError

The chunk does not exist in the store (reads as fill value).

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

class MissingChunkError(KeyError):
    """The chunk does not exist in the store (reads as fill value)."""

TransferStats dataclass

Thread-safe counters of what went over the wire.

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

@dataclass
class TransferStats:
    """Thread-safe counters of what went over the wire."""

    requests: int = 0
    bytes: int = 0
    seconds: float = 0.0
    throttled: int = 0  # 429/5xx answers that were retried (server pushing back)
    _lock: threading.Lock = field(default_factory=threading.Lock, repr=False, compare=False)

    def add(self, nbytes: int, seconds: float) -> None:
        """Account for one request of ``nbytes`` bytes taking ``seconds``."""
        with self._lock:
            self.requests += 1
            self.bytes += nbytes
            self.seconds += seconds

    def throttle(self) -> None:
        """Account for one 429/5xx answer."""
        with self._lock:
            self.throttled += 1

    def snapshot(self) -> dict[str, float]:
        """Current counters as a dict."""
        with self._lock:
            return {
                "requests": self.requests,
                "bytes": self.bytes,
                "seconds": self.seconds,
                "throttled": self.throttled,
            }

add(nbytes, seconds)

Account for one request of nbytes bytes taking seconds.

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

def add(self, nbytes: int, seconds: float) -> None:
    """Account for one request of ``nbytes`` bytes taking ``seconds``."""
    with self._lock:
        self.requests += 1
        self.bytes += nbytes
        self.seconds += seconds

snapshot()

Current counters as a dict.

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

def snapshot(self) -> dict[str, float]:
    """Current counters as a dict."""
    with self._lock:
        return {
            "requests": self.requests,
            "bytes": self.bytes,
            "seconds": self.seconds,
            "throttled": self.throttled,
        }

throttle()

Account for one 429/5xx answer.

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

def throttle(self) -> None:
    """Account for one 429/5xx answer."""
    with self._lock:
        self.throttled += 1

ZarrStore

Consolidated zarr-v2 store with region reads over a ByteStore.

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

class ZarrStore:
    """Consolidated zarr-v2 store with region reads over a ``ByteStore``."""

    def __init__(self, store: ByteStore, *, full_read_ratio: float = 0.8) -> None:
        self.store = store
        self.full_read_ratio = full_read_ratio
        meta = json.loads(store.get(".zmetadata"))["metadata"]
        self._meta = meta
        self._headers: Lru[str, BloscHeader | None] = Lru(_MAX_HEADERS)
        self._lock = threading.Lock()
        self._decoded = _DecodedRuns()

    def configure_decoded_cache(self, max_bytes: int, *, immutable: Callable[[str], bool]) -> None:
        """Retain exact decoded Blosc runs for explicitly immutable chunk keys.

        Zero disables retention. Reconfiguration clears entries. The predicate
        must identify immutable objects in this reader's fixed store; mutable
        forecast chunks must never qualify. Do not replace the backing store
        without clearing/reconfiguring this cache. Payload bytes alone are
        bounded, excluding headers, temporary decoding and caller arrays.
        """
        self._decoded.configure(max_bytes, immutable)

    def clear_decoded_cache(self) -> None:
        """Release retained decoded runs, preserving lifetime counters."""
        self._decoded.clear()

    def decoded_cache_info(self) -> dict[str, int]:
        """Retained payload and lifetime exact-run LRU counters."""
        return self._decoded.info()

    @property
    def attrs(self) -> dict[str, Any]:
        """Root attributes of the store."""
        return dict(self._meta.get(".zattrs", {}))

    def array(self, name: str) -> ArrayMeta:
        """Metadata of array ``name``."""
        try:
            za = self._meta[f"{name}/.zarray"]
        except KeyError as exc:
            raise KeyError(f"array {name!r} not in store") from exc
        attrs = self._meta.get(f"{name}/.zattrs", {})
        if za.get("filters"):
            raise NotImplementedError(f"{name}: zarr filters are not supported")
        if za.get("order", "C") != "C":
            raise NotImplementedError(f"{name}: only C order is supported")
        return ArrayMeta(
            name=name,
            shape=tuple(za["shape"]),
            chunks=tuple(za["chunks"]),
            dtype=np.dtype(za["dtype"]),
            compressor=za.get("compressor"),
            fill_value=_parse_fill(za.get("fill_value")),
            separator=za.get("dimension_separator", "."),
            dims=tuple(attrs.get("_ARRAY_DIMENSIONS", ())),
            attrs=attrs,
        )

    # -- whole-array reads (coordinates) ------------------------------------------------
    def read_1d(self, name: str, start: int = 0, stop: int | None = None) -> NDArray[Any]:
        """Values ``[start, stop)`` of a 1-D array (e.g. a coordinate)."""
        meta = self.array(name)
        if len(meta.shape) != 1:
            raise ValueError(f"{name} is not 1-D")
        stop = meta.shape[0] if stop is None else stop
        size = meta.chunks[0]
        parts = []
        for c in range(start // size, (stop - 1) // size + 1):
            chunk = self._decode_full(meta, (c,)).reshape(-1)
            lo, hi = max(start - c * size, 0), min(stop - c * size, size)
            parts.append(chunk[lo:hi])
        return np.concatenate(parts)

    def _decode_full(self, meta: ArrayMeta, index: tuple[int, ...]) -> NDArray[Any]:
        try:
            raw = self.store.get(meta.chunk_key(index))
        except MissingChunkError:
            return np.full(meta.chunks, meta.fill, dtype=meta.dtype)
        if meta.compressor is not None:
            raw = bytes(numcodecs.get_codec(meta.compressor).decode(raw))
        expected = int(np.prod(meta.chunks)) * meta.dtype.itemsize
        if len(raw) != expected:
            raise ValueError(
                f"{meta.chunk_key(index)}: {len(raw)} bytes decoded, expected {expected} "
                "(zarr v2 edge chunks must be padded to the full chunk shape)"
            )
        return np.frombuffer(raw, dtype=meta.dtype).reshape(meta.chunks)

    def block_rows(self, name: str, lead: Sequence[int]) -> int | None:
        """Latitude rows per Blosc block of ``name``, read from the chunk holding ``lead``.

        ``lead`` gives the leading (time, [level]) indices of an existing slab.
        Blocks do not align with level slabs, so this is the typical height of
        the latitude band one block holds, not an exact partition.
        """
        meta = self.array(name)
        nlead = len(meta.shape) - 2
        index = [int(i) // c for i, c in zip(lead, meta.chunks[:nlead], strict=True)]
        try:
            header = self._header(meta, meta.chunk_key([*index, 0, 0]))
        except MissingChunkError:
            return None
        if header is None:
            return None
        row = meta.shape[-1] * meta.dtype.itemsize
        return max(header.blocksize // row, 1)

    # -- regional reads -------------------------------------------------------------------
    def read_rows(
        self,
        name: str,
        lead: Sequence[Sequence[int]],
        rows: slice,
        *,
        executor: Executor | None = None,
    ) -> NDArray[np.float32]:
        """Read ``array[*lead_i, rows, :]`` for every leading index tuple in ``lead``.

        For an array ``(time, [level,] lat, lon)`` each ``lead_i`` gives the global
        ``(time, level)`` (or ``(time,)``) indices. The two trailing dimensions
        must be stored unchunked (true for ARCO-ERA5 0.25 deg). Every header and
        block-run request is independent and runs on ``executor`` if given.
        Returns ``float32`` of shape ``(len(lead), n_rows, n_lon)``.
        """
        meta = self.array(name)
        if meta.chunks[-2:] != meta.shape[-2:]:
            raise NotImplementedError(f"{name}: lat/lon must be unchunked")
        r0, r1, _ = rows.indices(meta.shape[-2])
        by_chunk = _plan_slabs(meta, lead, r0, r1)
        out = np.full((len(lead), r1 - r0, meta.shape[-1]), np.nan, dtype=np.float32)

        chunks = list(by_chunk)
        headers = _map(executor, lambda c: self._safe_header(meta, meta.chunk_key(c)), chunks)
        tasks: list[_Task] = []
        for cidx, header in zip(chunks, headers, strict=True):
            if header is _MISSING:
                for slab in by_chunk[cidx]:
                    out[slab.pos] = meta.fill
            else:
                tasks.extend(self._tasks_for(cidx, by_chunk[cidx], header))
        _map(executor, lambda t: self._execute(meta, t, by_chunk[t.chunk], out), tasks)
        return out

    def _tasks_for(self, cidx: ChunkIndex, slabs: list[_Slab], header: object) -> list[_Task]:
        if not isinstance(header, BloscHeader):
            return [_Task(cidx, None, None)]
        runs = _block_runs(header, [(sl.start, sl.stop) for sl in slabs])
        spans = [header.compressed_span(f, la) for f, la in runs]
        if sum(c1 - c0 for c0, c1 in spans) > self.full_read_ratio * header.cbytes:
            return [_Task(cidx, None, None)]
        return [_Task(cidx, header, run) for run in runs]

    def _execute(
        self, meta: ArrayMeta, task: _Task, slabs: list[_Slab], out: NDArray[np.float32]
    ) -> None:
        if task.header is None or task.run is None:
            data = self._decode_full(meta, task.chunk).tobytes()
            u0 = 0
        else:
            first, last = task.run
            header = task.header
            chunk = meta.chunk_key(task.chunk)

            def load() -> bytes:
                c0, c1 = header.compressed_span(first, last)
                return decode_run(header, first, last, self.store.get(chunk, c0, c1))

            data = self._decoded.get((chunk, task.run), load)
            u0 = header.block_span(first)[0]
        for slab in slabs:
            if u0 <= slab.start and slab.stop <= u0 + len(data):
                buf = data[slab.start - u0 : slab.stop - u0]
                out[slab.pos] = np.frombuffer(buf, dtype=meta.dtype).reshape(out.shape[1:])

    def _safe_header(self, meta: ArrayMeta, key: str) -> BloscHeader | object | None:
        try:
            return self._header(meta, key)
        except MissingChunkError:
            return _MISSING

    def _header(self, meta: ArrayMeta, key: str) -> BloscHeader | None:
        with self._lock:
            if key in self._headers:
                return self._headers[key]
        header: BloscHeader | None
        if meta.compressor is None or meta.compressor.get("id") != "blosc":
            header = None
        else:
            prefix = self.store.get(key, 0, _PREFIX_GUESS)
            try:
                need = header_prefix_size(prefix[:16])
                if need > len(prefix):
                    prefix = self.store.get(key, 0, need)
                header = parse_header(prefix)
            except UnsupportedFrameError:
                header = None
        with self._lock:
            self._headers[key] = header
        return header

attrs property

Root attributes of the store.

array(name)

Metadata of array name.

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

def array(self, name: str) -> ArrayMeta:
    """Metadata of array ``name``."""
    try:
        za = self._meta[f"{name}/.zarray"]
    except KeyError as exc:
        raise KeyError(f"array {name!r} not in store") from exc
    attrs = self._meta.get(f"{name}/.zattrs", {})
    if za.get("filters"):
        raise NotImplementedError(f"{name}: zarr filters are not supported")
    if za.get("order", "C") != "C":
        raise NotImplementedError(f"{name}: only C order is supported")
    return ArrayMeta(
        name=name,
        shape=tuple(za["shape"]),
        chunks=tuple(za["chunks"]),
        dtype=np.dtype(za["dtype"]),
        compressor=za.get("compressor"),
        fill_value=_parse_fill(za.get("fill_value")),
        separator=za.get("dimension_separator", "."),
        dims=tuple(attrs.get("_ARRAY_DIMENSIONS", ())),
        attrs=attrs,
    )

block_rows(name, lead)

Latitude rows per Blosc block of name, read from the chunk holding lead.

lead gives the leading (time, [level]) indices of an existing slab. Blocks do not align with level slabs, so this is the typical height of the latitude band one block holds, not an exact partition.

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

def block_rows(self, name: str, lead: Sequence[int]) -> int | None:
    """Latitude rows per Blosc block of ``name``, read from the chunk holding ``lead``.

    ``lead`` gives the leading (time, [level]) indices of an existing slab.
    Blocks do not align with level slabs, so this is the typical height of
    the latitude band one block holds, not an exact partition.
    """
    meta = self.array(name)
    nlead = len(meta.shape) - 2
    index = [int(i) // c for i, c in zip(lead, meta.chunks[:nlead], strict=True)]
    try:
        header = self._header(meta, meta.chunk_key([*index, 0, 0]))
    except MissingChunkError:
        return None
    if header is None:
        return None
    row = meta.shape[-1] * meta.dtype.itemsize
    return max(header.blocksize // row, 1)

clear_decoded_cache()

Release retained decoded runs, preserving lifetime counters.

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

def clear_decoded_cache(self) -> None:
    """Release retained decoded runs, preserving lifetime counters."""
    self._decoded.clear()

configure_decoded_cache(max_bytes, *, immutable)

Retain exact decoded Blosc runs for explicitly immutable chunk keys.

Zero disables retention. Reconfiguration clears entries. The predicate must identify immutable objects in this reader's fixed store; mutable forecast chunks must never qualify. Do not replace the backing store without clearing/reconfiguring this cache. Payload bytes alone are bounded, excluding headers, temporary decoding and caller arrays.

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

def configure_decoded_cache(self, max_bytes: int, *, immutable: Callable[[str], bool]) -> None:
    """Retain exact decoded Blosc runs for explicitly immutable chunk keys.

    Zero disables retention. Reconfiguration clears entries. The predicate
    must identify immutable objects in this reader's fixed store; mutable
    forecast chunks must never qualify. Do not replace the backing store
    without clearing/reconfiguring this cache. Payload bytes alone are
    bounded, excluding headers, temporary decoding and caller arrays.
    """
    self._decoded.configure(max_bytes, immutable)

decoded_cache_info()

Retained payload and lifetime exact-run LRU counters.

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

def decoded_cache_info(self) -> dict[str, int]:
    """Retained payload and lifetime exact-run LRU counters."""
    return self._decoded.info()

read_1d(name, start=0, stop=None)

Values [start, stop) of a 1-D array (e.g. a coordinate).

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

def read_1d(self, name: str, start: int = 0, stop: int | None = None) -> NDArray[Any]:
    """Values ``[start, stop)`` of a 1-D array (e.g. a coordinate)."""
    meta = self.array(name)
    if len(meta.shape) != 1:
        raise ValueError(f"{name} is not 1-D")
    stop = meta.shape[0] if stop is None else stop
    size = meta.chunks[0]
    parts = []
    for c in range(start // size, (stop - 1) // size + 1):
        chunk = self._decode_full(meta, (c,)).reshape(-1)
        lo, hi = max(start - c * size, 0), min(stop - c * size, size)
        parts.append(chunk[lo:hi])
    return np.concatenate(parts)

read_rows(name, lead, rows, *, executor=None)

Read array[*lead_i, rows, :] for every leading index tuple in lead.

For an array (time, [level,] lat, lon) each lead_i gives the global (time, level) (or (time,)) indices. The two trailing dimensions must be stored unchunked (true for ARCO-ERA5 0.25 deg). Every header and block-run request is independent and runs on executor if given. Returns float32 of shape (len(lead), n_rows, n_lon).

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

def read_rows(
    self,
    name: str,
    lead: Sequence[Sequence[int]],
    rows: slice,
    *,
    executor: Executor | None = None,
) -> NDArray[np.float32]:
    """Read ``array[*lead_i, rows, :]`` for every leading index tuple in ``lead``.

    For an array ``(time, [level,] lat, lon)`` each ``lead_i`` gives the global
    ``(time, level)`` (or ``(time,)``) indices. The two trailing dimensions
    must be stored unchunked (true for ARCO-ERA5 0.25 deg). Every header and
    block-run request is independent and runs on ``executor`` if given.
    Returns ``float32`` of shape ``(len(lead), n_rows, n_lon)``.
    """
    meta = self.array(name)
    if meta.chunks[-2:] != meta.shape[-2:]:
        raise NotImplementedError(f"{name}: lat/lon must be unchunked")
    r0, r1, _ = rows.indices(meta.shape[-2])
    by_chunk = _plan_slabs(meta, lead, r0, r1)
    out = np.full((len(lead), r1 - r0, meta.shape[-1]), np.nan, dtype=np.float32)

    chunks = list(by_chunk)
    headers = _map(executor, lambda c: self._safe_header(meta, meta.chunk_key(c)), chunks)
    tasks: list[_Task] = []
    for cidx, header in zip(chunks, headers, strict=True):
        if header is _MISSING:
            for slab in by_chunk[cidx]:
                out[slab.pos] = meta.fill
        else:
            tasks.extend(self._tasks_for(cidx, by_chunk[cidx], header))
    _map(executor, lambda t: self._execute(meta, t, by_chunk[t.chunk], out), tasks)
    return out

list_s3(base, prefix, *, dirs, stats=None)

Keys (dirs=False) or sub-folder prefixes (dirs=True) under prefix.

S3 ListObjectsV2 over HTTP, following continuation tokens (a page holds at most 1000 entries), under the per-host concurrency cap.

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

def list_s3(base: str, prefix: str, *, dirs: bool, stats: TransferStats | None = None) -> list[str]:
    """Keys (``dirs=False``) or sub-folder prefixes (``dirs=True``) under ``prefix``.

    S3 ``ListObjectsV2`` over HTTP, following continuation tokens (a page holds
    at most 1000 entries), under the per-host concurrency cap.
    """
    params = {"list-type": "2", "prefix": prefix, "max-keys": "1000"}
    if dirs:
        params["delimiter"] = "/"
    tag = _COMMON_PREFIX if dirs else _KEY
    names: list[str] = []
    url = base.rstrip("/") + "/"
    while True:
        t0 = time.perf_counter()
        with slot(url):
            resp = httpx.get(
                url,
                params=params,
                timeout=60.0,
                follow_redirects=True,
                headers={"User-Agent": USER_AGENT},
            )
        resp.raise_for_status()
        if stats is not None:
            stats.add(len(resp.content), time.perf_counter() - t0)
        names += tag.findall(resp.text)
        token = _NEXT_TOKEN.search(resp.text)
        if "<IsTruncated>true</IsTruncated>" not in resp.text or token is None:
            return names
        params["continuation-token"] = token.group(1)