Sentinel Hub — API reference

Sentinel Hub server-side-render data source subpackage — earthlens.sentinel_hub. Background, usage, authentication, the collection/recipe catalog, and the Statistical/Batch planes are covered on the other pages in this section; this page is the rendered API.

earthlens.sentinel_hub

Sentinel Hub server-side-render backend (defaults to Sentinel Hub on CDSE).

earthlens sends a bbox/geometry + time + an evalscript to one of Sentinel Hub's request planes (Process / Async / Batch render → raster GeoTIFF; Statistical / Batch-Statistical → tabular zonal stats); the server computes on-the-fly and earthlens collects the result. A request is variables={collection_or_recipe: [band, ...]} plus a bbox + date window, and the plane is chosen by api= (auto-selected by size + whether geometry= was given).

Public surface (re-exported from this package):

• :class:SentinelHub — the backend; instantiate with a date range, a bbox, a {collection_or_recipe: [band, ...]} request, and (optionally) evalscript, resolution, endpoint, mosaicking_order, api, geometry, then call :meth:SentinelHub.download.
• :class:Catalog — pydantic-backed loader for the bundled two-layer catalog under src/earthlens/sentinel_hub/catalog/, exposing datasets (collections), recipes, available_collections, and get_collection / get_recipe / is_recipe / resolve.
• :class:Collection / :class:EvalscriptRecipe / :class:Band / :class:Extent / :class:ResolvedRequest — the frozen value objects the catalog is built from.
• :func:read_evalscript — read a bundled .js evalscript by name.
• :class:SentinelHubAuth / :class:SentinelHubCredentials — the OAuth2 client-credentials auth wrapper and its credentials value object.
• :class:AuthenticationError — raised when no credentials are resolvable.
• :data:CATALOG_PATH / :data:EVALSCRIPTS_PATH — absolute paths to the bundled catalog directory and evalscript directory; monkey-patchable in tests.

The Sentinel Hub client ([sentinel-hub] extra) is imported lazily, so the EarthLens facade still imports without it.

AuthenticationError

Bases: AuthenticationError

Raised when the Sentinel Hub SHConfig cannot be assembled.

Wraps the missing-credentials case with an actionable message pointing at the CDSE Dashboard. A subclass of the cross-backend :class:earthlens.base.AuthenticationError so callers can catch every backend's auth failure with one except clause.

Source code in src/earthlens/sentinel_hub/auth.py

class AuthenticationError(_BaseAuthenticationError):
    """Raised when the Sentinel Hub `SHConfig` cannot be assembled.

    Wraps the missing-credentials case with an actionable message pointing at
    the CDSE Dashboard. A subclass of the cross-backend
    :class:`earthlens.base.AuthenticationError` so callers can catch every
    backend's auth failure with one `except` clause.
    """

Band

Bases: BaseModel

Per-band metadata for one band of a Sentinel Hub collection.

Frozen value object; the band name is the parent mapping key and is not repeated in the body. Every field is optional because band metadata is uneven across collections.

Attributes:

Name	Type	Description
common_name	str | None	Common name ("red", "nir"), or None.
description	str | None	Human description of the band, or None.
units	str | None	Physical unit string, or None.
resolution	float | None	Native ground sample distance in metres, or None.
center_wavelength	float | None	Central wavelength in micrometres (optical), or None.

Source code in src/earthlens/sentinel_hub/catalog.py

class Band(BaseModel):
    """Per-band metadata for one band of a Sentinel Hub collection.

    Frozen value object; the band name is the parent mapping key and is not
    repeated in the body. Every field is optional because band metadata is
    uneven across collections.

    Attributes:
        common_name: Common name (`"red"`, `"nir"`), or `None`.
        description: Human description of the band, or `None`.
        units: Physical unit string, or `None`.
        resolution: Native ground sample distance in metres, or `None`.
        center_wavelength: Central wavelength in micrometres (optical), or `None`.
    """

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

    common_name: str | None = None
    description: str | None = None
    units: str | None = None
    resolution: float | None = None
    center_wavelength: float | None = None

Catalog

Bases: AbstractCatalog

YAML-backed catalog of Sentinel Hub collections and evalscript recipes.

Reads every *.yaml under :data:CATALOG_PATH on construction and merges them into typed :class:Collection / :class:EvalscriptRecipe models. Collections are stored under the inherited :attr:datasets field (keyed by logical key); recipes live in :attr:recipes; the informational index lives in :attr:available_collections.

Attributes:

Name	Type	Description
datasets	dict[str, Collection]	Logical collection key → :class:Collection.
recipes	dict[str, EvalscriptRecipe]	Recipe key → :class:EvalscriptRecipe.
available_collections	list[str]	Every collection the backend can render (refreshed index; informational).

Examples:

• A recipe resolves to its bundled evalscript; a collection to a bare bind:

  >>> from earthlens.sentinel_hub import Catalog
  >>> cat = Catalog()
  >>> cat.is_recipe("sentinel-2-l2a-ndvi")
  True
  >>> cat.get_collection("sentinel-2-l2a").sh_collection
  'SENTINEL2_L2A'
  

• Resolving normalises both layers to one shape:

  >>> from earthlens.sentinel_hub import Catalog
  >>> r = Catalog().resolve("sentinel-2-l2a-ndvi")
  >>> r.sh_collection, r.evalscript, r.is_recipe
  ('SENTINEL2_L2A', 'ndvi.js', True)
  

Source code in src/earthlens/sentinel_hub/catalog.py

class Catalog(AbstractCatalog):
    """YAML-backed catalog of Sentinel Hub collections and evalscript recipes.

    Reads every `*.yaml` under :data:`CATALOG_PATH` on construction and merges
    them into typed :class:`Collection` / :class:`EvalscriptRecipe` models.
    Collections are stored under the inherited :attr:`datasets` field (keyed by
    logical key); recipes live in :attr:`recipes`; the informational index lives
    in :attr:`available_collections`.

    Attributes:
        datasets: Logical collection key → :class:`Collection`.
        recipes: Recipe key → :class:`EvalscriptRecipe`.
        available_collections: Every collection the backend can render
            (refreshed index; informational).

    Examples:
        - A recipe resolves to its bundled evalscript; a collection to a bare bind:
            ```python
            >>> from earthlens.sentinel_hub import Catalog
            >>> cat = Catalog()
            >>> cat.is_recipe("sentinel-2-l2a-ndvi")
            True
            >>> cat.get_collection("sentinel-2-l2a").sh_collection
            'SENTINEL2_L2A'

            ```
        - Resolving normalises both layers to one shape:
            ```python
            >>> from earthlens.sentinel_hub import Catalog
            >>> r = Catalog().resolve("sentinel-2-l2a-ndvi")
            >>> r.sh_collection, r.evalscript, r.is_recipe
            ('SENTINEL2_L2A', 'ndvi.js', True)

            ```
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    _catalog_kind: str = "Sentinel Hub catalog"

    datasets: dict[str, Collection] = Field(default_factory=dict)
    recipes: dict[str, EvalscriptRecipe] = Field(default_factory=dict)
    available_collections: list[str] = Field(default_factory=list)

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

        `Catalog()` with no args reads the bundled `catalog/` directory through
        the `(path, mtime)`-keyed cache. If the caller passed `datasets=` or
        `recipes=`, the disk read is skipped (in-memory catalogs for tests).
        The base `available_datasets` field is mirrored from
        `available_collections` so the index is discoverable through the
        `AbstractCatalog` contract, then `super().model_post_init` populates
        `catalog` from `get_catalog()`.

        Raises:
            ValueError: When auto-loading, propagates the loader's errors.
        """
        if not (self.datasets or self.recipes):
            loaded = Catalog.load()
            self.datasets = loaded.datasets
            self.recipes = loaded.recipes
            self.available_collections = loaded.available_collections
        if not self.available_datasets:
            self.available_datasets = list(self.available_collections)
        super().model_post_init(__context)

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

        Args:
            catalog_path: Catalog directory or single `*.yaml` file. Defaults to
                module-level :data:`CATALOG_PATH`.

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

        Raises:
            ValueError: Propagated from the loader.
        """
        catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
        collections, recipes, av_cols = _load_catalog_data(catalog_path)
        return cls(
            datasets=dict(collections),
            recipes=dict(recipes),
            available_collections=list(av_cols),
        )

    def get_catalog(self) -> dict[str, Collection]:
        """Return the curated collection map (logical key → :class:`Collection`)."""
        return self.datasets

    def get_collection(self, collection_key: str) -> Collection:
        """Return the :class:`Collection` for `collection_key` (did-you-mean on miss).

        Alias of the inherited :meth:`get_dataset`, named for clarity.

        Args:
            collection_key: Logical collection key.

        Returns:
            The matching :class:`Collection`.

        Raises:
            ValueError: If the key is unknown (message suggests the closest key).
        """
        return self.get_dataset(collection_key)

    def get_recipe(self, recipe_key: str) -> EvalscriptRecipe:
        """Return the :class:`EvalscriptRecipe` for `recipe_key` (did-you-mean on miss).

        Args:
            recipe_key: Recipe key.

        Returns:
            The matching :class:`EvalscriptRecipe`.

        Raises:
            ValueError: If the key is unknown (message suggests the closest key).
        """
        import difflib

        try:
            return self.recipes[recipe_key]
        except KeyError:
            close = difflib.get_close_matches(recipe_key, self.recipes, n=1)
            hint = f" Did you mean {close[0]!r}?" if close else ""
            raise ValueError(
                f"{recipe_key!r} is not a known recipe. "
                f"Known recipes: {sorted(self.recipes)}.{hint}"
            ) from None

    def is_recipe(self, key: str) -> bool:
        """Return whether `key` names a curated evalscript recipe.

        Args:
            key: A logical collection or recipe key.

        Returns:
            `True` when `key` is a recipe.
        """
        return key in self.recipes

    def resolve(self, key: str) -> ResolvedRequest:
        """Resolve a collection-or-recipe key to a uniform :class:`ResolvedRequest`.

        A recipe resolves to its bundled evalscript + bands + kind; a plain
        collection resolves with `evalscript=None` and its default bands (the
        backend then requires an explicit `evalscript=`).

        Args:
            key: A logical collection or recipe key.

        Returns:
            The normalised :class:`ResolvedRequest` the backend builds from.

        Raises:
            ValueError: If `key` is neither a known recipe nor collection
                (message suggests the closest key across both layers).

        Examples:
            - A recipe carries its evalscript + kind:
                ```python
                >>> from earthlens.sentinel_hub import Catalog
                >>> r = Catalog().resolve("sentinel-2-l2a-ndvi-stats")
                >>> r.kind, r.evalscript
                ('stats', 'ndvi_stats.js')

                ```
        """
        if key in self.recipes:
            recipe = self.recipes[key]
            return ResolvedRequest(
                key=key,
                sh_collection=recipe.base_collection,
                bands=list(recipe.bands),
                evalscript=recipe.evalscript,
                output_bands=recipe.output_bands,
                is_recipe=True,
                kind=recipe.kind,
            )
        if key in self.datasets:
            collection = self.datasets[key]
            return ResolvedRequest(
                key=key,
                sh_collection=collection.sh_collection,
                bands=collection.effective_bands,
                evalscript=None,
                is_recipe=False,
                kind="render",
            )
        import difflib

        known = sorted(set(self.recipes) | set(self.datasets))
        close = difflib.get_close_matches(key, known, n=1)
        hint = f" Did you mean {close[0]!r}?" if close else ""
        raise ValueError(
            f"{key!r} is not a known Sentinel Hub collection or recipe. "
            f"Known keys: {known}.{hint}"
        )

get_catalog()

Return the curated collection map (logical key → :class:Collection).

Source code in src/earthlens/sentinel_hub/catalog.py

def get_catalog(self) -> dict[str, Collection]:
    """Return the curated collection map (logical key → :class:`Collection`)."""
    return self.datasets

get_collection(collection_key)

Return the :class:Collection for collection_key (did-you-mean on miss).

Alias of the inherited :meth:get_dataset, named for clarity.

Parameters:

Name	Type	Description	Default
collection_key	str	Logical collection key.	required

Returns:

Type	Description
Collection	The matching :class:Collection.

Raises:

Type	Description
ValueError	If the key is unknown (message suggests the closest key).

Source code in src/earthlens/sentinel_hub/catalog.py

def get_collection(self, collection_key: str) -> Collection:
    """Return the :class:`Collection` for `collection_key` (did-you-mean on miss).

    Alias of the inherited :meth:`get_dataset`, named for clarity.

    Args:
        collection_key: Logical collection key.

    Returns:
        The matching :class:`Collection`.

    Raises:
        ValueError: If the key is unknown (message suggests the closest key).
    """
    return self.get_dataset(collection_key)

get_recipe(recipe_key)

Return the :class:EvalscriptRecipe for recipe_key (did-you-mean on miss).

Parameters:

Name	Type	Description	Default
recipe_key	str	Recipe key.	required

Returns:

Type	Description
EvalscriptRecipe	The matching :class:EvalscriptRecipe.

Raises:

Type	Description
ValueError	If the key is unknown (message suggests the closest key).

Source code in src/earthlens/sentinel_hub/catalog.py

def get_recipe(self, recipe_key: str) -> EvalscriptRecipe:
    """Return the :class:`EvalscriptRecipe` for `recipe_key` (did-you-mean on miss).

    Args:
        recipe_key: Recipe key.

    Returns:
        The matching :class:`EvalscriptRecipe`.

    Raises:
        ValueError: If the key is unknown (message suggests the closest key).
    """
    import difflib

    try:
        return self.recipes[recipe_key]
    except KeyError:
        close = difflib.get_close_matches(recipe_key, self.recipes, n=1)
        hint = f" Did you mean {close[0]!r}?" if close else ""
        raise ValueError(
            f"{recipe_key!r} is not a known recipe. "
            f"Known recipes: {sorted(self.recipes)}.{hint}"
        ) from None

is_recipe(key)

Return whether key names a curated evalscript recipe.

Parameters:

Name	Type	Description	Default
key	str	A logical collection or recipe key.	required

Returns:

Type	Description
bool	True when key is a recipe.

Source code in src/earthlens/sentinel_hub/catalog.py

def is_recipe(self, key: str) -> bool:
    """Return whether `key` names a curated evalscript recipe.

    Args:
        key: A logical collection or recipe key.

    Returns:
        `True` when `key` is a recipe.
    """
    return key in self.recipes

load(catalog_path=None) classmethod

Read the Sentinel Hub catalog from disk (cached).

Parameters:

Name	Type	Description	Default
catalog_path	Path | None	Catalog directory or single *.yaml file. Defaults to module-level :data:CATALOG_PATH.	None

Returns:

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

Raises:

Type	Description
ValueError	Propagated from the loader.

Source code in src/earthlens/sentinel_hub/catalog.py

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

    Args:
        catalog_path: Catalog directory or single `*.yaml` file. Defaults to
            module-level :data:`CATALOG_PATH`.

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

    Raises:
        ValueError: Propagated from the loader.
    """
    catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
    collections, recipes, av_cols = _load_catalog_data(catalog_path)
    return cls(
        datasets=dict(collections),
        recipes=dict(recipes),
        available_collections=list(av_cols),
    )

model_post_init(__context)

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

Catalog() with no args reads the bundled catalog/ directory through the (path, mtime)-keyed cache. If the caller passed datasets= or recipes=, the disk read is skipped (in-memory catalogs for tests). The base available_datasets field is mirrored from available_collections so the index is discoverable through the AbstractCatalog contract, then super().model_post_init populates catalog from get_catalog().

Raises:

Type	Description
ValueError	When auto-loading, propagates the loader's errors.

