EUMETSAT Data Store — API reference

The EUMETSAT Data Store data source subpackage — earthlens.eumetsat. Background, usage, authentication, the catalog, and the Data Tailor path are covered under the other pages in this section; this page is the rendered API.

earthlens.eumetsat

EUMETSAT Data Store backend (181 collections via eumdac).

One unified backend over the EUMETSAT Data Store: a single OAuth2 consumer key / secret mints a bearer token that reaches every collection — MTG-I1 FCI, MSG SEVIRI, Metop (ASCAT / IASI), Metop-SG, the Sentinel-3 / -5P / -6 mirrors, and the OSI SAF / CDR / FDR families. The backend fetches whole native products to disk, and supports server-side subset / reproject / reformat via the tailor=TailorConfig(...) Data Tailor path; native SEVIRI / FCI client-side reading (the satpy bridge) is a deferred follow-on.

Like the NASA Earthdata backend, the output shape is per-collection, not fixed — EUMETSAT sets OUTPUT_KIND from the resolved catalog row (raster / vector / tabular).

Public surface (re-exported from this package):

• EUMETSAT — the backend itself; instantiate with a date range, a bbox, and a {collection_key: [selector, ...]} mapping, then call EUMETSAT.download.
• Catalog — pydantic-backed loader for the bundled per-group catalog/ directory.
• EumetsatDataset — one curated dataset (collection) row (collection_id, group, output_kind, format, selectors, tailor_product_type, extent).
• DataStoreGroup — the Data Store group (mission family) enum.
• Extent / TemporalCoverage — the spatial / temporal coverage rows.
• EumetsatAuth — AbstractAuth wrapper over eumdac.AccessToken. Idempotent; safe to call repeatedly.
• EumetsatCredentials — frozen value object the auth class binds to.
• TailorConfig — frozen request shape for the Data Tailor server-side subset / reproject / reformat path (download(tailor=...)).
• AuthenticationError — raised when token minting fails; subclass of earthlens.base.AuthenticationError.
• CATALOG_PATH — absolute path to the bundled catalog/ directory.

The [eumetsat] extra pulls eumdac. The eumdac import is lazy, so this package imports without the extra installed.

AuthenticationError

Bases: AuthenticationError

Raised when eumdac cannot mint an OAuth2 token.

Wraps the underlying eumdac / HTTP failure with a message that names a fix: register a consumer key / secret at the EUMETSAT API-key page, set the EUMETSAT_CONSUMER_KEY / EUMETSAT_CONSUMER_SECRET environment variables, or write a ~/.eumdac/credentials file.

A subclass of the cross-backend earthlens.base.AuthenticationError so callers can catch every backend's auth failure with one except clause.

Source code in libs/providers/imagery/src/earthlens/eumetsat/auth.py

class AuthenticationError(_BaseAuthenticationError):
    """Raised when `eumdac` cannot mint an OAuth2 token.

    Wraps the underlying `eumdac` / HTTP failure with a message that
    names a fix: register a consumer key / secret at the EUMETSAT API-key
    page, set the `EUMETSAT_CONSUMER_KEY` / `EUMETSAT_CONSUMER_SECRET`
    environment variables, or write a `~/.eumdac/credentials` file.

    A subclass of the cross-backend `earthlens.base.AuthenticationError`
    so callers can catch every backend's auth failure with one `except`
    clause.
    """

Catalog

Bases: AbstractCatalog

Dataset catalog for the EUMETSAT Data Store backend.

Reads the bundled catalog/ directory (shipped as package data) and exposes its consumed top-level sections as typed pydantic fields. Instantiate with no arguments (Catalog()) — model_post_init parses the YAML and populates every field in one pass. Mirrors the earthlens.earthdata / earthlens.gee / earthlens.cmems catalogs: datasets (the curated map) and available_datasets (the informational index).

Attributes:

Name	Type	Description
available_datasets	list[str]	Informational list of every Data Store collection id the browse walk found. Runtime code does not consume it.
datasets	dict[str, EumetsatDataset]	Structural map keyed by the curated dataset key. Each value is an EumetsatDataset.

Examples:

• Resolve a curated dataset:

  >>> from earthlens.eumetsat import Catalog
  >>> "msg-hrseviri" in Catalog()
  True
  

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

class Catalog(AbstractCatalog):
    """Dataset catalog for the EUMETSAT Data Store backend.

    Reads the bundled `catalog/` directory (shipped as package data) and
    exposes its consumed top-level sections as typed pydantic fields.
    Instantiate with no arguments (`Catalog()`) — `model_post_init`
    parses the YAML and populates every field in one pass. Mirrors the
    `earthlens.earthdata` / `earthlens.gee` / `earthlens.cmems` catalogs:
    `datasets` (the curated map) and `available_datasets` (the
    informational index).

    Attributes:
        available_datasets: Informational list of every Data Store
            collection id the browse walk found. Runtime code does not
            consume it.
        datasets: Structural map keyed by the curated dataset key. Each
            value is an `EumetsatDataset`.

    Examples:
        - Resolve a curated dataset:
            ```python
            >>> from earthlens.eumetsat import Catalog
            >>> "msg-hrseviri" in Catalog()
            True

            ```
    """

    _catalog_kind: str = "EUMETSAT catalog"

    available_datasets: list[str] = Field(default_factory=list)
    datasets: dict[str, EumetsatDataset] = Field(default_factory=dict)

    @classmethod
    def _autoload(cls) -> dict[str, Any]:
        """Read the bundled catalog from disk.

        Returns:
            dict[str, Any]: The `available_datasets`, `datasets` read from
                the bundled catalog.
        """
        loaded = Catalog.load()
        return {
            "available_datasets": loaded.available_datasets,
            "datasets": loaded.datasets,
        }

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

        Args:
            catalog_path: Path to the `catalog/` directory or a single
                `*.yaml` file. Defaults to module-level `CATALOG_PATH`.

        Returns:
            A fully-populated `Catalog`.

        Raises:
            ValueError: Propagated from `_load_catalog_data`.
        """
        catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
        available, datasets = _load_catalog_data(catalog_path)
        return cls(
            available_datasets=list(available),
            datasets=dict(datasets),
        )

    def get_catalog(self) -> dict[str, EumetsatDataset]:
        """Return the structural per-dataset map.

        Satisfies the abstract base's contract; the actual parsing is
        done in `model_post_init`.

        Returns:
            dict[str, EumetsatDataset]: One entry per curated dataset.
                Same object as `datasets`.
        """
        return self.datasets

    def resolve(
        self, key: str, group: DataStoreGroup | str | None = None
    ) -> EumetsatDataset:
        """Resolve a dataset key, optionally disambiguated by group.

        Most keys map one-to-one to a curated row; the inherited
        :meth:`get_dataset` (with its did-you-mean) handles those. The
        `group=` filter additionally asserts which Data Store group the
        resolved dataset belongs to (`G2`) — the EUMETSAT analog of the
        Earthdata backend's `daac=` filter.

        Args:
            key: Curated dataset key (a member of `datasets`).
            group: Optional `DataStoreGroup` (or its string value) the
                resolved row's `group` must match.

        Returns:
            EumetsatDataset: The resolved row.

        Raises:
            ValueError: When `key` is unknown (with a did-you-mean hint),
                or `group=` is given but does not match the row's group.

        Examples:
            - Resolve a key and read its group:
                ```python
                >>> from earthlens.eumetsat import Catalog
                >>> Catalog().resolve("msg-hrseviri").group.value
                'MSG'

                ```
        """
        dataset = cast("EumetsatDataset", self.get_dataset(key))
        if group is not None:
            wanted = group.value if isinstance(group, DataStoreGroup) else str(group)
            if dataset.group.value != wanted:
                raise ValueError(
                    f"dataset {key!r} is in group "
                    f"{dataset.group.value!r}, not the requested "
                    f"group={wanted!r}."
                )
        return dataset

get_catalog()

Return the structural per-dataset map.

Satisfies the abstract base's contract; the actual parsing is done in model_post_init.

Returns:

Type	Description
dict[str, EumetsatDataset]	dict[str, EumetsatDataset]: One entry per curated dataset. Same object as datasets.

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

def get_catalog(self) -> dict[str, EumetsatDataset]:
    """Return the structural per-dataset map.

    Satisfies the abstract base's contract; the actual parsing is
    done in `model_post_init`.

    Returns:
        dict[str, EumetsatDataset]: One entry per curated dataset.
            Same object as `datasets`.
    """
    return self.datasets

load(catalog_path=None) classmethod

Read the EUMETSAT catalog from disk (cached).

Parameters:

Name	Type	Description	Default
catalog_path	Path | None	Path to the catalog/ directory or a single *.yaml file. Defaults to module-level CATALOG_PATH.	None

Returns:

Type	Description
Catalog	A fully-populated Catalog.

Raises:

Type	Description
ValueError	Propagated from _load_catalog_data.

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

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

    Args:
        catalog_path: Path to the `catalog/` directory or a single
            `*.yaml` file. Defaults to module-level `CATALOG_PATH`.

    Returns:
        A fully-populated `Catalog`.

    Raises:
        ValueError: Propagated from `_load_catalog_data`.
    """
    catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
    available, datasets = _load_catalog_data(catalog_path)
    return cls(
        available_datasets=list(available),
        datasets=dict(datasets),
    )

resolve(key, group=None)

Resolve a dataset key, optionally disambiguated by group.

Most keys map one-to-one to a curated row; the inherited :meth:get_dataset (with its did-you-mean) handles those. The group= filter additionally asserts which Data Store group the resolved dataset belongs to (G2) — the EUMETSAT analog of the Earthdata backend's daac= filter.

Parameters:

Name	Type	Description	Default
key	str	Curated dataset key (a member of datasets).	required
group	DataStoreGroup | str | None	Optional DataStoreGroup (or its string value) the resolved row's group must match.	None

Returns:

Name	Type	Description
EumetsatDataset	EumetsatDataset	The resolved row.

Raises:

Type	Description
ValueError	When key is unknown (with a did-you-mean hint), or group= is given but does not match the row's group.

Examples:

• Resolve a key and read its group:

  >>> from earthlens.eumetsat import Catalog
  >>> Catalog().resolve("msg-hrseviri").group.value
  'MSG'
  

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

def resolve(
    self, key: str, group: DataStoreGroup | str | None = None
) -> EumetsatDataset:
    """Resolve a dataset key, optionally disambiguated by group.

    Most keys map one-to-one to a curated row; the inherited
    :meth:`get_dataset` (with its did-you-mean) handles those. The
    `group=` filter additionally asserts which Data Store group the
    resolved dataset belongs to (`G2`) — the EUMETSAT analog of the
    Earthdata backend's `daac=` filter.

    Args:
        key: Curated dataset key (a member of `datasets`).
        group: Optional `DataStoreGroup` (or its string value) the
            resolved row's `group` must match.

    Returns:
        EumetsatDataset: The resolved row.

    Raises:
        ValueError: When `key` is unknown (with a did-you-mean hint),
            or `group=` is given but does not match the row's group.

    Examples:
        - Resolve a key and read its group:
            ```python
            >>> from earthlens.eumetsat import Catalog
            >>> Catalog().resolve("msg-hrseviri").group.value
            'MSG'

            ```
    """
    dataset = cast("EumetsatDataset", self.get_dataset(key))
    if group is not None:
        wanted = group.value if isinstance(group, DataStoreGroup) else str(group)
        if dataset.group.value != wanted:
            raise ValueError(
                f"dataset {key!r} is in group "
                f"{dataset.group.value!r}, not the requested "
                f"group={wanted!r}."
            )
    return dataset

DataStoreGroup

Bases: StrEnum

The EUMETSAT Data Store collection groups (mission families).

Each value is the human-readable group label carried on a catalog row's group: field and accepted by the backend's group= kwarg to disambiguate a collection key shared across groups (G2).

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

class DataStoreGroup(StrEnum):
    """The EUMETSAT Data Store collection groups (mission families).

    Each value is the human-readable group label carried on a catalog
    row's `group:` field and accepted by the backend's `group=` kwarg to
    disambiguate a collection key shared across groups (`G2`).
    """

    MTG = "MTG"
    MSG = "MSG"
    MFG = "MFG"
    METOP = "Metop"
    METOP_SG = "Metop-SG"
    SENTINEL_3 = "Sentinel-3"
    SENTINEL_5P = "Sentinel-5P"
    SENTINEL_6 = "Sentinel-6"
    OSI_SAF = "OSI-SAF"
    OTHER = "Other"

EUMETSAT

Bases: AbstractDataSource

EUMETSAT Data Store backend (per-collection output kind).

Wraps eumdac so a user can search a curated EUMETSAT collection by bbox + window and fetch its native products through the same download() shape every other earthlens backend uses. One OAuth2 consumer key / secret authenticates across every collection.

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	Class default "raster", overridden per instance in __init__ from the resolved collection row's output_kind (G1). The facade reads this instance value to gate aggregate=.

Source code in libs/providers/imagery/src/earthlens/eumetsat/backend.py

