AbstractDataSource

Base classes that define the interface for all data source implementations.

earthlens.base.AbstractDataSource

Bases: ABC

Blueprint for every concrete data-source backend.

Subclasses encapsulate the request shape, authentication, and download orchestration for a single provider (CHIRPS, ERA5 on AWS S3, ECMWF CDS, Google Earth Engine). The base class wires the abstract hooks (:meth:_initialize, :meth:_create_grid, :meth:_check_input_dates) into a uniform __init__ shape and exposes a single :meth:download entry point.

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	Class-level declaration of the natural output shape this backend emits. Read by :class:earthlens.earthlens.EarthLens at facade download() time to gate the aggregate= argument: "raster" accepts it (the existing pyramids-backed aggregate_netcdf flow); "vector" and "tabular" reject it with :class:NotImplementedError; "mixed" forwards it unchanged. Subclasses override the class attribute; the default is "raster". Most backends fix OUTPUT_KIND as a class attribute. A few backends whose output shape is only known once the requested dataset(s) are resolved set it per instance in __init__ instead — a sanctioned override: earthdata and eumetsat copy the resolved dataset's output_kind onto self.OUTPUT_KIND, and tropycal sets "tabular" for its ships product (else "vector"). The facade reads the instance attribute, so both forms work.

Source code in src/earthlens/base/abstractdatasource.py

