Copernicus DEM — API reference #

Anonymous Copernicus DEM backend subpackage — earthlens.dem. Background and usage are covered under the other pages in this section (Introduction, Usage, Available datasets); this page is the rendered API. The buckets are public, so there is no auth module.

`earthlens.dem` #

Anonymous Copernicus DEM backend.

Fetches raw Copernicus DEM GLO-30 / GLO-90 COG tiles from the public AWS Open Data buckets — no account, no SDK login, no key. Cropping, mosaicking, and reprojection are pyramids' job; this package returns whole 1° tiles as they came from the bucket. See docs/reference/dem/ for the usage guide and the shipped example notebook for a mosaic-with-pyramids walkthrough.

`Catalog` #

Bases: AbstractCatalog

Dataset catalog for the DEM backend.

Reads the bundled dem_data_catalog.yaml (shipped as package data) and exposes its datasets: block as a map of :class:DEMDataset rows keyed by dataset key under the inherited :attr:datasets field, which supplies the cat["cop-dem-glo-30"] / "cop-dem-glo-30" in cat / len(cat) dict-like surface and the did-you-mean error for free. Instantiate with no arguments (Catalog()); :func:model_post_init loads and validates the YAML in one pass and caches it by (path, mtime).

Attributes:

Name	Type	Description
`datasets`	`dict[str, DEMDataset]`	Map from dataset key to its :class:`DEMDataset` row.
`available_datasets`	`list[str]`	Sorted dataset keys — the curated datasets are the whole dataset universe for this backend.

Examples:

List datasets and resolve one:

>>> from earthlens.dem import Catalog
>>> cat = Catalog()
>>> sorted(cat.datasets)
['cop-dem-glo-30', 'cop-dem-glo-90']
>>> cat.get_dataset("cop-dem-glo-90").native_resolution_m
90

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

>>> from earthlens.dem import Catalog
>>> Catalog().get_dataset("cop-dem-glo-3")  # doctest: +ELLIPSIS
Traceback (most recent call last):
    ...
ValueError: 'cop-dem-glo-3' is not in the DEM catalog. ... Did you mean 'cop-dem-glo-30'?

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

class Catalog(AbstractCatalog):
    """Dataset catalog for the DEM backend.

    Reads the bundled `dem_data_catalog.yaml` (shipped as package data)
    and exposes its `datasets:` block as a map of :class:`DEMDataset`
    rows keyed by dataset key under the inherited :attr:`datasets`
    field, which supplies the `cat["cop-dem-glo-30"]` /
    `"cop-dem-glo-30" in cat` / `len(cat)` dict-like surface and the
    did-you-mean error for free. Instantiate with no arguments
    (`Catalog()`); :func:`model_post_init` loads and validates the YAML
    in one pass and caches it by `(path, mtime)`.

    Attributes:
        datasets: Map from dataset key to its :class:`DEMDataset` row.
        available_datasets: Sorted dataset keys — the curated datasets
            are the whole dataset universe for this backend.

    Examples:
        - List datasets and resolve one:
            ```python
            >>> from earthlens.dem import Catalog
            >>> cat = Catalog()
            >>> sorted(cat.datasets)
            ['cop-dem-glo-30', 'cop-dem-glo-90']
            >>> cat.get_dataset("cop-dem-glo-90").native_resolution_m
            90

            ```
        - An unknown key raises with a did-you-mean hint:
            ```python
            >>> from earthlens.dem import Catalog
            >>> Catalog().get_dataset("cop-dem-glo-3")  # doctest: +ELLIPSIS
            Traceback (most recent call last):
                ...
            ValueError: 'cop-dem-glo-3' is not in the DEM catalog. ... Did you mean 'cop-dem-glo-30'?

            ```
    """

    _catalog_kind: str = "DEM catalog"

    datasets: dict[str, DEMDataset] = Field(default_factory=dict)

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

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

        Raises:
            ValueError: Propagated from :meth:`load` when the YAML is
                missing, empty, or has a malformed row.
        """
        if not self.datasets:
            loaded = Catalog.load()
            self.datasets = loaded.datasets
        self.available_datasets = sorted(self.datasets)

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

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

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

        Raises:
            ValueError: If the file has no `datasets:` block, or any
                row fails validation.
        """
        path = catalog_path if catalog_path is not None else CATALOG_PATH
        try:
            mtime = path.stat().st_mtime_ns
        except FileNotFoundError:
            mtime = 0
        cache_key = (str(path.resolve()), mtime)
        cached = _CATALOG_CACHE.get(cache_key)
        if cached is not None:
            return cached

        data = load_yaml_strict(path) or {}
        datasets_yaml = data.get("datasets") or {}
        if not datasets_yaml:
            raise ValueError(
                f"{path} is missing or has an empty 'datasets:' block. "
                "The DEM catalog must list at least one dataset."
            )
        datasets: dict[str, DEMDataset] = {}
        for key, body in datasets_yaml.items():
            try:
                datasets[key] = DEMDataset(key=key, **dict(body or {}))
            except ValidationError as exc:
                raise ValueError(
                    f"{path} dataset {key!r} failed validation:\n{exc}"
                ) from exc
        catalog = cls(datasets=datasets)
        _CATALOG_CACHE[cache_key] = catalog
        return catalog

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

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

`get_catalog()` #

Return the dataset map (satisfies the abstract contract).

Returns:

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

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

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

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

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

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

Parameters:

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

Returns:

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

Raises:

Type	Description
`ValueError`	If the file has no `datasets:` block, or any row fails validation.

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

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

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

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

    Raises:
        ValueError: If the file has no `datasets:` block, or any
            row fails validation.
    """
    path = catalog_path if catalog_path is not None else CATALOG_PATH
    try:
        mtime = path.stat().st_mtime_ns
    except FileNotFoundError:
        mtime = 0
    cache_key = (str(path.resolve()), mtime)
    cached = _CATALOG_CACHE.get(cache_key)
    if cached is not None:
        return cached

    data = load_yaml_strict(path) or {}
    datasets_yaml = data.get("datasets") or {}
    if not datasets_yaml:
        raise ValueError(
            f"{path} is missing or has an empty 'datasets:' block. "
            "The DEM catalog must list at least one dataset."
        )
    datasets: dict[str, DEMDataset] = {}
    for key, body in datasets_yaml.items():
        try:
            datasets[key] = DEMDataset(key=key, **dict(body or {}))
        except ValidationError as exc:
            raise ValueError(
                f"{path} dataset {key!r} failed validation:\n{exc}"
            ) from exc
    catalog = cls(datasets=datasets)
    _CATALOG_CACHE[cache_key] = catalog
    return catalog

`model_post_init(__context)` #

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

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

Raises:

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

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

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

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

    Raises:
        ValueError: Propagated from :meth:`load` when the YAML is
            missing, empty, or has a malformed row.
    """
    if not self.datasets:
        loaded = Catalog.load()
        self.datasets = loaded.datasets
    self.available_datasets = sorted(self.datasets)

`DEM` #

Bases: AbstractDataSource

Anonymous Copernicus DEM backend (raw COG tiles).

Parameters:

Name	Type	Description	Default
`start`	`str \| None`	Advisory start date (parsed with `fmt`); DEM is time-invariant, so an omitted `start` defaults to today.	`None`
`end`	`str \| None`	Advisory end date; defaults to `start`.	`None`
`variables`	`list[str] \| dict[str, list[str]] \| None`	Ignored — a Copernicus DEM tile carries one elevation band. Kept in the signature for `AbstractDataSource` compatibility.	`None`
`lat_lim`	`list[float] \| None`	`[lat_min, lat_max]` in degrees. Selects which 1° tiles to fetch.	`None`
`lon_lim`	`list[float] \| None`	`[lon_min, lon_max]` in degrees.	`None`
`temporal_resolution`	`str`	Advisory label; DEM has no time cadence.	`'static'`
`path`	`Path \| str`	Output directory for the downloaded tiles.	`''`
`fmt`	`str`	`strptime` format for `start` / `end`.	`'%Y-%m-%d'`
`dataset`	`str`	Catalog key — `"cop-dem-glo-30"` (default) or `"cop-dem-glo-90"`.	`'cop-dem-glo-30'`
`catalog`	`Catalog \| None`	Optional pre-built :class:`Catalog` (tests inject one); defaults to the bundled catalog.	`None`

Examples:

Plan the tiles a bbox needs without touching the network:

>>> from earthlens.dem import DEM
>>> src = DEM(
...     start="2026-01-01", end="2026-01-01",
...     variables=[],
...     lat_lim=[30.2, 30.8], lon_lim=[31.2, 31.8],
... )
>>> [(t.lat, t.lon) for t in src.tiles()]
[(30, 31)]

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

class DEM(AbstractDataSource):
    """Anonymous Copernicus DEM backend (raw COG tiles).

    Args:
        start: Advisory start date (parsed with `fmt`); DEM is
            time-invariant, so an omitted `start` defaults to today.
        end: Advisory end date; defaults to `start`.
        variables: Ignored — a Copernicus DEM tile carries one
            elevation band. Kept in the signature for `AbstractDataSource`
            compatibility.
        lat_lim: `[lat_min, lat_max]` in degrees. Selects which 1° tiles
            to fetch.
        lon_lim: `[lon_min, lon_max]` in degrees.
        temporal_resolution: Advisory label; DEM has no time cadence.
        path: Output directory for the downloaded tiles.
        fmt: `strptime` format for `start` / `end`.
        dataset: Catalog key — `"cop-dem-glo-30"` (default) or
            `"cop-dem-glo-90"`.
        catalog: Optional pre-built :class:`Catalog` (tests inject
            one); defaults to the bundled catalog.

    Examples:
        - Plan the tiles a bbox needs without touching the network:
            ```python
            >>> from earthlens.dem import DEM
            >>> src = DEM(
            ...     start="2026-01-01", end="2026-01-01",
            ...     variables=[],
            ...     lat_lim=[30.2, 30.8], lon_lim=[31.2, 31.8],
            ... )
            >>> [(t.lat, t.lon) for t in src.tiles()]
            [(30, 31)]

            ```
    """

    OUTPUT_KIND = "raster"

    def __init__(
        self,
        start: str | None = None,
        end: str | None = None,
        variables: list[str] | dict[str, list[str]] | None = None,
        lat_lim: list[float] | None = None,
        lon_lim: list[float] | None = None,
        temporal_resolution: str = "static",
        path: Path | str = "",
        fmt: str = "%Y-%m-%d",
        *,
        dataset: str = "cop-dem-glo-30",
        catalog: Catalog | None = None,
    ):
        """Initialise a Copernicus DEM backend instance.

        Raises:
            ValueError: If `dataset` is not a curated DEM catalog key.
        """
        self._catalog = catalog if catalog is not None else Catalog()
        self._dataset_key = dataset
        self._dataset: DEMDataset = self._catalog.get_dataset(dataset)
        self._show_progress = True
        # `_initialize` (called by `super().__init__` below) builds `self._auth`.

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

    # -- abstract hooks -------------------------------------------------

    def _initialize(self) -> None:
        """Build the unsigned `boto3` client via the shared `S3Auth`.

        The client is lazy: `S3Auth.client()` builds it on first access
        and caches it. Returning `None` here avoids overwriting
        `self.client` — the auth object stays the source of truth.

        Returns:
            None: The auth object owns the client (accessed via
                :meth:`_client`).
        """
        self._auth = S3Auth(
            S3Credentials(region=self._dataset.region),
        )
        return None

    def _create_grid(
        self, lat_lim: list[float], lon_lim: list[float]
    ) -> SpatialExtent:
        """Wrap the user bbox into a :class:`SpatialExtent` (no snapping)."""
        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:
        """Return a single-instant :class:`TemporalExtent` (DEM is static).

        DEM has no time axis; the returned extent carries a single-date
        `DatetimeIndex` derived from `start` so the shared
        `TemporalExtent` validator passes.
        """
        start_date = pd.to_datetime(start, format=fmt)
        end_date = pd.to_datetime(end, format=fmt)
        return TemporalExtent(
            start_date=start_date,
            end_date=end_date,
            resolution="static",
            dates=pd.DatetimeIndex([start_date]),
        )

    # -- planning + fetch -----------------------------------------------

    def tiles(self) -> list[Tile]:
        """Return the 1° tiles the request bbox intersects.

        Pure arithmetic on the integer-degree grid — no network access.
        Antimeridian-straddling bboxes (`lon_min > lon_max`) are
        rejected at construction; pass the two halves in separate calls
        if you need to span the 180th meridian.

        Returns:
            list[Tile]: One tile per SW-corner integer degree pair in
                the bbox, in row-major (south → north, west → east)
                order.
        """
        return bbox_to_tiles(
            lat_min=self.space.latitude_min,
            lat_max=self.space.latitude_max,
            lon_min=self.space.longitude_min,
            lon_max=self.space.longitude_max,
        )

    def _search(self) -> list[RemoteProduct]:
        """Enumerate one candidate product per 1° tile in the bbox.

        The key is deterministic (no `list_objects_v2` call), so a
        request with a wide bbox does not walk the bucket. Ocean /
        outside-coverage tiles are only discovered as missing when
        `_fetch` calls `head_object` on each candidate.

        Returns:
            list[RemoteProduct]: One :class:`RemoteProduct` per tile;
                `href` carries the S3 key and `metadata` carries the
                bucket, the tile identifier, and the SW corner.
        """
        token = self._dataset.resolution_token
        bucket = self._dataset.bucket
        products: list[RemoteProduct] = []
        for tile in self.tiles():
            key = tile_key(tile, token)
            products.append(
                RemoteProduct(
                    id=key.split("/", 1)[0],
                    href=key,
                    metadata={
                        "bucket": bucket,
                        "dataset": self._dataset_key,
                        "tile_lat": tile.lat,
                        "tile_lon": tile.lon,
                    },
                )
            )
        return products

    def _fetch(self, products: list[RemoteProduct]) -> list[Path]:
        """Download every present tile; log and skip ocean / missing ones.

        The bucket carries one COG per land tile; a candidate key for an
        ocean tile (or outside coverage) simply returns a 404 on
        `head_object`. Those are warned about and skipped rather than
        propagated, so a coastal bbox returns a ragged but non-empty
        result — never a crash.

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

        Returns:
            list[Path]: One local path per tile that existed and was
                downloaded, in `products` order.
        """
        client = self._client()
        written: list[Path] = []
        missing = 0
        for product in tqdm(
            products,
            disable=not (self._show_progress and products),
            desc=f"dem/{self._dataset_key}",
            unit="tile",
        ):
            fetched = self._fetch_one(client, product)
            if fetched is None:
                missing += 1
                continue
            written.append(fetched)
        if missing:
            logger.info(
                f"dem: {len(written)}/{len(products)} tile(s) downloaded; "
                f"{missing} absent (ocean / outside coverage)."
            )
        return written

    def _fetch_one(self, client: Any, product: RemoteProduct) -> Path | None:
        """Head + download one candidate tile; return `None` if absent.

        Args:
            client: The unsigned `boto3` S3 client.
            product: The candidate :class:`RemoteProduct` to fetch.

        Returns:
            Path | None: The written local path, or `None` when the key
                is absent (ocean tile, outside coverage).
        """
        bucket: str = product.metadata["bucket"]
        key: str = product.href or ""
        target = self.root_dir / Path(key).name
        if target.exists():
            return target
        try:
            client.head_object(Bucket=bucket, Key=key)
        except Exception as exc:  # noqa: BLE001 - classify below
            if _is_missing_object(exc):
                logger.warning(
                    f"dem: tile absent, skipping: s3://{bucket}/{key}"
                )
                return None
            raise
        tmp = target.with_name(target.name + ".part")
        try:
            client.download_file(bucket, key, str(tmp))
            tmp.replace(target)
        except BaseException:
            tmp.unlink(missing_ok=True)
            raise
        return target

    def _client(self) -> Any:
        """Return the unsigned `boto3` client (built lazily by `S3Auth`)."""
        return self._auth.client()

    def _api(self, *args: Any, **kwargs: Any) -> list[Path]:
        """Compose `_search` + `_fetch` (the canonical post-C3 body)."""
        return self._api_via_search_fetch()

    def download(
        self,
        progress_bar: bool = True,
        aggregate: Any | None = None,
    ) -> list[Path]:
        """Fetch every 1° Copernicus DEM tile the bbox intersects.

        Args:
            progress_bar: Show a per-tile progress bar. Defaults to
                `True`.
            aggregate: Must be `None`. DEM is a raw-COG whole-tile
                backend (`G4` — no server-side subset, `G5` — no
                decode), so a time-window reduce has nothing to reduce.
                Cropping / mosaicking the tiles is pyramids' job, not
                the backend's.

        Returns:
            list[Path]: The local paths of the downloaded COG tiles, in
                bbox row-major order. Empty when every candidate tile is
                absent from the bucket.

        Raises:
            NotImplementedError: If `aggregate` is not `None`.
        """
        if aggregate is not None:
            raise NotImplementedError(
                "DEM.download(aggregate=...) is not supported — a DEM tile is "
                "time-invariant. Mosaic / crop the downloaded tiles with "
                "pyramids downstream (`pyramids.Dataset.read_file` + "
                "`.crop`, `pyramids.dataset.merge.merge_rasters` for a "
                "multi-tile mosaic)."
            )
        self._show_progress = progress_bar
        return self._api_via_search_fetch()