class EUMETSAT(AbstractDataSource):
    """EUMETSAT Data Store backend (per-collection output kind).

    Wraps `eumdac` so a user can search a curated EUMETSAT collection by
    bbox + window and fetch its native products through the same
    `download()` shape every other earthlens backend uses. One OAuth2
    consumer key / secret authenticates across every collection.

    Attributes:
        OUTPUT_KIND: Class default `"raster"`, **overridden per instance**
            in `__init__` from the resolved collection row's
            `output_kind` (`G1`). The facade reads this instance value to
            gate `aggregate=`.
    """

    OUTPUT_KIND: OutputKind = "raster"

    AGGREGATE_REFUSAL_REASON = (
        "the temporal reducer is not wired for this backend. Download the "
        "products (optionally with tailor= for a server-side subset / "
        "reproject) and reduce the NetCDF ones client-side with pyramids"
    )

    def __init__(
        self,
        start: str,
        end: str,
        variables: dict[str, list[str]],
        lat_lim: list[float],
        lon_lim: list[float],
        temporal_resolution: str = "daily",
        path: Path | str | None = None,
        fmt: str = "%Y-%m-%d",
        group: DataStoreGroup | str | None = None,
        consumer_key: str | None = None,
        consumer_secret: str | None = None,
        credentials_file: Path | str | None = None,
    ):
        """Initialise an EUMETSAT backend instance.

        Resolves every requested dataset key against the catalog
        **before** calling the parent constructor, so the per-instance
        `OUTPUT_KIND` is set from the resolved row(s). The parent
        `__init__` runs `_initialize` first (token mint), so the
        resolution cannot live there.

        Args:
            start: Inclusive start date as a string (parsed with `fmt`).
            end: Inclusive end date as a string.
            variables: Mapping from curated dataset key to a list of
                selectors, e.g. `{"msg-hrseviri": ["HRSEVIRI"]}`.
                Selectors are informational for the whole-product fetch
                (`G2`).
            lat_lim: `[lat_min, lat_max]` in degrees.
            lon_lim: `[lon_min, lon_max]` in degrees.
            temporal_resolution: Advisory cadence label. Defaults to
                `"daily"`.
            path: Output directory. Created by the parent class if it
                does not exist.
            fmt: `strptime` format for `start` / `end`. Defaults to
                `"%Y-%m-%d"`.
            group: Optional `DataStoreGroup` (or its string value) used to
                assert which Data Store group the requested collection(s)
                belong to (`G2`).
            consumer_key: EUMETSAT consumer key. Falls back to
                `EUMETSAT_CONSUMER_KEY`, then `~/.eumdac/credentials`.
            consumer_secret: EUMETSAT consumer secret. Falls back to
                `EUMETSAT_CONSUMER_SECRET`, then the credentials file.
            credentials_file: Optional explicit path to a `key,secret`
                credentials file.

        Raises:
            ValueError: When `variables` is empty, a dataset key is
                unknown, or the requested collections do not all share
                one `output_kind`.
        """
        self._group = group
        self._consumer_key = consumer_key
        self._consumer_secret = consumer_secret
        self._credentials_file = (
            Path(credentials_file) if credentials_file is not None else None
        )
        self._auth: EumetsatAuth | None = None
        self._show_progress = True

        self._catalog = Catalog()
        self._datasets: list[EumetsatDataset] = self._resolve_datasets(variables)
        self.OUTPUT_KIND = self._unify_output_kind(self._datasets)

        super().__init__(
            start=start,
            end=end,
            variables=variables,
            temporal_resolution=temporal_resolution,
            lat_lim=lat_lim,
            lon_lim=lon_lim,
            fmt=fmt,
            path=path,
        )

    def _resolve_datasets(
        self, variables: dict[str, list[str]]
    ) -> list[EumetsatDataset]:
        """Resolve every requested dataset key to a catalog row.

        Args:
            variables: The `{dataset_key: [selector, ...]}` request.

        Returns:
            list[EumetsatDataset]: One row per key, in request order.

        Raises:
            ValueError: When `variables` is empty or a key is unknown
                (the catalog's did-you-mean is surfaced in the message).
        """
        if not variables:
            raise ValueError(
                "EUMETSAT requires a non-empty `variables` mapping of "
                "{dataset_key: [selector, ...]}."
            )
        return [self._catalog.resolve(key, group=self._group) for key in variables]

    @staticmethod
    def _unify_output_kind(datasets: list[EumetsatDataset]) -> OutputKind:
        """Return the single `output_kind` shared by every requested row.

        A backend instance carries exactly one `OUTPUT_KIND`, so a
        request mixing (say) a raster and a vector dataset is ambiguous
        and rejected here.

        Args:
            datasets: The resolved dataset rows.

        Returns:
            OutputKind: The shared `output_kind`.

        Raises:
            ValueError: When the rows do not all share one `output_kind`.
        """
        kinds = {ds.output_kind for ds in datasets}
        if len(kinds) > 1:
            detail = ", ".join(
                f"{ds.collection_id}={ds.output_kind}" for ds in datasets
            )
            raise ValueError(
                "all datasets in one EUMETSAT request must share one "
                f"output_kind; got mixed kinds ({detail}). Split the "
                "request into one call per output kind."
            )
        return kinds.pop()

    def _initialize(self):
        """Build the `EumetsatAuth`; defer token minting.

        Returns `None` — `eumdac` keeps the token on the `EumetsatAuth`
        instance, so the parent class binds no opaque `self.client`. The
        token minting (`EumetsatAuth.configure`, which contacts the auth
        server) is deferred out of construction: it runs on the first
        :meth:`_search` (the `eumdac` collection search authenticates via
        the idempotent `configure()`), so constructing the backend never
        authenticates — but note that a dry-run `search()` does, since the
        `eumdac` data store needs a token.
        """
        creds = EumetsatCredentials(
            consumer_key=self._consumer_key,
            consumer_secret=(
                SecretStr(self._consumer_secret)
                if self._consumer_secret is not None
                else None
            ),
            credentials_file=self._credentials_file,
        )
        self._auth = EumetsatAuth(creds)
        return None

    def _check_input_dates(
        self,
        start: str,
        end: str,
        temporal_resolution: str,
        fmt: str,
    ) -> TemporalExtent:
        """Parse the date range into a `TemporalExtent`.

        Args:
            start: Inclusive start date as a string.
            end: Inclusive end date as a string.
            temporal_resolution: Advisory cadence label; mapped to a
                pandas frequency for the `dates` index when known.
            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.

        Raises:
            ValueError: If `temporal_resolution` is not one of the cadences
                `earthlens.base.CADENCE_ALIASES` accepts.

        Returns:
            TemporalExtent: Frozen model with parsed bounds.

        Raises:
            ValueError: If `start` parses to a date later than `end`.
        """
        self._end_is_date_only = end_is_date_only(end)
        return self._cadence_extent(
            start,
            end,
            fmt=fmt,
            cadence=temporal_resolution,
            accepted=CADENCE_ALIASES,
        )

    def _search(self) -> list[RemoteProduct]:
        """Query the Data Store for products of every requested collection.

        One `Collection.search(bbox=, dtstart=, dtend=)` per resolved
        collection row, scoped to the request bbox and time window. The
        bbox is the `eumdac` `W,S,E,N` comma-string the OpenSearch
        endpoint expects.

        How `end` is interpreted depends on whether it carries a time of
        day. A **date-only** `end` parses to midnight, which would collapse
        a same-day request (`start == end`) to a zero-width instant, so it
        is read as *inclusive of its whole calendar day* and `dtend` is
        widened to `23:59:59.999999`. An `end` that **names a time** means
        that instant and is passed through unchanged — widening it would
        pull every later product of the day, which for a 10-minute
        full-disk cadence is tens of gigabytes the caller never asked for.

        Each returned `eumdac` product becomes one `RemoteProduct` whose
        `metadata` carries the raw product handle and its collection row,
        so `_fetch` can stream without re-querying.

        Returns:
            list[RemoteProduct]: One product per matching Data Store
                product, across every requested collection. An empty list
                (no products in the window) short-circuits the fetch.

        Raises:
            ImportError: When the `[eumetsat]` extra (`eumdac`) is not
                installed — surfaced by `EumetsatAuth.datastore()`.
        """
        assert self._auth is not None  # set by _initialize
        self._auth.configure()
        store = self._auth.datastore()
        # A `SpatialExtent` constrains longitude to a single `[-180, 180]`
        # range with `west <= east`, so it cannot represent an
        # antimeridian-crossing box — a single search bbox always suffices.
        bbox = eumdac_bbox(
            self.space.west, self.space.south, self.space.east, self.space.north
        )
        dtstart = self.time.start_date
        dtend = expand_bare_date_end(
            self.time.end_date, date_only=self._end_is_date_only
        )
        products: list[RemoteProduct] = []
        for ds in self._datasets:
            collection = store.get_collection(ds.collection_id)
            for product in collection.search(
                bbox=bbox,
                dtstart=dtstart,
                dtend=dtend,
            ):
                products.append(
                    RemoteProduct(
                        id=str(product),
                        metadata={"product": product, "dataset": ds},
                    )
                )
        return products

    def _fetch(self, products: list[RemoteProduct]) -> list[Path]:
        """Stream every product `_search` returned to a local file.

        Each `eumdac` product is opened (`Product.open()` — a streaming
        context manager) and copied to `self.root_dir / <id>`, where the
        product id is reduced to a safe basename (`safe_product_filename`)
        so a server-supplied id with a path separator cannot write outside
        the output directory.

        Args:
            products: The products from `_search`.

        Returns:
            list[Path]: Local paths of every fetched product, in search
                order.
        """
        out_paths: list[Path] = []
        for rp in products:
            product = rp.metadata["product"]
            target = self.root_dir / safe_product_filename(str(product))
            with product.open() as src, open(target, "wb") as dst:
                shutil.copyfileobj(src, dst)
            out_paths.append(target)
        return out_paths

    def download(
        self,
        progress_bar: bool = True,
        tailor: TailorConfig | None = None,
    ) -> list[Path]:
        """Search the Data Store and return product paths (native or tailored).

        With no `tailor=`, composes `_search` and `_fetch` to pull every
        product matching the request bbox + window to `self.root_dir` as
        whole native products (unchanged behaviour).

        With `tailor=TailorConfig(...)`, each matching product is routed
        through EUMETSAT **Data Tailor** (server-side subset / reproject /
        reformat, `H4`): submit a customisation, poll it to `DONE`, stream
        the customised output(s) to `self.root_dir`, and delete the
        customisation (`G7`). The returned paths are then the customised
        GeoTIFF / NetCDF files, not the native products (`G6`).

        `aggregate=` is the **temporal** reducer (`G1`) and is a separate
        operation from the spatial `tailor=`; it is not implemented for
        EUMETSAT, so a non-`None` `aggregate` raises `NotImplementedError`.
        The two knobs compose — tailor server-side here, then reduce the
        result client-side with pyramids.

        The tailor branch is **fail-fast per batch**: if one product's
        customisation fails, the error propagates and paths already streamed
        for earlier products are not returned (their files remain on disk).
        This mirrors the native fetch; keep batches small when a partial
        result would be costly to recompute.

        Args:
            progress_bar: Reserved for parity with the other backends;
                `eumdac`'s streaming download has no built-in bar.
            tailor: Optional `TailorConfig` routing the request through
                Data Tailor. `None` (the default) keeps the native fetch.

        Returns:
            list[Path]: The native product paths, or — when `tailor=` is
                given — the customised output paths.

        Raises:
            ValueError: When `tailor=` names a dataset that is not
                Data-Tailor-eligible (`G5`).
        """
        self._show_progress = progress_bar
        if tailor is not None:
            return self._tailor(tailor)
        return self._api_via_search_fetch()

    def _tailor(self, tailor: TailorConfig) -> list[Path]:
        """Run the Data Tailor branch for every matching product (`G2`).

        Rejects a non-eligible request up front (`G5`), then searches the
        Data Store and customises each product in turn, returning the
        flattened list of customised output paths.

        Args:
            tailor: The `TailorConfig` describing the customisation.

        Returns:
            list[Path]: Every customised output path, in search order.

        Raises:
            ValueError: When any requested dataset lacks a
                `tailor_product_type` (not Data-Tailor-eligible).
        """
        ineligible = [ds for ds in self._datasets if ds.tailor_product_type is None]
        if ineligible:
            names = ", ".join(ds.collection_id for ds in ineligible)
            raise ValueError(
                f"{names} not Data-Tailor-eligible; download native (no "
                "tailor=) and reduce client-side with pyramids."
            )
        assert self._auth is not None  # set by _initialize
        self._auth.configure()
        datatailor = self._auth.datatailor()
        products = self._search()
        out_paths: list[Path] = []
        used_dirs: set[str] = set()
        for rp in products:
            out_paths.extend(self._tailor_one(rp, tailor, datatailor, used_dirs))
        return out_paths

    @staticmethod
    def _dedupe_name(name: str, used: set[str]) -> str:
        """Return `name`, suffixed if needed so it is unique within `used`.

        Guarantees a distinct on-disk name even when two products share an
        id (`L3`) or two customisation outputs sanitise to one basename
        (`L2`). Adds the chosen name to `used`.

        Args:
            name: The candidate (already path-safe) name.
            used: The set of names already taken; mutated in place.

        Returns:
            str: A name not already in `used`.
        """
        candidate = name
        counter = 1
        while candidate in used:
            candidate = f"{name}_{counter}"
            counter += 1
        used.add(candidate)
        return candidate

    def _tailor_one(
        self,
        product: RemoteProduct,
        tailor: TailorConfig,
        datatailor,
        used_dirs: set[str],
    ) -> list[Path]:
        """Customise one product via Data Tailor; always clean up (`G7`).

        Builds the `eumdac` `Chain` from `tailor`, the product's catalog
        row (`tailor_product_type`), and the request ROI (`tailor.bbox`
        else `self.space`), submits it, polls to a terminal state, streams
        every output to `self.root_dir`, and deletes the customisation in
        a `finally` — even on failure — so quota is always freed.

        Args:
            product: One `RemoteProduct` from `_search` (its `metadata`
                carries the raw `eumdac` product handle and catalog row).
            tailor: The customisation request.
            datatailor: The live `eumdac.DataTailor` client.
            used_dirs: Shared per-batch set of subdirectory names already
                taken, mutated here to keep each product's output directory
                unique across the request (`L3`).

        Returns:
            list[Path]: The customised output paths for this product.

        Raises:
            ValueError: When the product's dataset is not eligible (`G5`).
            RuntimeError: When the customisation ends `FAILED` / `KILLED`
                (with the server log), or a transient submit keeps failing.
            TimeoutError: When polling exceeds `TAILOR_POLL_TIMEOUT_S`.
        """
        import eumdac  # lazy — the [eumetsat] extra

        dataset: EumetsatDataset = product.metadata["dataset"]
        if dataset.tailor_product_type is None:
            raise ValueError(
                f"{dataset.collection_id!r} is not Data-Tailor-eligible; "
                "download native (no tailor=) and reduce client-side."
            )
        nswe = tailor.nswe or TailorConfig.nswe_from_extent(
            self.space.north, self.space.south, self.space.west, self.space.east
        )
        chain = eumdac.tailor_models.Chain(
            product=dataset.tailor_product_type,
            format=tailor.format,
            projection=tailor.crs,
            # eumdac types NSWE as Optional[str], but the Data Tailor ROI takes
            # a north/south/west/east list (see TailorConfig.nswe).
            roi=eumdac.tailor_models.RegionOfInterest(NSWE=nswe),  # type: ignore[arg-type]
            filter=(
                eumdac.tailor_models.Filter(bands=list(tailor.filter))
                if tailor.filter
                else None
            ),
            # eumdac types quicklook as a Quicklook/dict; the API accepts a truthy flag.
            quicklook=tailor.quicklook or None,  # type: ignore[arg-type]
        )
        product_handle = product.metadata["product"]
        # Choose the per-product output subdir *before* submitting, so nothing
        # between the submit and the `try/finally` can raise and orphan the
        # customisation (quota hygiene, G7). Namespacing avoids cross-granule
        # basename collisions (H1); the name is de-duped against this batch and
        # against any pre-existing native file of the same basename (L3).
        subdir = self._dedupe_name(
            safe_product_filename(str(product_handle)), used_dirs
        )
        while (self.root_dir / subdir).exists() and not (
            self.root_dir / subdir
        ).is_dir():
            subdir = self._dedupe_name(subdir, used_dirs)
        product_dir = self.root_dir / subdir
        cust = self._submit_customisation(datatailor, product_handle, chain)
        try:
            status = self._poll_customisation(cust)
            if status != "DONE":
                raise RuntimeError(
                    f"Data Tailor customisation {cust} ended {status}: "
                    f"{self._logfile_tail(cust)}"
                )
            product_dir.mkdir(parents=True, exist_ok=True)
            written: list[Path] = []
            used_names: set[str] = set()
            for name in cust.outputs:
                # De-dupe within a customisation so two outputs sharing a
                # basename do not overwrite each other (L2).
                out_name = self._dedupe_name(
                    safe_product_filename(str(name)), used_names
                )
                target = product_dir / out_name
                with cust.stream_output(name) as src, open(target, "wb") as fh:
                    shutil.copyfileobj(src, fh)
                written.append(target)
            return written
        finally:
            self._safe_delete(cust)  # ALWAYS free quota — even on failure (G7)

    @staticmethod
    def _safe_delete(cust) -> None:
        """Delete a customisation, never letting cleanup mask the real error.

        Called from the `_tailor_one` `finally` (`G7`). A `delete()` that
        itself raises must not replace the original `RuntimeError` /
        `TimeoutError`, nor abort the remaining products in the batch — so
        the failure is logged and swallowed.

        Args:
            cust: The `eumdac` `Customisation` handle to delete.
        """
        try:
            cust.delete()
        except Exception as exc:  # noqa: BLE001 - cleanup must never mask the cause
            logger.warning(f"Data Tailor customisation {cust} delete failed: {exc}")

    @staticmethod
    def _submit_customisation(datatailor, product, chain):
        """Submit a customisation, retrying transient EPCS failures (`G8`).

        The EUMETSAT EPCS endpoint intermittently returns `502 Bad
        Gateway`; a submit that fails with a transient marker is retried
        up to `TAILOR_SUBMIT_RETRIES` times with a linear backoff. A
        non-transient error (e.g. an invalid product id) is re-raised
        immediately.

        Note:
            If a create actually succeeds server-side but its response is
            lost (a dropped connection / timeout — the classified-transient
            cases), the retry submits a **second** customisation and the
            first is orphaned: the client never gets its handle, so it is
            not polled or deleted and lingers against the quota. The EPCS
            API offers no idempotency key to prevent this; recover by
            sweeping stale jobs (`eumdac.DataTailor(token).customisations`).

        Args:
            datatailor: The live `eumdac.DataTailor` client.
            product: The `eumdac` product handle to customise.
            chain: The `eumdac.tailor_models.Chain` describing the job.

        Returns:
            The `eumdac` `Customisation` handle for the submitted job.

        Raises:
            RuntimeError: When a transient submit keeps failing after
                `TAILOR_SUBMIT_RETRIES` attempts.
            Exception: Any non-transient submit error, re-raised as-is.
        """
        last_exc: Exception | None = None
        for attempt in range(1, TAILOR_SUBMIT_RETRIES + 1):
            try:
                return datatailor.new_customisation(product, chain)
            except Exception as exc:  # noqa: BLE001 - classified below
                message = str(exc).lower()
                if not any(mark in message for mark in _TRANSIENT_MARKERS):
                    raise
                last_exc = exc
                if attempt < TAILOR_SUBMIT_RETRIES:
                    time.sleep(TAILOR_SUBMIT_BACKOFF_S * attempt)
        raise RuntimeError(
            f"Data Tailor submit failed after {TAILOR_SUBMIT_RETRIES} "
            f"transient attempts: {last_exc}"
        ) from last_exc

    @staticmethod
    def _poll_customisation(cust) -> str:
        """Poll a customisation until it stops being active; return its status.

        Polls `cust.status` while it is `_TAILOR_ACTIVE` (`QUEUED` /
        `RUNNING`), sleeping `TAILOR_POLL_INITIAL_S` and growing the delay
        by `TAILOR_POLL_BACKOFF` up to `TAILOR_POLL_MAX_S`. Any other status
        is terminal and is returned immediately — so `DONE` succeeds and
        `FAILED` / `KILLED` / an unexpected stuck state (e.g. `INACTIVE`)
        fail fast rather than polling to the timeout. Gives up after
        `TAILOR_POLL_TIMEOUT_S` so a job stuck *active* cannot hang forever
        (`G8`); the final sleep is clamped to the remaining budget so the
        wall-clock never overshoots the timeout.

        Args:
            cust: The `eumdac` `Customisation` handle to poll.

        Returns:
            str: The terminal status (`"DONE"`, `"FAILED"`, `"KILLED"`, or
                any non-active value the service reports).

        Raises:
            TimeoutError: When the job is still active after
                `TAILOR_POLL_TIMEOUT_S`.
        """
        deadline = time.monotonic() + TAILOR_POLL_TIMEOUT_S
        delay = TAILOR_POLL_INITIAL_S
        while True:
            status = str(cust.status).upper()
            if status not in _TAILOR_ACTIVE:
                return status
            now = time.monotonic()
            if now >= deadline:
                raise TimeoutError(
                    f"Data Tailor customisation {cust} did not finish "
                    f"within {TAILOR_POLL_TIMEOUT_S:.0f}s (last status "
                    f"{status!r})."
                )
            time.sleep(min(delay, deadline - now))
            delay = min(delay * TAILOR_POLL_BACKOFF, TAILOR_POLL_MAX_S)

    @staticmethod
    def _logfile_tail(cust, limit: int = 1500) -> str:
        """Return the tail of a customisation's server log, if any.

        Args:
            cust: The `eumdac` `Customisation` handle.
            limit: Maximum number of trailing characters to return.

        Returns:
            str: The last `limit` characters of `cust.logfile`, or a
                placeholder when no log is available.
        """
        try:
            log = cust.logfile
        except Exception:  # noqa: BLE001 - a missing log must not mask the failure
            log = None
        if not log:
            return "(no customisation log available)"
        return str(log)[-limit:]

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

Initialise an EUMETSAT backend instance.

Resolves every requested dataset key against the catalog before calling the parent constructor, so the per-instance OUTPUT_KIND is set from the resolved row(s). The parent __init__ runs _initialize first (token mint), so the resolution cannot live there.

Parameters:

Name	Type	Description	Default
start	str	Inclusive start date as a string (parsed with fmt).	required
end	str	Inclusive end date as a string.	required
variables	dict[str, list[str]]	Mapping from curated dataset key to a list of selectors, e.g. {"msg-hrseviri": ["HRSEVIRI"]}. Selectors are informational for the whole-product fetch (G2).	required
lat_lim	list[float]	[lat_min, lat_max] in degrees.	required
lon_lim	list[float]	[lon_min, lon_max] in degrees.	required
temporal_resolution	str	Advisory cadence label. Defaults to "daily".	'daily'
path	Path | str | None	Output directory. Created by the parent class if it does not exist.	None
fmt	str	strptime format for start / end. Defaults to "%Y-%m-%d".	'%Y-%m-%d'
group	DataStoreGroup | str | None	Optional DataStoreGroup (or its string value) used to assert which Data Store group the requested collection(s) belong to (G2).	None
consumer_key	str | None	EUMETSAT consumer key. Falls back to EUMETSAT_CONSUMER_KEY, then ~/.eumdac/credentials.	None
consumer_secret	str | None	EUMETSAT consumer secret. Falls back to EUMETSAT_CONSUMER_SECRET, then the credentials file.	None
credentials_file	Path | str | None	Optional explicit path to a key,secret credentials file.	None

Raises:

Type	Description
ValueError	When variables is empty, a dataset key is unknown, or the requested collections do not all share one output_kind.

Source code in libs/providers/imagery/src/earthlens/eumetsat/backend.py

def __init__(
    self,
    start: str,
    end: str,
    variables: dict[str, list[str]],
    lat_lim: list[float],
    lon_lim: list[float],
    temporal_resolution: str = "daily",
    path: Path | str | None = None,
    fmt: str = "%Y-%m-%d",
    group: DataStoreGroup | str | None = None,
    consumer_key: str | None = None,
    consumer_secret: str | None = None,
    credentials_file: Path | str | None = None,
):
    """Initialise an EUMETSAT backend instance.

    Resolves every requested dataset key against the catalog
    **before** calling the parent constructor, so the per-instance
    `OUTPUT_KIND` is set from the resolved row(s). The parent
    `__init__` runs `_initialize` first (token mint), so the
    resolution cannot live there.

    Args:
        start: Inclusive start date as a string (parsed with `fmt`).
        end: Inclusive end date as a string.
        variables: Mapping from curated dataset key to a list of
            selectors, e.g. `{"msg-hrseviri": ["HRSEVIRI"]}`.
            Selectors are informational for the whole-product fetch
            (`G2`).
        lat_lim: `[lat_min, lat_max]` in degrees.
        lon_lim: `[lon_min, lon_max]` in degrees.
        temporal_resolution: Advisory cadence label. Defaults to
            `"daily"`.
        path: Output directory. Created by the parent class if it
            does not exist.
        fmt: `strptime` format for `start` / `end`. Defaults to
            `"%Y-%m-%d"`.
        group: Optional `DataStoreGroup` (or its string value) used to
            assert which Data Store group the requested collection(s)
            belong to (`G2`).
        consumer_key: EUMETSAT consumer key. Falls back to
            `EUMETSAT_CONSUMER_KEY`, then `~/.eumdac/credentials`.
        consumer_secret: EUMETSAT consumer secret. Falls back to
            `EUMETSAT_CONSUMER_SECRET`, then the credentials file.
        credentials_file: Optional explicit path to a `key,secret`
            credentials file.

    Raises:
        ValueError: When `variables` is empty, a dataset key is
            unknown, or the requested collections do not all share
            one `output_kind`.
    """
    self._group = group
    self._consumer_key = consumer_key
    self._consumer_secret = consumer_secret
    self._credentials_file = (
        Path(credentials_file) if credentials_file is not None else None
    )
    self._auth: EumetsatAuth | None = None
    self._show_progress = True

    self._catalog = Catalog()
    self._datasets: list[EumetsatDataset] = self._resolve_datasets(variables)
    self.OUTPUT_KIND = self._unify_output_kind(self._datasets)

    super().__init__(
        start=start,
        end=end,
        variables=variables,
        temporal_resolution=temporal_resolution,
        lat_lim=lat_lim,
        lon_lim=lon_lim,
        fmt=fmt,
        path=path,
    )

download(progress_bar=True, tailor=None)

Search the Data Store and return product paths (native or tailored).

With no tailor=, composes _search and _fetch to pull every product matching the request bbox + window to self.root_dir as whole native products (unchanged behaviour).

With tailor=TailorConfig(...), each matching product is routed through EUMETSAT Data Tailor (server-side subset / reproject / reformat, H4): submit a customisation, poll it to DONE, stream the customised output(s) to self.root_dir, and delete the customisation (G7). The returned paths are then the customised GeoTIFF / NetCDF files, not the native products (G6).

aggregate= is the temporal reducer (G1) and is a separate operation from the spatial tailor=; it is not implemented for EUMETSAT, so a non-None aggregate raises NotImplementedError. The two knobs compose — tailor server-side here, then reduce the result client-side with pyramids.

