JRC European flood hazard — API reference

JRC European Flood Hazard Map (EFHM) backend subpackage — earthlens.jrc_flood. Background and usage are covered under the other pages in this section (Introduction, Usage, Available datasets); this page is the rendered API. The EFHM is public (anonymous HTTPS), so there is no auth module.

earthlens.jrc_flood

JRC European flood-hazard (EFHM) backend for earthlens.

Exposes JRCFlood, the AbstractDataSource backend that reads the JRC European Flood Hazard Map (river-flood water depth per return period, Europe and the Mediterranean Basin) from the open JRC HTTPS directory via lazy /vsicurl windowed reads and crops it to the AOI via pyramids, plus its Catalog / Dataset catalog surface.

Catalog

Bases: AbstractCatalog

Product catalog for the JRC European flood-hazard backend.

Reads the bundled jrc_flood_data_catalog.yaml and exposes its single row under the inherited datasets field — which supplies the cat["efhm"] / "efhm" in cat / len(cat) surface and the did-you-mean error for free. Instantiate with no arguments; the base model_post_init auto-loads via _autoload, cached by (path, mtime).

Attributes:

Name	Type	Description
datasets	dict[str, Dataset]	Product key to its Dataset row.
available_datasets	list[str]	Sorted product keys.
license_id	str	SPDX-ish licence label ("CC-BY-4.0", permissive).
attribution	str	The citation the licence requires.

Examples:

• List products and read the licence:

  >>> from earthlens.jrc_flood import Catalog
  >>> cat = Catalog()
  >>> list(cat.datasets)
  ['efhm']
  >>> cat.license_id
  'CC-BY-4.0'
  

Source code in libs/providers/hazards/src/earthlens/jrc_flood/catalog.py

class Catalog(AbstractCatalog):
    """Product catalog for the JRC European flood-hazard backend.

    Reads the bundled `jrc_flood_data_catalog.yaml` and exposes its single row
    under the inherited `datasets` field — which supplies the `cat["efhm"]` /
    `"efhm" in cat` / `len(cat)` surface and the did-you-mean error for free.
    Instantiate with no arguments; the base `model_post_init` auto-loads via
    `_autoload`, cached by `(path, mtime)`.

    Attributes:
        datasets: Product key to its `Dataset` row.
        available_datasets: Sorted product keys.
        license_id: SPDX-ish licence label (`"CC-BY-4.0"`, permissive).
        attribution: The citation the licence requires.

    Examples:
        - List products and read the licence:
            ```python
            >>> from earthlens.jrc_flood import Catalog
            >>> cat = Catalog()
            >>> list(cat.datasets)
            ['efhm']
            >>> cat.license_id
            'CC-BY-4.0'

            ```
    """

    _catalog_kind: str = "JRC-flood catalog"

    datasets: dict[str, Dataset] = Field(default_factory=dict)
    license_id: str = ""
    attribution: str = ""

    @classmethod
    def _autoload(cls) -> dict[str, Any]:
        """Return the disk payload to fill an empty catalog (base post-init hook).

        Returns:
            dict[str, Any]: The parsed field → value map from `_parse_catalog`.
        """
        return load_catalog(
            CATALOG_PATH, _CATALOG_CACHE, _parse_catalog, provider="JRC-flood"
        )

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

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

        Returns:
            Catalog: A fully-populated catalog.

        Raises:
            ValueError: If `catalog_path` does not exist, has no `datasets:`
                block, or the row fails validation.
        """
        path = catalog_path if catalog_path is not None else CATALOG_PATH
        payload = load_catalog(
            path, _CATALOG_CACHE, _parse_catalog, provider="JRC-flood"
        )
        return cls(**payload)

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

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

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

        Args:
            key: A product key (`"efhm"`).

        Returns:
            Dataset: The matching product row.

        Raises:
            ValueError: If `key` is not a registered product.
        """
        return cast("Dataset", self.get_dataset(key))

get(key)

Return the Dataset for key, with a did-you-mean hint.

Parameters:

Name	Type	Description	Default
key	str	A product key ("efhm").	required

Returns:

Name	Type	Description
Dataset	Dataset	The matching product row.

Raises:

Type	Description
ValueError	If key is not a registered product.

Source code in libs/providers/hazards/src/earthlens/jrc_flood/catalog.py

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

    Args:
        key: A product key (`"efhm"`).

    Returns:
        Dataset: The matching product row.

    Raises:
        ValueError: If `key` is not a registered product.
    """
    return cast("Dataset", self.get_dataset(key))

get_catalog()

Return the product map (satisfies the abstract contract).

Returns:

Type	Description
dict[str, Dataset]	dict[str, Dataset]: Same object as datasets.

Source code in libs/providers/hazards/src/earthlens/jrc_flood/catalog.py

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

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

load(catalog_path=None) classmethod

Read and validate the JRC-flood catalog from disk (cached).

Parameters:

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

Returns:

Name	Type	Description
Catalog	Catalog	A fully-populated catalog.

Raises:

Type	Description
ValueError	If catalog_path does not exist, has no datasets: block, or the row fails validation.

Source code in libs/providers/hazards/src/earthlens/jrc_flood/catalog.py

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

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

    Returns:
        Catalog: A fully-populated catalog.

    Raises:
        ValueError: If `catalog_path` does not exist, has no `datasets:`
            block, or the row fails validation.
    """
    path = catalog_path if catalog_path is not None else CATALOG_PATH
    payload = load_catalog(
        path, _CATALOG_CACHE, _parse_catalog, provider="JRC-flood"
    )
    return cls(**payload)

Dataset

Bases: BaseModel

One EFHM product row.

Attributes:

Name	Type	Description
id	str	The catalog key ("efhm").
title	str	Human-readable product title.
band	str	The single water-depth band name ("water_depth").
long_name	str	Human-readable band description.
units	str	Physical units of the band ("m").
dtype	str	Pixel data type ("float32").
crs	str	Native CRS as an EPSG string ("EPSG:4326").
nodata	float	The raster no-data value.
spatial_resolution	float | None	Nominal resolution in metres (90).
base_url	str	The JRC directory root the return-period files live in.
filename_template	str	The per-return-period file-name template ("Europe_RP{rp}_filled_depth.tif").
return_periods	list[int]	The published return periods in years.
source_url	str	The dataset landing page.

Examples:

• Read the return periods and band:

  >>> from earthlens.jrc_flood import Catalog
  >>> row = Catalog().get("efhm")
  >>> row.band, row.return_periods[:3]
  ('water_depth', [10, 20, 30])
  

Source code in libs/providers/hazards/src/earthlens/jrc_flood/catalog.py