Source code in src/earthlens/sentinel_hub/catalog.py

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

    `Catalog()` with no args reads the bundled `catalog/` directory through
    the `(path, mtime)`-keyed cache. If the caller passed `datasets=` or
    `recipes=`, the disk read is skipped (in-memory catalogs for tests).
    The base `available_datasets` field is mirrored from
    `available_collections` so the index is discoverable through the
    `AbstractCatalog` contract, then `super().model_post_init` populates
    `catalog` from `get_catalog()`.

    Raises:
        ValueError: When auto-loading, propagates the loader's errors.
    """
    if not (self.datasets or self.recipes):
        loaded = Catalog.load()
        self.datasets = loaded.datasets
        self.recipes = loaded.recipes
        self.available_collections = loaded.available_collections
    if not self.available_datasets:
        self.available_datasets = list(self.available_collections)
    super().model_post_init(__context)

resolve(key)

Resolve a collection-or-recipe key to a uniform :class:ResolvedRequest.

A recipe resolves to its bundled evalscript + bands + kind; a plain collection resolves with evalscript=None and its default bands (the backend then requires an explicit evalscript=).

Parameters:

Name	Type	Description	Default
key	str	A logical collection or recipe key.	required

Returns:

Type	Description
ResolvedRequest	The normalised :class:ResolvedRequest the backend builds from.

Raises:

Type	Description
ValueError	If key is neither a known recipe nor collection (message suggests the closest key across both layers).

Examples:

• A recipe carries its evalscript + kind:

  >>> from earthlens.sentinel_hub import Catalog
  >>> r = Catalog().resolve("sentinel-2-l2a-ndvi-stats")
  >>> r.kind, r.evalscript
  ('stats', 'ndvi_stats.js')
  

Source code in src/earthlens/sentinel_hub/catalog.py

def resolve(self, key: str) -> ResolvedRequest:
    """Resolve a collection-or-recipe key to a uniform :class:`ResolvedRequest`.

    A recipe resolves to its bundled evalscript + bands + kind; a plain
    collection resolves with `evalscript=None` and its default bands (the
    backend then requires an explicit `evalscript=`).

    Args:
        key: A logical collection or recipe key.

    Returns:
        The normalised :class:`ResolvedRequest` the backend builds from.

    Raises:
        ValueError: If `key` is neither a known recipe nor collection
            (message suggests the closest key across both layers).

    Examples:
        - A recipe carries its evalscript + kind:
            ```python
            >>> from earthlens.sentinel_hub import Catalog
            >>> r = Catalog().resolve("sentinel-2-l2a-ndvi-stats")
            >>> r.kind, r.evalscript
            ('stats', 'ndvi_stats.js')

            ```
    """
    if key in self.recipes:
        recipe = self.recipes[key]
        return ResolvedRequest(
            key=key,
            sh_collection=recipe.base_collection,
            bands=list(recipe.bands),
            evalscript=recipe.evalscript,
            output_bands=recipe.output_bands,
            is_recipe=True,
            kind=recipe.kind,
        )
    if key in self.datasets:
        collection = self.datasets[key]
        return ResolvedRequest(
            key=key,
            sh_collection=collection.sh_collection,
            bands=collection.effective_bands,
            evalscript=None,
            is_recipe=False,
            kind="render",
        )
    import difflib

    known = sorted(set(self.recipes) | set(self.datasets))
    close = difflib.get_close_matches(key, known, n=1)
    hint = f" Did you mean {close[0]!r}?" if close else ""
    raise ValueError(
        f"{key!r} is not a known Sentinel Hub collection or recipe. "
        f"Known keys: {known}.{hint}"
    )

Collection

Bases: BaseModel

One curated Sentinel Hub data collection, addressed by a logical key.

Attributes:

Name	Type	Description
sh_collection	str	The sentinelhub.DataCollection member name this key binds to ("SENTINEL2_L2A", "SENTINEL1_IW", …).
bands	dict[str, Band]	Band name → :class:Band metadata for the bands the collection exposes.
default_bands	list[str]	Bands used when the request names none. Falls back to every key of bands when empty.
cadence	str | None	Native revisit cadence label, or None.
resolution	float | None	Native ground sample distance in metres, or None.
extent	Extent | None	Spatial/temporal coverage, or None.
description	str | None	One-line human summary, or None.

Source code in src/earthlens/sentinel_hub/catalog.py

class Collection(BaseModel):
    """One curated Sentinel Hub data collection, addressed by a logical key.

    Attributes:
        sh_collection: The `sentinelhub.DataCollection` member name this key
            binds to (`"SENTINEL2_L2A"`, `"SENTINEL1_IW"`, …).
        bands: Band name → :class:`Band` metadata for the bands the collection
            exposes.
        default_bands: Bands used when the request names none. Falls back to
            every key of `bands` when empty.
        cadence: Native revisit cadence label, or `None`.
        resolution: Native ground sample distance in metres, or `None`.
        extent: Spatial/temporal coverage, or `None`.
        description: One-line human summary, or `None`.
    """

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

    sh_collection: str
    bands: dict[str, Band] = Field(default_factory=dict)
    default_bands: list[str] = Field(default_factory=list)
    cadence: str | None = None
    resolution: float | None = None
    extent: Extent | None = None
    description: str | None = None

    @property
    def effective_bands(self) -> list[str]:
        """The band names to request by default (`default_bands`, else all bands)."""
        return list(self.default_bands or list(self.bands))

effective_bands property

The band names to request by default (default_bands, else all bands).

EvalscriptRecipe

Bases: BaseModel

One curated evalscript recipe fixing a base collection + a .js file.

Attributes:

Name	Type	Description
base_collection	str	The DataCollection member name the recipe renders (e.g. "SENTINEL2_L2A").
evalscript	str	The bundled .js filename under evalscripts/ (e.g. "ndvi.js").
bands	list[str]	The collection bands the evalscript consumes (informational + used by the refresh/validate tool).
output_bands	int	The number of bands the evalscript writes.
kind	str	"render" (writes a raster) or "stats" (emits a dataMask band for the Statistical API).
description	str | None	One-line human summary, or None.

Source code in src/earthlens/sentinel_hub/catalog.py

class EvalscriptRecipe(BaseModel):
    """One curated evalscript recipe fixing a base collection + a `.js` file.

    Attributes:
        base_collection: The `DataCollection` member name the recipe renders
            (e.g. `"SENTINEL2_L2A"`).
        evalscript: The bundled `.js` filename under `evalscripts/`
            (e.g. `"ndvi.js"`).
        bands: The collection bands the evalscript consumes (informational +
            used by the refresh/validate tool).
        output_bands: The number of bands the evalscript writes.
        kind: `"render"` (writes a raster) or `"stats"` (emits a `dataMask` band
            for the Statistical API).
        description: One-line human summary, or `None`.
    """

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

    base_collection: str
    evalscript: str
    bands: list[str] = Field(default_factory=list)
    output_bands: int = 1
    kind: str = "render"
    description: str | None = None

    @field_validator("kind")
    @classmethod
    def _check_kind(cls, value: str) -> str:
        """Validate `kind` is one of the recognised recipe kinds.

        Args:
            value: The recipe kind.

        Returns:
            The validated kind.

        Raises:
            ValueError: If `kind` is not `"render"` or `"stats"`.
        """
        if value not in _RECIPE_KINDS:
            raise ValueError(
                f"recipe kind must be one of {sorted(_RECIPE_KINDS)}, got {value!r}."
            )
        return value

Extent

Bases: BaseModel

Spatial/temporal coverage of a Sentinel Hub collection.

Attributes:

Name	Type	Description
start_date	str | None	First available date (YYYY-MM-DD), or None.
end_date	str | None	Last available date, or None for a rolling collection.
bbox	tuple[float, float, float, float] | None	[west, south, east, north] in EPSG:4326, or None for global.

Source code in src/earthlens/sentinel_hub/catalog.py

class Extent(BaseModel):
    """Spatial/temporal coverage of a Sentinel Hub collection.

    Attributes:
        start_date: First available date (`YYYY-MM-DD`), or `None`.
        end_date: Last available date, or `None` for a rolling collection.
        bbox: `[west, south, east, north]` in EPSG:4326, or `None` for global.
    """

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

    start_date: str | None = None
    end_date: str | None = None
    bbox: tuple[float, float, float, float] | None = None

ResolvedRequest

Bases: BaseModel

The uniform shape a resolved collection-or-recipe key takes.

Both a plain collection and a recipe resolve to this so the backend's request-builder consumes one type. A plain collection resolves with evalscript=None (the backend must then have an explicit evalscript=); a recipe carries its bundled .js filename.

Attributes:

Name	Type	Description
key	str	The logical key the request named.
sh_collection	str	The DataCollection member name to bind + render.
bands	list[str]	The bands to request (request override applied by the backend).
evalscript	str | None	The bundled .js filename, or None for a plain collection.
output_bands	int	Number of output bands the evalscript writes.
is_recipe	bool	Whether the key named a recipe (vs a plain collection).
kind	str	"render" or "stats" (recipes only; "render" for a plain collection).

Source code in src/earthlens/sentinel_hub/catalog.py

class ResolvedRequest(BaseModel):
    """The uniform shape a resolved collection-or-recipe key takes.

    Both a plain collection and a recipe resolve to this so the backend's
    request-builder consumes one type. A plain collection resolves with
    `evalscript=None` (the backend must then have an explicit `evalscript=`); a
    recipe carries its bundled `.js` filename.

    Attributes:
        key: The logical key the request named.
        sh_collection: The `DataCollection` member name to bind + render.
        bands: The bands to request (request override applied by the backend).
        evalscript: The bundled `.js` filename, or `None` for a plain collection.
        output_bands: Number of output bands the evalscript writes.
        is_recipe: Whether the key named a recipe (vs a plain collection).
        kind: `"render"` or `"stats"` (recipes only; `"render"` for a plain
            collection).
    """

    model_config = ConfigDict(frozen=True)

    key: str
    sh_collection: str
    bands: list[str] = Field(default_factory=list)
    evalscript: str | None = None
    output_bands: int = 1
    is_recipe: bool = False
    kind: str = "render"

SentinelHub

Bases: AbstractDataSource

Server-side Sentinel Hub backend on CDSE (raster or tabular by api=).

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	"mixed" — raster (Process/Async/Batch) or tabular (Statistical/Batch-Statistical) depending on the resolved api=. A fixed class attribute (never mutated per-instance) so the facade forwards aggregate= on every plane.

Source code in src/earthlens/sentinel_hub/backend.py

class SentinelHub(AbstractDataSource):
    """Server-side Sentinel Hub backend on CDSE (raster or tabular by `api=`).

    Attributes:
        OUTPUT_KIND: `"mixed"` — raster (Process/Async/Batch) or tabular
            (Statistical/Batch-Statistical) depending on the resolved `api=`.
            A *fixed* class attribute (never mutated per-instance) so the facade
            forwards `aggregate=` on every plane.
    """

    OUTPUT_KIND: OutputKind = "mixed"

    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 = "",
        fmt: str = "%Y-%m-%d",
        resolution: float = 10.0,
        evalscript: str | None = None,
        endpoint: str | None = None,
        mosaicking_order: str = _DEFAULT_MOSAICKING_ORDER,
        api: str | None = None,
        geometry: Any = None,
        maxcc: float | None = None,
        batch_output: dict[str, Any] | None = None,
        client_id: str | None = None,
        client_secret: str | None = None,
        profile: str | None = None,
    ):
        """Initialise a Sentinel Hub backend instance.

        Args:
            start: Inclusive start date string (parsed with `fmt`).
            end: Inclusive end date string.
            variables: `{collection_or_recipe_key: [band, ...]}`. A key is a
                catalog **collection** (needs an explicit `evalscript=`) or an
                **evalscript recipe** (pins its collection); an empty band list
                falls back to the recipe / collection default bands.
            lat_lim: `[lat_min, lat_max]` in degrees.
            lon_lim: `[lon_min, lon_max]` in degrees.
            temporal_resolution: Advisory cadence label; the render window is
                `start`/`end`.
            path: Output directory (created by the parent class).
            fmt: `strptime` format for `start` / `end`.
            resolution: Output pixel size in metres (→ `bbox_to_dimensions`).
            evalscript: A custom evalscript — an inline V3 JS string or a path to
                a `.js` file — that bypasses the recipe lookup (the collection
                then comes from the `variables` key). `None` uses the recipe's
                bundled `.js`.
            endpoint: Endpoint alias (`"cdse"`, `"commercial"`) or a full base
                URL. Defaults to CDSE-free.
            mosaicking_order: Per-pixel scene selection passed to
                `input_data` — `"mostRecent"` (default) / `"leastRecent"` /
                `"leastCC"` (the `MosaickingOrder` enum).
            api: The request plane (`"process"` / `"async"` / `"batch"` /
                `"statistical"` / `"batch-statistical"`), or `None` to
                auto-select by request size + whether `geometry=` was supplied.
            geometry: A shapely geometry / GeoJSON mapping / `FeatureCollection`
                for the Statistical planes (zonal stats over the polygon(s)).
            maxcc: Optional maximum cloud cover (0–1) passed to `input_data`
                (optical collections only).
            batch_output: S3 delivery spec (`{"bucket": ..., "iam_role_arn": ...}`)
                for the Batch planes.
            client_id: OAuth client id (else `SENTINELHUB_CLIENT_ID`).
            client_secret: OAuth client secret (else `SENTINELHUB_CLIENT_SECRET`).
            profile: A saved `SHConfig` profile name (else `SENTINELHUB_PROFILE`).

        Raises:
            ValueError: When `api` is an unknown plane, or `mosaicking_order` is
                not a recognised value.
        """
        validate_api(api)
        if mosaicking_order not in _VALID_MOSAICKING_ORDERS:
            raise ValueError(
                f"mosaicking_order must be one of {list(_VALID_MOSAICKING_ORDERS)}, "
                f"got {mosaicking_order!r}."
            )
        self._resolution = resolution
        self._evalscript = evalscript
        self._endpoint = endpoint
        self._mosaicking_order = mosaicking_order
        self._api_mode = api
        self._geometry = geometry
        self._maxcc = maxcc
        self._batch_output = batch_output
        env_creds = SentinelHubCredentials.from_env()
        self._credentials = SentinelHubCredentials(
            client_id=client_id or env_creds.client_id,
            client_secret=client_secret or env_creds.client_secret,
            profile=profile or env_creds.profile,
        )
        # Stored before super().__init__ because the base constructor calls
        # _initialize() (which needs the request) before it sets self.vars.
        self._variables = variables
        self._catalog: Catalog | None = None
        self._auth: SentinelHubAuth | None = None
        self._resolved: dict[str, ResolvedRequest] = {}
        # Aggregation request captured by download(); applied per plane.
        self._aggregate: AggregationConfig | None = None
        # Per-window time override set by the aggregate= render loop (C10).
        self._window_override: tuple[str, str] | None = None
        # Memoised resolved plane (deterministic from instance state; avoids a
        # second bbox_to_dimensions call across download() -> _fetch()).
        self._plane: str | None = None
        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 _initialize(self) -> None:
        """Load the catalog, resolve the requested keys, and build the auth.

        Returns `None` so the parent does not bind `self.client`; the auth
        wrapper is stored on `self._auth` and the resolved rows on
        `self._resolved`. Token minting is deferred to first request.

        Raises:
            TypeError: When `variables` is not a
                `{collection_or_recipe: [band, ...]}` mapping.
            ValueError: When `variables` is empty or names a key the catalog
                does not know (with a did-you-mean hint).
        """
        from earthlens.sentinel_hub.catalog import Catalog

        if not isinstance(self._variables, dict):
            raise TypeError(
                "Sentinel Hub requires variables as a "
                "{collection_or_recipe: [band, ...]} mapping, got "
                f"{type(self._variables).__name__}."
            )
        if not self._variables:
            raise ValueError(
                "Sentinel Hub requires variables={collection_or_recipe: "
                "[band, ...]} with at least one collection or recipe key."
            )
        self._catalog = Catalog()
        self._resolved = {key: self._catalog.resolve(key) for key in self._variables}
        self._auth = SentinelHubAuth(self._credentials, endpoint=self._endpoint)
        return None

    def _create_grid(self, lat_lim: list, lon_lim: list) -> SpatialExtent:
        """Build the WGS84 envelope the Sentinel Hub `BBox` is built from.

        Args:
            lat_lim: `[lat_min, lat_max]` in degrees.
            lon_lim: `[lon_min, lon_max]` in degrees.

        Returns:
            SpatialExtent: The request's bounding box.
        """
        return SpatialExtent.from_pairs(lat_lim=lat_lim, lon_lim=lon_lim)

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

        Args:
            start: Inclusive start date string (parsed with `fmt`).
            end: Inclusive end date string.
            temporal_resolution: Advisory cadence label (`"daily"`, `"monthly"`,
                `"hourly"`, `"yearly"`) → the pandas frequency on the extent.
            fmt: `strptime` format for `start` / `end`.

        Returns:
            TemporalExtent: The parsed window (inclusive `end_date`).

        Raises:
            ValueError: When `start` or `end` is `None`.
        """
        import datetime as dt

        import pandas as pd

        if start is None or end is None:
            raise ValueError(
                "Sentinel Hub requires both start and end dates (the render "
                "needs a time_interval); pass start=… and end=…."
            )
        start_dt = dt.datetime.strptime(start, fmt)
        end_dt = dt.datetime.strptime(end, fmt)
        freq_map = {"daily": "D", "monthly": "MS", "hourly": "h", "yearly": "YS"}
        resolution = freq_map.get(temporal_resolution, "D")
        dates = pd.date_range(start_dt, end_dt, freq=resolution)
        return TemporalExtent(
            start_date=start_dt, end_date=end_dt, resolution=resolution, dates=dates
        )

    def _bbox(self) -> Any:
        """Build the Sentinel Hub `BBox` from the request envelope (WGS84).

        Returns:
            A `sentinelhub.BBox` over the request bbox.
        """
        sentinelhub = import_sentinelhub()
        return sentinelhub.BBox(
            (self.space.west, self.space.south, self.space.east, self.space.north),
            crs=sentinelhub.CRS.WGS84,
        )

    def _request_size(self) -> tuple[int, int]:
        """Compute the render size in pixels via `bbox_to_dimensions`.

        Returns:
            `(width_px, height_px)` for the request bbox at `resolution`.
        """
        sentinelhub = import_sentinelhub()
        return sentinelhub.bbox_to_dimensions(self._bbox(), resolution=self._resolution)

    def _resolve_plane(self) -> str:
        """Resolve the request plane: explicit `api=`, else auto by size / geometry.

        Memoised: the plane is deterministic from instance state, so the result
        is computed once (one `bbox_to_dimensions` call) and reused across the
        `download` -> `_fetch` path.

        Returns:
            The resolved plane name.
        """
        if self._plane is None:
            has_geometry = self._geometry is not None
            has_s3 = self._batch_output is not None
            needs_size = self._api_mode is None or self._api_mode in RASTER_APIS
            max_side = max(self._request_size()) if needs_size else 0
            self._plane = resolve_api(self._api_mode, max_side, has_geometry, has_s3)
        return self._plane

    def _time_interval(self) -> tuple[str, str]:
        """Return the render time interval as ISO `(start, end)` date strings.

        Honours a per-window override set by the `aggregate=` render loop (C10);
        otherwise the full request window.
        """
        if self._window_override is not None:
            return self._window_override
        return (
            self.time.start_date.strftime("%Y-%m-%d"),
            self.time.end_date.strftime("%Y-%m-%d"),
        )

    def _api(self) -> list[Path]:
        """Compose `_search` and `_fetch` into the canonical search/fetch shape."""
        return self._api_via_search_fetch()

    def _search(self) -> list[RemoteProduct]:
        """List the planned requests without rendering (cheap dry-run).

        Returns one :class:`RemoteProduct` per requested key; the resolved row
        rides on `metadata["resolved"]`. No network call.

        Returns:
            One product per requested collection/recipe key.
        """
        return [
            RemoteProduct(id=key, metadata={"resolved": resolved})
            for key, resolved in self._resolved.items()
        ]

    def search(self, limit: int = 100) -> list[RemoteProduct]:
        """Query the Sentinel Hub Catalog API for scenes intersecting the request.

        A real STAC-style search (distinct from the internal :meth:`_search`
        fetch-planner): returns one :class:`RemoteProduct` per catalog item that
        intersects the request bbox + window, so a caller can enumerate coverage
        before rendering. An empty result is an empty list (not an error).

        Args:
            limit: Maximum number of items to return per requested collection.

        Returns:
            One product per catalog item (id + datetime + geometry on metadata).
        """
        sentinelhub = import_sentinelhub()
        cfg = self._auth.config()
        catalog = sentinelhub.SentinelHubCatalog(config=cfg)
        sh_bbox = self._bbox()
        time_interval = self._time_interval()
        products: list[RemoteProduct] = []
        seen_collections: set[str] = set()
        for key, resolved in self._resolved.items():
            if resolved.sh_collection in seen_collections:
                continue
            seen_collections.add(resolved.sh_collection)
            collection = cdse_collection(resolved.sh_collection, cfg.sh_base_url)
            for item in catalog.search(
                collection, bbox=sh_bbox, time=time_interval, limit=limit
            ):
                properties = item.get("properties", {})
                products.append(
                    RemoteProduct(
                        id=item.get("id"),
                        metadata={
                            "key": key,
                            "collection": resolved.sh_collection,
                            "datetime": properties.get("datetime"),
                            "geometry": item.get("geometry"),
                        },
                    )
                )
        return products

    def _fetch(self, products: list[RemoteProduct]) -> list[Any]:
        """Dispatch to the per-plane fetcher for the resolved `api`.

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            The written raster paths (raster planes) or table paths / S3 URIs
            (tabular / batch planes).
        """
        plane = self._resolve_plane()
        fetchers = {
            "process": self._fetch_process,
            "async": self._fetch_async,
            "tiling": self._fetch_tiling,
            "batch": self._fetch_batch,
            "statistical": self._fetch_statistical,
            "batch-statistical": self._fetch_batch_statistical,
        }
        return fetchers[plane](products)

    def _read_evalscript(self, resolved: ResolvedRequest) -> str:
        """Resolve the evalscript source for one request row.

        A custom `evalscript=` (an inline V3 JS string or a `.js` file path)
        wins over the recipe. Otherwise the recipe's bundled `.js` is read; a
        plain collection (no recipe evalscript and no custom one) is an error.

        Args:
            resolved: The resolved request row.

        Returns:
            The evalscript source string.

        Raises:
            ValueError: When the key is a plain collection and no `evalscript=`
                was supplied.
        """
        if self._evalscript is not None:
            candidate = self._evalscript
            try:
                path = Path(candidate)
                if path.is_file():
                    return path.read_text(encoding="utf-8")
            except OSError:
                pass
            return candidate
        if resolved.evalscript is None:
            raise ValueError(
                f"{resolved.key!r} is a plain collection, so it has no bundled "
                "evalscript. Pass evalscript= (an inline V3 JS string or a .js "
                "path), or request an evalscript recipe key instead."
            )
        return read_evalscript(resolved.evalscript)

    def _guard_process_size(self, size: tuple[int, int]) -> None:
        """Reject a Process request whose render exceeds the 2500 px cap.

        Args:
            size: The `(width, height)` render size in pixels.

        Raises:
            ValueError: When either side exceeds :data:`SH_MAX_DIMENSION`.
        """
        if max(size) > SH_MAX_DIMENSION:
            raise ValueError(
                f"the request renders to {size} px but the Process API caps a "
                f"single request at {SH_MAX_DIMENSION} px/side. Omit api= to "
                "auto-route to async / batch, or lower the resolution."
            )

    def _build_input_data(self, sentinelhub: Any, resolved: ResolvedRequest) -> dict:
        """Build the `SentinelHubRequest.input_data` block for one row.

        Args:
            sentinelhub: The imported `sentinelhub` module.
            resolved: The resolved request row.

        Returns:
            The `input_data` dict (collection, window, mosaicking order, maxcc).
        """
        cfg = self._auth.config()
        collection = cdse_collection(resolved.sh_collection, cfg.sh_base_url)
        kwargs: dict[str, Any] = {
            "data_collection": collection,
            "time_interval": self._time_interval(),
            "mosaicking_order": self._mosaicking_order,
        }
        if self._maxcc is not None:
            kwargs["maxcc"] = self._maxcc
        return sentinelhub.SentinelHubRequest.input_data(**kwargs)

    def _fetch_process(self, products: list[RemoteProduct]) -> list[Path]:
        """Render each product synchronously via the Process API → GeoTIFF on disk.

        Builds one `SentinelHubRequest` per resolved row over the request bbox +
        window at `resolution`, writes the rendered GeoTIFF under `path`, and
        returns the written file paths (resolved from the SDK rather than
        re-hashing).

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            The written GeoTIFF paths, one per product.

        Raises:
            ValueError: When the render exceeds the Process 2500 px cap.
        """
        sentinelhub = import_sentinelhub()
        sh_bbox = self._bbox()
        size = self._request_size()
        self._guard_process_size(size)
        out: list[Path] = []
        for product in products:
            resolved: ResolvedRequest = product.metadata["resolved"]
            out.append(
                self._render_process_tile(
                    sentinelhub, resolved, sh_bbox, size, str(self.root_dir)
                )
            )
        return out

    def _render_process_tile(
        self,
        sentinelhub: Any,
        resolved: ResolvedRequest,
        bbox: Any,
        size: tuple[int, int],
        data_folder: str,
    ) -> Path:
        """Build + run one Process request and return the written GeoTIFF path.

        Shared by :meth:`_fetch_process` (whole bbox) and :meth:`_fetch_tiling`
        (per tile).

        Args:
            sentinelhub: The imported `sentinelhub` module.
            resolved: The resolved request row.
            bbox: The `sentinelhub.BBox` to render.
            size: The `(width, height)` render size for this bbox.
            data_folder: The directory the SDK writes the GeoTIFF under.

        Returns:
            The written GeoTIFF path.
        """
        request = sentinelhub.SentinelHubRequest(
            evalscript=self._read_evalscript(resolved),
            input_data=[self._build_input_data(sentinelhub, resolved)],
            responses=[
                sentinelhub.SentinelHubRequest.output_response(
                    "default", sentinelhub.MimeType.TIFF
                )
            ],
            bbox=bbox,
            size=size,
            data_folder=data_folder,
            config=self._auth.config(),
        )
        request.get_data(save_data=True)
        return Path(data_folder) / request.get_filename_list()[0]

    def _require_s3_delivery(self, sentinelhub: Any) -> Any:
        """Build the S3 `AccessSpecification` from `batch_output`, or error.

        Args:
            sentinelhub: The imported `sentinelhub` module.

        Returns:
            The S3 delivery `AccessSpecification`.

        Raises:
            ValueError: When no `batch_output` was supplied, or it carries no
                `bucket` / `url`.
        """
        if not self._batch_output:
            raise ValueError(
                "the async / batch planes deliver server-side to S3, so they "
                "need batch_output={'bucket': 's3://…', 'iam_role_arn': '…'}. "
                "Omit api= (or pass api='tiling') for a no-S3 oversized render."
            )
        spec = dict(self._batch_output)
        url = spec.pop("bucket", None) or spec.pop("url", None)
        if not url:
            raise ValueError(
                "batch_output must include a 'bucket' (or 'url') S3 destination, "
                f"got {dict(self._batch_output)!r}."
            )
        return sentinelhub.AsyncProcessRequest.s3_specification(url=url, **spec)

    def _guard_async_size(self, size: tuple[int, int]) -> None:
        """Reject an Async request beyond the 10000 px ceiling.

        Args:
            size: The `(width, height)` render size in pixels.

        Raises:
            ValueError: When either side exceeds :data:`ASYNC_MAX_DIMENSION`.
        """
        if max(size) > ASYNC_MAX_DIMENSION:
            raise ValueError(
                f"the request renders to {size} px but the Async Processing API "
                f"caps a request at {ASYNC_MAX_DIMENSION} px/side. Use api='batch' "
                "for a larger AOI."
            )

    def _fetch_async(self, products: list[RemoteProduct]) -> list[str]:
        """Render via the Async Processing API (S3-delivered, ≤10000 px).

        Submits one `AsyncProcessRequest` per row delivering to the `batch_output`
        S3 bucket, polls `get_async_running_status` to completion, and returns the
        S3 delivery URIs. Requires an S3 `batch_output` (the SDK's async plane is
        not a direct synchronous download).

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            The S3 **delivery prefix** (the configured `batch_output` bucket), one
            per product — not the individual written object keys (enumerate the
            bucket for those; earthlens does not perform S3 listing).

        Raises:
            ValueError: When no `batch_output` is set or the render exceeds the
                10000 px Async ceiling.
        """
        sentinelhub = import_sentinelhub()
        cfg = self._auth.config()
        delivery = self._require_s3_delivery(sentinelhub)
        bucket_uri = self._batch_output.get("bucket") or self._batch_output.get("url")
        sh_bbox = self._bbox()
        size = self._request_size()
        self._guard_async_size(size)
        out: list[str] = []
        for product in products:
            resolved: ResolvedRequest = product.metadata["resolved"]
            evalscript = self._read_evalscript(resolved)
            request = sentinelhub.AsyncProcessRequest(
                evalscript=evalscript,
                input_data=[self._build_input_data(sentinelhub, resolved)],
                responses=[
                    sentinelhub.AsyncProcessRequest.output_response(
                        "default", sentinelhub.MimeType.TIFF
                    )
                ],
                delivery=delivery,
                bbox=sh_bbox,
                size=size,
                config=cfg,
            )
            # AsyncProcessRequest.get_data submits the job and returns the
            # submission JSON, which carries the async request id; poll on that
            # id (NOT the delivery URL — get_async_running_status resolves
            # `…/async/process/{id}`).
            request_id = _async_request_id(request.get_data(save_data=False))
            if request_id is not None:
                _wait_for_async(sentinelhub, [request_id], cfg)
            else:
                logger.warning(
                    "Sentinel Hub async: could not determine the request id from "
                    "the submission response; skipping the completion poll."
                )
            out.append(str(bucket_uri))
        logger.info(f"Sentinel Hub async: delivered {len(out)} object(s) to S3")
        return out

    def _fetch_tiling(self, products: list[RemoteProduct]) -> list[Path]:
        """Render an oversized AOI by local tiling + mosaic (no S3 needed).

        Splits the request bbox into ≤2500 px Process tiles, renders each tile,
        and mosaics them into one GeoTIFF per product with
        `pyramids.dataset.merge.merge_rasters`. Tile temporaries are written
        under a per-product subdirectory and removed after the merge.

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            The merged GeoTIFF paths, one per product (union bounds).

        Raises:
            ValueError: When a tile still renders above the Process cap (a
                rounding edge case); lower the `resolution` to re-tile finer.
        """
        import shutil

        from pyramids.dataset.merge import merge_rasters

        sentinelhub = import_sentinelhub()
        width, height = self._request_size()
        tiles = tile_bbox(
            (self.space.west, self.space.south, self.space.east, self.space.north),
            width,
            height,
        )
        # Pre-flight: size every tile once and guard against the Process cap
        # before rendering any, so a rounding edge case fails fast with a clear
        # error rather than a mid-run server 500 (and no partial tiles leak).
        tile_specs: list[tuple[Any, tuple[int, int]]] = []
        for tile in tiles:
            sh_bbox = sentinelhub.BBox(tile, crs=sentinelhub.CRS.WGS84)
            tile_size = sentinelhub.bbox_to_dimensions(
                sh_bbox, resolution=self._resolution
            )
            if max(tile_size) > SH_MAX_DIMENSION:
                raise ValueError(
                    f"a local tile renders to {tile_size} px, exceeding the "
                    f"{SH_MAX_DIMENSION} px Process cap; lower the resolution to "
                    "re-tile finer."
                )
            tile_specs.append((sh_bbox, tile_size))
        out: list[Path] = []
        for product in products:
            resolved: ResolvedRequest = product.metadata["resolved"]
            tile_dir = Path(self.root_dir) / f"_tiles_{_safe_name(product.id)}"
            tile_dir.mkdir(parents=True, exist_ok=True)
            tile_paths: list[str] = []
            for index, (sh_bbox, tile_size) in enumerate(tile_specs):
                rendered = self._render_process_tile(
                    sentinelhub,
                    resolved,
                    sh_bbox,
                    tile_size,
                    str(tile_dir / str(index)),
                )
                tile_paths.append(str(rendered))
            merged = Path(self.root_dir) / f"{_safe_name(product.id)}.tif"
            merge_rasters(tile_paths, str(merged))
            shutil.rmtree(tile_dir, ignore_errors=True)
            out.append(merged)
        logger.info(f"Sentinel Hub tiling: merged {len(tiles)} tile(s) per product")
        return out

    def _fetch_batch(self, products: list[RemoteProduct]) -> list[str]:
        """Render a very large AOI via the Batch Processing API, tiled to S3.

        Builds the base Process request, then creates a batch request with a
        server-side tiling grid delivering to the `batch_output` S3 bucket,
        runs analysis → start → monitor, and returns the S3 destination URIs.
        Requires an S3 `batch_output` with at least a bucket; `grid_id` selects
        the tiling grid (default 0).

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            The S3 **delivery prefix** (the configured `batch_output` bucket), one
            per product — the server tiles the output under that prefix; enumerate
            the bucket for the individual tile object keys (earthlens does not
            perform S3 listing).

        Raises:
            ValueError: When no `batch_output` (S3 bucket) was supplied, or the
                analysed `cost_PU` exceeds `batch_output['max_cost_pu']`.
        """
        sentinelhub = import_sentinelhub()
        if not self._batch_output:
            raise ValueError(
                "the Batch Processing plane tiles server-side to S3, so it needs "
                "batch_output={'bucket': 's3://…', 'iam_role_arn': '…', "
                "'grid_id': <int>}."
            )
        cfg = self._auth.config()
        client = sentinelhub.BatchProcessClient(config=cfg)
        spec = dict(self._batch_output)
        grid_id = spec.pop("grid_id", 0)
        buffer_x = spec.pop("buffer_x", None)
        buffer_y = spec.pop("buffer_y", None)
        max_cost_pu = spec.pop("max_cost_pu", None)
        url = spec.pop("bucket", None) or spec.pop("url", None)
        if not url:
            raise ValueError(
                "batch_output must include a 'bucket' (or 'url') S3 destination, "
                f"got {dict(self._batch_output)!r}."
            )
        delivery = client.s3_specification(url=url, **spec)
        tiling = client.tiling_grid_input(
            grid_id=grid_id,
            resolution=self._resolution,
            buffer_x=buffer_x,
            buffer_y=buffer_y,
        )
        output = client.raster_output(delivery=delivery)
        sh_bbox = self._bbox()
        out: list[str] = []
        for product in products:
            resolved: ResolvedRequest = product.metadata["resolved"]
            base = sentinelhub.SentinelHubRequest(
                evalscript=self._read_evalscript(resolved),
                input_data=[self._build_input_data(sentinelhub, resolved)],
                responses=[
                    sentinelhub.SentinelHubRequest.output_response(
                        "default", sentinelhub.MimeType.TIFF
                    )
                ],
                bbox=sh_bbox,
                config=cfg,
            )
            batch_request = client.create(base, input=tiling, output=output)
            # Analyse first so the tile count / cost is known before committing
            # the (potentially continental, costly) job. start_analysis only
            # *starts* the analysis phase, so wait for it to finish before
            # reading cost_PU (otherwise the guard below sees None and is a no-op).
            client.start_analysis(batch_request)
            sentinelhub.monitor_batch_process_analysis(batch_request, client)
            batch_request = client.get_request(batch_request)
            cost_pu = getattr(batch_request, "cost_PU", None)
            logger.info(
                f"Sentinel Hub batch: analysed {product.id!r} (cost_PU={cost_pu})"
            )
            if (
                max_cost_pu is not None
                and cost_pu is not None
                and cost_pu > max_cost_pu
            ):
                raise ValueError(
                    f"batch analysis estimates cost_PU={cost_pu} for {product.id!r}, "
                    f"which exceeds batch_output['max_cost_pu']={max_cost_pu}; raise "
                    "the limit or shrink the request/resolution."
                )
            client.start_job(batch_request)
            sentinelhub.monitor_batch_process_job(batch_request, client)
            out.append(str(url))
        logger.info(f"Sentinel Hub batch: {len(out)} job(s) delivered to S3")
        return out

    def _statistical_evalscript(self, resolved: ResolvedRequest) -> str:
        """Resolve the evalscript for a Statistical request (must emit `dataMask`).

        Args:
            resolved: The resolved request row.

        Returns:
            The evalscript source.

        Raises:
            ValueError: When the evalscript does not declare a `dataMask` band
                (the Statistical API needs it to exclude invalid pixels).
        """
        script = self._read_evalscript(resolved)
        if "dataMask" not in script:
            raise ValueError(
                f"the Statistical API requires the evalscript to emit a "
                f"'dataMask' output band; {resolved.key!r} does not. Use a "
                "stats recipe (kind='stats', e.g. '…-ndvi-stats') or add a "
                "dataMask band to your custom evalscript."
            )
        return script

    def _statistical_interval(self) -> str:
        """The Statistical `aggregation_interval`: from `aggregate=`, else `P1D`."""
        if self._aggregate is not None:
            return interval_for(self._aggregate.freq)
        return "P1D"

    def _fetch_statistical(self, products: list[RemoteProduct]) -> list[Path]:
        """Compute zonal statistics via the Statistical API → a tidy table.

        Builds one `SentinelHubStatistical` request per geometry per product,
        flattens the nested interval→output→band→stats tree into rows, and writes
        one CSV per product (`feature_id` carried for a `FeatureCollection`).

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            The written table paths, one per product.

        Raises:
            ValueError: When no `geometry=` was supplied, or the evalscript lacks
                a `dataMask` band.
        """
        sentinelhub = import_sentinelhub()
        if self._geometry is None:
            raise ValueError(
                "the Statistical API computes zonal stats over a polygon, so it "
                "needs geometry= (a shapely geometry, a GeoJSON mapping, or a "
                "FeatureCollection)."
            )
        cfg = self._auth.config()
        interval = self._statistical_interval()
        geometries = _iter_geometries(self._geometry)
        out: list[Path] = []
        for product in products:
            resolved: ResolvedRequest = product.metadata["resolved"]
            evalscript = self._statistical_evalscript(resolved)
            collection = cdse_collection(resolved.sh_collection, cfg.sh_base_url)
            rows: list[dict] = []
            for feature_id, geom in geometries:
                # Size the sampling grid in pixels from the geometry's WGS84
                # bounds via bbox_to_dimensions (which converts metres -> px with
                # latitude). Passing `resolution=` here would be read in the
                # geometry CRS units (degrees), so a metre value collapses to a
                # single pixel and the server rejects the effective resolution.
                geom_bbox = sentinelhub.BBox(
                    _geometry_bounds(geom), crs=sentinelhub.CRS.WGS84
                )
                size = sentinelhub.bbox_to_dimensions(
                    geom_bbox, resolution=self._resolution
                )
                aggregation = sentinelhub.SentinelHubStatistical.aggregation(
                    evalscript=evalscript,
                    time_interval=self._time_interval(),
                    aggregation_interval=interval,
                    size=size,
                )
                input_kwargs: dict[str, Any] = {}
                if self._maxcc is not None:
                    input_kwargs["maxcc"] = self._maxcc
                request = sentinelhub.SentinelHubStatistical(
                    aggregation=aggregation,
                    input_data=[
                        sentinelhub.SentinelHubStatistical.input_data(
                            collection, **input_kwargs
                        )
                    ],
                    geometry=sentinelhub.Geometry(geom, crs=sentinelhub.CRS.WGS84),
                    calculations=_STAT_CALCULATIONS,
                    config=cfg,
                )
                payload = request.get_data()[0]
                rows.extend(_flatten_statistics(payload, feature_id=feature_id))
            if not rows:
                logger.warning(
                    f"Sentinel Hub statistical: no data returned for "
                    f"{product.id!r} over the request geometry + window "
                    "(empty table written)."
                )
            target = Path(self.root_dir) / f"{_safe_name(product.id)}.csv"
            _stats_frame(rows).to_csv(target, index=False)
            out.append(target)
        logger.info(f"Sentinel Hub statistical: wrote {len(out)} table(s)")
        return out

    def _fetch_batch_statistical(self, products: list[RemoteProduct]) -> list[Path]:
        """Compute zonal stats over a huge `FeatureCollection` via Batch Statistical.

        Submits one async batch-statistical job per product against the features
        uploaded to S3 (`batch_output['input_features']`), monitors it, retrieves
        the per-feature JSON from S3 via `AwsBatchStatisticalResults`, flattens it
        (reusing the C7 flattener, keyed by `feature_id`), and writes one CSV per
        product. Requires `batch_output` with an `input_features` S3 GeoPackage
        and an output bucket.

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            The written table paths, one per product.

        Raises:
            ValueError: When `batch_output` is missing its `input_features` or
                output bucket, or the evalscript lacks a `dataMask` band.
        """
        from sentinelhub.aws import AwsBatchStatisticalResults

        sentinelhub = import_sentinelhub()
        if not self._batch_output:
            raise ValueError(
                "the Batch Statistical plane runs over a FeatureCollection on S3, "
                "so it needs batch_output={'input_features': 's3://…features.gpkg', "
                "'bucket': 's3://…out', 'iam_role_arn': '…'}."
            )
        spec = dict(self._batch_output)
        features_url = spec.pop("input_features", None)
        output_url = spec.pop("bucket", None) or spec.pop("output", None)
        feature_ids = spec.pop("feature_ids", None)
        if not features_url or not output_url:
            raise ValueError(
                "batch-statistical needs batch_output['input_features'] (the S3 "
                "GeoPackage of features) and an output bucket."
            )
        cfg = self._auth.config()
        interval = self._statistical_interval()
        client = sentinelhub.SentinelHubBatchStatistical(config=cfg)
        out: list[Path] = []
        for product in products:
            resolved: ResolvedRequest = product.metadata["resolved"]
            evalscript = self._statistical_evalscript(resolved)
            aggregation = sentinelhub.SentinelHubStatistical.aggregation(
                evalscript=evalscript,
                time_interval=self._time_interval(),
                aggregation_interval=interval,
                resolution=(self._resolution, self._resolution),
            )
            input_kwargs: dict[str, Any] = {}
            if self._maxcc is not None:
                input_kwargs["maxcc"] = self._maxcc
            batch_request = client.create(
                input_features=client.s3_specification(url=features_url, **spec),
                input_data=[
                    sentinelhub.SentinelHubStatistical.input_data(
                        cdse_collection(resolved.sh_collection, cfg.sh_base_url),
                        **input_kwargs,
                    )
                ],
                aggregation=aggregation,
                calculations=_STAT_CALCULATIONS,
                output=client.s3_specification(url=output_url, **spec),
            )
            client.start_analysis(batch_request)
            client.start_job(batch_request)
            sentinelhub.monitor_batch_statistical_job(batch_request, cfg)
            results = AwsBatchStatisticalResults(
                batch_request,
                feature_ids=feature_ids,
                data_folder=str(self.root_dir),
                config=cfg,
            )
            payloads = results.get_data(save_data=True)
            rows: list[dict] = []
            ids = feature_ids if feature_ids is not None else range(len(payloads))
            for feature_id, payload in zip(ids, payloads):
                rows.extend(_flatten_statistics(payload, feature_id=feature_id))
            target = Path(self.root_dir) / f"{_safe_name(product.id)}.csv"
            _stats_frame(rows).to_csv(target, index=False)
            out.append(target)
        logger.info(f"Sentinel Hub batch-statistical: wrote {len(out)} table(s)")
        return out

    def download(
        self,
        progress_bar: bool = True,
        aggregate: AggregationConfig | None = None,
    ) -> list[Any]:
        """Build the request(s), render server-side, and collect the output.

        Args:
            progress_bar: Reserved for parity with the other backends.
            aggregate: Optional aggregation request. Accepted (not rejected)
                because `OUTPUT_KIND` is `"mixed"`; applied per plane in C10.

        Returns:
            The written raster paths / table paths / S3 URIs, depending on the
            resolved plane.
        """
        self._aggregate = aggregate
        plane = self._resolve_plane()
        if aggregate is not None and plane in RASTER_APIS:
            results = self._aggregate_windows(aggregate)
        else:
            # Tabular planes apply aggregate via the Statistical
            # aggregation_interval (see _statistical_interval), so no loop here.
            results = self._api_via_search_fetch()
        logger.info(
            f"Sentinel Hub download: {len(results)} result(s) written to "
            f"{self.root_dir}"
        )
        return results

    def _aggregate_windows(self, aggregate: AggregationConfig) -> list[Any]:
        """Render one output per `aggregate.freq` window over the request span.

        Splits `[start, end]` into `freq` windows and renders the resolved raster
        plane once per window, stamping each local output
        `{key}_{freq}_{YYYYMMDD}.{suffix}` (the `ecmwf` / `cmems` per-window
        shape). S3-delivered planes (async / batch) return their per-window URIs
        unchanged.

        Args:
            aggregate: The aggregation request (its `freq` drives the windows).

        Returns:
            The per-window outputs across every requested key.
        """
        import pandas as pd

        edges = list(
            pd.date_range(self.time.start_date, self.time.end_date, freq=aggregate.freq)
        )
        if not edges or pd.Timestamp(edges[0]) > pd.Timestamp(self.time.start_date):
            edges.insert(0, pd.Timestamp(self.time.start_date))
        results: list[Any] = []
        keys = list(self._resolved)
        try:
            for index, window_start in enumerate(edges):
                if index + 1 < len(edges):
                    # End the day before the next window starts so adjacent
                    # windows don't both claim the shared boundary date (the SH
                    # time_interval is inclusive on both ends).
                    window_end = pd.Timestamp(edges[index + 1]) - pd.Timedelta(days=1)
                    if window_end < pd.Timestamp(window_start):
                        window_end = pd.Timestamp(window_start)
                else:
                    window_end = pd.Timestamp(self.time.end_date)
                self._window_override = (
                    pd.Timestamp(window_start).strftime("%Y-%m-%d"),
                    pd.Timestamp(window_end).strftime("%Y-%m-%d"),
                )
                stamp = pd.Timestamp(window_start).strftime("%Y%m%d")
                produced = self._api_via_search_fetch()
                for key, item in zip(keys, produced):
                    results.append(
                        self._stamp_window_output(key, item, aggregate, stamp)
                    )
        finally:
            self._window_override = None
        return results

    def _stamp_window_output(
        self, key: str, item: Any, aggregate: AggregationConfig, stamp: str
    ) -> Any:
        """Rename a local per-window raster to the stamped name; pass URIs through.

        Args:
            key: The requested collection/recipe key.
            item: A produced output (a local `Path`/str, or an S3 URI string).
            aggregate: The aggregation request (its `freq` labels the file).
            stamp: The window's `YYYYMMDD` start stamp.

        Returns:
            The renamed local path, or the original S3 URI string.
        """
        source = Path(str(item))
        if not source.exists():
            return item
        suffix = source.suffix or ".tif"
        target = (
            Path(self.root_dir) / f"{_safe_name(key)}_{aggregate.freq}_{stamp}{suffix}"
        )
        source.replace(target)
        return target

