Materialize

materialize

Build bounded native weather windows using Barograph's targeted range reader.

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)

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

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