class Dataset(BaseModel):
    """One EFHM product row.

    Attributes:
        id: The catalog key (`"efhm"`).
        title: Human-readable product title.
        band: The single water-depth band name (`"water_depth"`).
        long_name: Human-readable band description.
        units: Physical units of the band (`"m"`).
        dtype: Pixel data type (`"float32"`).
        crs: Native CRS as an EPSG string (`"EPSG:4326"`).
        nodata: The raster no-data value.
        spatial_resolution: Nominal resolution in metres (`90`).
        base_url: The JRC directory root the return-period files live in.
        filename_template: The per-return-period file-name template
            (`"Europe_RP{rp}_filled_depth.tif"`).
        return_periods: The published return periods in years.
        source_url: The dataset landing page.

    Examples:
        - Read the return periods and band:
            ```python
            >>> from earthlens.jrc_flood import Catalog
            >>> row = Catalog().get("efhm")
            >>> row.band, row.return_periods[:3]
            ('water_depth', [10, 20, 30])

            ```
    """

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

    id: str
    title: str = ""
    band: str = "water_depth"
    long_name: str = ""
    units: str = "m"
    dtype: str = "float32"
    crs: str = "EPSG:4326"
    nodata: float = -9999.0
    spatial_resolution: float | None = None
    base_url: str = ""
    filename_template: str = "Europe_RP{rp}_filled_depth.tif"
    return_periods: list[int] = Field(default_factory=list)
    source_url: str = ""

JRCFlood

Bases: AbstractDataSource

JRC European Flood Hazard Map backend (raster GeoTIFF output).

Fetches the EFHM water-depth grid for one or more return periods, cropped to the request bbox, via lazy /vsicurl windowed reads. The request is a search/fetch split: _search names one product per return period, _fetch realises each (windowed read → crop → GeoTIFF).

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	Fixed "raster"; each return period yields a gridded GeoTIFF. The facade reads it to gate aggregate= (rejected — the return periods are not a temporal axis).

Examples:

• A small AOI writes one cropped GeoTIFF per return period (marked +SKIP — it hits the live JRC directory):

  >>> from earthlens.earthlens import EarthLens
  >>> paths = EarthLens(  # doctest: +SKIP
  ...     data_source="jrc-flood",
  ...     lat_lim=[51.8, 52.0],
  ...     lon_lim=[4.8, 5.0],
  ...     return_periods=[100],
  ...     path="efhm_out",
  ... ).download()  # -> [Path('efhm_out/efhm_RP100.tif')]
  

Source code in libs/providers/hazards/src/earthlens/jrc_flood/backend.py