__init__(start, end, variables, lat_lim, lon_lim, temporal_resolution='daily', path='', fmt='%Y-%m-%d', resolution=10.0, evalscript=None, endpoint=None, mosaicking_order=_DEFAULT_MOSAICKING_ORDER, api=None, geometry=None, maxcc=None, batch_output=None, client_id=None, client_secret=None, profile=None)

Initialise a Sentinel Hub backend instance.

Parameters:

Name	Type	Description	Default
start	str	Inclusive start date string (parsed with fmt).	required
end	str	Inclusive end date string.	required
variables	dict[str, list[str]]	{collection_or_recipe_key: [band, ...]}. A key is a catalog collection (needs an explicit evalscript=) or an evalscript recipe (pins its collection); an empty band list falls back to the recipe / collection default bands.	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; the render window is start/end.	'daily'
path	Path | str	Output directory (created by the parent class).	''
fmt	str	strptime format for start / end.	'%Y-%m-%d'
resolution	float	Output pixel size in metres (→ bbox_to_dimensions).	10.0
evalscript	str | None	A custom evalscript — an inline V3 JS string or a path to a .js file — that bypasses the recipe lookup (the collection then comes from the variables key). None uses the recipe's bundled .js.	None
endpoint	str | None	Endpoint alias ("cdse", "commercial") or a full base URL. Defaults to CDSE-free.	None
mosaicking_order	str	Per-pixel scene selection passed to input_data — "mostRecent" (default) / "leastRecent" / "leastCC" (the MosaickingOrder enum).	_DEFAULT_MOSAICKING_ORDER
api	str | None	The request plane ("process" / "async" / "batch" / "statistical" / "batch-statistical"), or None to auto-select by request size + whether geometry= was supplied.	None
geometry	Any	A shapely geometry / GeoJSON mapping / FeatureCollection for the Statistical planes (zonal stats over the polygon(s)).	None
maxcc	float | None	Optional maximum cloud cover (0–1) passed to input_data (optical collections only).	None
batch_output	dict[str, Any] | None	S3 delivery spec ({"bucket": ..., "iam_role_arn": ...}) for the Batch planes.	None
client_id	str | None	OAuth client id (else SENTINELHUB_CLIENT_ID).	None
client_secret	str | None	OAuth client secret (else SENTINELHUB_CLIENT_SECRET).	None
profile	str | None	A saved SHConfig profile name (else SENTINELHUB_PROFILE).	None

Raises:

Type	Description
ValueError	When api is an unknown plane, or mosaicking_order is not a recognised value.

Source code in src/earthlens/sentinel_hub/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 = "",
    fmt: str = "%Y-%m-%d",
    resolution: float = 10.0,
    evalscript: str | None = None,
    endpoint: str | None = None,
    mosaicking_order: str = _DEFAULT_MOSAICKING_ORDER,
    api: str | None = None,
    geometry: Any = None,
    maxcc: float | None = None,
    batch_output: dict[str, Any] | None = None,
    client_id: str | None = None,
    client_secret: str | None = None,
    profile: str | None = None,
):
    """Initialise a Sentinel Hub backend instance.

    Args:
        start: Inclusive start date string (parsed with `fmt`).
        end: Inclusive end date string.
        variables: `{collection_or_recipe_key: [band, ...]}`. A key is a
            catalog **collection** (needs an explicit `evalscript=`) or an
            **evalscript recipe** (pins its collection); an empty band list
            falls back to the recipe / collection default bands.
        lat_lim: `[lat_min, lat_max]` in degrees.
        lon_lim: `[lon_min, lon_max]` in degrees.
        temporal_resolution: Advisory cadence label; the render window is
            `start`/`end`.
        path: Output directory (created by the parent class).
        fmt: `strptime` format for `start` / `end`.
        resolution: Output pixel size in metres (→ `bbox_to_dimensions`).
        evalscript: A custom evalscript — an inline V3 JS string or a path to
            a `.js` file — that bypasses the recipe lookup (the collection
            then comes from the `variables` key). `None` uses the recipe's
            bundled `.js`.
        endpoint: Endpoint alias (`"cdse"`, `"commercial"`) or a full base
            URL. Defaults to CDSE-free.
        mosaicking_order: Per-pixel scene selection passed to
            `input_data` — `"mostRecent"` (default) / `"leastRecent"` /
            `"leastCC"` (the `MosaickingOrder` enum).
        api: The request plane (`"process"` / `"async"` / `"batch"` /
            `"statistical"` / `"batch-statistical"`), or `None` to
            auto-select by request size + whether `geometry=` was supplied.
        geometry: A shapely geometry / GeoJSON mapping / `FeatureCollection`
            for the Statistical planes (zonal stats over the polygon(s)).
        maxcc: Optional maximum cloud cover (0–1) passed to `input_data`
            (optical collections only).
        batch_output: S3 delivery spec (`{"bucket": ..., "iam_role_arn": ...}`)
            for the Batch planes.
        client_id: OAuth client id (else `SENTINELHUB_CLIENT_ID`).
        client_secret: OAuth client secret (else `SENTINELHUB_CLIENT_SECRET`).
        profile: A saved `SHConfig` profile name (else `SENTINELHUB_PROFILE`).

    Raises:
        ValueError: When `api` is an unknown plane, or `mosaicking_order` is
            not a recognised value.
    """
    validate_api(api)
    if mosaicking_order not in _VALID_MOSAICKING_ORDERS:
        raise ValueError(
            f"mosaicking_order must be one of {list(_VALID_MOSAICKING_ORDERS)}, "
            f"got {mosaicking_order!r}."
        )
    self._resolution = resolution
    self._evalscript = evalscript
    self._endpoint = endpoint
    self._mosaicking_order = mosaicking_order
    self._api_mode = api
    self._geometry = geometry
    self._maxcc = maxcc
    self._batch_output = batch_output
    env_creds = SentinelHubCredentials.from_env()
    self._credentials = SentinelHubCredentials(
        client_id=client_id or env_creds.client_id,
        client_secret=client_secret or env_creds.client_secret,
        profile=profile or env_creds.profile,
    )
    # Stored before super().__init__ because the base constructor calls
    # _initialize() (which needs the request) before it sets self.vars.
    self._variables = variables
    self._catalog: Catalog | None = None
    self._auth: SentinelHubAuth | None = None
    self._resolved: dict[str, ResolvedRequest] = {}
    # Aggregation request captured by download(); applied per plane.
    self._aggregate: AggregationConfig | None = None
    # Per-window time override set by the aggregate= render loop (C10).
    self._window_override: tuple[str, str] | None = None
    # Memoised resolved plane (deterministic from instance state; avoids a
    # second bbox_to_dimensions call across download() -> _fetch()).
    self._plane: str | None = None
    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, aggregate=None)

Build the request(s), render server-side, and collect the output.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Reserved for parity with the other backends.	True
aggregate	AggregationConfig | None	Optional aggregation request. Accepted (not rejected) because OUTPUT_KIND is "mixed"; applied per plane in C10.	None

Returns:

Type	Description
list[Any]	The written raster paths / table paths / S3 URIs, depending on the
list[Any]	resolved plane.

Source code in src/earthlens/sentinel_hub/backend.py

def download(
    self,
    progress_bar: bool = True,
    aggregate: AggregationConfig | None = None,
) -> list[Any]:
    """Build the request(s), render server-side, and collect the output.

    Args:
        progress_bar: Reserved for parity with the other backends.
        aggregate: Optional aggregation request. Accepted (not rejected)
            because `OUTPUT_KIND` is `"mixed"`; applied per plane in C10.

    Returns:
        The written raster paths / table paths / S3 URIs, depending on the
        resolved plane.
    """
    self._aggregate = aggregate
    plane = self._resolve_plane()
    if aggregate is not None and plane in RASTER_APIS:
        results = self._aggregate_windows(aggregate)
    else:
        # Tabular planes apply aggregate via the Statistical
        # aggregation_interval (see _statistical_interval), so no loop here.
        results = self._api_via_search_fetch()
    logger.info(
        f"Sentinel Hub download: {len(results)} result(s) written to "
        f"{self.root_dir}"
    )
    return results

search(limit=100)

Query the Sentinel Hub Catalog API for scenes intersecting the request.

A real STAC-style search (distinct from the internal :meth:_search fetch-planner): returns one :class:RemoteProduct per catalog item that intersects the request bbox + window, so a caller can enumerate coverage before rendering. An empty result is an empty list (not an error).

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of items to return per requested collection.	100

Returns:

Type	Description
list[RemoteProduct]	One product per catalog item (id + datetime + geometry on metadata).

Source code in src/earthlens/sentinel_hub/backend.py

def search(self, limit: int = 100) -> list[RemoteProduct]:
    """Query the Sentinel Hub Catalog API for scenes intersecting the request.

    A real STAC-style search (distinct from the internal :meth:`_search`
    fetch-planner): returns one :class:`RemoteProduct` per catalog item that
    intersects the request bbox + window, so a caller can enumerate coverage
    before rendering. An empty result is an empty list (not an error).

    Args:
        limit: Maximum number of items to return per requested collection.

    Returns:
        One product per catalog item (id + datetime + geometry on metadata).
    """
    sentinelhub = import_sentinelhub()
    cfg = self._auth.config()
    catalog = sentinelhub.SentinelHubCatalog(config=cfg)
    sh_bbox = self._bbox()
    time_interval = self._time_interval()
    products: list[RemoteProduct] = []
    seen_collections: set[str] = set()
    for key, resolved in self._resolved.items():
        if resolved.sh_collection in seen_collections:
            continue
        seen_collections.add(resolved.sh_collection)
        collection = cdse_collection(resolved.sh_collection, cfg.sh_base_url)
        for item in catalog.search(
            collection, bbox=sh_bbox, time=time_interval, limit=limit
        ):
            properties = item.get("properties", {})
            products.append(
                RemoteProduct(
                    id=item.get("id"),
                    metadata={
                        "key": key,
                        "collection": resolved.sh_collection,
                        "datetime": properties.get("datetime"),
                        "geometry": item.get("geometry"),
                    },
                )
            )
    return products

SentinelHubAuth

Bases: AbstractAuth[SentinelHubCredentials]

Assemble and hold a Sentinel Hub SHConfig (non-interactive OAuth2).

Conforms to the cross-backend :class:earthlens.base.AbstractAuth contract: :meth:configure builds the SHConfig lazily and idempotently; :meth:is_authenticated is the cheap predicate; and :meth:config returns the configured SHConfig (configuring on first use). The token itself is minted and refreshed by sentinelhub-py from the config on the first request — there is no interactive flow.

Parameters:

Name	Type	Description	Default
credentials	SentinelHubCredentials | None	The OAuth credentials. Defaults to an empty :class:SentinelHubCredentials (env / profile path).	None
endpoint	str | None	A named endpoint alias ("cdse", "commercial") or a full base URL. Defaults to CDSE-free.	None

Examples:

• Construct against CDSE-free (no config built until configure()):

  >>> from earthlens.sentinel_hub.auth import SentinelHubAuth
  >>> auth = SentinelHubAuth()
  >>> auth.is_authenticated()
  False
  

Source code in src/earthlens/sentinel_hub/auth.py

