Wcs

wcs

Authenticated, geographically targeted Météo-France WCS 2.0.1 requests.

Credentials stay in headers and never enter cache keys. Only explicit, run-stamped GRIB coverages are persisted; the changing capabilities catalogue is not cached. Tokens may be callables so applications can manage OAuth renewal themselves.

WCSAuthenticationError

Bases: WCSRequestError

Credentials are absent, expired or not authorized for this API.

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

class WCSAuthenticationError(WCSRequestError):
    """Credentials are absent, expired or not authorized for this API."""

WCSClient

Small WCS client with credential-independent immutable response caching.

endpoint ends at the service name, before /GetCapabilities. auth='apikey' sends the token in the apikey header instead of Bearer. An injected HTTP client remains owned by its caller. No automatic retry is made for authentication failures. Transient HTTP failures retry twice by default, honoring Retry-After up to 60 seconds; longer delays raise rather than retry earlier than the server requested.

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

class WCSClient:
    """Small WCS client with credential-independent immutable response caching.

    ``endpoint`` ends at the service name, before ``/GetCapabilities``.
    ``auth='apikey'`` sends the token in the ``apikey`` header instead of Bearer.
    An injected HTTP client remains owned by its caller. No automatic retry is
    made for authentication failures. Transient HTTP failures retry twice by
    default, honoring Retry-After up to 60 seconds; longer delays raise rather
    than retry earlier than the server requested.
    """

    def __init__(  # noqa: PLR0913
        self,
        endpoint: str,
        token: str | Callable[[], str],
        *,
        client: httpx.Client | None = None,
        cache: RangeCache | None = None,
        auth: str = "bearer",
        retries: int = 2,
    ) -> None:
        url = httpx.URL(endpoint)
        if url.scheme != "https" or url.username or url.password or url.query or url.fragment:
            raise ValueError("WCS endpoint must be HTTPS without credentials, query or fragment")
        if auth not in {"bearer", "apikey"}:
            raise ValueError("auth must be 'bearer' or 'apikey'")
        if retries < 0:
            raise ValueError("retries must be nonnegative")
        self.retries = retries
        self.endpoint = endpoint.rstrip("/")
        self._token = token
        self._auth = auth
        self._client = client or httpx.Client(timeout=120.0, follow_redirects=False)
        self._owned = client is None
        self.cache = cache
        self.stats = TransferStats()
        self._namespace = namespace_for(self.endpoint)

    def _headers(self) -> dict[str, str]:
        token = self._token() if callable(self._token) else self._token
        if not token or not token.isascii() or not token.isprintable() or " " in token:
            raise WCSAuthenticationError("A valid nonempty Météo-France API token is required")
        return {"Authorization": f"Bearer {token}"} if self._auth == "bearer" else {"apikey": token}

    def _request(self, operation: str, params: list[tuple[str, str]]) -> bytes:
        for attempt in range(self.retries + 1):
            response = self._send(operation, params)
            if response.status_code in {401, 403}:
                raise WCSAuthenticationError(
                    f"WCS authentication failed (HTTP {response.status_code})"
                )
            if response.status_code in {429, 500, 502, 503, 504} and attempt < self.retries:
                delay = _retry_delay(response.headers.get("retry-after"), attempt)
                if delay <= 60:  # noqa: PLR2004
                    self.stats.throttle()
                    time.sleep(delay)
                    continue
            if not response.is_success:
                raise WCSRequestError(f"WCS request failed (HTTP {response.status_code})")
            return response.content
        raise AssertionError("unreachable")  # pragma: no cover

    def _send(self, operation: str, params: list[tuple[str, str]]) -> httpx.Response:
        headers = self._headers()
        started = time.monotonic()
        try:
            with slot(self.endpoint):
                response = self._client.get(
                    f"{self.endpoint}/{operation}",
                    params=tuple(params),
                    headers=headers,
                    follow_redirects=False,
                )
        except httpx.RequestError:
            raise WCSRequestError("WCS transport request failed") from None
        self.stats.add(len(response.content), time.monotonic() - started)
        return response

    @staticmethod
    def _params() -> list[tuple[str, str]]:
        return [("service", "WCS"), ("version", "2.0.1")]

    def capabilities(self) -> tuple[str, ...]:
        """Read the current catalogue without persisting its changing contents."""
        root = _xml(self._request("GetCapabilities", self._params()))
        return tuple(
            dict.fromkeys(
                node.text.strip()
                for node in root.iter()
                if node.tag.rsplit("}", 1)[-1] == "CoverageId" and node.text and node.text.strip()
            )
        )

    def describe(self, coverage_id: str) -> ET.Element:
        """Return the coverage metadata XML, including axis names and values."""
        return _xml(
            self._request("DescribeCoverage", [*self._params(), ("coverageid", coverage_id)])
        )

    def coverage(
        self,
        coverage_id: str,
        *,
        subsets: Sequence[str],
        validate: Callable[[bytes], None] | None = None,
    ) -> bytes:
        """Fetch one GRIB subset, retaining repeated ``subset`` query parameters.

        Optional ``validate`` runs before writing a new cache entry and on cache
        hits, allowing the caller to enforce field/run/interval metadata.
        """
        params = [*self._params(), ("coverageid", coverage_id), ("format", "application/wmo-grib")]
        params.extend(("subset", subset) for subset in sorted(subsets))
        validated = False

        def fetch() -> bytes:
            nonlocal validated
            data = self._request("GetCoverage", params)
            _grib(data)
            if validate is not None:
                validate(data)
                validated = True
            return data

        if self.cache is None or coverage_run(coverage_id) is None:
            return fetch()
        data = self.cache.get_full(self._namespace, urlencode(params), fetch)
        if validate is not None and not validated:
            validate(data)
        return data

    def close(self) -> None:
        """Close only the HTTP client created by this instance."""
        if self._owned:
            self._client.close()

    def __enter__(self) -> WCSClient:
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        traceback: TracebackType | None,
    ) -> None:
        self.close()