The tailor branch is fail-fast per batch: if one product's customisation fails, the error propagates and paths already streamed for earlier products are not returned (their files remain on disk). This mirrors the native fetch; keep batches small when a partial result would be costly to recompute.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Reserved for parity with the other backends; eumdac's streaming download has no built-in bar.	True
tailor	TailorConfig | None	Optional TailorConfig routing the request through Data Tailor. None (the default) keeps the native fetch.	None

Returns:

Type	Description
list[Path]	list[Path]: The native product paths, or — when tailor= is given — the customised output paths.

Raises:

Type	Description
ValueError	When tailor= names a dataset that is not Data-Tailor-eligible (G5).

Source code in libs/providers/imagery/src/earthlens/eumetsat/backend.py

def download(
    self,
    progress_bar: bool = True,
    tailor: TailorConfig | None = None,
) -> list[Path]:
    """Search the Data Store and return product paths (native or tailored).

    With no `tailor=`, composes `_search` and `_fetch` to pull every
    product matching the request bbox + window to `self.root_dir` as
    whole native products (unchanged behaviour).

    With `tailor=TailorConfig(...)`, each matching product is routed
    through EUMETSAT **Data Tailor** (server-side subset / reproject /
    reformat, `H4`): submit a customisation, poll it to `DONE`, stream
    the customised output(s) to `self.root_dir`, and delete the
    customisation (`G7`). The returned paths are then the customised
    GeoTIFF / NetCDF files, not the native products (`G6`).

    `aggregate=` is the **temporal** reducer (`G1`) and is a separate
    operation from the spatial `tailor=`; it is not implemented for
    EUMETSAT, so a non-`None` `aggregate` raises `NotImplementedError`.
    The two knobs compose — tailor server-side here, then reduce the
    result client-side with pyramids.

    The tailor branch is **fail-fast per batch**: if one product's
    customisation fails, the error propagates and paths already streamed
    for earlier products are not returned (their files remain on disk).
    This mirrors the native fetch; keep batches small when a partial
    result would be costly to recompute.

    Args:
        progress_bar: Reserved for parity with the other backends;
            `eumdac`'s streaming download has no built-in bar.
        tailor: Optional `TailorConfig` routing the request through
            Data Tailor. `None` (the default) keeps the native fetch.

    Returns:
        list[Path]: The native product paths, or — when `tailor=` is
            given — the customised output paths.

    Raises:
        ValueError: When `tailor=` names a dataset that is not
            Data-Tailor-eligible (`G5`).
    """
    self._show_progress = progress_bar
    if tailor is not None:
        return self._tailor(tailor)
    return self._api_via_search_fetch()

EumetsatAuth

Bases: AbstractAuth[EumetsatCredentials]

Authenticate against the EUMETSAT Data Store (OAuth2).

Wraps eumdac.AccessToken in the earthlens.base.AbstractAuth contract. configure() resolves a consumer key / secret pair (environment → constructor kwargs → ~/.eumdac/credentials) and mints an eumdac.AccessToken, which auto-refreshes the ~1 h bearer internally. The datastore / datatailor helpers build the matching eumdac clients from the live token.

The class is a context manager (inherited from AbstractAuth): with EumetsatAuth(creds) as auth: ... calls configure() on enter and the default no-op close() on exit.

Attributes:

Name	Type	Description
_creds		The EumetsatCredentials passed at construction. Read by configure to resolve the credential pair. Treated as write-once.
_token		The eumdac.AccessToken minted by a successful configure(), or None before it runs.

Examples:

• Build, configure, inspect — marked # doctest: +SKIP because it mints a real OAuth2 token:

  >>> from earthlens.eumetsat import EumetsatAuth, EumetsatCredentials
  >>> auth = EumetsatAuth(EumetsatCredentials())  # doctest: +SKIP
  >>> auth.configure()  # doctest: +SKIP
  >>> auth.is_authenticated()  # doctest: +SKIP
  True
  

Source code in libs/providers/imagery/src/earthlens/eumetsat/auth.py

class EumetsatAuth(AbstractAuth[EumetsatCredentials]):
    """Authenticate against the EUMETSAT Data Store (OAuth2).

    Wraps `eumdac.AccessToken` in the `earthlens.base.AbstractAuth`
    contract. `configure()` resolves a consumer key / secret pair
    (environment → constructor kwargs → `~/.eumdac/credentials`) and
    mints an `eumdac.AccessToken`, which auto-refreshes the ~1 h bearer
    internally. The `datastore` / `datatailor` helpers build the matching
    `eumdac` clients from the live token.

    The class is a context manager (inherited from `AbstractAuth`):
    `with EumetsatAuth(creds) as auth: ...` calls `configure()` on enter
    and the default no-op `close()` on exit.

    Attributes:
        _creds: The `EumetsatCredentials` passed at construction. Read by
            `configure` to resolve the credential pair. Treated as
            write-once.
        _token: The `eumdac.AccessToken` minted by a successful
            `configure()`, or `None` before it runs.

    Examples:
        - Build, configure, inspect — marked `# doctest: +SKIP` because
          it mints a real OAuth2 token:

            ```python
            >>> from earthlens.eumetsat import EumetsatAuth, EumetsatCredentials
            >>> auth = EumetsatAuth(EumetsatCredentials())  # doctest: +SKIP
            >>> auth.configure()  # doctest: +SKIP
            >>> auth.is_authenticated()  # doctest: +SKIP
            True

            ```
    """

    def __init__(self, credentials: EumetsatCredentials) -> None:
        """Store credentials; does not authenticate.

        Construction is side-effect-free — the user (or the backend's
        `_initialize`) must call `configure` (or use the context-manager
        form) to mint a token.

        Args:
            credentials: The `EumetsatCredentials` value object carrying
                the resolution rules.
        """
        super().__init__(credentials)
        self._token = None

    def _resolve_pair(self) -> tuple[str | None, str | None]:
        """Resolve the consumer key / secret pair.

        Resolution order: the `EUMETSAT_CONSUMER_KEY` /
        `EUMETSAT_CONSUMER_SECRET` environment variables, then the
        constructor kwargs, then a `key,secret` line in the
        credentials file (the explicit `credentials_file`, else
        `EUMDAC_CONFIG_DIR/credentials`, else `~/.eumdac/credentials`).
        The first source that yields both halves wins.

        Returns:
            tuple[str | None, str | None]: The `(consumer_key,
                consumer_secret)` pair; either element is `None` when no
                source supplied it.
        """
        key = os.getenv("EUMETSAT_CONSUMER_KEY") or self._creds.consumer_key
        secret = os.getenv("EUMETSAT_CONSUMER_SECRET")
        if not secret and self._creds.consumer_secret is not None:
            secret = self._creds.consumer_secret.get_secret_value()
        if key and secret:
            return key, secret
        file_key, file_secret = self._read_credentials_file()
        return key or file_key, secret or file_secret

    def _credentials_path(self) -> Path:
        """Return the credentials-file path to read.

        Honours an explicit `credentials_file`, then the
        `EUMDAC_CONFIG_DIR` environment variable `eumdac` itself reads,
        then the default `~/.eumdac/credentials`.

        Returns:
            Path: The resolved credentials-file path (which may not
                exist).
        """
        if self._creds.credentials_file is not None:
            return self._creds.credentials_file
        config_dir = os.getenv("EUMDAC_CONFIG_DIR")
        base = Path(config_dir) if config_dir else Path.home() / ".eumdac"
        return base / "credentials"

    def _read_credentials_file(self) -> tuple[str | None, str | None]:
        """Parse the `key,secret` credentials file, if present.

        Returns:
            tuple[str | None, str | None]: The `(key, secret)` parsed
                from the single `key,secret` line, or `(None, None)` when
                the file is missing or malformed.
        """
        path = self._credentials_path()
        try:
            content = path.read_text(encoding="utf-8").strip()
        except (FileNotFoundError, OSError):
            return None, None
        match = _CREDENTIALS_LINE.match(content)
        if match is None:
            return None, None
        return match.group(1), match.group(2)

    def configure(self) -> None:
        """Mint the OAuth2 token via `eumdac.AccessToken`.

        Idempotent — short-circuits when `is_authenticated` already
        returns `True`. Resolves the consumer key / secret pair
        (environment → kwargs → credentials file), then constructs an
        `eumdac.AccessToken((key, secret))`. The token object refreshes
        the ~1 h bearer internally; this wrapper re-mints a fresh one
        only after the cached token's `expiration` has passed (see
        `is_authenticated`).

        Raises:
            ImportError: When the `eumdac` SDK is not installed (the
                `[eumetsat]` extra is missing).
            AuthenticationError: When no credential pair resolves, or
                `eumdac` rejects the pair while minting the token.
        """
        if self.is_authenticated():
            return

        eumdac = _import_eumdac()  # lazy — only needed when authenticating

        key, secret = self._resolve_pair()
        if not key or not secret:
            raise AuthenticationError(
                "no EUMETSAT credentials resolved. Set "
                "EUMETSAT_CONSUMER_KEY / EUMETSAT_CONSUMER_SECRET, pass "
                "consumer_key= / consumer_secret=, or write a "
                f"'key,secret' line to ~/.eumdac/credentials. Register a "
                f"consumer key/secret at {_KEY_MGMT_URL}. See {_DOCS_URL}."
            )

        try:
            self._token = eumdac.AccessToken((key, secret))
        except Exception as exc:  # noqa: BLE001 - re-raised as AuthenticationError
            raise AuthenticationError(
                "EUMETSAT token request failed "
                f"({type(exc).__name__}: {exc}). Check the consumer "
                f"key/secret at {_KEY_MGMT_URL}."
            ) from exc

    def is_authenticated(self) -> bool:
        """Return whether a live, unexpired token is held.

        Cheap predicate — does not call the network. Returns `True` only
        when `configure()` has minted a token whose `expiration`
        (a `datetime`) is still in the future; an expired token reports
        `False` so `configure()` re-mints. A token whose `expiration`
        cannot be read is treated as live (the SDK refreshes it lazily).

        Returns:
            bool: `True` while the held token is valid, `False` before
                `configure` or once the token has expired.
        """
        if self._token is None:
            return False
        try:
            return datetime.now() < self._token.expiration
        except (AttributeError, TypeError, ValueError):
            # The SDK refreshes the bearer lazily, so a token whose
            # `expiration` is missing/unreadable is treated as live rather
            # than forcing a re-mint. Genuinely unexpected errors propagate.
            return True

    def datastore(self) -> Any:
        """Return an `eumdac.DataStore` bound to the live token.

        Returns:
            eumdac.DataStore: The Data Store client used to resolve
                collections and search products.

        Raises:
            ImportError: When the `[eumetsat]` extra (`eumdac`) is not
                installed.
            AuthenticationError: When `configure()` has not minted a
                token yet.
        """
        eumdac = _import_eumdac()
        if self._token is None:
            raise AuthenticationError(
                "datastore() called before configure(); authenticate "
                "first via configure() or the context-manager form."
            )
        return eumdac.DataStore(self._token)

    def datatailor(self) -> Any:
        """Return an `eumdac.DataTailor` bound to the live token.

        Used by the Data Tailor (server-side subset / reproject / reformat)
        path — `EUMETSAT.download(tailor=...)`. The native whole-product
        fetch does not call this.

        Returns:
            eumdac.DataTailor: The Data Tailor client.

        Raises:
            ImportError: When the `[eumetsat]` extra (`eumdac`) is not
                installed.
            AuthenticationError: When `configure()` has not minted a
                token yet.
        """
        eumdac = _import_eumdac()
        if self._token is None:
            raise AuthenticationError(
                "datatailor() called before configure(); authenticate "
                "first via configure() or the context-manager form."
            )
        return eumdac.DataTailor(self._token)

__init__(credentials)

Store credentials; does not authenticate.