class JRCFlood(AbstractDataSource):
    """JRC European Flood Hazard Map backend (raster GeoTIFF output).

    Fetches the EFHM water-depth grid for one or more return periods, cropped to
    the request bbox, via lazy `/vsicurl` windowed reads. The request is a
    search/fetch split: `_search` names one product per return period, `_fetch`
    realises each (windowed read → crop → GeoTIFF).

    Attributes:
        OUTPUT_KIND: Fixed `"raster"`; each return period yields a gridded
            GeoTIFF. The facade reads it to gate `aggregate=` (rejected — the
            return periods are not a temporal axis).

    Examples:
        - A small AOI writes one cropped GeoTIFF per return period (marked
          `+SKIP` — it hits the live JRC directory):

            ```python
            >>> from earthlens.earthlens import EarthLens
            >>> paths = EarthLens(  # doctest: +SKIP
            ...     data_source="jrc-flood",
            ...     lat_lim=[51.8, 52.0],
            ...     lon_lim=[4.8, 5.0],
            ...     return_periods=[100],
            ...     path="efhm_out",
            ... ).download()  # -> [Path('efhm_out/efhm_RP100.tif')]

            ```
    """

    OUTPUT_KIND: OutputKind = "raster"

    AGGREGATE_REFUSAL_REASON = "the JRC flood hazard map is a set of static per-return-period depth grids with no temporal axis, so there is nothing to reduce. Call download() without aggregate="

    #: Clips to the exact polygon when `aoi=` carries one, not just its bbox.
    SUPPORTS_POLYGON_AOI = True

    #: The EFHM is time-invariant, so a missing `start` / `end` is legal here.
    REQUIRES_TIME_WINDOW = False

    def __init__(
        self,
        start: str = "",
        end: str = "",
        lat_lim: list[float] | None = None,
        lon_lim: list[float] | None = None,
        return_periods: list[int | str] | int | str | None = None,
        temporal_resolution: str = "static",
        path: Path | str | None = None,
        fmt: str = "%Y-%m-%d",
        *,
        catalog: Catalog | None = None,
    ):
        """Initialise a JRC-flood backend instance.

        The EFHM has a single `water_depth` band, so the backend is facet-only
        (it declares no `variables` axis); the request axis is `return_periods`.

        Args:
            start: Accepted for facade parity; ignored (the EFHM is static).
            end: Accepted for facade parity; ignored.
            lat_lim: `[lat_min, lat_max]` bounding-box latitudes. Required.
            lon_lim: `[lon_min, lon_max]` bounding-box longitudes. Required.
            return_periods: One return period, or a list, in years — as ints
                (`100`) or strings (`"100"` / `"RP100"`). Defaults to `[100]`.
                Every value must be a published return period.
            temporal_resolution: Advisory label only (the EFHM is static).
            path: Output directory for the written GeoTIFF(s).
            fmt: Accepted for facade parity; unused.
            catalog: Optional pre-built `Catalog` (tests inject a faked one);
                defaults to the bundled catalog.

        Raises:
            ValueError: If the bounding box is missing or a requested return
                period is not published.
        """
        if lat_lim is None or lon_lim is None:
            raise ValueError(
                "JRCFlood requires a bounding box (lat_lim=[s, n], "
                "lon_lim=[w, e]) — a hazard-map subset has no default extent."
            )

        self._catalog = catalog if catalog is not None else Catalog()
        self._dataset: Dataset = self._catalog.get("efhm")
        self._return_periods = self._resolve_return_periods(return_periods)

        super().__init__(
            start=start,
            end=end,
            variables=[self._dataset.band],
            temporal_resolution=temporal_resolution,
            lat_lim=lat_lim,
            lon_lim=lon_lim,
            fmt=fmt,
            path=path,
        )

    def _resolve_return_periods(
        self, return_periods: list[int | str] | int | str | None
    ) -> list[int]:
        """Normalise + validate the requested return periods against the catalog.

        Accepts a single value or a list; each may be an int (`100`) or a string
        (`"100"` / `"RP100"`, case-insensitive). Defaults to `[100]`.

        Args:
            return_periods: The raw request value.

        Returns:
            list[int]: Sorted, de-duplicated return periods to fetch.

        Raises:
            ValueError: If a value is unparseable or is not a published return
                period.
        """
        available = self._dataset.return_periods
        if return_periods is None:
            requested_raw: list[int | str] = [100]
        elif isinstance(return_periods, (list, tuple)):
            requested_raw = list(return_periods)
        else:
            requested_raw = [return_periods]

        resolved: list[int] = []
        for value in requested_raw:
            resolved.append(self._parse_rp(value))
        unknown = [rp for rp in resolved if rp not in available]
        if unknown:
            raise ValueError(
                f"return period(s) {unknown} are not published for the EFHM; "
                f"available: {available}."
            )
        return sorted(set(resolved))

    @staticmethod
    def _parse_rp(value: int | str) -> int:
        """Parse one return-period token to an int (`100`, `"100"`, `"RP100"`).

        Args:
            value: The raw return-period value.

        Returns:
            int: The integer return period in years.

        Raises:
            ValueError: If `value` is not an int or an `RP`-prefixed / bare
                integer string.
        """
        if isinstance(value, int):
            return value
        text = str(value).strip().upper()
        if text.startswith("RP"):
            text = text[2:]
        try:
            return int(text)
        except ValueError:
            raise ValueError(
                f"could not parse return period {value!r} (expected e.g. 100, "
                "'100', or 'RP100')."
            ) from None

    def _initialize(self):
        """No-op initialiser — the EFHM is public + anonymous (no client).

        Returns:
            None: The parent binds no `self.client`.
        """
        return None

    def _check_input_dates(
        self, start: str, end: str, temporal_resolution: str, fmt: str
    ) -> TemporalExtent:
        """Return a degenerate (timeless) extent — the EFHM is static.

        Args:
            start: Ignored.
            end: Ignored.
            temporal_resolution: Recorded as the resolution label.
            fmt: Ignored.

        Returns:
            TemporalExtent: A frozen model with `None` bounds and an empty date
                index (a static hazard map has no time axis).
        """
        return self._static_extent(resolution=temporal_resolution or "static")

    @property
    def _bbox(self) -> tuple[float, float, float, float]:
        """The AOI as `(west, south, east, north)` in degrees."""
        return (self.space.west, self.space.south, self.space.east, self.space.north)

    def _bbox_overlaps(self, source: Any) -> bool:
        """Whether the AOI overlaps the source raster's geographic extent.

        Delegates to `earthlens.base.spatial.bbox_overlaps` so an AOI outside the
        EFHM's Europe / Mediterranean coverage is reported with a clear error
        before the windowed crop, rather than surfacing as an empty or opaque
        crop result.

        Args:
            source: An opened `pyramids.Dataset` exposing `geotransform`,
                `columns`, and `rows`.

        Returns:
            bool: `True` when the AOI bbox intersects the raster's extent.
        """
        return bbox_overlaps(source, self._bbox)

    def _is_cached(self, target: Path) -> bool:
        """Whether `target` already holds this exact AOI (AOI-aware skip).

        The output filename encodes the return period but not the AOI, so a bare
        exists-check would return a previous AOI's raster for a new bbox in the
        same `path`. The `<target>.aoi` sidecar records the AOI the file was
        written for (`earthlens.base.cache`); the skip only fires when it matches
        and `force` is off.

        Args:
            target: The candidate output GeoTIFF path.

        Returns:
            bool: `True` when a matching cached output exists and may be reused.
        """
        return not getattr(self, "_force", False) and sidecar_is_fresh(
            target, aoi_tag(self.space)
        )

    def download(
        self,
        progress_bar: bool = True,
        *,
        force: bool = False,
    ) -> list[Path]:
        """Fetch the EFHM subset(s) as one AOI-cropped GeoTIFF per return period.

        Args:
            progress_bar: Accepted for signature parity; one read per period.
            force: Re-fetch even when a complete output already exists.

        Returns:
            list[Path]: The written GeoTIFF path(s), one per return period.

        Raises:
            ValueError: If the AOI is outside the EFHM's Europe / Mediterranean
                coverage. (An antimeridian-crossing `west > east` AOI is already
                rejected by `SpatialExtent` at construction.)
        """
        self._force = force
        products = self._search()
        return self._fetch(products)

    def _search(self) -> list[RemoteProduct]:
        """Resolve the request to one `RemoteProduct` per return period.

        No network: each product carries its return period and EFHM URL.

        Returns:
            list[RemoteProduct]: The download plan, one per return period.
        """
        return [
            RemoteProduct(
                id=f"efhm_RP{rp}",
                metadata={
                    "rp": rp,
                    "url": efhm_url(
                        rp,
                        base_url=self._dataset.base_url,
                        template=self._dataset.filename_template,
                    ),
                },
            )
            for rp in self._return_periods
        ]

    def _fetch(self, products: list[RemoteProduct]) -> list[Path]:
        """Windowed-read + crop each return period to one GeoTIFF.

        Args:
            products: The plan from `_search`.

        Returns:
            list[Path]: The written GeoTIFF path(s).

        Raises:
            ValueError: If the AOI is outside the EFHM coverage for a period.
        """
        return [self._fetch_one(product) for product in products]

    def _fetch_one(self, product: RemoteProduct) -> Path:
        """Read the AOI window of one return-period GeoTIFF and write the crop.

        Opens the whole-Europe GeoTIFF lazily and windowed-crops it to the AOI
        with `pyramids.Dataset.crop(bbox=)`, whose fast path reads **only** the
        AOI's pixel window over `/vsicurl` (HTTP range requests, tuned via
        `vsicurl_config()` — readdir-suppression + retry/timeout) for an
        axis-aligned box in the source CRS, carrying the source grid, CRS and
        no-data through (with the
        catalog no-data stamped when the source declares none). A point AOI is
        widened to one pixel so the strict fast path still fires. `crop_to_aoi`
        then trims the all-touched window to the exact bbox — or to the exact
        polygon when the request carried an `aoi=` polygon. An in-coverage AOI
        that is entirely no-data (e.g. open sea) still writes an all-no-data
        raster rather than raising.

        Args:
            product: The `RemoteProduct` whose `metadata` carries `rp` + `url`.

        Returns:
            pathlib.Path: The written GeoTIFF at `<path>/efhm_RP{rp}.tif`.

        Raises:
            ValueError: If the AOI does not overlap the EFHM coverage (an
                in-coverage AOI is written even when it holds no valid data).
        """
        from pyramids.dataset import Dataset as PyramidsDataset

        from earthlens.base import close_quietly

        rp = product.metadata["rp"]
        url = product.metadata["url"]
        target = Path(self.path) / f"{product.id}.tif"
        if self._is_cached(target):
            logger.info(f"JRCFlood: {target.name} already holds this AOI; skipping.")
            return target

        # Tune the /vsicurl read (readdir-suppression + retry/timeout) for the
        # duration of the open + windowed crop; a plain read_file installs none.
        with vsicurl_config():
            source = PyramidsDataset.read_file(url)
            try:
                if not self._bbox_overlaps(source):
                    raise ValueError(
                        f"the AOI {self._bbox} is outside the EFHM's Europe / "
                        f"Mediterranean coverage; no RP{rp} data to write."
                    )
                logger.info(
                    f"JRCFlood RP{rp}: windowed /vsicurl crop of {self._bbox} "
                    f"from {url}"
                )
                # A point / cell-edge AOI (min == max on an axis) is widened to
                # one source pixel so crop(bbox=)'s fast path yields a 1x1 window
                # rather than raising on the zero-width box.
                geo = source.geotransform
                bbox = widen_degenerate_bbox(self._bbox, geo[1], geo[5])
                # The windowed fast path reads only the AOI pixel window from the
                # ~23 GB source; nodata / CRS / grid are carried onto the crop. An
                # in-coverage but all-no-data AOI keeps an all-no-data window
                # rather than raising.
                windowed = windowed_bbox_crop(source, bbox, epsg=4326)
            finally:
                close_quietly(source)

        try:
            # crop carries the source's own no-data through; when the source
            # declares none, fall back to the catalog value so the output stays
            # flagged and a polygon `aoi=` can trim exactly (matching the pre-crop
            # behaviour).
            windowed = ensure_no_data(windowed, self._dataset.nodata)
            # crop(bbox=) keeps every pixel the box overlaps (all-touched, up to
            # one extra pixel per edge); trim to the exact bbox (matching FABDEM)
            # — or to the exact polygon when the request carried an `aoi=` polygon.
            cropped = crop_to_aoi(
                windowed,
                self.space,
                bbox=[
                    self.space.west,
                    self.space.south,
                    self.space.east,
                    self.space.north,
                ],
                touch=False,
            )
            staged = target.with_name(f"{target.stem}.part{target.suffix}")
            try:
                cropped.to_file(str(staged))
                close_quietly(cropped)
                staged.replace(target)
            except BaseException:
                close_quietly(cropped)
                staged.unlink(missing_ok=True)
                raise
        finally:
            close_quietly(windowed)
        write_sidecar(target, aoi_tag(self.space))
        return target

