Weather

weather

Validated weather grids and optional contrail-engine exports.

These containers do not download, extrapolate, derive cirrus optical depth or convert accumulated radiation. Coordinates are effective interpolation times in UTC nanoseconds; callers resolve source-specific time conventions first.

RadiationCube dataclass

TOA flux grid with explicitly resolved timestamps.

Arrays have dimensions (lon, lat, time). time_semantics must be valid (instantaneous flux) or interval_midpoint (mean flux assigned to its averaging-interval midpoint). Interval-end or accumulated radiation must be converted upstream. No engine export shifts these timestamps.

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

@dataclass(kw_only=True)
class RadiationCube:
    """TOA flux grid with explicitly resolved timestamps.

    Arrays have dimensions ``(lon, lat, time)``. ``time_semantics`` must be
    ``valid`` (instantaneous flux) or ``interval_midpoint`` (mean flux assigned
    to its averaging-interval midpoint). Interval-end or accumulated radiation
    must be converted upstream. No engine export shifts these timestamps.
    """

    lon: npt.ArrayLike
    lat: npt.ArrayLike
    time_ns: npt.ArrayLike
    fields: Mapping[str, npt.ArrayLike]
    units: Mapping[str, str]
    provenance: Mapping[str, str]
    time_semantics: str

    def __post_init__(self) -> None:
        """Validate axes, shape, units, and timestamp semantics."""
        self.lon = _axis(self.lon, "longitude")
        self.lat = _axis(self.lat, "latitude")
        if np.any(np.abs(self.lat) > _MAX_LATITUDE):
            raise ValueError("latitude must be in [-90, 90]")
        self.time_ns = _times(self.time_ns)
        if self.time_semantics not in {"valid", "interval_midpoint"}:
            raise ValueError("Resolve radiation times to valid or interval_midpoint first")
        self.fields = _fields(self.fields, self.units, self.shape, self.specs)
        self.units = {k: self.specs[k][1] for k in self.fields}
        self.provenance = dict(self.provenance)

    @property
    def specs(self) -> Mapping[str, tuple[str, str, str]]:
        """Recognized variables, canonical and engine units."""
        return RAD_FIELDS

    @property
    def shape(self) -> tuple[int, ...]:
        """Required array shape in export axis order."""
        return (np.size(self.lon), np.size(self.lat), np.size(self.time_ns))

    @property
    def coordinates(self) -> dict[str, npt.NDArray[Any]]:
        """Coordinates in native engine dimension order."""
        return {
            "longitude": np.asarray(self.lon),
            "latitude": np.asarray(self.lat),
            "time": np.asarray(self.time_ns).astype("datetime64[ns]"),
        }

    @property
    def coverage(self) -> dict[str, tuple[Any, Any]]:
        """Inclusive coordinate bounds, without implying extrapolation support."""
        return {k: (v[0], v[-1]) for k, v in self.coordinates.items()}

    def _require_complete(self) -> None:
        missing = set(RAD_FIELDS) - self.fields.keys()
        if missing:
            raise ValueError(f"Missing required radiation fields: {sorted(missing)}")

    def to_diffcocip(self) -> dict[str, Array]:
        """Export the NumPy dict accepted by diffcocip, without importing JAX.

        Every interpolation axis needs at least two points. The engine's time
        bridge uses seconds since 2000; grids before this epoch are rejected.
        Metadata stays available on this container, not in the engine dict.
        """
        self._require_complete()
        if min(self.shape) < _MIN_INTERPOLATION_POINTS:
            raise ValueError("diffcocip requires at least two points on every axis")
        time_ns = np.asarray(self.time_ns)
        if np.any(time_ns < _EPOCH_NS):
            raise ValueError("diffcocip requires timestamps on or after 2000-01-01")
        relative = time_ns - _EPOCH_NS
        out = {
            "lon": np.asarray(self.lon).copy(),
            "lat": np.asarray(self.lat).copy(),
            "time_s": relative / 1e9,
            "days_since_2000": relative / _DAY_NS,
            "hours_of_day": np.remainder(time_ns, _DAY_NS) / 3_600_000_000_000,
        }
        if "level" in self.coordinates:
            out["level"] = self.coordinates["level"].copy()
        out.update({self.specs[k][0]: np.asarray(v).copy() for k, v in self.fields.items()})
        return out

    def to_pycontrails(self, *, require_cocip: bool = True) -> Any:
        """Export an optional pycontrails MetDataset with resolved time metadata.

        Import pycontrails and xarray only on this call. Radiation's time
        coordinate is marked as resolved, preventing a second engine shift.
        Set ``require_cocip=False`` to export partial fields for preprocessing.
        """
        if require_cocip:
            self._require_complete()
        try:
            xr = importlib.import_module("xarray")
            pc = importlib.import_module("pycontrails")
        except ImportError as exc:
            raise ImportError("Install pycontrails and xarray to export MetDataset") from exc
        coords = self.coordinates
        dims = tuple(coords)
        data = {
            self.specs[k][0]: (dims, np.asarray(v).copy(), {"units": self.specs[k][2]})
            for k, v in self.fields.items()
        }
        ds = xr.Dataset(data, coords=coords, attrs=dict(self.provenance))
        if "level" not in coords:
            ds = ds.expand_dims(level=[-1])
            ds["time"].attrs["shift_radiation_time"] = "already resolved by caller"
        return pc.MetDataset(ds)