class AbstractDataSource(ABC):
    """Blueprint for every concrete data-source backend.

    Subclasses encapsulate the request shape, authentication, and
    download orchestration for a single provider (CHIRPS, ERA5 on AWS
    S3, ECMWF CDS, Google Earth Engine). The base class wires the
    abstract hooks (:meth:`_initialize`, :meth:`_create_grid`,
    :meth:`_check_input_dates`) into a uniform `__init__` shape and
    exposes a single :meth:`download` entry point.

    Attributes:
        OUTPUT_KIND: Class-level declaration of the natural output
            shape this backend emits. Read by
            :class:`earthlens.earthlens.EarthLens` at facade
            `download()` time to gate the `aggregate=` argument:
            `"raster"` accepts it (the existing pyramids-backed
            `aggregate_netcdf` flow); `"vector"` and `"tabular"`
            reject it with :class:`NotImplementedError`; `"mixed"`
            forwards it unchanged. Subclasses override the class
            attribute; the default is `"raster"`.

            Most backends fix `OUTPUT_KIND` as a class attribute. A few
            backends whose output shape is only known once the requested
            dataset(s) are resolved set it **per instance** in
            `__init__` instead — a sanctioned override: earthdata and
            eumetsat copy the resolved dataset's `output_kind` onto
            `self.OUTPUT_KIND`, and tropycal sets `"tabular"` for its
            `ships` product (else `"vector"`). The facade reads the
            instance attribute, so both forms work.
    """

    OUTPUT_KIND: OutputKind = "raster"

    def __init_subclass__(cls, **kwargs: Any) -> None:
        """Give every backend the facade's ergonomic constructor sugar.

        Wraps each concrete backend's `__init__` so that — whether reached
        through the `EarthLens` facade or by constructing the backend
        class directly — it also accepts:

        * `aoi` (+ `buffer`): any shape :func:`earthlens.base.spatial.normalize_aoi`
          understands, reduced to `lat_lim` / `lon_lim`; a backend that
          declares its own `aoi` (WorldPop) keeps it;
        * `cadence`: a clearer alias for `temporal_resolution`;
        * `dataset`: split out of a single-key `variables` dict (or passed
          through to a backend with a native `dataset`, e.g. S3).

        The original `__init__` is preserved as the wrapper's `__wrapped__`,
        so signature introspection (e.g. `EarthLens.options_for`) and the
        facade's kwarg validation still see the backend's real parameters.

        Note:
            No backend currently subclasses another backend. If one ever
            does, its `__init__` must not forward the ergonomic kwargs
            (`aoi` / `buffer` / `cadence` / `dataset`) up to
            `super().__init__()`: the parent's wrapper would resolve them a
            second time (e.g. re-running `resolve_aoi`). Forward only the
            already-resolved native parameters (`lat_lim` / `lon_lim` /
            `temporal_resolution` / `variables`) instead.
        """
        super().__init_subclass__(**kwargs)
        orig = cls.__dict__.get("__init__")
        if orig is None or getattr(orig, "_ergonomic", False):
            return
        params = inspect.signature(orig).parameters
        native_aoi = "aoi" in params
        native_dataset = "dataset" in params

        @functools.wraps(orig)
        def __init__(
            self, *args, aoi=None, buffer=None, cadence=None, dataset=None, **kw
        ):
            clip_geometry = None
            if cadence is not None:
                kw["temporal_resolution"] = cadence
            if dataset is not None:
                if native_dataset:
                    kw["dataset"] = dataset
                elif isinstance(kw.get("variables"), dict):
                    raise ValueError(
                        "pass variables= as a list when using dataset=, or omit "
                        "dataset= and key the variables dict yourself"
                    )
                else:
                    v = kw.get("variables")
                    kw["variables"] = {dataset: list(v) if v is not None else []}
            if aoi is not None:
                if native_aoi:
                    if buffer is not None:
                        raise ValueError(
                            f"buffer= is not supported by {cls.__name__}, which "
                            "interprets aoi= itself"
                        )
                    kw["aoi"] = aoi
                else:
                    if kw.get("lat_lim") is not None or kw.get("lon_lim") is not None:
                        raise ValueError(
                            "pass either aoi= or lat_lim=/lon_lim=, not both"
                        )
                    from earthlens.base.spatial import resolve_aoi

                    kw["lat_lim"], kw["lon_lim"], clip_geometry = resolve_aoi(
                        aoi, buffer=buffer
                    )
            elif buffer is not None:
                raise ValueError(
                    "buffer= only applies to a point aoi=(lon, lat); pass aoi= too"
                )
            orig(self, *args, **kw)
            if clip_geometry is not None:
                self._attach_clip_geometry(clip_geometry)

        __init__._ergonomic = True
        cls.__init__ = __init__

    def __init__(
        self,
        start: str,
        end: str,
        variables: dict[str, list[str]] | list[str],
        lat_lim: list[float],
        lon_lim: list[float],
        temporal_resolution: str = "daily",
        fmt: str = "%Y-%m-%d",
        path: Path | str = "",
    ):
        """Initialize a data source instance.

        Captures the return values of the abstract hooks so subclasses
        do not have to wire them onto `self` themselves:

        * `self.client` — whatever :meth:`_initialize` returns (a CDS
          client, an S3 client, `None` for FTP). Subclasses that
          assign `self.client` inside :meth:`_initialize` (e.g.
          :class:`S3`) keep their own assignment; the parent only sets
          the attribute when :meth:`_initialize` returns a non-`None`
          value.
        * `self.space` — the dict returned by :meth:`_create_grid`,
          containing `lat_lim` and `lon_lim`. Subclasses that
          override :meth:`_create_grid` to set attributes directly (e.g.
          :class:`CHIRPS`) and return `None` are unaffected.
        * `self.time` — the dict returned by :meth:`_check_input_dates`,
          containing `start_date`, `end_date`, `time_freq` and
          `dates`. Same opt-in semantics as `self.space`.
        * `self.root_dir` — the absolute :class:`pathlib.Path` of the
          output directory. `self.path` is kept as a legacy alias so
          older backends (CHIRPS, S3) continue to work.

        Args:
            start: Inclusive start date as a string. Format controlled
                by `fmt`. Defaults to `None`.
            end: Inclusive end date as a string. Defaults to `None`.
            variables: List of variable short codes to download.
            temporal_resolution: `"daily"` or `"monthly"`. Defaults
                to `"daily"`.
            lat_lim: `[lat_min, lat_max]`.
            lon_lim: `[lon_min, lon_max]`.
            fmt: `strptime` format for `start` / `end`. Defaults
                to `"%Y-%m-%d"`.
            path: Output directory. Created if it does not exist.
                Defaults to the current working directory.
        """
        client = self._initialize()
        if client is not None:
            self.client = client

        self.temporal_resolution = temporal_resolution
        self.vars = variables

        space = self._create_grid(lat_lim, lon_lim)
        if isinstance(space, SpatialExtent):
            self.space = space
        elif isinstance(space, dict):
            self.space = SpatialExtent.from_pairs(
                lat_lim=space["lat_lim"], lon_lim=space["lon_lim"]
            )

        time = self._check_input_dates(start, end, temporal_resolution, fmt)
        if isinstance(time, TemporalExtent):
            self.time = time
        elif isinstance(time, dict):
            self.time = TemporalExtent(
                start_date=time["start_date"],
                end_date=time["end_date"],
                resolution=time.get("resolution", time.get("time_freq")),
                dates=time["dates"],
            )

        self.root_dir = Path(path).absolute()
        self.path = self.root_dir
        if not os.path.exists(self.root_dir):
            os.makedirs(self.root_dir)

    def _attach_clip_geometry(self, geometry: Any) -> None:
        """Record a polygon mask on `self.space` for precise clipping.

        Called by the ergonomic `__init__` wrapper when the request's
        `aoi=` was a polygon rather than a plain bbox. The geometry is
        stored on the (frozen) :class:`SpatialExtent` via a copy so that
        raster backends clipping through `pyramids.Dataset.crop` can mask
        the fetched bbox down to the exact shape. A no-op when `self.space`
        is not a :class:`SpatialExtent`.

        Args:
            geometry: A WGS84 `GeoDataFrame` polygon mask.
        """
        space = getattr(self, "space", None)
        if isinstance(space, SpatialExtent):
            self.space = space.model_copy(update={"geometry": geometry})

    def authenticate(self) -> AbstractDataSource:
        """Eagerly establish the backend's authenticated connection.

        The explicit, fail-fast counterpart to the lazy authentication
        that otherwise happens on the first :meth:`download` / `search`:
        it opens the network client for backends that have one (those
        mixing in :class:`LazyClientMixin` — e.g. GEE, ECMWF, STAC) or
        runs the credential `configure()` step for backends that hold an
        auth object (CMEMS, Earthdata, EUMETSAT, …), raising
        :class:`~earthlens.base.AuthenticationError` on failure. It is a
        no-op for credential-free backends (CHIRPS, GDACS, Overture, …),
        and is idempotent.

        Returns:
            The backend instance, so callers can chain
            `EarthLens(...).authenticate().download()`.

        Raises:
            AuthenticationError: If the backend cannot authenticate.
        """
        if isinstance(self, LazyClientMixin):
            # Accessing `client` runs the cached `_open_client` (auth).
            _ = self.client
        elif getattr(self, "_auth", None) is not None:
            self._auth.configure()
        return self

    @abstractmethod
    def _check_input_dates(
        self, start: str, end: str, temporal_resolution: str, fmt: str
    ):
        """Check validity of input dates. Called by `__init__`."""
        pass

    @abstractmethod
    def _initialize(self, *args, **kwargs):
        """Initialize connection with the data source server (for non-FTP servers).

        Called once by :meth:`__init__`; the return value is captured
        into `self.client` when non-`None`.
        """
        pass

    @abstractmethod
    def _create_grid(self, lat_lim: list, lon_lim: list):
        """Create a grid from the lat/lon boundaries. Called by `__init__`."""
        pass

    @abstractmethod
    def download(self):
        """Download every requested variable and return the produced artifacts.

        The return shape tracks :attr:`OUTPUT_KIND`: `"raster"` /
        `"mixed"` file-writing backends return the list of written
        paths (`list[Path]`); `"vector"` backends return an in-memory
        `FeatureCollection` (radar returns a `GeoDataFrame`);
        `"tabular"` backends return a `pandas.DataFrame`. Every backend
        now returns its produced artifacts (the legacy CHIRPS / ECMWF
        backends return their written `list[Path]` and also leave the
        files on disk under `self.root_dir`).

        Partial-failure policy is per backend and currently varies:
        most multi-item backends are **skip-and-continue** — a single
        failed `(dataset, variable)` / chunk / sensor is logged and the
        batch proceeds, with a success/failure summary at the end
        (CHIRPS, CMEMS, FDSN, FIRMS, …) — while single-shot backends
        propagate the error. NWP exposes this as an explicit
        `errors="warn" | "raise" | "ignore"` argument; new backends
        with a per-item loop should follow that `errors=` convention so
        the policy becomes uniformly caller-controllable.
        """
        # loop over dates if the downloaded rasters/netcdf are for a specific date out of the required
        # list of dates
        pass

    def _download_dataset(self):
        """Download a single variable/dataset (called by :meth:`download`)."""
        pass

    @abstractmethod
    def _api(self, *args, **kwargs):
        """Send / receive a single request to the data source server.

        Called by :meth:`download` (or :meth:`_download_dataset`) once
        per `(dataset, variable)` pair. The signature is
        backend-specific.

        New backends (C3 onward) should implement :meth:`_search` and
        :meth:`_fetch` instead and let the default
        :meth:`_api_via_search_fetch` compose them; existing backends
        (CHIRPS, S3, ECMWF, GEE) continue to override `_api` directly.
        """
        pass

    # ------------------------------------------------------------------
    # C3 — optional search/fetch decomposition.
    #
    # The existing four backends (CHIRPS, S3, ECMWF, GEE) keep their
    # `_api` overrides unchanged: nothing below is abstract, so they do
    # not have to implement `_search` / `_fetch` to stay green.
    #
    # New backends (earthlens.stac, earthlens.earthdata, earthlens.fdsn,
    # earthlens.openaq, …) should override `_search` and `_fetch`
    # instead — `_search` returns a list of `RemoteProduct`s and
    # `_fetch` consumes them. The :meth:`_api_via_search_fetch` helper
    # is the canonical composition; backends can opt into it by
    # overriding `_api` as `return self._api_via_search_fetch()`.
    # ------------------------------------------------------------------

    def _search(self) -> list[RemoteProduct]:
        """List the remote products that satisfy this download request.

        Default raises `NotImplementedError` so backends that do not
        opt into the search/fetch split (the four shipped before C3)
        keep their `_api`-only flow unchanged. Backends that opt in
        override this to return one `RemoteProduct` per item the
        server's catalog says they should download.

        The split exists to make dry-run inspection cheap (`_search`
        does not hit the bulk-download endpoint) and to make
        per-product parallelism explicit (`_fetch` is the
        parallelisable half).

        Returns:
            list[RemoteProduct]: One item per product to download.
                The empty list is a legal result (the catalog matched
                nothing) and short-circuits `_api_via_search_fetch`
                without ever calling `_fetch`.

        Raises:
            NotImplementedError: When the subclass keeps the legacy
                `_api`-only flow. The message names the subclass
                class so the user can find the offending backend.
        """
        raise NotImplementedError(
            f"{type(self).__name__} does not implement _search; "
            f"either override _api directly (legacy) or override both "
            f"_search and _fetch (post-C3)."
        )

    def _count(self) -> int:
        """Return how many products :meth:`_search` would yield, without fetching.

        Default implementation runs :meth:`_search` and counts the
        result. Backends with a cheap server-side total (e.g. a STAC
        `numberMatched` read with `limit=1`) should override this to
        avoid materialising the whole product list.

        Returns:
            int: The number of products the current request matches.

        Raises:
            NotImplementedError: When the backend keeps the legacy
                `_api`-only flow and implements no :meth:`_search`.
        """
        return len(self._search())

    def _fetch(self, products: list[RemoteProduct]) -> list[Any]:
        """Download the bytes of every product `_search` returned.

        Default raises `NotImplementedError` (see `_search`).
        Backends that opt into the search/fetch split override this
        to iterate over `products` — either sequentially or via
        `joblib.Parallel` / `concurrent.futures` — and write each
        one to disk (or build it in memory).

        Args:
            products: The list returned by `_search` (or a
                user-filtered subset). The empty list is allowed and
                returns an empty list.

        Returns:
            list[Any]: One element per product, in `products` order.
                The element type tracks :attr:`OUTPUT_KIND`: written
                `Path`s for `"raster"` / `"mixed"`, `FeatureCollection`
                fragments for `"vector"`, and `DataFrame` fragments for
                `"tabular"` (these are concatenated by the backend's
                `download`). Empty list when `products` is empty (no-op
                fetch is legal).

        Raises:
            NotImplementedError: When the subclass keeps the legacy
                `_api`-only flow.
        """
        raise NotImplementedError(
            f"{type(self).__name__} does not implement _fetch; "
            f"either override _api directly (legacy) or override both "
            f"_search and _fetch (post-C3)."
        )

    def _api_via_search_fetch(self) -> list[Any]:
        """Canonical `_api` body for backends using the C3 split.

        Backends that override `_search` and `_fetch` usually want
        `_api` to just compose them; this helper is that
        composition, factored once so each new backend's `_api`
        body becomes a single line:

        ```python
        def _api(self):
            return self._api_via_search_fetch()
        ```

        The helper short-circuits on an empty search result so
        `_fetch` is only called when there is something to fetch —
        a tiny but meaningful win when many backends are queried in
        parallel and most return nothing.

        Returns:
            list[Any]: Whatever `_fetch` returned (element type tracks
                :attr:`OUTPUT_KIND` — see :meth:`_fetch`). An empty list
                when `_search` returned no products.
        """
        products = self._search()
        if not products:
            return []
        return self._fetch(products)

    def _fetch_one(self, product: RemoteProduct) -> Any:
        """Fetch a single product — the per-product hook for `_search_fetch_each`.

        Default raises `NotImplementedError`. Backends that want a
        per-item progress bar override this (instead of, or alongside,
        the whole-list `_fetch`) so `_search_fetch_each` can map it over
        the `_search` results under a `tqdm` bar.

        Raises:
            NotImplementedError: When the backend does not opt into the
                per-product fetch hook.
        """
        raise NotImplementedError(
            f"{type(self).__name__} does not implement _fetch_one."
        )

    def _search_fetch_each(
        self,
        *,
        progress_bar: bool = False,
        desc: str | None = None,
        unit: str = "item",
    ) -> list[Any]:
        """C3 composition with an optional per-product `tqdm` progress bar.

        Like :meth:`_api_via_search_fetch`, but maps the per-product
        :meth:`_fetch_one` hook over the `_search` results so a `tqdm`
        bar can show per-item progress — the shared form of the
        progress-aware composition several backends (FIRMS, OpenAQ)
        previously duplicated. Backends that fetch the whole product
        list at once, or need bespoke progress / partial-failure
        handling (e.g. CMEMS), keep their own composition.

        Args:
            progress_bar: Show the per-product `tqdm` bar when `True`.
            desc: `tqdm` description; defaults to the class name.
            unit: `tqdm` unit label.

        Returns:
            list[Any]: One :meth:`_fetch_one` result per product
                (element type tracks :attr:`OUTPUT_KIND`), or `[]` when
                `_search` matched nothing.
        """
        products = self._search()
        if not products:
            return []
        from tqdm import tqdm

        iterator = tqdm(
            products,
            disable=not progress_bar,
            desc=desc or type(self).__name__,
            unit=unit,
        )
        return [self._fetch_one(product) for product in iterator]

