terraflow.cmip6

CMIP6 NetCDF scenario ingest behind the optional [cmip6] install extra. The module opens climate scenario NetCDFs (typically daily tas or pr cubes), samples them at station coordinates, and returns the long-format station time-series that the terraflow.temporal aggregation engine consumes.

Install

pip install "terraflow-agro[cmip6]"

This installs xarray>=2024.1 and netcdf4>=1.7. They are imported lazily inside each callable so users on the historical CSV path never pay the import cost.

Overview

• cmip6_metadata(path) — SHA-256 + size + CMIP6 global attrs (variant_label, source_id, experiment_id, institution_id, table_id). Exposed as a public helper so callers can fold NetCDF provenance into their own manifest. The main terraflow run pipeline does not yet ingest NetCDFs directly — users today convert each NetCDF to a long-format station CSV via cmip6_to_station_timeseries(...) and the CSV's SHA-256 enters the run fingerprint via the existing _collect_input_paths path. First-class climate.cmip6_scenarios: config support (per-NetCDF SHA-256 + unit conversion + calendar handling inside the pipeline) ships in v0.6.0 — see issue #148.
• load_cmip6_scenario(path, variable, period) — opens a NetCDF lazily and slices the time axis by an inclusive (year_min, year_max) window. Handles non-Gregorian calendars (365_day, noleap, 360_day) via the calendar-aware dt.year path.
• extract_station_timeseries(da, stations, output_variable) — vectorised nearest-neighbour selection at station coordinates. Returns a DataFrame in the shape (station_id, lat, lon, date, <output_variable>).
• cmip6_to_station_timeseries(path, stations, ...) — one-call wrapper around the previous two.

Tolerates both lat/lon and latitude/longitude coord names commonly seen across CMIP6 sources.

Quick example

import pandas as pd
from terraflow.cmip6 import cmip6_to_station_timeseries

stations = pd.DataFrame({
    "station_id": ["KS01", "KS02", "KS03"],
    "lat":        [38.5,   39.0,   39.5],
    "lon":        [-100.5, -100.0, -99.5],
})

ts = cmip6_to_station_timeseries(
    "data/cmip6/tas_day_MPI-ESM1-2-HR_historical_19910101-20201231.nc",
    stations,
    period=(1991, 2020),
    output_variable="temperature_c",
)

API Reference

cmip6