coordinates property

Coordinates in native engine dimension order.

coverage property

Inclusive coordinate bounds, without implying extrapolation support.

shape property

Required array shape in export axis order.

specs property

Recognized variables, canonical and engine units.

__post_init__()

Validate axes, shape, units, and timestamp semantics.

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

def __post_init__(self) -> None:
    """Validate axes, shape, units, and timestamp semantics."""
    self.lon = _axis(self.lon, "longitude")
    self.lat = _axis(self.lat, "latitude")
    if np.any(np.abs(self.lat) > _MAX_LATITUDE):
        raise ValueError("latitude must be in [-90, 90]")
    self.time_ns = _times(self.time_ns)
    if self.time_semantics not in {"valid", "interval_midpoint"}:
        raise ValueError("Resolve radiation times to valid or interval_midpoint first")
    self.fields = _fields(self.fields, self.units, self.shape, self.specs)
    self.units = {k: self.specs[k][1] for k in self.fields}
    self.provenance = dict(self.provenance)

to_diffcocip()

Export the NumPy dict accepted by diffcocip, without importing JAX.

Every interpolation axis needs at least two points. The engine's time bridge uses seconds since 2000; grids before this epoch are rejected. Metadata stays available on this container, not in the engine dict.

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

def to_diffcocip(self) -> dict[str, Array]:
    """Export the NumPy dict accepted by diffcocip, without importing JAX.

    Every interpolation axis needs at least two points. The engine's time
    bridge uses seconds since 2000; grids before this epoch are rejected.
    Metadata stays available on this container, not in the engine dict.
    """
    self._require_complete()
    if min(self.shape) < _MIN_INTERPOLATION_POINTS:
        raise ValueError("diffcocip requires at least two points on every axis")
    time_ns = np.asarray(self.time_ns)
    if np.any(time_ns < _EPOCH_NS):
        raise ValueError("diffcocip requires timestamps on or after 2000-01-01")
    relative = time_ns - _EPOCH_NS
    out = {
        "lon": np.asarray(self.lon).copy(),
        "lat": np.asarray(self.lat).copy(),
        "time_s": relative / 1e9,
        "days_since_2000": relative / _DAY_NS,
        "hours_of_day": np.remainder(time_ns, _DAY_NS) / 3_600_000_000_000,
    }
    if "level" in self.coordinates:
        out["level"] = self.coordinates["level"].copy()
    out.update({self.specs[k][0]: np.asarray(v).copy() for k, v in self.fields.items()})
    return out

to_pycontrails(*, require_cocip=True)

Export an optional pycontrails MetDataset with resolved time metadata.