capabilities()

Read the current catalogue without persisting its changing contents.

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

def capabilities(self) -> tuple[str, ...]:
    """Read the current catalogue without persisting its changing contents."""
    root = _xml(self._request("GetCapabilities", self._params()))
    return tuple(
        dict.fromkeys(
            node.text.strip()
            for node in root.iter()
            if node.tag.rsplit("}", 1)[-1] == "CoverageId" and node.text and node.text.strip()
        )
    )

close()

Close only the HTTP client created by this instance.

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

def close(self) -> None:
    """Close only the HTTP client created by this instance."""
    if self._owned:
        self._client.close()

coverage(coverage_id, *, subsets, validate=None)

Fetch one GRIB subset, retaining repeated subset query parameters.

Optional validate runs before writing a new cache entry and on cache hits, allowing the caller to enforce field/run/interval metadata.

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

def coverage(
    self,
    coverage_id: str,
    *,
    subsets: Sequence[str],
    validate: Callable[[bytes], None] | None = None,
) -> bytes:
    """Fetch one GRIB subset, retaining repeated ``subset`` query parameters.

    Optional ``validate`` runs before writing a new cache entry and on cache
    hits, allowing the caller to enforce field/run/interval metadata.
    """
    params = [*self._params(), ("coverageid", coverage_id), ("format", "application/wmo-grib")]
    params.extend(("subset", subset) for subset in sorted(subsets))
    validated = False

    def fetch() -> bytes:
        nonlocal validated
        data = self._request("GetCoverage", params)
        _grib(data)
        if validate is not None:
            validate(data)
            validated = True
        return data

    if self.cache is None or coverage_run(coverage_id) is None:
        return fetch()
    data = self.cache.get_full(self._namespace, urlencode(params), fetch)
    if validate is not None and not validated:
        validate(data)
    return data

describe(coverage_id)

Return the coverage metadata XML, including axis names and values.

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

def describe(self, coverage_id: str) -> ET.Element:
    """Return the coverage metadata XML, including axis names and values."""
    return _xml(
        self._request("DescribeCoverage", [*self._params(), ("coverageid", coverage_id)])
    )

WCSRequestError

Bases: RuntimeError

An HTTP or OGC service error, excluding missing local credentials.

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

class WCSRequestError(RuntimeError):
    """An HTTP or OGC service error, excluding missing local credentials."""

coverage_run(coverage_id)

UTC initialization encoded in a Météo-France coverage identifier.

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

def coverage_run(coverage_id: str) -> np.datetime64 | None:
    """UTC initialization encoded in a Météo-France coverage identifier."""
    match = _RUN.search(coverage_id)
    if match is None:
        return None
    return np.datetime64(":".join(match.groups()), "ns")