__init__(start, end, variables, lat_lim, lon_lim, temporal_resolution='daily', fmt='%Y-%m-%d', path='')

Initialize a data source instance.

Captures the return values of the abstract hooks so subclasses do not have to wire them onto self themselves:

• self.client — whatever :meth:_initialize returns (a CDS client, an S3 client, None for FTP). Subclasses that assign self.client inside :meth:_initialize (e.g. :class:S3) keep their own assignment; the parent only sets the attribute when :meth:_initialize returns a non-None value.
• self.space — the dict returned by :meth:_create_grid, containing lat_lim and lon_lim. Subclasses that override :meth:_create_grid to set attributes directly (e.g. :class:CHIRPS) and return None are unaffected.
• self.time — the dict returned by :meth:_check_input_dates, containing start_date, end_date, time_freq and dates. Same opt-in semantics as self.space.
• self.root_dir — the absolute :class:pathlib.Path of the output directory. self.path is kept as a legacy alias so older backends (CHIRPS, S3) continue to work.

Parameters:

Name	Type	Description	Default
start	str	Inclusive start date as a string. Format controlled by fmt. Defaults to None.	required
end	str	Inclusive end date as a string. Defaults to None.	required
variables	dict[str, list[str]] | list[str]	List of variable short codes to download.	required
temporal_resolution	str	"daily" or "monthly". Defaults to "daily".	'daily'
lat_lim	list[float]	[lat_min, lat_max].	required
lon_lim	list[float]	[lon_min, lon_max].	required
fmt	str	strptime format for start / end. Defaults to "%Y-%m-%d".	'%Y-%m-%d'
path	Path | str	Output directory. Created if it does not exist. Defaults to the current working directory.	''

Source code in src/earthlens/base/abstractdatasource.py