class SentinelHubAuth(AbstractAuth[SentinelHubCredentials]):
    """Assemble and hold a Sentinel Hub `SHConfig` (non-interactive OAuth2).

    Conforms to the cross-backend :class:`earthlens.base.AbstractAuth` contract:
    :meth:`configure` builds the `SHConfig` lazily and idempotently;
    :meth:`is_authenticated` is the cheap predicate; and :meth:`config` returns
    the configured `SHConfig` (configuring on first use). The token itself is
    minted and refreshed by `sentinelhub-py` from the config on the first
    request — there is no interactive flow.

    Args:
        credentials: The OAuth credentials. Defaults to an empty
            :class:`SentinelHubCredentials` (env / profile path).
        endpoint: A named endpoint alias (`"cdse"`, `"commercial"`) or a full
            base URL. Defaults to CDSE-free.

    Examples:
        - Construct against CDSE-free (no config built until `configure()`):
            ```python
            >>> from earthlens.sentinel_hub.auth import SentinelHubAuth
            >>> auth = SentinelHubAuth()
            >>> auth.is_authenticated()
            False

            ```
    """

    def __init__(
        self,
        credentials: SentinelHubCredentials | None = None,
        endpoint: str | None = None,
    ) -> None:
        """Store the credentials + resolved endpoint; does not build the config yet.

        Args:
            credentials: OAuth credentials, or `None` for the env / profile path.
            endpoint: Endpoint alias or base URL; resolved eagerly so a bad alias
                fails at construction.
        """
        super().__init__(credentials or SentinelHubCredentials())
        self._base_url, self._token_url = resolve_endpoint(endpoint)
        self._config: Any = None

    @property
    def base_url(self) -> str:
        """The resolved Sentinel Hub base URL this auth targets."""
        return self._base_url

    def _resolve_pair(self) -> tuple[str | None, str | None]:
        """Return the `(client_id, client_secret)` to use (env → kwargs).

        Explicit credentials on the object win over the environment.

        Returns:
            The resolved id and secret (either may be `None`).
        """
        env = SentinelHubCredentials.from_env()
        client_id = self._creds.client_id or env.client_id
        secret_obj = self._creds.client_secret or env.client_secret
        secret = secret_obj.get_secret_value() if secret_obj else None
        return client_id, secret

    def configure(self) -> None:
        """Build the `SHConfig` (base/token urls + credentials); idempotent.

        Resolves credentials env → kwargs → profile. A second call after
        :meth:`is_authenticated` returns `True` is a no-op.

        Raises:
            ImportError: When the `sentinel-hub` extra is not installed.
            AuthenticationError: When no credentials are resolvable (no id/secret
                and no profile).
        """
        if self.is_authenticated():
            return
        sentinelhub = import_sentinelhub()
        client_id, client_secret = self._resolve_pair()
        if not (client_id and client_secret) and not self._creds.profile:
            raise AuthenticationError(
                "no Sentinel Hub credentials found: set SENTINELHUB_CLIENT_ID / "
                "SENTINELHUB_CLIENT_SECRET (mint an OAuth client_credentials pair "
                f"in the CDSE Dashboard at {_DASHBOARD_URL}, Grant Type "
                "'Client Credentials'), pass client_id= / client_secret=, or "
                "supply a saved SHConfig profile= name."
            )
        cfg = (
            sentinelhub.SHConfig(self._creds.profile)
            if self._creds.profile
            else sentinelhub.SHConfig()
        )
        cfg.sh_base_url = self._base_url
        cfg.sh_token_url = self._token_url
        if client_id:
            cfg.sh_client_id = client_id
        if client_secret:
            cfg.sh_client_secret = client_secret
        self._config = cfg

    def is_authenticated(self) -> bool:
        """`True` once :meth:`configure` has assembled an `SHConfig`.

        Cheap predicate — does not call the network. Returns `True` exactly when
        a config object has been built.

        Returns:
            Whether the `SHConfig` is ready to issue requests.
        """
        return self._config is not None

    def config(self) -> Any:
        """Return the assembled `SHConfig`, configuring on first use.

        Returns:
            The `sentinelhub.SHConfig` carrying the base/token urls + credentials.

        Raises:
            ImportError: When the `sentinel-hub` extra is not installed.
            AuthenticationError: When no credentials are resolvable.
        """
        self.configure()
        return self._config

base_url property

The resolved Sentinel Hub base URL this auth targets.

__init__(credentials=None, endpoint=None)

Store the credentials + resolved endpoint; does not build the config yet.

Parameters:

Name	Type	Description	Default
credentials	SentinelHubCredentials | None	OAuth credentials, or None for the env / profile path.	None
endpoint	str | None	Endpoint alias or base URL; resolved eagerly so a bad alias fails at construction.	None

Source code in src/earthlens/sentinel_hub/auth.py

def __init__(
    self,
    credentials: SentinelHubCredentials | None = None,
    endpoint: str | None = None,
) -> None:
    """Store the credentials + resolved endpoint; does not build the config yet.

    Args:
        credentials: OAuth credentials, or `None` for the env / profile path.
        endpoint: Endpoint alias or base URL; resolved eagerly so a bad alias
            fails at construction.
    """
    super().__init__(credentials or SentinelHubCredentials())
    self._base_url, self._token_url = resolve_endpoint(endpoint)
    self._config: Any = None

config()

Return the assembled SHConfig, configuring on first use.

Returns:

Type	Description
Any	The sentinelhub.SHConfig carrying the base/token urls + credentials.

Raises:

Type	Description
ImportError	When the sentinel-hub extra is not installed.
AuthenticationError	When no credentials are resolvable.

Source code in src/earthlens/sentinel_hub/auth.py

def config(self) -> Any:
    """Return the assembled `SHConfig`, configuring on first use.

    Returns:
        The `sentinelhub.SHConfig` carrying the base/token urls + credentials.

    Raises:
        ImportError: When the `sentinel-hub` extra is not installed.
        AuthenticationError: When no credentials are resolvable.
    """
    self.configure()
    return self._config

configure()

Build the SHConfig (base/token urls + credentials); idempotent.

Resolves credentials env → kwargs → profile. A second call after :meth:is_authenticated returns True is a no-op.

Raises:

Type	Description
ImportError	When the sentinel-hub extra is not installed.
AuthenticationError	When no credentials are resolvable (no id/secret and no profile).

Source code in src/earthlens/sentinel_hub/auth.py

def configure(self) -> None:
    """Build the `SHConfig` (base/token urls + credentials); idempotent.

    Resolves credentials env → kwargs → profile. A second call after
    :meth:`is_authenticated` returns `True` is a no-op.

    Raises:
        ImportError: When the `sentinel-hub` extra is not installed.
        AuthenticationError: When no credentials are resolvable (no id/secret
            and no profile).
    """
    if self.is_authenticated():
        return
    sentinelhub = import_sentinelhub()
    client_id, client_secret = self._resolve_pair()
    if not (client_id and client_secret) and not self._creds.profile:
        raise AuthenticationError(
            "no Sentinel Hub credentials found: set SENTINELHUB_CLIENT_ID / "
            "SENTINELHUB_CLIENT_SECRET (mint an OAuth client_credentials pair "
            f"in the CDSE Dashboard at {_DASHBOARD_URL}, Grant Type "
            "'Client Credentials'), pass client_id= / client_secret=, or "
            "supply a saved SHConfig profile= name."
        )
    cfg = (
        sentinelhub.SHConfig(self._creds.profile)
        if self._creds.profile
        else sentinelhub.SHConfig()
    )
    cfg.sh_base_url = self._base_url
    cfg.sh_token_url = self._token_url
    if client_id:
        cfg.sh_client_id = client_id
    if client_secret:
        cfg.sh_client_secret = client_secret
    self._config = cfg

is_authenticated()

True once :meth:configure has assembled an SHConfig.

Cheap predicate — does not call the network. Returns True exactly when a config object has been built.

Returns:

Type	Description
bool	Whether the SHConfig is ready to issue requests.

Source code in src/earthlens/sentinel_hub/auth.py

def is_authenticated(self) -> bool:
    """`True` once :meth:`configure` has assembled an `SHConfig`.

    Cheap predicate — does not call the network. Returns `True` exactly when
    a config object has been built.

    Returns:
        Whether the `SHConfig` is ready to issue requests.
    """
    return self._config is not None

SentinelHubCredentials

Bases: BaseModel

Frozen value object holding the Sentinel Hub OAuth credentials.

All fields are optional: an empty object falls back to the SENTINELHUB_CLIENT_ID / SENTINELHUB_CLIENT_SECRET environment (or the SH_* fallback), then to the named profile. Supplying client_id (+ client_secret) selects the explicit client-credentials pair (which wins over the environment).

Attributes:

Name	Type	Description
client_id	str | None	OAuth client id, or None to read SENTINELHUB_CLIENT_ID.
client_secret	SecretStr | None	OAuth client secret (kept as a SecretStr), or None to read SENTINELHUB_CLIENT_SECRET.
profile	str | None	A saved SHConfig profile name to load credentials from when no id/secret is supplied, or None for the default profile.

Examples:

• An empty object carries no id (the env / profile path):

  >>> from earthlens.sentinel_hub.auth import SentinelHubCredentials
  >>> SentinelHubCredentials().client_id is None
  True
  

• The secret is carried opaquely:

  >>> from earthlens.sentinel_hub.auth import SentinelHubCredentials
  >>> creds = SentinelHubCredentials(client_id="abc", client_secret="shh")
  >>> creds.client_secret.get_secret_value()
  'shh'
  

Source code in src/earthlens/sentinel_hub/auth.py

class SentinelHubCredentials(BaseModel):
    """Frozen value object holding the Sentinel Hub OAuth credentials.

    All fields are optional: an empty object falls back to the
    `SENTINELHUB_CLIENT_ID` / `SENTINELHUB_CLIENT_SECRET` environment (or the
    `SH_*` fallback), then to the named `profile`. Supplying `client_id`
    (+ `client_secret`) selects the explicit client-credentials pair (which wins
    over the environment).

    Attributes:
        client_id: OAuth client id, or `None` to read `SENTINELHUB_CLIENT_ID`.
        client_secret: OAuth client secret (kept as a `SecretStr`), or `None` to
            read `SENTINELHUB_CLIENT_SECRET`.
        profile: A saved `SHConfig` profile name to load credentials from when
            no id/secret is supplied, or `None` for the default profile.

    Examples:
        - An empty object carries no id (the env / profile path):
            ```python
            >>> from earthlens.sentinel_hub.auth import SentinelHubCredentials
            >>> SentinelHubCredentials().client_id is None
            True

            ```
        - The secret is carried opaquely:
            ```python
            >>> from earthlens.sentinel_hub.auth import SentinelHubCredentials
            >>> creds = SentinelHubCredentials(client_id="abc", client_secret="shh")
            >>> creds.client_secret.get_secret_value()
            'shh'

            ```
    """

    model_config = ConfigDict(frozen=True)

    client_id: str | None = None
    client_secret: SecretStr | None = None
    profile: str | None = None

    @classmethod
    def from_env(cls) -> SentinelHubCredentials:
        """Build credentials from the `SENTINELHUB_*` (or `SH_*`) environment.

        Reads the descriptive `SENTINELHUB_CLIENT_ID` / `SENTINELHUB_CLIENT_SECRET`
        / `SENTINELHUB_PROFILE` variables, falling back to the `sentinelhub-py`
        native `SH_CLIENT_ID` / `SH_CLIENT_SECRET` / `SH_PROFILE` for users coming
        from the SDK. Absent variables stay `None`.

        Returns:
            The credentials object built from the environment.
        """
        return cls(
            client_id=(
                os.environ.get("SENTINELHUB_CLIENT_ID")
                or os.environ.get("SH_CLIENT_ID")
            ),
            client_secret=(
                os.environ.get("SENTINELHUB_CLIENT_SECRET")
                or os.environ.get("SH_CLIENT_SECRET")
            ),
            profile=(
                os.environ.get("SENTINELHUB_PROFILE") or os.environ.get("SH_PROFILE")
            ),
        )

from_env() classmethod

Build credentials from the SENTINELHUB_* (or SH_*) environment.

Reads the descriptive SENTINELHUB_CLIENT_ID / SENTINELHUB_CLIENT_SECRET / SENTINELHUB_PROFILE variables, falling back to the sentinelhub-py native SH_CLIENT_ID / SH_CLIENT_SECRET / SH_PROFILE for users coming from the SDK. Absent variables stay None.

Returns:

Type	Description
SentinelHubCredentials	The credentials object built from the environment.

Source code in src/earthlens/sentinel_hub/auth.py

@classmethod
def from_env(cls) -> SentinelHubCredentials:
    """Build credentials from the `SENTINELHUB_*` (or `SH_*`) environment.

    Reads the descriptive `SENTINELHUB_CLIENT_ID` / `SENTINELHUB_CLIENT_SECRET`
    / `SENTINELHUB_PROFILE` variables, falling back to the `sentinelhub-py`
    native `SH_CLIENT_ID` / `SH_CLIENT_SECRET` / `SH_PROFILE` for users coming
    from the SDK. Absent variables stay `None`.

    Returns:
        The credentials object built from the environment.
    """
    return cls(
        client_id=(
            os.environ.get("SENTINELHUB_CLIENT_ID")
            or os.environ.get("SH_CLIENT_ID")
        ),
        client_secret=(
            os.environ.get("SENTINELHUB_CLIENT_SECRET")
            or os.environ.get("SH_CLIENT_SECRET")
        ),
        profile=(
            os.environ.get("SENTINELHUB_PROFILE") or os.environ.get("SH_PROFILE")
        ),
    )

read_evalscript(filename)

Read a bundled evalscript .js file by name.

Parameters:

Name	Type	Description	Default
filename	str	The .js filename (e.g. "ndvi.js") under :data:EVALSCRIPTS_PATH.	required

Returns:

Type	Description
str	The evalscript source.

Raises:

Type	Description
FileNotFoundError	When no such bundled evalscript exists.

Source code in src/earthlens/sentinel_hub/catalog.py

def read_evalscript(filename: str) -> str:
    """Read a bundled evalscript `.js` file by name.

    Args:
        filename: The `.js` filename (e.g. `"ndvi.js"`) under
            :data:`EVALSCRIPTS_PATH`.

    Returns:
        The evalscript source.

    Raises:
        FileNotFoundError: When no such bundled evalscript exists.
    """
    path = EVALSCRIPTS_PATH / filename
    if not path.is_file():
        available = sorted(p.name for p in EVALSCRIPTS_PATH.glob("*.js"))
        raise FileNotFoundError(
            f"bundled evalscript {filename!r} not found under {EVALSCRIPTS_PATH}. "
            f"Available: {available}."
        )
    return path.read_text(encoding="utf-8")

earthlens.sentinel_hub.backend

Sentinel Hub server-side-render backend (raster or tabular output).

SentinelHub(AbstractDataSource) sends a bbox/geometry + time + an evalscript to one of Sentinel Hub's request planes on CDSE (sh.dataspace.copernicus.eu, free with a CDSE account); the server computes on-the-fly and earthlens collects the result. Its closest siblings are the GEE and openEO backends — the server does the band math / compositing / rendering and earthlens (a) authenticates, (b) builds the request from a curated evalscript library, (c) triggers, (d) collects the output.

A request is variables={collection_or_recipe: [band, ...]} plus a bbox + date window. A key may name a collection (then an explicit evalscript= is required) or an evalscript recipe (a bundled .js that pins its collection). The plane is chosen by the api= kwarg ("process" / "async" / "batch" / "statistical" / "batch-statistical"), auto-selected by request size + whether a geometry= was supplied when api= is omitted.

Because it returns raster or tabular depending on api=, the backend is OUTPUT_KIND="mixed" (a fixed class attribute, never mutated per-instance): the EarthLens facade forwards aggregate= for {"raster", "mixed"}, and aggregate= is supported on every plane (including the statistical plane, where it maps to the Statistical aggregation_interval), so demoting the instance to "tabular" would wrongly make the facade reject it.

SentinelHub

Bases: AbstractDataSource

Server-side Sentinel Hub backend on CDSE (raster or tabular by api=).

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	"mixed" — raster (Process/Async/Batch) or tabular (Statistical/Batch-Statistical) depending on the resolved api=. A fixed class attribute (never mutated per-instance) so the facade forwards aggregate= on every plane.

Source code in src/earthlens/sentinel_hub/backend.py