`init(start=None, end=None, variables=None, lat_lim=None, lon_lim=None, temporal_resolution='static', path='', fmt='%Y-%m-%d', *, dataset='cop-dem-glo-30', catalog=None)` #

Initialise a Copernicus DEM backend instance.

Raises:

Type	Description
`ValueError`	If `dataset` is not a curated DEM catalog key.

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

def __init__(
    self,
    start: str | None = None,
    end: str | None = None,
    variables: list[str] | dict[str, list[str]] | None = None,
    lat_lim: list[float] | None = None,
    lon_lim: list[float] | None = None,
    temporal_resolution: str = "static",
    path: Path | str = "",
    fmt: str = "%Y-%m-%d",
    *,
    dataset: str = "cop-dem-glo-30",
    catalog: Catalog | None = None,
):
    """Initialise a Copernicus DEM backend instance.

    Raises:
        ValueError: If `dataset` is not a curated DEM catalog key.
    """
    self._catalog = catalog if catalog is not None else Catalog()
    self._dataset_key = dataset
    self._dataset: DEMDataset = self._catalog.get_dataset(dataset)
    self._show_progress = True
    # `_initialize` (called by `super().__init__` below) builds `self._auth`.

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

`download(progress_bar=True, aggregate=None)` #

Fetch every 1° Copernicus DEM tile the bbox intersects.

Parameters:

Name	Type	Description	Default
`progress_bar`	`bool`	Show a per-tile progress bar. Defaults to `True`.	`True`
`aggregate`	`Any \| None`	Must be `None`. DEM is a raw-COG whole-tile backend (`G4` — no server-side subset, `G5` — no decode), so a time-window reduce has nothing to reduce. Cropping / mosaicking the tiles is pyramids' job, not the backend's.	`None`

Returns:

Type	Description
`list[Path]`	list[Path]: The local paths of the downloaded COG tiles, in bbox row-major order. Empty when every candidate tile is absent from the bucket.

Raises:

Type	Description
`NotImplementedError`	If `aggregate` is not `None`.

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

def download(
    self,
    progress_bar: bool = True,
    aggregate: Any | None = None,
) -> list[Path]:
    """Fetch every 1° Copernicus DEM tile the bbox intersects.

    Args:
        progress_bar: Show a per-tile progress bar. Defaults to
            `True`.
        aggregate: Must be `None`. DEM is a raw-COG whole-tile
            backend (`G4` — no server-side subset, `G5` — no
            decode), so a time-window reduce has nothing to reduce.
            Cropping / mosaicking the tiles is pyramids' job, not
            the backend's.

    Returns:
        list[Path]: The local paths of the downloaded COG tiles, in
            bbox row-major order. Empty when every candidate tile is
            absent from the bucket.

    Raises:
        NotImplementedError: If `aggregate` is not `None`.
    """
    if aggregate is not None:
        raise NotImplementedError(
            "DEM.download(aggregate=...) is not supported — a DEM tile is "
            "time-invariant. Mosaic / crop the downloaded tiles with "
            "pyramids downstream (`pyramids.Dataset.read_file` + "
            "`.crop`, `pyramids.dataset.merge.merge_rasters` for a "
            "multi-tile mosaic)."
        )
    self._show_progress = progress_bar
    return self._api_via_search_fetch()

`tiles()` #

Return the 1° tiles the request bbox intersects.

Pure arithmetic on the integer-degree grid — no network access. Antimeridian-straddling bboxes (lon_min > lon_max) are rejected at construction; pass the two halves in separate calls if you need to span the 180th meridian.

Returns:

Type	Description
`list[Tile]`	list[Tile]: One tile per SW-corner integer degree pair in the bbox, in row-major (south → north, west → east) order.

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

def tiles(self) -> list[Tile]:
    """Return the 1° tiles the request bbox intersects.

    Pure arithmetic on the integer-degree grid — no network access.
    Antimeridian-straddling bboxes (`lon_min > lon_max`) are
    rejected at construction; pass the two halves in separate calls
    if you need to span the 180th meridian.

    Returns:
        list[Tile]: One tile per SW-corner integer degree pair in
            the bbox, in row-major (south → north, west → east)
            order.
    """
    return bbox_to_tiles(
        lat_min=self.space.latitude_min,
        lat_max=self.space.latitude_max,
        lon_min=self.space.longitude_min,
        lon_max=self.space.longitude_max,
    )

`DEMDataset` #

Bases: BaseModel

One Copernicus DEM dataset row.

A frozen value object that pins the exact S3 bucket, region, and the resolution token embedded in every tile key (_10_ for GLO-30 or _30_ for GLO-90 — the tokens Copernicus uses in the object names, not the pixel size).

Attributes:

Name	Type	Description
`key`	`str`	Dataset key (`"cop-dem-glo-30"`, `"cop-dem-glo-90"`) — the value passed to `EarthLens(..., dataset=...)`.
`long_name`	`str`	Human-readable label used in docs and logs.
`bucket`	`str`	Anonymous S3 bucket that holds the tiles.
`region`	`str`	AWS region of the bucket (`"eu-central-1"`).
`resolution_token`	`str`	The `{token}` fragment in the tile file name (`"10"` for GLO-30, `"30"` for GLO-90).
`tile_degrees`	`int`	Edge length of one tile in degrees (`1`).
`native_resolution_m`	`int`	Approximate pixel size in metres (30 / 90).
`vertical_datum`	`str`	Vertical datum of the elevation values.
`horizontal_datum`	`str`	Horizontal datum of the tile grid.
`attribution`	`str`	Required attribution string; surfaced in docs.
`description`	`str`	Human-readable summary of the dataset.