def __init__(
    self,
    start: str,
    end: str,
    variables: dict[str, list[str]] | list[str],
    lat_lim: list[float],
    lon_lim: list[float],
    temporal_resolution: str = "daily",
    fmt: str = "%Y-%m-%d",
    path: Path | str = "",
):
    """Initialize a data source instance.

    Captures the return values of the abstract hooks so subclasses
    do not have to wire them onto `self` themselves:

    * `self.client` — whatever :meth:`_initialize` returns (a CDS
      client, an S3 client, `None` for FTP). Subclasses that
      assign `self.client` inside :meth:`_initialize` (e.g.
      :class:`S3`) keep their own assignment; the parent only sets
      the attribute when :meth:`_initialize` returns a non-`None`
      value.
    * `self.space` — the dict returned by :meth:`_create_grid`,
      containing `lat_lim` and `lon_lim`. Subclasses that
      override :meth:`_create_grid` to set attributes directly (e.g.
      :class:`CHIRPS`) and return `None` are unaffected.
    * `self.time` — the dict returned by :meth:`_check_input_dates`,
      containing `start_date`, `end_date`, `time_freq` and
      `dates`. Same opt-in semantics as `self.space`.
    * `self.root_dir` — the absolute :class:`pathlib.Path` of the
      output directory. `self.path` is kept as a legacy alias so
      older backends (CHIRPS, S3) continue to work.

    Args:
        start: Inclusive start date as a string. Format controlled
            by `fmt`. Defaults to `None`.
        end: Inclusive end date as a string. Defaults to `None`.
        variables: List of variable short codes to download.
        temporal_resolution: `"daily"` or `"monthly"`. Defaults
            to `"daily"`.
        lat_lim: `[lat_min, lat_max]`.
        lon_lim: `[lon_min, lon_max]`.
        fmt: `strptime` format for `start` / `end`. Defaults
            to `"%Y-%m-%d"`.
        path: Output directory. Created if it does not exist.
            Defaults to the current working directory.
    """
    client = self._initialize()
    if client is not None:
        self.client = client

    self.temporal_resolution = temporal_resolution
    self.vars = variables

    space = self._create_grid(lat_lim, lon_lim)
    if isinstance(space, SpatialExtent):
        self.space = space
    elif isinstance(space, dict):
        self.space = SpatialExtent.from_pairs(
            lat_lim=space["lat_lim"], lon_lim=space["lon_lim"]
        )

    time = self._check_input_dates(start, end, temporal_resolution, fmt)
    if isinstance(time, TemporalExtent):
        self.time = time
    elif isinstance(time, dict):
        self.time = TemporalExtent(
            start_date=time["start_date"],
            end_date=time["end_date"],
            resolution=time.get("resolution", time.get("time_freq")),
            dates=time["dates"],
        )

    self.root_dir = Path(path).absolute()
    self.path = self.root_dir
    if not os.path.exists(self.root_dir):
        os.makedirs(self.root_dir)

__init_subclass__(**kwargs)

Give every backend the facade's ergonomic constructor sugar.

Wraps each concrete backend's __init__ so that — whether reached through the EarthLens facade or by constructing the backend class directly — it also accepts:

• aoi (+ buffer): any shape :func:earthlens.base.spatial.normalize_aoi understands, reduced to lat_lim / lon_lim; a backend that declares its own aoi (WorldPop) keeps it;
• cadence: a clearer alias for temporal_resolution;
• dataset: split out of a single-key variables dict (or passed through to a backend with a native dataset, e.g. S3).

The original __init__ is preserved as the wrapper's __wrapped__, so signature introspection (e.g. EarthLens.options_for) and the facade's kwarg validation still see the backend's real parameters.

Note

No backend currently subclasses another backend. If one ever does, its __init__ must not forward the ergonomic kwargs (aoi / buffer / cadence / dataset) up to super().__init__(): the parent's wrapper would resolve them a second time (e.g. re-running resolve_aoi). Forward only the already-resolved native parameters (lat_lim / lon_lim / temporal_resolution / variables) instead.

Source code in src/earthlens/base/abstractdatasource.py

def __init_subclass__(cls, **kwargs: Any) -> None:
    """Give every backend the facade's ergonomic constructor sugar.

    Wraps each concrete backend's `__init__` so that — whether reached
    through the `EarthLens` facade or by constructing the backend
    class directly — it also accepts:

    * `aoi` (+ `buffer`): any shape :func:`earthlens.base.spatial.normalize_aoi`
      understands, reduced to `lat_lim` / `lon_lim`; a backend that
      declares its own `aoi` (WorldPop) keeps it;
    * `cadence`: a clearer alias for `temporal_resolution`;
    * `dataset`: split out of a single-key `variables` dict (or passed
      through to a backend with a native `dataset`, e.g. S3).

    The original `__init__` is preserved as the wrapper's `__wrapped__`,
    so signature introspection (e.g. `EarthLens.options_for`) and the
    facade's kwarg validation still see the backend's real parameters.

    Note:
        No backend currently subclasses another backend. If one ever
        does, its `__init__` must not forward the ergonomic kwargs
        (`aoi` / `buffer` / `cadence` / `dataset`) up to
        `super().__init__()`: the parent's wrapper would resolve them a
        second time (e.g. re-running `resolve_aoi`). Forward only the
        already-resolved native parameters (`lat_lim` / `lon_lim` /
        `temporal_resolution` / `variables`) instead.
    """
    super().__init_subclass__(**kwargs)
    orig = cls.__dict__.get("__init__")
    if orig is None or getattr(orig, "_ergonomic", False):
        return
    params = inspect.signature(orig).parameters
    native_aoi = "aoi" in params
    native_dataset = "dataset" in params

    @functools.wraps(orig)
    def __init__(
        self, *args, aoi=None, buffer=None, cadence=None, dataset=None, **kw
    ):
        clip_geometry = None
        if cadence is not None:
            kw["temporal_resolution"] = cadence
        if dataset is not None:
            if native_dataset:
                kw["dataset"] = dataset
            elif isinstance(kw.get("variables"), dict):
                raise ValueError(
                    "pass variables= as a list when using dataset=, or omit "
                    "dataset= and key the variables dict yourself"
                )
            else:
                v = kw.get("variables")
                kw["variables"] = {dataset: list(v) if v is not None else []}
        if aoi is not None:
            if native_aoi:
                if buffer is not None:
                    raise ValueError(
                        f"buffer= is not supported by {cls.__name__}, which "
                        "interprets aoi= itself"
                    )
                kw["aoi"] = aoi
            else:
                if kw.get("lat_lim") is not None or kw.get("lon_lim") is not None:
                    raise ValueError(
                        "pass either aoi= or lat_lim=/lon_lim=, not both"
                    )
                from earthlens.base.spatial import resolve_aoi

                kw["lat_lim"], kw["lon_lim"], clip_geometry = resolve_aoi(
                    aoi, buffer=buffer
                )
        elif buffer is not None:
            raise ValueError(
                "buffer= only applies to a point aoi=(lon, lat); pass aoi= too"
            )
        orig(self, *args, **kw)
        if clip_geometry is not None:
            self._attach_clip_geometry(clip_geometry)

    __init__._ergonomic = True
    cls.__init__ = __init__