Import pycontrails and xarray only on this call. Radiation's time coordinate is marked as resolved, preventing a second engine shift. Set require_cocip=False to export partial fields for preprocessing.

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

def to_pycontrails(self, *, require_cocip: bool = True) -> Any:
    """Export an optional pycontrails MetDataset with resolved time metadata.

    Import pycontrails and xarray only on this call. Radiation's time
    coordinate is marked as resolved, preventing a second engine shift.
    Set ``require_cocip=False`` to export partial fields for preprocessing.
    """
    if require_cocip:
        self._require_complete()
    try:
        xr = importlib.import_module("xarray")
        pc = importlib.import_module("pycontrails")
    except ImportError as exc:
        raise ImportError("Install pycontrails and xarray to export MetDataset") from exc
    coords = self.coordinates
    dims = tuple(coords)
    data = {
        self.specs[k][0]: (dims, np.asarray(v).copy(), {"units": self.specs[k][2]})
        for k, v in self.fields.items()
    }
    ds = xr.Dataset(data, coords=coords, attrs=dict(self.provenance))
    if "level" not in coords:
        ds = ds.expand_dims(level=[-1])
        ds["time"].attrs["shift_radiation_time"] = "already resolved by caller"
    return pc.MetDataset(ds)

WeatherCube dataclass

Bases: RadiationCube

Pressure-level grid with dimensions (lon, lat, level_hpa, time).

A cube may contain wind only; CoCiP exports require u, v, t, q, omega, tau_cirrus. Optical depth must be supplied after a correct full-column calculation; it cannot safely be derived from a cropped pressure slab.

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

@dataclass(kw_only=True)
class WeatherCube(RadiationCube):
    """Pressure-level grid with dimensions ``(lon, lat, level_hpa, time)``.

    A cube may contain wind only; CoCiP exports require ``u, v, t, q, omega,
    tau_cirrus``. Optical depth must be supplied after a correct full-column
    calculation; it cannot safely be derived from a cropped pressure slab.
    """

    level_hpa: npt.ArrayLike
    time_semantics: str = "valid"

    def __post_init__(self) -> None:
        """Validate the pressure axis and inherited grid contract."""
        self.level_hpa = _axis(self.level_hpa, "level_hpa")
        if np.any(np.asarray(self.level_hpa) <= 0):
            raise ValueError("level_hpa must be positive")
        if self.time_semantics != "valid":
            raise ValueError("Pressure-level weather must use valid timestamps")
        super().__post_init__()

    @property
    def specs(self) -> Mapping[str, tuple[str, str, str]]:
        """Pressure-level fields and units."""
        return MET_FIELDS

    @property
    def shape(self) -> tuple[int, ...]:
        """Required array shape in export axis order."""
        return (
            np.size(self.lon),
            np.size(self.lat),
            np.size(self.level_hpa),
            np.size(self.time_ns),
        )

    @property
    def coordinates(self) -> dict[str, npt.NDArray[Any]]:
        """Engine coordinates, including pressure in hPa."""
        return {
            "longitude": np.asarray(self.lon),
            "latitude": np.asarray(self.lat),
            "level": np.asarray(self.level_hpa),
            "time": np.asarray(self.time_ns).astype("datetime64[ns]"),
        }

    def _require_complete(self) -> None:
        missing = COCIP_FIELDS - self.fields.keys()
        if missing:
            raise ValueError(f"Missing required CoCiP fields: {sorted(missing)}")

coordinates property

Engine coordinates, including pressure in hPa.

shape property

Required array shape in export axis order.

specs property

Pressure-level fields and units.

__post_init__()

Validate the pressure axis and inherited grid contract.

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

def __post_init__(self) -> None:
    """Validate the pressure axis and inherited grid contract."""
    self.level_hpa = _axis(self.level_hpa, "level_hpa")
    if np.any(np.asarray(self.level_hpa) <= 0):
        raise ValueError("level_hpa must be positive")
    if self.time_semantics != "valid":
        raise ValueError("Pressure-level weather must use valid timestamps")
    super().__post_init__()