__init__(start='', end='', lat_lim=None, lon_lim=None, return_periods=None, temporal_resolution='static', path=None, fmt='%Y-%m-%d', *, catalog=None)

Initialise a JRC-flood backend instance.

The EFHM has a single water_depth band, so the backend is facet-only (it declares no variables axis); the request axis is return_periods.

Parameters:

Name	Type	Description	Default
start	str	Accepted for facade parity; ignored (the EFHM is static).	''
end	str	Accepted for facade parity; ignored.	''
lat_lim	list[float] | None	[lat_min, lat_max] bounding-box latitudes. Required.	None
lon_lim	list[float] | None	[lon_min, lon_max] bounding-box longitudes. Required.	None
return_periods	list[int | str] | int | str | None	One return period, or a list, in years — as ints (100) or strings ("100" / "RP100"). Defaults to [100]. Every value must be a published return period.	None
temporal_resolution	str	Advisory label only (the EFHM is static).	'static'
path	Path | str | None	Output directory for the written GeoTIFF(s).	None
fmt	str	Accepted for facade parity; unused.	'%Y-%m-%d'
catalog	Catalog | None	Optional pre-built Catalog (tests inject a faked one); defaults to the bundled catalog.	None

Raises:

Type	Description
ValueError	If the bounding box is missing or a requested return period is not published.

Source code in libs/providers/hazards/src/earthlens/jrc_flood/backend.py

def __init__(
    self,
    start: str = "",
    end: str = "",
    lat_lim: list[float] | None = None,
    lon_lim: list[float] | None = None,
    return_periods: list[int | str] | int | str | None = None,
    temporal_resolution: str = "static",
    path: Path | str | None = None,
    fmt: str = "%Y-%m-%d",
    *,
    catalog: Catalog | None = None,
):
    """Initialise a JRC-flood backend instance.

    The EFHM has a single `water_depth` band, so the backend is facet-only
    (it declares no `variables` axis); the request axis is `return_periods`.

    Args:
        start: Accepted for facade parity; ignored (the EFHM is static).
        end: Accepted for facade parity; ignored.
        lat_lim: `[lat_min, lat_max]` bounding-box latitudes. Required.
        lon_lim: `[lon_min, lon_max]` bounding-box longitudes. Required.
        return_periods: One return period, or a list, in years — as ints
            (`100`) or strings (`"100"` / `"RP100"`). Defaults to `[100]`.
            Every value must be a published return period.
        temporal_resolution: Advisory label only (the EFHM is static).
        path: Output directory for the written GeoTIFF(s).
        fmt: Accepted for facade parity; unused.
        catalog: Optional pre-built `Catalog` (tests inject a faked one);
            defaults to the bundled catalog.

    Raises:
        ValueError: If the bounding box is missing or a requested return
            period is not published.
    """
    if lat_lim is None or lon_lim is None:
        raise ValueError(
            "JRCFlood requires a bounding box (lat_lim=[s, n], "
            "lon_lim=[w, e]) — a hazard-map subset has no default extent."
        )

    self._catalog = catalog if catalog is not None else Catalog()
    self._dataset: Dataset = self._catalog.get("efhm")
    self._return_periods = self._resolve_return_periods(return_periods)

    super().__init__(
        start=start,
        end=end,
        variables=[self._dataset.band],
        temporal_resolution=temporal_resolution,
        lat_lim=lat_lim,
        lon_lim=lon_lim,
        fmt=fmt,
        path=path,
    )

download(progress_bar=True, *, force=False)

Fetch the EFHM subset(s) as one AOI-cropped GeoTIFF per return period.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Accepted for signature parity; one read per period.	True
force	bool	Re-fetch even when a complete output already exists.	False

Returns:

Type	Description
list[Path]	list[Path]: The written GeoTIFF path(s), one per return period.

Raises:

Type	Description
ValueError	If the AOI is outside the EFHM's Europe / Mediterranean coverage. (An antimeridian-crossing west > east AOI is already rejected by SpatialExtent at construction.)

Source code in libs/providers/hazards/src/earthlens/jrc_flood/backend.py

def download(
    self,
    progress_bar: bool = True,
    *,
    force: bool = False,
) -> list[Path]:
    """Fetch the EFHM subset(s) as one AOI-cropped GeoTIFF per return period.

    Args:
        progress_bar: Accepted for signature parity; one read per period.
        force: Re-fetch even when a complete output already exists.

    Returns:
        list[Path]: The written GeoTIFF path(s), one per return period.

    Raises:
        ValueError: If the AOI is outside the EFHM's Europe / Mediterranean
            coverage. (An antimeridian-crossing `west > east` AOI is already
            rejected by `SpatialExtent` at construction.)
    """
    self._force = force
    products = self._search()
    return self._fetch(products)

clear_catalog_cache()

Empty the module-level JRC-flood catalog parse cache.

Source code in libs/providers/hazards/src/earthlens/jrc_flood/catalog.py

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

earthlens.jrc_flood.backend

JRC European flood-hazard backend — JRCFlood(AbstractDataSource).

JRCFlood is a download-and-localise raster backend (OUTPUT_KIND="raster") for the JRC European Flood Hazard Map (EFHM): "River flood hazard maps for Europe and the Mediterranean Basin". Each cell value is river-flood water depth (m) for a chosen return period.

A request is a bbox (lat_lim / lon_lim) plus one or more return_periods. The product is static, so start / end are accepted for facade parity and ignored, and the facade-forwarded aggregate= is rejected (return periods are not a reducible time axis). Each return period is one whole-Europe EPSG:4326 GeoTIFF of ~23 GB uncompressed, so the backend never reads it whole: it opens the file lazily over GDAL's /vsicurl (HTTP range requests), reads only the AOI's pixel window through pyramids, and writes one cropped GeoTIFF per return period. An AOI outside the Europe / Mediterranean coverage raises a clear ValueError rather than writing an empty raster.

The product is public and CC-BY-4.0 (permissive), so there is no auth module and no LicenseWarning. The raster read happens through pyramids (a windowed read_array), so this is a genuine pyramids-consuming backend — no xarray.

JRCFlood

Bases: AbstractDataSource

JRC European Flood Hazard Map backend (raster GeoTIFF output).

Fetches the EFHM water-depth grid for one or more return periods, cropped to the request bbox, via lazy /vsicurl windowed reads. The request is a search/fetch split: _search names one product per return period, _fetch realises each (windowed read → crop → GeoTIFF).

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	Fixed "raster"; each return period yields a gridded GeoTIFF. The facade reads it to gate aggregate= (rejected — the return periods are not a temporal axis).

Examples:

• A small AOI writes one cropped GeoTIFF per return period (marked +SKIP — it hits the live JRC directory):

  >>> from earthlens.earthlens import EarthLens
  >>> paths = EarthLens(  # doctest: +SKIP
  ...     data_source="jrc-flood",
  ...     lat_lim=[51.8, 52.0],
  ...     lon_lim=[4.8, 5.0],
  ...     return_periods=[100],
  ...     path="efhm_out",
  ... ).download()  # -> [Path('efhm_out/efhm_RP100.tif')]
  