Examples:

Load the shipped GLO-30 row:

>>> from earthlens.dem import Catalog
>>> row = Catalog().get_dataset("cop-dem-glo-30")
>>> row.bucket
'copernicus-dem-30m'
>>> row.resolution_token
'10'

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

class DEMDataset(BaseModel):
    """One Copernicus DEM dataset row.

    A frozen value object that pins the exact S3 bucket, region, and the
    resolution token embedded in every tile key (`_10_` for GLO-30 or
    `_30_` for GLO-90 — the tokens Copernicus uses in the object names,
    not the pixel size).

    Attributes:
        key: Dataset key (`"cop-dem-glo-30"`, `"cop-dem-glo-90"`) — the
            value passed to `EarthLens(..., dataset=...)`.
        long_name: Human-readable label used in docs and logs.
        bucket: Anonymous S3 bucket that holds the tiles.
        region: AWS region of the bucket (`"eu-central-1"`).
        resolution_token: The `{token}` fragment in the tile file name
            (`"10"` for GLO-30, `"30"` for GLO-90).
        tile_degrees: Edge length of one tile in degrees (`1`).
        native_resolution_m: Approximate pixel size in metres (30 / 90).
        vertical_datum: Vertical datum of the elevation values.
        horizontal_datum: Horizontal datum of the tile grid.
        attribution: Required attribution string; surfaced in docs.
        description: Human-readable summary of the dataset.

    Examples:
        - Load the shipped GLO-30 row:
            ```python
            >>> from earthlens.dem import Catalog
            >>> row = Catalog().get_dataset("cop-dem-glo-30")
            >>> row.bucket
            'copernicus-dem-30m'
            >>> row.resolution_token
            '10'

            ```
    """

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

    key: str
    long_name: str = ""
    bucket: str
    region: str = "eu-central-1"
    resolution_token: str
    tile_degrees: int = 1
    native_resolution_m: int = 0
    vertical_datum: str = ""
    horizontal_datum: str = ""
    attribution: str = ""
    description: str = ""

`clear_catalog_cache()` #

Empty the module-level DEM catalog parse cache.

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

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

`earthlens.dem.backend` #

Copernicus DEM backend — anonymous, no account, raw COG tiles.

DEM(AbstractDataSource) fetches Copernicus DEM GLO-30 / GLO-90 tiles from the anonymous AWS Open Data buckets copernicus-dem-30m and copernicus-dem-90m. The backend is unusual on two counts:

Zero credentials. The buckets are public and unsigned; every other shipped path to a global DEM (gee, stac, earthdata) requires an account. This backend earns its place by offering Copernicus DEM without one — the whole G0 justification.
No decode. A request downloads the raw 1° x 1° COG GeoTIFF(s) as shipped by the bucket and returns their list[Path]. Cropping, reprojecting, and mosaicking are pyramids' job (its COG reader already reads and mosaics COGs). This module does not import rasterio, gdal, osgeo, or xarray.

Request shape: a dataset key ("cop-dem-glo-30" default, or "cop-dem-glo-90") and a WGS84 bbox (lat_lim, lon_lim). The backend is time-invariant — the start / end dates are accepted for the shared AbstractDataSource signature but are advisory only. Ocean and outside-coverage tiles are absent from the bucket and are logged and skipped, never fatal (G6).

`DEM` #

Bases: AbstractDataSource

Anonymous Copernicus DEM backend (raw COG tiles).

Parameters:

Name	Type	Description	Default
`start`	`str \| None`	Advisory start date (parsed with `fmt`); DEM is time-invariant, so an omitted `start` defaults to today.	`None`
`end`	`str \| None`	Advisory end date; defaults to `start`.	`None`
`variables`	`list[str] \| dict[str, list[str]] \| None`	Ignored — a Copernicus DEM tile carries one elevation band. Kept in the signature for `AbstractDataSource` compatibility.	`None`
`lat_lim`	`list[float] \| None`	`[lat_min, lat_max]` in degrees. Selects which 1° tiles to fetch.	`None`
`lon_lim`	`list[float] \| None`	`[lon_min, lon_max]` in degrees.	`None`
`temporal_resolution`	`str`	Advisory label; DEM has no time cadence.	`'static'`
`path`	`Path \| str`	Output directory for the downloaded tiles.	`''`
`fmt`	`str`	`strptime` format for `start` / `end`.	`'%Y-%m-%d'`
`dataset`	`str`	Catalog key — `"cop-dem-glo-30"` (default) or `"cop-dem-glo-90"`.	`'cop-dem-glo-30'`
`catalog`	`Catalog \| None`	Optional pre-built :class:`Catalog` (tests inject one); defaults to the bundled catalog.	`None`

Examples:

Plan the tiles a bbox needs without touching the network:

>>> from earthlens.dem import DEM
>>> src = DEM(
...     start="2026-01-01", end="2026-01-01",
...     variables=[],
...     lat_lim=[30.2, 30.8], lon_lim=[31.2, 31.8],
... )
>>> [(t.lat, t.lon) for t in src.tiles()]
[(30, 31)]

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