authenticate()

Eagerly establish the backend's authenticated connection.

The explicit, fail-fast counterpart to the lazy authentication that otherwise happens on the first :meth:download / search: it opens the network client for backends that have one (those mixing in :class:LazyClientMixin — e.g. GEE, ECMWF, STAC) or runs the credential configure() step for backends that hold an auth object (CMEMS, Earthdata, EUMETSAT, …), raising :class:~earthlens.base.AuthenticationError on failure. It is a no-op for credential-free backends (CHIRPS, GDACS, Overture, …), and is idempotent.

Returns:

Type	Description
AbstractDataSource	The backend instance, so callers can chain
AbstractDataSource	EarthLens(...).authenticate().download().

Raises:

Type	Description
AuthenticationError	If the backend cannot authenticate.

Source code in src/earthlens/base/abstractdatasource.py

def authenticate(self) -> AbstractDataSource:
    """Eagerly establish the backend's authenticated connection.

    The explicit, fail-fast counterpart to the lazy authentication
    that otherwise happens on the first :meth:`download` / `search`:
    it opens the network client for backends that have one (those
    mixing in :class:`LazyClientMixin` — e.g. GEE, ECMWF, STAC) or
    runs the credential `configure()` step for backends that hold an
    auth object (CMEMS, Earthdata, EUMETSAT, …), raising
    :class:`~earthlens.base.AuthenticationError` on failure. It is a
    no-op for credential-free backends (CHIRPS, GDACS, Overture, …),
    and is idempotent.

    Returns:
        The backend instance, so callers can chain
        `EarthLens(...).authenticate().download()`.

    Raises:
        AuthenticationError: If the backend cannot authenticate.
    """
    if isinstance(self, LazyClientMixin):
        # Accessing `client` runs the cached `_open_client` (auth).
        _ = self.client
    elif getattr(self, "_auth", None) is not None:
        self._auth.configure()
    return self

download() abstractmethod

Download every requested variable and return the produced artifacts.

The return shape tracks :attr:OUTPUT_KIND: "raster" / "mixed" file-writing backends return the list of written paths (list[Path]); "vector" backends return an in-memory FeatureCollection (radar returns a GeoDataFrame); "tabular" backends return a pandas.DataFrame. Every backend now returns its produced artifacts (the legacy CHIRPS / ECMWF backends return their written list[Path] and also leave the files on disk under self.root_dir).

Partial-failure policy is per backend and currently varies: most multi-item backends are skip-and-continue — a single failed (dataset, variable) / chunk / sensor is logged and the batch proceeds, with a success/failure summary at the end (CHIRPS, CMEMS, FDSN, FIRMS, …) — while single-shot backends propagate the error. NWP exposes this as an explicit errors="warn" | "raise" | "ignore" argument; new backends with a per-item loop should follow that errors= convention so the policy becomes uniformly caller-controllable.

Source code in src/earthlens/base/abstractdatasource.py

@abstractmethod
def download(self):
    """Download every requested variable and return the produced artifacts.

    The return shape tracks :attr:`OUTPUT_KIND`: `"raster"` /
    `"mixed"` file-writing backends return the list of written
    paths (`list[Path]`); `"vector"` backends return an in-memory
    `FeatureCollection` (radar returns a `GeoDataFrame`);
    `"tabular"` backends return a `pandas.DataFrame`. Every backend
    now returns its produced artifacts (the legacy CHIRPS / ECMWF
    backends return their written `list[Path]` and also leave the
    files on disk under `self.root_dir`).

    Partial-failure policy is per backend and currently varies:
    most multi-item backends are **skip-and-continue** — a single
    failed `(dataset, variable)` / chunk / sensor is logged and the
    batch proceeds, with a success/failure summary at the end
    (CHIRPS, CMEMS, FDSN, FIRMS, …) — while single-shot backends
    propagate the error. NWP exposes this as an explicit
    `errors="warn" | "raise" | "ignore"` argument; new backends
    with a per-item loop should follow that `errors=` convention so
    the policy becomes uniformly caller-controllable.
    """
    # loop over dates if the downloaded rasters/netcdf are for a specific date out of the required
    # list of dates
    pass

earthlens.base.AbstractCatalog

Bases: BaseModel

Abstract base class for per-data-source variable catalogs.

Subclasses load a backend-specific catalog (a YAML file, an in-code dict, or a remote query) in :meth:get_catalog and expose individual entries via :meth:get_variable. The :func:model_post_init hook eagerly populates :attr:catalog after pydantic validation runs, so subclasses can treat the catalog as a dict thereafter without writing their own __init__.

Subclasses pass through pydantic's normal BaseModel.__init__ — declare any backend-specific construction parameters as pydantic fields rather than __init__ arguments. Override :meth:get_catalog (and optionally :meth:get_variable); the base implementations raise :class:NotImplementedError to flag a missing override at first use rather than silently returning an empty mapping.

Attributes:

Name	Type	Description
catalog	dict[str, Any]	The full catalog mapping returned by :meth:get_catalog. Populated post-init; defaults to an empty dict so the field is always present. Type and shape are backend-specific (a concrete subclass typically stores typed value objects, e.g. dict[str, Variable] for the ECMWF backend).

Source code in src/earthlens/base/abstractdatasource.py