class SentinelHub(AbstractDataSource):
    """Server-side Sentinel Hub backend on CDSE (raster or tabular by `api=`).

    Attributes:
        OUTPUT_KIND: `"mixed"` — raster (Process/Async/Batch) or tabular
            (Statistical/Batch-Statistical) depending on the resolved `api=`.
            A *fixed* class attribute (never mutated per-instance) so the facade
            forwards `aggregate=` on every plane.
    """

    OUTPUT_KIND: OutputKind = "mixed"

    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 = "",
        fmt: str = "%Y-%m-%d",
        resolution: float = 10.0,
        evalscript: str | None = None,
        endpoint: str | None = None,
        mosaicking_order: str = _DEFAULT_MOSAICKING_ORDER,
        api: str | None = None,
        geometry: Any = None,
        maxcc: float | None = None,
        batch_output: dict[str, Any] | None = None,
        client_id: str | None = None,
        client_secret: str | None = None,
        profile: str | None = None,
    ):
        """Initialise a Sentinel Hub backend instance.

        Args:
            start: Inclusive start date string (parsed with `fmt`).
            end: Inclusive end date string.
            variables: `{collection_or_recipe_key: [band, ...]}`. A key is a
                catalog **collection** (needs an explicit `evalscript=`) or an
                **evalscript recipe** (pins its collection); an empty band list
                falls back to the recipe / collection default bands.
            lat_lim: `[lat_min, lat_max]` in degrees.
            lon_lim: `[lon_min, lon_max]` in degrees.
            temporal_resolution: Advisory cadence label; the render window is
                `start`/`end`.
            path: Output directory (created by the parent class).
            fmt: `strptime` format for `start` / `end`.
            resolution: Output pixel size in metres (→ `bbox_to_dimensions`).
            evalscript: A custom evalscript — an inline V3 JS string or a path to
                a `.js` file — that bypasses the recipe lookup (the collection
                then comes from the `variables` key). `None` uses the recipe's
                bundled `.js`.
            endpoint: Endpoint alias (`"cdse"`, `"commercial"`) or a full base
                URL. Defaults to CDSE-free.
            mosaicking_order: Per-pixel scene selection passed to
                `input_data` — `"mostRecent"` (default) / `"leastRecent"` /
                `"leastCC"` (the `MosaickingOrder` enum).
            api: The request plane (`"process"` / `"async"` / `"batch"` /
                `"statistical"` / `"batch-statistical"`), or `None` to
                auto-select by request size + whether `geometry=` was supplied.
            geometry: A shapely geometry / GeoJSON mapping / `FeatureCollection`
                for the Statistical planes (zonal stats over the polygon(s)).
            maxcc: Optional maximum cloud cover (0–1) passed to `input_data`
                (optical collections only).
            batch_output: S3 delivery spec (`{"bucket": ..., "iam_role_arn": ...}`)
                for the Batch planes.
            client_id: OAuth client id (else `SENTINELHUB_CLIENT_ID`).
            client_secret: OAuth client secret (else `SENTINELHUB_CLIENT_SECRET`).
            profile: A saved `SHConfig` profile name (else `SENTINELHUB_PROFILE`).

        Raises:
            ValueError: When `api` is an unknown plane, or `mosaicking_order` is
                not a recognised value.
        """
        validate_api(api)
        if mosaicking_order not in _VALID_MOSAICKING_ORDERS:
            raise ValueError(
                f"mosaicking_order must be one of {list(_VALID_MOSAICKING_ORDERS)}, "
                f"got {mosaicking_order!r}."
            )
        self._resolution = resolution
        self._evalscript = evalscript
        self._endpoint = endpoint
        self._mosaicking_order = mosaicking_order
        self._api_mode = api
        self._geometry = geometry
        self._maxcc = maxcc
        self._batch_output = batch_output
        env_creds = SentinelHubCredentials.from_env()
        self._credentials = SentinelHubCredentials(
            client_id=client_id or env_creds.client_id,
            client_secret=client_secret or env_creds.client_secret,
            profile=profile or env_creds.profile,
        )
        # Stored before super().__init__ because the base constructor calls
        # _initialize() (which needs the request) before it sets self.vars.
        self._variables = variables
        self._catalog: Catalog | None = None
        self._auth: SentinelHubAuth | None = None
        self._resolved: dict[str, ResolvedRequest] = {}
        # Aggregation request captured by download(); applied per plane.
        self._aggregate: AggregationConfig | None = None
        # Per-window time override set by the aggregate= render loop (C10).
        self._window_override: tuple[str, str] | None = None
        # Memoised resolved plane (deterministic from instance state; avoids a
        # second bbox_to_dimensions call across download() -> _fetch()).
        self._plane: str | None = None
        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 _initialize(self) -> None:
        """Load the catalog, resolve the requested keys, and build the auth.

        Returns `None` so the parent does not bind `self.client`; the auth
        wrapper is stored on `self._auth` and the resolved rows on
        `self._resolved`. Token minting is deferred to first request.

        Raises:
            TypeError: When `variables` is not a
                `{collection_or_recipe: [band, ...]}` mapping.
            ValueError: When `variables` is empty or names a key the catalog
                does not know (with a did-you-mean hint).
        """
        from earthlens.sentinel_hub.catalog import Catalog

        if not isinstance(self._variables, dict):
            raise TypeError(
                "Sentinel Hub requires variables as a "
                "{collection_or_recipe: [band, ...]} mapping, got "
                f"{type(self._variables).__name__}."
            )
        if not self._variables:
            raise ValueError(
                "Sentinel Hub requires variables={collection_or_recipe: "
                "[band, ...]} with at least one collection or recipe key."
            )
        self._catalog = Catalog()
        self._resolved = {key: self._catalog.resolve(key) for key in self._variables}
        self._auth = SentinelHubAuth(self._credentials, endpoint=self._endpoint)
        return None

    def _create_grid(self, lat_lim: list, lon_lim: list) -> SpatialExtent:
        """Build the WGS84 envelope the Sentinel Hub `BBox` is built from.

        Args:
            lat_lim: `[lat_min, lat_max]` in degrees.
            lon_lim: `[lon_min, lon_max]` in degrees.

        Returns:
            SpatialExtent: The request's bounding box.
        """
        return SpatialExtent.from_pairs(lat_lim=lat_lim, lon_lim=lon_lim)

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

        Args:
            start: Inclusive start date string (parsed with `fmt`).
            end: Inclusive end date string.
            temporal_resolution: Advisory cadence label (`"daily"`, `"monthly"`,
                `"hourly"`, `"yearly"`) → the pandas frequency on the extent.
            fmt: `strptime` format for `start` / `end`.

        Returns:
            TemporalExtent: The parsed window (inclusive `end_date`).

        Raises:
            ValueError: When `start` or `end` is `None`.
        """
        import datetime as dt

        import pandas as pd

        if start is None or end is None:
            raise ValueError(
                "Sentinel Hub requires both start and end dates (the render "
                "needs a time_interval); pass start=… and end=…."
            )
        start_dt = dt.datetime.strptime(start, fmt)
        end_dt = dt.datetime.strptime(end, fmt)
        freq_map = {"daily": "D", "monthly": "MS", "hourly": "h", "yearly": "YS"}
        resolution = freq_map.get(temporal_resolution, "D")
        dates = pd.date_range(start_dt, end_dt, freq=resolution)
        return TemporalExtent(
            start_date=start_dt, end_date=end_dt, resolution=resolution, dates=dates
        )

    def _bbox(self) -> Any:
        """Build the Sentinel Hub `BBox` from the request envelope (WGS84).

        Returns:
            A `sentinelhub.BBox` over the request bbox.
        """
        sentinelhub = import_sentinelhub()
        return sentinelhub.BBox(
            (self.space.west, self.space.south, self.space.east, self.space.north),
            crs=sentinelhub.CRS.WGS84,
        )

    def _request_size(self) -> tuple[int, int]:
        """Compute the render size in pixels via `bbox_to_dimensions`.

        Returns:
            `(width_px, height_px)` for the request bbox at `resolution`.
        """
        sentinelhub = import_sentinelhub()
        return sentinelhub.bbox_to_dimensions(self._bbox(), resolution=self._resolution)

    def _resolve_plane(self) -> str:
        """Resolve the request plane: explicit `api=`, else auto by size / geometry.

        Memoised: the plane is deterministic from instance state, so the result
        is computed once (one `bbox_to_dimensions` call) and reused across the
        `download` -> `_fetch` path.

        Returns:
            The resolved plane name.
        """
        if self._plane is None:
            has_geometry = self._geometry is not None
            has_s3 = self._batch_output is not None
            needs_size = self._api_mode is None or self._api_mode in RASTER_APIS
            max_side = max(self._request_size()) if needs_size else 0
            self._plane = resolve_api(self._api_mode, max_side, has_geometry, has_s3)
        return self._plane

    def _time_interval(self) -> tuple[str, str]:
        """Return the render time interval as ISO `(start, end)` date strings.

        Honours a per-window override set by the `aggregate=` render loop (C10);
        otherwise the full request window.
        """
        if self._window_override is not None:
            return self._window_override
        return (
            self.time.start_date.strftime("%Y-%m-%d"),
            self.time.end_date.strftime("%Y-%m-%d"),
        )

    def _api(self) -> list[Path]:
        """Compose `_search` and `_fetch` into the canonical search/fetch shape."""
        return self._api_via_search_fetch()

    def _search(self) -> list[RemoteProduct]:
        """List the planned requests without rendering (cheap dry-run).

        Returns one :class:`RemoteProduct` per requested key; the resolved row
        rides on `metadata["resolved"]`. No network call.

        Returns:
            One product per requested collection/recipe key.
        """
        return [
            RemoteProduct(id=key, metadata={"resolved": resolved})
            for key, resolved in self._resolved.items()
        ]

    def search(self, limit: int = 100) -> list[RemoteProduct]:
        """Query the Sentinel Hub Catalog API for scenes intersecting the request.

        A real STAC-style search (distinct from the internal :meth:`_search`
        fetch-planner): returns one :class:`RemoteProduct` per catalog item that
        intersects the request bbox + window, so a caller can enumerate coverage
        before rendering. An empty result is an empty list (not an error).

        Args:
            limit: Maximum number of items to return per requested collection.

        Returns:
            One product per catalog item (id + datetime + geometry on metadata).
        """
        sentinelhub = import_sentinelhub()
        cfg = self._auth.config()
        catalog = sentinelhub.SentinelHubCatalog(config=cfg)
        sh_bbox = self._bbox()
        time_interval = self._time_interval()
        products: list[RemoteProduct] = []
        seen_collections: set[str] = set()
        for key, resolved in self._resolved.items():
            if resolved.sh_collection in seen_collections:
                continue
            seen_collections.add(resolved.sh_collection)
            collection = cdse_collection(resolved.sh_collection, cfg.sh_base_url)
            for item in catalog.search(
                collection, bbox=sh_bbox, time=time_interval, limit=limit
            ):
                properties = item.get("properties", {})
                products.append(
                    RemoteProduct(
                        id=item.get("id"),
                        metadata={
                            "key": key,
                            "collection": resolved.sh_collection,
                            "datetime": properties.get("datetime"),
                            "geometry": item.get("geometry"),
                        },
                    )
                )
        return products

    def _fetch(self, products: list[RemoteProduct]) -> list[Any]:
        """Dispatch to the per-plane fetcher for the resolved `api`.

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            The written raster paths (raster planes) or table paths / S3 URIs
            (tabular / batch planes).
        """
        plane = self._resolve_plane()
        fetchers = {
            "process": self._fetch_process,
            "async": self._fetch_async,
            "tiling": self._fetch_tiling,
            "batch": self._fetch_batch,
            "statistical": self._fetch_statistical,
            "batch-statistical": self._fetch_batch_statistical,
        }
        return fetchers[plane](products)

    def _read_evalscript(self, resolved: ResolvedRequest) -> str:
        """Resolve the evalscript source for one request row.

        A custom `evalscript=` (an inline V3 JS string or a `.js` file path)
        wins over the recipe. Otherwise the recipe's bundled `.js` is read; a
        plain collection (no recipe evalscript and no custom one) is an error.

        Args:
            resolved: The resolved request row.

        Returns:
            The evalscript source string.

        Raises:
            ValueError: When the key is a plain collection and no `evalscript=`
                was supplied.
        """
        if self._evalscript is not None:
            candidate = self._evalscript
            try:
                path = Path(candidate)
                if path.is_file():
                    return path.read_text(encoding="utf-8")
            except OSError:
                pass
            return candidate
        if resolved.evalscript is None:
            raise ValueError(
                f"{resolved.key!r} is a plain collection, so it has no bundled "
                "evalscript. Pass evalscript= (an inline V3 JS string or a .js "
                "path), or request an evalscript recipe key instead."
            )
        return read_evalscript(resolved.evalscript)

    def _guard_process_size(self, size: tuple[int, int]) -> None:
        """Reject a Process request whose render exceeds the 2500 px cap.

        Args:
            size: The `(width, height)` render size in pixels.

        Raises:
            ValueError: When either side exceeds :data:`SH_MAX_DIMENSION`.
        """
        if max(size) > SH_MAX_DIMENSION:
            raise ValueError(
                f"the request renders to {size} px but the Process API caps a "
                f"single request at {SH_MAX_DIMENSION} px/side. Omit api= to "
                "auto-route to async / batch, or lower the resolution."
            )

    def _build_input_data(self, sentinelhub: Any, resolved: ResolvedRequest) -> dict:
        """Build the `SentinelHubRequest.input_data` block for one row.

        Args:
            sentinelhub: The imported `sentinelhub` module.
            resolved: The resolved request row.

        Returns:
            The `input_data` dict (collection, window, mosaicking order, maxcc).
        """
        cfg = self._auth.config()
        collection = cdse_collection(resolved.sh_collection, cfg.sh_base_url)
        kwargs: dict[str, Any] = {
            "data_collection": collection,
            "time_interval": self._time_interval(),
            "mosaicking_order": self._mosaicking_order,
        }
        if self._maxcc is not None:
            kwargs["maxcc"] = self._maxcc
        return sentinelhub.SentinelHubRequest.input_data(**kwargs)

    def _fetch_process(self, products: list[RemoteProduct]) -> list[Path]:
        """Render each product synchronously via the Process API → GeoTIFF on disk.

        Builds one `SentinelHubRequest` per resolved row over the request bbox +
        window at `resolution`, writes the rendered GeoTIFF under `path`, and
        returns the written file paths (resolved from the SDK rather than
        re-hashing).

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            The written GeoTIFF paths, one per product.

        Raises:
            ValueError: When the render exceeds the Process 2500 px cap.
        """
        sentinelhub = import_sentinelhub()
        sh_bbox = self._bbox()
        size = self._request_size()
        self._guard_process_size(size)
        out: list[Path] = []
        for product in products:
            resolved: ResolvedRequest = product.metadata["resolved"]
            out.append(
                self._render_process_tile(
                    sentinelhub, resolved, sh_bbox, size, str(self.root_dir)
                )
            )
        return out

    def _render_process_tile(
        self,
        sentinelhub: Any,
        resolved: ResolvedRequest,
        bbox: Any,
        size: tuple[int, int],
        data_folder: str,
    ) -> Path:
        """Build + run one Process request and return the written GeoTIFF path.

        Shared by :meth:`_fetch_process` (whole bbox) and :meth:`_fetch_tiling`
        (per tile).

        Args:
            sentinelhub: The imported `sentinelhub` module.
            resolved: The resolved request row.
            bbox: The `sentinelhub.BBox` to render.
            size: The `(width, height)` render size for this bbox.
            data_folder: The directory the SDK writes the GeoTIFF under.

        Returns:
            The written GeoTIFF path.
        """
        request = sentinelhub.SentinelHubRequest(
            evalscript=self._read_evalscript(resolved),
            input_data=[self._build_input_data(sentinelhub, resolved)],
            responses=[
                sentinelhub.SentinelHubRequest.output_response(
                    "default", sentinelhub.MimeType.TIFF
                )
            ],
            bbox=bbox,
            size=size,
            data_folder=data_folder,
            config=self._auth.config(),
        )
        request.get_data(save_data=True)
        return Path(data_folder) / request.get_filename_list()[0]

    def _require_s3_delivery(self, sentinelhub: Any) -> Any:
        """Build the S3 `AccessSpecification` from `batch_output`, or error.

        Args:
            sentinelhub: The imported `sentinelhub` module.

        Returns:
            The S3 delivery `AccessSpecification`.

        Raises:
            ValueError: When no `batch_output` was supplied, or it carries no
                `bucket` / `url`.
        """
        if not self._batch_output:
            raise ValueError(
                "the async / batch planes deliver server-side to S3, so they "
                "need batch_output={'bucket': 's3://…', 'iam_role_arn': '…'}. "
                "Omit api= (or pass api='tiling') for a no-S3 oversized render."
            )
        spec = dict(self._batch_output)
        url = spec.pop("bucket", None) or spec.pop("url", None)
        if not url:
            raise ValueError(
                "batch_output must include a 'bucket' (or 'url') S3 destination, "
                f"got {dict(self._batch_output)!r}."
            )
        return sentinelhub.AsyncProcessRequest.s3_specification(url=url, **spec)

    def _guard_async_size(self, size: tuple[int, int]) -> None:
        """Reject an Async request beyond the 10000 px ceiling.

        Args:
            size: The `(width, height)` render size in pixels.

        Raises:
            ValueError: When either side exceeds :data:`ASYNC_MAX_DIMENSION`.
        """
        if max(size) > ASYNC_MAX_DIMENSION:
            raise ValueError(
                f"the request renders to {size} px but the Async Processing API "
                f"caps a request at {ASYNC_MAX_DIMENSION} px/side. Use api='batch' "
                "for a larger AOI."
            )

    def _fetch_async(self, products: list[RemoteProduct]) -> list[str]:
        """Render via the Async Processing API (S3-delivered, ≤10000 px).

        Submits one `AsyncProcessRequest` per row delivering to the `batch_output`
        S3 bucket, polls `get_async_running_status` to completion, and returns the
        S3 delivery URIs. Requires an S3 `batch_output` (the SDK's async plane is
        not a direct synchronous download).

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            The S3 **delivery prefix** (the configured `batch_output` bucket), one
            per product — not the individual written object keys (enumerate the
            bucket for those; earthlens does not perform S3 listing).

        Raises:
            ValueError: When no `batch_output` is set or the render exceeds the
                10000 px Async ceiling.
        """
        sentinelhub = import_sentinelhub()
        cfg = self._auth.config()
        delivery = self._require_s3_delivery(sentinelhub)
        bucket_uri = self._batch_output.get("bucket") or self._batch_output.get("url")
        sh_bbox = self._bbox()
        size = self._request_size()
        self._guard_async_size(size)
        out: list[str] = []
        for product in products:
            resolved: ResolvedRequest = product.metadata["resolved"]
            evalscript = self._read_evalscript(resolved)
            request = sentinelhub.AsyncProcessRequest(
                evalscript=evalscript,
                input_data=[self._build_input_data(sentinelhub, resolved)],
                responses=[
                    sentinelhub.AsyncProcessRequest.output_response(
                        "default", sentinelhub.MimeType.TIFF
                    )
                ],
                delivery=delivery,
                bbox=sh_bbox,
                size=size,
                config=cfg,
            )
            # AsyncProcessRequest.get_data submits the job and returns the
            # submission JSON, which carries the async request id; poll on that
            # id (NOT the delivery URL — get_async_running_status resolves
            # `…/async/process/{id}`).
            request_id = _async_request_id(request.get_data(save_data=False))
            if request_id is not None:
                _wait_for_async(sentinelhub, [request_id], cfg)
            else:
                logger.warning(
                    "Sentinel Hub async: could not determine the request id from "
                    "the submission response; skipping the completion poll."
                )
            out.append(str(bucket_uri))
        logger.info(f"Sentinel Hub async: delivered {len(out)} object(s) to S3")
        return out

    def _fetch_tiling(self, products: list[RemoteProduct]) -> list[Path]:
        """Render an oversized AOI by local tiling + mosaic (no S3 needed).

        Splits the request bbox into ≤2500 px Process tiles, renders each tile,
        and mosaics them into one GeoTIFF per product with
        `pyramids.dataset.merge.merge_rasters`. Tile temporaries are written
        under a per-product subdirectory and removed after the merge.

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            The merged GeoTIFF paths, one per product (union bounds).

        Raises:
            ValueError: When a tile still renders above the Process cap (a
                rounding edge case); lower the `resolution` to re-tile finer.
        """
        import shutil

        from pyramids.dataset.merge import merge_rasters

        sentinelhub = import_sentinelhub()
        width, height = self._request_size()
        tiles = tile_bbox(
            (self.space.west, self.space.south, self.space.east, self.space.north),
            width,
            height,
        )
        # Pre-flight: size every tile once and guard against the Process cap
        # before rendering any, so a rounding edge case fails fast with a clear
        # error rather than a mid-run server 500 (and no partial tiles leak).
        tile_specs: list[tuple[Any, tuple[int, int]]] = []
        for tile in tiles:
            sh_bbox = sentinelhub.BBox(tile, crs=sentinelhub.CRS.WGS84)
            tile_size = sentinelhub.bbox_to_dimensions(
                sh_bbox, resolution=self._resolution
            )
            if max(tile_size) > SH_MAX_DIMENSION:
                raise ValueError(
                    f"a local tile renders to {tile_size} px, exceeding the "
                    f"{SH_MAX_DIMENSION} px Process cap; lower the resolution to "
                    "re-tile finer."
                )
            tile_specs.append((sh_bbox, tile_size))
        out: list[Path] = []
        for product in products:
            resolved: ResolvedRequest = product.metadata["resolved"]
            tile_dir = Path(self.root_dir) / f"_tiles_{_safe_name(product.id)}"
            tile_dir.mkdir(parents=True, exist_ok=True)
            tile_paths: list[str] = []
            for index, (sh_bbox, tile_size) in enumerate(tile_specs):
                rendered = self._render_process_tile(
                    sentinelhub,
                    resolved,
                    sh_bbox,
                    tile_size,
                    str(tile_dir / str(index)),
                )
                tile_paths.append(str(rendered))
            merged = Path(self.root_dir) / f"{_safe_name(product.id)}.tif"
            merge_rasters(tile_paths, str(merged))
            shutil.rmtree(tile_dir, ignore_errors=True)
            out.append(merged)
        logger.info(f"Sentinel Hub tiling: merged {len(tiles)} tile(s) per product")
        return out

    def _fetch_batch(self, products: list[RemoteProduct]) -> list[str]:
        """Render a very large AOI via the Batch Processing API, tiled to S3.

        Builds the base Process request, then creates a batch request with a
        server-side tiling grid delivering to the `batch_output` S3 bucket,
        runs analysis → start → monitor, and returns the S3 destination URIs.
        Requires an S3 `batch_output` with at least a bucket; `grid_id` selects
        the tiling grid (default 0).

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            The S3 **delivery prefix** (the configured `batch_output` bucket), one
            per product — the server tiles the output under that prefix; enumerate
            the bucket for the individual tile object keys (earthlens does not
            perform S3 listing).

        Raises:
            ValueError: When no `batch_output` (S3 bucket) was supplied, or the
                analysed `cost_PU` exceeds `batch_output['max_cost_pu']`.
        """
        sentinelhub = import_sentinelhub()
        if not self._batch_output:
            raise ValueError(
                "the Batch Processing plane tiles server-side to S3, so it needs "
                "batch_output={'bucket': 's3://…', 'iam_role_arn': '…', "
                "'grid_id': <int>}."
            )
        cfg = self._auth.config()
        client = sentinelhub.BatchProcessClient(config=cfg)
        spec = dict(self._batch_output)
        grid_id = spec.pop("grid_id", 0)
        buffer_x = spec.pop("buffer_x", None)
        buffer_y = spec.pop("buffer_y", None)
        max_cost_pu = spec.pop("max_cost_pu", None)
        url = spec.pop("bucket", None) or spec.pop("url", None)
        if not url:
            raise ValueError(
                "batch_output must include a 'bucket' (or 'url') S3 destination, "
                f"got {dict(self._batch_output)!r}."
            )
        delivery = client.s3_specification(url=url, **spec)
        tiling = client.tiling_grid_input(
            grid_id=grid_id,
            resolution=self._resolution,
            buffer_x=buffer_x,
            buffer_y=buffer_y,
        )
        output = client.raster_output(delivery=delivery)
        sh_bbox = self._bbox()
        out: list[str] = []
        for product in products:
            resolved: ResolvedRequest = product.metadata["resolved"]
            base = sentinelhub.SentinelHubRequest(
                evalscript=self._read_evalscript(resolved),
                input_data=[self._build_input_data(sentinelhub, resolved)],
                responses=[
                    sentinelhub.SentinelHubRequest.output_response(
                        "default", sentinelhub.MimeType.TIFF
                    )
                ],
                bbox=sh_bbox,
                config=cfg,
            )
            batch_request = client.create(base, input=tiling, output=output)
            # Analyse first so the tile count / cost is known before committing
            # the (potentially continental, costly) job. start_analysis only
            # *starts* the analysis phase, so wait for it to finish before
            # reading cost_PU (otherwise the guard below sees None and is a no-op).
            client.start_analysis(batch_request)
            sentinelhub.monitor_batch_process_analysis(batch_request, client)
            batch_request = client.get_request(batch_request)
            cost_pu = getattr(batch_request, "cost_PU", None)
            logger.info(
                f"Sentinel Hub batch: analysed {product.id!r} (cost_PU={cost_pu})"
            )
            if (
                max_cost_pu is not None
                and cost_pu is not None
                and cost_pu > max_cost_pu
            ):
                raise ValueError(
                    f"batch analysis estimates cost_PU={cost_pu} for {product.id!r}, "
                    f"which exceeds batch_output['max_cost_pu']={max_cost_pu}; raise "
                    "the limit or shrink the request/resolution."
                )
            client.start_job(batch_request)
            sentinelhub.monitor_batch_process_job(batch_request, client)
            out.append(str(url))
        logger.info(f"Sentinel Hub batch: {len(out)} job(s) delivered to S3")
        return out

    def _statistical_evalscript(self, resolved: ResolvedRequest) -> str:
        """Resolve the evalscript for a Statistical request (must emit `dataMask`).

        Args:
            resolved: The resolved request row.

        Returns:
            The evalscript source.

        Raises:
            ValueError: When the evalscript does not declare a `dataMask` band
                (the Statistical API needs it to exclude invalid pixels).
        """
        script = self._read_evalscript(resolved)
        if "dataMask" not in script:
            raise ValueError(
                f"the Statistical API requires the evalscript to emit a "
                f"'dataMask' output band; {resolved.key!r} does not. Use a "
                "stats recipe (kind='stats', e.g. '…-ndvi-stats') or add a "
                "dataMask band to your custom evalscript."
            )
        return script

    def _statistical_interval(self) -> str:
        """The Statistical `aggregation_interval`: from `aggregate=`, else `P1D`."""
        if self._aggregate is not None:
            return interval_for(self._aggregate.freq)
        return "P1D"

    def _fetch_statistical(self, products: list[RemoteProduct]) -> list[Path]:
        """Compute zonal statistics via the Statistical API → a tidy table.

        Builds one `SentinelHubStatistical` request per geometry per product,
        flattens the nested interval→output→band→stats tree into rows, and writes
        one CSV per product (`feature_id` carried for a `FeatureCollection`).

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            The written table paths, one per product.

        Raises:
            ValueError: When no `geometry=` was supplied, or the evalscript lacks
                a `dataMask` band.
        """
        sentinelhub = import_sentinelhub()
        if self._geometry is None:
            raise ValueError(
                "the Statistical API computes zonal stats over a polygon, so it "
                "needs geometry= (a shapely geometry, a GeoJSON mapping, or a "
                "FeatureCollection)."
            )
        cfg = self._auth.config()
        interval = self._statistical_interval()
        geometries = _iter_geometries(self._geometry)
        out: list[Path] = []
        for product in products:
            resolved: ResolvedRequest = product.metadata["resolved"]
            evalscript = self._statistical_evalscript(resolved)
            collection = cdse_collection(resolved.sh_collection, cfg.sh_base_url)
            rows: list[dict] = []
            for feature_id, geom in geometries:
                # Size the sampling grid in pixels from the geometry's WGS84
                # bounds via bbox_to_dimensions (which converts metres -> px with
                # latitude). Passing `resolution=` here would be read in the
                # geometry CRS units (degrees), so a metre value collapses to a
                # single pixel and the server rejects the effective resolution.
                geom_bbox = sentinelhub.BBox(
                    _geometry_bounds(geom), crs=sentinelhub.CRS.WGS84
                )
                size = sentinelhub.bbox_to_dimensions(
                    geom_bbox, resolution=self._resolution
                )
                aggregation = sentinelhub.SentinelHubStatistical.aggregation(
                    evalscript=evalscript,
                    time_interval=self._time_interval(),
                    aggregation_interval=interval,
                    size=size,
                )
                input_kwargs: dict[str, Any] = {}
                if self._maxcc is not None:
                    input_kwargs["maxcc"] = self._maxcc
                request = sentinelhub.SentinelHubStatistical(
                    aggregation=aggregation,
                    input_data=[
                        sentinelhub.SentinelHubStatistical.input_data(
                            collection, **input_kwargs
                        )
                    ],
                    geometry=sentinelhub.Geometry(geom, crs=sentinelhub.CRS.WGS84),
                    calculations=_STAT_CALCULATIONS,
                    config=cfg,
                )
                payload = request.get_data()[0]
                rows.extend(_flatten_statistics(payload, feature_id=feature_id))
            if not rows:
                logger.warning(
                    f"Sentinel Hub statistical: no data returned for "
                    f"{product.id!r} over the request geometry + window "
                    "(empty table written)."
                )
            target = Path(self.root_dir) / f"{_safe_name(product.id)}.csv"
            _stats_frame(rows).to_csv(target, index=False)
            out.append(target)
        logger.info(f"Sentinel Hub statistical: wrote {len(out)} table(s)")
        return out

    def _fetch_batch_statistical(self, products: list[RemoteProduct]) -> list[Path]:
        """Compute zonal stats over a huge `FeatureCollection` via Batch Statistical.

        Submits one async batch-statistical job per product against the features
        uploaded to S3 (`batch_output['input_features']`), monitors it, retrieves
        the per-feature JSON from S3 via `AwsBatchStatisticalResults`, flattens it
        (reusing the C7 flattener, keyed by `feature_id`), and writes one CSV per
        product. Requires `batch_output` with an `input_features` S3 GeoPackage
        and an output bucket.

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            The written table paths, one per product.

        Raises:
            ValueError: When `batch_output` is missing its `input_features` or
                output bucket, or the evalscript lacks a `dataMask` band.
        """
        from sentinelhub.aws import AwsBatchStatisticalResults

        sentinelhub = import_sentinelhub()
        if not self._batch_output:
            raise ValueError(
                "the Batch Statistical plane runs over a FeatureCollection on S3, "
                "so it needs batch_output={'input_features': 's3://…features.gpkg', "
                "'bucket': 's3://…out', 'iam_role_arn': '…'}."
            )
        spec = dict(self._batch_output)
        features_url = spec.pop("input_features", None)
        output_url = spec.pop("bucket", None) or spec.pop("output", None)
        feature_ids = spec.pop("feature_ids", None)
        if not features_url or not output_url:
            raise ValueError(
                "batch-statistical needs batch_output['input_features'] (the S3 "
                "GeoPackage of features) and an output bucket."
            )
        cfg = self._auth.config()
        interval = self._statistical_interval()
        client = sentinelhub.SentinelHubBatchStatistical(config=cfg)
        out: list[Path] = []
        for product in products:
            resolved: ResolvedRequest = product.metadata["resolved"]
            evalscript = self._statistical_evalscript(resolved)
            aggregation = sentinelhub.SentinelHubStatistical.aggregation(
                evalscript=evalscript,
                time_interval=self._time_interval(),
                aggregation_interval=interval,
                resolution=(self._resolution, self._resolution),
            )
            input_kwargs: dict[str, Any] = {}
            if self._maxcc is not None:
                input_kwargs["maxcc"] = self._maxcc
            batch_request = client.create(
                input_features=client.s3_specification(url=features_url, **spec),
                input_data=[
                    sentinelhub.SentinelHubStatistical.input_data(
                        cdse_collection(resolved.sh_collection, cfg.sh_base_url),
                        **input_kwargs,
                    )
                ],
                aggregation=aggregation,
                calculations=_STAT_CALCULATIONS,
                output=client.s3_specification(url=output_url, **spec),
            )
            client.start_analysis(batch_request)
            client.start_job(batch_request)
            sentinelhub.monitor_batch_statistical_job(batch_request, cfg)
            results = AwsBatchStatisticalResults(
                batch_request,
                feature_ids=feature_ids,
                data_folder=str(self.root_dir),
                config=cfg,
            )
            payloads = results.get_data(save_data=True)
            rows: list[dict] = []
            ids = feature_ids if feature_ids is not None else range(len(payloads))
            for feature_id, payload in zip(ids, payloads):
                rows.extend(_flatten_statistics(payload, feature_id=feature_id))
            target = Path(self.root_dir) / f"{_safe_name(product.id)}.csv"
            _stats_frame(rows).to_csv(target, index=False)
            out.append(target)
        logger.info(f"Sentinel Hub batch-statistical: wrote {len(out)} table(s)")
        return out

    def download(
        self,
        progress_bar: bool = True,
        aggregate: AggregationConfig | None = None,
    ) -> list[Any]:
        """Build the request(s), render server-side, and collect the output.

        Args:
            progress_bar: Reserved for parity with the other backends.
            aggregate: Optional aggregation request. Accepted (not rejected)
                because `OUTPUT_KIND` is `"mixed"`; applied per plane in C10.

        Returns:
            The written raster paths / table paths / S3 URIs, depending on the
            resolved plane.
        """
        self._aggregate = aggregate
        plane = self._resolve_plane()
        if aggregate is not None and plane in RASTER_APIS:
            results = self._aggregate_windows(aggregate)
        else:
            # Tabular planes apply aggregate via the Statistical
            # aggregation_interval (see _statistical_interval), so no loop here.
            results = self._api_via_search_fetch()
        logger.info(
            f"Sentinel Hub download: {len(results)} result(s) written to "
            f"{self.root_dir}"
        )
        return results

    def _aggregate_windows(self, aggregate: AggregationConfig) -> list[Any]:
        """Render one output per `aggregate.freq` window over the request span.

        Splits `[start, end]` into `freq` windows and renders the resolved raster
        plane once per window, stamping each local output
        `{key}_{freq}_{YYYYMMDD}.{suffix}` (the `ecmwf` / `cmems` per-window
        shape). S3-delivered planes (async / batch) return their per-window URIs
        unchanged.

        Args:
            aggregate: The aggregation request (its `freq` drives the windows).

        Returns:
            The per-window outputs across every requested key.
        """
        import pandas as pd

        edges = list(
            pd.date_range(self.time.start_date, self.time.end_date, freq=aggregate.freq)
        )
        if not edges or pd.Timestamp(edges[0]) > pd.Timestamp(self.time.start_date):
            edges.insert(0, pd.Timestamp(self.time.start_date))
        results: list[Any] = []
        keys = list(self._resolved)
        try:
            for index, window_start in enumerate(edges):
                if index + 1 < len(edges):
                    # End the day before the next window starts so adjacent
                    # windows don't both claim the shared boundary date (the SH
                    # time_interval is inclusive on both ends).
                    window_end = pd.Timestamp(edges[index + 1]) - pd.Timedelta(days=1)
                    if window_end < pd.Timestamp(window_start):
                        window_end = pd.Timestamp(window_start)
                else:
                    window_end = pd.Timestamp(self.time.end_date)
                self._window_override = (
                    pd.Timestamp(window_start).strftime("%Y-%m-%d"),
                    pd.Timestamp(window_end).strftime("%Y-%m-%d"),
                )
                stamp = pd.Timestamp(window_start).strftime("%Y%m%d")
                produced = self._api_via_search_fetch()
                for key, item in zip(keys, produced):
                    results.append(
                        self._stamp_window_output(key, item, aggregate, stamp)
                    )
        finally:
            self._window_override = None
        return results

    def _stamp_window_output(
        self, key: str, item: Any, aggregate: AggregationConfig, stamp: str
    ) -> Any:
        """Rename a local per-window raster to the stamped name; pass URIs through.

        Args:
            key: The requested collection/recipe key.
            item: A produced output (a local `Path`/str, or an S3 URI string).
            aggregate: The aggregation request (its `freq` labels the file).
            stamp: The window's `YYYYMMDD` start stamp.

        Returns:
            The renamed local path, or the original S3 URI string.
        """
        source = Path(str(item))
        if not source.exists():
            return item
        suffix = source.suffix or ".tif"
        target = (
            Path(self.root_dir) / f"{_safe_name(key)}_{aggregate.freq}_{stamp}{suffix}"
        )
        source.replace(target)
        return target

__init__(start, end, variables, lat_lim, lon_lim, temporal_resolution='daily', path='', fmt='%Y-%m-%d', resolution=10.0, evalscript=None, endpoint=None, mosaicking_order=_DEFAULT_MOSAICKING_ORDER, api=None, geometry=None, maxcc=None, batch_output=None, client_id=None, client_secret=None, profile=None)

Initialise a Sentinel Hub backend instance.

Parameters:

Name	Type	Description	Default
start	str	Inclusive start date string (parsed with fmt).	required
end	str	Inclusive end date string.	required
variables	dict[str, list[str]]	{collection_or_recipe_key: [band, ...]}. A key is a catalog collection (needs an explicit evalscript=) or an evalscript recipe (pins its collection); an empty band list falls back to the recipe / collection default bands.	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; the render window is start/end.	'daily'
path	Path | str	Output directory (created by the parent class).	''
fmt	str	strptime format for start / end.	'%Y-%m-%d'
resolution	float	Output pixel size in metres (→ bbox_to_dimensions).	10.0
evalscript	str | None	A custom evalscript — an inline V3 JS string or a path to a .js file — that bypasses the recipe lookup (the collection then comes from the variables key). None uses the recipe's bundled .js.	None
endpoint	str | None	Endpoint alias ("cdse", "commercial") or a full base URL. Defaults to CDSE-free.	None
mosaicking_order	str	Per-pixel scene selection passed to input_data — "mostRecent" (default) / "leastRecent" / "leastCC" (the MosaickingOrder enum).	_DEFAULT_MOSAICKING_ORDER
api	str | None	The request plane ("process" / "async" / "batch" / "statistical" / "batch-statistical"), or None to auto-select by request size + whether geometry= was supplied.	None
geometry	Any	A shapely geometry / GeoJSON mapping / FeatureCollection for the Statistical planes (zonal stats over the polygon(s)).	None
maxcc	float | None	Optional maximum cloud cover (0–1) passed to input_data (optical collections only).	None
batch_output	dict[str, Any] | None	S3 delivery spec ({"bucket": ..., "iam_role_arn": ...}) for the Batch planes.	None
client_id	str | None	OAuth client id (else SENTINELHUB_CLIENT_ID).	None
client_secret	str | None	OAuth client secret (else SENTINELHUB_CLIENT_SECRET).	None
profile	str | None	A saved SHConfig profile name (else SENTINELHUB_PROFILE).	None

Raises:

Type	Description
ValueError	When api is an unknown plane, or mosaicking_order is not a recognised value.

Source code in src/earthlens/sentinel_hub/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 = "",
    fmt: str = "%Y-%m-%d",
    resolution: float = 10.0,
    evalscript: str | None = None,
    endpoint: str | None = None,
    mosaicking_order: str = _DEFAULT_MOSAICKING_ORDER,
    api: str | None = None,
    geometry: Any = None,
    maxcc: float | None = None,
    batch_output: dict[str, Any] | None = None,
    client_id: str | None = None,
    client_secret: str | None = None,
    profile: str | None = None,
):
    """Initialise a Sentinel Hub backend instance.

    Args:
        start: Inclusive start date string (parsed with `fmt`).
        end: Inclusive end date string.
        variables: `{collection_or_recipe_key: [band, ...]}`. A key is a
            catalog **collection** (needs an explicit `evalscript=`) or an
            **evalscript recipe** (pins its collection); an empty band list
            falls back to the recipe / collection default bands.
        lat_lim: `[lat_min, lat_max]` in degrees.
        lon_lim: `[lon_min, lon_max]` in degrees.
        temporal_resolution: Advisory cadence label; the render window is
            `start`/`end`.
        path: Output directory (created by the parent class).
        fmt: `strptime` format for `start` / `end`.
        resolution: Output pixel size in metres (→ `bbox_to_dimensions`).
        evalscript: A custom evalscript — an inline V3 JS string or a path to
            a `.js` file — that bypasses the recipe lookup (the collection
            then comes from the `variables` key). `None` uses the recipe's
            bundled `.js`.
        endpoint: Endpoint alias (`"cdse"`, `"commercial"`) or a full base
            URL. Defaults to CDSE-free.
        mosaicking_order: Per-pixel scene selection passed to
            `input_data` — `"mostRecent"` (default) / `"leastRecent"` /
            `"leastCC"` (the `MosaickingOrder` enum).
        api: The request plane (`"process"` / `"async"` / `"batch"` /
            `"statistical"` / `"batch-statistical"`), or `None` to
            auto-select by request size + whether `geometry=` was supplied.
        geometry: A shapely geometry / GeoJSON mapping / `FeatureCollection`
            for the Statistical planes (zonal stats over the polygon(s)).
        maxcc: Optional maximum cloud cover (0–1) passed to `input_data`
            (optical collections only).
        batch_output: S3 delivery spec (`{"bucket": ..., "iam_role_arn": ...}`)
            for the Batch planes.
        client_id: OAuth client id (else `SENTINELHUB_CLIENT_ID`).
        client_secret: OAuth client secret (else `SENTINELHUB_CLIENT_SECRET`).
        profile: A saved `SHConfig` profile name (else `SENTINELHUB_PROFILE`).

    Raises:
        ValueError: When `api` is an unknown plane, or `mosaicking_order` is
            not a recognised value.
    """
    validate_api(api)
    if mosaicking_order not in _VALID_MOSAICKING_ORDERS:
        raise ValueError(
            f"mosaicking_order must be one of {list(_VALID_MOSAICKING_ORDERS)}, "
            f"got {mosaicking_order!r}."
        )
    self._resolution = resolution
    self._evalscript = evalscript
    self._endpoint = endpoint
    self._mosaicking_order = mosaicking_order
    self._api_mode = api
    self._geometry = geometry
    self._maxcc = maxcc
    self._batch_output = batch_output
    env_creds = SentinelHubCredentials.from_env()
    self._credentials = SentinelHubCredentials(
        client_id=client_id or env_creds.client_id,
        client_secret=client_secret or env_creds.client_secret,
        profile=profile or env_creds.profile,
    )
    # Stored before super().__init__ because the base constructor calls
    # _initialize() (which needs the request) before it sets self.vars.
    self._variables = variables
    self._catalog: Catalog | None = None
    self._auth: SentinelHubAuth | None = None
    self._resolved: dict[str, ResolvedRequest] = {}
    # Aggregation request captured by download(); applied per plane.
    self._aggregate: AggregationConfig | None = None
    # Per-window time override set by the aggregate= render loop (C10).
    self._window_override: tuple[str, str] | None = None
    # Memoised resolved plane (deterministic from instance state; avoids a
    # second bbox_to_dimensions call across download() -> _fetch()).
    self._plane: str | None = None
    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, aggregate=None)

Build the request(s), render server-side, and collect the output.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Reserved for parity with the other backends.	True
aggregate	AggregationConfig | None	Optional aggregation request. Accepted (not rejected) because OUTPUT_KIND is "mixed"; applied per plane in C10.	None

Returns:

Type	Description
list[Any]	The written raster paths / table paths / S3 URIs, depending on the
list[Any]	resolved plane.

Source code in src/earthlens/sentinel_hub/backend.py

def download(
    self,
    progress_bar: bool = True,
    aggregate: AggregationConfig | None = None,
) -> list[Any]:
    """Build the request(s), render server-side, and collect the output.

    Args:
        progress_bar: Reserved for parity with the other backends.
        aggregate: Optional aggregation request. Accepted (not rejected)
            because `OUTPUT_KIND` is `"mixed"`; applied per plane in C10.

    Returns:
        The written raster paths / table paths / S3 URIs, depending on the
        resolved plane.
    """
    self._aggregate = aggregate
    plane = self._resolve_plane()
    if aggregate is not None and plane in RASTER_APIS:
        results = self._aggregate_windows(aggregate)
    else:
        # Tabular planes apply aggregate via the Statistical
        # aggregation_interval (see _statistical_interval), so no loop here.
        results = self._api_via_search_fetch()
    logger.info(
        f"Sentinel Hub download: {len(results)} result(s) written to "
        f"{self.root_dir}"
    )
    return results

search(limit=100)

Query the Sentinel Hub Catalog API for scenes intersecting the request.

A real STAC-style search (distinct from the internal :meth:_search fetch-planner): returns one :class:RemoteProduct per catalog item that intersects the request bbox + window, so a caller can enumerate coverage before rendering. An empty result is an empty list (not an error).

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of items to return per requested collection.	100

Returns:

Type	Description
list[RemoteProduct]	One product per catalog item (id + datetime + geometry on metadata).

Source code in src/earthlens/sentinel_hub/backend.py

def search(self, limit: int = 100) -> list[RemoteProduct]:
    """Query the Sentinel Hub Catalog API for scenes intersecting the request.

    A real STAC-style search (distinct from the internal :meth:`_search`
    fetch-planner): returns one :class:`RemoteProduct` per catalog item that
    intersects the request bbox + window, so a caller can enumerate coverage
    before rendering. An empty result is an empty list (not an error).

    Args:
        limit: Maximum number of items to return per requested collection.

    Returns:
        One product per catalog item (id + datetime + geometry on metadata).
    """
    sentinelhub = import_sentinelhub()
    cfg = self._auth.config()
    catalog = sentinelhub.SentinelHubCatalog(config=cfg)
    sh_bbox = self._bbox()
    time_interval = self._time_interval()
    products: list[RemoteProduct] = []
    seen_collections: set[str] = set()
    for key, resolved in self._resolved.items():
        if resolved.sh_collection in seen_collections:
            continue
        seen_collections.add(resolved.sh_collection)
        collection = cdse_collection(resolved.sh_collection, cfg.sh_base_url)
        for item in catalog.search(
            collection, bbox=sh_bbox, time=time_interval, limit=limit
        ):
            properties = item.get("properties", {})
            products.append(
                RemoteProduct(
                    id=item.get("id"),
                    metadata={
                        "key": key,
                        "collection": resolved.sh_collection,
                        "datetime": properties.get("datetime"),
                        "geometry": item.get("geometry"),
                    },
                )
            )
    return products

earthlens.sentinel_hub.catalog

Two-layer catalog for the Sentinel Hub backend: collections + evalscripts.

Sentinel Hub renders from an evalscript — there is no granule to fetch — so the "catalog" has two layers (the headline design decision, G1):

• Collections — the underlying Sentinel Hub DataCollections (SENTINEL2_L2A, SENTINEL1_IW, SENTINEL3_OLCI, …) with their bands. A request names a collection (plus an explicit evalscript=).
• Evalscript recipes — a curated library of named, parametric .js files bundled under evalscripts/ (sentinel-2-l2a-ndvi, …-true-colour, …), each fixing a base collection + the render logic. A recipe is either a "render" recipe (writes a raster) or a "stats" recipe (emits the dataMask band the Statistical API requires). A request that names a recipe gets that fixed evalscript over its collection.

The catalog ships as a directory of YAML files at src/earthlens/sentinel_hub/catalog/ (collections.yaml, recipes.yaml, plus _index.yaml carrying the informational available_collections: index rebuilt by tools/sentinel_hub/refresh_sh_catalog.py). The loader unions the per-file collections: / recipes: blocks into one :class:Catalog (a key declared in two files is a load-time error) and caches on (path, mtime_ns), the same way the openEO / GEE / STAC catalogs do. The evalscript .js files live alongside under evalscripts/ and are read at request time (not parsed here).

The bundled catalog directory lives at :data:CATALOG_PATH and the evalscript directory at :data:EVALSCRIPTS_PATH; tests can monkey-patch either to redirect the loader at a temporary directory.

Band

Bases: BaseModel

Per-band metadata for one band of a Sentinel Hub collection.

Frozen value object; the band name is the parent mapping key and is not repeated in the body. Every field is optional because band metadata is uneven across collections.

Attributes:

Name	Type	Description
common_name	str | None	Common name ("red", "nir"), or None.
description	str | None	Human description of the band, or None.
units	str | None	Physical unit string, or None.
resolution	float | None	Native ground sample distance in metres, or None.
center_wavelength	float | None	Central wavelength in micrometres (optical), or None.

Source code in src/earthlens/sentinel_hub/catalog.py

class Band(BaseModel):
    """Per-band metadata for one band of a Sentinel Hub collection.

    Frozen value object; the band name is the parent mapping key and is not
    repeated in the body. Every field is optional because band metadata is
    uneven across collections.

    Attributes:
        common_name: Common name (`"red"`, `"nir"`), or `None`.
        description: Human description of the band, or `None`.
        units: Physical unit string, or `None`.
        resolution: Native ground sample distance in metres, or `None`.
        center_wavelength: Central wavelength in micrometres (optical), or `None`.
    """

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

    common_name: str | None = None
    description: str | None = None
    units: str | None = None
    resolution: float | None = None
    center_wavelength: float | None = None

Catalog

Bases: AbstractCatalog

YAML-backed catalog of Sentinel Hub collections and evalscript recipes.

Reads every *.yaml under :data:CATALOG_PATH on construction and merges them into typed :class:Collection / :class:EvalscriptRecipe models. Collections are stored under the inherited :attr:datasets field (keyed by logical key); recipes live in :attr:recipes; the informational index lives in :attr:available_collections.

Attributes:

Name	Type	Description
datasets	dict[str, Collection]	Logical collection key → :class:Collection.
recipes	dict[str, EvalscriptRecipe]	Recipe key → :class:EvalscriptRecipe.
available_collections	list[str]	Every collection the backend can render (refreshed index; informational).

Examples:

• A recipe resolves to its bundled evalscript; a collection to a bare bind:

  >>> from earthlens.sentinel_hub import Catalog
  >>> cat = Catalog()
  >>> cat.is_recipe("sentinel-2-l2a-ndvi")
  True
  >>> cat.get_collection("sentinel-2-l2a").sh_collection
  'SENTINEL2_L2A'
  

• Resolving normalises both layers to one shape:

  >>> from earthlens.sentinel_hub import Catalog
  >>> r = Catalog().resolve("sentinel-2-l2a-ndvi")
  >>> r.sh_collection, r.evalscript, r.is_recipe
  ('SENTINEL2_L2A', 'ndvi.js', True)
  

Source code in src/earthlens/sentinel_hub/catalog.py

class Catalog(AbstractCatalog):
    """YAML-backed catalog of Sentinel Hub collections and evalscript recipes.

    Reads every `*.yaml` under :data:`CATALOG_PATH` on construction and merges
    them into typed :class:`Collection` / :class:`EvalscriptRecipe` models.
    Collections are stored under the inherited :attr:`datasets` field (keyed by
    logical key); recipes live in :attr:`recipes`; the informational index lives
    in :attr:`available_collections`.

    Attributes:
        datasets: Logical collection key → :class:`Collection`.
        recipes: Recipe key → :class:`EvalscriptRecipe`.
        available_collections: Every collection the backend can render
            (refreshed index; informational).

    Examples:
        - A recipe resolves to its bundled evalscript; a collection to a bare bind:
            ```python
            >>> from earthlens.sentinel_hub import Catalog
            >>> cat = Catalog()
            >>> cat.is_recipe("sentinel-2-l2a-ndvi")
            True
            >>> cat.get_collection("sentinel-2-l2a").sh_collection
            'SENTINEL2_L2A'

            ```
        - Resolving normalises both layers to one shape:
            ```python
            >>> from earthlens.sentinel_hub import Catalog
            >>> r = Catalog().resolve("sentinel-2-l2a-ndvi")
            >>> r.sh_collection, r.evalscript, r.is_recipe
            ('SENTINEL2_L2A', 'ndvi.js', True)

            ```
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    _catalog_kind: str = "Sentinel Hub catalog"

    datasets: dict[str, Collection] = Field(default_factory=dict)
    recipes: dict[str, EvalscriptRecipe] = Field(default_factory=dict)
    available_collections: list[str] = Field(default_factory=list)

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

        `Catalog()` with no args reads the bundled `catalog/` directory through
        the `(path, mtime)`-keyed cache. If the caller passed `datasets=` or
        `recipes=`, the disk read is skipped (in-memory catalogs for tests).
        The base `available_datasets` field is mirrored from
        `available_collections` so the index is discoverable through the
        `AbstractCatalog` contract, then `super().model_post_init` populates
        `catalog` from `get_catalog()`.

        Raises:
            ValueError: When auto-loading, propagates the loader's errors.
        """
        if not (self.datasets or self.recipes):
            loaded = Catalog.load()
            self.datasets = loaded.datasets
            self.recipes = loaded.recipes
            self.available_collections = loaded.available_collections
        if not self.available_datasets:
            self.available_datasets = list(self.available_collections)
        super().model_post_init(__context)

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

        Args:
            catalog_path: Catalog directory or single `*.yaml` file. Defaults to
                module-level :data:`CATALOG_PATH`.

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

        Raises:
            ValueError: Propagated from the loader.
        """
        catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
        collections, recipes, av_cols = _load_catalog_data(catalog_path)
        return cls(
            datasets=dict(collections),
            recipes=dict(recipes),
            available_collections=list(av_cols),
        )

    def get_catalog(self) -> dict[str, Collection]:
        """Return the curated collection map (logical key → :class:`Collection`)."""
        return self.datasets

    def get_collection(self, collection_key: str) -> Collection:
        """Return the :class:`Collection` for `collection_key` (did-you-mean on miss).

        Alias of the inherited :meth:`get_dataset`, named for clarity.

        Args:
            collection_key: Logical collection key.

        Returns:
            The matching :class:`Collection`.

        Raises:
            ValueError: If the key is unknown (message suggests the closest key).
        """
        return self.get_dataset(collection_key)

    def get_recipe(self, recipe_key: str) -> EvalscriptRecipe:
        """Return the :class:`EvalscriptRecipe` for `recipe_key` (did-you-mean on miss).

        Args:
            recipe_key: Recipe key.

        Returns:
            The matching :class:`EvalscriptRecipe`.

        Raises:
            ValueError: If the key is unknown (message suggests the closest key).
        """
        import difflib

        try:
            return self.recipes[recipe_key]
        except KeyError:
            close = difflib.get_close_matches(recipe_key, self.recipes, n=1)
            hint = f" Did you mean {close[0]!r}?" if close else ""
            raise ValueError(
                f"{recipe_key!r} is not a known recipe. "
                f"Known recipes: {sorted(self.recipes)}.{hint}"
            ) from None

    def is_recipe(self, key: str) -> bool:
        """Return whether `key` names a curated evalscript recipe.

        Args:
            key: A logical collection or recipe key.

        Returns:
            `True` when `key` is a recipe.
        """
        return key in self.recipes

    def resolve(self, key: str) -> ResolvedRequest:
        """Resolve a collection-or-recipe key to a uniform :class:`ResolvedRequest`.

        A recipe resolves to its bundled evalscript + bands + kind; a plain
        collection resolves with `evalscript=None` and its default bands (the
        backend then requires an explicit `evalscript=`).

        Args:
            key: A logical collection or recipe key.

        Returns:
            The normalised :class:`ResolvedRequest` the backend builds from.

        Raises:
            ValueError: If `key` is neither a known recipe nor collection
                (message suggests the closest key across both layers).

        Examples:
            - A recipe carries its evalscript + kind:
                ```python
                >>> from earthlens.sentinel_hub import Catalog
                >>> r = Catalog().resolve("sentinel-2-l2a-ndvi-stats")
                >>> r.kind, r.evalscript
                ('stats', 'ndvi_stats.js')

                ```
        """
        if key in self.recipes:
            recipe = self.recipes[key]
            return ResolvedRequest(
                key=key,
                sh_collection=recipe.base_collection,
                bands=list(recipe.bands),
                evalscript=recipe.evalscript,
                output_bands=recipe.output_bands,
                is_recipe=True,
                kind=recipe.kind,
            )
        if key in self.datasets:
            collection = self.datasets[key]
            return ResolvedRequest(
                key=key,
                sh_collection=collection.sh_collection,
                bands=collection.effective_bands,
                evalscript=None,
                is_recipe=False,
                kind="render",
            )
        import difflib

        known = sorted(set(self.recipes) | set(self.datasets))
        close = difflib.get_close_matches(key, known, n=1)
        hint = f" Did you mean {close[0]!r}?" if close else ""
        raise ValueError(
            f"{key!r} is not a known Sentinel Hub collection or recipe. "
            f"Known keys: {known}.{hint}"
        )