class DEM(AbstractDataSource):
    """Anonymous Copernicus DEM backend (raw COG tiles).

    Args:
        start: Advisory start date (parsed with `fmt`); DEM is
            time-invariant, so an omitted `start` defaults to today.
        end: Advisory end date; defaults to `start`.
        variables: Ignored — a Copernicus DEM tile carries one
            elevation band. Kept in the signature for `AbstractDataSource`
            compatibility.
        lat_lim: `[lat_min, lat_max]` in degrees. Selects which 1° tiles
            to fetch.
        lon_lim: `[lon_min, lon_max]` in degrees.
        temporal_resolution: Advisory label; DEM has no time cadence.
        path: Output directory for the downloaded tiles.
        fmt: `strptime` format for `start` / `end`.
        dataset: Catalog key — `"cop-dem-glo-30"` (default) or
            `"cop-dem-glo-90"`.
        catalog: Optional pre-built :class:`Catalog` (tests inject
            one); defaults to the bundled catalog.

    Examples:
        - Plan the tiles a bbox needs without touching the network:
            ```python
            >>> from earthlens.dem import DEM
            >>> src = DEM(
            ...     start="2026-01-01", end="2026-01-01",
            ...     variables=[],
            ...     lat_lim=[30.2, 30.8], lon_lim=[31.2, 31.8],
            ... )
            >>> [(t.lat, t.lon) for t in src.tiles()]
            [(30, 31)]

            ```
    """

    OUTPUT_KIND = "raster"

    def __init__(
        self,
        start: str | None = None,
        end: str | None = None,
        variables: list[str] | dict[str, list[str]] | None = None,
        lat_lim: list[float] | None = None,
        lon_lim: list[float] | None = None,
        temporal_resolution: str = "static",
        path: Path | str = "",
        fmt: str = "%Y-%m-%d",
        *,
        dataset: str = "cop-dem-glo-30",
        catalog: Catalog | None = None,
    ):
        """Initialise a Copernicus DEM backend instance.

        Raises:
            ValueError: If `dataset` is not a curated DEM catalog key.
        """
        self._catalog = catalog if catalog is not None else Catalog()
        self._dataset_key = dataset
        self._dataset: DEMDataset = self._catalog.get_dataset(dataset)
        self._show_progress = True
        # `_initialize` (called by `super().__init__` below) builds `self._auth`.

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

    # -- abstract hooks -------------------------------------------------

    def _initialize(self) -> None:
        """Build the unsigned `boto3` client via the shared `S3Auth`.

        The client is lazy: `S3Auth.client()` builds it on first access
        and caches it. Returning `None` here avoids overwriting
        `self.client` — the auth object stays the source of truth.

        Returns:
            None: The auth object owns the client (accessed via
                :meth:`_client`).
        """
        self._auth = S3Auth(
            S3Credentials(region=self._dataset.region),
        )
        return None

    def _create_grid(
        self, lat_lim: list[float], lon_lim: list[float]
    ) -> SpatialExtent:
        """Wrap the user bbox into a :class:`SpatialExtent` (no snapping)."""
        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:
        """Return a single-instant :class:`TemporalExtent` (DEM is static).

        DEM has no time axis; the returned extent carries a single-date
        `DatetimeIndex` derived from `start` so the shared
        `TemporalExtent` validator passes.
        """
        start_date = pd.to_datetime(start, format=fmt)
        end_date = pd.to_datetime(end, format=fmt)
        return TemporalExtent(
            start_date=start_date,
            end_date=end_date,
            resolution="static",
            dates=pd.DatetimeIndex([start_date]),
        )

    # -- planning + fetch -----------------------------------------------

    def tiles(self) -> list[Tile]:
        """Return the 1° tiles the request bbox intersects.

        Pure arithmetic on the integer-degree grid — no network access.
        Antimeridian-straddling bboxes (`lon_min > lon_max`) are
        rejected at construction; pass the two halves in separate calls
        if you need to span the 180th meridian.

        Returns:
            list[Tile]: One tile per SW-corner integer degree pair in
                the bbox, in row-major (south → north, west → east)
                order.
        """
        return bbox_to_tiles(
            lat_min=self.space.latitude_min,
            lat_max=self.space.latitude_max,
            lon_min=self.space.longitude_min,
            lon_max=self.space.longitude_max,
        )

    def _search(self) -> list[RemoteProduct]:
        """Enumerate one candidate product per 1° tile in the bbox.

        The key is deterministic (no `list_objects_v2` call), so a
        request with a wide bbox does not walk the bucket. Ocean /
        outside-coverage tiles are only discovered as missing when
        `_fetch` calls `head_object` on each candidate.

        Returns:
            list[RemoteProduct]: One :class:`RemoteProduct` per tile;
                `href` carries the S3 key and `metadata` carries the
                bucket, the tile identifier, and the SW corner.
        """
        token = self._dataset.resolution_token
        bucket = self._dataset.bucket
        products: list[RemoteProduct] = []
        for tile in self.tiles():
            key = tile_key(tile, token)
            products.append(
                RemoteProduct(
                    id=key.split("/", 1)[0],
                    href=key,
                    metadata={
                        "bucket": bucket,
                        "dataset": self._dataset_key,
                        "tile_lat": tile.lat,
                        "tile_lon": tile.lon,
                    },
                )
            )
        return products

    def _fetch(self, products: list[RemoteProduct]) -> list[Path]:
        """Download every present tile; log and skip ocean / missing ones.

        The bucket carries one COG per land tile; a candidate key for an
        ocean tile (or outside coverage) simply returns a 404 on
        `head_object`. Those are warned about and skipped rather than
        propagated, so a coastal bbox returns a ragged but non-empty
        result — never a crash.

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

        Returns:
            list[Path]: One local path per tile that existed and was
                downloaded, in `products` order.
        """
        client = self._client()
        written: list[Path] = []
        missing = 0
        for product in tqdm(
            products,
            disable=not (self._show_progress and products),
            desc=f"dem/{self._dataset_key}",
            unit="tile",
        ):
            fetched = self._fetch_one(client, product)
            if fetched is None:
                missing += 1
                continue
            written.append(fetched)
        if missing:
            logger.info(
                f"dem: {len(written)}/{len(products)} tile(s) downloaded; "
                f"{missing} absent (ocean / outside coverage)."
            )
        return written

    def _fetch_one(self, client: Any, product: RemoteProduct) -> Path | None:
        """Head + download one candidate tile; return `None` if absent.

        Args:
            client: The unsigned `boto3` S3 client.
            product: The candidate :class:`RemoteProduct` to fetch.

        Returns:
            Path | None: The written local path, or `None` when the key
                is absent (ocean tile, outside coverage).
        """
        bucket: str = product.metadata["bucket"]
        key: str = product.href or ""
        target = self.root_dir / Path(key).name
        if target.exists():
            return target
        try:
            client.head_object(Bucket=bucket, Key=key)
        except Exception as exc:  # noqa: BLE001 - classify below
            if _is_missing_object(exc):
                logger.warning(
                    f"dem: tile absent, skipping: s3://{bucket}/{key}"
                )
                return None
            raise
        tmp = target.with_name(target.name + ".part")
        try:
            client.download_file(bucket, key, str(tmp))
            tmp.replace(target)
        except BaseException:
            tmp.unlink(missing_ok=True)
            raise
        return target

    def _client(self) -> Any:
        """Return the unsigned `boto3` client (built lazily by `S3Auth`)."""
        return self._auth.client()

    def _api(self, *args: Any, **kwargs: Any) -> list[Path]:
        """Compose `_search` + `_fetch` (the canonical post-C3 body)."""
        return self._api_via_search_fetch()

    def download(
        self,
        progress_bar: bool = True,
        aggregate: Any | None = None,
    ) -> list[Path]:
        """Fetch every 1° Copernicus DEM tile the bbox intersects.

        Args:
            progress_bar: Show a per-tile progress bar. Defaults to
                `True`.
            aggregate: Must be `None`. DEM is a raw-COG whole-tile
                backend (`G4` — no server-side subset, `G5` — no
                decode), so a time-window reduce has nothing to reduce.
                Cropping / mosaicking the tiles is pyramids' job, not
                the backend's.

        Returns:
            list[Path]: The local paths of the downloaded COG tiles, in
                bbox row-major order. Empty when every candidate tile is
                absent from the bucket.

        Raises:
            NotImplementedError: If `aggregate` is not `None`.
        """
        if aggregate is not None:
            raise NotImplementedError(
                "DEM.download(aggregate=...) is not supported — a DEM tile is "
                "time-invariant. Mosaic / crop the downloaded tiles with "
                "pyramids downstream (`pyramids.Dataset.read_file` + "
                "`.crop`, `pyramids.dataset.merge.merge_rasters` for a "
                "multi-tile mosaic)."
            )
        self._show_progress = progress_bar
        return self._api_via_search_fetch()

`init(start=None, end=None, variables=None, lat_lim=None, lon_lim=None, temporal_resolution='static', path='', fmt='%Y-%m-%d', *, dataset='cop-dem-glo-30', catalog=None)` #

Initialise a Copernicus DEM backend instance.

Raises:

Type	Description
`ValueError`	If `dataset` is not a curated DEM catalog key.

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

def __init__(
    self,
    start: str | None = None,
    end: str | None = None,
    variables: list[str] | dict[str, list[str]] | None = None,
    lat_lim: list[float] | None = None,
    lon_lim: list[float] | None = None,
    temporal_resolution: str = "static",
    path: Path | str = "",
    fmt: str = "%Y-%m-%d",
    *,
    dataset: str = "cop-dem-glo-30",
    catalog: Catalog | None = None,
):
    """Initialise a Copernicus DEM backend instance.

    Raises:
        ValueError: If `dataset` is not a curated DEM catalog key.
    """
    self._catalog = catalog if catalog is not None else Catalog()
    self._dataset_key = dataset
    self._dataset: DEMDataset = self._catalog.get_dataset(dataset)
    self._show_progress = True
    # `_initialize` (called by `super().__init__` below) builds `self._auth`.

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

`download(progress_bar=True, aggregate=None)` #