class AbstractCatalog(BaseModel):
    """Abstract base class for per-data-source variable catalogs.

    Subclasses load a backend-specific catalog (a YAML file, an
    in-code dict, or a remote query) in :meth:`get_catalog` and
    expose individual entries via :meth:`get_variable`. The
    :func:`model_post_init` hook eagerly populates :attr:`catalog`
    after pydantic validation runs, so subclasses can treat the
    catalog as a dict thereafter without writing their own
    `__init__`.

    Subclasses pass through pydantic's normal `BaseModel.__init__`
    — declare any backend-specific construction parameters as
    pydantic fields rather than `__init__` arguments. Override
    :meth:`get_catalog` (and optionally :meth:`get_variable`); the
    base implementations raise :class:`NotImplementedError` to flag
    a missing override at first use rather than silently returning
    an empty mapping.

    Attributes:
        catalog: The full catalog mapping returned by
            :meth:`get_catalog`. Populated post-init; defaults to an
            empty dict so the field is always present. Type and
            shape are backend-specific (a concrete subclass typically
            stores typed value objects, e.g. `dict[str, Variable]`
            for the ECMWF backend).
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    #: Short label used by :meth:`get_dataset`'s did-you-mean error
    #: message — concrete subclasses override (e.g. `"GEE catalog"`,
    #: `"CDS catalog"`, `"CHC catalog"`) so the user sees which
    #: catalog they failed against.
    _catalog_kind: str = "catalog"

    #: Plural noun for the catalog entries, used in :meth:`get_dataset`'s
    #: did-you-mean message (`"Known {noun}: [...]"`). Defaults to
    #: `"datasets"`; subclasses whose entries are not "datasets" override
    #: it (e.g. `"parameters"` for the openaq / usgs_water catalogs).
    _entry_noun: str = "datasets"

    catalog: dict[str, Any] = Field(default_factory=dict)
    available_datasets: list[str] = Field(default_factory=list)
    datasets: dict[str, Any] = Field(default_factory=dict)
    providers: dict[str, Any] = Field(default_factory=dict)

    def model_post_init(self, __context: Any) -> None:
        """Populate :attr:`catalog` after pydantic validation runs.

        Pydantic calls this hook automatically; subclasses that need
        their own post-init wiring should override it and call
        `super().model_post_init(__context)` first to keep the
        catalog-loading behaviour.
        """
        self.catalog = self.get_catalog()

    def get_catalog(self) -> Any:
        """Read the catalog of the datasource from disk or retrieve it from server.

        Abstract; concrete subclasses must override and return their
        backend-specific catalog object (e.g. a pydantic `Catalog`
        instance, a `dict`, or whatever shape the backend uses).

        Raises:
            NotImplementedError: Always, until overridden by a subclass.
        """
        raise NotImplementedError

    def get_variable(self, dataset_key: str, variable_name: str) -> Any:
        """Return one leaf (variable / band / asset) of a dataset.

        Shared two-argument contract for the two-level catalogs: a leaf
        is addressed by its `(dataset_key, variable_name)` pair, because
        the same leaf code can appear under more than one dataset (e.g.
        `"2m-temperature"` lives under several CDS datasets). Concrete
        overrides return their typed leaf row and raise `ValueError`
        (with a did-you-mean hint) on an unknown key:

        * chc / ecmwf / cmems — return a `Variable`.
        * gee — return a `Band` (also exposed as `get_band`).
        * firms — return a `SensorColumn` (also exposed as `get_column`).
        * tropycal — return a `TrackField` (also exposed as `get_field`).

        Single-level catalogs (where one row *is* the leaf — fdsn, gdacs,
        radar, openaq, overture, usgs_water) do not implement this; their
        rows are addressed directly with :meth:`get_dataset` / `[key]`.

        Note:
            This supersedes the former single-argument
            `get_variable(var_name)`, which returned `self.catalog.get(var_name)`.
            External callers/subclassers that relied on the one-argument
            form must pass the parent `dataset_key` as well.

        Args:
            dataset_key: The parent dataset / collection key.
            variable_name: The leaf code within that dataset.

        Returns:
            The backend-specific leaf row.

        Raises:
            NotImplementedError: If the backend has no per-dataset leaf
                level.
        """
        raise NotImplementedError(
            f"{type(self).__name__} has no per-dataset variable level; "
            "address its rows with get_dataset() / [key]."
        )

    # -- shared dict-like surface over `datasets` (M1 from catalog-cross-backend-comparison)

    def get_dataset(self, name: str) -> Any:
        """Return the dataset record for `name`, with a did-you-mean hint on miss.

        Backend-generic: looks up `name` in :attr:`datasets` and raises
        `ValueError` (not `KeyError`) with the closest known name when
        absent. Concrete subclasses can override to narrow the return
        type or customise the error message.

        Args:
            name: Catalog key (e.g. CDS dataset short name, EE asset id,
                CHC dataset key).

        Returns:
            The matching dataset record (type depends on the subclass).

        Raises:
            ValueError: If `name` is not a key of :attr:`datasets`.
        """
        try:
            return self.datasets[name]
        except KeyError:
            close = difflib.get_close_matches(name, self.datasets, n=1)
            hint = f" Did you mean {close[0]!r}?" if close else ""
            raise ValueError(
                f"{name!r} is not in the {self._catalog_kind}. "
                f"Known {self._entry_noun}: {sorted(self.datasets)}.{hint}"
            ) from None

    def __getitem__(self, name: str) -> Any:
        """`cat[name]` — dict-style lookup; raises `KeyError` on miss."""
        try:
            return self.get_dataset(name)
        except ValueError as exc:
            raise KeyError(name) from exc

    def __contains__(self, name: object) -> bool:
        """`name in cat` — True when `name` is a curated dataset."""
        return name in self.datasets

    def __iter__(self):
        """Iterate over the curated dataset keys."""
        return iter(self.datasets)

    def __len__(self) -> int:
        """Number of curated datasets in the catalog."""
        return len(self.datasets)

    def __repr__(self) -> str:
        """Compact developer repr — counts, not contents."""
        return (
            f"{type(self).__name__}(datasets={len(self.datasets)}, "
            f"available_datasets={len(self.available_datasets)})"
        )

    def get_provider(self, slug: str) -> Any:
        """Return the provider record for `slug` (with a did-you-mean hint on miss).

        The value type depends on the backend's :attr:`providers` field:
        most backends store an :class:`earthlens.base.Provider`, but some
        mirror a domain-specific record (earthdata mirrors its
        `EarthdataDAAC` from `daacs`, stac its `Endpoint` from
        `endpoints`).

        Args:
            slug: A registered provider slug (e.g. `"nasa-lp-daac"`,
                `"ucsb-chc"`, `"copernicus"`).

        Returns:
            The matching provider record (a `Provider`, or the backend's
            domain-specific provider model).

        Raises:
            ValueError: If `slug` is not a registered provider.
        """
        try:
            return self.providers[slug]
        except KeyError:
            close = difflib.get_close_matches(slug, self.providers, n=1)
            hint = f" Did you mean {close[0]!r}?" if close else ""
            raise ValueError(
                f"{slug!r} is not a registered provider. "
                f"Known providers: {sorted(self.providers)}.{hint}"
            ) from None

    def resolve(self, key: str, *args: Any, **kwargs: Any) -> Any:
        """Map a user-facing key to the concrete thing a request needs.

        Shared convention for every backend that implements a resolve
        step: take a *logical* catalog key (a friendly name, collection
        key, or model key) and return the backend-specific value the
        download path consumes. The return type and any extra
        positional / keyword arguments are backend-specific by
        necessity — the catalogs resolve to different things — so this
        base method only fixes the *verb*, not the signature. The
        concrete overrides:

        * `nwp.resolve(model_key)` / `usgs_water.resolve(code_or_name)`
          — return a model key / 5-digit parameter code (`str`).
        * `stac.resolve(endpoint, collection_key)` — return the upstream
          collection id for that endpoint (`str`).
        * `openeo.resolve(key)` / `sentinel_hub.resolve(key)` — return a
          normalised request object (a `ResolvedGraph` / `ResolvedRequest`)
          covering both plain collections and recipes.
        * `earthdata.resolve(key, daac=None)` /
          `eumetsat.resolve(key, group=None)` — return the dataset row,
          with an optional second argument to disambiguate a key shared
          across DAACs / mission groups.

        Backends without a resolve step address their catalog directly
        through :meth:`get_dataset` / `__getitem__`.

        Args:
            key: The logical catalog key to resolve.
            *args: Backend-specific positional arguments (e.g. the STAC
                endpoint).
            **kwargs: Backend-specific keyword arguments (e.g.
                `daac=` / `group=`).

        Returns:
            The backend-specific resolved value (see the override list).

        Raises:
            NotImplementedError: If the backend has no resolve step.
        """
        raise NotImplementedError(
            f"{type(self).__name__} has no resolve() step; address its "
            "catalog with get_dataset() / [key] instead."
        )

    def __str__(self) -> str:
        """Pretty-print the curated `datasets` map as YAML.

        `None`-valued fields are omitted so the output stays readable;
        the ordering of keys follows insertion. Concrete subclasses
        whose dataset values aren't pydantic `BaseModel`s (rare) must
        override.
        """
        import yaml

        body = {}
        for key, dataset in self.datasets.items():
            if isinstance(dataset, BaseModel):
                body[key] = dataset.model_dump(exclude_none=True)
            else:
                body[key] = dataset
        return yaml.safe_dump(
            body, default_flow_style=False, sort_keys=False, allow_unicode=True
        )

__contains__(name)

name in cat — True when name is a curated dataset.

Source code in src/earthlens/base/abstractdatasource.py

def __contains__(self, name: object) -> bool:
    """`name in cat` — True when `name` is a curated dataset."""
    return name in self.datasets

__getitem__(name)

cat[name] — dict-style lookup; raises KeyError on miss.

Source code in src/earthlens/base/abstractdatasource.py

def __getitem__(self, name: str) -> Any:
    """`cat[name]` — dict-style lookup; raises `KeyError` on miss."""
    try:
        return self.get_dataset(name)
    except ValueError as exc:
        raise KeyError(name) from exc

__iter__()

Iterate over the curated dataset keys.

Source code in src/earthlens/base/abstractdatasource.py

def __iter__(self):
    """Iterate over the curated dataset keys."""
    return iter(self.datasets)

__len__()

Number of curated datasets in the catalog.

Source code in src/earthlens/base/abstractdatasource.py

def __len__(self) -> int:
    """Number of curated datasets in the catalog."""
    return len(self.datasets)

__repr__()

Compact developer repr — counts, not contents.

Source code in src/earthlens/base/abstractdatasource.py

def __repr__(self) -> str:
    """Compact developer repr — counts, not contents."""
    return (
        f"{type(self).__name__}(datasets={len(self.datasets)}, "
        f"available_datasets={len(self.available_datasets)})"
    )

__str__()

Pretty-print the curated datasets map as YAML.

None-valued fields are omitted so the output stays readable; the ordering of keys follows insertion. Concrete subclasses whose dataset values aren't pydantic BaseModels (rare) must override.

Source code in src/earthlens/base/abstractdatasource.py

def __str__(self) -> str:
    """Pretty-print the curated `datasets` map as YAML.

    `None`-valued fields are omitted so the output stays readable;
    the ordering of keys follows insertion. Concrete subclasses
    whose dataset values aren't pydantic `BaseModel`s (rare) must
    override.
    """
    import yaml

    body = {}
    for key, dataset in self.datasets.items():
        if isinstance(dataset, BaseModel):
            body[key] = dataset.model_dump(exclude_none=True)
        else:
            body[key] = dataset
    return yaml.safe_dump(
        body, default_flow_style=False, sort_keys=False, allow_unicode=True
    )

get_catalog()

Read the catalog of the datasource from disk or retrieve it from server.

Abstract; concrete subclasses must override and return their backend-specific catalog object (e.g. a pydantic Catalog instance, a dict, or whatever shape the backend uses).

Raises:

Type	Description
NotImplementedError	Always, until overridden by a subclass.

Source code in src/earthlens/base/abstractdatasource.py

def get_catalog(self) -> Any:
    """Read the catalog of the datasource from disk or retrieve it from server.

    Abstract; concrete subclasses must override and return their
    backend-specific catalog object (e.g. a pydantic `Catalog`
    instance, a `dict`, or whatever shape the backend uses).

    Raises:
        NotImplementedError: Always, until overridden by a subclass.
    """
    raise NotImplementedError

get_dataset(name)

Return the dataset record for name, with a did-you-mean hint on miss.

Backend-generic: looks up name in :attr:datasets and raises ValueError (not KeyError) with the closest known name when absent. Concrete subclasses can override to narrow the return type or customise the error message.

Parameters:

Name	Type	Description	Default
name	str	Catalog key (e.g. CDS dataset short name, EE asset id, CHC dataset key).	required

Returns:

Type	Description
Any	The matching dataset record (type depends on the subclass).

Raises:

Type	Description
ValueError	If name is not a key of :attr:datasets.

Source code in src/earthlens/base/abstractdatasource.py

def get_dataset(self, name: str) -> Any:
    """Return the dataset record for `name`, with a did-you-mean hint on miss.

    Backend-generic: looks up `name` in :attr:`datasets` and raises
    `ValueError` (not `KeyError`) with the closest known name when
    absent. Concrete subclasses can override to narrow the return
    type or customise the error message.

    Args:
        name: Catalog key (e.g. CDS dataset short name, EE asset id,
            CHC dataset key).

    Returns:
        The matching dataset record (type depends on the subclass).

    Raises:
        ValueError: If `name` is not a key of :attr:`datasets`.
    """
    try:
        return self.datasets[name]
    except KeyError:
        close = difflib.get_close_matches(name, self.datasets, n=1)
        hint = f" Did you mean {close[0]!r}?" if close else ""
        raise ValueError(
            f"{name!r} is not in the {self._catalog_kind}. "
            f"Known {self._entry_noun}: {sorted(self.datasets)}.{hint}"
        ) from None

get_provider(slug)

Return the provider record for slug (with a did-you-mean hint on miss).

The value type depends on the backend's :attr:providers field: most backends store an :class:earthlens.base.Provider, but some mirror a domain-specific record (earthdata mirrors its EarthdataDAAC from daacs, stac its Endpoint from endpoints).

Parameters:

Name	Type	Description	Default
slug	str	A registered provider slug (e.g. "nasa-lp-daac", "ucsb-chc", "copernicus").	required

Returns:

Type	Description
Any	The matching provider record (a Provider, or the backend's
Any	domain-specific provider model).

Raises:

Type	Description
ValueError	If slug is not a registered provider.

Source code in src/earthlens/base/abstractdatasource.py

def get_provider(self, slug: str) -> Any:
    """Return the provider record for `slug` (with a did-you-mean hint on miss).

    The value type depends on the backend's :attr:`providers` field:
    most backends store an :class:`earthlens.base.Provider`, but some
    mirror a domain-specific record (earthdata mirrors its
    `EarthdataDAAC` from `daacs`, stac its `Endpoint` from
    `endpoints`).

    Args:
        slug: A registered provider slug (e.g. `"nasa-lp-daac"`,
            `"ucsb-chc"`, `"copernicus"`).

    Returns:
        The matching provider record (a `Provider`, or the backend's
        domain-specific provider model).

    Raises:
        ValueError: If `slug` is not a registered provider.
    """
    try:
        return self.providers[slug]
    except KeyError:
        close = difflib.get_close_matches(slug, self.providers, n=1)
        hint = f" Did you mean {close[0]!r}?" if close else ""
        raise ValueError(
            f"{slug!r} is not a registered provider. "
            f"Known providers: {sorted(self.providers)}.{hint}"
        ) from None

get_variable(dataset_key, variable_name)

Return one leaf (variable / band / asset) of a dataset.

Shared two-argument contract for the two-level catalogs: a leaf is addressed by its (dataset_key, variable_name) pair, because the same leaf code can appear under more than one dataset (e.g. "2m-temperature" lives under several CDS datasets). Concrete overrides return their typed leaf row and raise ValueError (with a did-you-mean hint) on an unknown key:

• chc / ecmwf / cmems — return a Variable.
• gee — return a Band (also exposed as get_band).
• firms — return a SensorColumn (also exposed as get_column).
• tropycal — return a TrackField (also exposed as get_field).

Single-level catalogs (where one row is the leaf — fdsn, gdacs, radar, openaq, overture, usgs_water) do not implement this; their rows are addressed directly with :meth:get_dataset / [key].

Note

This supersedes the former single-argument get_variable(var_name), which returned self.catalog.get(var_name). External callers/subclassers that relied on the one-argument form must pass the parent dataset_key as well.

Parameters:

Name	Type	Description	Default
dataset_key	str	The parent dataset / collection key.	required
variable_name	str	The leaf code within that dataset.	required

Returns:

Type	Description
Any	The backend-specific leaf row.

Raises:

Type	Description
NotImplementedError	If the backend has no per-dataset leaf level.

Source code in src/earthlens/base/abstractdatasource.py

def get_variable(self, dataset_key: str, variable_name: str) -> Any:
    """Return one leaf (variable / band / asset) of a dataset.

    Shared two-argument contract for the two-level catalogs: a leaf
    is addressed by its `(dataset_key, variable_name)` pair, because
    the same leaf code can appear under more than one dataset (e.g.
    `"2m-temperature"` lives under several CDS datasets). Concrete
    overrides return their typed leaf row and raise `ValueError`
    (with a did-you-mean hint) on an unknown key:

    * chc / ecmwf / cmems — return a `Variable`.
    * gee — return a `Band` (also exposed as `get_band`).
    * firms — return a `SensorColumn` (also exposed as `get_column`).
    * tropycal — return a `TrackField` (also exposed as `get_field`).

    Single-level catalogs (where one row *is* the leaf — fdsn, gdacs,
    radar, openaq, overture, usgs_water) do not implement this; their
    rows are addressed directly with :meth:`get_dataset` / `[key]`.

    Note:
        This supersedes the former single-argument
        `get_variable(var_name)`, which returned `self.catalog.get(var_name)`.
        External callers/subclassers that relied on the one-argument
        form must pass the parent `dataset_key` as well.

    Args:
        dataset_key: The parent dataset / collection key.
        variable_name: The leaf code within that dataset.

    Returns:
        The backend-specific leaf row.

    Raises:
        NotImplementedError: If the backend has no per-dataset leaf
            level.
    """
    raise NotImplementedError(
        f"{type(self).__name__} has no per-dataset variable level; "
        "address its rows with get_dataset() / [key]."
    )

model_post_init(__context)

Populate :attr:catalog after pydantic validation runs.

Pydantic calls this hook automatically; subclasses that need their own post-init wiring should override it and call super().model_post_init(__context) first to keep the catalog-loading behaviour.

Source code in src/earthlens/base/abstractdatasource.py

def model_post_init(self, __context: Any) -> None:
    """Populate :attr:`catalog` after pydantic validation runs.

    Pydantic calls this hook automatically; subclasses that need
    their own post-init wiring should override it and call
    `super().model_post_init(__context)` first to keep the
    catalog-loading behaviour.
    """
    self.catalog = self.get_catalog()

resolve(key, *args, **kwargs)

Map a user-facing key to the concrete thing a request needs.

Shared convention for every backend that implements a resolve step: take a logical catalog key (a friendly name, collection key, or model key) and return the backend-specific value the download path consumes. The return type and any extra positional / keyword arguments are backend-specific by necessity — the catalogs resolve to different things — so this base method only fixes the verb, not the signature. The concrete overrides:

• nwp.resolve(model_key) / usgs_water.resolve(code_or_name) — return a model key / 5-digit parameter code (str).
• stac.resolve(endpoint, collection_key) — return the upstream collection id for that endpoint (str).
• openeo.resolve(key) / sentinel_hub.resolve(key) — return a normalised request object (a ResolvedGraph / ResolvedRequest) covering both plain collections and recipes.
• earthdata.resolve(key, daac=None) / eumetsat.resolve(key, group=None) — return the dataset row, with an optional second argument to disambiguate a key shared across DAACs / mission groups.

Backends without a resolve step address their catalog directly through :meth:get_dataset / __getitem__.

Parameters:

Name	Type	Description	Default
key	str	The logical catalog key to resolve.	required
*args	Any	Backend-specific positional arguments (e.g. the STAC endpoint).	()
**kwargs	Any	Backend-specific keyword arguments (e.g. daac= / group=).	{}

Returns:

Type	Description
Any	The backend-specific resolved value (see the override list).

Raises:

Type	Description
NotImplementedError	If the backend has no resolve step.

Source code in src/earthlens/base/abstractdatasource.py

def resolve(self, key: str, *args: Any, **kwargs: Any) -> Any:
    """Map a user-facing key to the concrete thing a request needs.

    Shared convention for every backend that implements a resolve
    step: take a *logical* catalog key (a friendly name, collection
    key, or model key) and return the backend-specific value the
    download path consumes. The return type and any extra
    positional / keyword arguments are backend-specific by
    necessity — the catalogs resolve to different things — so this
    base method only fixes the *verb*, not the signature. The
    concrete overrides:

    * `nwp.resolve(model_key)` / `usgs_water.resolve(code_or_name)`
      — return a model key / 5-digit parameter code (`str`).
    * `stac.resolve(endpoint, collection_key)` — return the upstream
      collection id for that endpoint (`str`).
    * `openeo.resolve(key)` / `sentinel_hub.resolve(key)` — return a
      normalised request object (a `ResolvedGraph` / `ResolvedRequest`)
      covering both plain collections and recipes.
    * `earthdata.resolve(key, daac=None)` /
      `eumetsat.resolve(key, group=None)` — return the dataset row,
      with an optional second argument to disambiguate a key shared
      across DAACs / mission groups.

    Backends without a resolve step address their catalog directly
    through :meth:`get_dataset` / `__getitem__`.

    Args:
        key: The logical catalog key to resolve.
        *args: Backend-specific positional arguments (e.g. the STAC
            endpoint).
        **kwargs: Backend-specific keyword arguments (e.g.
            `daac=` / `group=`).

    Returns:
        The backend-specific resolved value (see the override list).

    Raises:
        NotImplementedError: If the backend has no resolve step.
    """
    raise NotImplementedError(
        f"{type(self).__name__} has no resolve() step; address its "
        "catalog with get_dataset() / [key] instead."
    )