Source code in libs/providers/hazards/src/earthlens/jrc_flood/backend.py

class JRCFlood(AbstractDataSource):
    """JRC European Flood Hazard Map backend (raster GeoTIFF output).

    Fetches the EFHM water-depth grid for one or more return periods, cropped to
    the request bbox, via lazy `/vsicurl` windowed reads. The request is a
    search/fetch split: `_search` names one product per return period, `_fetch`
    realises each (windowed read → crop → GeoTIFF).

    Attributes:
        OUTPUT_KIND: Fixed `"raster"`; each return period yields a gridded
            GeoTIFF. The facade reads it to gate `aggregate=` (rejected — the
            return periods are not a temporal axis).

    Examples:
        - A small AOI writes one cropped GeoTIFF per return period (marked
          `+SKIP` — it hits the live JRC directory):

            ```python
            >>> from earthlens.earthlens import EarthLens
            >>> paths = EarthLens(  # doctest: +SKIP
            ...     data_source="jrc-flood",
            ...     lat_lim=[51.8, 52.0],
            ...     lon_lim=[4.8, 5.0],
            ...     return_periods=[100],
            ...     path="efhm_out",
            ... ).download()  # -> [Path('efhm_out/efhm_RP100.tif')]

            ```
    """

    OUTPUT_KIND: OutputKind = "raster"

    AGGREGATE_REFUSAL_REASON = "the JRC flood hazard map is a set of static per-return-period depth grids with no temporal axis, so there is nothing to reduce. Call download() without aggregate="

    #: Clips to the exact polygon when `aoi=` carries one, not just its bbox.
    SUPPORTS_POLYGON_AOI = True

    #: The EFHM is time-invariant, so a missing `start` / `end` is legal here.
    REQUIRES_TIME_WINDOW = False

    def __init__(
        self,
        start: str = "",
        end: str = "",
        lat_lim: list[float] | None = None,
        lon_lim: list[float] | None = None,
        return_periods: list[int | str] | int | str | None = None,
        temporal_resolution: str = "static",
        path: Path | str | None = None,
        fmt: str = "%Y-%m-%d",
        *,
        catalog: Catalog | None = None,
    ):
        """Initialise a JRC-flood backend instance.

        The EFHM has a single `water_depth` band, so the backend is facet-only
        (it declares no `variables` axis); the request axis is `return_periods`.

        Args:
            start: Accepted for facade parity; ignored (the EFHM is static).
            end: Accepted for facade parity; ignored.
            lat_lim: `[lat_min, lat_max]` bounding-box latitudes. Required.
            lon_lim: `[lon_min, lon_max]` bounding-box longitudes. Required.
            return_periods: One return period, or a list, in years — as ints
                (`100`) or strings (`"100"` / `"RP100"`). Defaults to `[100]`.
                Every value must be a published return period.
            temporal_resolution: Advisory label only (the EFHM is static).
            path: Output directory for the written GeoTIFF(s).
            fmt: Accepted for facade parity; unused.
            catalog: Optional pre-built `Catalog` (tests inject a faked one);
                defaults to the bundled catalog.

        Raises:
            ValueError: If the bounding box is missing or a requested return
                period is not published.
        """
        if lat_lim is None or lon_lim is None:
            raise ValueError(
                "JRCFlood requires a bounding box (lat_lim=[s, n], "
                "lon_lim=[w, e]) — a hazard-map subset has no default extent."
            )

        self._catalog = catalog if catalog is not None else Catalog()
        self._dataset: Dataset = self._catalog.get("efhm")
        self._return_periods = self._resolve_return_periods(return_periods)

        super().__init__(
            start=start,
            end=end,
            variables=[self._dataset.band],
            temporal_resolution=temporal_resolution,
            lat_lim=lat_lim,
            lon_lim=lon_lim,
            fmt=fmt,
            path=path,
        )

    def _resolve_return_periods(
        self, return_periods: list[int | str] | int | str | None
    ) -> list[int]:
        """Normalise + validate the requested return periods against the catalog.

        Accepts a single value or a list; each may be an int (`100`) or a string
        (`"100"` / `"RP100"`, case-insensitive). Defaults to `[100]`.

        Args:
            return_periods: The raw request value.

        Returns:
            list[int]: Sorted, de-duplicated return periods to fetch.

        Raises:
            ValueError: If a value is unparseable or is not a published return
                period.
        """
        available = self._dataset.return_periods
        if return_periods is None:
            requested_raw: list[int | str] = [100]
        elif isinstance(return_periods, (list, tuple)):
            requested_raw = list(return_periods)
        else:
            requested_raw = [return_periods]

        resolved: list[int] = []
        for value in requested_raw:
            resolved.append(self._parse_rp(value))
        unknown = [rp for rp in resolved if rp not in available]
        if unknown:
            raise ValueError(
                f"return period(s) {unknown} are not published for the EFHM; "
                f"available: {available}."
            )
        return sorted(set(resolved))

    @staticmethod
    def _parse_rp(value: int | str) -> int:
        """Parse one return-period token to an int (`100`, `"100"`, `"RP100"`).

        Args:
            value: The raw return-period value.

        Returns:
            int: The integer return period in years.

        Raises:
            ValueError: If `value` is not an int or an `RP`-prefixed / bare
                integer string.
        """
        if isinstance(value, int):
            return value
        text = str(value).strip().upper()
        if text.startswith("RP"):
            text = text[2:]
        try:
            return int(text)
        except ValueError:
            raise ValueError(
                f"could not parse return period {value!r} (expected e.g. 100, "
                "'100', or 'RP100')."
            ) from None

    def _initialize(self):
        """No-op initialiser — the EFHM is public + anonymous (no client).

        Returns:
            None: The parent binds no `self.client`.
        """
        return None

    def _check_input_dates(
        self, start: str, end: str, temporal_resolution: str, fmt: str
    ) -> TemporalExtent:
        """Return a degenerate (timeless) extent — the EFHM is static.

        Args:
            start: Ignored.
            end: Ignored.
            temporal_resolution: Recorded as the resolution label.
            fmt: Ignored.

        Returns:
            TemporalExtent: A frozen model with `None` bounds and an empty date
                index (a static hazard map has no time axis).
        """
        return self._static_extent(resolution=temporal_resolution or "static")

    @property
    def _bbox(self) -> tuple[float, float, float, float]:
        """The AOI as `(west, south, east, north)` in degrees."""
        return (self.space.west, self.space.south, self.space.east, self.space.north)

    def _bbox_overlaps(self, source: Any) -> bool:
        """Whether the AOI overlaps the source raster's geographic extent.

        Delegates to `earthlens.base.spatial.bbox_overlaps` so an AOI outside the
        EFHM's Europe / Mediterranean coverage is reported with a clear error
        before the windowed crop, rather than surfacing as an empty or opaque
        crop result.

        Args:
            source: An opened `pyramids.Dataset` exposing `geotransform`,
                `columns`, and `rows`.

        Returns:
            bool: `True` when the AOI bbox intersects the raster's extent.
        """
        return bbox_overlaps(source, self._bbox)

    def _is_cached(self, target: Path) -> bool:
        """Whether `target` already holds this exact AOI (AOI-aware skip).

        The output filename encodes the return period but not the AOI, so a bare
        exists-check would return a previous AOI's raster for a new bbox in the
        same `path`. The `<target>.aoi` sidecar records the AOI the file was
        written for (`earthlens.base.cache`); the skip only fires when it matches
        and `force` is off.

        Args:
            target: The candidate output GeoTIFF path.

        Returns:
            bool: `True` when a matching cached output exists and may be reused.
        """
        return not getattr(self, "_force", False) and sidecar_is_fresh(
            target, aoi_tag(self.space)
        )

    def download(
        self,
        progress_bar: bool = True,
        *,
        force: bool = False,
    ) -> list[Path]:
        """Fetch the EFHM subset(s) as one AOI-cropped GeoTIFF per return period.

        Args:
            progress_bar: Accepted for signature parity; one read per period.
            force: Re-fetch even when a complete output already exists.

        Returns:
            list[Path]: The written GeoTIFF path(s), one per return period.

        Raises:
            ValueError: If the AOI is outside the EFHM's Europe / Mediterranean
                coverage. (An antimeridian-crossing `west > east` AOI is already
                rejected by `SpatialExtent` at construction.)
        """
        self._force = force
        products = self._search()
        return self._fetch(products)

    def _search(self) -> list[RemoteProduct]:
        """Resolve the request to one `RemoteProduct` per return period.

        No network: each product carries its return period and EFHM URL.

        Returns:
            list[RemoteProduct]: The download plan, one per return period.
        """
        return [
            RemoteProduct(
                id=f"efhm_RP{rp}",
                metadata={
                    "rp": rp,
                    "url": efhm_url(
                        rp,
                        base_url=self._dataset.base_url,
                        template=self._dataset.filename_template,
                    ),
                },
            )
            for rp in self._return_periods
        ]

    def _fetch(self, products: list[RemoteProduct]) -> list[Path]:
        """Windowed-read + crop each return period to one GeoTIFF.

        Args:
            products: The plan from `_search`.

        Returns:
            list[Path]: The written GeoTIFF path(s).

        Raises:
            ValueError: If the AOI is outside the EFHM coverage for a period.
        """
        return [self._fetch_one(product) for product in products]

    def _fetch_one(self, product: RemoteProduct) -> Path:
        """Read the AOI window of one return-period GeoTIFF and write the crop.

        Opens the whole-Europe GeoTIFF lazily and windowed-crops it to the AOI
        with `pyramids.Dataset.crop(bbox=)`, whose fast path reads **only** the
        AOI's pixel window over `/vsicurl` (HTTP range requests, tuned via
        `vsicurl_config()` — readdir-suppression + retry/timeout) for an
        axis-aligned box in the source CRS, carrying the source grid, CRS and
        no-data through (with the
        catalog no-data stamped when the source declares none). A point AOI is
        widened to one pixel so the strict fast path still fires. `crop_to_aoi`
        then trims the all-touched window to the exact bbox — or to the exact
        polygon when the request carried an `aoi=` polygon. An in-coverage AOI
        that is entirely no-data (e.g. open sea) still writes an all-no-data
        raster rather than raising.

        Args:
            product: The `RemoteProduct` whose `metadata` carries `rp` + `url`.

        Returns:
            pathlib.Path: The written GeoTIFF at `<path>/efhm_RP{rp}.tif`.

        Raises:
            ValueError: If the AOI does not overlap the EFHM coverage (an
                in-coverage AOI is written even when it holds no valid data).
        """
        from pyramids.dataset import Dataset as PyramidsDataset

        from earthlens.base import close_quietly

        rp = product.metadata["rp"]
        url = product.metadata["url"]
        target = Path(self.path) / f"{product.id}.tif"
        if self._is_cached(target):
            logger.info(f"JRCFlood: {target.name} already holds this AOI; skipping.")
            return target

        # Tune the /vsicurl read (readdir-suppression + retry/timeout) for the
        # duration of the open + windowed crop; a plain read_file installs none.
        with vsicurl_config():
            source = PyramidsDataset.read_file(url)
            try:
                if not self._bbox_overlaps(source):
                    raise ValueError(
                        f"the AOI {self._bbox} is outside the EFHM's Europe / "
                        f"Mediterranean coverage; no RP{rp} data to write."
                    )
                logger.info(
                    f"JRCFlood RP{rp}: windowed /vsicurl crop of {self._bbox} "
                    f"from {url}"
                )
                # A point / cell-edge AOI (min == max on an axis) is widened to
                # one source pixel so crop(bbox=)'s fast path yields a 1x1 window
                # rather than raising on the zero-width box.
                geo = source.geotransform
                bbox = widen_degenerate_bbox(self._bbox, geo[1], geo[5])
                # The windowed fast path reads only the AOI pixel window from the
                # ~23 GB source; nodata / CRS / grid are carried onto the crop. An
                # in-coverage but all-no-data AOI keeps an all-no-data window
                # rather than raising.
                windowed = windowed_bbox_crop(source, bbox, epsg=4326)
            finally:
                close_quietly(source)

        try:
            # crop carries the source's own no-data through; when the source
            # declares none, fall back to the catalog value so the output stays
            # flagged and a polygon `aoi=` can trim exactly (matching the pre-crop
            # behaviour).
            windowed = ensure_no_data(windowed, self._dataset.nodata)
            # crop(bbox=) keeps every pixel the box overlaps (all-touched, up to
            # one extra pixel per edge); trim to the exact bbox (matching FABDEM)
            # — or to the exact polygon when the request carried an `aoi=` polygon.
            cropped = crop_to_aoi(
                windowed,
                self.space,
                bbox=[
                    self.space.west,
                    self.space.south,
                    self.space.east,
                    self.space.north,
                ],
                touch=False,
            )
            staged = target.with_name(f"{target.stem}.part{target.suffix}")
            try:
                cropped.to_file(str(staged))
                close_quietly(cropped)
                staged.replace(target)
            except BaseException:
                close_quietly(cropped)
                staged.unlink(missing_ok=True)
                raise
        finally:
            close_quietly(windowed)
        write_sidecar(target, aoi_tag(self.space))
        return target