Fetch every 1° Copernicus DEM tile the bbox intersects.

Parameters:

Name	Type	Description	Default
`progress_bar`	`bool`	Show a per-tile progress bar. Defaults to `True`.	`True`
`aggregate`	`Any \| None`	Must be `None`. DEM is a raw-COG whole-tile backend (`G4` — no server-side subset, `G5` — no decode), so a time-window reduce has nothing to reduce. Cropping / mosaicking the tiles is pyramids' job, not the backend's.	`None`

Returns:

Type	Description
`list[Path]`	list[Path]: The local paths of the downloaded COG tiles, in bbox row-major order. Empty when every candidate tile is absent from the bucket.

Raises:

Type	Description
`NotImplementedError`	If `aggregate` is not `None`.

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

def download(
    self,
    progress_bar: bool = True,
    aggregate: Any | None = None,
) -> list[Path]:
    """Fetch every 1° Copernicus DEM tile the bbox intersects.

    Args:
        progress_bar: Show a per-tile progress bar. Defaults to
            `True`.
        aggregate: Must be `None`. DEM is a raw-COG whole-tile
            backend (`G4` — no server-side subset, `G5` — no
            decode), so a time-window reduce has nothing to reduce.
            Cropping / mosaicking the tiles is pyramids' job, not
            the backend's.

    Returns:
        list[Path]: The local paths of the downloaded COG tiles, in
            bbox row-major order. Empty when every candidate tile is
            absent from the bucket.

    Raises:
        NotImplementedError: If `aggregate` is not `None`.
    """
    if aggregate is not None:
        raise NotImplementedError(
            "DEM.download(aggregate=...) is not supported — a DEM tile is "
            "time-invariant. Mosaic / crop the downloaded tiles with "
            "pyramids downstream (`pyramids.Dataset.read_file` + "
            "`.crop`, `pyramids.dataset.merge.merge_rasters` for a "
            "multi-tile mosaic)."
        )
    self._show_progress = progress_bar
    return self._api_via_search_fetch()

`tiles()` #

Return the 1° tiles the request bbox intersects.

Pure arithmetic on the integer-degree grid — no network access. Antimeridian-straddling bboxes (lon_min > lon_max) are rejected at construction; pass the two halves in separate calls if you need to span the 180th meridian.

Returns:

Type	Description
`list[Tile]`	list[Tile]: One tile per SW-corner integer degree pair in the bbox, in row-major (south → north, west → east) order.

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

def tiles(self) -> list[Tile]:
    """Return the 1° tiles the request bbox intersects.

    Pure arithmetic on the integer-degree grid — no network access.
    Antimeridian-straddling bboxes (`lon_min > lon_max`) are
    rejected at construction; pass the two halves in separate calls
    if you need to span the 180th meridian.

    Returns:
        list[Tile]: One tile per SW-corner integer degree pair in
            the bbox, in row-major (south → north, west → east)
            order.
    """
    return bbox_to_tiles(
        lat_min=self.space.latitude_min,
        lat_max=self.space.latitude_max,
        lon_min=self.space.longitude_min,
        lon_max=self.space.longitude_max,
    )

`earthlens.dem.catalog` #

Dataset catalog for the Copernicus DEM backend.

Hosts :class:Catalog, the pydantic-backed reader for the bundled dem_data_catalog.yaml. The catalog is a small single-family map from a dataset key (cop-dem-glo-30, cop-dem-glo-90) to the anonymous S3 bucket, region, and the resolution token that appears in the tile key.

The DEM backend is one-axis: a request picks one dataset key and a bbox; the tile enumeration is arithmetic on the integer-degree grid. There are no variables (a Copernicus DEM tile is one elevation band), so the catalog has no per-variable sub-block.

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

`Catalog` #

Bases: AbstractCatalog

Dataset catalog for the DEM backend.

Reads the bundled dem_data_catalog.yaml (shipped as package data) and exposes its datasets: block as a map of :class:DEMDataset rows keyed by dataset key under the inherited :attr:datasets field, which supplies the cat["cop-dem-glo-30"] / "cop-dem-glo-30" in cat / len(cat) dict-like surface and the did-you-mean error for free. Instantiate with no arguments (Catalog()); :func:model_post_init loads and validates the YAML in one pass and caches it by (path, mtime).

Attributes:

Name	Type	Description
`datasets`	`dict[str, DEMDataset]`	Map from dataset key to its :class:`DEMDataset` row.
`available_datasets`	`list[str]`	Sorted dataset keys — the curated datasets are the whole dataset universe for this backend.

Examples:

List datasets and resolve one:

>>> from earthlens.dem import Catalog
>>> cat = Catalog()
>>> sorted(cat.datasets)
['cop-dem-glo-30', 'cop-dem-glo-90']
>>> cat.get_dataset("cop-dem-glo-90").native_resolution_m
90

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

>>> from earthlens.dem import Catalog
>>> Catalog().get_dataset("cop-dem-glo-3")  # doctest: +ELLIPSIS
Traceback (most recent call last):
    ...
ValueError: 'cop-dem-glo-3' is not in the DEM catalog. ... Did you mean 'cop-dem-glo-30'?

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

class Catalog(AbstractCatalog):
    """Dataset catalog for the DEM backend.

    Reads the bundled `dem_data_catalog.yaml` (shipped as package data)
    and exposes its `datasets:` block as a map of :class:`DEMDataset`
    rows keyed by dataset key under the inherited :attr:`datasets`
    field, which supplies the `cat["cop-dem-glo-30"]` /
    `"cop-dem-glo-30" in cat` / `len(cat)` dict-like surface and the
    did-you-mean error for free. Instantiate with no arguments
    (`Catalog()`); :func:`model_post_init` loads and validates the YAML
    in one pass and caches it by `(path, mtime)`.

    Attributes:
        datasets: Map from dataset key to its :class:`DEMDataset` row.
        available_datasets: Sorted dataset keys — the curated datasets
            are the whole dataset universe for this backend.

    Examples:
        - List datasets and resolve one:
            ```python
            >>> from earthlens.dem import Catalog
            >>> cat = Catalog()
            >>> sorted(cat.datasets)
            ['cop-dem-glo-30', 'cop-dem-glo-90']
            >>> cat.get_dataset("cop-dem-glo-90").native_resolution_m
            90

            ```
        - An unknown key raises with a did-you-mean hint:
            ```python
            >>> from earthlens.dem import Catalog
            >>> Catalog().get_dataset("cop-dem-glo-3")  # doctest: +ELLIPSIS
            Traceback (most recent call last):
                ...
            ValueError: 'cop-dem-glo-3' is not in the DEM catalog. ... Did you mean 'cop-dem-glo-30'?

            ```
    """

    _catalog_kind: str = "DEM catalog"

    datasets: dict[str, DEMDataset] = Field(default_factory=dict)

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

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

        Raises:
            ValueError: Propagated from :meth:`load` when the YAML is
                missing, empty, or has a malformed row.
        """
        if not self.datasets:
            loaded = Catalog.load()
            self.datasets = loaded.datasets
        self.available_datasets = sorted(self.datasets)

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

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

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

        Raises:
            ValueError: If the file has no `datasets:` block, or any
                row fails validation.
        """
        path = catalog_path if catalog_path is not None else CATALOG_PATH
        try:
            mtime = path.stat().st_mtime_ns
        except FileNotFoundError:
            mtime = 0
        cache_key = (str(path.resolve()), mtime)
        cached = _CATALOG_CACHE.get(cache_key)
        if cached is not None:
            return cached

        data = load_yaml_strict(path) or {}
        datasets_yaml = data.get("datasets") or {}
        if not datasets_yaml:
            raise ValueError(
                f"{path} is missing or has an empty 'datasets:' block. "
                "The DEM catalog must list at least one dataset."
            )
        datasets: dict[str, DEMDataset] = {}
        for key, body in datasets_yaml.items():
            try:
                datasets[key] = DEMDataset(key=key, **dict(body or {}))
            except ValidationError as exc:
                raise ValueError(
                    f"{path} dataset {key!r} failed validation:\n{exc}"
                ) from exc
        catalog = cls(datasets=datasets)
        _CATALOG_CACHE[cache_key] = catalog
        return catalog

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

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

