CMIP6 — API reference #

CMIP6 climate-projections data source subpackage — earthlens.cmip6. Background, the facet request shape, and the config + curated-vocabulary catalog are covered under the other pages in this section; this page is the rendered API.

`earthlens.cmip6` #

CMIP6 climate-projections backend (Pangeo ARCO mirror on gs://cmip6).

Exposes the raw, full CMIP6 archive — every ScenarioMIP / CMIP experiment, every ESM, on its native grid — as analysis-ready cloud Zarr on the open Pangeo Google Cloud mirror (gs://cmip6), indexed by a plain consolidated-stores CSV (no auth). This is the whole model x scenario x variable x member matrix, not the single pre-downscaled product the gee backend exposes (NASA/GDDP-CMIP6) nor the CHC-CMIP6 precipitation deltas the chc backend exposes.

A request is a CMIP6 facet tuple — source_id (model), experiment_id (scenario), variable_id, table_id (+ optional member_id / grid_label / version) — which the resolver maps to the matching zstore (gs://cmip6/...) URI(s). The backend is file-writing: download() has pyramids open the Zarr and write a bbox/time NetCDF subset, returning the list[Path]. earthlens never imports xarray / zarr / gcsfs — pyramids owns the read (via GDAL's /vsigs/ multidim driver, read anonymously; no gcsfs needed).

Public surface (re-exported from this package):

:class:CMIP6 — the backend; instantiate with a date window, a bbox, and a facet tuple (source_id / experiment_id / variable_id / table_id), then call :meth:CMIP6.download.
:class:Catalog — loader for the bundled cmip6_data_catalog.yaml (config + curated vocabulary).
:class:Cmip6Variable / :class:Experiment / :class:Table / :class:Source — one curated variable / experiment / table / source row.
:data:CATALOG_PATH — path to the bundled YAML; monkey-patchable in tests.
:func:clear_catalog_cache — empty the catalog parse cache.
:class:StoreResolver / :class:ResolvedStore — facet -> zstore resolution over the consolidated-stores CSV.

Examples:

Resolve a curated variable's metadata:

>>> from earthlens.cmip6 import Catalog
>>> Catalog().get_dataset("tas").long_name
'Near-surface (2 m) air temperature'

`CMIP6` #

Bases: AbstractDataSource

CMIP6 climate-projections backend (raw archive on gs://cmip6).

Wraps the open Pangeo CMIP6 ARCO mirror so a user pulls a model / scenario / variable / member subset of the raw CMIP6 archive through the same download() shape every other earthlens backend uses. The output is one gridded NetCDF per resolved store.

Attributes:

Name	Type	Description
`OUTPUT_KIND`	`OutputKind`	`"raster"` — the written artefacts are gridded NetCDFs.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/backend.py

class CMIP6(AbstractDataSource):
    """CMIP6 climate-projections backend (raw archive on `gs://cmip6`).

    Wraps the open Pangeo CMIP6 ARCO mirror so a user pulls a
    model / scenario / variable / member subset of the raw CMIP6 archive through
    the same `download()` shape every other earthlens backend uses. The output is
    one gridded NetCDF per resolved store.

    Attributes:
        OUTPUT_KIND: `"raster"` — the written artefacts are gridded NetCDFs.
    """

    OUTPUT_KIND: OutputKind = "raster"

    AGGREGATE_REFUSAL_REASON = "the backend writes gridded NetCDF subsets; reduce them separately with earthlens.aggregate.aggregate_netcdf"

    def __init__(
        self,
        start: str,
        end: str,
        *,
        source_id: str | None = None,
        experiment_id: str | None = None,
        variable_id: str | None = None,
        table_id: str | None = None,
        lat_lim: list[float] | None = None,
        lon_lim: list[float] | None = None,
        member_id: str | None = None,
        grid_label: str | None = None,
        version: str = "latest",
        activity_id: str | None = None,
        whole_time: bool = False,
        temporal_resolution: str = "monthly",
        path: Path | str | None = None,
        fmt: str = "%Y-%m-%d",
        catalog: Catalog | None = None,
        resolver: StoreResolver | None = None,
    ):
        """Initialise a CMIP6 backend instance.

        Args:
            start: Inclusive start of the date window (parsed with `fmt`).
            end: Inclusive end of the date window.
            source_id: Model key (`"CanESM5"`, `"GFDL-ESM4"`).
            experiment_id: Scenario / experiment (`"ssp585"`, `"historical"`).
            variable_id: The CMIP6 variable to fetch (`"tas"`, `"pr"`).
            table_id: The MIP table (`"Amon"`, `"day"`, `"Omon"`).
            lat_lim: `[lat_min, lat_max]` in degrees. A whole-Earth box
                (`[-90, 90]`, the default) is "no spatial subset" — a whole-grid
                download, warned; a narrower box crops the native grid.
            lon_lim: `[lon_min, lon_max]` in degrees. Defaults to `[-180, 180]`.
            member_id: Variant label; `None` uses the catalog default
                (`r1i1p1f1`).
            grid_label: Grid label (`"gn"`, `"gr"`); `None` fans out over the
                grids present for the other facets.
            version: `"latest"` (newest publication per store) or an explicit
                version string.
            activity_id: MIP the experiment belongs to; `None` leaves it
                unconstrained (inferred from the experiment).
            whole_time: Skip the date-window time subset and write the whole
                series (warned). Defaults to `False`.
            temporal_resolution: Advisory cadence label (fixed by `table_id`).
            path: Output directory for the written NetCDFs.
            fmt: `strptime` format for `start` / `end`.
            catalog: Optional pre-built :class:`Catalog`; defaults to the
                bundled catalog.
            resolver: Optional pre-built
                :class:`~earthlens.cmip6.resolver.StoreResolver`; defaults to one
                built from the catalog's CSV URL and facet columns.

        Raises:
            ValueError: If a required facet (`source_id` / `experiment_id` /
                `variable_id` / `table_id`) or a date bound is omitted or empty.
        """
        for name, value in (
            ("source_id", source_id),
            ("experiment_id", experiment_id),
            ("variable_id", variable_id),
            ("table_id", table_id),
        ):
            if not value:
                raise ValueError(f"CMIP6 requires a non-empty {name}.")
        if not start or not end:
            raise ValueError(
                "CMIP6 requires a start and end date, e.g. "
                "start='2050-01-01', end='2050-12-31'."
            )

        # The loop above raised on any empty required id; narrow for the type
        # checker so the downstream str-typed uses see non-optional values.
        assert source_id is not None
        assert experiment_id is not None
        assert variable_id is not None
        assert table_id is not None

        self._catalog = catalog if catalog is not None else Catalog()
        self._resolver = (
            resolver
            if resolver is not None
            else StoreResolver(self._catalog.csv_url, self._catalog.facet_columns)
        )
        self._source_id = source_id
        self._experiment_id = experiment_id
        self._variable_id = variable_id
        self._table_id = table_id
        self._member_id = member_id or self._catalog.default_member_id
        self._grid_label = grid_label
        self._version = version
        self._activity_id = activity_id
        self._whole_time = whole_time
        self._show_progress = True

        super().__init__(
            start=start,
            end=end,
            variables=[variable_id],
            temporal_resolution=temporal_resolution,
            lat_lim=[-90.0, 90.0] if lat_lim is None else lat_lim,
            lon_lim=[-180.0, 180.0] if lon_lim is None else lon_lim,
            fmt=fmt,
            path=path,
        )

    def _check_input_dates(
        self, start: str, end: str, temporal_resolution: str, fmt: str
    ) -> TemporalExtent:
        """Parse the `[start, end]` window into a :class:`TemporalExtent`.

        Args:
            start: Inclusive window start.
            end: Inclusive window end.
            temporal_resolution: Advisory cadence label.
            fmt: `strptime` format tried first for a string `start` /
                `end`; a non-matching string falls back to an ISO-8601
                parse, and a `datetime` / `date` ignores it.

        Returns:
            TemporalExtent: Frozen model with the parsed bounds.
        """
        return self._whole_window_extent(start, end, fmt=fmt, resolution="raw")

    def _wants_spatial_subset(self) -> bool:
        """Return whether the request narrows the grid (a bbox crop).

        Returns:
            bool: `True` when the bbox is narrower than whole-Earth.
        """
        return not (
            self.space.latitude_min <= -90.0
            and self.space.latitude_max >= 90.0
            and self.space.longitude_min <= -180.0
            and self.space.longitude_max >= 180.0
        )

    def _bbox(self) -> tuple[float, float, float, float] | None:
        """Return the request bbox as `(west, south, east, north)`, or `None`.

        Returns:
            tuple | None: The crop window, or `None` for a whole-grid request.
        """
        if not self._wants_spatial_subset():
            return None
        return (self.space.west, self.space.south, self.space.east, self.space.north)

    def _search(self) -> list[RemoteProduct]:
        """Resolve the facet tuple to one product per matching `zstore`.

        Returns:
            list[RemoteProduct]: One product per resolved store; each carries the
                `zstore` as `href` and the store's facets as `metadata`.

        Raises:
            ValueError: If no store matches (the resolver names the offending
                facet and lists the available values).
        """
        stores = self._resolver.resolve(
            source_id=self._source_id,
            experiment_id=self._experiment_id,
            variable_id=self._variable_id,
            table_id=self._table_id,
            member_id=self._member_id,
            grid_label=self._grid_label,
            version=self._version,
            activity_id=self._activity_id,
        )
        return [
            RemoteProduct(
                id=store.slug,
                href=store.zstore,
                metadata={"store": store},
            )
            for store in stores
        ]

    def _fetch(self, products: list[RemoteProduct]) -> list[Path]:
        """Write a bbox/time NetCDF subset for each resolved store.

        For each store: map the `[start, end]` window to an integer time-index
        range (unless `whole_time`), then read the gridded `(variable, time,
        bbox)` window through pyramids and write it to NetCDF.

        Args:
            products: The products from :meth:`_search`.

        Returns:
            list[Path]: One written NetCDF path per store, in order.
        """
        bbox = self._bbox()
        if bbox is None:
            logger.warning(
                f"cmip6: no bbox — writing the whole native grid for "
                f"{self._variable_id}/{self._experiment_id}; pass lat_lim/lon_lim "
                "to subset (CMIP6 stores can be large)."
            )
        out: list[Path] = []
        for product in tqdm(
            products, disable=not self._show_progress, desc="cmip6", unit="store"
        ):
            store: ResolvedStore = product.metadata["store"]
            time_sel = self._time_selector(store)
            stem = accessor.store_output_stem(
                store, self.time.start_date, self.time.end_date
            )
            out_path = self.root_dir / f"{stem}.nc"
            out.append(
                accessor.write_subset(
                    store.zstore,
                    self._variable_id,
                    bbox=bbox,
                    time=time_sel,
                    out_path=out_path,
                )
            )
        return out

    def _time_selector(
        self, store: ResolvedStore
    ) -> int | tuple[int, int] | slice | None:
        """Resolve the time selector for one store from the request window.

        Args:
            store: The resolved store to read.

        Returns:
            The integer time selector for :func:`accessor.write_subset` — a
            `(i0, i1)` index range for the date window, or `slice(None)` for a
            `whole_time` request.
        """
        if self._whole_time:
            return slice(None)
        return accessor.resolve_time_window(
            store.zstore,
            self._variable_id,
            self.time.start_date,
            self.time.end_date,
        )

    def terms_note(self) -> str:
        """Return the attribution note for the requested source model.

        Returns:
            str: The per-model `terms_note`, else the catalog default.
        """
        return self._catalog.terms_note(self._source_id)

    def download(
        self,
        progress_bar: bool = True,
    ) -> list[Path]:
        """Fetch the requested CMIP6 subset(s) and return the written paths.

        Runs the cheap :meth:`_search` (facet -> `zstore` resolution) then
        :meth:`_fetch`, which writes one bbox/time NetCDF subset per resolved
        store.

        Args:
            progress_bar: Show a per-store progress bar. Defaults to `True`.

        Returns:
            list[Path]: The written NetCDF paths, one per resolved store (never
                empty — a facet tuple that matches no store raises rather than
                returning an empty list).

        Raises:
            ValueError: If the facet tuple matches no store.
        """
        self._show_progress = progress_bar
        return self._api_via_search_fetch()

`init(start, end, *, source_id=None, experiment_id=None, variable_id=None, table_id=None, lat_lim=None, lon_lim=None, member_id=None, grid_label=None, version='latest', activity_id=None, whole_time=False, temporal_resolution='monthly', path=None, fmt='%Y-%m-%d', catalog=None, resolver=None)` #

Initialise a CMIP6 backend instance.

Parameters:

Name	Type	Description	Default
`start`	`str`	Inclusive start of the date window (parsed with `fmt`).	required
`end`	`str`	Inclusive end of the date window.	required
`source_id`	`str \| None`	Model key (`"CanESM5"`, `"GFDL-ESM4"`).	`None`
`experiment_id`	`str \| None`	Scenario / experiment (`"ssp585"`, `"historical"`).	`None`
`variable_id`	`str \| None`	The CMIP6 variable to fetch (`"tas"`, `"pr"`).	`None`
`table_id`	`str \| None`	The MIP table (`"Amon"`, `"day"`, `"Omon"`).	`None`
`lat_lim`	`list[float] \| None`	`[lat_min, lat_max]` in degrees. A whole-Earth box (`[-90, 90]`, the default) is "no spatial subset" — a whole-grid download, warned; a narrower box crops the native grid.	`None`
`lon_lim`	`list[float] \| None`	`[lon_min, lon_max]` in degrees. Defaults to `[-180, 180]`.	`None`
`member_id`	`str \| None`	Variant label; `None` uses the catalog default (`r1i1p1f1`).	`None`
`grid_label`	`str \| None`	Grid label (`"gn"`, `"gr"`); `None` fans out over the grids present for the other facets.	`None`
`version`	`str`	`"latest"` (newest publication per store) or an explicit version string.	`'latest'`
`activity_id`	`str \| None`	MIP the experiment belongs to; `None` leaves it unconstrained (inferred from the experiment).	`None`
`whole_time`	`bool`	Skip the date-window time subset and write the whole series (warned). Defaults to `False`.	`False`
`temporal_resolution`	`str`	Advisory cadence label (fixed by `table_id`).	`'monthly'`
`path`	`Path \| str \| None`	Output directory for the written NetCDFs.	`None`
`fmt`	`str`	`strptime` format for `start` / `end`.	`'%Y-%m-%d'`
`catalog`	`Catalog \| None`	Optional pre-built :class:`Catalog`; defaults to the bundled catalog.	`None`
`resolver`	`StoreResolver \| None`	Optional pre-built :class:`~earthlens.cmip6.resolver.StoreResolver`; defaults to one built from the catalog's CSV URL and facet columns.	`None`

Raises:

Type	Description
`ValueError`	If a required facet (`source_id` / `experiment_id` / `variable_id` / `table_id`) or a date bound is omitted or empty.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/backend.py

def __init__(
    self,
    start: str,
    end: str,
    *,
    source_id: str | None = None,
    experiment_id: str | None = None,
    variable_id: str | None = None,
    table_id: str | None = None,
    lat_lim: list[float] | None = None,
    lon_lim: list[float] | None = None,
    member_id: str | None = None,
    grid_label: str | None = None,
    version: str = "latest",
    activity_id: str | None = None,
    whole_time: bool = False,
    temporal_resolution: str = "monthly",
    path: Path | str | None = None,
    fmt: str = "%Y-%m-%d",
    catalog: Catalog | None = None,
    resolver: StoreResolver | None = None,
):
    """Initialise a CMIP6 backend instance.

    Args:
        start: Inclusive start of the date window (parsed with `fmt`).
        end: Inclusive end of the date window.
        source_id: Model key (`"CanESM5"`, `"GFDL-ESM4"`).
        experiment_id: Scenario / experiment (`"ssp585"`, `"historical"`).
        variable_id: The CMIP6 variable to fetch (`"tas"`, `"pr"`).
        table_id: The MIP table (`"Amon"`, `"day"`, `"Omon"`).
        lat_lim: `[lat_min, lat_max]` in degrees. A whole-Earth box
            (`[-90, 90]`, the default) is "no spatial subset" — a whole-grid
            download, warned; a narrower box crops the native grid.
        lon_lim: `[lon_min, lon_max]` in degrees. Defaults to `[-180, 180]`.
        member_id: Variant label; `None` uses the catalog default
            (`r1i1p1f1`).
        grid_label: Grid label (`"gn"`, `"gr"`); `None` fans out over the
            grids present for the other facets.
        version: `"latest"` (newest publication per store) or an explicit
            version string.
        activity_id: MIP the experiment belongs to; `None` leaves it
            unconstrained (inferred from the experiment).
        whole_time: Skip the date-window time subset and write the whole
            series (warned). Defaults to `False`.
        temporal_resolution: Advisory cadence label (fixed by `table_id`).
        path: Output directory for the written NetCDFs.
        fmt: `strptime` format for `start` / `end`.
        catalog: Optional pre-built :class:`Catalog`; defaults to the
            bundled catalog.
        resolver: Optional pre-built
            :class:`~earthlens.cmip6.resolver.StoreResolver`; defaults to one
            built from the catalog's CSV URL and facet columns.

    Raises:
        ValueError: If a required facet (`source_id` / `experiment_id` /
            `variable_id` / `table_id`) or a date bound is omitted or empty.
    """
    for name, value in (
        ("source_id", source_id),
        ("experiment_id", experiment_id),
        ("variable_id", variable_id),
        ("table_id", table_id),
    ):
        if not value:
            raise ValueError(f"CMIP6 requires a non-empty {name}.")
    if not start or not end:
        raise ValueError(
            "CMIP6 requires a start and end date, e.g. "
            "start='2050-01-01', end='2050-12-31'."
        )

    # The loop above raised on any empty required id; narrow for the type
    # checker so the downstream str-typed uses see non-optional values.
    assert source_id is not None
    assert experiment_id is not None
    assert variable_id is not None
    assert table_id is not None

    self._catalog = catalog if catalog is not None else Catalog()
    self._resolver = (
        resolver
        if resolver is not None
        else StoreResolver(self._catalog.csv_url, self._catalog.facet_columns)
    )
    self._source_id = source_id
    self._experiment_id = experiment_id
    self._variable_id = variable_id
    self._table_id = table_id
    self._member_id = member_id or self._catalog.default_member_id
    self._grid_label = grid_label
    self._version = version
    self._activity_id = activity_id
    self._whole_time = whole_time
    self._show_progress = True

    super().__init__(
        start=start,
        end=end,
        variables=[variable_id],
        temporal_resolution=temporal_resolution,
        lat_lim=[-90.0, 90.0] if lat_lim is None else lat_lim,
        lon_lim=[-180.0, 180.0] if lon_lim is None else lon_lim,
        fmt=fmt,
        path=path,
    )

`download(progress_bar=True)` #

Fetch the requested CMIP6 subset(s) and return the written paths.

Runs the cheap :meth:_search (facet -> zstore resolution) then :meth:_fetch, which writes one bbox/time NetCDF subset per resolved store.

Parameters:

Name	Type	Description	Default
`progress_bar`	`bool`	Show a per-store progress bar. Defaults to `True`.	`True`

Returns:

Type	Description
`list[Path]`	list[Path]: The written NetCDF paths, one per resolved store (never empty — a facet tuple that matches no store raises rather than returning an empty list).

Raises:

Type	Description
`ValueError`	If the facet tuple matches no store.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/backend.py

def download(
    self,
    progress_bar: bool = True,
) -> list[Path]:
    """Fetch the requested CMIP6 subset(s) and return the written paths.

    Runs the cheap :meth:`_search` (facet -> `zstore` resolution) then
    :meth:`_fetch`, which writes one bbox/time NetCDF subset per resolved
    store.

    Args:
        progress_bar: Show a per-store progress bar. Defaults to `True`.

    Returns:
        list[Path]: The written NetCDF paths, one per resolved store (never
            empty — a facet tuple that matches no store raises rather than
            returning an empty list).

    Raises:
        ValueError: If the facet tuple matches no store.
    """
    self._show_progress = progress_bar
    return self._api_via_search_fetch()

`terms_note()` #

Return the attribution note for the requested source model.

Returns:

Name	Type	Description
`str`	`str`	The per-model `terms_note`, else the catalog default.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/backend.py

def terms_note(self) -> str:
    """Return the attribution note for the requested source model.

    Returns:
        str: The per-model `terms_note`, else the catalog default.
    """
    return self._catalog.terms_note(self._source_id)

`Catalog` #

Bases: AbstractCatalog

Config + curated-vocabulary catalog for the CMIP6 backend.

Reads the bundled cmip6_data_catalog.yaml (shipped as package data) and exposes its variables: block as a map of :class:Cmip6Variable rows keyed by variable_id under the inherited :attr:datasets field, plus parallel :attr:experiments, :attr:tables, and :attr:sources maps and the resolution config (:attr:csv_url, :attr:bucket, :attr:facet_columns, :attr:default_member_id, :attr:default_version). Instantiate with no arguments (Catalog()); :func:model_post_init loads and validates the YAML in one pass and caches it by (path, mtime).

Attributes:

Name	Type	Description
`csv_url`	`str`	URL of the consolidated-stores CSV (the full per-store index).
`bucket`	`str`	The public GCS bucket the `zstore` URIs live on (`"cmip6"`).
`facet_columns`	`list[str]`	The CSV facet columns, in file order.
`default_member_id`	`str`	Member label applied when a request omits it.
`default_version`	`str`	Version-selection policy (`"latest"`).
`default_terms_note`	`str`	Attribution fallback for an uncurated source.
`datasets`	`dict[str, Cmip6Variable]`	Map from `variable_id` to its :class:`Cmip6Variable` row.
`experiments`	`dict[str, Experiment]`	Map from `experiment_id` to its :class:`Experiment` row.
`tables`	`dict[str, Table]`	Map from `table_id` to its :class:`Table` row.
`sources`	`dict[str, Source]`	Map from `source_id` to its :class:`Source` row.

Examples:

List curated variables and resolve one:

>>> from earthlens.cmip6 import Catalog
>>> cat = Catalog()
>>> "tas" in cat
True
>>> cat.get_dataset("tas").units
'K'
>>> cat.bucket
'cmip6'

An unknown variable raises with a did-you-mean hint:

>>> from earthlens.cmip6 import Catalog
>>> Catalog().get_dataset("rainfall")  # doctest: +ELLIPSIS
Traceback (most recent call last):
    ...
ValueError: 'rainfall' is not in the CMIP6 catalog. Known variables: [...].

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

class Catalog(AbstractCatalog):
    """Config + curated-vocabulary catalog for the CMIP6 backend.

    Reads the bundled `cmip6_data_catalog.yaml` (shipped as package data) and
    exposes its `variables:` block as a map of :class:`Cmip6Variable` rows keyed
    by `variable_id` under the inherited :attr:`datasets` field, plus parallel
    :attr:`experiments`, :attr:`tables`, and :attr:`sources` maps and the
    resolution config (:attr:`csv_url`, :attr:`bucket`, :attr:`facet_columns`,
    :attr:`default_member_id`, :attr:`default_version`). Instantiate with no
    arguments (`Catalog()`); :func:`model_post_init` loads and validates the YAML
    in one pass and caches it by `(path, mtime)`.

    Attributes:
        csv_url: URL of the consolidated-stores CSV (the full per-store index).
        bucket: The public GCS bucket the `zstore` URIs live on (`"cmip6"`).
        facet_columns: The CSV facet columns, in file order.
        default_member_id: Member label applied when a request omits it.
        default_version: Version-selection policy (`"latest"`).
        default_terms_note: Attribution fallback for an uncurated source.
        datasets: Map from `variable_id` to its :class:`Cmip6Variable` row.
        experiments: Map from `experiment_id` to its :class:`Experiment` row.
        tables: Map from `table_id` to its :class:`Table` row.
        sources: Map from `source_id` to its :class:`Source` row.

    Examples:
        - List curated variables and resolve one:
            ```python
            >>> from earthlens.cmip6 import Catalog
            >>> cat = Catalog()
            >>> "tas" in cat
            True
            >>> cat.get_dataset("tas").units
            'K'
            >>> cat.bucket
            'cmip6'

            ```
        - An unknown variable raises with a did-you-mean hint:
            ```python
            >>> from earthlens.cmip6 import Catalog
            >>> Catalog().get_dataset("rainfall")  # doctest: +ELLIPSIS
            Traceback (most recent call last):
                ...
            ValueError: 'rainfall' is not in the CMIP6 catalog. Known variables: [...].

            ```
    """

    _catalog_kind: str = "CMIP6 catalog"
    _entry_noun: str = "variables"

    csv_url: str = ""
    bucket: str = "cmip6"
    facet_columns: list[str] = Field(default_factory=list)
    default_member_id: str = "r1i1p1f1"
    default_version: str = "latest"
    default_terms_note: str = ""

    datasets: dict[str, Cmip6Variable] = Field(default_factory=dict)
    experiments: dict[str, Experiment] = Field(default_factory=dict)
    tables: dict[str, Table] = Field(default_factory=dict)
    sources: dict[str, Source] = Field(default_factory=dict)

    def model_post_init(self, __context: Any) -> None:
        """Auto-load the bundled catalog when no variables were supplied.

        `Catalog()` with no args reads :data:`CATALOG_PATH` (cached by
        `(path, mtime)`); passing `datasets=...` skips the disk read (used in
        tests). Either way the `available_datasets` index is derived from the
        loaded variable map.

        Raises:
            ValueError: Propagated from :meth:`load` when the YAML is missing,
                empty, or has a malformed row.
        """
        if not self.datasets and not self.csv_url:
            loaded = Catalog.load()
            self.csv_url = loaded.csv_url
            self.bucket = loaded.bucket
            self.facet_columns = loaded.facet_columns
            self.default_member_id = loaded.default_member_id
            self.default_version = loaded.default_version
            self.default_terms_note = loaded.default_terms_note
            self.datasets = loaded.datasets
            self.experiments = loaded.experiments
            self.tables = loaded.tables
            self.sources = loaded.sources
        self.available_datasets = sorted(self.datasets)

    @classmethod
    def load(cls, catalog_path: Path | None = None) -> Catalog:
        """Read and validate the CMIP6 catalog from disk (cached).

        Args:
            catalog_path: Path to the catalog YAML. Defaults to the module-level
                :data:`CATALOG_PATH`.

        Returns:
            A fully-populated :class:`Catalog`.

        Raises:
            ValueError: If `catalog_path` does not exist, or if the file is missing
                its `csv_url`, or any curated row fails validation.
        """
        path = catalog_path if catalog_path is not None else CATALOG_PATH
        payload = load_catalog(path, _CATALOG_CACHE, _parse_catalog, provider="CMIP6")
        return cls(**payload)

    @staticmethod
    def _parse_block(path: Path, block: Any, model: type[BaseModel]) -> dict[str, Any]:
        """Validate one YAML mapping block into `{key: model(...)}`.

        Args:
            path: Catalog path, for error messages.
            block: The raw mapping from the YAML (or `None` when absent).
            model: The pydantic row type to build.

        Returns:
            dict[str, Any]: The validated rows keyed by their YAML key.

        Raises:
            ValueError: If any row fails validation.
        """
        out: dict[str, Any] = {}
        for key, body in (block or {}).items():
            try:
                out[str(key)] = model(**dict(body or {}))
            except ValidationError as exc:
                raise ValueError(
                    f"{path} {model.__name__} {key!r} failed validation:\n{exc}"
                ) from exc
        return out

    def get_catalog(self) -> dict[str, Cmip6Variable]:
        """Return the curated variable map (satisfies the abstract contract).

        Returns:
            dict[str, Cmip6Variable]: Same object as :attr:`datasets`.
        """
        return self.datasets

    def get_experiment(self, key: str) -> Experiment:
        """Return the :class:`Experiment` for `key`, with a did-you-mean hint.

        Args:
            key: An `experiment_id` (`"ssp585"`, `"historical"`).

        Returns:
            Experiment: The matching experiment row.

        Raises:
            ValueError: If `key` is not a curated experiment.
        """
        return cast("Experiment", self._get_from(self.experiments, key, "experiment"))

    def get_table(self, key: str) -> Table:
        """Return the :class:`Table` for `key`, with a did-you-mean hint.

        Args:
            key: A `table_id` (`"Amon"`, `"day"`, `"Omon"`).

        Returns:
            Table: The matching table row.

        Raises:
            ValueError: If `key` is not a curated table.
        """
        return cast("Table", self._get_from(self.tables, key, "table"))

    def get_source(self, key: str) -> Source:
        """Return the :class:`Source` for `key`, with a did-you-mean hint.

        Args:
            key: A `source_id` (`"CanESM5"`, `"GFDL-ESM4"`).

        Returns:
            Source: The matching source row.

        Raises:
            ValueError: If `key` is not a curated source.
        """
        return cast("Source", self._get_from(self.sources, key, "source"))

    def terms_note(self, source_id: str) -> str:
        """Return the attribution note for `source_id`.

        Falls back to :attr:`default_terms_note` for an uncurated source (or a
        curated one with no per-model note).

        Args:
            source_id: The model key (`"CanESM5"`).

        Returns:
            str: The per-model `terms_note`, else the catalog default.
        """
        source = self.sources.get(source_id)
        if source is not None and source.terms_note:
            return source.terms_note
        return self.default_terms_note

    @staticmethod
    def _get_from(mapping: dict[str, Any], key: str, noun: str) -> Any:
        """Look up `key` in `mapping`, raising a did-you-mean `ValueError`.

        Args:
            mapping: The curated map to search.
            key: The requested key.
            noun: Singular noun for the error message (`"experiment"`).

        Returns:
            The matching row.

        Raises:
            ValueError: If `key` is absent.
        """
        try:
            return mapping[key]
        except KeyError:
            import difflib

            close = difflib.get_close_matches(key, mapping, n=1)
            hint = f" Did you mean {close[0]!r}?" if close else ""
            raise ValueError(
                f"{key!r} is not a curated CMIP6 {noun}. "
                f"Known {noun}s: {sorted(mapping)}.{hint}"
            ) from None

`get_catalog()` #

Return the curated variable map (satisfies the abstract contract).

Returns:

Type	Description
`dict[str, Cmip6Variable]`	dict[str, Cmip6Variable]: Same object as :attr:`datasets`.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

def get_catalog(self) -> dict[str, Cmip6Variable]:
    """Return the curated variable map (satisfies the abstract contract).

    Returns:
        dict[str, Cmip6Variable]: Same object as :attr:`datasets`.
    """
    return self.datasets

`get_experiment(key)` #

Return the :class:Experiment for key, with a did-you-mean hint.

Parameters:

Name	Type	Description	Default
`key`	`str`	An `experiment_id` (`"ssp585"`, `"historical"`).	required

Returns:

Name	Type	Description
`Experiment`	`Experiment`	The matching experiment row.

Raises:

Type	Description
`ValueError`	If `key` is not a curated experiment.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

def get_experiment(self, key: str) -> Experiment:
    """Return the :class:`Experiment` for `key`, with a did-you-mean hint.

    Args:
        key: An `experiment_id` (`"ssp585"`, `"historical"`).

    Returns:
        Experiment: The matching experiment row.

    Raises:
        ValueError: If `key` is not a curated experiment.
    """
    return cast("Experiment", self._get_from(self.experiments, key, "experiment"))

`get_source(key)` #

Return the :class:Source for key, with a did-you-mean hint.

Parameters:

Name	Type	Description	Default
`key`	`str`	A `source_id` (`"CanESM5"`, `"GFDL-ESM4"`).	required

Returns:

Name	Type	Description
`Source`	`Source`	The matching source row.

Raises:

Type	Description
`ValueError`	If `key` is not a curated source.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

def get_source(self, key: str) -> Source:
    """Return the :class:`Source` for `key`, with a did-you-mean hint.

    Args:
        key: A `source_id` (`"CanESM5"`, `"GFDL-ESM4"`).

    Returns:
        Source: The matching source row.

    Raises:
        ValueError: If `key` is not a curated source.
    """
    return cast("Source", self._get_from(self.sources, key, "source"))

`get_table(key)` #

Return the :class:Table for key, with a did-you-mean hint.

Parameters:

Name	Type	Description	Default
`key`	`str`	A `table_id` (`"Amon"`, `"day"`, `"Omon"`).	required

Returns:

Name	Type	Description
`Table`	`Table`	The matching table row.

Raises:

Type	Description
`ValueError`	If `key` is not a curated table.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

def get_table(self, key: str) -> Table:
    """Return the :class:`Table` for `key`, with a did-you-mean hint.

    Args:
        key: A `table_id` (`"Amon"`, `"day"`, `"Omon"`).

    Returns:
        Table: The matching table row.

    Raises:
        ValueError: If `key` is not a curated table.
    """
    return cast("Table", self._get_from(self.tables, key, "table"))

`load(catalog_path=None)` `classmethod` #

Read and validate the CMIP6 catalog from disk (cached).

Parameters:

Name	Type	Description	Default
`catalog_path`	`Path \| None`	Path to the catalog YAML. Defaults to the module-level :data:`CATALOG_PATH`.	`None`

Returns:

Type	Description
`Catalog`	A fully-populated :class:`Catalog`.

Raises:

Type	Description
`ValueError`	If `catalog_path` does not exist, or if the file is missing its `csv_url`, or any curated row fails validation.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

@classmethod
def load(cls, catalog_path: Path | None = None) -> Catalog:
    """Read and validate the CMIP6 catalog from disk (cached).

    Args:
        catalog_path: Path to the catalog YAML. Defaults to the module-level
            :data:`CATALOG_PATH`.

    Returns:
        A fully-populated :class:`Catalog`.

    Raises:
        ValueError: If `catalog_path` does not exist, or if the file is missing
            its `csv_url`, or any curated row fails validation.
    """
    path = catalog_path if catalog_path is not None else CATALOG_PATH
    payload = load_catalog(path, _CATALOG_CACHE, _parse_catalog, provider="CMIP6")
    return cls(**payload)

`model_post_init(__context)` #

Auto-load the bundled catalog when no variables were supplied.

Catalog() with no args reads :data:CATALOG_PATH (cached by (path, mtime)); passing datasets=... skips the disk read (used in tests). Either way the available_datasets index is derived from the loaded variable map.

Raises:

Type	Description
`ValueError`	Propagated from :meth:`load` when the YAML is missing, empty, or has a malformed row.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

def model_post_init(self, __context: Any) -> None:
    """Auto-load the bundled catalog when no variables were supplied.

    `Catalog()` with no args reads :data:`CATALOG_PATH` (cached by
    `(path, mtime)`); passing `datasets=...` skips the disk read (used in
    tests). Either way the `available_datasets` index is derived from the
    loaded variable map.

    Raises:
        ValueError: Propagated from :meth:`load` when the YAML is missing,
            empty, or has a malformed row.
    """
    if not self.datasets and not self.csv_url:
        loaded = Catalog.load()
        self.csv_url = loaded.csv_url
        self.bucket = loaded.bucket
        self.facet_columns = loaded.facet_columns
        self.default_member_id = loaded.default_member_id
        self.default_version = loaded.default_version
        self.default_terms_note = loaded.default_terms_note
        self.datasets = loaded.datasets
        self.experiments = loaded.experiments
        self.tables = loaded.tables
        self.sources = loaded.sources
    self.available_datasets = sorted(self.datasets)

`terms_note(source_id)` #

Return the attribution note for source_id.

Falls back to :attr:default_terms_note for an uncurated source (or a curated one with no per-model note).

Parameters:

Name	Type	Description	Default
`source_id`	`str`	The model key (`"CanESM5"`).	required

Returns:

Name	Type	Description
`str`	`str`	The per-model `terms_note`, else the catalog default.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

def terms_note(self, source_id: str) -> str:
    """Return the attribution note for `source_id`.

    Falls back to :attr:`default_terms_note` for an uncurated source (or a
    curated one with no per-model note).

    Args:
        source_id: The model key (`"CanESM5"`).

    Returns:
        str: The per-model `terms_note`, else the catalog default.
    """
    source = self.sources.get(source_id)
    if source is not None and source.terms_note:
        return source.terms_note
    return self.default_terms_note

`Cmip6Variable` #

Bases: BaseModel

One curated CMIP6 variable row (the variable_id leaf).

A frozen value object with descriptive metadata only — CMIP6 variables carry no request-shaping parameters; the facet tuple selects the store and pyramids reads the array. Curated rows are optional: an uncurated variable_id still resolves against the CSV, it just lacks these labels.

Attributes:

Name	Type	Description
`units`	`str`	CMIP6 CMOR unit (`"K"`, `"kg m-2 s-1"`, `"1"` for a dimensionless fraction).
`long_name`	`str`	Human-readable description used in docs and logs.
`realm`	`str`	Modelling realm the variable belongs to (`"atmos"`, `"ocean"`, `"land"`, `"seaIce"`, ...).

Examples:

Build a variable row directly:

>>> from earthlens.cmip6 import Cmip6Variable
>>> v = Cmip6Variable(units="K", long_name="Near-surface air temperature")
>>> v.units
'K'

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

class Cmip6Variable(BaseModel):
    """One curated CMIP6 variable row (the `variable_id` leaf).

    A frozen value object with descriptive metadata only — CMIP6 variables carry
    no request-shaping parameters; the facet tuple selects the store and pyramids
    reads the array. Curated rows are optional: an uncurated `variable_id` still
    resolves against the CSV, it just lacks these labels.

    Attributes:
        units: CMIP6 CMOR unit (`"K"`, `"kg m-2 s-1"`, `"1"` for a
            dimensionless fraction).
        long_name: Human-readable description used in docs and logs.
        realm: Modelling realm the variable belongs to (`"atmos"`, `"ocean"`,
            `"land"`, `"seaIce"`, ...).

    Examples:
        - Build a variable row directly:
            ```python
            >>> from earthlens.cmip6 import Cmip6Variable
            >>> v = Cmip6Variable(units="K", long_name="Near-surface air temperature")
            >>> v.units
            'K'

            ```
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    units: str = ""
    long_name: str = ""
    realm: str = ""

`Experiment` #

Bases: BaseModel

One curated CMIP6 experiment (scenario / diagnostic) row.

Attributes:

Name	Type	Description
`activity_id`	`str`	The MIP the experiment belongs to (`"CMIP"` for the DECK / historical runs, `"ScenarioMIP"` for the SSPs).
`description`	`str`	Human-readable summary.

Examples:

Inspect an experiment's activity:

>>> from earthlens.cmip6 import Catalog
>>> Catalog().get_experiment("ssp585").activity_id
'ScenarioMIP'

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

class Experiment(BaseModel):
    """One curated CMIP6 experiment (scenario / diagnostic) row.

    Attributes:
        activity_id: The MIP the experiment belongs to (`"CMIP"` for the
            DECK / historical runs, `"ScenarioMIP"` for the SSPs).
        description: Human-readable summary.

    Examples:
        - Inspect an experiment's activity:
            ```python
            >>> from earthlens.cmip6 import Catalog
            >>> Catalog().get_experiment("ssp585").activity_id
            'ScenarioMIP'

            ```
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    activity_id: str = ""
    description: str = ""

`ResolvedStore` `dataclass` #

One CMIP6 Zarr store resolved from a facet tuple.

Carries the zstore URI plus the facet values that identify it, so the backend can name the output file and log the provenance without re-querying the CSV.

Attributes:

Name	Type	Description
`zstore`	`str`	The `gs://cmip6/...` store URI (ends in `/`).
`source_id`	`str`	Model that produced the store.
`experiment_id`	`str`	Scenario / diagnostic experiment.
`variable_id`	`str`	The CMIP6 variable.
`table_id`	`str`	The MIP table (realm x cadence).
`member_id`	`str`	The variant label (`r1i1p1f1`).
`grid_label`	`str`	Grid label (`gn` native, `gr` regridded, ...).
`version`	`str`	Data-publication version (an integer date, as a string).
`activity_id`	`str`	The MIP the experiment belongs to.

Examples:

Build one directly:

>>> from earthlens.cmip6.resolver import ResolvedStore
>>> s = ResolvedStore(
...     zstore="gs://cmip6/CMIP6/ScenarioMIP/.../tas/gn/v20190101/",
...     source_id="CanESM5", experiment_id="ssp585", variable_id="tas",
...     table_id="Amon", member_id="r1i1p1f1", grid_label="gn",
...     version="20190101", activity_id="ScenarioMIP",
... )
>>> s.variable_id
'tas'

Source code in libs/providers/atmosphere/src/earthlens/cmip6/resolver.py

@dataclass(frozen=True)
class ResolvedStore:
    """One CMIP6 Zarr store resolved from a facet tuple.

    Carries the `zstore` URI plus the facet values that identify it, so the
    backend can name the output file and log the provenance without re-querying
    the CSV.

    Attributes:
        zstore: The `gs://cmip6/...` store URI (ends in `/`).
        source_id: Model that produced the store.
        experiment_id: Scenario / diagnostic experiment.
        variable_id: The CMIP6 variable.
        table_id: The MIP table (realm x cadence).
        member_id: The variant label (`r1i1p1f1`).
        grid_label: Grid label (`gn` native, `gr` regridded, ...).
        version: Data-publication version (an integer date, as a string).
        activity_id: The MIP the experiment belongs to.

    Examples:
        - Build one directly:
            ```python
            >>> from earthlens.cmip6.resolver import ResolvedStore
            >>> s = ResolvedStore(
            ...     zstore="gs://cmip6/CMIP6/ScenarioMIP/.../tas/gn/v20190101/",
            ...     source_id="CanESM5", experiment_id="ssp585", variable_id="tas",
            ...     table_id="Amon", member_id="r1i1p1f1", grid_label="gn",
            ...     version="20190101", activity_id="ScenarioMIP",
            ... )
            >>> s.variable_id
            'tas'

            ```
    """

    zstore: str
    source_id: str
    experiment_id: str
    variable_id: str
    table_id: str
    member_id: str
    grid_label: str
    version: str
    activity_id: str = ""

    @property
    def slug(self) -> str:
        """A filesystem-safe stem identifying this store.

        Returns:
            str: `<source>_<experiment>_<variable>_<table>_<member>_<grid>` with
                any path separators removed.
        """
        parts = [
            self.source_id,
            self.experiment_id,
            self.variable_id,
            self.table_id,
            self.member_id,
            self.grid_label,
        ]
        return "_".join(str(p).replace("/", "-") for p in parts if p)

`slug` `property` #

A filesystem-safe stem identifying this store.

Returns:

Name	Type	Description
`str`	`str`	`<source>_<experiment>_<variable>_<table>_<member>_<grid>` with any path separators removed.

`Source` #

Bases: BaseModel

One curated CMIP6 source-model (GCM) row.

Attributes:

Name	Type	Description
`institution_id`	`str`	The modelling centre that produced the model.
`terms_note`	`str`	Any per-model licence / attribution nuance (most CMIP6 models are CC BY 4.0, cited via the source GCM).
`description`	`str`	Optional human-readable summary.

Examples:

Read a source's institution:

>>> from earthlens.cmip6 import Catalog
>>> Catalog().get_source("CanESM5").institution_id
'CCCma'

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

class Source(BaseModel):
    """One curated CMIP6 source-model (GCM) row.

    Attributes:
        institution_id: The modelling centre that produced the model.
        terms_note: Any per-model licence / attribution nuance (most CMIP6
            models are CC BY 4.0, cited via the source GCM).
        description: Optional human-readable summary.

    Examples:
        - Read a source's institution:
            ```python
            >>> from earthlens.cmip6 import Catalog
            >>> Catalog().get_source("CanESM5").institution_id
            'CCCma'

            ```
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    institution_id: str = ""
    terms_note: str = ""
    description: str = ""

`StoreResolver` #

Resolve CMIP6 facet tuples to zstore URIs over the consolidated CSV.

Fetches + caches the CSV once, then filters it per request. Construct with the catalog's csv_url + facet_columns; inject a frame= or cache_path= to run offline.

Parameters:

Name	Type	Description	Default
`csv_url`	`str`	URL of the consolidated-stores CSV.	required
`facet_columns`	`list[str]`	The CSV facet column names (from the catalog), kept as schema documentation; `resolve()` filters on :data:`_FILTER_FACETS`, not on this list.	required
`cache_path`	`Path \| str \| None`	Where to cache the downloaded CSV. Defaults to :func:`default_cache_path`.	`None`
`frame`	`DataFrame \| None`	A pre-loaded `DataFrame` to use verbatim, skipping all I/O.	`None`
`timeout`	`float`	Per-request network timeout, in seconds.	`120.0`

Source code in libs/providers/atmosphere/src/earthlens/cmip6/resolver.py

class StoreResolver:
    """Resolve CMIP6 facet tuples to `zstore` URIs over the consolidated CSV.

    Fetches + caches the CSV once, then filters it per request. Construct with
    the catalog's `csv_url` + `facet_columns`; inject a `frame=` or `cache_path=`
    to run offline.

    Args:
        csv_url: URL of the consolidated-stores CSV.
        facet_columns: The CSV facet column names (from the catalog), kept as
            schema documentation; `resolve()` filters on :data:`_FILTER_FACETS`,
            not on this list.
        cache_path: Where to cache the downloaded CSV. Defaults to
            :func:`default_cache_path`.
        frame: A pre-loaded `DataFrame` to use verbatim, skipping all I/O.
        timeout: Per-request network timeout, in seconds.
    """

    def __init__(
        self,
        csv_url: str,
        facet_columns: list[str],
        *,
        cache_path: Path | str | None = None,
        frame: pd.DataFrame | None = None,
        timeout: float = 120.0,
    ):
        self.csv_url = csv_url
        self.facet_columns = list(facet_columns)
        self.cache_path = (
            Path(cache_path) if cache_path is not None else default_cache_path()
        )
        self.timeout = timeout
        self._frame = frame

    @property
    def frame(self) -> pd.DataFrame:
        """The consolidated-stores table, loaded + cached on first access.

        Returns:
            pandas.DataFrame: The full store index.
        """
        if self._frame is None:
            self._frame = self._load()
        return self._frame

    def _load(self) -> pd.DataFrame:
        """Read the CSV into a `DataFrame`, downloading + caching it if needed.

        Returns:
            pandas.DataFrame: The parsed CSV.
        """
        import pandas as pd

        path = self._ensure_csv()
        return pd.read_csv(path, low_memory=False)

    def _ensure_csv(self) -> Path:
        """Return the cached CSV path, downloading it once if absent.

        Streams through :class:`~earthlens.base.http.HttpClient`'s atomic
        `download` (temp `.part` file + rename, cleanup on failure), then
        enforces the non-empty invariant: a zero-byte transfer unlinks the
        cache and raises rather than caching a useless file.

        Returns:
            Path: The local cache path (guaranteed to exist and be non-empty).

        Raises:
            requests.RequestException: On a transport error or a non-2xx
                status from the download (`HttpClient.download` calls
                `raise_for_status` — this covers `HTTPError` for a 4xx/5xx
                on the CSV, `ConnectionError` for a network failure, and
                `Timeout` if the transfer stalls past `self.timeout`).
            OSError: If the atomic rename of the `.part` temp to
                `cache_path` fails.
            RuntimeError: When the download succeeds but yields a
                zero-byte CSV (the cache is unlinked before raising).
        """
        if self.cache_path.exists() and self.cache_path.stat().st_size > 0:
            return self.cache_path
        client = HttpClient(
            timeout=self.timeout,
            max_retries=0,
            status_forcelist=(),
            raise_for_status=True,
        )
        client.download(
            self.csv_url,
            self.cache_path,
            chunk=1 << 20,
            atomic=True,
            progress=False,
        )
        if self.cache_path.stat().st_size == 0:
            self.cache_path.unlink(missing_ok=True)
            raise RuntimeError(f"downloaded an empty CSV from {self.csv_url}")
        return self.cache_path

    def resolve(
        self,
        *,
        source_id: str,
        experiment_id: str,
        variable_id: str,
        table_id: str,
        member_id: str | None = None,
        grid_label: str | None = None,
        version: str = "latest",
        activity_id: str | None = None,
    ) -> list[ResolvedStore]:
        """Resolve a facet tuple to the matching `zstore` store(s).

        Filters the CSV by every pinned facet (unset facets fan out), then
        reduces `version="latest"` to the newest publication per store. Returns
        one :class:`ResolvedStore` per surviving row.

        Args:
            source_id: Model (required).
            experiment_id: Scenario / experiment (required).
            variable_id: Variable (required).
            table_id: MIP table (required).
            member_id: Variant label; `None` fans out over all members.
            grid_label: Grid label; `None` fans out over all grids.
            version: `"latest"` (newest per store) or an explicit version
                string.
            activity_id: MIP; `None` leaves it unconstrained.

        Returns:
            list[ResolvedStore]: One entry per matching store. For
                `version="latest"` the entries are ordered by their identity
                facets; for an explicit version they follow CSV row order.

        Raises:
            ValueError: If no store matches; the message names the facet that
                eliminated every row and lists the values that were available.
        """
        requested = {
            "activity_id": activity_id,
            "source_id": source_id,
            "experiment_id": experiment_id,
            "variable_id": variable_id,
            "table_id": table_id,
            "member_id": member_id,
            "grid_label": grid_label,
        }
        frame = self.frame
        for facet in _FILTER_FACETS:
            value = requested.get(facet)
            if value is None or facet not in frame.columns:
                continue
            narrowed = frame[frame[facet].astype(str) == str(value)]
            if narrowed.empty:
                available = sorted(frame[facet].dropna().astype(str).unique())
                raise ValueError(
                    f"no CMIP6 store matches {facet}={value!r} for the requested "
                    f"facets so far ({self._describe(requested, facet)}); "
                    f"{self._available_hint(facet, value, available)}"
                )
            frame = narrowed
        frame = self._select_version(frame, version)
        return [self._row_to_store(row) for _, row in frame.iterrows()]

    def _select_version(self, frame: pd.DataFrame, version: str) -> pd.DataFrame:
        """Apply the version policy — `latest` (newest per store) or exact.

        Args:
            frame: The facet-filtered frame.
            version: `"latest"` or an explicit version string.

        Returns:
            pandas.DataFrame: The frame reduced to the chosen version(s),
                sorted by the store slug facets.

        Raises:
            ValueError: If an explicit `version` matches nothing.
        """
        if "version" not in frame.columns:
            return frame
        if str(version).lower() != "latest":
            narrowed = frame[frame["version"].astype(str) == str(version)]
            if narrowed.empty:
                available = sorted(frame["version"].dropna().astype(str).unique())
                raise ValueError(
                    f"no CMIP6 store matches version={version!r}; "
                    f"available versions: {available}."
                )
            return narrowed
        keys = [f for f in _IDENTITY_FACETS if f in frame.columns]
        ordered = frame.sort_values("version", ascending=False)
        if keys:
            ordered = ordered.drop_duplicates(subset=keys, keep="first")
            ordered = ordered.sort_values(keys)
        return ordered

    @staticmethod
    def _row_to_store(row: Any) -> ResolvedStore:
        """Build a :class:`ResolvedStore` from one CSV row.

        Args:
            row: A `pandas.Series` for one store.

        Returns:
            ResolvedStore: The typed store descriptor.
        """
        return ResolvedStore(
            zstore=str(row["zstore"]),
            source_id=str(row.get("source_id", "")),
            experiment_id=str(row.get("experiment_id", "")),
            variable_id=str(row.get("variable_id", "")),
            table_id=str(row.get("table_id", "")),
            member_id=str(row.get("member_id", "")),
            grid_label=str(row.get("grid_label", "")),
            version=str(row.get("version", "")),
            activity_id=str(row.get("activity_id", "")),
        )

    @staticmethod
    def _describe(requested: dict[str, str | None], up_to: str) -> str:
        """Summarise the facets pinned before the one that failed.

        Args:
            requested: The full requested-facet mapping.
            up_to: The facet that eliminated every row.

        Returns:
            str: A `k=v` list of the facets applied before `up_to`, or
                `"no prior facets"` when it was the first.
        """
        applied = []
        for facet in _FILTER_FACETS:
            if facet == up_to:
                break
            value = requested.get(facet)
            if value is not None:
                applied.append(f"{facet}={value}")
        return ", ".join(applied) if applied else "no prior facets"

    @staticmethod
    def _available_hint(
        facet: str, value: str, available: list[str], limit: int = 20
    ) -> str:
        """Build a concise "available values" hint with a did-you-mean.

        Keeps the miss message readable on a high-cardinality facet (a
        `source_id` / `variable_id` miss can leave 100s of candidates) by
        capping the listed values and adding the closest match as a
        did-you-mean, mirroring the catalog's `difflib` lookups.

        Args:
            facet: The facet that eliminated every row.
            value: The requested (unmatched) value.
            available: The sorted values that were still available.
            limit: Maximum number of values to list before truncating.

        Returns:
            str: `available {facet}: [v1, …][, +K more]. Did you mean 'x'?`.
        """
        import difflib

        close = difflib.get_close_matches(str(value), available, n=1)
        hint = f" Did you mean {close[0]!r}?" if close else ""
        shown = available[:limit]
        tail = f", +{len(available) - limit} more" if len(available) > limit else ""
        return f"available {facet}: {shown}{tail}.{hint}"

`frame` `property` #

The consolidated-stores table, loaded + cached on first access.

Returns:

Type	Description
`DataFrame`	pandas.DataFrame: The full store index.

`resolve(*, source_id, experiment_id, variable_id, table_id, member_id=None, grid_label=None, version='latest', activity_id=None)` #

Resolve a facet tuple to the matching zstore store(s).

Filters the CSV by every pinned facet (unset facets fan out), then reduces version="latest" to the newest publication per store. Returns one :class:ResolvedStore per surviving row.

Parameters:

Name	Type	Description	Default
`source_id`	`str`	Model (required).	required
`experiment_id`	`str`	Scenario / experiment (required).	required
`variable_id`	`str`	Variable (required).	required
`table_id`	`str`	MIP table (required).	required
`member_id`	`str \| None`	Variant label; `None` fans out over all members.	`None`
`grid_label`	`str \| None`	Grid label; `None` fans out over all grids.	`None`
`version`	`str`	`"latest"` (newest per store) or an explicit version string.	`'latest'`
`activity_id`	`str \| None`	MIP; `None` leaves it unconstrained.	`None`

Returns:

Type	Description
`list[ResolvedStore]`	list[ResolvedStore]: One entry per matching store. For `version="latest"` the entries are ordered by their identity facets; for an explicit version they follow CSV row order.

Raises:

Type	Description
`ValueError`	If no store matches; the message names the facet that eliminated every row and lists the values that were available.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/resolver.py

def resolve(
    self,
    *,
    source_id: str,
    experiment_id: str,
    variable_id: str,
    table_id: str,
    member_id: str | None = None,
    grid_label: str | None = None,
    version: str = "latest",
    activity_id: str | None = None,
) -> list[ResolvedStore]:
    """Resolve a facet tuple to the matching `zstore` store(s).

    Filters the CSV by every pinned facet (unset facets fan out), then
    reduces `version="latest"` to the newest publication per store. Returns
    one :class:`ResolvedStore` per surviving row.

    Args:
        source_id: Model (required).
        experiment_id: Scenario / experiment (required).
        variable_id: Variable (required).
        table_id: MIP table (required).
        member_id: Variant label; `None` fans out over all members.
        grid_label: Grid label; `None` fans out over all grids.
        version: `"latest"` (newest per store) or an explicit version
            string.
        activity_id: MIP; `None` leaves it unconstrained.

    Returns:
        list[ResolvedStore]: One entry per matching store. For
            `version="latest"` the entries are ordered by their identity
            facets; for an explicit version they follow CSV row order.

    Raises:
        ValueError: If no store matches; the message names the facet that
            eliminated every row and lists the values that were available.
    """
    requested = {
        "activity_id": activity_id,
        "source_id": source_id,
        "experiment_id": experiment_id,
        "variable_id": variable_id,
        "table_id": table_id,
        "member_id": member_id,
        "grid_label": grid_label,
    }
    frame = self.frame
    for facet in _FILTER_FACETS:
        value = requested.get(facet)
        if value is None or facet not in frame.columns:
            continue
        narrowed = frame[frame[facet].astype(str) == str(value)]
        if narrowed.empty:
            available = sorted(frame[facet].dropna().astype(str).unique())
            raise ValueError(
                f"no CMIP6 store matches {facet}={value!r} for the requested "
                f"facets so far ({self._describe(requested, facet)}); "
                f"{self._available_hint(facet, value, available)}"
            )
        frame = narrowed
    frame = self._select_version(frame, version)
    return [self._row_to_store(row) for _, row in frame.iterrows()]

`Table` #

Bases: BaseModel

One curated CMIP6 MIP-table row (a realm x cadence bundle).

Attributes:

Name	Type	Description
`realm`	`str`	Modelling realm (`"atmos"`, `"ocean"`, `"land"`, ...).
`cadence`	`str`	Output cadence (`"monthly"`, `"daily"`, `"3-hourly"`, `"yearly"`, `"fixed"`).
`description`	`str`	Human-readable summary.

Examples:

Read a table's cadence:

>>> from earthlens.cmip6 import Catalog
>>> Catalog().get_table("Amon").cadence
'monthly'

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

class Table(BaseModel):
    """One curated CMIP6 MIP-table row (a realm x cadence bundle).

    Attributes:
        realm: Modelling realm (`"atmos"`, `"ocean"`, `"land"`, ...).
        cadence: Output cadence (`"monthly"`, `"daily"`, `"3-hourly"`,
            `"yearly"`, `"fixed"`).
        description: Human-readable summary.

    Examples:
        - Read a table's cadence:
            ```python
            >>> from earthlens.cmip6 import Catalog
            >>> Catalog().get_table("Amon").cadence
            'monthly'

            ```
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    realm: str = ""
    cadence: str = ""
    description: str = ""

`clear_catalog_cache()` #

Empty the module-level CMIP6 catalog parse cache.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

def clear_catalog_cache() -> None:
    """Empty the module-level CMIP6 catalog parse cache."""
    _CATALOG_CACHE.clear()

`earthlens.cmip6.backend` #

Backend that fetches raw CMIP6 climate projections from the Pangeo ARCO mirror.

CMIP6(AbstractDataSource) exposes the full raw CMIP6 archive — every ScenarioMIP / CMIP experiment, every ESM, on its native grid — as analysis-ready Zarr on the open gs://cmip6 Google Cloud bucket (Pangeo), indexed by a plain consolidated-stores CSV with no auth. This is the whole model x scenario x variable x member matrix, unlike the single pre-downscaled NASA/GDDP-CMIP6 product the gee backend exposes or the CHC-CMIP6 precipitation deltas the chc backend exposes.

A request is a CMIP6 facet tuple — source_id (model), experiment_id (scenario), variable_id, table_id (+ optional member_id / grid_label / version). :meth:_search resolves it against the CSV (:class:~earthlens.cmip6.resolver.StoreResolver) to the matching zstore URI(s) — a tuple that pins fewer facets fans out, one output per store. :meth:_fetch then has pyramids open each store and write a bbox/time NetCDF subset (:mod:earthlens.cmip6.accessor): the [start, end] window maps to an integer time-index range, the lat_lim/lon_lim box crops the grid, and only the requested cells are fetched. earthlens never imports xarray / zarr / gcsfs — pyramids owns the read (GDAL /vsigs/, anonymous).

The archive is on each model's native grid and stores are large, so a subset is the default; a whole-grid download (lat_lim/lon_lim left at whole-Earth) is allowed but warned. Aggregation (aggregate=) is not supported — the written NetCDFs can be aggregated separately.

`CMIP6` #

Bases: AbstractDataSource

CMIP6 climate-projections backend (raw archive on gs://cmip6).

Wraps the open Pangeo CMIP6 ARCO mirror so a user pulls a model / scenario / variable / member subset of the raw CMIP6 archive through the same download() shape every other earthlens backend uses. The output is one gridded NetCDF per resolved store.

Attributes:

Name	Type	Description
`OUTPUT_KIND`	`OutputKind`	`"raster"` — the written artefacts are gridded NetCDFs.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/backend.py

class CMIP6(AbstractDataSource):
    """CMIP6 climate-projections backend (raw archive on `gs://cmip6`).

    Wraps the open Pangeo CMIP6 ARCO mirror so a user pulls a
    model / scenario / variable / member subset of the raw CMIP6 archive through
    the same `download()` shape every other earthlens backend uses. The output is
    one gridded NetCDF per resolved store.

    Attributes:
        OUTPUT_KIND: `"raster"` — the written artefacts are gridded NetCDFs.
    """

    OUTPUT_KIND: OutputKind = "raster"

    AGGREGATE_REFUSAL_REASON = "the backend writes gridded NetCDF subsets; reduce them separately with earthlens.aggregate.aggregate_netcdf"

    def __init__(
        self,
        start: str,
        end: str,
        *,
        source_id: str | None = None,
        experiment_id: str | None = None,
        variable_id: str | None = None,
        table_id: str | None = None,
        lat_lim: list[float] | None = None,
        lon_lim: list[float] | None = None,
        member_id: str | None = None,
        grid_label: str | None = None,
        version: str = "latest",
        activity_id: str | None = None,
        whole_time: bool = False,
        temporal_resolution: str = "monthly",
        path: Path | str | None = None,
        fmt: str = "%Y-%m-%d",
        catalog: Catalog | None = None,
        resolver: StoreResolver | None = None,
    ):
        """Initialise a CMIP6 backend instance.

        Args:
            start: Inclusive start of the date window (parsed with `fmt`).
            end: Inclusive end of the date window.
            source_id: Model key (`"CanESM5"`, `"GFDL-ESM4"`).
            experiment_id: Scenario / experiment (`"ssp585"`, `"historical"`).
            variable_id: The CMIP6 variable to fetch (`"tas"`, `"pr"`).
            table_id: The MIP table (`"Amon"`, `"day"`, `"Omon"`).
            lat_lim: `[lat_min, lat_max]` in degrees. A whole-Earth box
                (`[-90, 90]`, the default) is "no spatial subset" — a whole-grid
                download, warned; a narrower box crops the native grid.
            lon_lim: `[lon_min, lon_max]` in degrees. Defaults to `[-180, 180]`.
            member_id: Variant label; `None` uses the catalog default
                (`r1i1p1f1`).
            grid_label: Grid label (`"gn"`, `"gr"`); `None` fans out over the
                grids present for the other facets.
            version: `"latest"` (newest publication per store) or an explicit
                version string.
            activity_id: MIP the experiment belongs to; `None` leaves it
                unconstrained (inferred from the experiment).
            whole_time: Skip the date-window time subset and write the whole
                series (warned). Defaults to `False`.
            temporal_resolution: Advisory cadence label (fixed by `table_id`).
            path: Output directory for the written NetCDFs.
            fmt: `strptime` format for `start` / `end`.
            catalog: Optional pre-built :class:`Catalog`; defaults to the
                bundled catalog.
            resolver: Optional pre-built
                :class:`~earthlens.cmip6.resolver.StoreResolver`; defaults to one
                built from the catalog's CSV URL and facet columns.

        Raises:
            ValueError: If a required facet (`source_id` / `experiment_id` /
                `variable_id` / `table_id`) or a date bound is omitted or empty.
        """
        for name, value in (
            ("source_id", source_id),
            ("experiment_id", experiment_id),
            ("variable_id", variable_id),
            ("table_id", table_id),
        ):
            if not value:
                raise ValueError(f"CMIP6 requires a non-empty {name}.")
        if not start or not end:
            raise ValueError(
                "CMIP6 requires a start and end date, e.g. "
                "start='2050-01-01', end='2050-12-31'."
            )

        # The loop above raised on any empty required id; narrow for the type
        # checker so the downstream str-typed uses see non-optional values.
        assert source_id is not None
        assert experiment_id is not None
        assert variable_id is not None
        assert table_id is not None

        self._catalog = catalog if catalog is not None else Catalog()
        self._resolver = (
            resolver
            if resolver is not None
            else StoreResolver(self._catalog.csv_url, self._catalog.facet_columns)
        )
        self._source_id = source_id
        self._experiment_id = experiment_id
        self._variable_id = variable_id
        self._table_id = table_id
        self._member_id = member_id or self._catalog.default_member_id
        self._grid_label = grid_label
        self._version = version
        self._activity_id = activity_id
        self._whole_time = whole_time
        self._show_progress = True

        super().__init__(
            start=start,
            end=end,
            variables=[variable_id],
            temporal_resolution=temporal_resolution,
            lat_lim=[-90.0, 90.0] if lat_lim is None else lat_lim,
            lon_lim=[-180.0, 180.0] if lon_lim is None else lon_lim,
            fmt=fmt,
            path=path,
        )

    def _check_input_dates(
        self, start: str, end: str, temporal_resolution: str, fmt: str
    ) -> TemporalExtent:
        """Parse the `[start, end]` window into a :class:`TemporalExtent`.

        Args:
            start: Inclusive window start.
            end: Inclusive window end.
            temporal_resolution: Advisory cadence label.
            fmt: `strptime` format tried first for a string `start` /
                `end`; a non-matching string falls back to an ISO-8601
                parse, and a `datetime` / `date` ignores it.

        Returns:
            TemporalExtent: Frozen model with the parsed bounds.
        """
        return self._whole_window_extent(start, end, fmt=fmt, resolution="raw")

    def _wants_spatial_subset(self) -> bool:
        """Return whether the request narrows the grid (a bbox crop).

        Returns:
            bool: `True` when the bbox is narrower than whole-Earth.
        """
        return not (
            self.space.latitude_min <= -90.0
            and self.space.latitude_max >= 90.0
            and self.space.longitude_min <= -180.0
            and self.space.longitude_max >= 180.0
        )

    def _bbox(self) -> tuple[float, float, float, float] | None:
        """Return the request bbox as `(west, south, east, north)`, or `None`.

        Returns:
            tuple | None: The crop window, or `None` for a whole-grid request.
        """
        if not self._wants_spatial_subset():
            return None
        return (self.space.west, self.space.south, self.space.east, self.space.north)

    def _search(self) -> list[RemoteProduct]:
        """Resolve the facet tuple to one product per matching `zstore`.

        Returns:
            list[RemoteProduct]: One product per resolved store; each carries the
                `zstore` as `href` and the store's facets as `metadata`.

        Raises:
            ValueError: If no store matches (the resolver names the offending
                facet and lists the available values).
        """
        stores = self._resolver.resolve(
            source_id=self._source_id,
            experiment_id=self._experiment_id,
            variable_id=self._variable_id,
            table_id=self._table_id,
            member_id=self._member_id,
            grid_label=self._grid_label,
            version=self._version,
            activity_id=self._activity_id,
        )
        return [
            RemoteProduct(
                id=store.slug,
                href=store.zstore,
                metadata={"store": store},
            )
            for store in stores
        ]

    def _fetch(self, products: list[RemoteProduct]) -> list[Path]:
        """Write a bbox/time NetCDF subset for each resolved store.

        For each store: map the `[start, end]` window to an integer time-index
        range (unless `whole_time`), then read the gridded `(variable, time,
        bbox)` window through pyramids and write it to NetCDF.

        Args:
            products: The products from :meth:`_search`.

        Returns:
            list[Path]: One written NetCDF path per store, in order.
        """
        bbox = self._bbox()
        if bbox is None:
            logger.warning(
                f"cmip6: no bbox — writing the whole native grid for "
                f"{self._variable_id}/{self._experiment_id}; pass lat_lim/lon_lim "
                "to subset (CMIP6 stores can be large)."
            )
        out: list[Path] = []
        for product in tqdm(
            products, disable=not self._show_progress, desc="cmip6", unit="store"
        ):
            store: ResolvedStore = product.metadata["store"]
            time_sel = self._time_selector(store)
            stem = accessor.store_output_stem(
                store, self.time.start_date, self.time.end_date
            )
            out_path = self.root_dir / f"{stem}.nc"
            out.append(
                accessor.write_subset(
                    store.zstore,
                    self._variable_id,
                    bbox=bbox,
                    time=time_sel,
                    out_path=out_path,
                )
            )
        return out

    def _time_selector(
        self, store: ResolvedStore
    ) -> int | tuple[int, int] | slice | None:
        """Resolve the time selector for one store from the request window.

        Args:
            store: The resolved store to read.

        Returns:
            The integer time selector for :func:`accessor.write_subset` — a
            `(i0, i1)` index range for the date window, or `slice(None)` for a
            `whole_time` request.
        """
        if self._whole_time:
            return slice(None)
        return accessor.resolve_time_window(
            store.zstore,
            self._variable_id,
            self.time.start_date,
            self.time.end_date,
        )

    def terms_note(self) -> str:
        """Return the attribution note for the requested source model.

        Returns:
            str: The per-model `terms_note`, else the catalog default.
        """
        return self._catalog.terms_note(self._source_id)

    def download(
        self,
        progress_bar: bool = True,
    ) -> list[Path]:
        """Fetch the requested CMIP6 subset(s) and return the written paths.

        Runs the cheap :meth:`_search` (facet -> `zstore` resolution) then
        :meth:`_fetch`, which writes one bbox/time NetCDF subset per resolved
        store.

        Args:
            progress_bar: Show a per-store progress bar. Defaults to `True`.

        Returns:
            list[Path]: The written NetCDF paths, one per resolved store (never
                empty — a facet tuple that matches no store raises rather than
                returning an empty list).

        Raises:
            ValueError: If the facet tuple matches no store.
        """
        self._show_progress = progress_bar
        return self._api_via_search_fetch()

`init(start, end, *, source_id=None, experiment_id=None, variable_id=None, table_id=None, lat_lim=None, lon_lim=None, member_id=None, grid_label=None, version='latest', activity_id=None, whole_time=False, temporal_resolution='monthly', path=None, fmt='%Y-%m-%d', catalog=None, resolver=None)` #

Initialise a CMIP6 backend instance.

Parameters:

Name	Type	Description	Default
`start`	`str`	Inclusive start of the date window (parsed with `fmt`).	required
`end`	`str`	Inclusive end of the date window.	required
`source_id`	`str \| None`	Model key (`"CanESM5"`, `"GFDL-ESM4"`).	`None`
`experiment_id`	`str \| None`	Scenario / experiment (`"ssp585"`, `"historical"`).	`None`
`variable_id`	`str \| None`	The CMIP6 variable to fetch (`"tas"`, `"pr"`).	`None`
`table_id`	`str \| None`	The MIP table (`"Amon"`, `"day"`, `"Omon"`).	`None`
`lat_lim`	`list[float] \| None`	`[lat_min, lat_max]` in degrees. A whole-Earth box (`[-90, 90]`, the default) is "no spatial subset" — a whole-grid download, warned; a narrower box crops the native grid.	`None`
`lon_lim`	`list[float] \| None`	`[lon_min, lon_max]` in degrees. Defaults to `[-180, 180]`.	`None`
`member_id`	`str \| None`	Variant label; `None` uses the catalog default (`r1i1p1f1`).	`None`
`grid_label`	`str \| None`	Grid label (`"gn"`, `"gr"`); `None` fans out over the grids present for the other facets.	`None`
`version`	`str`	`"latest"` (newest publication per store) or an explicit version string.	`'latest'`
`activity_id`	`str \| None`	MIP the experiment belongs to; `None` leaves it unconstrained (inferred from the experiment).	`None`
`whole_time`	`bool`	Skip the date-window time subset and write the whole series (warned). Defaults to `False`.	`False`
`temporal_resolution`	`str`	Advisory cadence label (fixed by `table_id`).	`'monthly'`
`path`	`Path \| str \| None`	Output directory for the written NetCDFs.	`None`
`fmt`	`str`	`strptime` format for `start` / `end`.	`'%Y-%m-%d'`
`catalog`	`Catalog \| None`	Optional pre-built :class:`Catalog`; defaults to the bundled catalog.	`None`
`resolver`	`StoreResolver \| None`	Optional pre-built :class:`~earthlens.cmip6.resolver.StoreResolver`; defaults to one built from the catalog's CSV URL and facet columns.	`None`

Raises:

Type	Description
`ValueError`	If a required facet (`source_id` / `experiment_id` / `variable_id` / `table_id`) or a date bound is omitted or empty.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/backend.py

def __init__(
    self,
    start: str,
    end: str,
    *,
    source_id: str | None = None,
    experiment_id: str | None = None,
    variable_id: str | None = None,
    table_id: str | None = None,
    lat_lim: list[float] | None = None,
    lon_lim: list[float] | None = None,
    member_id: str | None = None,
    grid_label: str | None = None,
    version: str = "latest",
    activity_id: str | None = None,
    whole_time: bool = False,
    temporal_resolution: str = "monthly",
    path: Path | str | None = None,
    fmt: str = "%Y-%m-%d",
    catalog: Catalog | None = None,
    resolver: StoreResolver | None = None,
):
    """Initialise a CMIP6 backend instance.

    Args:
        start: Inclusive start of the date window (parsed with `fmt`).
        end: Inclusive end of the date window.
        source_id: Model key (`"CanESM5"`, `"GFDL-ESM4"`).
        experiment_id: Scenario / experiment (`"ssp585"`, `"historical"`).
        variable_id: The CMIP6 variable to fetch (`"tas"`, `"pr"`).
        table_id: The MIP table (`"Amon"`, `"day"`, `"Omon"`).
        lat_lim: `[lat_min, lat_max]` in degrees. A whole-Earth box
            (`[-90, 90]`, the default) is "no spatial subset" — a whole-grid
            download, warned; a narrower box crops the native grid.
        lon_lim: `[lon_min, lon_max]` in degrees. Defaults to `[-180, 180]`.
        member_id: Variant label; `None` uses the catalog default
            (`r1i1p1f1`).
        grid_label: Grid label (`"gn"`, `"gr"`); `None` fans out over the
            grids present for the other facets.
        version: `"latest"` (newest publication per store) or an explicit
            version string.
        activity_id: MIP the experiment belongs to; `None` leaves it
            unconstrained (inferred from the experiment).
        whole_time: Skip the date-window time subset and write the whole
            series (warned). Defaults to `False`.
        temporal_resolution: Advisory cadence label (fixed by `table_id`).
        path: Output directory for the written NetCDFs.
        fmt: `strptime` format for `start` / `end`.
        catalog: Optional pre-built :class:`Catalog`; defaults to the
            bundled catalog.
        resolver: Optional pre-built
            :class:`~earthlens.cmip6.resolver.StoreResolver`; defaults to one
            built from the catalog's CSV URL and facet columns.

    Raises:
        ValueError: If a required facet (`source_id` / `experiment_id` /
            `variable_id` / `table_id`) or a date bound is omitted or empty.
    """
    for name, value in (
        ("source_id", source_id),
        ("experiment_id", experiment_id),
        ("variable_id", variable_id),
        ("table_id", table_id),
    ):
        if not value:
            raise ValueError(f"CMIP6 requires a non-empty {name}.")
    if not start or not end:
        raise ValueError(
            "CMIP6 requires a start and end date, e.g. "
            "start='2050-01-01', end='2050-12-31'."
        )

    # The loop above raised on any empty required id; narrow for the type
    # checker so the downstream str-typed uses see non-optional values.
    assert source_id is not None
    assert experiment_id is not None
    assert variable_id is not None
    assert table_id is not None

    self._catalog = catalog if catalog is not None else Catalog()
    self._resolver = (
        resolver
        if resolver is not None
        else StoreResolver(self._catalog.csv_url, self._catalog.facet_columns)
    )
    self._source_id = source_id
    self._experiment_id = experiment_id
    self._variable_id = variable_id
    self._table_id = table_id
    self._member_id = member_id or self._catalog.default_member_id
    self._grid_label = grid_label
    self._version = version
    self._activity_id = activity_id
    self._whole_time = whole_time
    self._show_progress = True

    super().__init__(
        start=start,
        end=end,
        variables=[variable_id],
        temporal_resolution=temporal_resolution,
        lat_lim=[-90.0, 90.0] if lat_lim is None else lat_lim,
        lon_lim=[-180.0, 180.0] if lon_lim is None else lon_lim,
        fmt=fmt,
        path=path,
    )

`download(progress_bar=True)` #

Fetch the requested CMIP6 subset(s) and return the written paths.

Runs the cheap :meth:_search (facet -> zstore resolution) then :meth:_fetch, which writes one bbox/time NetCDF subset per resolved store.

Parameters:

Name	Type	Description	Default
`progress_bar`	`bool`	Show a per-store progress bar. Defaults to `True`.	`True`

Returns:

Type	Description
`list[Path]`	list[Path]: The written NetCDF paths, one per resolved store (never empty — a facet tuple that matches no store raises rather than returning an empty list).

Raises:

Type	Description
`ValueError`	If the facet tuple matches no store.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/backend.py

def download(
    self,
    progress_bar: bool = True,
) -> list[Path]:
    """Fetch the requested CMIP6 subset(s) and return the written paths.

    Runs the cheap :meth:`_search` (facet -> `zstore` resolution) then
    :meth:`_fetch`, which writes one bbox/time NetCDF subset per resolved
    store.

    Args:
        progress_bar: Show a per-store progress bar. Defaults to `True`.

    Returns:
        list[Path]: The written NetCDF paths, one per resolved store (never
            empty — a facet tuple that matches no store raises rather than
            returning an empty list).

    Raises:
        ValueError: If the facet tuple matches no store.
    """
    self._show_progress = progress_bar
    return self._api_via_search_fetch()

`terms_note()` #

Return the attribution note for the requested source model.

Returns:

Name	Type	Description
`str`	`str`	The per-model `terms_note`, else the catalog default.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/backend.py

def terms_note(self) -> str:
    """Return the attribution note for the requested source model.

    Returns:
        str: The per-model `terms_note`, else the catalog default.
    """
    return self._catalog.terms_note(self._source_id)

`earthlens.cmip6.catalog` #

Config + curated-vocabulary catalog for the CMIP6 backend.

Hosts :class:Catalog, the pydantic-backed reader for the bundled cmip6_data_catalog.yaml. CMIP6 is addressed by a facet tuple (source_id, experiment_id, variable_id, table_id, plus optional member_id / grid_label / version), and the per-store index — one Zarr store per facet combination — is far too large to inline (~515k rows). That index is the consolidated-stores CSV at :attr:Catalog.csv_url, fetched and cached by :mod:earthlens.cmip6.resolver; this catalog holds only the config (CSV URL, bucket, facet columns, defaults) plus a curated vocabulary of the common variables / experiments / tables / sources used for output metadata, docs, and did-you-mean hints.

The curated variables: block is exposed under the inherited :attr:~earthlens.base.AbstractCatalog.datasets field, so a variable resolves with cat["tas"] / "tas" in cat / the did-you-mean error for free; experiments: / tables: / sources: hang off parallel maps. Resolution itself runs against the full CSV, so an uncurated facet still downloads — the curated rows only enrich metadata and error messages.

:data:CATALOG_PATH is the path to the bundled YAML; :func:clear_catalog_cache empties the (path, mtime) parse cache.

`Catalog` #

Bases: AbstractCatalog

Config + curated-vocabulary catalog for the CMIP6 backend.

Reads the bundled cmip6_data_catalog.yaml (shipped as package data) and exposes its variables: block as a map of :class:Cmip6Variable rows keyed by variable_id under the inherited :attr:datasets field, plus parallel :attr:experiments, :attr:tables, and :attr:sources maps and the resolution config (:attr:csv_url, :attr:bucket, :attr:facet_columns, :attr:default_member_id, :attr:default_version). Instantiate with no arguments (Catalog()); :func:model_post_init loads and validates the YAML in one pass and caches it by (path, mtime).

Attributes:

Name	Type	Description
`csv_url`	`str`	URL of the consolidated-stores CSV (the full per-store index).
`bucket`	`str`	The public GCS bucket the `zstore` URIs live on (`"cmip6"`).
`facet_columns`	`list[str]`	The CSV facet columns, in file order.
`default_member_id`	`str`	Member label applied when a request omits it.
`default_version`	`str`	Version-selection policy (`"latest"`).
`default_terms_note`	`str`	Attribution fallback for an uncurated source.
`datasets`	`dict[str, Cmip6Variable]`	Map from `variable_id` to its :class:`Cmip6Variable` row.
`experiments`	`dict[str, Experiment]`	Map from `experiment_id` to its :class:`Experiment` row.
`tables`	`dict[str, Table]`	Map from `table_id` to its :class:`Table` row.
`sources`	`dict[str, Source]`	Map from `source_id` to its :class:`Source` row.

Examples:

List curated variables and resolve one:

>>> from earthlens.cmip6 import Catalog
>>> cat = Catalog()
>>> "tas" in cat
True
>>> cat.get_dataset("tas").units
'K'
>>> cat.bucket
'cmip6'

An unknown variable raises with a did-you-mean hint:

>>> from earthlens.cmip6 import Catalog
>>> Catalog().get_dataset("rainfall")  # doctest: +ELLIPSIS
Traceback (most recent call last):
    ...
ValueError: 'rainfall' is not in the CMIP6 catalog. Known variables: [...].

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

class Catalog(AbstractCatalog):
    """Config + curated-vocabulary catalog for the CMIP6 backend.

    Reads the bundled `cmip6_data_catalog.yaml` (shipped as package data) and
    exposes its `variables:` block as a map of :class:`Cmip6Variable` rows keyed
    by `variable_id` under the inherited :attr:`datasets` field, plus parallel
    :attr:`experiments`, :attr:`tables`, and :attr:`sources` maps and the
    resolution config (:attr:`csv_url`, :attr:`bucket`, :attr:`facet_columns`,
    :attr:`default_member_id`, :attr:`default_version`). Instantiate with no
    arguments (`Catalog()`); :func:`model_post_init` loads and validates the YAML
    in one pass and caches it by `(path, mtime)`.

    Attributes:
        csv_url: URL of the consolidated-stores CSV (the full per-store index).
        bucket: The public GCS bucket the `zstore` URIs live on (`"cmip6"`).
        facet_columns: The CSV facet columns, in file order.
        default_member_id: Member label applied when a request omits it.
        default_version: Version-selection policy (`"latest"`).
        default_terms_note: Attribution fallback for an uncurated source.
        datasets: Map from `variable_id` to its :class:`Cmip6Variable` row.
        experiments: Map from `experiment_id` to its :class:`Experiment` row.
        tables: Map from `table_id` to its :class:`Table` row.
        sources: Map from `source_id` to its :class:`Source` row.

    Examples:
        - List curated variables and resolve one:
            ```python
            >>> from earthlens.cmip6 import Catalog
            >>> cat = Catalog()
            >>> "tas" in cat
            True
            >>> cat.get_dataset("tas").units
            'K'
            >>> cat.bucket
            'cmip6'

            ```
        - An unknown variable raises with a did-you-mean hint:
            ```python
            >>> from earthlens.cmip6 import Catalog
            >>> Catalog().get_dataset("rainfall")  # doctest: +ELLIPSIS
            Traceback (most recent call last):
                ...
            ValueError: 'rainfall' is not in the CMIP6 catalog. Known variables: [...].

            ```
    """

    _catalog_kind: str = "CMIP6 catalog"
    _entry_noun: str = "variables"

    csv_url: str = ""
    bucket: str = "cmip6"
    facet_columns: list[str] = Field(default_factory=list)
    default_member_id: str = "r1i1p1f1"
    default_version: str = "latest"
    default_terms_note: str = ""

    datasets: dict[str, Cmip6Variable] = Field(default_factory=dict)
    experiments: dict[str, Experiment] = Field(default_factory=dict)
    tables: dict[str, Table] = Field(default_factory=dict)
    sources: dict[str, Source] = Field(default_factory=dict)

    def model_post_init(self, __context: Any) -> None:
        """Auto-load the bundled catalog when no variables were supplied.

        `Catalog()` with no args reads :data:`CATALOG_PATH` (cached by
        `(path, mtime)`); passing `datasets=...` skips the disk read (used in
        tests). Either way the `available_datasets` index is derived from the
        loaded variable map.

        Raises:
            ValueError: Propagated from :meth:`load` when the YAML is missing,
                empty, or has a malformed row.
        """
        if not self.datasets and not self.csv_url:
            loaded = Catalog.load()
            self.csv_url = loaded.csv_url
            self.bucket = loaded.bucket
            self.facet_columns = loaded.facet_columns
            self.default_member_id = loaded.default_member_id
            self.default_version = loaded.default_version
            self.default_terms_note = loaded.default_terms_note
            self.datasets = loaded.datasets
            self.experiments = loaded.experiments
            self.tables = loaded.tables
            self.sources = loaded.sources
        self.available_datasets = sorted(self.datasets)

    @classmethod
    def load(cls, catalog_path: Path | None = None) -> Catalog:
        """Read and validate the CMIP6 catalog from disk (cached).

        Args:
            catalog_path: Path to the catalog YAML. Defaults to the module-level
                :data:`CATALOG_PATH`.

        Returns:
            A fully-populated :class:`Catalog`.

        Raises:
            ValueError: If `catalog_path` does not exist, or if the file is missing
                its `csv_url`, or any curated row fails validation.
        """
        path = catalog_path if catalog_path is not None else CATALOG_PATH
        payload = load_catalog(path, _CATALOG_CACHE, _parse_catalog, provider="CMIP6")
        return cls(**payload)

    @staticmethod
    def _parse_block(path: Path, block: Any, model: type[BaseModel]) -> dict[str, Any]:
        """Validate one YAML mapping block into `{key: model(...)}`.

        Args:
            path: Catalog path, for error messages.
            block: The raw mapping from the YAML (or `None` when absent).
            model: The pydantic row type to build.

        Returns:
            dict[str, Any]: The validated rows keyed by their YAML key.

        Raises:
            ValueError: If any row fails validation.
        """
        out: dict[str, Any] = {}
        for key, body in (block or {}).items():
            try:
                out[str(key)] = model(**dict(body or {}))
            except ValidationError as exc:
                raise ValueError(
                    f"{path} {model.__name__} {key!r} failed validation:\n{exc}"
                ) from exc
        return out

    def get_catalog(self) -> dict[str, Cmip6Variable]:
        """Return the curated variable map (satisfies the abstract contract).

        Returns:
            dict[str, Cmip6Variable]: Same object as :attr:`datasets`.
        """
        return self.datasets

    def get_experiment(self, key: str) -> Experiment:
        """Return the :class:`Experiment` for `key`, with a did-you-mean hint.

        Args:
            key: An `experiment_id` (`"ssp585"`, `"historical"`).

        Returns:
            Experiment: The matching experiment row.

        Raises:
            ValueError: If `key` is not a curated experiment.
        """
        return cast("Experiment", self._get_from(self.experiments, key, "experiment"))

    def get_table(self, key: str) -> Table:
        """Return the :class:`Table` for `key`, with a did-you-mean hint.

        Args:
            key: A `table_id` (`"Amon"`, `"day"`, `"Omon"`).

        Returns:
            Table: The matching table row.

        Raises:
            ValueError: If `key` is not a curated table.
        """
        return cast("Table", self._get_from(self.tables, key, "table"))

    def get_source(self, key: str) -> Source:
        """Return the :class:`Source` for `key`, with a did-you-mean hint.

        Args:
            key: A `source_id` (`"CanESM5"`, `"GFDL-ESM4"`).

        Returns:
            Source: The matching source row.

        Raises:
            ValueError: If `key` is not a curated source.
        """
        return cast("Source", self._get_from(self.sources, key, "source"))

    def terms_note(self, source_id: str) -> str:
        """Return the attribution note for `source_id`.

        Falls back to :attr:`default_terms_note` for an uncurated source (or a
        curated one with no per-model note).

        Args:
            source_id: The model key (`"CanESM5"`).

        Returns:
            str: The per-model `terms_note`, else the catalog default.
        """
        source = self.sources.get(source_id)
        if source is not None and source.terms_note:
            return source.terms_note
        return self.default_terms_note

    @staticmethod
    def _get_from(mapping: dict[str, Any], key: str, noun: str) -> Any:
        """Look up `key` in `mapping`, raising a did-you-mean `ValueError`.

        Args:
            mapping: The curated map to search.
            key: The requested key.
            noun: Singular noun for the error message (`"experiment"`).

        Returns:
            The matching row.

        Raises:
            ValueError: If `key` is absent.
        """
        try:
            return mapping[key]
        except KeyError:
            import difflib

            close = difflib.get_close_matches(key, mapping, n=1)
            hint = f" Did you mean {close[0]!r}?" if close else ""
            raise ValueError(
                f"{key!r} is not a curated CMIP6 {noun}. "
                f"Known {noun}s: {sorted(mapping)}.{hint}"
            ) from None

`get_catalog()` #

Return the curated variable map (satisfies the abstract contract).

Returns:

Type	Description
`dict[str, Cmip6Variable]`	dict[str, Cmip6Variable]: Same object as :attr:`datasets`.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

def get_catalog(self) -> dict[str, Cmip6Variable]:
    """Return the curated variable map (satisfies the abstract contract).

    Returns:
        dict[str, Cmip6Variable]: Same object as :attr:`datasets`.
    """
    return self.datasets

`get_experiment(key)` #

Return the :class:Experiment for key, with a did-you-mean hint.

Parameters:

Name	Type	Description	Default
`key`	`str`	An `experiment_id` (`"ssp585"`, `"historical"`).	required

Returns:

Name	Type	Description
`Experiment`	`Experiment`	The matching experiment row.

Raises:

Type	Description
`ValueError`	If `key` is not a curated experiment.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

def get_experiment(self, key: str) -> Experiment:
    """Return the :class:`Experiment` for `key`, with a did-you-mean hint.

    Args:
        key: An `experiment_id` (`"ssp585"`, `"historical"`).

    Returns:
        Experiment: The matching experiment row.

    Raises:
        ValueError: If `key` is not a curated experiment.
    """
    return cast("Experiment", self._get_from(self.experiments, key, "experiment"))

`get_source(key)` #

Return the :class:Source for key, with a did-you-mean hint.

Parameters:

Name	Type	Description	Default
`key`	`str`	A `source_id` (`"CanESM5"`, `"GFDL-ESM4"`).	required

Returns:

Name	Type	Description
`Source`	`Source`	The matching source row.

Raises:

Type	Description
`ValueError`	If `key` is not a curated source.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

def get_source(self, key: str) -> Source:
    """Return the :class:`Source` for `key`, with a did-you-mean hint.

    Args:
        key: A `source_id` (`"CanESM5"`, `"GFDL-ESM4"`).

    Returns:
        Source: The matching source row.

    Raises:
        ValueError: If `key` is not a curated source.
    """
    return cast("Source", self._get_from(self.sources, key, "source"))

`get_table(key)` #

Return the :class:Table for key, with a did-you-mean hint.

Parameters:

Name	Type	Description	Default
`key`	`str`	A `table_id` (`"Amon"`, `"day"`, `"Omon"`).	required

Returns:

Name	Type	Description
`Table`	`Table`	The matching table row.

Raises:

Type	Description
`ValueError`	If `key` is not a curated table.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

def get_table(self, key: str) -> Table:
    """Return the :class:`Table` for `key`, with a did-you-mean hint.

    Args:
        key: A `table_id` (`"Amon"`, `"day"`, `"Omon"`).

    Returns:
        Table: The matching table row.

    Raises:
        ValueError: If `key` is not a curated table.
    """
    return cast("Table", self._get_from(self.tables, key, "table"))

`load(catalog_path=None)` `classmethod` #

Read and validate the CMIP6 catalog from disk (cached).

Parameters:

Name	Type	Description	Default
`catalog_path`	`Path \| None`	Path to the catalog YAML. Defaults to the module-level :data:`CATALOG_PATH`.	`None`

Returns:

Type	Description
`Catalog`	A fully-populated :class:`Catalog`.

Raises:

Type	Description
`ValueError`	If `catalog_path` does not exist, or if the file is missing its `csv_url`, or any curated row fails validation.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

@classmethod
def load(cls, catalog_path: Path | None = None) -> Catalog:
    """Read and validate the CMIP6 catalog from disk (cached).

    Args:
        catalog_path: Path to the catalog YAML. Defaults to the module-level
            :data:`CATALOG_PATH`.

    Returns:
        A fully-populated :class:`Catalog`.

    Raises:
        ValueError: If `catalog_path` does not exist, or if the file is missing
            its `csv_url`, or any curated row fails validation.
    """
    path = catalog_path if catalog_path is not None else CATALOG_PATH
    payload = load_catalog(path, _CATALOG_CACHE, _parse_catalog, provider="CMIP6")
    return cls(**payload)

`model_post_init(__context)` #

Auto-load the bundled catalog when no variables were supplied.

Catalog() with no args reads :data:CATALOG_PATH (cached by (path, mtime)); passing datasets=... skips the disk read (used in tests). Either way the available_datasets index is derived from the loaded variable map.

Raises:

Type	Description
`ValueError`	Propagated from :meth:`load` when the YAML is missing, empty, or has a malformed row.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

def model_post_init(self, __context: Any) -> None:
    """Auto-load the bundled catalog when no variables were supplied.

    `Catalog()` with no args reads :data:`CATALOG_PATH` (cached by
    `(path, mtime)`); passing `datasets=...` skips the disk read (used in
    tests). Either way the `available_datasets` index is derived from the
    loaded variable map.

    Raises:
        ValueError: Propagated from :meth:`load` when the YAML is missing,
            empty, or has a malformed row.
    """
    if not self.datasets and not self.csv_url:
        loaded = Catalog.load()
        self.csv_url = loaded.csv_url
        self.bucket = loaded.bucket
        self.facet_columns = loaded.facet_columns
        self.default_member_id = loaded.default_member_id
        self.default_version = loaded.default_version
        self.default_terms_note = loaded.default_terms_note
        self.datasets = loaded.datasets
        self.experiments = loaded.experiments
        self.tables = loaded.tables
        self.sources = loaded.sources
    self.available_datasets = sorted(self.datasets)

`terms_note(source_id)` #

Return the attribution note for source_id.

Falls back to :attr:default_terms_note for an uncurated source (or a curated one with no per-model note).

Parameters:

Name	Type	Description	Default
`source_id`	`str`	The model key (`"CanESM5"`).	required

Returns:

Name	Type	Description
`str`	`str`	The per-model `terms_note`, else the catalog default.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

def terms_note(self, source_id: str) -> str:
    """Return the attribution note for `source_id`.

    Falls back to :attr:`default_terms_note` for an uncurated source (or a
    curated one with no per-model note).

    Args:
        source_id: The model key (`"CanESM5"`).

    Returns:
        str: The per-model `terms_note`, else the catalog default.
    """
    source = self.sources.get(source_id)
    if source is not None and source.terms_note:
        return source.terms_note
    return self.default_terms_note

`Cmip6Variable` #

Bases: BaseModel

One curated CMIP6 variable row (the variable_id leaf).

A frozen value object with descriptive metadata only — CMIP6 variables carry no request-shaping parameters; the facet tuple selects the store and pyramids reads the array. Curated rows are optional: an uncurated variable_id still resolves against the CSV, it just lacks these labels.

Attributes:

Name	Type	Description
`units`	`str`	CMIP6 CMOR unit (`"K"`, `"kg m-2 s-1"`, `"1"` for a dimensionless fraction).
`long_name`	`str`	Human-readable description used in docs and logs.
`realm`	`str`	Modelling realm the variable belongs to (`"atmos"`, `"ocean"`, `"land"`, `"seaIce"`, ...).

Examples:

Build a variable row directly:

>>> from earthlens.cmip6 import Cmip6Variable
>>> v = Cmip6Variable(units="K", long_name="Near-surface air temperature")
>>> v.units
'K'

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

class Cmip6Variable(BaseModel):
    """One curated CMIP6 variable row (the `variable_id` leaf).

    A frozen value object with descriptive metadata only — CMIP6 variables carry
    no request-shaping parameters; the facet tuple selects the store and pyramids
    reads the array. Curated rows are optional: an uncurated `variable_id` still
    resolves against the CSV, it just lacks these labels.

    Attributes:
        units: CMIP6 CMOR unit (`"K"`, `"kg m-2 s-1"`, `"1"` for a
            dimensionless fraction).
        long_name: Human-readable description used in docs and logs.
        realm: Modelling realm the variable belongs to (`"atmos"`, `"ocean"`,
            `"land"`, `"seaIce"`, ...).

    Examples:
        - Build a variable row directly:
            ```python
            >>> from earthlens.cmip6 import Cmip6Variable
            >>> v = Cmip6Variable(units="K", long_name="Near-surface air temperature")
            >>> v.units
            'K'

            ```
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    units: str = ""
    long_name: str = ""
    realm: str = ""

`Experiment` #

Bases: BaseModel

One curated CMIP6 experiment (scenario / diagnostic) row.

Attributes:

Name	Type	Description
`activity_id`	`str`	The MIP the experiment belongs to (`"CMIP"` for the DECK / historical runs, `"ScenarioMIP"` for the SSPs).
`description`	`str`	Human-readable summary.

Examples:

Inspect an experiment's activity:

>>> from earthlens.cmip6 import Catalog
>>> Catalog().get_experiment("ssp585").activity_id
'ScenarioMIP'

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

class Experiment(BaseModel):
    """One curated CMIP6 experiment (scenario / diagnostic) row.

    Attributes:
        activity_id: The MIP the experiment belongs to (`"CMIP"` for the
            DECK / historical runs, `"ScenarioMIP"` for the SSPs).
        description: Human-readable summary.

    Examples:
        - Inspect an experiment's activity:
            ```python
            >>> from earthlens.cmip6 import Catalog
            >>> Catalog().get_experiment("ssp585").activity_id
            'ScenarioMIP'

            ```
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    activity_id: str = ""
    description: str = ""

`Source` #

Bases: BaseModel

One curated CMIP6 source-model (GCM) row.

Attributes:

Name	Type	Description
`institution_id`	`str`	The modelling centre that produced the model.
`terms_note`	`str`	Any per-model licence / attribution nuance (most CMIP6 models are CC BY 4.0, cited via the source GCM).
`description`	`str`	Optional human-readable summary.

Examples:

Read a source's institution:

>>> from earthlens.cmip6 import Catalog
>>> Catalog().get_source("CanESM5").institution_id
'CCCma'

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

class Source(BaseModel):
    """One curated CMIP6 source-model (GCM) row.

    Attributes:
        institution_id: The modelling centre that produced the model.
        terms_note: Any per-model licence / attribution nuance (most CMIP6
            models are CC BY 4.0, cited via the source GCM).
        description: Optional human-readable summary.

    Examples:
        - Read a source's institution:
            ```python
            >>> from earthlens.cmip6 import Catalog
            >>> Catalog().get_source("CanESM5").institution_id
            'CCCma'

            ```
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    institution_id: str = ""
    terms_note: str = ""
    description: str = ""

`Table` #

Bases: BaseModel

One curated CMIP6 MIP-table row (a realm x cadence bundle).

Attributes:

Name	Type	Description
`realm`	`str`	Modelling realm (`"atmos"`, `"ocean"`, `"land"`, ...).
`cadence`	`str`	Output cadence (`"monthly"`, `"daily"`, `"3-hourly"`, `"yearly"`, `"fixed"`).
`description`	`str`	Human-readable summary.

Examples:

Read a table's cadence:

>>> from earthlens.cmip6 import Catalog
>>> Catalog().get_table("Amon").cadence
'monthly'

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

class Table(BaseModel):
    """One curated CMIP6 MIP-table row (a realm x cadence bundle).

    Attributes:
        realm: Modelling realm (`"atmos"`, `"ocean"`, `"land"`, ...).
        cadence: Output cadence (`"monthly"`, `"daily"`, `"3-hourly"`,
            `"yearly"`, `"fixed"`).
        description: Human-readable summary.

    Examples:
        - Read a table's cadence:
            ```python
            >>> from earthlens.cmip6 import Catalog
            >>> Catalog().get_table("Amon").cadence
            'monthly'

            ```
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    realm: str = ""
    cadence: str = ""
    description: str = ""

`clear_catalog_cache()` #

Empty the module-level CMIP6 catalog parse cache.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/catalog.py

def clear_catalog_cache() -> None:
    """Empty the module-level CMIP6 catalog parse cache."""
    _CATALOG_CACHE.clear()

`earthlens.cmip6.resolver` #

Facet -> zstore resolver over the CMIP6 consolidated-stores CSV.

The Pangeo CMIP6 ARCO index is a single flat CSV — one row per Zarr store, keyed by the CMIP6 facets (source_id, experiment_id, variable_id, table_id, member_id, grid_label, version, ...) with the store URI in the zstore column. :class:StoreResolver fetches and caches that CSV (with requests, read with pandas — both core), then filters it by a requested facet tuple to the matching zstore URI(s).

The resolver is deliberately stateless and injectable: pass a pre-loaded frame= or a local cache_path= to run with no network. On a miss it raises a ValueError that names the facet which eliminated every row and lists the values that were available — so a typo in a model or scenario name is easy to fix.

No intake-esm (it would drag in xarray); no xarray / zarr / gcsfs here — this module only resolves URIs. Opening the store is :mod:earthlens.cmip6.accessor's job (pyramids).

`ResolvedStore` `dataclass` #

One CMIP6 Zarr store resolved from a facet tuple.

Carries the zstore URI plus the facet values that identify it, so the backend can name the output file and log the provenance without re-querying the CSV.

Attributes:

Name	Type	Description
`zstore`	`str`	The `gs://cmip6/...` store URI (ends in `/`).
`source_id`	`str`	Model that produced the store.
`experiment_id`	`str`	Scenario / diagnostic experiment.
`variable_id`	`str`	The CMIP6 variable.
`table_id`	`str`	The MIP table (realm x cadence).
`member_id`	`str`	The variant label (`r1i1p1f1`).
`grid_label`	`str`	Grid label (`gn` native, `gr` regridded, ...).
`version`	`str`	Data-publication version (an integer date, as a string).
`activity_id`	`str`	The MIP the experiment belongs to.

Examples:

Build one directly:

>>> from earthlens.cmip6.resolver import ResolvedStore
>>> s = ResolvedStore(
...     zstore="gs://cmip6/CMIP6/ScenarioMIP/.../tas/gn/v20190101/",
...     source_id="CanESM5", experiment_id="ssp585", variable_id="tas",
...     table_id="Amon", member_id="r1i1p1f1", grid_label="gn",
...     version="20190101", activity_id="ScenarioMIP",
... )
>>> s.variable_id
'tas'

Source code in libs/providers/atmosphere/src/earthlens/cmip6/resolver.py

@dataclass(frozen=True)
class ResolvedStore:
    """One CMIP6 Zarr store resolved from a facet tuple.

    Carries the `zstore` URI plus the facet values that identify it, so the
    backend can name the output file and log the provenance without re-querying
    the CSV.

    Attributes:
        zstore: The `gs://cmip6/...` store URI (ends in `/`).
        source_id: Model that produced the store.
        experiment_id: Scenario / diagnostic experiment.
        variable_id: The CMIP6 variable.
        table_id: The MIP table (realm x cadence).
        member_id: The variant label (`r1i1p1f1`).
        grid_label: Grid label (`gn` native, `gr` regridded, ...).
        version: Data-publication version (an integer date, as a string).
        activity_id: The MIP the experiment belongs to.

    Examples:
        - Build one directly:
            ```python
            >>> from earthlens.cmip6.resolver import ResolvedStore
            >>> s = ResolvedStore(
            ...     zstore="gs://cmip6/CMIP6/ScenarioMIP/.../tas/gn/v20190101/",
            ...     source_id="CanESM5", experiment_id="ssp585", variable_id="tas",
            ...     table_id="Amon", member_id="r1i1p1f1", grid_label="gn",
            ...     version="20190101", activity_id="ScenarioMIP",
            ... )
            >>> s.variable_id
            'tas'

            ```
    """

    zstore: str
    source_id: str
    experiment_id: str
    variable_id: str
    table_id: str
    member_id: str
    grid_label: str
    version: str
    activity_id: str = ""

    @property
    def slug(self) -> str:
        """A filesystem-safe stem identifying this store.

        Returns:
            str: `<source>_<experiment>_<variable>_<table>_<member>_<grid>` with
                any path separators removed.
        """
        parts = [
            self.source_id,
            self.experiment_id,
            self.variable_id,
            self.table_id,
            self.member_id,
            self.grid_label,
        ]
        return "_".join(str(p).replace("/", "-") for p in parts if p)

`slug` `property` #

A filesystem-safe stem identifying this store.

Returns:

Name	Type	Description
`str`	`str`	`<source>_<experiment>_<variable>_<table>_<member>_<grid>` with any path separators removed.

`StoreResolver` #

Resolve CMIP6 facet tuples to zstore URIs over the consolidated CSV.

Fetches + caches the CSV once, then filters it per request. Construct with the catalog's csv_url + facet_columns; inject a frame= or cache_path= to run offline.

Parameters:

Name	Type	Description	Default
`csv_url`	`str`	URL of the consolidated-stores CSV.	required
`facet_columns`	`list[str]`	The CSV facet column names (from the catalog), kept as schema documentation; `resolve()` filters on :data:`_FILTER_FACETS`, not on this list.	required
`cache_path`	`Path \| str \| None`	Where to cache the downloaded CSV. Defaults to :func:`default_cache_path`.	`None`
`frame`	`DataFrame \| None`	A pre-loaded `DataFrame` to use verbatim, skipping all I/O.	`None`
`timeout`	`float`	Per-request network timeout, in seconds.	`120.0`

Source code in libs/providers/atmosphere/src/earthlens/cmip6/resolver.py

class StoreResolver:
    """Resolve CMIP6 facet tuples to `zstore` URIs over the consolidated CSV.

    Fetches + caches the CSV once, then filters it per request. Construct with
    the catalog's `csv_url` + `facet_columns`; inject a `frame=` or `cache_path=`
    to run offline.

    Args:
        csv_url: URL of the consolidated-stores CSV.
        facet_columns: The CSV facet column names (from the catalog), kept as
            schema documentation; `resolve()` filters on :data:`_FILTER_FACETS`,
            not on this list.
        cache_path: Where to cache the downloaded CSV. Defaults to
            :func:`default_cache_path`.
        frame: A pre-loaded `DataFrame` to use verbatim, skipping all I/O.
        timeout: Per-request network timeout, in seconds.
    """

    def __init__(
        self,
        csv_url: str,
        facet_columns: list[str],
        *,
        cache_path: Path | str | None = None,
        frame: pd.DataFrame | None = None,
        timeout: float = 120.0,
    ):
        self.csv_url = csv_url
        self.facet_columns = list(facet_columns)
        self.cache_path = (
            Path(cache_path) if cache_path is not None else default_cache_path()
        )
        self.timeout = timeout
        self._frame = frame

    @property
    def frame(self) -> pd.DataFrame:
        """The consolidated-stores table, loaded + cached on first access.

        Returns:
            pandas.DataFrame: The full store index.
        """
        if self._frame is None:
            self._frame = self._load()
        return self._frame

    def _load(self) -> pd.DataFrame:
        """Read the CSV into a `DataFrame`, downloading + caching it if needed.

        Returns:
            pandas.DataFrame: The parsed CSV.
        """
        import pandas as pd

        path = self._ensure_csv()
        return pd.read_csv(path, low_memory=False)

    def _ensure_csv(self) -> Path:
        """Return the cached CSV path, downloading it once if absent.

        Streams through :class:`~earthlens.base.http.HttpClient`'s atomic
        `download` (temp `.part` file + rename, cleanup on failure), then
        enforces the non-empty invariant: a zero-byte transfer unlinks the
        cache and raises rather than caching a useless file.

        Returns:
            Path: The local cache path (guaranteed to exist and be non-empty).

        Raises:
            requests.RequestException: On a transport error or a non-2xx
                status from the download (`HttpClient.download` calls
                `raise_for_status` — this covers `HTTPError` for a 4xx/5xx
                on the CSV, `ConnectionError` for a network failure, and
                `Timeout` if the transfer stalls past `self.timeout`).
            OSError: If the atomic rename of the `.part` temp to
                `cache_path` fails.
            RuntimeError: When the download succeeds but yields a
                zero-byte CSV (the cache is unlinked before raising).
        """
        if self.cache_path.exists() and self.cache_path.stat().st_size > 0:
            return self.cache_path
        client = HttpClient(
            timeout=self.timeout,
            max_retries=0,
            status_forcelist=(),
            raise_for_status=True,
        )
        client.download(
            self.csv_url,
            self.cache_path,
            chunk=1 << 20,
            atomic=True,
            progress=False,
        )
        if self.cache_path.stat().st_size == 0:
            self.cache_path.unlink(missing_ok=True)
            raise RuntimeError(f"downloaded an empty CSV from {self.csv_url}")
        return self.cache_path

    def resolve(
        self,
        *,
        source_id: str,
        experiment_id: str,
        variable_id: str,
        table_id: str,
        member_id: str | None = None,
        grid_label: str | None = None,
        version: str = "latest",
        activity_id: str | None = None,
    ) -> list[ResolvedStore]:
        """Resolve a facet tuple to the matching `zstore` store(s).

        Filters the CSV by every pinned facet (unset facets fan out), then
        reduces `version="latest"` to the newest publication per store. Returns
        one :class:`ResolvedStore` per surviving row.

        Args:
            source_id: Model (required).
            experiment_id: Scenario / experiment (required).
            variable_id: Variable (required).
            table_id: MIP table (required).
            member_id: Variant label; `None` fans out over all members.
            grid_label: Grid label; `None` fans out over all grids.
            version: `"latest"` (newest per store) or an explicit version
                string.
            activity_id: MIP; `None` leaves it unconstrained.

        Returns:
            list[ResolvedStore]: One entry per matching store. For
                `version="latest"` the entries are ordered by their identity
                facets; for an explicit version they follow CSV row order.

        Raises:
            ValueError: If no store matches; the message names the facet that
                eliminated every row and lists the values that were available.
        """
        requested = {
            "activity_id": activity_id,
            "source_id": source_id,
            "experiment_id": experiment_id,
            "variable_id": variable_id,
            "table_id": table_id,
            "member_id": member_id,
            "grid_label": grid_label,
        }
        frame = self.frame
        for facet in _FILTER_FACETS:
            value = requested.get(facet)
            if value is None or facet not in frame.columns:
                continue
            narrowed = frame[frame[facet].astype(str) == str(value)]
            if narrowed.empty:
                available = sorted(frame[facet].dropna().astype(str).unique())
                raise ValueError(
                    f"no CMIP6 store matches {facet}={value!r} for the requested "
                    f"facets so far ({self._describe(requested, facet)}); "
                    f"{self._available_hint(facet, value, available)}"
                )
            frame = narrowed
        frame = self._select_version(frame, version)
        return [self._row_to_store(row) for _, row in frame.iterrows()]

    def _select_version(self, frame: pd.DataFrame, version: str) -> pd.DataFrame:
        """Apply the version policy — `latest` (newest per store) or exact.

        Args:
            frame: The facet-filtered frame.
            version: `"latest"` or an explicit version string.

        Returns:
            pandas.DataFrame: The frame reduced to the chosen version(s),
                sorted by the store slug facets.

        Raises:
            ValueError: If an explicit `version` matches nothing.
        """
        if "version" not in frame.columns:
            return frame
        if str(version).lower() != "latest":
            narrowed = frame[frame["version"].astype(str) == str(version)]
            if narrowed.empty:
                available = sorted(frame["version"].dropna().astype(str).unique())
                raise ValueError(
                    f"no CMIP6 store matches version={version!r}; "
                    f"available versions: {available}."
                )
            return narrowed
        keys = [f for f in _IDENTITY_FACETS if f in frame.columns]
        ordered = frame.sort_values("version", ascending=False)
        if keys:
            ordered = ordered.drop_duplicates(subset=keys, keep="first")
            ordered = ordered.sort_values(keys)
        return ordered

    @staticmethod
    def _row_to_store(row: Any) -> ResolvedStore:
        """Build a :class:`ResolvedStore` from one CSV row.

        Args:
            row: A `pandas.Series` for one store.

        Returns:
            ResolvedStore: The typed store descriptor.
        """
        return ResolvedStore(
            zstore=str(row["zstore"]),
            source_id=str(row.get("source_id", "")),
            experiment_id=str(row.get("experiment_id", "")),
            variable_id=str(row.get("variable_id", "")),
            table_id=str(row.get("table_id", "")),
            member_id=str(row.get("member_id", "")),
            grid_label=str(row.get("grid_label", "")),
            version=str(row.get("version", "")),
            activity_id=str(row.get("activity_id", "")),
        )

    @staticmethod
    def _describe(requested: dict[str, str | None], up_to: str) -> str:
        """Summarise the facets pinned before the one that failed.

        Args:
            requested: The full requested-facet mapping.
            up_to: The facet that eliminated every row.

        Returns:
            str: A `k=v` list of the facets applied before `up_to`, or
                `"no prior facets"` when it was the first.
        """
        applied = []
        for facet in _FILTER_FACETS:
            if facet == up_to:
                break
            value = requested.get(facet)
            if value is not None:
                applied.append(f"{facet}={value}")
        return ", ".join(applied) if applied else "no prior facets"

    @staticmethod
    def _available_hint(
        facet: str, value: str, available: list[str], limit: int = 20
    ) -> str:
        """Build a concise "available values" hint with a did-you-mean.

        Keeps the miss message readable on a high-cardinality facet (a
        `source_id` / `variable_id` miss can leave 100s of candidates) by
        capping the listed values and adding the closest match as a
        did-you-mean, mirroring the catalog's `difflib` lookups.

        Args:
            facet: The facet that eliminated every row.
            value: The requested (unmatched) value.
            available: The sorted values that were still available.
            limit: Maximum number of values to list before truncating.

        Returns:
            str: `available {facet}: [v1, …][, +K more]. Did you mean 'x'?`.
        """
        import difflib

        close = difflib.get_close_matches(str(value), available, n=1)
        hint = f" Did you mean {close[0]!r}?" if close else ""
        shown = available[:limit]
        tail = f", +{len(available) - limit} more" if len(available) > limit else ""
        return f"available {facet}: {shown}{tail}.{hint}"

`frame` `property` #

The consolidated-stores table, loaded + cached on first access.

Returns:

Type	Description
`DataFrame`	pandas.DataFrame: The full store index.

`resolve(*, source_id, experiment_id, variable_id, table_id, member_id=None, grid_label=None, version='latest', activity_id=None)` #

Resolve a facet tuple to the matching zstore store(s).

Filters the CSV by every pinned facet (unset facets fan out), then reduces version="latest" to the newest publication per store. Returns one :class:ResolvedStore per surviving row.

Parameters:

Name	Type	Description	Default
`source_id`	`str`	Model (required).	required
`experiment_id`	`str`	Scenario / experiment (required).	required
`variable_id`	`str`	Variable (required).	required
`table_id`	`str`	MIP table (required).	required
`member_id`	`str \| None`	Variant label; `None` fans out over all members.	`None`
`grid_label`	`str \| None`	Grid label; `None` fans out over all grids.	`None`
`version`	`str`	`"latest"` (newest per store) or an explicit version string.	`'latest'`
`activity_id`	`str \| None`	MIP; `None` leaves it unconstrained.	`None`

Returns:

Type	Description
`list[ResolvedStore]`	list[ResolvedStore]: One entry per matching store. For `version="latest"` the entries are ordered by their identity facets; for an explicit version they follow CSV row order.

Raises:

Type	Description
`ValueError`	If no store matches; the message names the facet that eliminated every row and lists the values that were available.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/resolver.py

def resolve(
    self,
    *,
    source_id: str,
    experiment_id: str,
    variable_id: str,
    table_id: str,
    member_id: str | None = None,
    grid_label: str | None = None,
    version: str = "latest",
    activity_id: str | None = None,
) -> list[ResolvedStore]:
    """Resolve a facet tuple to the matching `zstore` store(s).

    Filters the CSV by every pinned facet (unset facets fan out), then
    reduces `version="latest"` to the newest publication per store. Returns
    one :class:`ResolvedStore` per surviving row.

    Args:
        source_id: Model (required).
        experiment_id: Scenario / experiment (required).
        variable_id: Variable (required).
        table_id: MIP table (required).
        member_id: Variant label; `None` fans out over all members.
        grid_label: Grid label; `None` fans out over all grids.
        version: `"latest"` (newest per store) or an explicit version
            string.
        activity_id: MIP; `None` leaves it unconstrained.

    Returns:
        list[ResolvedStore]: One entry per matching store. For
            `version="latest"` the entries are ordered by their identity
            facets; for an explicit version they follow CSV row order.

    Raises:
        ValueError: If no store matches; the message names the facet that
            eliminated every row and lists the values that were available.
    """
    requested = {
        "activity_id": activity_id,
        "source_id": source_id,
        "experiment_id": experiment_id,
        "variable_id": variable_id,
        "table_id": table_id,
        "member_id": member_id,
        "grid_label": grid_label,
    }
    frame = self.frame
    for facet in _FILTER_FACETS:
        value = requested.get(facet)
        if value is None or facet not in frame.columns:
            continue
        narrowed = frame[frame[facet].astype(str) == str(value)]
        if narrowed.empty:
            available = sorted(frame[facet].dropna().astype(str).unique())
            raise ValueError(
                f"no CMIP6 store matches {facet}={value!r} for the requested "
                f"facets so far ({self._describe(requested, facet)}); "
                f"{self._available_hint(facet, value, available)}"
            )
        frame = narrowed
    frame = self._select_version(frame, version)
    return [self._row_to_store(row) for _, row in frame.iterrows()]

`default_cache_path()` #

Return the default on-disk location for the cached CSV.

Resolved from the shared earthlens cache directory (set_cache_dir() / EARTHLENS_CACHE). The CMIP6 CSV lands under a cmip6/ subdirectory.

Returns:

Name	Type	Description
`Path`	`Path`	`<cache_dir()>/cmip6/pangeo-cmip6.csv`.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/resolver.py

def default_cache_path() -> Path:
    """Return the default on-disk location for the cached CSV.

    Resolved from the shared earthlens cache directory (`set_cache_dir()` /
    `EARTHLENS_CACHE`). The CMIP6 CSV lands under a `cmip6/` subdirectory.

    Returns:
        Path: `<cache_dir()>/cmip6/pangeo-cmip6.csv`.
    """
    return cache_dir() / "cmip6" / "pangeo-cmip6.csv"

`earthlens.cmip6.accessor` #

Thin pyramids accessor for the CMIP6 Zarr stores (the read boundary).

Everything that opens a gs://cmip6 Zarr store, windows it, and writes a NetCDF subset lives here — the one place earthlens touches pyramids for CMIP6. earthlens itself never imports xarray / zarr / gcsfs; pyramids reads the store through GDAL's /vsigs/ multidimensional driver (no gcsfs needed), and this module only:

rewrites a gs://cmip6/<path>/ zstore URI to the GDAL ZARR:"/vsigs/..." form (:func:zstore_to_vsi);
forces anonymous GCS access with GS_NO_SIGN_REQUEST around every read (:func:anonymous_gcs) — pyramids' anon=True only sets the AWS flag, and this machine may carry ambient GCS credentials, so the flag is set explicitly;
maps a CF [start, end] date window to an integer time-index range (:func:resolve_time_window) — the gridded reader selects time by integer index, and LabeledDataset.select_time is the public path that decodes CF time / non-standard calendars;
reads the gridded (time, bbox) slice and writes it to NetCDF (:func:write_subset), a windowed read that fetches only the requested cells.

The pyramids reader classes are imported lazily behind an install-hint so the package imports (and the backend constructs) without a read ever happening.

`anonymous_gcs()` #

Force anonymous GCS access for the duration of the with block.

Sets GS_NO_SIGN_REQUEST=YES so GDAL's /vsigs/ driver reads the public gs://cmip6 bucket without signing — even when the environment carries GCS credentials (e.g. a Google Earth Engine service account) — then restores the prior value. The flag must stay live for the lazy data-chunk reads, not just the open, so wrap the whole read + write.

Yields:

Name	Type	Description
`None`	`None`	control to the `with` body.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/accessor.py

@contextlib.contextmanager
def anonymous_gcs() -> Iterator[None]:
    """Force anonymous GCS access for the duration of the `with` block.

    Sets `GS_NO_SIGN_REQUEST=YES` so GDAL's `/vsigs/` driver reads the public
    `gs://cmip6` bucket without signing — even when the environment carries GCS
    credentials (e.g. a Google Earth Engine service account) — then restores the
    prior value. The flag must stay live for the lazy data-chunk reads, not just
    the open, so wrap the whole read + write.

    Yields:
        None: control to the `with` body.
    """
    previous = os.environ.get(GS_NO_SIGN_ENV)
    os.environ[GS_NO_SIGN_ENV] = "YES"
    try:
        yield
    finally:
        if previous is None:
            os.environ.pop(GS_NO_SIGN_ENV, None)
        else:
            os.environ[GS_NO_SIGN_ENV] = previous

`resolve_time_window(zstore, variable, start=None, end=None, *, time_dim='time')` #

Map a CF [start, end] date window to an integer time-index range.

The gridded reader (:func:write_subset) selects time by integer index; CMIP6 ARCO stores do not surface CF time units through GDAL's multidim path, but LabeledDataset.select_time decodes the store's own time units + calendar (so noleap / 360_day work). The half-open window is recovered from public counts alone:

i0 = N - select_time(start=start).sizes[time] (steps at or after start)
i1 = select_time(end=end).sizes[time] (steps at or before end)

Parameters:

Name	Type	Description	Default
`zstore`	`str`	The `gs://cmip6/...` store URI.	required
`variable`	`str`	The data variable to open (keeps the read light).	required
`start`	`Any`	Inclusive window start (`datetime` / `"YYYY-MM-DD"` / `None`).	`None`
`end`	`Any`	Inclusive window end; `None` runs to the last step.	`None`
`time_dim`	`str`	Name of the time dimension.	`'time'`

Returns:

Type	Description
`tuple[int, int] \| None`	tuple[int, int] \| None: The half-open `(i0, i1)` index range, or `None` when neither bound is given (the caller should read the whole series).

Raises:

Type	Description
`ValueError`	If the window selects no timesteps.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/accessor.py

def resolve_time_window(
    zstore: str,
    variable: str,
    start: Any = None,
    end: Any = None,
    *,
    time_dim: str = "time",
) -> tuple[int, int] | None:
    """Map a CF `[start, end]` date window to an integer time-index range.

    The gridded reader (:func:`write_subset`) selects time by **integer index**;
    CMIP6 ARCO stores do not surface CF time units through GDAL's multidim path,
    but `LabeledDataset.select_time` decodes the store's own time `units` +
    `calendar` (so `noleap` / `360_day` work). The half-open window is recovered
    from public counts alone:

    * `i0 = N - select_time(start=start).sizes[time]` (steps at or after `start`)
    * `i1 = select_time(end=end).sizes[time]` (steps at or before `end`)

    Args:
        zstore: The `gs://cmip6/...` store URI.
        variable: The data variable to open (keeps the read light).
        start: Inclusive window start (`datetime` / `"YYYY-MM-DD"` / `None`).
        end: Inclusive window end; `None` runs to the last step.
        time_dim: Name of the time dimension.

    Returns:
        tuple[int, int] | None: The half-open `(i0, i1)` index range, or `None`
            when neither bound is given (the caller should read the whole series).

    Raises:
        ValueError: If the window selects no timesteps.
    """
    if start is None and end is None:
        return None
    labeled = _labeled_reader()
    with anonymous_gcs():
        # engine="zarr" forces the Zarr driver explicitly: a CMIP6 store URI ends
        # in /v<YYYYMMDD>/ (never `.zarr`), so pyramids' suffix-based auto-detect
        # would otherwise open it down the NetCDF branch — asymmetric with
        # write_subset, which opens the same store through the ZARR: /vsigs/ path.
        dataset = labeled.read_file(
            zstore, variables=[variable], anon=True, engine="zarr"
        )
        try:
            total = int(dataset.sizes.get(time_dim, 0))
            if total == 0:
                return None
            i0 = (
                0
                if start is None
                else total
                - int(
                    dataset.select_time(start=start, time_dim=time_dim).sizes[time_dim]
                )
            )
            i1 = (
                total
                if end is None
                else int(
                    dataset.select_time(end=end, time_dim=time_dim).sizes[time_dim]
                )
            )
        except ValueError as exc:
            raise ValueError(
                f"no CMIP6 timesteps fall in the window [{start}, {end}] for "
                f"{variable!r} in {zstore!r}: {exc}"
            ) from exc
        finally:
            close_quietly(dataset)
    if i1 <= i0:
        raise ValueError(
            f"no CMIP6 timesteps fall in the window [{start}, {end}] for "
            f"{variable!r} in {zstore!r}."
        )
    return (i0, i1)

`store_output_stem(store, start, end)` #

Compose a unique output-file stem for one resolved store + window.

The store version is folded in so two calls that pin different explicit version= values for the same identity (or a CSV carrying duplicate (identity, version) rows) write to distinct files instead of the second silently overwriting the first.

Parameters:

Name	Type	Description	Default
`store`	`ResolvedStore`	The resolved store (supplies the facet slug + version).	required
`start`	`Any`	Window start (its `%Y%m%d` is appended when it has one).	required
`end`	`Any`	Window end.	required

Returns:

Name	Type	Description
`str`	`str`	`<facet-slug>[_v<version>]_<startYYYYMMDD>_<endYYYYMMDD>` (the version tag is added when the store carries one; dates are omitted when unavailable).

Source code in libs/providers/atmosphere/src/earthlens/cmip6/accessor.py

def store_output_stem(store: ResolvedStore, start: Any, end: Any) -> str:
    """Compose a unique output-file stem for one resolved store + window.

    The store `version` is folded in so two calls that pin different explicit
    `version=` values for the same identity (or a CSV carrying duplicate
    `(identity, version)` rows) write to distinct files instead of the second
    silently overwriting the first.

    Args:
        store: The resolved store (supplies the facet slug + version).
        start: Window start (its `%Y%m%d` is appended when it has one).
        end: Window end.

    Returns:
        str: `<facet-slug>[_v<version>]_<startYYYYMMDD>_<endYYYYMMDD>` (the
            version tag is added when the store carries one; dates are omitted
            when unavailable).
    """
    stem = f"{store.slug}_v{store.version}" if store.version else store.slug
    tokens = [token for token in (_date_token(start), _date_token(end)) if token]
    fragment = "_".join(tokens)
    return f"{stem}_{fragment}" if fragment else stem

`write_subset(zstore, variable, *, bbox, time, out_path, crs=4326)` #

Read a (variable, time, bbox) window of a store and write it to NetCDF.

Opens the resolved Zarr store through pyramids' NetCDF reader (GDAL /vsigs/, anonymous) and writes the windowed slice — only the requested cells are fetched. The read + write both run inside :func:anonymous_gcs.

Parameters:

Name	Type	Description	Default
`zstore`	`str`	The `gs://cmip6/...` store URI.	required
`variable`	`str`	The data variable to read (`"tas"`).	required
`bbox`	`tuple[float, float, float, float] \| None`	`(west, south, east, north)` crop in `crs`, or `None` for the full grid.	required
`time`	`int \| tuple[int, int] \| slice \| None`	Integer time selector (`int` / `(start, stop)` / `slice` / `None`). `None` is valid only when the time dimension has length 1.	required
`out_path`	`Path \| str`	Destination path for the written NetCDF.	required
`crs`	`int \| str`	CRS of `bbox`. Defaults to `4326` (lon/lat).	`4326`

Returns:

Name	Type	Description
`Path`	`Path`	The written NetCDF path.

Source code in libs/providers/atmosphere/src/earthlens/cmip6/accessor.py

def write_subset(
    zstore: str,
    variable: str,
    *,
    bbox: tuple[float, float, float, float] | None,
    time: int | tuple[int, int] | slice | None,
    out_path: Path | str,
    crs: int | str = 4326,
) -> Path:
    """Read a `(variable, time, bbox)` window of a store and write it to NetCDF.

    Opens the resolved Zarr store through pyramids' `NetCDF` reader (GDAL
    `/vsigs/`, anonymous) and writes the windowed slice — only the requested
    cells are fetched. The read + write both run inside :func:`anonymous_gcs`.

    Args:
        zstore: The `gs://cmip6/...` store URI.
        variable: The data variable to read (`"tas"`).
        bbox: `(west, south, east, north)` crop in `crs`, or `None` for the full
            grid.
        time: Integer time selector (`int` / `(start, stop)` / `slice` / `None`).
            `None` is valid only when the time dimension has length 1.
        out_path: Destination path for the written NetCDF.
        crs: CRS of `bbox`. Defaults to `4326` (lon/lat).

    Returns:
        Path: The written NetCDF path.
    """
    netcdf = _netcdf_reader()
    vsi = zstore_to_vsi(zstore)
    out = Path(out_path)
    with anonymous_gcs():
        container = netcdf.read_file(vsi)
        try:
            subset = container.subset(variable, time=time, bbox=bbox, crs=crs)
            subset.to_file(str(out))
        finally:
            close_quietly(container)
    return out

`zstore_to_vsi(zstore)` #

Rewrite a gs:// store URI to the GDAL ZARR:"/vsigs/..." form.

Parameters:

Name	Type	Description	Default
`zstore`	`str`	A `gs://cmip6/<path>/` store URI (or an already-rewritten `/vsigs/...` path).	required

Returns:

Name	Type	Description
`str`	`str`	The `ZARR:"/vsigs/<bucket>/<path>/"` path GDAL's multidim Zarr driver opens.

Raises:

Type	Description
`ValueError`	If `zstore` is neither a `gs://` URI nor a `/vsigs/` path.

Examples:

Rewrite a store URI:

>>> from earthlens.cmip6.accessor import zstore_to_vsi
>>> zstore_to_vsi("gs://cmip6/CMIP6/ScenarioMIP/x/tas/gn/v1/")
'ZARR:"/vsigs/cmip6/CMIP6/ScenarioMIP/x/tas/gn/v1/"'

Source code in libs/providers/atmosphere/src/earthlens/cmip6/accessor.py

def zstore_to_vsi(zstore: str) -> str:
    """Rewrite a `gs://` store URI to the GDAL `ZARR:"/vsigs/..."` form.

    Args:
        zstore: A `gs://cmip6/<path>/` store URI (or an already-rewritten
            `/vsigs/...` path).

    Returns:
        str: The `ZARR:"/vsigs/<bucket>/<path>/"` path GDAL's multidim Zarr
            driver opens.

    Raises:
        ValueError: If `zstore` is neither a `gs://` URI nor a `/vsigs/` path.

    Examples:
        - Rewrite a store URI:
            ```python
            >>> from earthlens.cmip6.accessor import zstore_to_vsi
            >>> zstore_to_vsi("gs://cmip6/CMIP6/ScenarioMIP/x/tas/gn/v1/")
            'ZARR:"/vsigs/cmip6/CMIP6/ScenarioMIP/x/tas/gn/v1/"'

            ```
    """
    if zstore.startswith("gs://"):
        vsi = "/vsigs/" + zstore[len("gs://") :]
    elif zstore.startswith("/vsigs/"):
        vsi = zstore
    elif zstore.startswith('ZARR:"'):
        return zstore
    else:
        raise ValueError(
            f"cannot rewrite {zstore!r} to a GDAL /vsigs/ path: expected a "
            "'gs://' store URI."
        )
    return f'ZARR:"{vsi}"'