__init__(start='', end='', lat_lim=None, lon_lim=None, return_periods=None, temporal_resolution='static', path=None, fmt='%Y-%m-%d', *, catalog=None)

Initialise a JRC-flood backend instance.

The EFHM has a single water_depth band, so the backend is facet-only (it declares no variables axis); the request axis is return_periods.

Parameters:

Name	Type	Description	Default
start	str	Accepted for facade parity; ignored (the EFHM is static).	''
end	str	Accepted for facade parity; ignored.	''
lat_lim	list[float] | None	[lat_min, lat_max] bounding-box latitudes. Required.	None
lon_lim	list[float] | None	[lon_min, lon_max] bounding-box longitudes. Required.	None
return_periods	list[int | str] | int | str | None	One return period, or a list, in years — as ints (100) or strings ("100" / "RP100"). Defaults to [100]. Every value must be a published return period.	None
temporal_resolution	str	Advisory label only (the EFHM is static).	'static'
path	Path | str | None	Output directory for the written GeoTIFF(s).	None
fmt	str	Accepted for facade parity; unused.	'%Y-%m-%d'
catalog	Catalog | None	Optional pre-built Catalog (tests inject a faked one); defaults to the bundled catalog.	None

Raises:

Type	Description
ValueError	If the bounding box is missing or a requested return period is not published.