`get_catalog()` #

Return the dataset map (satisfies the abstract contract).

Returns:

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

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

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

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

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

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

Parameters:

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

Returns:

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

Raises:

Type	Description
`ValueError`	If the file has no `datasets:` block, or any row fails validation.

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

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

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

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

    Raises:
        ValueError: If the file has no `datasets:` block, or any
            row fails validation.
    """
    path = catalog_path if catalog_path is not None else CATALOG_PATH
    try:
        mtime = path.stat().st_mtime_ns
    except FileNotFoundError:
        mtime = 0
    cache_key = (str(path.resolve()), mtime)
    cached = _CATALOG_CACHE.get(cache_key)
    if cached is not None:
        return cached

    data = load_yaml_strict(path) or {}
    datasets_yaml = data.get("datasets") or {}
    if not datasets_yaml:
        raise ValueError(
            f"{path} is missing or has an empty 'datasets:' block. "
            "The DEM catalog must list at least one dataset."
        )
    datasets: dict[str, DEMDataset] = {}
    for key, body in datasets_yaml.items():
        try:
            datasets[key] = DEMDataset(key=key, **dict(body or {}))
        except ValidationError as exc:
            raise ValueError(
                f"{path} dataset {key!r} failed validation:\n{exc}"
            ) from exc
    catalog = cls(datasets=datasets)
    _CATALOG_CACHE[cache_key] = catalog
    return catalog

`model_post_init(__context)` #

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

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

Raises:

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

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

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

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

    Raises:
        ValueError: Propagated from :meth:`load` when the YAML is
            missing, empty, or has a malformed row.
    """
    if not self.datasets:
        loaded = Catalog.load()
        self.datasets = loaded.datasets
    self.available_datasets = sorted(self.datasets)

`DEMDataset` #

Bases: BaseModel

One Copernicus DEM dataset row.

A frozen value object that pins the exact S3 bucket, region, and the resolution token embedded in every tile key (_10_ for GLO-30 or _30_ for GLO-90 — the tokens Copernicus uses in the object names, not the pixel size).

Attributes:

Name	Type	Description
`key`	`str`	Dataset key (`"cop-dem-glo-30"`, `"cop-dem-glo-90"`) — the value passed to `EarthLens(..., dataset=...)`.
`long_name`	`str`	Human-readable label used in docs and logs.
`bucket`	`str`	Anonymous S3 bucket that holds the tiles.
`region`	`str`	AWS region of the bucket (`"eu-central-1"`).
`resolution_token`	`str`	The `{token}` fragment in the tile file name (`"10"` for GLO-30, `"30"` for GLO-90).
`tile_degrees`	`int`	Edge length of one tile in degrees (`1`).
`native_resolution_m`	`int`	Approximate pixel size in metres (30 / 90).
`vertical_datum`	`str`	Vertical datum of the elevation values.
`horizontal_datum`	`str`	Horizontal datum of the tile grid.
`attribution`	`str`	Required attribution string; surfaced in docs.
`description`	`str`	Human-readable summary of the dataset.

Examples:

Load the shipped GLO-30 row:

>>> from earthlens.dem import Catalog
>>> row = Catalog().get_dataset("cop-dem-glo-30")
>>> row.bucket
'copernicus-dem-30m'
>>> row.resolution_token
'10'

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

class DEMDataset(BaseModel):
    """One Copernicus DEM dataset row.

    A frozen value object that pins the exact S3 bucket, region, and the
    resolution token embedded in every tile key (`_10_` for GLO-30 or
    `_30_` for GLO-90 — the tokens Copernicus uses in the object names,
    not the pixel size).

    Attributes:
        key: Dataset key (`"cop-dem-glo-30"`, `"cop-dem-glo-90"`) — the
            value passed to `EarthLens(..., dataset=...)`.
        long_name: Human-readable label used in docs and logs.
        bucket: Anonymous S3 bucket that holds the tiles.
        region: AWS region of the bucket (`"eu-central-1"`).
        resolution_token: The `{token}` fragment in the tile file name
            (`"10"` for GLO-30, `"30"` for GLO-90).
        tile_degrees: Edge length of one tile in degrees (`1`).
        native_resolution_m: Approximate pixel size in metres (30 / 90).
        vertical_datum: Vertical datum of the elevation values.
        horizontal_datum: Horizontal datum of the tile grid.
        attribution: Required attribution string; surfaced in docs.
        description: Human-readable summary of the dataset.

    Examples:
        - Load the shipped GLO-30 row:
            ```python
            >>> from earthlens.dem import Catalog
            >>> row = Catalog().get_dataset("cop-dem-glo-30")
            >>> row.bucket
            'copernicus-dem-30m'
            >>> row.resolution_token
            '10'

            ```
    """

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

    key: str
    long_name: str = ""
    bucket: str
    region: str = "eu-central-1"
    resolution_token: str
    tile_degrees: int = 1
    native_resolution_m: int = 0
    vertical_datum: str = ""
    horizontal_datum: str = ""
    attribution: str = ""
    description: str = ""

`clear_catalog_cache()` #

Empty the module-level DEM catalog parse cache.

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

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

`earthlens.dem._helpers` #

Pure tile-key arithmetic for the Copernicus DEM buckets.

The Copernicus DEM COGs live one per 1° x 1° tile on the anonymous AWS buckets. Every tile is addressed by its SW-corner integer degrees on a fixed grid, encoded in the object name as N##/S## for latitude (2-digit) and E###/W### for longitude (3-digit). This module is the one place that arithmetic lives, so the backend, the tests, and the gated e2e all resolve a bbox to the same list of tile keys.

Nothing here touches the network — the enumeration is deterministic integer arithmetic.

`Tile` `dataclass` #

One Copernicus DEM 1° x 1° tile addressed by its SW corner.

Attributes:

Name	Type	Description
`lat`	`int`	SW-corner latitude in integer degrees, in `[-90, 89]`.
`lon`	`int`	SW-corner longitude in integer degrees, in `[-180, 179]`.

Examples:

Nile Delta cell (30 N, 31 E):

>>> from earthlens.dem._helpers import Tile
>>> t = Tile(lat=30, lon=31)
>>> t.lat, t.lon
(30, 31)

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

@dataclass(frozen=True)
class Tile:
    """One Copernicus DEM 1° x 1° tile addressed by its SW corner.

    Attributes:
        lat: SW-corner latitude in integer degrees, in `[-90, 89]`.
        lon: SW-corner longitude in integer degrees, in `[-180, 179]`.

    Examples:
        - Nile Delta cell (30 N, 31 E):
            ```python
            >>> from earthlens.dem._helpers import Tile
            >>> t = Tile(lat=30, lon=31)
            >>> t.lat, t.lon
            (30, 31)

            ```
    """

    lat: int
    lon: int

`bbox_to_tiles(lat_min, lat_max, lon_min, lon_max)` #

Enumerate every 1° tile whose SW corner lies inside the bbox.

Snaps the bbox to the integer-degree tile grid: for each axis the minimum corner is floored and the maximum corner is included up to (and excluding) the ceiling. A bbox that lies entirely inside a single tile still returns that tile. Latitude and longitude values are clamped to the WGS84 grid range.