CMIP6 NetCDF scenario ingest (flagship #138c).

Reads a CMIP6 scenario NetCDF (typically a CF-compliant daily-or-monthly temperature_c or precipitation_mm cube) and samples it at station coordinates, producing the long-format DataFrame shape that the :mod:terraflow.temporal aggregation engine consumes.

The module is gated behind the optional [cmip6] install extra so the default install stays lean::

pip install "terraflow-agro[cmip6]"

Imports xarray and netCDF4 lazily inside each callable so users who only run the historical CSV path never pay the import cost.

Expected NetCDF shape (CF convention): - Dimensions: time, lat / latitude, lon / longitude. - A single primary data variable (e.g. tas for near-surface temperature, pr for precipitation). - Global attributes optionally containing variant_label, source_id, and experiment_id (folded into the run fingerprint by the pipeline in #138e).

The module deliberately does NOT depend on terraflow.config — it is a thin adapter so the temporal engine can stay format-agnostic.

cmip6_metadata(path)

Extract CMIP6 provenance attributes from a NetCDF file.

Returns a dict with keys sha256 (file content hash), size_bytes, and a subset of variant_label, source_id, experiment_id, institution_id, table_id when those global attributes are set on the file.

The pipeline (in #138e) folds this dict into the run fingerprint so different CMIP6 variants of the same SSP scenario yield distinct cache directories.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to a CMIP6-conformant NetCDF file. The file is opened twice — once for SHA-256 hashing of bytes, once via xarray to read global attrs.	required

Source code in terraflow/cmip6.py

def cmip6_metadata(path: str | Path) -> dict[str, str]:
    """Extract CMIP6 provenance attributes from a NetCDF file.

    Returns a dict with keys ``sha256`` (file content hash),
    ``size_bytes``, and a subset of ``variant_label``, ``source_id``,
    ``experiment_id``, ``institution_id``, ``table_id`` when those
    global attributes are set on the file.

    The pipeline (in #138e) folds this dict into the run fingerprint
    so different CMIP6 variants of the same SSP scenario yield distinct
    cache directories.

    Parameters
    ----------
    path:
        Path to a CMIP6-conformant NetCDF file. The file is opened
        twice — once for SHA-256 hashing of bytes, once via xarray to
        read global attrs.
    """
    xr = _require_xarray()
    path = Path(path)
    if not path.is_file():
        raise FileNotFoundError(f"CMIP6 NetCDF file not found: {path}")

    digest = hashlib.sha256()
    size = 0
    with path.open("rb") as fh:
        while True:
            chunk = fh.read(1024 * 1024)
            if not chunk:
                break
            digest.update(chunk)
            size += len(chunk)

    metadata: dict[str, str] = {
        "sha256": digest.hexdigest(),
        "size_bytes": str(size),
    }
    with xr.open_dataset(path, decode_times=True) as ds:
        for key in _CMIP6_METADATA_KEYS:
            value = ds.attrs.get(key)
            if value is not None:
                metadata[key] = str(value)
    return metadata

cmip6_to_station_timeseries(path, stations, *, variable=None, period=None, output_variable)

Convenience: open a CMIP6 NetCDF and return a station time-series.

One-call wrapper around :func:load_cmip6_scenario + :func:extract_station_timeseries. Yields a DataFrame in the shape :mod:terraflow.temporal consumes.

Source code in terraflow/cmip6.py

def cmip6_to_station_timeseries(
    path: str | Path,
    stations: pd.DataFrame,
    *,
    variable: str | None = None,
    period: tuple[int, int] | None = None,
    output_variable: str,
) -> pd.DataFrame:
    """Convenience: open a CMIP6 NetCDF and return a station time-series.

    One-call wrapper around :func:`load_cmip6_scenario` +
    :func:`extract_station_timeseries`. Yields a DataFrame in the shape
    :mod:`terraflow.temporal` consumes.
    """
    da = load_cmip6_scenario(path, variable=variable, period=period)
    return extract_station_timeseries(da, stations, output_variable=output_variable)

extract_station_timeseries(da, stations, *, output_variable)

Sample a CMIP6 DataArray at each station's (lat, lon) coordinates.

Uses nearest-neighbour selection on the lat/lon coordinates. The output is a long-format DataFrame in the shape consumed by :mod:terraflow.temporal.

Parameters:

Name	Type	Description	Default
da	Any	An xarray DataArray with time, lat, lon dims (latitude / longitude variants tolerated).	required
stations	DataFrame	DataFrame with at minimum station_id, lat, lon.	required
output_variable	str	Name of the climate variable column in the output (e.g. temperature_c or precipitation_mm). Lets a caller line up the column with the schema expected by :func:terraflow.temporal.aggregate_per_station.	required

Returns:

Type	Description
pd.DataFrame:	Columns station_id, lat, lon, date, <output_variable>. One row per (station, time) pair.

Source code in terraflow/cmip6.py

def extract_station_timeseries(
    da: Any,
    stations: pd.DataFrame,
    *,
    output_variable: str,
) -> pd.DataFrame:
    """Sample a CMIP6 DataArray at each station's (lat, lon) coordinates.

    Uses nearest-neighbour selection on the lat/lon coordinates. The
    output is a long-format DataFrame in the shape consumed by
    :mod:`terraflow.temporal`.

    Parameters
    ----------
    da:
        An xarray DataArray with ``time, lat, lon`` dims (latitude /
        longitude variants tolerated).
    stations:
        DataFrame with at minimum ``station_id``, ``lat``, ``lon``.
    output_variable:
        Name of the climate variable column in the output (e.g.
        ``temperature_c`` or ``precipitation_mm``). Lets a caller line
        up the column with the schema expected by
        :func:`terraflow.temporal.aggregate_per_station`.

    Returns
    -------
    pd.DataFrame:
        Columns ``station_id, lat, lon, date, <output_variable>``. One
        row per (station, time) pair.
    """
    if STATION_ID_COL not in stations.columns:
        raise ValueError(
            f"stations DataFrame missing required column {STATION_ID_COL!r}; "
            f"got columns: {sorted(stations.columns)}"
        )
    for required in (LAT_COL, LON_COL):
        if required not in stations.columns:
            raise ValueError(
                f"stations DataFrame missing required column {required!r}; "
                f"got columns: {sorted(stations.columns)}"
            )

    lat_dim, lon_dim = _resolve_lat_lon_names(da)
    if "time" not in da.dims:
        raise ValueError(
            f"DataArray missing 'time' dimension; got dims {list(da.dims)}"
        )

    station_lats = stations[LAT_COL].to_numpy()
    station_lons = stations[LON_COL].to_numpy()
    station_ids = stations[STATION_ID_COL].to_numpy()

    # xarray supports vectorised nearest-neighbour selection by passing
    # DataArrays with a shared "station" dim — that returns a (time x
    # station) slab in one pass.
    xr = _require_xarray()
    sampled = da.sel(
        {
            lat_dim: xr.DataArray(station_lats, dims=("station",)),
            lon_dim: xr.DataArray(station_lons, dims=("station",)),
        },
        method="nearest",
    )

    # Materialise into a long-format DataFrame.
    times = _times_to_pandas(sampled["time"].to_numpy())
    arr = sampled.to_numpy()  # shape: (time, station)
    if arr.ndim == 1:
        # single-station case — xarray returns (time,)
        arr = arr.reshape(arr.shape[0], 1)

    n_time, n_stations = arr.shape
    out = pd.DataFrame(
        {
            STATION_ID_COL: np.repeat(station_ids, n_time),
            LAT_COL: np.repeat(station_lats, n_time),
            LON_COL: np.repeat(station_lons, n_time),
            DATE_COL: np.tile(times.to_numpy(), n_stations),
            output_variable: arr.T.reshape(-1),
        }
    )
    out[DATE_COL] = pd.to_datetime(out[DATE_COL])
    out[output_variable] = out[output_variable].astype(float)
    # Drop the station_dim coord — it confused round-tripping otherwise.
    return out.reset_index(drop=True)

load_cmip6_scenario(path, *, variable=None, period=None)

Open a CMIP6 NetCDF and return the requested DataArray (lazy).

The result is an xarray DataArray indexed by time, lat, lon (or time, latitude, longitude). The underlying data is loaded lazily — pulling a station's full time-series is what materialises values from disk.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to a CMIP6 NetCDF file.	required
variable	str | None	Data variable name to extract. Optional when the file contains exactly one variable.	None
period	tuple[int, int] | None	(year_min, year_max) inclusive year window. Slices the time axis. When None, the full time range is kept.	None

Source code in terraflow/cmip6.py

def load_cmip6_scenario(
    path: str | Path,
    *,
    variable: str | None = None,
    period: tuple[int, int] | None = None,
) -> Any:
    """Open a CMIP6 NetCDF and return the requested DataArray (lazy).

    The result is an xarray DataArray indexed by ``time, lat, lon``
    (or ``time, latitude, longitude``). The underlying data is loaded
    lazily — pulling a station's full time-series is what materialises
    values from disk.

    Parameters
    ----------
    path:
        Path to a CMIP6 NetCDF file.
    variable:
        Data variable name to extract. Optional when the file contains
        exactly one variable.
    period:
        ``(year_min, year_max)`` inclusive year window. Slices the
        ``time`` axis. When ``None``, the full time range is kept.
    """
    xr = _require_xarray()
    path = Path(path)
    if not path.is_file():
        raise FileNotFoundError(f"CMIP6 NetCDF file not found: {path}")

    ds = xr.open_dataset(path, decode_times=True)
    var_name = _resolve_data_variable(ds, variable)
    da = ds[var_name]
    if period is not None:
        year_min, year_max = period
        if "time" not in da.dims:
            raise ValueError(
                f"variable {var_name!r} has no 'time' dimension; cannot apply period filter"
            )
        years = da["time"].dt.year.to_numpy()
        mask = (years >= year_min) & (years <= year_max)
        da = da.isel(time=mask)
    return da