Source code in libs/providers/hazards/src/earthlens/jrc_flood/backend.py

def __init__(
    self,
    start: str = "",
    end: str = "",
    lat_lim: list[float] | None = None,
    lon_lim: list[float] | None = None,
    return_periods: list[int | str] | int | str | None = None,
    temporal_resolution: str = "static",
    path: Path | str | None = None,
    fmt: str = "%Y-%m-%d",
    *,
    catalog: Catalog | None = None,
):
    """Initialise a JRC-flood backend instance.

    The EFHM has a single `water_depth` band, so the backend is facet-only
    (it declares no `variables` axis); the request axis is `return_periods`.

    Args:
        start: Accepted for facade parity; ignored (the EFHM is static).
        end: Accepted for facade parity; ignored.
        lat_lim: `[lat_min, lat_max]` bounding-box latitudes. Required.
        lon_lim: `[lon_min, lon_max]` bounding-box longitudes. Required.
        return_periods: One return period, or a list, in years — as ints
            (`100`) or strings (`"100"` / `"RP100"`). Defaults to `[100]`.
            Every value must be a published return period.
        temporal_resolution: Advisory label only (the EFHM is static).
        path: Output directory for the written GeoTIFF(s).
        fmt: Accepted for facade parity; unused.
        catalog: Optional pre-built `Catalog` (tests inject a faked one);
            defaults to the bundled catalog.

    Raises:
        ValueError: If the bounding box is missing or a requested return
            period is not published.
    """
    if lat_lim is None or lon_lim is None:
        raise ValueError(
            "JRCFlood requires a bounding box (lat_lim=[s, n], "
            "lon_lim=[w, e]) — a hazard-map subset has no default extent."
        )

    self._catalog = catalog if catalog is not None else Catalog()
    self._dataset: Dataset = self._catalog.get("efhm")
    self._return_periods = self._resolve_return_periods(return_periods)

    super().__init__(
        start=start,
        end=end,
        variables=[self._dataset.band],
        temporal_resolution=temporal_resolution,
        lat_lim=lat_lim,
        lon_lim=lon_lim,
        fmt=fmt,
        path=path,
    )

download(progress_bar=True, *, force=False)

Fetch the EFHM subset(s) as one AOI-cropped GeoTIFF per return period.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Accepted for signature parity; one read per period.	True
force	bool	Re-fetch even when a complete output already exists.	False

Returns:

Type	Description
list[Path]	list[Path]: The written GeoTIFF path(s), one per return period.

Raises:

Type	Description
ValueError	If the AOI is outside the EFHM's Europe / Mediterranean coverage. (An antimeridian-crossing west > east AOI is already rejected by SpatialExtent at construction.)

Source code in libs/providers/hazards/src/earthlens/jrc_flood/backend.py

def download(
    self,
    progress_bar: bool = True,
    *,
    force: bool = False,
) -> list[Path]:
    """Fetch the EFHM subset(s) as one AOI-cropped GeoTIFF per return period.

    Args:
        progress_bar: Accepted for signature parity; one read per period.
        force: Re-fetch even when a complete output already exists.

    Returns:
        list[Path]: The written GeoTIFF path(s), one per return period.

    Raises:
        ValueError: If the AOI is outside the EFHM's Europe / Mediterranean
            coverage. (An antimeridian-crossing `west > east` AOI is already
            rejected by `SpatialExtent` at construction.)
    """
    self._force = force
    products = self._search()
    return self._fetch(products)

earthlens.jrc_flood.catalog

Catalog loader for the JRC European flood-hazard (EFHM) backend.

EFHM is a single product served as one whole-Europe GeoTIFF per return period, so the catalog is one jrc_flood_data_catalog.yaml at the package root holding the dataset row (band, CRS, the return-period list, the URL template) plus the CC-BY-4.0 licence / attribution. It loads through the shared strict YAML loader and the (path, mtime) parse cache, and exposes the row via the inherited AbstractCatalog surface (cat["efhm"], get_dataset, the did-you-mean error).

CATALOG_PATH is the bundled YAML; clear_catalog_cache empties the parse cache (used by tests that monkey-patch CATALOG_PATH).

Catalog

Bases: AbstractCatalog

Product catalog for the JRC European flood-hazard backend.

Reads the bundled jrc_flood_data_catalog.yaml and exposes its single row under the inherited datasets field — which supplies the cat["efhm"] / "efhm" in cat / len(cat) surface and the did-you-mean error for free. Instantiate with no arguments; the base model_post_init auto-loads via _autoload, cached by (path, mtime).

Attributes:

Name	Type	Description
datasets	dict[str, Dataset]	Product key to its Dataset row.
available_datasets	list[str]	Sorted product keys.
license_id	str	SPDX-ish licence label ("CC-BY-4.0", permissive).
attribution	str	The citation the licence requires.

Examples:

• List products and read the licence:

  >>> from earthlens.jrc_flood import Catalog
  >>> cat = Catalog()
  >>> list(cat.datasets)
  ['efhm']
  >>> cat.license_id
  'CC-BY-4.0'
  

Source code in libs/providers/hazards/src/earthlens/jrc_flood/catalog.py

class Catalog(AbstractCatalog):
    """Product catalog for the JRC European flood-hazard backend.

    Reads the bundled `jrc_flood_data_catalog.yaml` and exposes its single row
    under the inherited `datasets` field — which supplies the `cat["efhm"]` /
    `"efhm" in cat` / `len(cat)` surface and the did-you-mean error for free.
    Instantiate with no arguments; the base `model_post_init` auto-loads via
    `_autoload`, cached by `(path, mtime)`.

    Attributes:
        datasets: Product key to its `Dataset` row.
        available_datasets: Sorted product keys.
        license_id: SPDX-ish licence label (`"CC-BY-4.0"`, permissive).
        attribution: The citation the licence requires.

    Examples:
        - List products and read the licence:
            ```python
            >>> from earthlens.jrc_flood import Catalog
            >>> cat = Catalog()
            >>> list(cat.datasets)
            ['efhm']
            >>> cat.license_id
            'CC-BY-4.0'

            ```
    """

    _catalog_kind: str = "JRC-flood catalog"

    datasets: dict[str, Dataset] = Field(default_factory=dict)
    license_id: str = ""
    attribution: str = ""

    @classmethod
    def _autoload(cls) -> dict[str, Any]:
        """Return the disk payload to fill an empty catalog (base post-init hook).

        Returns:
            dict[str, Any]: The parsed field → value map from `_parse_catalog`.
        """
        return load_catalog(
            CATALOG_PATH, _CATALOG_CACHE, _parse_catalog, provider="JRC-flood"
        )

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

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

        Returns:
            Catalog: A fully-populated catalog.

        Raises:
            ValueError: If `catalog_path` does not exist, has no `datasets:`
                block, or the row fails validation.
        """
        path = catalog_path if catalog_path is not None else CATALOG_PATH
        payload = load_catalog(
            path, _CATALOG_CACHE, _parse_catalog, provider="JRC-flood"
        )
        return cls(**payload)

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

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

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

        Args:
            key: A product key (`"efhm"`).

        Returns:
            Dataset: The matching product row.

        Raises:
            ValueError: If `key` is not a registered product.
        """
        return cast("Dataset", self.get_dataset(key))

