Openmeteo surface

openmeteo_surface

Surface weather from the selectively read Open-Meteo stitched archive.

The archive preserves valid times, not original run identities or issue times. Use a run-aware source to compare forecasts available before a cutoff. Native wind components are instantaneous model values at 10 m, not hourly means.

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},
    )