get_catalog()

Return the curated collection map (logical key → :class:Collection).

Source code in src/earthlens/sentinel_hub/catalog.py

def get_catalog(self) -> dict[str, Collection]:
    """Return the curated collection map (logical key → :class:`Collection`)."""
    return self.datasets

get_collection(collection_key)

Return the :class:Collection for collection_key (did-you-mean on miss).

Alias of the inherited :meth:get_dataset, named for clarity.

Parameters:

Name	Type	Description	Default
collection_key	str	Logical collection key.	required

Returns:

Type	Description
Collection	The matching :class:Collection.

Raises:

Type	Description
ValueError	If the key is unknown (message suggests the closest key).

Source code in src/earthlens/sentinel_hub/catalog.py

def get_collection(self, collection_key: str) -> Collection:
    """Return the :class:`Collection` for `collection_key` (did-you-mean on miss).

    Alias of the inherited :meth:`get_dataset`, named for clarity.

    Args:
        collection_key: Logical collection key.

    Returns:
        The matching :class:`Collection`.

    Raises:
        ValueError: If the key is unknown (message suggests the closest key).
    """
    return self.get_dataset(collection_key)

get_recipe(recipe_key)

Return the :class:EvalscriptRecipe for recipe_key (did-you-mean on miss).

Parameters:

Name	Type	Description	Default
recipe_key	str	Recipe key.	required

Returns:

Type	Description
EvalscriptRecipe	The matching :class:EvalscriptRecipe.

Raises:

Type	Description
ValueError	If the key is unknown (message suggests the closest key).

Source code in src/earthlens/sentinel_hub/catalog.py

