Skip to content

Wcs catalog

wcs_catalog

Strict geographic and discrete-axis metadata for Météo-France WCS.

CoverageGrid dataclass

Advertised grid centres and exact available forecast leads in seconds.

Source code in packages/barograph-met/src/baro/met/wcs_catalog.py
@dataclass(frozen=True)
class CoverageGrid:
    """Advertised grid centres and exact available forecast leads in seconds."""

    lat: NDArray[np.float64]
    lon: NDArray[np.float64]
    lead_seconds: tuple[int, ...]
    heights: tuple[float, ...]

parse_description(root)

Parse a single regular lon/lat surface coverage, rejecting unknown axes.

Météo-France documents time coefficients as forecast lead seconds and height coefficients as metres. No interpolation of missing leads is implied here.

Source code in packages/barograph-met/src/baro/met/wcs_catalog.py
def parse_description(root: Element) -> CoverageGrid:
    """Parse a single regular lon/lat surface coverage, rejecting unknown axes.

    Météo-France documents time coefficients as forecast lead seconds and height
    coefficients as metres. No interpolation of missing leads is implied here.
    """
    axes = _axes(root)
    envelope = _one(root, "EnvelopeWithTimePeriod")
    labels = envelope.attrib.get("axisLabels", "").split()
    if labels not in (["long", "lat", "time"], ["long", "lat", "height", "time"]):
        raise ValueError("Unsupported WCS envelope coordinate order")
    lower = _numbers(_one(envelope, "lowerCorner").text)
    upper = _numbers(_one(envelope, "upperCorner").text)
    if len(lower) != len(labels) - 1 or len(upper) != len(lower):
        raise ValueError("WCS envelope coordinate dimensions do not match")
    lon = _spatial(lower[0], upper[0], axes["long"][1][0])
    lat = _spatial(lower[1], upper[1], axes["lat"][1][0])[::-1].copy()
    leads = _discrete(axes["time"][0])
    if np.any(leads < 0) or np.any(leads != np.floor(leads)):
        raise ValueError("WCS forecast leads must be nonnegative integer seconds")
    heights = tuple(float(v) for v in _discrete(axes["height"][0])) if "height" in axes else ()
    return CoverageGrid(lat, lon, tuple(int(v) for v in leads), heights)

select_coverage(ids, field, run, step_seconds, override=None)

Select exact semantic names and run, preferring native interval products.

Plain names may represent cumulative precipitation; callers must inspect GRIB statistical metadata. No meaning or interval is inferred from substrings. Explicit overrides still require the requested run and catalog membership.

Source code in packages/barograph-met/src/baro/met/wcs_catalog.py
def select_coverage(
    ids: tuple[str, ...],
    field: str,
    run: np.datetime64,
    step_seconds: int,
    override: str | None = None,
) -> str:
    """Select exact semantic names and run, preferring native interval products.

    Plain names may represent cumulative precipitation; callers must inspect GRIB
    statistical metadata. No meaning or interval is inferred from substrings.
    Explicit overrides still require the requested run and catalog membership.
    """
    stamp = str(np.datetime_as_string(run, unit="s")).replace(":", ".") + "Z"
    if override is not None:
        if override not in ids or override.split("___")[-1].split("_")[0] != stamp:
            raise ValueError("WCS coverage override must exist and match the requested run")
        return override
    for candidate in _coverage_candidates(field, stamp, step_seconds):
        if candidate in ids:
            return candidate
    raise ValueError(f"No supported WCS coverage for {field} at {stamp}")