Construction is side-effect-free — the user (or the backend's _initialize) must call configure (or use the context-manager form) to mint a token.

Parameters:

Name	Type	Description	Default
credentials	EumetsatCredentials	The EumetsatCredentials value object carrying the resolution rules.	required

Source code in libs/providers/imagery/src/earthlens/eumetsat/auth.py

def __init__(self, credentials: EumetsatCredentials) -> None:
    """Store credentials; does not authenticate.

    Construction is side-effect-free — the user (or the backend's
    `_initialize`) must call `configure` (or use the context-manager
    form) to mint a token.

    Args:
        credentials: The `EumetsatCredentials` value object carrying
            the resolution rules.
    """
    super().__init__(credentials)
    self._token = None

configure()

Mint the OAuth2 token via eumdac.AccessToken.

Idempotent — short-circuits when is_authenticated already returns True. Resolves the consumer key / secret pair (environment → kwargs → credentials file), then constructs an eumdac.AccessToken((key, secret)). The token object refreshes the ~1 h bearer internally; this wrapper re-mints a fresh one only after the cached token's expiration has passed (see is_authenticated).

Raises:

Type	Description
ImportError	When the eumdac SDK is not installed (the [eumetsat] extra is missing).
AuthenticationError	When no credential pair resolves, or eumdac rejects the pair while minting the token.

Source code in libs/providers/imagery/src/earthlens/eumetsat/auth.py

def configure(self) -> None:
    """Mint the OAuth2 token via `eumdac.AccessToken`.

    Idempotent — short-circuits when `is_authenticated` already
    returns `True`. Resolves the consumer key / secret pair
    (environment → kwargs → credentials file), then constructs an
    `eumdac.AccessToken((key, secret))`. The token object refreshes
    the ~1 h bearer internally; this wrapper re-mints a fresh one
    only after the cached token's `expiration` has passed (see
    `is_authenticated`).

    Raises:
        ImportError: When the `eumdac` SDK is not installed (the
            `[eumetsat]` extra is missing).
        AuthenticationError: When no credential pair resolves, or
            `eumdac` rejects the pair while minting the token.
    """
    if self.is_authenticated():
        return

    eumdac = _import_eumdac()  # lazy — only needed when authenticating

    key, secret = self._resolve_pair()
    if not key or not secret:
        raise AuthenticationError(
            "no EUMETSAT credentials resolved. Set "
            "EUMETSAT_CONSUMER_KEY / EUMETSAT_CONSUMER_SECRET, pass "
            "consumer_key= / consumer_secret=, or write a "
            f"'key,secret' line to ~/.eumdac/credentials. Register a "
            f"consumer key/secret at {_KEY_MGMT_URL}. See {_DOCS_URL}."
        )

    try:
        self._token = eumdac.AccessToken((key, secret))
    except Exception as exc:  # noqa: BLE001 - re-raised as AuthenticationError
        raise AuthenticationError(
            "EUMETSAT token request failed "
            f"({type(exc).__name__}: {exc}). Check the consumer "
            f"key/secret at {_KEY_MGMT_URL}."
        ) from exc

datastore()

Return an eumdac.DataStore bound to the live token.

Returns:

Type	Description
Any	eumdac.DataStore: The Data Store client used to resolve collections and search products.

Raises:

Type	Description
ImportError	When the [eumetsat] extra (eumdac) is not installed.
AuthenticationError	When configure() has not minted a token yet.

Source code in libs/providers/imagery/src/earthlens/eumetsat/auth.py

def datastore(self) -> Any:
    """Return an `eumdac.DataStore` bound to the live token.

    Returns:
        eumdac.DataStore: The Data Store client used to resolve
            collections and search products.

    Raises:
        ImportError: When the `[eumetsat]` extra (`eumdac`) is not
            installed.
        AuthenticationError: When `configure()` has not minted a
            token yet.
    """
    eumdac = _import_eumdac()
    if self._token is None:
        raise AuthenticationError(
            "datastore() called before configure(); authenticate "
            "first via configure() or the context-manager form."
        )
    return eumdac.DataStore(self._token)

datatailor()

Return an eumdac.DataTailor bound to the live token.

Used by the Data Tailor (server-side subset / reproject / reformat) path — EUMETSAT.download(tailor=...). The native whole-product fetch does not call this.

Returns:

Type	Description
Any	eumdac.DataTailor: The Data Tailor client.

Raises:

Type	Description
ImportError	When the [eumetsat] extra (eumdac) is not installed.
AuthenticationError	When configure() has not minted a token yet.

Source code in libs/providers/imagery/src/earthlens/eumetsat/auth.py

def datatailor(self) -> Any:
    """Return an `eumdac.DataTailor` bound to the live token.

    Used by the Data Tailor (server-side subset / reproject / reformat)
    path — `EUMETSAT.download(tailor=...)`. The native whole-product
    fetch does not call this.

    Returns:
        eumdac.DataTailor: The Data Tailor client.

    Raises:
        ImportError: When the `[eumetsat]` extra (`eumdac`) is not
            installed.
        AuthenticationError: When `configure()` has not minted a
            token yet.
    """
    eumdac = _import_eumdac()
    if self._token is None:
        raise AuthenticationError(
            "datatailor() called before configure(); authenticate "
            "first via configure() or the context-manager form."
        )
    return eumdac.DataTailor(self._token)

is_authenticated()

Return whether a live, unexpired token is held.

Cheap predicate — does not call the network. Returns True only when configure() has minted a token whose expiration (a datetime) is still in the future; an expired token reports False so configure() re-mints. A token whose expiration cannot be read is treated as live (the SDK refreshes it lazily).

Returns:

Name	Type	Description
bool	bool	True while the held token is valid, False before configure or once the token has expired.

Source code in libs/providers/imagery/src/earthlens/eumetsat/auth.py

def is_authenticated(self) -> bool:
    """Return whether a live, unexpired token is held.

    Cheap predicate — does not call the network. Returns `True` only
    when `configure()` has minted a token whose `expiration`
    (a `datetime`) is still in the future; an expired token reports
    `False` so `configure()` re-mints. A token whose `expiration`
    cannot be read is treated as live (the SDK refreshes it lazily).

    Returns:
        bool: `True` while the held token is valid, `False` before
            `configure` or once the token has expired.
    """
    if self._token is None:
        return False
    try:
        return datetime.now() < self._token.expiration
    except (AttributeError, TypeError, ValueError):
        # The SDK refreshes the bearer lazily, so a token whose
        # `expiration` is missing/unreadable is treated as live rather
        # than forcing a re-mint. Genuinely unexpected errors propagate.
        return True

EumetsatCredentials

Bases: BaseModel

Frozen value object holding the EUMETSAT consumer key / secret.

Every field is optional — the auth wrapper resolves the actual pair at configure() time from the environment, these fields, then the ~/.eumdac/credentials file. Validation is intentionally permissive; the real "do these creds work?" gate is EumetsatAuth.configure, which mints a token against the OAuth2 endpoint.

Attributes:

Name	Type	Description
consumer_key	str | None	The EUMETSAT consumer key (the OAuth2 client id). None means "look at the environment / credentials file".
consumer_secret	SecretStr | None	The matching consumer secret, stored as a pydantic.SecretStr so it is never echoed by repr(creds) or in logs. None means same as consumer_key.
credentials_file	Path | None	Optional explicit path to a key,secret credentials file. None falls back to EUMDAC_CONFIG_DIR/credentials, then ~/.eumdac/credentials.

Examples:

• All fields optional — rely on env / credentials file:

  >>> from earthlens.eumetsat import EumetsatCredentials
  >>> creds = EumetsatCredentials()
  >>> creds.consumer_key is None and creds.consumer_secret is None
  True
  

• SecretStr hides the secret in repr:

  >>> from earthlens.eumetsat import EumetsatCredentials
  >>> creds = EumetsatCredentials(consumer_key="k", consumer_secret="topsecret")
  >>> "topsecret" in repr(creds)
  False
  

Source code in libs/providers/imagery/src/earthlens/eumetsat/auth.py

class EumetsatCredentials(BaseModel):
    """Frozen value object holding the EUMETSAT consumer key / secret.

    Every field is optional — the auth wrapper resolves the actual pair
    at `configure()` time from the environment, these fields, then the
    `~/.eumdac/credentials` file. Validation is intentionally permissive;
    the real "do these creds work?" gate is
    `EumetsatAuth.configure`, which mints a token against the OAuth2
    endpoint.

    Attributes:
        consumer_key: The EUMETSAT consumer key (the OAuth2 client id).
            `None` means "look at the environment / credentials file".
        consumer_secret: The matching consumer secret, stored as a
            `pydantic.SecretStr` so it is never echoed by `repr(creds)`
            or in logs. `None` means same as `consumer_key`.
        credentials_file: Optional explicit path to a `key,secret`
            credentials file. `None` falls back to
            `EUMDAC_CONFIG_DIR/credentials`, then `~/.eumdac/credentials`.

    Examples:
        - All fields optional — rely on env / credentials file:
            ```python
            >>> from earthlens.eumetsat import EumetsatCredentials
            >>> creds = EumetsatCredentials()
            >>> creds.consumer_key is None and creds.consumer_secret is None
            True

            ```
        - SecretStr hides the secret in repr:
            ```python
            >>> from earthlens.eumetsat import EumetsatCredentials
            >>> creds = EumetsatCredentials(consumer_key="k", consumer_secret="topsecret")
            >>> "topsecret" in repr(creds)
            False

            ```
    """

    model_config = ConfigDict(frozen=True)

    consumer_key: str | None = None
    consumer_secret: SecretStr | None = None
    credentials_file: Path | None = None

EumetsatDataset

Bases: BaseModel

One curated EUMETSAT Data Store dataset (collection) row.

Mirrors a single datasets.<key>: block in one of the per-group catalog/*.yaml files. The friendly dataset key is the parent key in Catalog.datasets and is not stored on the row.

Attributes:

Name	Type	Description
collection_id	str	The real Data Store collection id the search uses, e.g. "EO:EUM:DAT:MSG:HRSEVIRI".
group	DataStoreGroup	The Data Store group (mission family) this dataset belongs to — used by the backend's group= disambiguation.
mission	str	Short mission tag ("msg", "mtg", "metop", "sentinel-3", …). Advisory.
output_kind	OutputKindLiteral	The per-dataset output shape — "raster", "vector", or "tabular". Copied onto the backend instance's OUTPUT_KIND (G1).
format	str	On-disk product format ("native", "netcdf", "grib", "bufr", …). "native" SEVIRI / FCI needs the satpy bridge (PY-2) to read; "netcdf" is pyramids-readable today (G4).
cadence	CadenceLiteral	Native temporal cadence (advisory).
timeliness	TimelinessLiteral | None	Delivery timeliness of the dataset — "nrt", "reprocessed", or "offline" — or None when the distinction does not apply (e.g. geostationary L1.5 imagery). Recorded so a caller can tell a near-real-time stream from a reprocessed archive (the Sentinel-5P collections mix the two).
selectors	list[str]	Informational product-type / band selectors (G2). EUMETSAT delivers whole products, so selectors do not subset the download; they seed catalog metadata and the Data Tailor Chain.
tailor_product_type	str | None	The Data Tailor product-type id for the tailor= server-side subset / reproject / reformat path; None when the collection is not Data-Tailor-eligible.
extent	Extent	Extent — lat / lon coverage.
temporal	TemporalCoverage	TemporalCoverage — start / end dates.

Examples:

• Inspect a curated raster row:

  >>> from earthlens.eumetsat import Catalog
  >>> ds = Catalog().get_dataset("msg-hrseviri")
  >>> ds.collection_id
  'EO:EUM:DAT:MSG:HRSEVIRI'
  >>> ds.output_kind
  'raster'
  

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

class EumetsatDataset(BaseModel):
    """One curated EUMETSAT Data Store dataset (collection) row.

    Mirrors a single `datasets.<key>:` block in one of the per-group
    `catalog/*.yaml` files. The friendly dataset key is the parent key
    in `Catalog.datasets` and is not stored on the row.

    Attributes:
        collection_id: The real Data Store collection id the search
            uses, e.g. `"EO:EUM:DAT:MSG:HRSEVIRI"`.
        group: The Data Store group (mission family) this dataset
            belongs to — used by the backend's `group=` disambiguation.
        mission: Short mission tag (`"msg"`, `"mtg"`, `"metop"`,
            `"sentinel-3"`, …). Advisory.
        output_kind: The per-dataset output shape — `"raster"`,
            `"vector"`, or `"tabular"`. Copied onto the backend
            instance's `OUTPUT_KIND` (`G1`).
        format: On-disk product format (`"native"`, `"netcdf"`,
            `"grib"`, `"bufr"`, …). `"native"` SEVIRI / FCI needs the
            satpy bridge (`PY-2`) to read; `"netcdf"` is pyramids-readable
            today (`G4`).
        cadence: Native temporal cadence (advisory).
        timeliness: Delivery timeliness of the dataset — `"nrt"`,
            `"reprocessed"`, or `"offline"` — or `None` when the
            distinction does not apply (e.g. geostationary L1.5 imagery).
            Recorded so a caller can tell a near-real-time stream from a
            reprocessed archive (the Sentinel-5P collections mix the two).
        selectors: Informational product-type / band selectors (`G2`).
            EUMETSAT delivers whole products, so selectors do not subset
            the download; they seed catalog metadata and the Data Tailor
            `Chain`.
        tailor_product_type: The Data Tailor product-type id for the
            `tailor=` server-side subset / reproject / reformat path;
            `None` when the collection is not Data-Tailor-eligible.
        extent: `Extent` — lat / lon coverage.
        temporal: `TemporalCoverage` — start / end dates.

    Examples:
        - Inspect a curated raster row:
            ```python
            >>> from earthlens.eumetsat import Catalog
            >>> ds = Catalog().get_dataset("msg-hrseviri")
            >>> ds.collection_id
            'EO:EUM:DAT:MSG:HRSEVIRI'
            >>> ds.output_kind
            'raster'

            ```
    """

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

    collection_id: str
    group: DataStoreGroup
    mission: str = ""
    output_kind: OutputKindLiteral = "raster"
    format: str = ""
    cadence: CadenceLiteral = "irregular"
    timeliness: TimelinessLiteral | None = None
    selectors: list[str] = Field(default_factory=list)
    tailor_product_type: str | None = None
    extent: Extent = Field(default_factory=Extent)
    temporal: TemporalCoverage = Field(default_factory=TemporalCoverage)

Extent

Bases: BaseModel

Spatial coverage of an EUMETSAT collection (lat / lon bounds).

Mirrors the extent: block in the YAML. Whole-disk geostationary products (SEVIRI / FCI) use the ~±79° sub-satellite extent; polar / mirror products are global.

Attributes:

Name	Type	Description
lat	list[float]	[lat_min, lat_max] in degrees. Empty means unspecified.
lon	list[float]	[lon_min, lon_max] in degrees. Empty means unspecified.

Examples:

• A geostationary 0° disk extent:

  >>> from earthlens.eumetsat.catalog import Extent
  >>> Extent(lat=[-79, 79], lon=[-79, 79]).lat
  [-79.0, 79.0]
  

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

class Extent(BaseModel):
    """Spatial coverage of an EUMETSAT collection (lat / lon bounds).

    Mirrors the `extent:` block in the YAML. Whole-disk geostationary
    products (SEVIRI / FCI) use the ~±79° sub-satellite extent; polar /
    mirror products are global.

    Attributes:
        lat: `[lat_min, lat_max]` in degrees. Empty means unspecified.
        lon: `[lon_min, lon_max]` in degrees. Empty means unspecified.

    Examples:
        - A geostationary 0° disk extent:
            ```python
            >>> from earthlens.eumetsat.catalog import Extent
            >>> Extent(lat=[-79, 79], lon=[-79, 79]).lat
            [-79.0, 79.0]

            ```
    """

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

    lat: list[float] = Field(default_factory=list)
    lon: list[float] = Field(default_factory=list)

TailorConfig

Bases: BaseModel

Server-side customisation request for the EUMETSAT Data Tailor.

A frozen value object passed to EUMETSAT.download(tailor=...) (or the EarthLens(..., tailor=...) facade kwarg) to route a request through Data Tailor instead of the native whole-product fetch. It carries the output format, the target crs / projection, an optional bbox crop, an optional band filter, and a quicklook flag. The Data Tailor product-type is not set here — it comes from the resolved catalog row's tailor_product_type (G4 / G5).

Attributes:

Name	Type	Description
format	str	Data Tailor output format, e.g. "geotiff", "netcdf4". Maps to Chain.format. Defaults to "geotiff".
crs	str	Target projection / CRS, e.g. "geographic". Maps to Chain.projection. Defaults to "geographic".
bbox	tuple[float, float, float, float] | None	Optional crop as (west, south, east, north) in degrees (the GeoJSON / OGC bbox order). When None, the backend falls back to the request's own spatial extent (lat_lim / lon_lim). Maps to the Data Tailor ROI as an NSWE list.
filter	list[str] | None	Optional list of band / layer names to keep. Maps to Chain.filter (a Filter(bands=...)). None keeps every band.
quicklook	bool	When True, request a quicklook rendering alongside the customised data. Defaults to False.

Examples:

• A reproject + crop + reformat request:

  >>> from earthlens.eumetsat import TailorConfig
  >>> cfg = TailorConfig(format="geotiff", crs="geographic", bbox=(4, 48, 8, 52))
  >>> cfg.nswe
  [52.0, 48.0, 4.0, 8.0]
  

• No bbox falls back to the request extent (ROI derived later):

  >>> from earthlens.eumetsat import TailorConfig
  >>> TailorConfig().nswe is None
  True
  

Source code in libs/providers/imagery/src/earthlens/eumetsat/tailor.py

class TailorConfig(BaseModel):
    """Server-side customisation request for the EUMETSAT Data Tailor.

    A frozen value object passed to `EUMETSAT.download(tailor=...)` (or the
    `EarthLens(..., tailor=...)` facade kwarg) to route a request through
    Data Tailor instead of the native whole-product fetch. It carries the
    output `format`, the target `crs` / projection, an optional `bbox`
    crop, an optional band `filter`, and a `quicklook` flag. The Data
    Tailor **product-type** is not set here — it comes from the resolved
    catalog row's `tailor_product_type` (`G4` / `G5`).

    Attributes:
        format: Data Tailor output format, e.g. `"geotiff"`, `"netcdf4"`.
            Maps to `Chain.format`. Defaults to `"geotiff"`.
        crs: Target projection / CRS, e.g. `"geographic"`. Maps to
            `Chain.projection`. Defaults to `"geographic"`.
        bbox: Optional crop as `(west, south, east, north)` in degrees
            (the GeoJSON / OGC bbox order). When `None`, the backend falls
            back to the request's own spatial extent (`lat_lim` /
            `lon_lim`). Maps to the Data Tailor ROI as an `NSWE` list.
        filter: Optional list of band / layer names to keep. Maps to
            `Chain.filter` (a `Filter(bands=...)`). `None` keeps every
            band.
        quicklook: When `True`, request a quicklook rendering alongside
            the customised data. Defaults to `False`.

    Examples:
        - A reproject + crop + reformat request:
            ```python
            >>> from earthlens.eumetsat import TailorConfig
            >>> cfg = TailorConfig(format="geotiff", crs="geographic", bbox=(4, 48, 8, 52))
            >>> cfg.nswe
            [52.0, 48.0, 4.0, 8.0]

            ```
        - No bbox falls back to the request extent (ROI derived later):
            ```python
            >>> from earthlens.eumetsat import TailorConfig
            >>> TailorConfig().nswe is None
            True

            ```
    """

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

    format: str = DEFAULT_FORMAT
    crs: str = DEFAULT_CRS
    bbox: tuple[float, float, float, float] | None = None
    filter: list[str] | None = None
    quicklook: bool = False

    @field_validator("format", "crs")
    @classmethod
    def _non_empty(cls, value: str) -> str:
        """Reject an empty `format` / `crs` string.

        Args:
            value: The candidate `format` or `crs` value.

        Returns:
            The stripped value.

        Raises:
            ValueError: When the value is blank.
        """
        stripped = value.strip()
        if not stripped:
            raise ValueError("must be a non-empty string")
        return stripped

    @field_validator("bbox")
    @classmethod
    def _valid_bbox(
        cls, value: tuple[float, float, float, float] | None
    ) -> tuple[float, float, float, float] | None:
        """Validate a `bbox` is a well-ordered, in-range `(w, s, e, n)`.

        Args:
            value: The candidate `(west, south, east, north)` box, or
                `None`.

        Returns:
            The validated box, or `None`.

        Raises:
            ValueError: When the box is inverted (`west > east` or
                `south > north`) or out of the WGS84 range.
        """
        if value is None:
            return None
        west, south, east, north = value
        if west > east:
            raise ValueError(f"bbox west ({west}) must be <= east ({east})")
        if south > north:
            raise ValueError(f"bbox south ({south}) must be <= north ({north})")
        if not (-180.0 <= west <= 180.0 and -180.0 <= east <= 180.0):
            raise ValueError(f"bbox longitudes must be in [-180, 180]: {value}")
        if not (-90.0 <= south <= 90.0 and -90.0 <= north <= 90.0):
            raise ValueError(f"bbox latitudes must be in [-90, 90]: {value}")
        return value

    @property
    def nswe(self) -> list[float] | None:
        """Return the Data Tailor ROI as an `[N, S, W, E]` list, or `None`.

        Data Tailor's `RegionOfInterest` takes an `NSWE` list of four
        numbers `[north, south, west, east]` (`A1`). This converts the
        stored `bbox` (`(west, south, east, north)`) into that order.
        Returns `None` when no `bbox` was set, so the backend can fall
        back to the request's spatial extent.

        Returns:
            list[float] | None: `[north, south, west, east]`, or `None`
                when `bbox` is unset.
        """
        if self.bbox is None:
            return None
        west, south, east, north = self.bbox
        return [north, south, west, east]

    @staticmethod
    def nswe_from_extent(
        north: float, south: float, west: float, east: float
    ) -> list[float]:
        """Build a Data Tailor `NSWE` list from spatial-extent bounds.

        The backend calls this with `self.space` bounds when a
        `TailorConfig` carries no explicit `bbox`, so the request's
        `lat_lim` / `lon_lim` become the ROI.

        Args:
            north: Northern latitude bound in degrees.
            south: Southern latitude bound in degrees.
            west: Western longitude bound in degrees.
            east: Eastern longitude bound in degrees.

        Returns:
            list[float]: `[north, south, west, east]`.
        """
        return [north, south, west, east]

nswe property

Return the Data Tailor ROI as an [N, S, W, E] list, or None.

Data Tailor's RegionOfInterest takes an NSWE list of four numbers [north, south, west, east] (A1). This converts the stored bbox ((west, south, east, north)) into that order. Returns None when no bbox was set, so the backend can fall back to the request's spatial extent.

Returns:

Type	Description
list[float] | None	list[float] | None: [north, south, west, east], or None when bbox is unset.

nswe_from_extent(north, south, west, east) staticmethod

Build a Data Tailor NSWE list from spatial-extent bounds.

The backend calls this with self.space bounds when a TailorConfig carries no explicit bbox, so the request's lat_lim / lon_lim become the ROI.

Parameters:

Name	Type	Description	Default
north	float	Northern latitude bound in degrees.	required
south	float	Southern latitude bound in degrees.	required
west	float	Western longitude bound in degrees.	required
east	float	Eastern longitude bound in degrees.	required

Returns:

Type	Description
list[float]	list[float]: [north, south, west, east].

Source code in libs/providers/imagery/src/earthlens/eumetsat/tailor.py

@staticmethod
def nswe_from_extent(
    north: float, south: float, west: float, east: float
) -> list[float]:
    """Build a Data Tailor `NSWE` list from spatial-extent bounds.

    The backend calls this with `self.space` bounds when a
    `TailorConfig` carries no explicit `bbox`, so the request's
    `lat_lim` / `lon_lim` become the ROI.

    Args:
        north: Northern latitude bound in degrees.
        south: Southern latitude bound in degrees.
        west: Western longitude bound in degrees.
        east: Eastern longitude bound in degrees.

    Returns:
        list[float]: `[north, south, west, east]`.
    """
    return [north, south, west, east]

TemporalCoverage

Bases: BaseModel

Temporal coverage of an EUMETSAT collection (start + optional end).

Mirrors the temporal: block in the YAML. end: null (or a missing end) means the collection is ongoing / rolling.

Attributes:

Name	Type	Description
start	str | None	First date with data, as a YYYY-MM-DD string. May be None when the start date is not pinned in the YAML.
end	str | None	Last date with data, or None for an ongoing collection.

Examples:

• An ongoing collection:

  >>> from earthlens.eumetsat.catalog import TemporalCoverage
  >>> tc = TemporalCoverage(start="2004-01-19", end=None)
  >>> tc.start, tc.end
  ('2004-01-19', None)
  

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

class TemporalCoverage(BaseModel):
    """Temporal coverage of an EUMETSAT collection (start + optional end).

    Mirrors the `temporal:` block in the YAML. `end: null` (or a missing
    `end`) means the collection is ongoing / rolling.

    Attributes:
        start: First date with data, as a `YYYY-MM-DD` string. May be
            `None` when the start date is not pinned in the YAML.
        end: Last date with data, or `None` for an ongoing collection.

    Examples:
        - An ongoing collection:
            ```python
            >>> from earthlens.eumetsat.catalog import TemporalCoverage
            >>> tc = TemporalCoverage(start="2004-01-19", end=None)
            >>> tc.start, tc.end
            ('2004-01-19', None)

            ```
    """

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

    start: str | None = None
    end: str | None = None

    @field_validator("start", "end", mode="before")
    @classmethod
    def _coerce_date(cls, value: Any) -> Any:
        """Accept a `datetime.date` (PyYAML's native parse) as ISO string.

        Args:
            value: Raw YAML value — a string, a `datetime.date`, or
                `None`.

        Returns:
            An ISO-format string (`"YYYY-MM-DD"`) or `None`.
        """
        if isinstance(value, _dt.date):
            return value.isoformat()
        return value

clear_catalog_cache()

Empty the module-level catalog parse cache.

Useful in tests that rewrite the catalog on disk and want to force a re-parse. Production callers do not need this — the cache keys include every contributing file's st_mtime_ns, so any real file mutation invalidates the entry on its own.

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

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

    Useful in tests that rewrite the catalog on disk and want to force a
    re-parse. Production callers do not need this — the cache keys
    include every contributing file's `st_mtime_ns`, so any real file
    mutation invalidates the entry on its own.
    """
    _CATALOG_CACHE.clear()

earthlens.eumetsat.tailor

Typed request shape for the EUMETSAT Data Tailor (server-side customisation).

Hosts TailorConfig, the frozen pydantic value object that turns a download(tailor=...) call into a EUMETSAT Data Tailor Chain. Data Tailor is EUMETSAT's server-side subset / reproject / reformat service (the analogue of NASA Harmony) — a spatial operation, distinct from the temporal earthlens.aggregate.AggregationConfig. The two knobs compose: tailor= reshapes each product server-side, then an optional aggregate= reduces the result client-side.

TailorConfig is deliberately SDK-free — it holds only plain values and knows how to derive the Data Tailor region-of-interest (NSWE) from its bbox. The backend builds the actual eumdac.tailor_models.Chain (a lazy eumdac import) from a TailorConfig, the resolved catalog row's tailor_product_type, and the request's spatial extent.

TailorConfig

Bases: BaseModel

Server-side customisation request for the EUMETSAT Data Tailor.

A frozen value object passed to EUMETSAT.download(tailor=...) (or the EarthLens(..., tailor=...) facade kwarg) to route a request through Data Tailor instead of the native whole-product fetch. It carries the output format, the target crs / projection, an optional bbox crop, an optional band filter, and a quicklook flag. The Data Tailor product-type is not set here — it comes from the resolved catalog row's tailor_product_type (G4 / G5).

Attributes:

Name	Type	Description
format	str	Data Tailor output format, e.g. "geotiff", "netcdf4". Maps to Chain.format. Defaults to "geotiff".
crs	str	Target projection / CRS, e.g. "geographic". Maps to Chain.projection. Defaults to "geographic".
bbox	tuple[float, float, float, float] | None	Optional crop as (west, south, east, north) in degrees (the GeoJSON / OGC bbox order). When None, the backend falls back to the request's own spatial extent (lat_lim / lon_lim). Maps to the Data Tailor ROI as an NSWE list.
filter	list[str] | None	Optional list of band / layer names to keep. Maps to Chain.filter (a Filter(bands=...)). None keeps every band.
quicklook	bool	When True, request a quicklook rendering alongside the customised data. Defaults to False.

Examples:

• A reproject + crop + reformat request:

  >>> from earthlens.eumetsat import TailorConfig
  >>> cfg = TailorConfig(format="geotiff", crs="geographic", bbox=(4, 48, 8, 52))
  >>> cfg.nswe
  [52.0, 48.0, 4.0, 8.0]
  

• No bbox falls back to the request extent (ROI derived later):

  >>> from earthlens.eumetsat import TailorConfig
  >>> TailorConfig().nswe is None
  True
  

Source code in libs/providers/imagery/src/earthlens/eumetsat/tailor.py

class TailorConfig(BaseModel):
    """Server-side customisation request for the EUMETSAT Data Tailor.

    A frozen value object passed to `EUMETSAT.download(tailor=...)` (or the
    `EarthLens(..., tailor=...)` facade kwarg) to route a request through
    Data Tailor instead of the native whole-product fetch. It carries the
    output `format`, the target `crs` / projection, an optional `bbox`
    crop, an optional band `filter`, and a `quicklook` flag. The Data
    Tailor **product-type** is not set here — it comes from the resolved
    catalog row's `tailor_product_type` (`G4` / `G5`).

    Attributes:
        format: Data Tailor output format, e.g. `"geotiff"`, `"netcdf4"`.
            Maps to `Chain.format`. Defaults to `"geotiff"`.
        crs: Target projection / CRS, e.g. `"geographic"`. Maps to
            `Chain.projection`. Defaults to `"geographic"`.
        bbox: Optional crop as `(west, south, east, north)` in degrees
            (the GeoJSON / OGC bbox order). When `None`, the backend falls
            back to the request's own spatial extent (`lat_lim` /
            `lon_lim`). Maps to the Data Tailor ROI as an `NSWE` list.
        filter: Optional list of band / layer names to keep. Maps to
            `Chain.filter` (a `Filter(bands=...)`). `None` keeps every
            band.
        quicklook: When `True`, request a quicklook rendering alongside
            the customised data. Defaults to `False`.

    Examples:
        - A reproject + crop + reformat request:
            ```python
            >>> from earthlens.eumetsat import TailorConfig
            >>> cfg = TailorConfig(format="geotiff", crs="geographic", bbox=(4, 48, 8, 52))
            >>> cfg.nswe
            [52.0, 48.0, 4.0, 8.0]

            ```
        - No bbox falls back to the request extent (ROI derived later):
            ```python
            >>> from earthlens.eumetsat import TailorConfig
            >>> TailorConfig().nswe is None
            True

            ```
    """

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

    format: str = DEFAULT_FORMAT
    crs: str = DEFAULT_CRS
    bbox: tuple[float, float, float, float] | None = None
    filter: list[str] | None = None
    quicklook: bool = False

    @field_validator("format", "crs")
    @classmethod
    def _non_empty(cls, value: str) -> str:
        """Reject an empty `format` / `crs` string.

        Args:
            value: The candidate `format` or `crs` value.

        Returns:
            The stripped value.

        Raises:
            ValueError: When the value is blank.
        """
        stripped = value.strip()
        if not stripped:
            raise ValueError("must be a non-empty string")
        return stripped

    @field_validator("bbox")
    @classmethod
    def _valid_bbox(
        cls, value: tuple[float, float, float, float] | None
    ) -> tuple[float, float, float, float] | None:
        """Validate a `bbox` is a well-ordered, in-range `(w, s, e, n)`.

        Args:
            value: The candidate `(west, south, east, north)` box, or
                `None`.

        Returns:
            The validated box, or `None`.

        Raises:
            ValueError: When the box is inverted (`west > east` or
                `south > north`) or out of the WGS84 range.
        """
        if value is None:
            return None
        west, south, east, north = value
        if west > east:
            raise ValueError(f"bbox west ({west}) must be <= east ({east})")
        if south > north:
            raise ValueError(f"bbox south ({south}) must be <= north ({north})")
        if not (-180.0 <= west <= 180.0 and -180.0 <= east <= 180.0):
            raise ValueError(f"bbox longitudes must be in [-180, 180]: {value}")
        if not (-90.0 <= south <= 90.0 and -90.0 <= north <= 90.0):
            raise ValueError(f"bbox latitudes must be in [-90, 90]: {value}")
        return value

    @property
    def nswe(self) -> list[float] | None:
        """Return the Data Tailor ROI as an `[N, S, W, E]` list, or `None`.

        Data Tailor's `RegionOfInterest` takes an `NSWE` list of four
        numbers `[north, south, west, east]` (`A1`). This converts the
        stored `bbox` (`(west, south, east, north)`) into that order.
        Returns `None` when no `bbox` was set, so the backend can fall
        back to the request's spatial extent.

        Returns:
            list[float] | None: `[north, south, west, east]`, or `None`
                when `bbox` is unset.
        """
        if self.bbox is None:
            return None
        west, south, east, north = self.bbox
        return [north, south, west, east]

    @staticmethod
    def nswe_from_extent(
        north: float, south: float, west: float, east: float
    ) -> list[float]:
        """Build a Data Tailor `NSWE` list from spatial-extent bounds.

        The backend calls this with `self.space` bounds when a
        `TailorConfig` carries no explicit `bbox`, so the request's
        `lat_lim` / `lon_lim` become the ROI.

        Args:
            north: Northern latitude bound in degrees.
            south: Southern latitude bound in degrees.
            west: Western longitude bound in degrees.
            east: Eastern longitude bound in degrees.

        Returns:
            list[float]: `[north, south, west, east]`.
        """
        return [north, south, west, east]

nswe property

Return the Data Tailor ROI as an [N, S, W, E] list, or None.

Data Tailor's RegionOfInterest takes an NSWE list of four numbers [north, south, west, east] (A1). This converts the stored bbox ((west, south, east, north)) into that order. Returns None when no bbox was set, so the backend can fall back to the request's spatial extent.

Returns:

Type	Description
list[float] | None	list[float] | None: [north, south, west, east], or None when bbox is unset.

nswe_from_extent(north, south, west, east) staticmethod

Build a Data Tailor NSWE list from spatial-extent bounds.

The backend calls this with self.space bounds when a TailorConfig carries no explicit bbox, so the request's lat_lim / lon_lim become the ROI.

Parameters:

Name	Type	Description	Default
north	float	Northern latitude bound in degrees.	required
south	float	Southern latitude bound in degrees.	required
west	float	Western longitude bound in degrees.	required
east	float	Eastern longitude bound in degrees.	required

Returns:

Type	Description
list[float]	list[float]: [north, south, west, east].

Source code in libs/providers/imagery/src/earthlens/eumetsat/tailor.py

@staticmethod
def nswe_from_extent(
    north: float, south: float, west: float, east: float
) -> list[float]:
    """Build a Data Tailor `NSWE` list from spatial-extent bounds.

    The backend calls this with `self.space` bounds when a
    `TailorConfig` carries no explicit `bbox`, so the request's
    `lat_lim` / `lon_lim` become the ROI.

    Args:
        north: Northern latitude bound in degrees.
        south: Southern latitude bound in degrees.
        west: Western longitude bound in degrees.
        east: Eastern longitude bound in degrees.

    Returns:
        list[float]: `[north, south, west, east]`.
    """
    return [north, south, west, east]

earthlens.eumetsat.backend

Backend that fetches EUMETSAT Data Store products via eumdac.

EUMETSAT(AbstractDataSource) accepts the same constructor surface as the other earthlens backends — start, end, variables, lat_lim, lon_lim, temporal_resolution, path — plus a few backend-specific kwargs for the OAuth2 consumer key / secret and group disambiguation. Each (dataset_key, [selector, ...]) pair in the variables mapping names one curated Data Store collection to search (bbox + window) and fetch.

This backend's OUTPUT_KIND is per-instance, not fixed (G1). EUMETSAT spans gridded imagery (SEVIRI / FCI L1.5/L1c, OLCI / SLSTR grids) and swath / sounding products (S5P TROPOMI L2, ASCAT, IASI). The class default is "raster"; __init__ resolves the requested collection row(s) and copies the row's output_kind onto self.OUTPUT_KIND. The earthlens.earthlens.EarthLens facade reads that per-instance value at download() time to gate aggregate=.

A single request may name several collections, but they must all share one output_kind — a mixed raster+vector request is rejected at construction.

By default the backend fetches whole native products to disk (G4): eumdac's Product.open() stream copied to a file. The selectors in variables are informational for a whole-product fetch (G2), and bbox is a search filter (which products intersect), not a pixel crop.

Server-side subset / reproject / reformat is EUMETSAT's Data Tailor service, reached with download(tailor=TailorConfig(...)) (H4). That routes each matching product through Data Tailor (submit → poll → stream → delete) and returns the customised GeoTIFF / NetCDF paths instead of the native product; every customisation is deleted afterwards for quota hygiene (G7). Only catalog rows carrying a tailor_product_type are Data-Tailor-eligible (G5).

tailor= is spatial; the temporal reducer is a separate aggregate= knob (G1). download(aggregate=...) still raises NotImplementedError — EUMETSAT native NetCDF products can be reduced client-side with pyramids after download, and the two knobs compose (tailor server-side, then reduce client-side).

EUMETSAT

Bases: AbstractDataSource

EUMETSAT Data Store backend (per-collection output kind).

Wraps eumdac so a user can search a curated EUMETSAT collection by bbox + window and fetch its native products through the same download() shape every other earthlens backend uses. One OAuth2 consumer key / secret authenticates across every collection.

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	Class default "raster", overridden per instance in __init__ from the resolved collection row's output_kind (G1). The facade reads this instance value to gate aggregate=.

Source code in libs/providers/imagery/src/earthlens/eumetsat/backend.py

class EUMETSAT(AbstractDataSource):
    """EUMETSAT Data Store backend (per-collection output kind).

    Wraps `eumdac` so a user can search a curated EUMETSAT collection by
    bbox + window and fetch its native products through the same
    `download()` shape every other earthlens backend uses. One OAuth2
    consumer key / secret authenticates across every collection.

    Attributes:
        OUTPUT_KIND: Class default `"raster"`, **overridden per instance**
            in `__init__` from the resolved collection row's
            `output_kind` (`G1`). The facade reads this instance value to
            gate `aggregate=`.
    """

    OUTPUT_KIND: OutputKind = "raster"

    AGGREGATE_REFUSAL_REASON = (
        "the temporal reducer is not wired for this backend. Download the "
        "products (optionally with tailor= for a server-side subset / "
        "reproject) and reduce the NetCDF ones client-side with pyramids"
    )

    def __init__(
        self,
        start: str,
        end: str,
        variables: dict[str, list[str]],
        lat_lim: list[float],
        lon_lim: list[float],
        temporal_resolution: str = "daily",
        path: Path | str | None = None,
        fmt: str = "%Y-%m-%d",
        group: DataStoreGroup | str | None = None,
        consumer_key: str | None = None,
        consumer_secret: str | None = None,
        credentials_file: Path | str | None = None,
    ):
        """Initialise an EUMETSAT backend instance.

        Resolves every requested dataset key against the catalog
        **before** calling the parent constructor, so the per-instance
        `OUTPUT_KIND` is set from the resolved row(s). The parent
        `__init__` runs `_initialize` first (token mint), so the
        resolution cannot live there.

        Args:
            start: Inclusive start date as a string (parsed with `fmt`).
            end: Inclusive end date as a string.
            variables: Mapping from curated dataset key to a list of
                selectors, e.g. `{"msg-hrseviri": ["HRSEVIRI"]}`.
                Selectors are informational for the whole-product fetch
                (`G2`).
            lat_lim: `[lat_min, lat_max]` in degrees.
            lon_lim: `[lon_min, lon_max]` in degrees.
            temporal_resolution: Advisory cadence label. Defaults to
                `"daily"`.
            path: Output directory. Created by the parent class if it
                does not exist.
            fmt: `strptime` format for `start` / `end`. Defaults to
                `"%Y-%m-%d"`.
            group: Optional `DataStoreGroup` (or its string value) used to
                assert which Data Store group the requested collection(s)
                belong to (`G2`).
            consumer_key: EUMETSAT consumer key. Falls back to
                `EUMETSAT_CONSUMER_KEY`, then `~/.eumdac/credentials`.
            consumer_secret: EUMETSAT consumer secret. Falls back to
                `EUMETSAT_CONSUMER_SECRET`, then the credentials file.
            credentials_file: Optional explicit path to a `key,secret`
                credentials file.

        Raises:
            ValueError: When `variables` is empty, a dataset key is
                unknown, or the requested collections do not all share
                one `output_kind`.
        """
        self._group = group
        self._consumer_key = consumer_key
        self._consumer_secret = consumer_secret
        self._credentials_file = (
            Path(credentials_file) if credentials_file is not None else None
        )
        self._auth: EumetsatAuth | None = None
        self._show_progress = True

        self._catalog = Catalog()
        self._datasets: list[EumetsatDataset] = self._resolve_datasets(variables)
        self.OUTPUT_KIND = self._unify_output_kind(self._datasets)

        super().__init__(
            start=start,
            end=end,
            variables=variables,
            temporal_resolution=temporal_resolution,
            lat_lim=lat_lim,
            lon_lim=lon_lim,
            fmt=fmt,
            path=path,
        )

    def _resolve_datasets(
        self, variables: dict[str, list[str]]
    ) -> list[EumetsatDataset]:
        """Resolve every requested dataset key to a catalog row.

        Args:
            variables: The `{dataset_key: [selector, ...]}` request.

        Returns:
            list[EumetsatDataset]: One row per key, in request order.

        Raises:
            ValueError: When `variables` is empty or a key is unknown
                (the catalog's did-you-mean is surfaced in the message).
        """
        if not variables:
            raise ValueError(
                "EUMETSAT requires a non-empty `variables` mapping of "
                "{dataset_key: [selector, ...]}."
            )
        return [self._catalog.resolve(key, group=self._group) for key in variables]

    @staticmethod
    def _unify_output_kind(datasets: list[EumetsatDataset]) -> OutputKind:
        """Return the single `output_kind` shared by every requested row.

        A backend instance carries exactly one `OUTPUT_KIND`, so a
        request mixing (say) a raster and a vector dataset is ambiguous
        and rejected here.

        Args:
            datasets: The resolved dataset rows.

        Returns:
            OutputKind: The shared `output_kind`.

        Raises:
            ValueError: When the rows do not all share one `output_kind`.
        """
        kinds = {ds.output_kind for ds in datasets}
        if len(kinds) > 1:
            detail = ", ".join(
                f"{ds.collection_id}={ds.output_kind}" for ds in datasets
            )
            raise ValueError(
                "all datasets in one EUMETSAT request must share one "
                f"output_kind; got mixed kinds ({detail}). Split the "
                "request into one call per output kind."
            )
        return kinds.pop()

    def _initialize(self):
        """Build the `EumetsatAuth`; defer token minting.

        Returns `None` — `eumdac` keeps the token on the `EumetsatAuth`
        instance, so the parent class binds no opaque `self.client`. The
        token minting (`EumetsatAuth.configure`, which contacts the auth
        server) is deferred out of construction: it runs on the first
        :meth:`_search` (the `eumdac` collection search authenticates via
        the idempotent `configure()`), so constructing the backend never
        authenticates — but note that a dry-run `search()` does, since the
        `eumdac` data store needs a token.
        """
        creds = EumetsatCredentials(
            consumer_key=self._consumer_key,
            consumer_secret=(
                SecretStr(self._consumer_secret)
                if self._consumer_secret is not None
                else None
            ),
            credentials_file=self._credentials_file,
        )
        self._auth = EumetsatAuth(creds)
        return None

    def _check_input_dates(
        self,
        start: str,
        end: str,
        temporal_resolution: str,
        fmt: str,
    ) -> TemporalExtent:
        """Parse the date range into a `TemporalExtent`.

        Args:
            start: Inclusive start date as a string.
            end: Inclusive end date as a string.
            temporal_resolution: Advisory cadence label; mapped to a
                pandas frequency for the `dates` index when known.
            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.

        Raises:
            ValueError: If `temporal_resolution` is not one of the cadences
                `earthlens.base.CADENCE_ALIASES` accepts.

        Returns:
            TemporalExtent: Frozen model with parsed bounds.

        Raises:
            ValueError: If `start` parses to a date later than `end`.
        """
        self._end_is_date_only = end_is_date_only(end)
        return self._cadence_extent(
            start,
            end,
            fmt=fmt,
            cadence=temporal_resolution,
            accepted=CADENCE_ALIASES,
        )

    def _search(self) -> list[RemoteProduct]:
        """Query the Data Store for products of every requested collection.

        One `Collection.search(bbox=, dtstart=, dtend=)` per resolved
        collection row, scoped to the request bbox and time window. The
        bbox is the `eumdac` `W,S,E,N` comma-string the OpenSearch
        endpoint expects.

        How `end` is interpreted depends on whether it carries a time of
        day. A **date-only** `end` parses to midnight, which would collapse
        a same-day request (`start == end`) to a zero-width instant, so it
        is read as *inclusive of its whole calendar day* and `dtend` is
        widened to `23:59:59.999999`. An `end` that **names a time** means
        that instant and is passed through unchanged — widening it would
        pull every later product of the day, which for a 10-minute
        full-disk cadence is tens of gigabytes the caller never asked for.

        Each returned `eumdac` product becomes one `RemoteProduct` whose
        `metadata` carries the raw product handle and its collection row,
        so `_fetch` can stream without re-querying.

        Returns:
            list[RemoteProduct]: One product per matching Data Store
                product, across every requested collection. An empty list
                (no products in the window) short-circuits the fetch.

        Raises:
            ImportError: When the `[eumetsat]` extra (`eumdac`) is not
                installed — surfaced by `EumetsatAuth.datastore()`.
        """
        assert self._auth is not None  # set by _initialize
        self._auth.configure()
        store = self._auth.datastore()
        # A `SpatialExtent` constrains longitude to a single `[-180, 180]`
        # range with `west <= east`, so it cannot represent an
        # antimeridian-crossing box — a single search bbox always suffices.
        bbox = eumdac_bbox(
            self.space.west, self.space.south, self.space.east, self.space.north
        )
        dtstart = self.time.start_date
        dtend = expand_bare_date_end(
            self.time.end_date, date_only=self._end_is_date_only
        )
        products: list[RemoteProduct] = []
        for ds in self._datasets:
            collection = store.get_collection(ds.collection_id)
            for product in collection.search(
                bbox=bbox,
                dtstart=dtstart,
                dtend=dtend,
            ):
                products.append(
                    RemoteProduct(
                        id=str(product),
                        metadata={"product": product, "dataset": ds},
                    )
                )
        return products

    def _fetch(self, products: list[RemoteProduct]) -> list[Path]:
        """Stream every product `_search` returned to a local file.

        Each `eumdac` product is opened (`Product.open()` — a streaming
        context manager) and copied to `self.root_dir / <id>`, where the
        product id is reduced to a safe basename (`safe_product_filename`)
        so a server-supplied id with a path separator cannot write outside
        the output directory.

        Args:
            products: The products from `_search`.

        Returns:
            list[Path]: Local paths of every fetched product, in search
                order.
        """
        out_paths: list[Path] = []
        for rp in products:
            product = rp.metadata["product"]
            target = self.root_dir / safe_product_filename(str(product))
            with product.open() as src, open(target, "wb") as dst:
                shutil.copyfileobj(src, dst)
            out_paths.append(target)
        return out_paths

    def download(
        self,
        progress_bar: bool = True,
        tailor: TailorConfig | None = None,
    ) -> list[Path]:
        """Search the Data Store and return product paths (native or tailored).

        With no `tailor=`, composes `_search` and `_fetch` to pull every
        product matching the request bbox + window to `self.root_dir` as
        whole native products (unchanged behaviour).

        With `tailor=TailorConfig(...)`, each matching product is routed
        through EUMETSAT **Data Tailor** (server-side subset / reproject /
        reformat, `H4`): submit a customisation, poll it to `DONE`, stream
        the customised output(s) to `self.root_dir`, and delete the
        customisation (`G7`). The returned paths are then the customised
        GeoTIFF / NetCDF files, not the native products (`G6`).

        `aggregate=` is the **temporal** reducer (`G1`) and is a separate
        operation from the spatial `tailor=`; it is not implemented for
        EUMETSAT, so a non-`None` `aggregate` raises `NotImplementedError`.
        The two knobs compose — tailor server-side here, then reduce the
        result client-side with pyramids.

        The tailor branch is **fail-fast per batch**: if one product's
        customisation fails, the error propagates and paths already streamed
        for earlier products are not returned (their files remain on disk).
        This mirrors the native fetch; keep batches small when a partial
        result would be costly to recompute.

        Args:
            progress_bar: Reserved for parity with the other backends;
                `eumdac`'s streaming download has no built-in bar.
            tailor: Optional `TailorConfig` routing the request through
                Data Tailor. `None` (the default) keeps the native fetch.

        Returns:
            list[Path]: The native product paths, or — when `tailor=` is
                given — the customised output paths.

        Raises:
            ValueError: When `tailor=` names a dataset that is not
                Data-Tailor-eligible (`G5`).
        """
        self._show_progress = progress_bar
        if tailor is not None:
            return self._tailor(tailor)
        return self._api_via_search_fetch()

    def _tailor(self, tailor: TailorConfig) -> list[Path]:
        """Run the Data Tailor branch for every matching product (`G2`).

        Rejects a non-eligible request up front (`G5`), then searches the
        Data Store and customises each product in turn, returning the
        flattened list of customised output paths.

        Args:
            tailor: The `TailorConfig` describing the customisation.

        Returns:
            list[Path]: Every customised output path, in search order.

        Raises:
            ValueError: When any requested dataset lacks a
                `tailor_product_type` (not Data-Tailor-eligible).
        """
        ineligible = [ds for ds in self._datasets if ds.tailor_product_type is None]
        if ineligible:
            names = ", ".join(ds.collection_id for ds in ineligible)
            raise ValueError(
                f"{names} not Data-Tailor-eligible; download native (no "
                "tailor=) and reduce client-side with pyramids."
            )
        assert self._auth is not None  # set by _initialize
        self._auth.configure()
        datatailor = self._auth.datatailor()
        products = self._search()
        out_paths: list[Path] = []
        used_dirs: set[str] = set()
        for rp in products:
            out_paths.extend(self._tailor_one(rp, tailor, datatailor, used_dirs))
        return out_paths

    @staticmethod
    def _dedupe_name(name: str, used: set[str]) -> str:
        """Return `name`, suffixed if needed so it is unique within `used`.

        Guarantees a distinct on-disk name even when two products share an
        id (`L3`) or two customisation outputs sanitise to one basename
        (`L2`). Adds the chosen name to `used`.

        Args:
            name: The candidate (already path-safe) name.
            used: The set of names already taken; mutated in place.

        Returns:
            str: A name not already in `used`.
        """
        candidate = name
        counter = 1
        while candidate in used:
            candidate = f"{name}_{counter}"
            counter += 1
        used.add(candidate)
        return candidate

    def _tailor_one(
        self,
        product: RemoteProduct,
        tailor: TailorConfig,
        datatailor,
        used_dirs: set[str],
    ) -> list[Path]:
        """Customise one product via Data Tailor; always clean up (`G7`).

        Builds the `eumdac` `Chain` from `tailor`, the product's catalog
        row (`tailor_product_type`), and the request ROI (`tailor.bbox`
        else `self.space`), submits it, polls to a terminal state, streams
        every output to `self.root_dir`, and deletes the customisation in
        a `finally` — even on failure — so quota is always freed.

        Args:
            product: One `RemoteProduct` from `_search` (its `metadata`
                carries the raw `eumdac` product handle and catalog row).
            tailor: The customisation request.
            datatailor: The live `eumdac.DataTailor` client.
            used_dirs: Shared per-batch set of subdirectory names already
                taken, mutated here to keep each product's output directory
                unique across the request (`L3`).

        Returns:
            list[Path]: The customised output paths for this product.

        Raises:
            ValueError: When the product's dataset is not eligible (`G5`).
            RuntimeError: When the customisation ends `FAILED` / `KILLED`
                (with the server log), or a transient submit keeps failing.
            TimeoutError: When polling exceeds `TAILOR_POLL_TIMEOUT_S`.
        """
        import eumdac  # lazy — the [eumetsat] extra

        dataset: EumetsatDataset = product.metadata["dataset"]
        if dataset.tailor_product_type is None:
            raise ValueError(
                f"{dataset.collection_id!r} is not Data-Tailor-eligible; "
                "download native (no tailor=) and reduce client-side."
            )
        nswe = tailor.nswe or TailorConfig.nswe_from_extent(
            self.space.north, self.space.south, self.space.west, self.space.east
        )
        chain = eumdac.tailor_models.Chain(
            product=dataset.tailor_product_type,
            format=tailor.format,
            projection=tailor.crs,
            # eumdac types NSWE as Optional[str], but the Data Tailor ROI takes
            # a north/south/west/east list (see TailorConfig.nswe).
            roi=eumdac.tailor_models.RegionOfInterest(NSWE=nswe),  # type: ignore[arg-type]
            filter=(
                eumdac.tailor_models.Filter(bands=list(tailor.filter))
                if tailor.filter
                else None
            ),
            # eumdac types quicklook as a Quicklook/dict; the API accepts a truthy flag.
            quicklook=tailor.quicklook or None,  # type: ignore[arg-type]
        )
        product_handle = product.metadata["product"]
        # Choose the per-product output subdir *before* submitting, so nothing
        # between the submit and the `try/finally` can raise and orphan the
        # customisation (quota hygiene, G7). Namespacing avoids cross-granule
        # basename collisions (H1); the name is de-duped against this batch and
        # against any pre-existing native file of the same basename (L3).
        subdir = self._dedupe_name(
            safe_product_filename(str(product_handle)), used_dirs
        )
        while (self.root_dir / subdir).exists() and not (
            self.root_dir / subdir
        ).is_dir():
            subdir = self._dedupe_name(subdir, used_dirs)
        product_dir = self.root_dir / subdir
        cust = self._submit_customisation(datatailor, product_handle, chain)
        try:
            status = self._poll_customisation(cust)
            if status != "DONE":
                raise RuntimeError(
                    f"Data Tailor customisation {cust} ended {status}: "
                    f"{self._logfile_tail(cust)}"
                )
            product_dir.mkdir(parents=True, exist_ok=True)
            written: list[Path] = []
            used_names: set[str] = set()
            for name in cust.outputs:
                # De-dupe within a customisation so two outputs sharing a
                # basename do not overwrite each other (L2).
                out_name = self._dedupe_name(
                    safe_product_filename(str(name)), used_names
                )
                target = product_dir / out_name
                with cust.stream_output(name) as src, open(target, "wb") as fh:
                    shutil.copyfileobj(src, fh)
                written.append(target)
            return written
        finally:
            self._safe_delete(cust)  # ALWAYS free quota — even on failure (G7)

    @staticmethod
    def _safe_delete(cust) -> None:
        """Delete a customisation, never letting cleanup mask the real error.

        Called from the `_tailor_one` `finally` (`G7`). A `delete()` that
        itself raises must not replace the original `RuntimeError` /
        `TimeoutError`, nor abort the remaining products in the batch — so
        the failure is logged and swallowed.

        Args:
            cust: The `eumdac` `Customisation` handle to delete.
        """
        try:
            cust.delete()
        except Exception as exc:  # noqa: BLE001 - cleanup must never mask the cause
            logger.warning(f"Data Tailor customisation {cust} delete failed: {exc}")

    @staticmethod
    def _submit_customisation(datatailor, product, chain):
        """Submit a customisation, retrying transient EPCS failures (`G8`).

        The EUMETSAT EPCS endpoint intermittently returns `502 Bad
        Gateway`; a submit that fails with a transient marker is retried
        up to `TAILOR_SUBMIT_RETRIES` times with a linear backoff. A
        non-transient error (e.g. an invalid product id) is re-raised
        immediately.

        Note:
            If a create actually succeeds server-side but its response is
            lost (a dropped connection / timeout — the classified-transient
            cases), the retry submits a **second** customisation and the
            first is orphaned: the client never gets its handle, so it is
            not polled or deleted and lingers against the quota. The EPCS
            API offers no idempotency key to prevent this; recover by
            sweeping stale jobs (`eumdac.DataTailor(token).customisations`).

        Args:
            datatailor: The live `eumdac.DataTailor` client.
            product: The `eumdac` product handle to customise.
            chain: The `eumdac.tailor_models.Chain` describing the job.

        Returns:
            The `eumdac` `Customisation` handle for the submitted job.

        Raises:
            RuntimeError: When a transient submit keeps failing after
                `TAILOR_SUBMIT_RETRIES` attempts.
            Exception: Any non-transient submit error, re-raised as-is.
        """
        last_exc: Exception | None = None
        for attempt in range(1, TAILOR_SUBMIT_RETRIES + 1):
            try:
                return datatailor.new_customisation(product, chain)
            except Exception as exc:  # noqa: BLE001 - classified below
                message = str(exc).lower()
                if not any(mark in message for mark in _TRANSIENT_MARKERS):
                    raise
                last_exc = exc
                if attempt < TAILOR_SUBMIT_RETRIES:
                    time.sleep(TAILOR_SUBMIT_BACKOFF_S * attempt)
        raise RuntimeError(
            f"Data Tailor submit failed after {TAILOR_SUBMIT_RETRIES} "
            f"transient attempts: {last_exc}"
        ) from last_exc

    @staticmethod
    def _poll_customisation(cust) -> str:
        """Poll a customisation until it stops being active; return its status.

        Polls `cust.status` while it is `_TAILOR_ACTIVE` (`QUEUED` /
        `RUNNING`), sleeping `TAILOR_POLL_INITIAL_S` and growing the delay
        by `TAILOR_POLL_BACKOFF` up to `TAILOR_POLL_MAX_S`. Any other status
        is terminal and is returned immediately — so `DONE` succeeds and
        `FAILED` / `KILLED` / an unexpected stuck state (e.g. `INACTIVE`)
        fail fast rather than polling to the timeout. Gives up after
        `TAILOR_POLL_TIMEOUT_S` so a job stuck *active* cannot hang forever
        (`G8`); the final sleep is clamped to the remaining budget so the
        wall-clock never overshoots the timeout.

        Args:
            cust: The `eumdac` `Customisation` handle to poll.

        Returns:
            str: The terminal status (`"DONE"`, `"FAILED"`, `"KILLED"`, or
                any non-active value the service reports).

        Raises:
            TimeoutError: When the job is still active after
                `TAILOR_POLL_TIMEOUT_S`.
        """
        deadline = time.monotonic() + TAILOR_POLL_TIMEOUT_S
        delay = TAILOR_POLL_INITIAL_S
        while True:
            status = str(cust.status).upper()
            if status not in _TAILOR_ACTIVE:
                return status
            now = time.monotonic()
            if now >= deadline:
                raise TimeoutError(
                    f"Data Tailor customisation {cust} did not finish "
                    f"within {TAILOR_POLL_TIMEOUT_S:.0f}s (last status "
                    f"{status!r})."
                )
            time.sleep(min(delay, deadline - now))
            delay = min(delay * TAILOR_POLL_BACKOFF, TAILOR_POLL_MAX_S)

    @staticmethod
    def _logfile_tail(cust, limit: int = 1500) -> str:
        """Return the tail of a customisation's server log, if any.

        Args:
            cust: The `eumdac` `Customisation` handle.
            limit: Maximum number of trailing characters to return.

        Returns:
            str: The last `limit` characters of `cust.logfile`, or a
                placeholder when no log is available.
        """
        try:
            log = cust.logfile
        except Exception:  # noqa: BLE001 - a missing log must not mask the failure
            log = None
        if not log:
            return "(no customisation log available)"
        return str(log)[-limit:]

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

Initialise an EUMETSAT backend instance.

Resolves every requested dataset key against the catalog before calling the parent constructor, so the per-instance OUTPUT_KIND is set from the resolved row(s). The parent __init__ runs _initialize first (token mint), so the resolution cannot live there.

Parameters:

Name	Type	Description	Default
start	str	Inclusive start date as a string (parsed with fmt).	required
end	str	Inclusive end date as a string.	required
variables	dict[str, list[str]]	Mapping from curated dataset key to a list of selectors, e.g. {"msg-hrseviri": ["HRSEVIRI"]}. Selectors are informational for the whole-product fetch (G2).	required
lat_lim	list[float]	[lat_min, lat_max] in degrees.	required
lon_lim	list[float]	[lon_min, lon_max] in degrees.	required
temporal_resolution	str	Advisory cadence label. Defaults to "daily".	'daily'
path	Path | str | None	Output directory. Created by the parent class if it does not exist.	None
fmt	str	strptime format for start / end. Defaults to "%Y-%m-%d".	'%Y-%m-%d'
group	DataStoreGroup | str | None	Optional DataStoreGroup (or its string value) used to assert which Data Store group the requested collection(s) belong to (G2).	None
consumer_key	str | None	EUMETSAT consumer key. Falls back to EUMETSAT_CONSUMER_KEY, then ~/.eumdac/credentials.	None
consumer_secret	str | None	EUMETSAT consumer secret. Falls back to EUMETSAT_CONSUMER_SECRET, then the credentials file.	None
credentials_file	Path | str | None	Optional explicit path to a key,secret credentials file.	None

Raises:

Type	Description
ValueError	When variables is empty, a dataset key is unknown, or the requested collections do not all share one output_kind.

Source code in libs/providers/imagery/src/earthlens/eumetsat/backend.py

def __init__(
    self,
    start: str,
    end: str,
    variables: dict[str, list[str]],
    lat_lim: list[float],
    lon_lim: list[float],
    temporal_resolution: str = "daily",
    path: Path | str | None = None,
    fmt: str = "%Y-%m-%d",
    group: DataStoreGroup | str | None = None,
    consumer_key: str | None = None,
    consumer_secret: str | None = None,
    credentials_file: Path | str | None = None,
):
    """Initialise an EUMETSAT backend instance.

    Resolves every requested dataset key against the catalog
    **before** calling the parent constructor, so the per-instance
    `OUTPUT_KIND` is set from the resolved row(s). The parent
    `__init__` runs `_initialize` first (token mint), so the
    resolution cannot live there.

    Args:
        start: Inclusive start date as a string (parsed with `fmt`).
        end: Inclusive end date as a string.
        variables: Mapping from curated dataset key to a list of
            selectors, e.g. `{"msg-hrseviri": ["HRSEVIRI"]}`.
            Selectors are informational for the whole-product fetch
            (`G2`).
        lat_lim: `[lat_min, lat_max]` in degrees.
        lon_lim: `[lon_min, lon_max]` in degrees.
        temporal_resolution: Advisory cadence label. Defaults to
            `"daily"`.
        path: Output directory. Created by the parent class if it
            does not exist.
        fmt: `strptime` format for `start` / `end`. Defaults to
            `"%Y-%m-%d"`.
        group: Optional `DataStoreGroup` (or its string value) used to
            assert which Data Store group the requested collection(s)
            belong to (`G2`).
        consumer_key: EUMETSAT consumer key. Falls back to
            `EUMETSAT_CONSUMER_KEY`, then `~/.eumdac/credentials`.
        consumer_secret: EUMETSAT consumer secret. Falls back to
            `EUMETSAT_CONSUMER_SECRET`, then the credentials file.
        credentials_file: Optional explicit path to a `key,secret`
            credentials file.

    Raises:
        ValueError: When `variables` is empty, a dataset key is
            unknown, or the requested collections do not all share
            one `output_kind`.
    """
    self._group = group
    self._consumer_key = consumer_key
    self._consumer_secret = consumer_secret
    self._credentials_file = (
        Path(credentials_file) if credentials_file is not None else None
    )
    self._auth: EumetsatAuth | None = None
    self._show_progress = True

    self._catalog = Catalog()
    self._datasets: list[EumetsatDataset] = self._resolve_datasets(variables)
    self.OUTPUT_KIND = self._unify_output_kind(self._datasets)

    super().__init__(
        start=start,
        end=end,
        variables=variables,
        temporal_resolution=temporal_resolution,
        lat_lim=lat_lim,
        lon_lim=lon_lim,
        fmt=fmt,
        path=path,
    )

download(progress_bar=True, tailor=None)

Search the Data Store and return product paths (native or tailored).

With no tailor=, composes _search and _fetch to pull every product matching the request bbox + window to self.root_dir as whole native products (unchanged behaviour).

With tailor=TailorConfig(...), each matching product is routed through EUMETSAT Data Tailor (server-side subset / reproject / reformat, H4): submit a customisation, poll it to DONE, stream the customised output(s) to self.root_dir, and delete the customisation (G7). The returned paths are then the customised GeoTIFF / NetCDF files, not the native products (G6).

aggregate= is the temporal reducer (G1) and is a separate operation from the spatial tailor=; it is not implemented for EUMETSAT, so a non-None aggregate raises NotImplementedError. The two knobs compose — tailor server-side here, then reduce the result client-side with pyramids.

The tailor branch is fail-fast per batch: if one product's customisation fails, the error propagates and paths already streamed for earlier products are not returned (their files remain on disk). This mirrors the native fetch; keep batches small when a partial result would be costly to recompute.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Reserved for parity with the other backends; eumdac's streaming download has no built-in bar.	True
tailor	TailorConfig | None	Optional TailorConfig routing the request through Data Tailor. None (the default) keeps the native fetch.	None

Returns:

Type	Description
list[Path]	list[Path]: The native product paths, or — when tailor= is given — the customised output paths.

Raises:

Type	Description
ValueError	When tailor= names a dataset that is not Data-Tailor-eligible (G5).

Source code in libs/providers/imagery/src/earthlens/eumetsat/backend.py

def download(
    self,
    progress_bar: bool = True,
    tailor: TailorConfig | None = None,
) -> list[Path]:
    """Search the Data Store and return product paths (native or tailored).

    With no `tailor=`, composes `_search` and `_fetch` to pull every
    product matching the request bbox + window to `self.root_dir` as
    whole native products (unchanged behaviour).

    With `tailor=TailorConfig(...)`, each matching product is routed
    through EUMETSAT **Data Tailor** (server-side subset / reproject /
    reformat, `H4`): submit a customisation, poll it to `DONE`, stream
    the customised output(s) to `self.root_dir`, and delete the
    customisation (`G7`). The returned paths are then the customised
    GeoTIFF / NetCDF files, not the native products (`G6`).

    `aggregate=` is the **temporal** reducer (`G1`) and is a separate
    operation from the spatial `tailor=`; it is not implemented for
    EUMETSAT, so a non-`None` `aggregate` raises `NotImplementedError`.
    The two knobs compose — tailor server-side here, then reduce the
    result client-side with pyramids.

    The tailor branch is **fail-fast per batch**: if one product's
    customisation fails, the error propagates and paths already streamed
    for earlier products are not returned (their files remain on disk).
    This mirrors the native fetch; keep batches small when a partial
    result would be costly to recompute.

    Args:
        progress_bar: Reserved for parity with the other backends;
            `eumdac`'s streaming download has no built-in bar.
        tailor: Optional `TailorConfig` routing the request through
            Data Tailor. `None` (the default) keeps the native fetch.

    Returns:
        list[Path]: The native product paths, or — when `tailor=` is
            given — the customised output paths.

    Raises:
        ValueError: When `tailor=` names a dataset that is not
            Data-Tailor-eligible (`G5`).
    """
    self._show_progress = progress_bar
    if tailor is not None:
        return self._tailor(tailor)
    return self._api_via_search_fetch()

earthlens.eumetsat.catalog

Dataset-catalog loader for the EUMETSAT Data Store backend.

Hosts Catalog, the pydantic-backed reader for the bundled EUMETSAT catalog. Mirrors the shape of earthlens.earthdata.catalog and earthlens.gee.catalog: the catalog ships as a directory of per-group YAML files at src/earthlens/eumetsat/catalog/ (mtg.yaml, msg.yaml, metop.yaml, sentinel3.yaml, sentinel5p.yaml, …) plus a single _index.yaml carrying the merged available_datasets: list. Each per-group file contributes its datasets: block; the loader unions them into one Catalog at construction time.

A friendly dataset key (e.g. "msg-hrseviri") resolves to an EumetsatDataset via Catalog.get_dataset / Catalog()["..."] / Catalog.resolve. The row maps the friendly key to the real EUMETSAT EO:EUM:DAT:… collection id (the string eumdac's get_collection needs) and carries the fields the backend shapes a search and a fetch from: group, mission, the per-instance output_kind (G1), the on-disk format (so the aggregate= path knows which products are pyramids-readable NetCDF vs native), the informational selectors (G2), the tailor_product_type for the Data Tailor Chain (the tailor= server-side path), and the spatial / temporal coverage.

available_datasets: is the informational index of every Data Store collection id the browse walk found (the C7 auto-generated index); the curated datasets: map is the vetted subset (here, the whole catalog). The path to the bundled catalog directory lives at CATALOG_PATH; tests redirect the loader by pointing that module attribute at a temporary directory or a single YAML file.

Catalog

Bases: AbstractCatalog

Dataset catalog for the EUMETSAT Data Store backend.

Reads the bundled catalog/ directory (shipped as package data) and exposes its consumed top-level sections as typed pydantic fields. Instantiate with no arguments (Catalog()) — model_post_init parses the YAML and populates every field in one pass. Mirrors the earthlens.earthdata / earthlens.gee / earthlens.cmems catalogs: datasets (the curated map) and available_datasets (the informational index).

Attributes:

Name	Type	Description
available_datasets	list[str]	Informational list of every Data Store collection id the browse walk found. Runtime code does not consume it.
datasets	dict[str, EumetsatDataset]	Structural map keyed by the curated dataset key. Each value is an EumetsatDataset.

Examples:

• Resolve a curated dataset:

  >>> from earthlens.eumetsat import Catalog
  >>> "msg-hrseviri" in Catalog()
  True
  

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

class Catalog(AbstractCatalog):
    """Dataset catalog for the EUMETSAT Data Store backend.

    Reads the bundled `catalog/` directory (shipped as package data) and
    exposes its consumed top-level sections as typed pydantic fields.
    Instantiate with no arguments (`Catalog()`) — `model_post_init`
    parses the YAML and populates every field in one pass. Mirrors the
    `earthlens.earthdata` / `earthlens.gee` / `earthlens.cmems` catalogs:
    `datasets` (the curated map) and `available_datasets` (the
    informational index).

    Attributes:
        available_datasets: Informational list of every Data Store
            collection id the browse walk found. Runtime code does not
            consume it.
        datasets: Structural map keyed by the curated dataset key. Each
            value is an `EumetsatDataset`.

    Examples:
        - Resolve a curated dataset:
            ```python
            >>> from earthlens.eumetsat import Catalog
            >>> "msg-hrseviri" in Catalog()
            True

            ```
    """

    _catalog_kind: str = "EUMETSAT catalog"

    available_datasets: list[str] = Field(default_factory=list)
    datasets: dict[str, EumetsatDataset] = Field(default_factory=dict)

    @classmethod
    def _autoload(cls) -> dict[str, Any]:
        """Read the bundled catalog from disk.

        Returns:
            dict[str, Any]: The `available_datasets`, `datasets` read from
                the bundled catalog.
        """
        loaded = Catalog.load()
        return {
            "available_datasets": loaded.available_datasets,
            "datasets": loaded.datasets,
        }

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

        Args:
            catalog_path: Path to the `catalog/` directory or a single
                `*.yaml` file. Defaults to module-level `CATALOG_PATH`.

        Returns:
            A fully-populated `Catalog`.

        Raises:
            ValueError: Propagated from `_load_catalog_data`.
        """
        catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
        available, datasets = _load_catalog_data(catalog_path)
        return cls(
            available_datasets=list(available),
            datasets=dict(datasets),
        )

    def get_catalog(self) -> dict[str, EumetsatDataset]:
        """Return the structural per-dataset map.

        Satisfies the abstract base's contract; the actual parsing is
        done in `model_post_init`.

        Returns:
            dict[str, EumetsatDataset]: One entry per curated dataset.
                Same object as `datasets`.
        """
        return self.datasets

    def resolve(
        self, key: str, group: DataStoreGroup | str | None = None
    ) -> EumetsatDataset:
        """Resolve a dataset key, optionally disambiguated by group.

        Most keys map one-to-one to a curated row; the inherited
        :meth:`get_dataset` (with its did-you-mean) handles those. The
        `group=` filter additionally asserts which Data Store group the
        resolved dataset belongs to (`G2`) — the EUMETSAT analog of the
        Earthdata backend's `daac=` filter.

        Args:
            key: Curated dataset key (a member of `datasets`).
            group: Optional `DataStoreGroup` (or its string value) the
                resolved row's `group` must match.

        Returns:
            EumetsatDataset: The resolved row.

        Raises:
            ValueError: When `key` is unknown (with a did-you-mean hint),
                or `group=` is given but does not match the row's group.

        Examples:
            - Resolve a key and read its group:
                ```python
                >>> from earthlens.eumetsat import Catalog
                >>> Catalog().resolve("msg-hrseviri").group.value
                'MSG'

                ```
        """
        dataset = cast("EumetsatDataset", self.get_dataset(key))
        if group is not None:
            wanted = group.value if isinstance(group, DataStoreGroup) else str(group)
            if dataset.group.value != wanted:
                raise ValueError(
                    f"dataset {key!r} is in group "
                    f"{dataset.group.value!r}, not the requested "
                    f"group={wanted!r}."
                )
        return dataset

get_catalog()

Return the structural per-dataset map.

Satisfies the abstract base's contract; the actual parsing is done in model_post_init.

Returns:

Type	Description
dict[str, EumetsatDataset]	dict[str, EumetsatDataset]: One entry per curated dataset. Same object as datasets.

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

def get_catalog(self) -> dict[str, EumetsatDataset]:
    """Return the structural per-dataset map.

    Satisfies the abstract base's contract; the actual parsing is
    done in `model_post_init`.

    Returns:
        dict[str, EumetsatDataset]: One entry per curated dataset.
            Same object as `datasets`.
    """
    return self.datasets

load(catalog_path=None) classmethod

Read the EUMETSAT catalog from disk (cached).

Parameters:

Name	Type	Description	Default
catalog_path	Path | None	Path to the catalog/ directory or a single *.yaml file. Defaults to module-level CATALOG_PATH.	None

Returns:

Type	Description
Catalog	A fully-populated Catalog.

Raises:

Type	Description
ValueError	Propagated from _load_catalog_data.

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

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

    Args:
        catalog_path: Path to the `catalog/` directory or a single
            `*.yaml` file. Defaults to module-level `CATALOG_PATH`.

    Returns:
        A fully-populated `Catalog`.

    Raises:
        ValueError: Propagated from `_load_catalog_data`.
    """
    catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
    available, datasets = _load_catalog_data(catalog_path)
    return cls(
        available_datasets=list(available),
        datasets=dict(datasets),
    )

resolve(key, group=None)

Resolve a dataset key, optionally disambiguated by group.

Most keys map one-to-one to a curated row; the inherited :meth:get_dataset (with its did-you-mean) handles those. The group= filter additionally asserts which Data Store group the resolved dataset belongs to (G2) — the EUMETSAT analog of the Earthdata backend's daac= filter.

Parameters:

Name	Type	Description	Default
key	str	Curated dataset key (a member of datasets).	required
group	DataStoreGroup | str | None	Optional DataStoreGroup (or its string value) the resolved row's group must match.	None

Returns:

Name	Type	Description
EumetsatDataset	EumetsatDataset	The resolved row.

Raises:

Type	Description
ValueError	When key is unknown (with a did-you-mean hint), or group= is given but does not match the row's group.

Examples:

• Resolve a key and read its group:

  >>> from earthlens.eumetsat import Catalog
  >>> Catalog().resolve("msg-hrseviri").group.value
  'MSG'
  

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

def resolve(
    self, key: str, group: DataStoreGroup | str | None = None
) -> EumetsatDataset:
    """Resolve a dataset key, optionally disambiguated by group.

    Most keys map one-to-one to a curated row; the inherited
    :meth:`get_dataset` (with its did-you-mean) handles those. The
    `group=` filter additionally asserts which Data Store group the
    resolved dataset belongs to (`G2`) — the EUMETSAT analog of the
    Earthdata backend's `daac=` filter.

    Args:
        key: Curated dataset key (a member of `datasets`).
        group: Optional `DataStoreGroup` (or its string value) the
            resolved row's `group` must match.

    Returns:
        EumetsatDataset: The resolved row.

    Raises:
        ValueError: When `key` is unknown (with a did-you-mean hint),
            or `group=` is given but does not match the row's group.

    Examples:
        - Resolve a key and read its group:
            ```python
            >>> from earthlens.eumetsat import Catalog
            >>> Catalog().resolve("msg-hrseviri").group.value
            'MSG'

            ```
    """
    dataset = cast("EumetsatDataset", self.get_dataset(key))
    if group is not None:
        wanted = group.value if isinstance(group, DataStoreGroup) else str(group)
        if dataset.group.value != wanted:
            raise ValueError(
                f"dataset {key!r} is in group "
                f"{dataset.group.value!r}, not the requested "
                f"group={wanted!r}."
            )
    return dataset

DataStoreGroup

Bases: StrEnum

The EUMETSAT Data Store collection groups (mission families).

Each value is the human-readable group label carried on a catalog row's group: field and accepted by the backend's group= kwarg to disambiguate a collection key shared across groups (G2).

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

class DataStoreGroup(StrEnum):
    """The EUMETSAT Data Store collection groups (mission families).

    Each value is the human-readable group label carried on a catalog
    row's `group:` field and accepted by the backend's `group=` kwarg to
    disambiguate a collection key shared across groups (`G2`).
    """

    MTG = "MTG"
    MSG = "MSG"
    MFG = "MFG"
    METOP = "Metop"
    METOP_SG = "Metop-SG"
    SENTINEL_3 = "Sentinel-3"
    SENTINEL_5P = "Sentinel-5P"
    SENTINEL_6 = "Sentinel-6"
    OSI_SAF = "OSI-SAF"
    OTHER = "Other"

EumetsatDataset

Bases: BaseModel

One curated EUMETSAT Data Store dataset (collection) row.

Mirrors a single datasets.<key>: block in one of the per-group catalog/*.yaml files. The friendly dataset key is the parent key in Catalog.datasets and is not stored on the row.

Attributes:

Name	Type	Description
collection_id	str	The real Data Store collection id the search uses, e.g. "EO:EUM:DAT:MSG:HRSEVIRI".
group	DataStoreGroup	The Data Store group (mission family) this dataset belongs to — used by the backend's group= disambiguation.
mission	str	Short mission tag ("msg", "mtg", "metop", "sentinel-3", …). Advisory.
output_kind	OutputKindLiteral	The per-dataset output shape — "raster", "vector", or "tabular". Copied onto the backend instance's OUTPUT_KIND (G1).
format	str	On-disk product format ("native", "netcdf", "grib", "bufr", …). "native" SEVIRI / FCI needs the satpy bridge (PY-2) to read; "netcdf" is pyramids-readable today (G4).
cadence	CadenceLiteral	Native temporal cadence (advisory).
timeliness	TimelinessLiteral | None	Delivery timeliness of the dataset — "nrt", "reprocessed", or "offline" — or None when the distinction does not apply (e.g. geostationary L1.5 imagery). Recorded so a caller can tell a near-real-time stream from a reprocessed archive (the Sentinel-5P collections mix the two).
selectors	list[str]	Informational product-type / band selectors (G2). EUMETSAT delivers whole products, so selectors do not subset the download; they seed catalog metadata and the Data Tailor Chain.
tailor_product_type	str | None	The Data Tailor product-type id for the tailor= server-side subset / reproject / reformat path; None when the collection is not Data-Tailor-eligible.
extent	Extent	Extent — lat / lon coverage.
temporal	TemporalCoverage	TemporalCoverage — start / end dates.

Examples:

• Inspect a curated raster row:

  >>> from earthlens.eumetsat import Catalog
  >>> ds = Catalog().get_dataset("msg-hrseviri")
  >>> ds.collection_id
  'EO:EUM:DAT:MSG:HRSEVIRI'
  >>> ds.output_kind
  'raster'
  

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

class EumetsatDataset(BaseModel):
    """One curated EUMETSAT Data Store dataset (collection) row.

    Mirrors a single `datasets.<key>:` block in one of the per-group
    `catalog/*.yaml` files. The friendly dataset key is the parent key
    in `Catalog.datasets` and is not stored on the row.

    Attributes:
        collection_id: The real Data Store collection id the search
            uses, e.g. `"EO:EUM:DAT:MSG:HRSEVIRI"`.
        group: The Data Store group (mission family) this dataset
            belongs to — used by the backend's `group=` disambiguation.
        mission: Short mission tag (`"msg"`, `"mtg"`, `"metop"`,
            `"sentinel-3"`, …). Advisory.
        output_kind: The per-dataset output shape — `"raster"`,
            `"vector"`, or `"tabular"`. Copied onto the backend
            instance's `OUTPUT_KIND` (`G1`).
        format: On-disk product format (`"native"`, `"netcdf"`,
            `"grib"`, `"bufr"`, …). `"native"` SEVIRI / FCI needs the
            satpy bridge (`PY-2`) to read; `"netcdf"` is pyramids-readable
            today (`G4`).
        cadence: Native temporal cadence (advisory).
        timeliness: Delivery timeliness of the dataset — `"nrt"`,
            `"reprocessed"`, or `"offline"` — or `None` when the
            distinction does not apply (e.g. geostationary L1.5 imagery).
            Recorded so a caller can tell a near-real-time stream from a
            reprocessed archive (the Sentinel-5P collections mix the two).
        selectors: Informational product-type / band selectors (`G2`).
            EUMETSAT delivers whole products, so selectors do not subset
            the download; they seed catalog metadata and the Data Tailor
            `Chain`.
        tailor_product_type: The Data Tailor product-type id for the
            `tailor=` server-side subset / reproject / reformat path;
            `None` when the collection is not Data-Tailor-eligible.
        extent: `Extent` — lat / lon coverage.
        temporal: `TemporalCoverage` — start / end dates.

    Examples:
        - Inspect a curated raster row:
            ```python
            >>> from earthlens.eumetsat import Catalog
            >>> ds = Catalog().get_dataset("msg-hrseviri")
            >>> ds.collection_id
            'EO:EUM:DAT:MSG:HRSEVIRI'
            >>> ds.output_kind
            'raster'

            ```
    """

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

    collection_id: str
    group: DataStoreGroup
    mission: str = ""
    output_kind: OutputKindLiteral = "raster"
    format: str = ""
    cadence: CadenceLiteral = "irregular"
    timeliness: TimelinessLiteral | None = None
    selectors: list[str] = Field(default_factory=list)
    tailor_product_type: str | None = None
    extent: Extent = Field(default_factory=Extent)
    temporal: TemporalCoverage = Field(default_factory=TemporalCoverage)

Extent

Bases: BaseModel

Spatial coverage of an EUMETSAT collection (lat / lon bounds).

Mirrors the extent: block in the YAML. Whole-disk geostationary products (SEVIRI / FCI) use the ~±79° sub-satellite extent; polar / mirror products are global.

Attributes:

Name	Type	Description
lat	list[float]	[lat_min, lat_max] in degrees. Empty means unspecified.
lon	list[float]	[lon_min, lon_max] in degrees. Empty means unspecified.

Examples:

• A geostationary 0° disk extent:

  >>> from earthlens.eumetsat.catalog import Extent
  >>> Extent(lat=[-79, 79], lon=[-79, 79]).lat
  [-79.0, 79.0]
  

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

class Extent(BaseModel):
    """Spatial coverage of an EUMETSAT collection (lat / lon bounds).

    Mirrors the `extent:` block in the YAML. Whole-disk geostationary
    products (SEVIRI / FCI) use the ~±79° sub-satellite extent; polar /
    mirror products are global.

    Attributes:
        lat: `[lat_min, lat_max]` in degrees. Empty means unspecified.
        lon: `[lon_min, lon_max]` in degrees. Empty means unspecified.

    Examples:
        - A geostationary 0° disk extent:
            ```python
            >>> from earthlens.eumetsat.catalog import Extent
            >>> Extent(lat=[-79, 79], lon=[-79, 79]).lat
            [-79.0, 79.0]

            ```
    """

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

    lat: list[float] = Field(default_factory=list)
    lon: list[float] = Field(default_factory=list)

TemporalCoverage

Bases: BaseModel

Temporal coverage of an EUMETSAT collection (start + optional end).

Mirrors the temporal: block in the YAML. end: null (or a missing end) means the collection is ongoing / rolling.

Attributes:

Name	Type	Description
start	str | None	First date with data, as a YYYY-MM-DD string. May be None when the start date is not pinned in the YAML.
end	str | None	Last date with data, or None for an ongoing collection.

Examples:

• An ongoing collection:

  >>> from earthlens.eumetsat.catalog import TemporalCoverage
  >>> tc = TemporalCoverage(start="2004-01-19", end=None)
  >>> tc.start, tc.end
  ('2004-01-19', None)
  

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

class TemporalCoverage(BaseModel):
    """Temporal coverage of an EUMETSAT collection (start + optional end).

    Mirrors the `temporal:` block in the YAML. `end: null` (or a missing
    `end`) means the collection is ongoing / rolling.

    Attributes:
        start: First date with data, as a `YYYY-MM-DD` string. May be
            `None` when the start date is not pinned in the YAML.
        end: Last date with data, or `None` for an ongoing collection.

    Examples:
        - An ongoing collection:
            ```python
            >>> from earthlens.eumetsat.catalog import TemporalCoverage
            >>> tc = TemporalCoverage(start="2004-01-19", end=None)
            >>> tc.start, tc.end
            ('2004-01-19', None)

            ```
    """

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

    start: str | None = None
    end: str | None = None

    @field_validator("start", "end", mode="before")
    @classmethod
    def _coerce_date(cls, value: Any) -> Any:
        """Accept a `datetime.date` (PyYAML's native parse) as ISO string.

        Args:
            value: Raw YAML value — a string, a `datetime.date`, or
                `None`.

        Returns:
            An ISO-format string (`"YYYY-MM-DD"`) or `None`.
        """
        if isinstance(value, _dt.date):
            return value.isoformat()
        return value

clear_catalog_cache()

Empty the module-level catalog parse cache.

Useful in tests that rewrite the catalog on disk and want to force a re-parse. Production callers do not need this — the cache keys include every contributing file's st_mtime_ns, so any real file mutation invalidates the entry on its own.

Source code in libs/providers/imagery/src/earthlens/eumetsat/catalog.py

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

    Useful in tests that rewrite the catalog on disk and want to force a
    re-parse. Production callers do not need this — the cache keys
    include every contributing file's `st_mtime_ns`, so any real file
    mutation invalidates the entry on its own.
    """
    _CATALOG_CACHE.clear()

earthlens.eumetsat.auth

Credentials and authentication for the EUMETSAT Data Store backend.

Hosts EumetsatAuth, an earthlens.base.AbstractAuth subclass that wraps eumdac.AccessToken. A single EUMETSAT consumer key / secret pair mints the OAuth2 bearer token that unlocks every Data Store collection the backend reaches — MTG-I1 FCI, MSG SEVIRI, Metop (ASCAT / IASI), the Sentinel-3 / -5P / -6 mirrors, and the OSI SAF / CDR / FDR families. The bearer is short-lived (~1 h) and eumdac refreshes it transparently; this wrapper re-mints a fresh AccessToken only when the cached one has expired.

The auth wrapper exists so that:

• The backend builds an EumetsatCredentials value object up front, validates it with pydantic, and passes it through super().__init__(creds) — consistent with EarthdataAuth / CmemsAuth.
• configure() is idempotent — a second call after is_authenticated() returns True short-circuits, so it is safe to call from long-lived workers without re-minting on every download().
• eumdac auth failures surface as the cross-backend earthlens.base.AuthenticationError, so a caller can write one except AuthenticationError clause across CMEMS / Earthdata / EUMETSAT.

The credentials-resolution priority used by configure() is:

1. EUMETSAT_CONSUMER_KEY / EUMETSAT_CONSUMER_SECRET environment variables.
2. The consumer_key / consumer_secret passed to the constructor.
3. A ~/.eumdac/credentials file (a single key,secret line, the format eumdac set-credentials writes; the directory is overridable via EUMDAC_CONFIG_DIR, or pointed at explicitly with credentials_file).

AuthenticationError

Bases: AuthenticationError

Raised when eumdac cannot mint an OAuth2 token.

Wraps the underlying eumdac / HTTP failure with a message that names a fix: register a consumer key / secret at the EUMETSAT API-key page, set the EUMETSAT_CONSUMER_KEY / EUMETSAT_CONSUMER_SECRET environment variables, or write a ~/.eumdac/credentials file.

A subclass of the cross-backend earthlens.base.AuthenticationError so callers can catch every backend's auth failure with one except clause.

Source code in libs/providers/imagery/src/earthlens/eumetsat/auth.py

class AuthenticationError(_BaseAuthenticationError):
    """Raised when `eumdac` cannot mint an OAuth2 token.

    Wraps the underlying `eumdac` / HTTP failure with a message that
    names a fix: register a consumer key / secret at the EUMETSAT API-key
    page, set the `EUMETSAT_CONSUMER_KEY` / `EUMETSAT_CONSUMER_SECRET`
    environment variables, or write a `~/.eumdac/credentials` file.

    A subclass of the cross-backend `earthlens.base.AuthenticationError`
    so callers can catch every backend's auth failure with one `except`
    clause.
    """

EumetsatAuth

Bases: AbstractAuth[EumetsatCredentials]

Authenticate against the EUMETSAT Data Store (OAuth2).

Wraps eumdac.AccessToken in the earthlens.base.AbstractAuth contract. configure() resolves a consumer key / secret pair (environment → constructor kwargs → ~/.eumdac/credentials) and mints an eumdac.AccessToken, which auto-refreshes the ~1 h bearer internally. The datastore / datatailor helpers build the matching eumdac clients from the live token.

The class is a context manager (inherited from AbstractAuth): with EumetsatAuth(creds) as auth: ... calls configure() on enter and the default no-op close() on exit.

Attributes:

Name	Type	Description
_creds		The EumetsatCredentials passed at construction. Read by configure to resolve the credential pair. Treated as write-once.
_token		The eumdac.AccessToken minted by a successful configure(), or None before it runs.

Examples:

• Build, configure, inspect — marked # doctest: +SKIP because it mints a real OAuth2 token:

  >>> from earthlens.eumetsat import EumetsatAuth, EumetsatCredentials
  >>> auth = EumetsatAuth(EumetsatCredentials())  # doctest: +SKIP
  >>> auth.configure()  # doctest: +SKIP
  >>> auth.is_authenticated()  # doctest: +SKIP
  True
  

Source code in libs/providers/imagery/src/earthlens/eumetsat/auth.py

class EumetsatAuth(AbstractAuth[EumetsatCredentials]):
    """Authenticate against the EUMETSAT Data Store (OAuth2).

    Wraps `eumdac.AccessToken` in the `earthlens.base.AbstractAuth`
    contract. `configure()` resolves a consumer key / secret pair
    (environment → constructor kwargs → `~/.eumdac/credentials`) and
    mints an `eumdac.AccessToken`, which auto-refreshes the ~1 h bearer
    internally. The `datastore` / `datatailor` helpers build the matching
    `eumdac` clients from the live token.

    The class is a context manager (inherited from `AbstractAuth`):
    `with EumetsatAuth(creds) as auth: ...` calls `configure()` on enter
    and the default no-op `close()` on exit.

    Attributes:
        _creds: The `EumetsatCredentials` passed at construction. Read by
            `configure` to resolve the credential pair. Treated as
            write-once.
        _token: The `eumdac.AccessToken` minted by a successful
            `configure()`, or `None` before it runs.

    Examples:
        - Build, configure, inspect — marked `# doctest: +SKIP` because
          it mints a real OAuth2 token:

            ```python
            >>> from earthlens.eumetsat import EumetsatAuth, EumetsatCredentials
            >>> auth = EumetsatAuth(EumetsatCredentials())  # doctest: +SKIP
            >>> auth.configure()  # doctest: +SKIP
            >>> auth.is_authenticated()  # doctest: +SKIP
            True

            ```
    """

    def __init__(self, credentials: EumetsatCredentials) -> None:
        """Store credentials; does not authenticate.

        Construction is side-effect-free — the user (or the backend's
        `_initialize`) must call `configure` (or use the context-manager
        form) to mint a token.

        Args:
            credentials: The `EumetsatCredentials` value object carrying
                the resolution rules.
        """
        super().__init__(credentials)
        self._token = None

    def _resolve_pair(self) -> tuple[str | None, str | None]:
        """Resolve the consumer key / secret pair.

        Resolution order: the `EUMETSAT_CONSUMER_KEY` /
        `EUMETSAT_CONSUMER_SECRET` environment variables, then the
        constructor kwargs, then a `key,secret` line in the
        credentials file (the explicit `credentials_file`, else
        `EUMDAC_CONFIG_DIR/credentials`, else `~/.eumdac/credentials`).
        The first source that yields both halves wins.

        Returns:
            tuple[str | None, str | None]: The `(consumer_key,
                consumer_secret)` pair; either element is `None` when no
                source supplied it.
        """
        key = os.getenv("EUMETSAT_CONSUMER_KEY") or self._creds.consumer_key
        secret = os.getenv("EUMETSAT_CONSUMER_SECRET")
        if not secret and self._creds.consumer_secret is not None:
            secret = self._creds.consumer_secret.get_secret_value()
        if key and secret:
            return key, secret
        file_key, file_secret = self._read_credentials_file()
        return key or file_key, secret or file_secret

    def _credentials_path(self) -> Path:
        """Return the credentials-file path to read.

        Honours an explicit `credentials_file`, then the
        `EUMDAC_CONFIG_DIR` environment variable `eumdac` itself reads,
        then the default `~/.eumdac/credentials`.

        Returns:
            Path: The resolved credentials-file path (which may not
                exist).
        """
        if self._creds.credentials_file is not None:
            return self._creds.credentials_file
        config_dir = os.getenv("EUMDAC_CONFIG_DIR")
        base = Path(config_dir) if config_dir else Path.home() / ".eumdac"
        return base / "credentials"

    def _read_credentials_file(self) -> tuple[str | None, str | None]:
        """Parse the `key,secret` credentials file, if present.

        Returns:
            tuple[str | None, str | None]: The `(key, secret)` parsed
                from the single `key,secret` line, or `(None, None)` when
                the file is missing or malformed.
        """
        path = self._credentials_path()
        try:
            content = path.read_text(encoding="utf-8").strip()
        except (FileNotFoundError, OSError):
            return None, None
        match = _CREDENTIALS_LINE.match(content)
        if match is None:
            return None, None
        return match.group(1), match.group(2)

    def configure(self) -> None:
        """Mint the OAuth2 token via `eumdac.AccessToken`.

        Idempotent — short-circuits when `is_authenticated` already
        returns `True`. Resolves the consumer key / secret pair
        (environment → kwargs → credentials file), then constructs an
        `eumdac.AccessToken((key, secret))`. The token object refreshes
        the ~1 h bearer internally; this wrapper re-mints a fresh one
        only after the cached token's `expiration` has passed (see
        `is_authenticated`).

        Raises:
            ImportError: When the `eumdac` SDK is not installed (the
                `[eumetsat]` extra is missing).
            AuthenticationError: When no credential pair resolves, or
                `eumdac` rejects the pair while minting the token.
        """
        if self.is_authenticated():
            return

        eumdac = _import_eumdac()  # lazy — only needed when authenticating

        key, secret = self._resolve_pair()
        if not key or not secret:
            raise AuthenticationError(
                "no EUMETSAT credentials resolved. Set "
                "EUMETSAT_CONSUMER_KEY / EUMETSAT_CONSUMER_SECRET, pass "
                "consumer_key= / consumer_secret=, or write a "
                f"'key,secret' line to ~/.eumdac/credentials. Register a "
                f"consumer key/secret at {_KEY_MGMT_URL}. See {_DOCS_URL}."
            )

        try:
            self._token = eumdac.AccessToken((key, secret))
        except Exception as exc:  # noqa: BLE001 - re-raised as AuthenticationError
            raise AuthenticationError(
                "EUMETSAT token request failed "
                f"({type(exc).__name__}: {exc}). Check the consumer "
                f"key/secret at {_KEY_MGMT_URL}."
            ) from exc

    def is_authenticated(self) -> bool:
        """Return whether a live, unexpired token is held.

        Cheap predicate — does not call the network. Returns `True` only
        when `configure()` has minted a token whose `expiration`
        (a `datetime`) is still in the future; an expired token reports
        `False` so `configure()` re-mints. A token whose `expiration`
        cannot be read is treated as live (the SDK refreshes it lazily).

        Returns:
            bool: `True` while the held token is valid, `False` before
                `configure` or once the token has expired.
        """
        if self._token is None:
            return False
        try:
            return datetime.now() < self._token.expiration
        except (AttributeError, TypeError, ValueError):
            # The SDK refreshes the bearer lazily, so a token whose
            # `expiration` is missing/unreadable is treated as live rather
            # than forcing a re-mint. Genuinely unexpected errors propagate.
            return True

    def datastore(self) -> Any:
        """Return an `eumdac.DataStore` bound to the live token.

        Returns:
            eumdac.DataStore: The Data Store client used to resolve
                collections and search products.

        Raises:
            ImportError: When the `[eumetsat]` extra (`eumdac`) is not
                installed.
            AuthenticationError: When `configure()` has not minted a
                token yet.
        """
        eumdac = _import_eumdac()
        if self._token is None:
            raise AuthenticationError(
                "datastore() called before configure(); authenticate "
                "first via configure() or the context-manager form."
            )
        return eumdac.DataStore(self._token)

    def datatailor(self) -> Any:
        """Return an `eumdac.DataTailor` bound to the live token.

        Used by the Data Tailor (server-side subset / reproject / reformat)
        path — `EUMETSAT.download(tailor=...)`. The native whole-product
        fetch does not call this.

        Returns:
            eumdac.DataTailor: The Data Tailor client.

        Raises:
            ImportError: When the `[eumetsat]` extra (`eumdac`) is not
                installed.
            AuthenticationError: When `configure()` has not minted a
                token yet.
        """
        eumdac = _import_eumdac()
        if self._token is None:
            raise AuthenticationError(
                "datatailor() called before configure(); authenticate "
                "first via configure() or the context-manager form."
            )
        return eumdac.DataTailor(self._token)

__init__(credentials)

Store credentials; does not authenticate.

Construction is side-effect-free — the user (or the backend's _initialize) must call configure (or use the context-manager form) to mint a token.

Parameters:

Name	Type	Description	Default
credentials	EumetsatCredentials	The EumetsatCredentials value object carrying the resolution rules.	required

Source code in libs/providers/imagery/src/earthlens/eumetsat/auth.py

def __init__(self, credentials: EumetsatCredentials) -> None:
    """Store credentials; does not authenticate.

    Construction is side-effect-free — the user (or the backend's
    `_initialize`) must call `configure` (or use the context-manager
    form) to mint a token.

    Args:
        credentials: The `EumetsatCredentials` value object carrying
            the resolution rules.
    """
    super().__init__(credentials)
    self._token = None

configure()

Mint the OAuth2 token via eumdac.AccessToken.

Idempotent — short-circuits when is_authenticated already returns True. Resolves the consumer key / secret pair (environment → kwargs → credentials file), then constructs an eumdac.AccessToken((key, secret)). The token object refreshes the ~1 h bearer internally; this wrapper re-mints a fresh one only after the cached token's expiration has passed (see is_authenticated).

Raises:

Type	Description
ImportError	When the eumdac SDK is not installed (the [eumetsat] extra is missing).
AuthenticationError	When no credential pair resolves, or eumdac rejects the pair while minting the token.

Source code in libs/providers/imagery/src/earthlens/eumetsat/auth.py

def configure(self) -> None:
    """Mint the OAuth2 token via `eumdac.AccessToken`.

    Idempotent — short-circuits when `is_authenticated` already
    returns `True`. Resolves the consumer key / secret pair
    (environment → kwargs → credentials file), then constructs an
    `eumdac.AccessToken((key, secret))`. The token object refreshes
    the ~1 h bearer internally; this wrapper re-mints a fresh one
    only after the cached token's `expiration` has passed (see
    `is_authenticated`).

    Raises:
        ImportError: When the `eumdac` SDK is not installed (the
            `[eumetsat]` extra is missing).
        AuthenticationError: When no credential pair resolves, or
            `eumdac` rejects the pair while minting the token.
    """
    if self.is_authenticated():
        return

    eumdac = _import_eumdac()  # lazy — only needed when authenticating

    key, secret = self._resolve_pair()
    if not key or not secret:
        raise AuthenticationError(
            "no EUMETSAT credentials resolved. Set "
            "EUMETSAT_CONSUMER_KEY / EUMETSAT_CONSUMER_SECRET, pass "
            "consumer_key= / consumer_secret=, or write a "
            f"'key,secret' line to ~/.eumdac/credentials. Register a "
            f"consumer key/secret at {_KEY_MGMT_URL}. See {_DOCS_URL}."
        )

    try:
        self._token = eumdac.AccessToken((key, secret))
    except Exception as exc:  # noqa: BLE001 - re-raised as AuthenticationError
        raise AuthenticationError(
            "EUMETSAT token request failed "
            f"({type(exc).__name__}: {exc}). Check the consumer "
            f"key/secret at {_KEY_MGMT_URL}."
        ) from exc

datastore()

Return an eumdac.DataStore bound to the live token.

Returns:

Type	Description
Any	eumdac.DataStore: The Data Store client used to resolve collections and search products.

Raises:

Type	Description
ImportError	When the [eumetsat] extra (eumdac) is not installed.
AuthenticationError	When configure() has not minted a token yet.

Source code in libs/providers/imagery/src/earthlens/eumetsat/auth.py

def datastore(self) -> Any:
    """Return an `eumdac.DataStore` bound to the live token.

    Returns:
        eumdac.DataStore: The Data Store client used to resolve
            collections and search products.

    Raises:
        ImportError: When the `[eumetsat]` extra (`eumdac`) is not
            installed.
        AuthenticationError: When `configure()` has not minted a
            token yet.
    """
    eumdac = _import_eumdac()
    if self._token is None:
        raise AuthenticationError(
            "datastore() called before configure(); authenticate "
            "first via configure() or the context-manager form."
        )
    return eumdac.DataStore(self._token)

datatailor()

Return an eumdac.DataTailor bound to the live token.

Used by the Data Tailor (server-side subset / reproject / reformat) path — EUMETSAT.download(tailor=...). The native whole-product fetch does not call this.

Returns:

Type	Description
Any	eumdac.DataTailor: The Data Tailor client.

Raises:

Type	Description
ImportError	When the [eumetsat] extra (eumdac) is not installed.
AuthenticationError	When configure() has not minted a token yet.

Source code in libs/providers/imagery/src/earthlens/eumetsat/auth.py

def datatailor(self) -> Any:
    """Return an `eumdac.DataTailor` bound to the live token.

    Used by the Data Tailor (server-side subset / reproject / reformat)
    path — `EUMETSAT.download(tailor=...)`. The native whole-product
    fetch does not call this.

    Returns:
        eumdac.DataTailor: The Data Tailor client.

    Raises:
        ImportError: When the `[eumetsat]` extra (`eumdac`) is not
            installed.
        AuthenticationError: When `configure()` has not minted a
            token yet.
    """
    eumdac = _import_eumdac()
    if self._token is None:
        raise AuthenticationError(
            "datatailor() called before configure(); authenticate "
            "first via configure() or the context-manager form."
        )
    return eumdac.DataTailor(self._token)

is_authenticated()

Return whether a live, unexpired token is held.

Cheap predicate — does not call the network. Returns True only when configure() has minted a token whose expiration (a datetime) is still in the future; an expired token reports False so configure() re-mints. A token whose expiration cannot be read is treated as live (the SDK refreshes it lazily).

Returns:

Name	Type	Description
bool	bool	True while the held token is valid, False before configure or once the token has expired.

Source code in libs/providers/imagery/src/earthlens/eumetsat/auth.py

def is_authenticated(self) -> bool:
    """Return whether a live, unexpired token is held.

    Cheap predicate — does not call the network. Returns `True` only
    when `configure()` has minted a token whose `expiration`
    (a `datetime`) is still in the future; an expired token reports
    `False` so `configure()` re-mints. A token whose `expiration`
    cannot be read is treated as live (the SDK refreshes it lazily).

    Returns:
        bool: `True` while the held token is valid, `False` before
            `configure` or once the token has expired.
    """
    if self._token is None:
        return False
    try:
        return datetime.now() < self._token.expiration
    except (AttributeError, TypeError, ValueError):
        # The SDK refreshes the bearer lazily, so a token whose
        # `expiration` is missing/unreadable is treated as live rather
        # than forcing a re-mint. Genuinely unexpected errors propagate.
        return True

EumetsatCredentials

Bases: BaseModel

Frozen value object holding the EUMETSAT consumer key / secret.

Every field is optional — the auth wrapper resolves the actual pair at configure() time from the environment, these fields, then the ~/.eumdac/credentials file. Validation is intentionally permissive; the real "do these creds work?" gate is EumetsatAuth.configure, which mints a token against the OAuth2 endpoint.

Attributes:

Name	Type	Description
consumer_key	str | None	The EUMETSAT consumer key (the OAuth2 client id). None means "look at the environment / credentials file".
consumer_secret	SecretStr | None	The matching consumer secret, stored as a pydantic.SecretStr so it is never echoed by repr(creds) or in logs. None means same as consumer_key.
credentials_file	Path | None	Optional explicit path to a key,secret credentials file. None falls back to EUMDAC_CONFIG_DIR/credentials, then ~/.eumdac/credentials.

Examples:

• All fields optional — rely on env / credentials file:

  >>> from earthlens.eumetsat import EumetsatCredentials
  >>> creds = EumetsatCredentials()
  >>> creds.consumer_key is None and creds.consumer_secret is None
  True
  

• SecretStr hides the secret in repr:

  >>> from earthlens.eumetsat import EumetsatCredentials
  >>> creds = EumetsatCredentials(consumer_key="k", consumer_secret="topsecret")
  >>> "topsecret" in repr(creds)
  False
  

Source code in libs/providers/imagery/src/earthlens/eumetsat/auth.py

class EumetsatCredentials(BaseModel):
    """Frozen value object holding the EUMETSAT consumer key / secret.

    Every field is optional — the auth wrapper resolves the actual pair
    at `configure()` time from the environment, these fields, then the
    `~/.eumdac/credentials` file. Validation is intentionally permissive;
    the real "do these creds work?" gate is
    `EumetsatAuth.configure`, which mints a token against the OAuth2
    endpoint.

    Attributes:
        consumer_key: The EUMETSAT consumer key (the OAuth2 client id).
            `None` means "look at the environment / credentials file".
        consumer_secret: The matching consumer secret, stored as a
            `pydantic.SecretStr` so it is never echoed by `repr(creds)`
            or in logs. `None` means same as `consumer_key`.
        credentials_file: Optional explicit path to a `key,secret`
            credentials file. `None` falls back to
            `EUMDAC_CONFIG_DIR/credentials`, then `~/.eumdac/credentials`.

    Examples:
        - All fields optional — rely on env / credentials file:
            ```python
            >>> from earthlens.eumetsat import EumetsatCredentials
            >>> creds = EumetsatCredentials()
            >>> creds.consumer_key is None and creds.consumer_secret is None
            True

            ```
        - SecretStr hides the secret in repr:
            ```python
            >>> from earthlens.eumetsat import EumetsatCredentials
            >>> creds = EumetsatCredentials(consumer_key="k", consumer_secret="topsecret")
            >>> "topsecret" in repr(creds)
            False

            ```
    """

    model_config = ConfigDict(frozen=True)

    consumer_key: str | None = None
    consumer_secret: SecretStr | None = None
    credentials_file: Path | None = None