def get_recipe(self, recipe_key: str) -> EvalscriptRecipe:
    """Return the :class:`EvalscriptRecipe` for `recipe_key` (did-you-mean on miss).

    Args:
        recipe_key: Recipe key.

    Returns:
        The matching :class:`EvalscriptRecipe`.

    Raises:
        ValueError: If the key is unknown (message suggests the closest key).
    """
    import difflib

    try:
        return self.recipes[recipe_key]
    except KeyError:
        close = difflib.get_close_matches(recipe_key, self.recipes, n=1)
        hint = f" Did you mean {close[0]!r}?" if close else ""
        raise ValueError(
            f"{recipe_key!r} is not a known recipe. "
            f"Known recipes: {sorted(self.recipes)}.{hint}"
        ) from None

is_recipe(key)

Return whether key names a curated evalscript recipe.

Parameters:

Name	Type	Description	Default
key	str	A logical collection or recipe key.	required

Returns:

Type	Description
bool	True when key is a recipe.

Source code in src/earthlens/sentinel_hub/catalog.py

def is_recipe(self, key: str) -> bool:
    """Return whether `key` names a curated evalscript recipe.

    Args:
        key: A logical collection or recipe key.

    Returns:
        `True` when `key` is a recipe.
    """
    return key in self.recipes

load(catalog_path=None) classmethod

Read the Sentinel Hub catalog from disk (cached).

Parameters:

Name	Type	Description	Default
catalog_path	Path | None	Catalog directory or single *.yaml file. Defaults to module-level :data:CATALOG_PATH.	None

Returns:

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

Raises:

Type	Description
ValueError	Propagated from the loader.

Source code in src/earthlens/sentinel_hub/catalog.py

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

    Args:
        catalog_path: Catalog directory or single `*.yaml` file. Defaults to
            module-level :data:`CATALOG_PATH`.

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

    Raises:
        ValueError: Propagated from the loader.
    """
    catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
    collections, recipes, av_cols = _load_catalog_data(catalog_path)
    return cls(
        datasets=dict(collections),
        recipes=dict(recipes),
        available_collections=list(av_cols),
    )

model_post_init(__context)

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

Catalog() with no args reads the bundled catalog/ directory through the (path, mtime)-keyed cache. If the caller passed datasets= or recipes=, the disk read is skipped (in-memory catalogs for tests). The base available_datasets field is mirrored from available_collections so the index is discoverable through the AbstractCatalog contract, then super().model_post_init populates catalog from get_catalog().

Raises:

Type	Description
ValueError	When auto-loading, propagates the loader's errors.

Source code in src/earthlens/sentinel_hub/catalog.py

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

    `Catalog()` with no args reads the bundled `catalog/` directory through
    the `(path, mtime)`-keyed cache. If the caller passed `datasets=` or
    `recipes=`, the disk read is skipped (in-memory catalogs for tests).
    The base `available_datasets` field is mirrored from
    `available_collections` so the index is discoverable through the
    `AbstractCatalog` contract, then `super().model_post_init` populates
    `catalog` from `get_catalog()`.

    Raises:
        ValueError: When auto-loading, propagates the loader's errors.
    """
    if not (self.datasets or self.recipes):
        loaded = Catalog.load()
        self.datasets = loaded.datasets
        self.recipes = loaded.recipes
        self.available_collections = loaded.available_collections
    if not self.available_datasets:
        self.available_datasets = list(self.available_collections)
    super().model_post_init(__context)

resolve(key)

Resolve a collection-or-recipe key to a uniform :class:ResolvedRequest.

A recipe resolves to its bundled evalscript + bands + kind; a plain collection resolves with evalscript=None and its default bands (the backend then requires an explicit evalscript=).

Parameters:

Name	Type	Description	Default
key	str	A logical collection or recipe key.	required

Returns:

Type	Description
ResolvedRequest	The normalised :class:ResolvedRequest the backend builds from.

Raises:

Type	Description
ValueError	If key is neither a known recipe nor collection (message suggests the closest key across both layers).

Examples:

• A recipe carries its evalscript + kind:

  >>> from earthlens.sentinel_hub import Catalog
  >>> r = Catalog().resolve("sentinel-2-l2a-ndvi-stats")
  >>> r.kind, r.evalscript
  ('stats', 'ndvi_stats.js')
  

Source code in src/earthlens/sentinel_hub/catalog.py

def resolve(self, key: str) -> ResolvedRequest:
    """Resolve a collection-or-recipe key to a uniform :class:`ResolvedRequest`.

    A recipe resolves to its bundled evalscript + bands + kind; a plain
    collection resolves with `evalscript=None` and its default bands (the
    backend then requires an explicit `evalscript=`).

    Args:
        key: A logical collection or recipe key.

    Returns:
        The normalised :class:`ResolvedRequest` the backend builds from.

    Raises:
        ValueError: If `key` is neither a known recipe nor collection
            (message suggests the closest key across both layers).

    Examples:
        - A recipe carries its evalscript + kind:
            ```python
            >>> from earthlens.sentinel_hub import Catalog
            >>> r = Catalog().resolve("sentinel-2-l2a-ndvi-stats")
            >>> r.kind, r.evalscript
            ('stats', 'ndvi_stats.js')

            ```
    """
    if key in self.recipes:
        recipe = self.recipes[key]
        return ResolvedRequest(
            key=key,
            sh_collection=recipe.base_collection,
            bands=list(recipe.bands),
            evalscript=recipe.evalscript,
            output_bands=recipe.output_bands,
            is_recipe=True,
            kind=recipe.kind,
        )
    if key in self.datasets:
        collection = self.datasets[key]
        return ResolvedRequest(
            key=key,
            sh_collection=collection.sh_collection,
            bands=collection.effective_bands,
            evalscript=None,
            is_recipe=False,
            kind="render",
        )
    import difflib

    known = sorted(set(self.recipes) | set(self.datasets))
    close = difflib.get_close_matches(key, known, n=1)
    hint = f" Did you mean {close[0]!r}?" if close else ""
    raise ValueError(
        f"{key!r} is not a known Sentinel Hub collection or recipe. "
        f"Known keys: {known}.{hint}"
    )

Collection

Bases: BaseModel

One curated Sentinel Hub data collection, addressed by a logical key.

Attributes:

Name	Type	Description
sh_collection	str	The sentinelhub.DataCollection member name this key binds to ("SENTINEL2_L2A", "SENTINEL1_IW", …).
bands	dict[str, Band]	Band name → :class:Band metadata for the bands the collection exposes.
default_bands	list[str]	Bands used when the request names none. Falls back to every key of bands when empty.
cadence	str | None	Native revisit cadence label, or None.
resolution	float | None	Native ground sample distance in metres, or None.
extent	Extent | None	Spatial/temporal coverage, or None.
description	str | None	One-line human summary, or None.

Source code in src/earthlens/sentinel_hub/catalog.py

class Collection(BaseModel):
    """One curated Sentinel Hub data collection, addressed by a logical key.

    Attributes:
        sh_collection: The `sentinelhub.DataCollection` member name this key
            binds to (`"SENTINEL2_L2A"`, `"SENTINEL1_IW"`, …).
        bands: Band name → :class:`Band` metadata for the bands the collection
            exposes.
        default_bands: Bands used when the request names none. Falls back to
            every key of `bands` when empty.
        cadence: Native revisit cadence label, or `None`.
        resolution: Native ground sample distance in metres, or `None`.
        extent: Spatial/temporal coverage, or `None`.
        description: One-line human summary, or `None`.
    """

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

    sh_collection: str
    bands: dict[str, Band] = Field(default_factory=dict)
    default_bands: list[str] = Field(default_factory=list)
    cadence: str | None = None
    resolution: float | None = None
    extent: Extent | None = None
    description: str | None = None

    @property
    def effective_bands(self) -> list[str]:
        """The band names to request by default (`default_bands`, else all bands)."""
        return list(self.default_bands or list(self.bands))

effective_bands property

The band names to request by default (default_bands, else all bands).

EvalscriptRecipe

Bases: BaseModel

One curated evalscript recipe fixing a base collection + a .js file.

Attributes:

Name	Type	Description
base_collection	str	The DataCollection member name the recipe renders (e.g. "SENTINEL2_L2A").
evalscript	str	The bundled .js filename under evalscripts/ (e.g. "ndvi.js").
bands	list[str]	The collection bands the evalscript consumes (informational + used by the refresh/validate tool).
output_bands	int	The number of bands the evalscript writes.
kind	str	"render" (writes a raster) or "stats" (emits a dataMask band for the Statistical API).
description	str | None	One-line human summary, or None.

Source code in src/earthlens/sentinel_hub/catalog.py

class EvalscriptRecipe(BaseModel):
    """One curated evalscript recipe fixing a base collection + a `.js` file.

    Attributes:
        base_collection: The `DataCollection` member name the recipe renders
            (e.g. `"SENTINEL2_L2A"`).
        evalscript: The bundled `.js` filename under `evalscripts/`
            (e.g. `"ndvi.js"`).
        bands: The collection bands the evalscript consumes (informational +
            used by the refresh/validate tool).
        output_bands: The number of bands the evalscript writes.
        kind: `"render"` (writes a raster) or `"stats"` (emits a `dataMask` band
            for the Statistical API).
        description: One-line human summary, or `None`.
    """

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

    base_collection: str
    evalscript: str
    bands: list[str] = Field(default_factory=list)
    output_bands: int = 1
    kind: str = "render"
    description: str | None = None

    @field_validator("kind")
    @classmethod
    def _check_kind(cls, value: str) -> str:
        """Validate `kind` is one of the recognised recipe kinds.

        Args:
            value: The recipe kind.

        Returns:
            The validated kind.

        Raises:
            ValueError: If `kind` is not `"render"` or `"stats"`.
        """
        if value not in _RECIPE_KINDS:
            raise ValueError(
                f"recipe kind must be one of {sorted(_RECIPE_KINDS)}, got {value!r}."
            )
        return value

Extent

Bases: BaseModel

Spatial/temporal coverage of a Sentinel Hub collection.

Attributes:

Name	Type	Description
start_date	str | None	First available date (YYYY-MM-DD), or None.
end_date	str | None	Last available date, or None for a rolling collection.
bbox	tuple[float, float, float, float] | None	[west, south, east, north] in EPSG:4326, or None for global.

Source code in src/earthlens/sentinel_hub/catalog.py

class Extent(BaseModel):
    """Spatial/temporal coverage of a Sentinel Hub collection.

    Attributes:
        start_date: First available date (`YYYY-MM-DD`), or `None`.
        end_date: Last available date, or `None` for a rolling collection.
        bbox: `[west, south, east, north]` in EPSG:4326, or `None` for global.
    """

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

    start_date: str | None = None
    end_date: str | None = None
    bbox: tuple[float, float, float, float] | None = None

ResolvedRequest

Bases: BaseModel

The uniform shape a resolved collection-or-recipe key takes.

Both a plain collection and a recipe resolve to this so the backend's request-builder consumes one type. A plain collection resolves with evalscript=None (the backend must then have an explicit evalscript=); a recipe carries its bundled .js filename.

Attributes:

Name	Type	Description
key	str	The logical key the request named.
sh_collection	str	The DataCollection member name to bind + render.
bands	list[str]	The bands to request (request override applied by the backend).
evalscript	str | None	The bundled .js filename, or None for a plain collection.
output_bands	int	Number of output bands the evalscript writes.
is_recipe	bool	Whether the key named a recipe (vs a plain collection).
kind	str	"render" or "stats" (recipes only; "render" for a plain collection).

Source code in src/earthlens/sentinel_hub/catalog.py

class ResolvedRequest(BaseModel):
    """The uniform shape a resolved collection-or-recipe key takes.

    Both a plain collection and a recipe resolve to this so the backend's
    request-builder consumes one type. A plain collection resolves with
    `evalscript=None` (the backend must then have an explicit `evalscript=`); a
    recipe carries its bundled `.js` filename.

    Attributes:
        key: The logical key the request named.
        sh_collection: The `DataCollection` member name to bind + render.
        bands: The bands to request (request override applied by the backend).
        evalscript: The bundled `.js` filename, or `None` for a plain collection.
        output_bands: Number of output bands the evalscript writes.
        is_recipe: Whether the key named a recipe (vs a plain collection).
        kind: `"render"` or `"stats"` (recipes only; `"render"` for a plain
            collection).
    """

    model_config = ConfigDict(frozen=True)

    key: str
    sh_collection: str
    bands: list[str] = Field(default_factory=list)
    evalscript: str | None = None
    output_bands: int = 1
    is_recipe: bool = False
    kind: str = "render"

clear_catalog_cache()

Empty the module-level catalog parse cache (for tests that rewrite YAML).

Source code in src/earthlens/sentinel_hub/catalog.py

def clear_catalog_cache() -> None:
    """Empty the module-level catalog parse cache (for tests that rewrite YAML)."""
    _CATALOG_CACHE.clear()

read_evalscript(filename)

Read a bundled evalscript .js file by name.

Parameters:

Name	Type	Description	Default
filename	str	The .js filename (e.g. "ndvi.js") under :data:EVALSCRIPTS_PATH.	required

Returns:

Type	Description
str	The evalscript source.

Raises:

Type	Description
FileNotFoundError	When no such bundled evalscript exists.

Source code in src/earthlens/sentinel_hub/catalog.py

def read_evalscript(filename: str) -> str:
    """Read a bundled evalscript `.js` file by name.

    Args:
        filename: The `.js` filename (e.g. `"ndvi.js"`) under
            :data:`EVALSCRIPTS_PATH`.

    Returns:
        The evalscript source.

    Raises:
        FileNotFoundError: When no such bundled evalscript exists.
    """
    path = EVALSCRIPTS_PATH / filename
    if not path.is_file():
        available = sorted(p.name for p in EVALSCRIPTS_PATH.glob("*.js"))
        raise FileNotFoundError(
            f"bundled evalscript {filename!r} not found under {EVALSCRIPTS_PATH}. "
            f"Available: {available}."
        )
    return path.read_text(encoding="utf-8")

earthlens.sentinel_hub.auth

OAuth2 client-credentials authentication for the Sentinel Hub backend.

Hosts :class:SentinelHubAuth, a thin wrapper that builds and holds a sentinelhub.SHConfig. Unlike the openEO backend's interactive OIDC device flow, Sentinel Hub authenticates non-interactively with an OAuth2 client-credentials pair (a client_id / client_secret minted in the CDSE Dashboard) — closest to GEE's service-account model. sentinelhub-py mints and refreshes the bearer token internally from the config, so there is no explicit login step here: :meth:SentinelHubAuth.configure only assembles the SHConfig (base URL, token URL, credentials).

The credential resolution order in :meth:SentinelHubAuth.configure is:

1. environment — SENTINELHUB_CLIENT_ID / SENTINELHUB_CLIENT_SECRET (with SH_CLIENT_ID / SH_CLIENT_SECRET accepted as a sentinelhub-py-native fallback).
2. kwargs — an explicit client_id / client_secret on the credentials object (these win over the environment).
3. saved profile — a named SHConfig profile written earlier with SHConfig.save(profile) (used when neither of the above is set).

endpoint= switches the CDSE-free deployment (the default) and the commercial one (services.sentinel-hub.com, a different token URL). Any failure is wrapped as :class:AuthenticationError with a pointer at the CDSE Dashboard OAuth-client page, never a raw sentinelhub exception.

AuthenticationError

Bases: AuthenticationError

Raised when the Sentinel Hub SHConfig cannot be assembled.

Wraps the missing-credentials case with an actionable message pointing at the CDSE Dashboard. A subclass of the cross-backend :class:earthlens.base.AuthenticationError so callers can catch every backend's auth failure with one except clause.

Source code in src/earthlens/sentinel_hub/auth.py

class AuthenticationError(_BaseAuthenticationError):
    """Raised when the Sentinel Hub `SHConfig` cannot be assembled.

    Wraps the missing-credentials case with an actionable message pointing at
    the CDSE Dashboard. A subclass of the cross-backend
    :class:`earthlens.base.AuthenticationError` so callers can catch every
    backend's auth failure with one `except` clause.
    """

SentinelHubAuth

Bases: AbstractAuth[SentinelHubCredentials]

Assemble and hold a Sentinel Hub SHConfig (non-interactive OAuth2).

Conforms to the cross-backend :class:earthlens.base.AbstractAuth contract: :meth:configure builds the SHConfig lazily and idempotently; :meth:is_authenticated is the cheap predicate; and :meth:config returns the configured SHConfig (configuring on first use). The token itself is minted and refreshed by sentinelhub-py from the config on the first request — there is no interactive flow.

Parameters:

Name	Type	Description	Default
credentials	SentinelHubCredentials | None	The OAuth credentials. Defaults to an empty :class:SentinelHubCredentials (env / profile path).	None
endpoint	str | None	A named endpoint alias ("cdse", "commercial") or a full base URL. Defaults to CDSE-free.	None

Examples:

• Construct against CDSE-free (no config built until configure()):

  >>> from earthlens.sentinel_hub.auth import SentinelHubAuth
  >>> auth = SentinelHubAuth()
  >>> auth.is_authenticated()
  False
  

Source code in src/earthlens/sentinel_hub/auth.py

class SentinelHubAuth(AbstractAuth[SentinelHubCredentials]):
    """Assemble and hold a Sentinel Hub `SHConfig` (non-interactive OAuth2).

    Conforms to the cross-backend :class:`earthlens.base.AbstractAuth` contract:
    :meth:`configure` builds the `SHConfig` lazily and idempotently;
    :meth:`is_authenticated` is the cheap predicate; and :meth:`config` returns
    the configured `SHConfig` (configuring on first use). The token itself is
    minted and refreshed by `sentinelhub-py` from the config on the first
    request — there is no interactive flow.

    Args:
        credentials: The OAuth credentials. Defaults to an empty
            :class:`SentinelHubCredentials` (env / profile path).
        endpoint: A named endpoint alias (`"cdse"`, `"commercial"`) or a full
            base URL. Defaults to CDSE-free.

    Examples:
        - Construct against CDSE-free (no config built until `configure()`):
            ```python
            >>> from earthlens.sentinel_hub.auth import SentinelHubAuth
            >>> auth = SentinelHubAuth()
            >>> auth.is_authenticated()
            False

            ```
    """

    def __init__(
        self,
        credentials: SentinelHubCredentials | None = None,
        endpoint: str | None = None,
    ) -> None:
        """Store the credentials + resolved endpoint; does not build the config yet.

        Args:
            credentials: OAuth credentials, or `None` for the env / profile path.
            endpoint: Endpoint alias or base URL; resolved eagerly so a bad alias
                fails at construction.
        """
        super().__init__(credentials or SentinelHubCredentials())
        self._base_url, self._token_url = resolve_endpoint(endpoint)
        self._config: Any = None

    @property
    def base_url(self) -> str:
        """The resolved Sentinel Hub base URL this auth targets."""
        return self._base_url

    def _resolve_pair(self) -> tuple[str | None, str | None]:
        """Return the `(client_id, client_secret)` to use (env → kwargs).

        Explicit credentials on the object win over the environment.

        Returns:
            The resolved id and secret (either may be `None`).
        """
        env = SentinelHubCredentials.from_env()
        client_id = self._creds.client_id or env.client_id
        secret_obj = self._creds.client_secret or env.client_secret
        secret = secret_obj.get_secret_value() if secret_obj else None
        return client_id, secret

    def configure(self) -> None:
        """Build the `SHConfig` (base/token urls + credentials); idempotent.

        Resolves credentials env → kwargs → profile. A second call after
        :meth:`is_authenticated` returns `True` is a no-op.

        Raises:
            ImportError: When the `sentinel-hub` extra is not installed.
            AuthenticationError: When no credentials are resolvable (no id/secret
                and no profile).
        """
        if self.is_authenticated():
            return
        sentinelhub = import_sentinelhub()
        client_id, client_secret = self._resolve_pair()
        if not (client_id and client_secret) and not self._creds.profile:
            raise AuthenticationError(
                "no Sentinel Hub credentials found: set SENTINELHUB_CLIENT_ID / "
                "SENTINELHUB_CLIENT_SECRET (mint an OAuth client_credentials pair "
                f"in the CDSE Dashboard at {_DASHBOARD_URL}, Grant Type "
                "'Client Credentials'), pass client_id= / client_secret=, or "
                "supply a saved SHConfig profile= name."
            )
        cfg = (
            sentinelhub.SHConfig(self._creds.profile)
            if self._creds.profile
            else sentinelhub.SHConfig()
        )
        cfg.sh_base_url = self._base_url
        cfg.sh_token_url = self._token_url
        if client_id:
            cfg.sh_client_id = client_id
        if client_secret:
            cfg.sh_client_secret = client_secret
        self._config = cfg

    def is_authenticated(self) -> bool:
        """`True` once :meth:`configure` has assembled an `SHConfig`.

        Cheap predicate — does not call the network. Returns `True` exactly when
        a config object has been built.

        Returns:
            Whether the `SHConfig` is ready to issue requests.
        """
        return self._config is not None

    def config(self) -> Any:
        """Return the assembled `SHConfig`, configuring on first use.

        Returns:
            The `sentinelhub.SHConfig` carrying the base/token urls + credentials.

        Raises:
            ImportError: When the `sentinel-hub` extra is not installed.
            AuthenticationError: When no credentials are resolvable.
        """
        self.configure()
        return self._config

base_url property

The resolved Sentinel Hub base URL this auth targets.

__init__(credentials=None, endpoint=None)

Store the credentials + resolved endpoint; does not build the config yet.

Parameters:

Name	Type	Description	Default
credentials	SentinelHubCredentials | None	OAuth credentials, or None for the env / profile path.	None
endpoint	str | None	Endpoint alias or base URL; resolved eagerly so a bad alias fails at construction.	None

Source code in src/earthlens/sentinel_hub/auth.py

def __init__(
    self,
    credentials: SentinelHubCredentials | None = None,
    endpoint: str | None = None,
) -> None:
    """Store the credentials + resolved endpoint; does not build the config yet.

    Args:
        credentials: OAuth credentials, or `None` for the env / profile path.
        endpoint: Endpoint alias or base URL; resolved eagerly so a bad alias
            fails at construction.
    """
    super().__init__(credentials or SentinelHubCredentials())
    self._base_url, self._token_url = resolve_endpoint(endpoint)
    self._config: Any = None

config()

Return the assembled SHConfig, configuring on first use.

Returns:

Type	Description
Any	The sentinelhub.SHConfig carrying the base/token urls + credentials.

Raises:

Type	Description
ImportError	When the sentinel-hub extra is not installed.
AuthenticationError	When no credentials are resolvable.

Source code in src/earthlens/sentinel_hub/auth.py

def config(self) -> Any:
    """Return the assembled `SHConfig`, configuring on first use.

    Returns:
        The `sentinelhub.SHConfig` carrying the base/token urls + credentials.

    Raises:
        ImportError: When the `sentinel-hub` extra is not installed.
        AuthenticationError: When no credentials are resolvable.
    """
    self.configure()
    return self._config

configure()

Build the SHConfig (base/token urls + credentials); idempotent.

Resolves credentials env → kwargs → profile. A second call after :meth:is_authenticated returns True is a no-op.

Raises:

Type	Description
ImportError	When the sentinel-hub extra is not installed.
AuthenticationError	When no credentials are resolvable (no id/secret and no profile).

Source code in src/earthlens/sentinel_hub/auth.py

def configure(self) -> None:
    """Build the `SHConfig` (base/token urls + credentials); idempotent.

    Resolves credentials env → kwargs → profile. A second call after
    :meth:`is_authenticated` returns `True` is a no-op.

    Raises:
        ImportError: When the `sentinel-hub` extra is not installed.
        AuthenticationError: When no credentials are resolvable (no id/secret
            and no profile).
    """
    if self.is_authenticated():
        return
    sentinelhub = import_sentinelhub()
    client_id, client_secret = self._resolve_pair()
    if not (client_id and client_secret) and not self._creds.profile:
        raise AuthenticationError(
            "no Sentinel Hub credentials found: set SENTINELHUB_CLIENT_ID / "
            "SENTINELHUB_CLIENT_SECRET (mint an OAuth client_credentials pair "
            f"in the CDSE Dashboard at {_DASHBOARD_URL}, Grant Type "
            "'Client Credentials'), pass client_id= / client_secret=, or "
            "supply a saved SHConfig profile= name."
        )
    cfg = (
        sentinelhub.SHConfig(self._creds.profile)
        if self._creds.profile
        else sentinelhub.SHConfig()
    )
    cfg.sh_base_url = self._base_url
    cfg.sh_token_url = self._token_url
    if client_id:
        cfg.sh_client_id = client_id
    if client_secret:
        cfg.sh_client_secret = client_secret
    self._config = cfg

is_authenticated()

True once :meth:configure has assembled an SHConfig.

Cheap predicate — does not call the network. Returns True exactly when a config object has been built.