get(key)

Return the Dataset for key, with a did-you-mean hint.

Parameters:

Name	Type	Description	Default
key	str	A product key ("efhm").	required

Returns:

Name	Type	Description
Dataset	Dataset	The matching product row.

Raises:

Type	Description
ValueError	If key is not a registered product.

Source code in libs/providers/hazards/src/earthlens/jrc_flood/catalog.py

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

    Args:
        key: A product key (`"efhm"`).

    Returns:
        Dataset: The matching product row.

    Raises:
        ValueError: If `key` is not a registered product.
    """
    return cast("Dataset", self.get_dataset(key))

get_catalog()

Return the product map (satisfies the abstract contract).

Returns:

Type	Description
dict[str, Dataset]	dict[str, Dataset]: Same object as datasets.

Source code in libs/providers/hazards/src/earthlens/jrc_flood/catalog.py

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

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

load(catalog_path=None) classmethod

Read and validate the JRC-flood catalog from disk (cached).

Parameters:

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

Returns:

Name	Type	Description
Catalog	Catalog	A fully-populated catalog.

Raises:

Type	Description
ValueError	If catalog_path does not exist, has no datasets: block, or the row fails validation.

Source code in libs/providers/hazards/src/earthlens/jrc_flood/catalog.py

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

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

    Returns:
        Catalog: A fully-populated catalog.

    Raises:
        ValueError: If `catalog_path` does not exist, has no `datasets:`
            block, or the row fails validation.
    """
    path = catalog_path if catalog_path is not None else CATALOG_PATH
    payload = load_catalog(
        path, _CATALOG_CACHE, _parse_catalog, provider="JRC-flood"
    )
    return cls(**payload)

Dataset

Bases: BaseModel

One EFHM product row.

Attributes:

Name	Type	Description
id	str	The catalog key ("efhm").
title	str	Human-readable product title.
band	str	The single water-depth band name ("water_depth").
long_name	str	Human-readable band description.
units	str	Physical units of the band ("m").
dtype	str	Pixel data type ("float32").
crs	str	Native CRS as an EPSG string ("EPSG:4326").
nodata	float	The raster no-data value.
spatial_resolution	float | None	Nominal resolution in metres (90).
base_url	str	The JRC directory root the return-period files live in.
filename_template	str	The per-return-period file-name template ("Europe_RP{rp}_filled_depth.tif").
return_periods	list[int]	The published return periods in years.
source_url	str	The dataset landing page.

Examples:

• Read the return periods and band:

  >>> from earthlens.jrc_flood import Catalog
  >>> row = Catalog().get("efhm")
  >>> row.band, row.return_periods[:3]
  ('water_depth', [10, 20, 30])
  

Source code in libs/providers/hazards/src/earthlens/jrc_flood/catalog.py

class Dataset(BaseModel):
    """One EFHM product row.

    Attributes:
        id: The catalog key (`"efhm"`).
        title: Human-readable product title.
        band: The single water-depth band name (`"water_depth"`).
        long_name: Human-readable band description.
        units: Physical units of the band (`"m"`).
        dtype: Pixel data type (`"float32"`).
        crs: Native CRS as an EPSG string (`"EPSG:4326"`).
        nodata: The raster no-data value.
        spatial_resolution: Nominal resolution in metres (`90`).
        base_url: The JRC directory root the return-period files live in.
        filename_template: The per-return-period file-name template
            (`"Europe_RP{rp}_filled_depth.tif"`).
        return_periods: The published return periods in years.
        source_url: The dataset landing page.

    Examples:
        - Read the return periods and band:
            ```python
            >>> from earthlens.jrc_flood import Catalog
            >>> row = Catalog().get("efhm")
            >>> row.band, row.return_periods[:3]
            ('water_depth', [10, 20, 30])

            ```
    """

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

    id: str
    title: str = ""
    band: str = "water_depth"
    long_name: str = ""
    units: str = "m"
    dtype: str = "float32"
    crs: str = "EPSG:4326"
    nodata: float = -9999.0
    spatial_resolution: float | None = None
    base_url: str = ""
    filename_template: str = "Europe_RP{rp}_filled_depth.tif"
    return_periods: list[int] = Field(default_factory=list)
    source_url: str = ""

clear_catalog_cache()

Empty the module-level JRC-flood catalog parse cache.

Source code in libs/providers/hazards/src/earthlens/jrc_flood/catalog.py

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

earthlens.jrc_flood._helpers

URL builder for the JRC European flood-hazard backend.

The JRC serves the European Flood Hazard Map (EFHM) over a deterministic, anonymous HTTPS directory (verified live 2026-08-09): one whole-Europe GeoTIFF per return period at {BASE_URL}/Europe_RP{rp}_filled_depth.tif. Each file is a single-band EPSG:4326 Float32 grid at ~0.000833 deg (~90 m; documented 100 m), covering Europe and the Mediterranean Basin — 110162x51992 px (~23 GB uncompressed), so it is never read whole. The backend opens it lazily over pyramids.dataset.Dataset.crop(bbox=), whose windowed fast path reads only the AOI's pixel window over GDAL's /vsicurl (HTTP range requests) for an axis-aligned box in the source CRS — the case here (an EPSG:4326 AOI against the 4326 EFHM). This module only builds the per-return-period URL; the windowed read + crop live in the backend, which wraps them in vsicurl_config() for the /vsicurl readdir-suppression + retry/timeout tuning.

efhm_url(rp, *, base_url=BASE_URL, template=FILENAME_TEMPLATE)

Build the EFHM GeoTIFF URL for one return period.

Parameters:

Name	Type	Description	Default
rp	int	The integer return period in years (e.g. 100).	required
base_url	str	The directory root; defaults to BASE_URL.	BASE_URL
template	str	The file-name template; defaults to FILENAME_TEMPLATE.	FILENAME_TEMPLATE

Returns:

Name	Type	Description
str	str	The fully-qualified .tif URL.

Examples:

• The verified RP100 URL:

  >>> from earthlens.jrc_flood._helpers import efhm_url
  >>> efhm_url(100)
  'https://jeodpp.jrc.ec.europa.eu/ftp/jrc-opendata/CEMS-EFAS/flood_hazard/Europe_RP100_filled_depth.tif'
  

Source code in libs/providers/hazards/src/earthlens/jrc_flood/_helpers.py

def efhm_url(
    rp: int, *, base_url: str = BASE_URL, template: str = FILENAME_TEMPLATE
) -> str:
    """Build the EFHM GeoTIFF URL for one return period.

    Args:
        rp: The integer return period in years (e.g. `100`).
        base_url: The directory root; defaults to `BASE_URL`.
        template: The file-name template; defaults to `FILENAME_TEMPLATE`.

    Returns:
        str: The fully-qualified `.tif` URL.

    Examples:
        - The verified RP100 URL:
            ```python
            >>> from earthlens.jrc_flood._helpers import efhm_url
            >>> efhm_url(100)
            'https://jeodpp.jrc.ec.europa.eu/ftp/jrc-opendata/CEMS-EFAS/flood_hazard/Europe_RP100_filled_depth.tif'

            ```
    """
    return f"{base_url}/{template.format(rp=rp)}"