Antimeridian-straddling bboxes (lon_min > lon_max) are not supported in this first cut — pass the two halves in separate calls.

Parameters:

Name	Type	Description	Default
`lat_min`	`float`	Southern edge of the bbox in degrees.	required
`lat_max`	`float`	Northern edge of the bbox in degrees.	required
`lon_min`	`float`	Western edge of the bbox in degrees.	required
`lon_max`	`float`	Eastern edge of the bbox in degrees.	required

Returns:

Type	Description
`list[Tile]`	list[Tile]: Every 1° tile whose SW corner lies within the bbox, in row-major order (south -> north, then west -> east).

Raises:

Type	Description
`ValueError`	If `lat_min > lat_max`, `lon_min > lon_max`, or an axis value is outside the WGS84 range.

Examples:

A single tile bbox:

>>> from earthlens.dem._helpers import bbox_to_tiles
>>> [(t.lat, t.lon) for t in bbox_to_tiles(30.2, 30.8, 31.2, 31.8)]
[(30, 31)]

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

def bbox_to_tiles(
    lat_min: float,
    lat_max: float,
    lon_min: float,
    lon_max: float,
) -> list[Tile]:
    """Enumerate every 1° tile whose SW corner lies inside the bbox.

    Snaps the bbox to the integer-degree tile grid: for each axis the
    minimum corner is floored and the maximum corner is included up to
    (and excluding) the ceiling. A bbox that lies entirely inside a
    single tile still returns that tile. Latitude and longitude values
    are clamped to the WGS84 grid range.

    Antimeridian-straddling bboxes (`lon_min > lon_max`) are not
    supported in this first cut — pass the two halves in separate calls.

    Args:
        lat_min: Southern edge of the bbox in degrees.
        lat_max: Northern edge of the bbox in degrees.
        lon_min: Western edge of the bbox in degrees.
        lon_max: Eastern edge of the bbox in degrees.

    Returns:
        list[Tile]: Every 1° tile whose SW corner lies within the
            bbox, in row-major order (south -> north, then west ->
            east).

    Raises:
        ValueError: If `lat_min > lat_max`, `lon_min > lon_max`, or an
            axis value is outside the WGS84 range.

    Examples:
        - A single tile bbox:
            ```python
            >>> from earthlens.dem._helpers import bbox_to_tiles
            >>> [(t.lat, t.lon) for t in bbox_to_tiles(30.2, 30.8, 31.2, 31.8)]
            [(30, 31)]

            ```
    """
    if lat_min > lat_max:
        raise ValueError(f"lat_min {lat_min} > lat_max {lat_max}")
    if lon_min > lon_max:
        raise ValueError(
            f"lon_min {lon_min} > lon_max {lon_max} — antimeridian-straddling "
            "bboxes are not supported; pass the two halves in separate calls."
        )
    if not (-90.0 <= lat_min <= 90.0 and -90.0 <= lat_max <= 90.0):
        raise ValueError(
            f"latitude out of [-90, 90]: lat_min={lat_min}, lat_max={lat_max}"
        )
    if not (-180.0 <= lon_min <= 180.0 and -180.0 <= lon_max <= 180.0):
        raise ValueError(
            f"longitude out of [-180, 180]: lon_min={lon_min}, lon_max={lon_max}"
        )

    lat_origins = _axis_origins(lat_min, lat_max, min_index=-90, max_index=89)
    lon_origins = _axis_origins(lon_min, lon_max, min_index=-180, max_index=179)
    return [Tile(lat=lat, lon=lon) for lat in lat_origins for lon in lon_origins]

`tile_key(tile, resolution_token)` #

Return the bucket-relative object key for the tile's DEM COG.

A tile directory in the bucket carries the DEM .tif at <name>/<name>.tif plus a set of sidecars (AUXFILES/, PREVIEW/, INFO/) that the DEM backend deliberately ignores.

Parameters:

Name	Type	Description	Default
`tile`	`Tile`	The tile addressed by its SW-corner integer degrees.	required
`resolution_token`	`str`	`"10"` for GLO-30, `"30"` for GLO-90.	required

Returns:

Name	Type	Description
`str`	`str`	The bucket-relative key to the DEM COG.

Examples:

GLO-30 Nile Delta:

>>> from earthlens.dem._helpers import Tile, tile_key
>>> tile_key(Tile(lat=30, lon=31), "10")
'Copernicus_DSM_COG_10_N30_00_E031_00_DEM/Copernicus_DSM_COG_10_N30_00_E031_00_DEM.tif'

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

def tile_key(tile: Tile, resolution_token: str) -> str:
    """Return the bucket-relative object key for the tile's DEM COG.

    A tile directory in the bucket carries the DEM `.tif` at
    `<name>/<name>.tif` plus a set of sidecars (`AUXFILES/`,
    `PREVIEW/`, `INFO/`) that the DEM backend deliberately ignores.

    Args:
        tile: The tile addressed by its SW-corner integer degrees.
        resolution_token: `"10"` for GLO-30, `"30"` for GLO-90.

    Returns:
        str: The bucket-relative key to the DEM COG.

    Examples:
        - GLO-30 Nile Delta:
            ```python
            >>> from earthlens.dem._helpers import Tile, tile_key
            >>> tile_key(Tile(lat=30, lon=31), "10")
            'Copernicus_DSM_COG_10_N30_00_E031_00_DEM/Copernicus_DSM_COG_10_N30_00_E031_00_DEM.tif'

            ```
    """
    name = tile_name(tile, resolution_token)
    return f"{name}/{name}.tif"

`tile_name(tile, resolution_token)` #

Return the tile identifier (also the top-level bucket "folder").

The identifier follows the Copernicus naming convention verified against the live buckets: Copernicus_DSM_COG_{token}_{LAT}_00_{LON}_00_DEM, where the _00_ markers are the arc-minute portion of the corner (always 00 on the 1° grid).

Parameters:

Name	Type	Description	Default
`tile`	`Tile`	The tile addressed by its SW-corner integer degrees.	required
`resolution_token`	`str`	`"10"` for GLO-30 or `"30"` for GLO-90 — the token embedded in the tile name, taken from the catalog row.	required

Returns:

Name	Type	Description
`str`	`str`	The tile identifier without a trailing suffix.

Examples:

GLO-30 Nile Delta (30 N, 31 E):

>>> from earthlens.dem._helpers import Tile, tile_name
>>> tile_name(Tile(lat=30, lon=31), "10")
'Copernicus_DSM_COG_10_N30_00_E031_00_DEM'

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

def tile_name(tile: Tile, resolution_token: str) -> str:
    """Return the tile identifier (also the top-level bucket "folder").

    The identifier follows the Copernicus naming convention verified
    against the live buckets:
    `Copernicus_DSM_COG_{token}_{LAT}_00_{LON}_00_DEM`, where the
    `_00_` markers are the arc-minute portion of the corner (always
    `00` on the 1° grid).

    Args:
        tile: The tile addressed by its SW-corner integer degrees.
        resolution_token: `"10"` for GLO-30 or `"30"` for GLO-90 —
            the token embedded in the tile name, taken from the
            catalog row.

    Returns:
        str: The tile identifier without a trailing suffix.

    Examples:
        - GLO-30 Nile Delta (30 N, 31 E):
            ```python
            >>> from earthlens.dem._helpers import Tile, tile_name
            >>> tile_name(Tile(lat=30, lon=31), "10")
            'Copernicus_DSM_COG_10_N30_00_E031_00_DEM'

            ```
    """
    return (
        f"Copernicus_DSM_COG_{resolution_token}_"
        f"{_fmt_lat(tile.lat)}_00_{_fmt_lon(tile.lon)}_00_DEM"
    )