Returns:

Type	Description
bool	Whether the SHConfig is ready to issue requests.

Source code in src/earthlens/sentinel_hub/auth.py

def is_authenticated(self) -> bool:
    """`True` once :meth:`configure` has assembled an `SHConfig`.

    Cheap predicate — does not call the network. Returns `True` exactly when
    a config object has been built.

    Returns:
        Whether the `SHConfig` is ready to issue requests.
    """
    return self._config is not None

SentinelHubCredentials

Bases: BaseModel

Frozen value object holding the Sentinel Hub OAuth credentials.

All fields are optional: an empty object falls back to the SENTINELHUB_CLIENT_ID / SENTINELHUB_CLIENT_SECRET environment (or the SH_* fallback), then to the named profile. Supplying client_id (+ client_secret) selects the explicit client-credentials pair (which wins over the environment).

Attributes:

Name	Type	Description
client_id	str | None	OAuth client id, or None to read SENTINELHUB_CLIENT_ID.
client_secret	SecretStr | None	OAuth client secret (kept as a SecretStr), or None to read SENTINELHUB_CLIENT_SECRET.
profile	str | None	A saved SHConfig profile name to load credentials from when no id/secret is supplied, or None for the default profile.

Examples:

• An empty object carries no id (the env / profile path):

  >>> from earthlens.sentinel_hub.auth import SentinelHubCredentials
  >>> SentinelHubCredentials().client_id is None
  True
  

• The secret is carried opaquely:

  >>> from earthlens.sentinel_hub.auth import SentinelHubCredentials
  >>> creds = SentinelHubCredentials(client_id="abc", client_secret="shh")
  >>> creds.client_secret.get_secret_value()
  'shh'
  

Source code in src/earthlens/sentinel_hub/auth.py

class SentinelHubCredentials(BaseModel):
    """Frozen value object holding the Sentinel Hub OAuth credentials.

    All fields are optional: an empty object falls back to the
    `SENTINELHUB_CLIENT_ID` / `SENTINELHUB_CLIENT_SECRET` environment (or the
    `SH_*` fallback), then to the named `profile`. Supplying `client_id`
    (+ `client_secret`) selects the explicit client-credentials pair (which wins
    over the environment).

    Attributes:
        client_id: OAuth client id, or `None` to read `SENTINELHUB_CLIENT_ID`.
        client_secret: OAuth client secret (kept as a `SecretStr`), or `None` to
            read `SENTINELHUB_CLIENT_SECRET`.
        profile: A saved `SHConfig` profile name to load credentials from when
            no id/secret is supplied, or `None` for the default profile.

    Examples:
        - An empty object carries no id (the env / profile path):
            ```python
            >>> from earthlens.sentinel_hub.auth import SentinelHubCredentials
            >>> SentinelHubCredentials().client_id is None
            True

            ```
        - The secret is carried opaquely:
            ```python
            >>> from earthlens.sentinel_hub.auth import SentinelHubCredentials
            >>> creds = SentinelHubCredentials(client_id="abc", client_secret="shh")
            >>> creds.client_secret.get_secret_value()
            'shh'

            ```
    """

    model_config = ConfigDict(frozen=True)

    client_id: str | None = None
    client_secret: SecretStr | None = None
    profile: str | None = None

    @classmethod
    def from_env(cls) -> SentinelHubCredentials:
        """Build credentials from the `SENTINELHUB_*` (or `SH_*`) environment.

        Reads the descriptive `SENTINELHUB_CLIENT_ID` / `SENTINELHUB_CLIENT_SECRET`
        / `SENTINELHUB_PROFILE` variables, falling back to the `sentinelhub-py`
        native `SH_CLIENT_ID` / `SH_CLIENT_SECRET` / `SH_PROFILE` for users coming
        from the SDK. Absent variables stay `None`.

        Returns:
            The credentials object built from the environment.
        """
        return cls(
            client_id=(
                os.environ.get("SENTINELHUB_CLIENT_ID")
                or os.environ.get("SH_CLIENT_ID")
            ),
            client_secret=(
                os.environ.get("SENTINELHUB_CLIENT_SECRET")
                or os.environ.get("SH_CLIENT_SECRET")
            ),
            profile=(
                os.environ.get("SENTINELHUB_PROFILE") or os.environ.get("SH_PROFILE")
            ),
        )

from_env() classmethod

Build credentials from the SENTINELHUB_* (or SH_*) environment.

Reads the descriptive SENTINELHUB_CLIENT_ID / SENTINELHUB_CLIENT_SECRET / SENTINELHUB_PROFILE variables, falling back to the sentinelhub-py native SH_CLIENT_ID / SH_CLIENT_SECRET / SH_PROFILE for users coming from the SDK. Absent variables stay None.

Returns:

Type	Description
SentinelHubCredentials	The credentials object built from the environment.

Source code in src/earthlens/sentinel_hub/auth.py

@classmethod
def from_env(cls) -> SentinelHubCredentials:
    """Build credentials from the `SENTINELHUB_*` (or `SH_*`) environment.

    Reads the descriptive `SENTINELHUB_CLIENT_ID` / `SENTINELHUB_CLIENT_SECRET`
    / `SENTINELHUB_PROFILE` variables, falling back to the `sentinelhub-py`
    native `SH_CLIENT_ID` / `SH_CLIENT_SECRET` / `SH_PROFILE` for users coming
    from the SDK. Absent variables stay `None`.

    Returns:
        The credentials object built from the environment.
    """
    return cls(
        client_id=(
            os.environ.get("SENTINELHUB_CLIENT_ID")
            or os.environ.get("SH_CLIENT_ID")
        ),
        client_secret=(
            os.environ.get("SENTINELHUB_CLIENT_SECRET")
            or os.environ.get("SH_CLIENT_SECRET")
        ),
        profile=(
            os.environ.get("SENTINELHUB_PROFILE") or os.environ.get("SH_PROFILE")
        ),
    )

earthlens.sentinel_hub._dispatch

Request-plane selection for the Sentinel Hub backend.

Pure, network-free helpers that decide which request plane a download uses. The backend passes the resolved render size (px/side) and whether a geometry= was supplied; these functions return the plane name (one of :data:earthlens.sentinel_hub._helpers.VALID_APIS). Kept separate from backend.py so the routing rules are unit-testable in isolation.

The auto-routing rule (G4 / G5), when api= is omitted:

• a geometry= request → "statistical" (zonal stats over the polygon);
• otherwise a raster render routed by size: ≤ SH_MAX_DIMENSION → "process"; larger → "async" (within the Async ceiling) / "batch" (above it) when an S3 batch_output is configured, else "tiling" (the local split + mosaic path, which needs no S3 bucket).

An explicit api= is honoured verbatim (only validated), so a user can force "process" on a request the auto-rule would route elsewhere — the size guard in the backend then raises if that forced plane cannot satisfy the request.

auto_select_api(max_side_px, has_geometry, has_s3=False)

Pick the plane for a request when api= was omitted.

Parameters:

Name	Type	Description	Default
max_side_px	int	The larger of the render's two pixel dimensions.	required
has_geometry	bool	Whether a geometry= (polygon / FeatureCollection) was supplied (selects the tabular Statistical plane).	required
has_s3	bool	Whether an S3 batch_output is configured (enables the S3-delivered async / batch planes; otherwise oversized rasters fall back to local tiling).	False

Returns:

Type	Description
str	The selected plane name.

Examples:

• A geometry request goes to the Statistical plane:

  >>> from earthlens.sentinel_hub._dispatch import auto_select_api
  >>> auto_select_api(512, has_geometry=True)
  'statistical'
  

• A small raster goes to Process; an oversized one without S3 to tiling:

  >>> from earthlens.sentinel_hub._dispatch import auto_select_api
  >>> auto_select_api(1024, has_geometry=False)
  'process'
  >>> auto_select_api(40000, has_geometry=False)
  'tiling'
  

• With an S3 bucket, oversized rasters go to async / batch:

  >>> from earthlens.sentinel_hub._dispatch import auto_select_api
  >>> auto_select_api(8000, has_geometry=False, has_s3=True)
  'async'
  >>> auto_select_api(40000, has_geometry=False, has_s3=True)
  'batch'
  

Source code in src/earthlens/sentinel_hub/_dispatch.py

def auto_select_api(max_side_px: int, has_geometry: bool, has_s3: bool = False) -> str:
    """Pick the plane for a request when `api=` was omitted.

    Args:
        max_side_px: The larger of the render's two pixel dimensions.
        has_geometry: Whether a `geometry=` (polygon / FeatureCollection) was
            supplied (selects the tabular Statistical plane).
        has_s3: Whether an S3 `batch_output` is configured (enables the
            S3-delivered async / batch planes; otherwise oversized rasters fall
            back to local tiling).

    Returns:
        The selected plane name.

    Examples:
        - A geometry request goes to the Statistical plane:
            ```python
            >>> from earthlens.sentinel_hub._dispatch import auto_select_api
            >>> auto_select_api(512, has_geometry=True)
            'statistical'

            ```
        - A small raster goes to Process; an oversized one without S3 to tiling:
            ```python
            >>> from earthlens.sentinel_hub._dispatch import auto_select_api
            >>> auto_select_api(1024, has_geometry=False)
            'process'
            >>> auto_select_api(40000, has_geometry=False)
            'tiling'

            ```
        - With an S3 bucket, oversized rasters go to async / batch:
            ```python
            >>> from earthlens.sentinel_hub._dispatch import auto_select_api
            >>> auto_select_api(8000, has_geometry=False, has_s3=True)
            'async'
            >>> auto_select_api(40000, has_geometry=False, has_s3=True)
            'batch'

            ```
    """
    if has_geometry:
        return "statistical"
    if max_side_px <= SH_MAX_DIMENSION:
        return "process"
    if not has_s3:
        return "tiling"
    if max_side_px <= ASYNC_MAX_DIMENSION:
        return "async"
    return "batch"

resolve_api(api, max_side_px, has_geometry, has_s3=False)

Validate an explicit api=, or auto-select one by size / geometry / S3.

Parameters:

Name	Type	Description	Default
api	str | None	The requested plane, or None for auto-selection.	required
max_side_px	int	The larger render dimension in pixels.	required
has_geometry	bool	Whether a geometry= was supplied.	required
has_s3	bool	Whether an S3 batch_output is configured.	False

Returns:

Type	Description
str	The resolved plane name.

Raises:

Type	Description
ValueError	When api is a non-None value outside :data:VALID_APIS.

Source code in src/earthlens/sentinel_hub/_dispatch.py

def resolve_api(
    api: str | None, max_side_px: int, has_geometry: bool, has_s3: bool = False
) -> str:
    """Validate an explicit `api=`, or auto-select one by size / geometry / S3.

    Args:
        api: The requested plane, or `None` for auto-selection.
        max_side_px: The larger render dimension in pixels.
        has_geometry: Whether a `geometry=` was supplied.
        has_s3: Whether an S3 `batch_output` is configured.

    Returns:
        The resolved plane name.

    Raises:
        ValueError: When `api` is a non-`None` value outside :data:`VALID_APIS`.
    """
    validate_api(api)
    if api is not None:
        return api
    return auto_select_api(max_side_px, has_geometry, has_s3)

validate_api(api)

Validate an explicit api= value, if any.

Parameters:

Name	Type	Description	Default
api	str | None	The requested plane, or None for auto-selection.	required

Raises:

Type	Description
ValueError	When api is a non-None value outside :data:VALID_APIS.

Source code in src/earthlens/sentinel_hub/_dispatch.py

def validate_api(api: str | None) -> None:
    """Validate an explicit `api=` value, if any.

    Args:
        api: The requested plane, or `None` for auto-selection.

    Raises:
        ValueError: When `api` is a non-`None` value outside :data:`VALID_APIS`.
    """
    if api is not None and api not in VALID_APIS:
        raise ValueError(
            f"unknown api={api!r}: choose one of {list(VALID_APIS)} or omit it "
            "for size-based auto-selection."
        )

earthlens.sentinel_hub._helpers

Stateless helpers for the Sentinel Hub backend.

Hosts the small, pure functions and constant tables the backend, auth, and dispatch modules share: the request-plane vocabulary (VALID_APIS), the endpoint registry (CDSE-free vs commercial), the Process/Async pixel ceilings (SH_MAX_DIMENSION / ASYNC_MAX_DIMENSION), the lazy sentinelhub SDK importer (which turns a missing [sentinel-hub] extra into a friendly ImportError), the CDSE data-collection binding (cdse_collection), and the pandas-frequency → Statistical aggregation_interval (ISO-8601) mapping used by aggregate=.

None of these touch the network, so they are trivially unit testable and safe to import without the sentinelhub SDK installed (the SDK is imported only inside import_sentinelhub / cdse_collection, never at module load).

cdse_collection(name, base_url)

Bind a DataCollection name to the CDSE deployment (or return it as-is).

The stock sentinelhub.DataCollection enum members point at the commercial deployment, so on CDSE each must be rebound to the CDSE service URL via DataCollection.<NAME>.define_from(..., service_url=base_url) (verified, A1). On a commercial base_url the stock member is returned unchanged.

Parameters:

Name	Type	Description	Default
name	str	A DataCollection member name (e.g. "SENTINEL2_L2A").	required
base_url	str	The configured sh_base_url (CDSE vs commercial).	required

Returns:

Type	Description
Any	The (possibly rebound) DataCollection to pass to input_data.

Raises:

Type	Description
ImportError	When the sentinel-hub extra is not installed.
ValueError	When name is not a known DataCollection member.

Source code in src/earthlens/sentinel_hub/_helpers.py

def cdse_collection(name: str, base_url: str) -> Any:
    """Bind a `DataCollection` name to the CDSE deployment (or return it as-is).

    The stock `sentinelhub.DataCollection` enum members point at the commercial
    deployment, so on CDSE each must be rebound to the CDSE service URL via
    `DataCollection.<NAME>.define_from(..., service_url=base_url)` (verified, A1).
    On a commercial `base_url` the stock member is returned unchanged.

    Args:
        name: A `DataCollection` member name (e.g. `"SENTINEL2_L2A"`).
        base_url: The configured `sh_base_url` (CDSE vs commercial).

    Returns:
        The (possibly rebound) `DataCollection` to pass to `input_data`.

    Raises:
        ImportError: When the `sentinel-hub` extra is not installed.
        ValueError: When `name` is not a known `DataCollection` member.
    """
    sentinelhub = import_sentinelhub()
    try:
        base = getattr(sentinelhub.DataCollection, name)
    except AttributeError as exc:
        raise ValueError(
            f"{name!r} is not a known sentinelhub DataCollection."
        ) from exc
    if "dataspace" in base_url:
        return base.define_from(f"{name.lower()}_cdse", service_url=base_url)
    return base

import_sentinelhub()

Import and return the sentinelhub SDK, or raise a friendly ImportError.

The SDK is an optional dependency (pip install earthlens[sentinel-hub]); this helper centralises the lazy import so every call site surfaces the same actionable message when the extra is missing.

Returns:

Type	Description
Any	The imported sentinelhub module.

Raises:

Type	Description
ImportError	When the sentinel-hub extra is not installed.

Source code in src/earthlens/sentinel_hub/_helpers.py

def import_sentinelhub() -> Any:
    """Import and return the `sentinelhub` SDK, or raise a friendly `ImportError`.

    The SDK is an optional dependency (`pip install earthlens[sentinel-hub]`);
    this helper centralises the lazy import so every call site surfaces the same
    actionable message when the extra is missing.

    Returns:
        The imported `sentinelhub` module.

    Raises:
        ImportError: When the `sentinel-hub` extra is not installed.
    """
    try:
        import sentinelhub
    except ImportError as exc:
        raise ImportError(
            "the Sentinel Hub backend requires the 'sentinelhub' client. Install "
            "it with `pip install earthlens[sentinel-hub]`."
        ) from exc
    return sentinelhub

interval_for(freq)

Map a pandas frequency alias to a Statistical aggregation_interval.

Strips any leading repeat count ("1MS" → "MS") before the lookup, except for the count-prefixed day windows ("7D", "10D") handled directly.

Parameters:

Name	Type	Description	Default
freq	str	A pandas offset alias, optionally count-prefixed ("1D", "7D", "1MS", "YS", …).	required

Returns:

Type	Description
str	The ISO-8601 duration string the Statistical API expects ("P1D",
str	"P7D", "P1M", "P1Y", …).

Raises:

Type	Description
NotImplementedError	If freq has no ISO-8601 interval equivalent.

Examples:

• Month-start maps to a one-month interval:

  >>> from earthlens.sentinel_hub._helpers import interval_for
  >>> interval_for("1MS")
  'P1M'
  

• A daily window maps to P1D:

  >>> from earthlens.sentinel_hub._helpers import interval_for
  >>> interval_for("D")
  'P1D'
  

Source code in src/earthlens/sentinel_hub/_helpers.py

def interval_for(freq: str) -> str:
    """Map a pandas frequency alias to a Statistical `aggregation_interval`.

    Strips any leading repeat count (`"1MS"` → `"MS"`) before the lookup, except
    for the count-prefixed day windows (`"7D"`, `"10D"`) handled directly.

    Args:
        freq: A pandas offset alias, optionally count-prefixed (`"1D"`, `"7D"`,
            `"1MS"`, `"YS"`, …).

    Returns:
        The ISO-8601 duration string the Statistical API expects (`"P1D"`,
        `"P7D"`, `"P1M"`, `"P1Y"`, …).

    Raises:
        NotImplementedError: If `freq` has no ISO-8601 interval equivalent.

    Examples:
        - Month-start maps to a one-month interval:
            ```python
            >>> from earthlens.sentinel_hub._helpers import interval_for
            >>> interval_for("1MS")
            'P1M'

            ```
        - A daily window maps to `P1D`:
            ```python
            >>> from earthlens.sentinel_hub._helpers import interval_for
            >>> interval_for("D")
            'P1D'

            ```
    """
    if freq in _FREQ_TO_INTERVAL:
        return _FREQ_TO_INTERVAL[freq]
    key = freq.lstrip("0123456789")
    if key in _FREQ_TO_INTERVAL:
        return _FREQ_TO_INTERVAL[key]
    raise NotImplementedError(
        f"the Statistical API has no aggregation_interval for freq={freq!r}; "
        f"supported pandas aliases map to {sorted(set(_FREQ_TO_INTERVAL.values()))}."
    )

resolve_endpoint(endpoint)

Resolve an endpoint= value to a (base_url, token_url) pair.

Accepts a named alias ("cdse", "commercial"), a full http(s):// base URL (paired with the matching token URL by host), or None (the CDSE-free default).

Parameters:

Name	Type	Description	Default
endpoint	str | None	A named alias, a full base URL, or None.	required

Returns:

Type	Description
tuple[str, str]	The (sh_base_url, sh_token_url) pair to write onto an SHConfig.

Raises:

Type	Description
ValueError	When endpoint is an unknown non-URL string.

Examples:

• The default alias resolves to CDSE-free:

  >>> from earthlens.sentinel_hub._helpers import resolve_endpoint
  >>> base, _ = resolve_endpoint(None)
  >>> base
  'https://sh.dataspace.copernicus.eu'
  

• A full URL on a CDSE host keeps the CDSE Keycloak token URL:

  >>> from earthlens.sentinel_hub._helpers import resolve_endpoint
  >>> _, token = resolve_endpoint("https://sh.dataspace.copernicus.eu")
  >>> "dataspace" in token
  True
  

Source code in src/earthlens/sentinel_hub/_helpers.py

def resolve_endpoint(endpoint: str | None) -> tuple[str, str]:
    """Resolve an `endpoint=` value to a `(base_url, token_url)` pair.

    Accepts a named alias (`"cdse"`, `"commercial"`), a full `http(s)://` base
    URL (paired with the matching token URL by host), or `None` (the CDSE-free
    default).

    Args:
        endpoint: A named alias, a full base URL, or `None`.

    Returns:
        The `(sh_base_url, sh_token_url)` pair to write onto an `SHConfig`.

    Raises:
        ValueError: When `endpoint` is an unknown non-URL string.

    Examples:
        - The default alias resolves to CDSE-free:
            ```python
            >>> from earthlens.sentinel_hub._helpers import resolve_endpoint
            >>> base, _ = resolve_endpoint(None)
            >>> base
            'https://sh.dataspace.copernicus.eu'

            ```
        - A full URL on a CDSE host keeps the CDSE Keycloak token URL:
            ```python
            >>> from earthlens.sentinel_hub._helpers import resolve_endpoint
            >>> _, token = resolve_endpoint("https://sh.dataspace.copernicus.eu")
            >>> "dataspace" in token
            True

            ```
    """
    if endpoint is None:
        return SH_ENDPOINTS[DEFAULT_ENDPOINT]
    if endpoint in SH_ENDPOINTS:
        return SH_ENDPOINTS[endpoint]
    if endpoint.startswith("http://") or endpoint.startswith("https://"):
        token = (
            SH_ENDPOINTS["cdse"][1]
            if "dataspace" in endpoint
            else SH_ENDPOINTS["commercial"][1]
        )
        return endpoint, token
    raise ValueError(
        f"unknown Sentinel Hub endpoint {endpoint!r}: pass one of "
        f"{sorted(SH_ENDPOINTS)} or a full http(s):// base URL."
    )

tile_bbox(bbox, width_px, height_px, max_dim=SH_MAX_DIMENSION)

Split a bbox into a grid of sub-bboxes each ≤ max_dim px per side.

Request-shaping arithmetic for the Process-API 2500 px cap: the bbox is divided into ceil(width_px / max_dim) × ceil(height_px / max_dim) equal tiles in degree space (row-major, south-to-north then west-to-east). The mosaic of the rendered tiles is the responsibility of pyramids — this only decides the windows.

Parameters:

Name	Type	Description	Default
bbox	tuple[float, float, float, float]	The (west, south, east, north) envelope in degrees.	required
width_px	int	The full render width in pixels (at the target resolution).	required
height_px	int	The full render height in pixels.	required
max_dim	int	The per-tile pixel ceiling (defaults to :data:SH_MAX_DIMENSION).	SH_MAX_DIMENSION

Returns:

Type	Description
list[tuple[float, float, float, float]]	The list of (west, south, east, north) sub-bboxes (one per tile).

Examples:

• A render within the cap is a single tile (the bbox itself):

  >>> from earthlens.sentinel_hub._helpers import tile_bbox
  >>> tile_bbox((0.0, 0.0, 1.0, 1.0), 1000, 1000)
  [(0.0, 0.0, 1.0, 1.0)]
  

• A 2× oversized render splits into a 2×2 grid:

  >>> from earthlens.sentinel_hub._helpers import tile_bbox
  >>> len(tile_bbox((0.0, 0.0, 1.0, 1.0), 5000, 5000))
  4
  

Source code in src/earthlens/sentinel_hub/_helpers.py

def tile_bbox(
    bbox: tuple[float, float, float, float],
    width_px: int,
    height_px: int,
    max_dim: int = SH_MAX_DIMENSION,
) -> list[tuple[float, float, float, float]]:
    """Split a bbox into a grid of sub-bboxes each ≤ `max_dim` px per side.

    Request-shaping arithmetic for the Process-API 2500 px cap: the bbox is
    divided into `ceil(width_px / max_dim)` × `ceil(height_px / max_dim)` equal
    tiles in degree space (row-major, south-to-north then west-to-east). The
    mosaic of the rendered tiles is the responsibility of `pyramids` — this only
    decides the windows.

    Args:
        bbox: The `(west, south, east, north)` envelope in degrees.
        width_px: The full render width in pixels (at the target resolution).
        height_px: The full render height in pixels.
        max_dim: The per-tile pixel ceiling (defaults to :data:`SH_MAX_DIMENSION`).

    Returns:
        The list of `(west, south, east, north)` sub-bboxes (one per tile).

    Examples:
        - A render within the cap is a single tile (the bbox itself):
            ```python
            >>> from earthlens.sentinel_hub._helpers import tile_bbox
            >>> tile_bbox((0.0, 0.0, 1.0, 1.0), 1000, 1000)
            [(0.0, 0.0, 1.0, 1.0)]

            ```
        - A 2× oversized render splits into a 2×2 grid:
            ```python
            >>> from earthlens.sentinel_hub._helpers import tile_bbox
            >>> len(tile_bbox((0.0, 0.0, 1.0, 1.0), 5000, 5000))
            4

            ```
    """
    import math

    west, south, east, north = bbox
    n_x = max(1, math.ceil(width_px / max_dim))
    n_y = max(1, math.ceil(height_px / max_dim))
    step_x = (east - west) / n_x
    step_y = (north - south) / n_y
    tiles: list[tuple[float, float, float, float]] = []
    for row in range(n_y):
        for col in range(n_x):
            tiles.append(
                (
                    west + col * step_x,
                    south + row * step_y,
                    west + (col + 1) * step_x,
                    south + (row + 1) * step_y,
                )
            )
    return tiles