HANZE — API reference

HANZE historical European flood-impacts data source subpackage — earthlens.hanze. Background and usage are covered under the other pages in this section (Introduction, Usage); this page is the rendered API. HANZE needs no credentials — the Zenodo record is public (CC-BY-4.0).

earthlens.hanze

HANZE historical-flood-impacts backend.

Fetches the HANZE (Historical Analysis of Natural Hazards in Europe) database of observed European flood events and their impacts (Paprotny et al.) from its pinned static Zenodo release, and returns the event / impact records as a :class:pandas.DataFrame. It is the observed hazard -> loss record: real floods with fatalities, persons affected, area flooded and economic losses, against which a modelled event set can be validated. Companion to the global emdat backend.

This is a tabular backend by default: the result is a table of event / impact rows, not a gridded array, so the :class:earthlens.earthlens.EarthLens facade rejects an aggregate= argument. Passing with_geometry=True instead returns a pyramids :class:~pyramids.feature.collection.FeatureCollection of the affected NUTS-3 regions (a per-instance vector output).

HANZE needs no credentials — the Zenodo record is public (CC-BY-4.0) — so there is no auth class and no [hanze] extra: the only dependencies (HttpClient, pandas, base/archive, pyramids) are all core.

Public surface (re-exported from this package):

• :class:HANZE — the backend; instantiate with a date range and optional country= / region= / flood_type= filters, then call :meth:HANZE.download.
• :class:Catalog — loader for the bundled hanze_data_catalog.yaml.
• :class:ZenodoRecord / :class:HanzeFile / :class:FloodType / :class:GeometryJoin — the catalog's frozen row models.
• :func:join_events_to_regions / :func:empty_region_fc — the event -> region-geometry join and its empty-result counterpart.
• :data:CATALOG_PATH — path to the bundled catalog YAML; monkey-patchable in tests.

Examples:

• List the flood-type vocabulary:

  >>> from earthlens.hanze import Catalog
  >>> Catalog().flood_types()
  ['Coastal', 'Flash', 'River', 'River/Coastal']
  

Catalog

Bases: AbstractCatalog

Catalog for the HANZE backend.

Reads the bundled hanze_data_catalog.yaml (shipped as package data) and exposes the pinned Zenodo record, the per-file names, the flood-Type vocabulary (as :class:FloodType rows keyed by type under the inherited :attr:datasets field — the cat["River"] / "River" in cat / len(cat) dict surface), the friendly-name -> CSV-header map, and the region-geometry join configuration. Instantiate with no arguments (Catalog()).

Attributes:

Name	Type	Description
datasets	dict[str, FloodType]	Map from a flood-Type string to its :class:FloodType row.
record	ZenodoRecord | None	The pinned :class:ZenodoRecord.
files	dict[str, HanzeFile]	Map from a logical key ("events", "regions", "region_names") to its :class:HanzeFile.
geometry	GeometryJoin | None	The :class:GeometryJoin for the region attach.
columns	dict[str, str]	Friendly name -> exact HANZE CSV header.

Examples:

• List the flood types and resolve one, and read the pinned record:

  >>> from earthlens.hanze import Catalog
  >>> cat = Catalog()
  >>> cat.flood_types()
  ['Coastal', 'Flash', 'River', 'River/Coastal']
  >>> cat.get_flood_type("River").description
  'Riverine (fluvial) floods.'
  >>> "Coastal" in cat
  True
  >>> cat.column("country_code")
  'Country code'
  

• An unknown flood type raises with a did-you-mean hint:

  >>> from earthlens.hanze import Catalog
  >>> Catalog().get_flood_type("Rivers")
  Traceback (most recent call last):
      ...
  ValueError: 'Rivers' is not in the HANZE catalog. Known flood types: ['Coastal', 'Flash', 'River', 'River/Coastal']. Did you mean 'River'?
  

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

class Catalog(AbstractCatalog):
    """Catalog for the HANZE backend.

    Reads the bundled `hanze_data_catalog.yaml` (shipped as package data) and
    exposes the pinned Zenodo record, the per-file names, the flood-`Type`
    vocabulary (as :class:`FloodType` rows keyed by type under the inherited
    :attr:`datasets` field — the `cat["River"]` / `"River" in cat` / `len(cat)`
    dict surface), the friendly-name -> CSV-header map, and the region-geometry
    join configuration. Instantiate with no arguments (`Catalog()`).

    Attributes:
        datasets: Map from a flood-`Type` string to its :class:`FloodType` row.
        record: The pinned :class:`ZenodoRecord`.
        files: Map from a logical key (`"events"`, `"regions"`,
            `"region_names"`) to its :class:`HanzeFile`.
        geometry: The :class:`GeometryJoin` for the region attach.
        columns: Friendly name -> exact HANZE CSV header.

    Examples:
        - List the flood types and resolve one, and read the pinned record:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> cat = Catalog()
            >>> cat.flood_types()
            ['Coastal', 'Flash', 'River', 'River/Coastal']
            >>> cat.get_flood_type("River").description
            'Riverine (fluvial) floods.'
            >>> "Coastal" in cat
            True
            >>> cat.column("country_code")
            'Country code'

            ```
        - An unknown flood type raises with a did-you-mean hint:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> Catalog().get_flood_type("Rivers")
            Traceback (most recent call last):
                ...
            ValueError: 'Rivers' is not in the HANZE catalog. Known flood types: ['Coastal', 'Flash', 'River', 'River/Coastal']. Did you mean 'River'?

            ```
    """

    _catalog_kind: str = "HANZE catalog"
    _entry_noun: str = "flood types"

    datasets: dict[str, FloodType] = Field(default_factory=dict)
    #: `record` / `geometry` default to `None` rather than a placeholder model:
    #: the base :meth:`model_post_init` autoload fills a field only when its
    #: current value is falsy, and a placeholder model instance is truthy, so it
    #: would be skipped and `Catalog()` would keep the placeholder. `None` is
    #: falsy, so the bundled record / geometry load as intended. `load()` and a
    #: test passing `datasets=` + these fields populate them directly.
    record: ZenodoRecord | None = Field(default=None, repr=False)
    files: dict[str, HanzeFile] = Field(default_factory=dict, repr=False)
    geometry: GeometryJoin | None = Field(default=None, repr=False)
    columns: dict[str, str] = Field(default_factory=dict, repr=False)

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

        Returns:
            dict[str, Any]: The full field payload read from the bundled catalog.
        """
        return dict(_parse_catalog([CATALOG_PATH]))

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

        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 `catalog_path` does not exist, a required block is
                missing, or a row fails validation.

        Examples:
            - Loading the bundled catalog yields the pinned record and types:
                ```python
                >>> from earthlens.hanze import Catalog
                >>> cat = Catalog.load()
                >>> cat.record.record
                20478847
                >>> cat.flood_types()
                ['Coastal', 'Flash', 'River', 'River/Coastal']

                ```
        """
        catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
        parsed = load_catalog(
            catalog_path, _CATALOG_CACHE, _parse_catalog, provider="HANZE"
        )
        return cls(**parsed)

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

        Returns:
            dict[str, FloodType]: Same object as :attr:`datasets`.

        Examples:
            - The flood-type map is keyed by the `Type` string:
                ```python
                >>> from earthlens.hanze import Catalog
                >>> sorted(Catalog().get_catalog())
                ['Coastal', 'Flash', 'River', 'River/Coastal']

                ```
        """
        return self.datasets

    def get_flood_type(self, flood_type: str) -> FloodType:
        """Return the :class:`FloodType` for `flood_type`, with a did-you-mean hint.

        Thin alias over :meth:`~earthlens.base.AbstractCatalog.get_dataset`.

        Args:
            flood_type: A HANZE flood-`Type` string (`"River"`, `"Coastal"`,
                `"Flash"`, `"River/Coastal"`).

        Returns:
            FloodType: The matching row.

        Raises:
            ValueError: If `flood_type` is not a registered flood type.

        Examples:
            - Resolve a type and read its description:
                ```python
                >>> from earthlens.hanze import Catalog
                >>> Catalog().get_flood_type("Coastal").description
                'Coastal (storm-surge) floods.'

                ```
        """
        return cast("FloodType", self.get_dataset(flood_type))

    def flood_types(self) -> list[str]:
        """Return the registered flood-`Type` strings, sorted.

        Returns:
            list[str]: The flood types
                (`["Coastal", "Flash", "River", "River/Coastal"]`).

        Examples:
            - The registered types come back sorted:
                ```python
                >>> from earthlens.hanze import Catalog
                >>> Catalog().flood_types()
                ['Coastal', 'Flash', 'River', 'River/Coastal']

                ```
        """
        return sorted(self.datasets)

    def file(self, key: str) -> HanzeFile:
        """Return the :class:`HanzeFile` for a logical key.

        Args:
            key: `"events"`, `"regions"`, or `"region_names"`.

        Returns:
            HanzeFile: The matching file descriptor.

        Raises:
            KeyError: If `key` is not a known file.

        Examples:
            - Resolve the events and region file names:
                ```python
                >>> from earthlens.hanze import Catalog
                >>> cat = Catalog()
                >>> cat.file("events").name
                'HANZE_events_v3_0_1b.csv'
                >>> cat.file("regions").name
                'Regions_v2024_simplified.zip'

                ```
        """
        return self.files[key]

    def column(self, friendly: str) -> str:
        """Return the exact HANZE CSV header for a friendly column name.

        Args:
            friendly: A friendly key from the catalog's `columns:` map
                (`"country_code"`, `"type"`, `"regions_nuts3"`, ...).

        Returns:
            str: The exact CSV header (`"Country code"`).

        Raises:
            KeyError: If `friendly` is not a mapped column.

        Examples:
            - Map friendly keys to their exact HANZE headers:
                ```python
                >>> from earthlens.hanze import Catalog
                >>> cat = Catalog()
                >>> cat.column("country_code")
                'Country code'
                >>> cat.column("regions_nuts3")
                'Regions affected (NUTS 3)'

                ```
        """
        return self.columns[friendly]

column(friendly)

Return the exact HANZE CSV header for a friendly column name.

Parameters:

Name	Type	Description	Default
friendly	str	A friendly key from the catalog's columns: map ("country_code", "type", "regions_nuts3", ...).	required

Returns:

Name	Type	Description
str	str	The exact CSV header ("Country code").

Raises:

Type	Description
KeyError	If friendly is not a mapped column.

Examples:

• Map friendly keys to their exact HANZE headers:

  >>> from earthlens.hanze import Catalog
  >>> cat = Catalog()
  >>> cat.column("country_code")
  'Country code'
  >>> cat.column("regions_nuts3")
  'Regions affected (NUTS 3)'
  

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

def column(self, friendly: str) -> str:
    """Return the exact HANZE CSV header for a friendly column name.

    Args:
        friendly: A friendly key from the catalog's `columns:` map
            (`"country_code"`, `"type"`, `"regions_nuts3"`, ...).

    Returns:
        str: The exact CSV header (`"Country code"`).

    Raises:
        KeyError: If `friendly` is not a mapped column.

    Examples:
        - Map friendly keys to their exact HANZE headers:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> cat = Catalog()
            >>> cat.column("country_code")
            'Country code'
            >>> cat.column("regions_nuts3")
            'Regions affected (NUTS 3)'

            ```
    """
    return self.columns[friendly]

file(key)

Return the :class:HanzeFile for a logical key.

Parameters:

Name	Type	Description	Default
key	str	"events", "regions", or "region_names".	required

Returns:

Name	Type	Description
HanzeFile	HanzeFile	The matching file descriptor.

Raises:

Type	Description
KeyError	If key is not a known file.

Examples:

• Resolve the events and region file names:

  >>> from earthlens.hanze import Catalog
  >>> cat = Catalog()
  >>> cat.file("events").name
  'HANZE_events_v3_0_1b.csv'
  >>> cat.file("regions").name
  'Regions_v2024_simplified.zip'
  

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

def file(self, key: str) -> HanzeFile:
    """Return the :class:`HanzeFile` for a logical key.

    Args:
        key: `"events"`, `"regions"`, or `"region_names"`.

    Returns:
        HanzeFile: The matching file descriptor.

    Raises:
        KeyError: If `key` is not a known file.

    Examples:
        - Resolve the events and region file names:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> cat = Catalog()
            >>> cat.file("events").name
            'HANZE_events_v3_0_1b.csv'
            >>> cat.file("regions").name
            'Regions_v2024_simplified.zip'

            ```
    """
    return self.files[key]

flood_types()

Return the registered flood-Type strings, sorted.

Returns:

Type	Description
list[str]	list[str]: The flood types (["Coastal", "Flash", "River", "River/Coastal"]).

Examples:

• The registered types come back sorted:

  >>> from earthlens.hanze import Catalog
  >>> Catalog().flood_types()
  ['Coastal', 'Flash', 'River', 'River/Coastal']
  

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

def flood_types(self) -> list[str]:
    """Return the registered flood-`Type` strings, sorted.

    Returns:
        list[str]: The flood types
            (`["Coastal", "Flash", "River", "River/Coastal"]`).

    Examples:
        - The registered types come back sorted:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> Catalog().flood_types()
            ['Coastal', 'Flash', 'River', 'River/Coastal']

            ```
    """
    return sorted(self.datasets)

get_catalog()

Return the flood-type map (satisfies the abstract contract).

Returns:

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

Examples:

• The flood-type map is keyed by the Type string:

  >>> from earthlens.hanze import Catalog
  >>> sorted(Catalog().get_catalog())
  ['Coastal', 'Flash', 'River', 'River/Coastal']
  

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

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

    Returns:
        dict[str, FloodType]: Same object as :attr:`datasets`.

    Examples:
        - The flood-type map is keyed by the `Type` string:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> sorted(Catalog().get_catalog())
            ['Coastal', 'Flash', 'River', 'River/Coastal']

            ```
    """
    return self.datasets

get_flood_type(flood_type)

Return the :class:FloodType for flood_type, with a did-you-mean hint.

Thin alias over :meth:~earthlens.base.AbstractCatalog.get_dataset.

Parameters:

Name	Type	Description	Default
flood_type	str	A HANZE flood-Type string ("River", "Coastal", "Flash", "River/Coastal").	required

Returns:

Name	Type	Description
FloodType	FloodType	The matching row.

Raises:

Type	Description
ValueError	If flood_type is not a registered flood type.

Examples:

• Resolve a type and read its description:

  >>> from earthlens.hanze import Catalog
  >>> Catalog().get_flood_type("Coastal").description
  'Coastal (storm-surge) floods.'
  

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

def get_flood_type(self, flood_type: str) -> FloodType:
    """Return the :class:`FloodType` for `flood_type`, with a did-you-mean hint.

    Thin alias over :meth:`~earthlens.base.AbstractCatalog.get_dataset`.

    Args:
        flood_type: A HANZE flood-`Type` string (`"River"`, `"Coastal"`,
            `"Flash"`, `"River/Coastal"`).

    Returns:
        FloodType: The matching row.

    Raises:
        ValueError: If `flood_type` is not a registered flood type.

    Examples:
        - Resolve a type and read its description:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> Catalog().get_flood_type("Coastal").description
            'Coastal (storm-surge) floods.'

            ```
    """
    return cast("FloodType", self.get_dataset(flood_type))

load(catalog_path=None) classmethod

Read the HANZE catalog from disk.

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 catalog_path does not exist, a required block is missing, or a row fails validation.

Examples:

• Loading the bundled catalog yields the pinned record and types:

  >>> from earthlens.hanze import Catalog
  >>> cat = Catalog.load()
  >>> cat.record.record
  20478847
  >>> cat.flood_types()
  ['Coastal', 'Flash', 'River', 'River/Coastal']
  

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

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

    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 `catalog_path` does not exist, a required block is
            missing, or a row fails validation.

    Examples:
        - Loading the bundled catalog yields the pinned record and types:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> cat = Catalog.load()
            >>> cat.record.record
            20478847
            >>> cat.flood_types()
            ['Coastal', 'Flash', 'River', 'River/Coastal']

            ```
    """
    catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
    parsed = load_catalog(
        catalog_path, _CATALOG_CACHE, _parse_catalog, provider="HANZE"
    )
    return cls(**parsed)

FloodType

Bases: BaseModel

One entry of the HANZE flood-Type vocabulary.

The type string ("River", "River/Coastal") is the parent key in :attr:Catalog.datasets and is not stored on the row.

Attributes:

Name	Type	Description
description	str	Short note on what the flood type covers.

Examples:

• Build a row directly:

  >>> from earthlens.hanze import FloodType
  >>> FloodType(description="Riverine (fluvial) floods.").description
  'Riverine (fluvial) floods.'
  

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

class FloodType(BaseModel):
    """One entry of the HANZE flood-`Type` vocabulary.

    The type string (`"River"`, `"River/Coastal"`) is the parent key in
    :attr:`Catalog.datasets` and is not stored on the row.

    Attributes:
        description: Short note on what the flood type covers.

    Examples:
        - Build a row directly:
            ```python
            >>> from earthlens.hanze import FloodType
            >>> FloodType(description="Riverine (fluvial) floods.").description
            'Riverine (fluvial) floods.'

            ```
    """

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

    description: str = ""

GeometryJoin

Bases: BaseModel

The region-shapefile join configuration for with_geometry.

Attributes:

Name	Type	Description
member_stem	str	The shapefile member stem inside the region zip ("NUTS3_regions_v2024_simplified"); the .shp and its sidecars share it.
join_field	str	The shapefile attribute holding the NUTS-3 code ("Code"), joined to the semicolon-split Regions affected (NUTS 3) list.
name_field	str	The shapefile attribute holding the region name ("Name").
crs	str	The shapefile's stored CRS ("EPSG:3035", ETRS89-LAEA Europe). The backend reprojects to WGS84 for a degree bbox filter and for parity with the other vector backends.

Examples:

• The join field and CRS are what the geometry attach reads:

  >>> from earthlens.hanze import Catalog
  >>> geometry = Catalog().geometry
  >>> geometry.join_field, geometry.crs
  ('Code', 'EPSG:3035')
  

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

class GeometryJoin(BaseModel):
    """The region-shapefile join configuration for `with_geometry`.

    Attributes:
        member_stem: The shapefile member stem inside the region zip
            (`"NUTS3_regions_v2024_simplified"`); the `.shp` and its sidecars
            share it.
        join_field: The shapefile attribute holding the NUTS-3 code (`"Code"`),
            joined to the semicolon-split `Regions affected (NUTS 3)` list.
        name_field: The shapefile attribute holding the region name (`"Name"`).
        crs: The shapefile's stored CRS (`"EPSG:3035"`, ETRS89-LAEA Europe). The
            backend reprojects to WGS84 for a degree bbox filter and for parity
            with the other vector backends.

    Examples:
        - The join field and CRS are what the geometry attach reads:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> geometry = Catalog().geometry
            >>> geometry.join_field, geometry.crs
            ('Code', 'EPSG:3035')

            ```
    """

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

    member_stem: str
    join_field: str = "Code"
    name_field: str = "Name"
    crs: str = "EPSG:3035"

HANZE

Bases: AbstractDataSource

HANZE historical-flood-impacts backend (per-instance output kind).

Downloads the HANZE events / impacts table from its pinned Zenodo release, filters it by country / region / flood type / date window, and returns a :class:pandas.DataFrame. With with_geometry=True it instead returns a :class:~pyramids.feature.collection.FeatureCollection of the affected NUTS-3 regions.

The record is public (CC-BY-4.0); no credentials are needed.

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	Set per instance in :meth:__init__ — "tabular" by default, "vector" when with_geometry=True. The facade reads it to know the return shape and rejects aggregate= for both.
REQUIRES_TIME_WINDOW		False — a request without a window returns every year the record covers.

Examples:

• Pull DE + NL flood events, or the affected-region geometry, through the facade (both fetch from Zenodo, so this is illustrative, not a doctest):

  from earthlens.core import EarthLens
  
  events = EarthLens(
      "hanze", start="1950", end="2020", country=["DE", "NL"]
  ).download()  # a pandas.DataFrame of events + impacts
  
  regions = EarthLens(
      "hanze", start="1990", end="2020", country="DE", with_geometry=True
  ).download()  # a FeatureCollection of the affected NUTS-3 regions
  

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

class HANZE(AbstractDataSource):
    """HANZE historical-flood-impacts backend (per-instance output kind).

    Downloads the HANZE events / impacts table from its pinned Zenodo release,
    filters it by country / region / flood type / date window, and returns a
    :class:`pandas.DataFrame`. With `with_geometry=True` it instead returns a
    :class:`~pyramids.feature.collection.FeatureCollection` of the affected
    NUTS-3 regions.

    The record is public (CC-BY-4.0); no credentials are needed.

    Attributes:
        OUTPUT_KIND: Set **per instance** in :meth:`__init__` — `"tabular"` by
            default, `"vector"` when `with_geometry=True`. The facade reads it
            to know the return shape and rejects `aggregate=` for both.
        REQUIRES_TIME_WINDOW: `False` — a request without a window returns every
            year the record covers.

    Examples:
        - Pull DE + NL flood events, or the affected-region geometry, through the
          facade (both fetch from Zenodo, so this is illustrative, not a
          doctest):

            ```python
            from earthlens.core import EarthLens

            events = EarthLens(
                "hanze", start="1950", end="2020", country=["DE", "NL"]
            ).download()  # a pandas.DataFrame of events + impacts

            regions = EarthLens(
                "hanze", start="1990", end="2020", country="DE", with_geometry=True
            ).download()  # a FeatureCollection of the affected NUTS-3 regions
            ```
    """

    OUTPUT_KIND: OutputKind = "tabular"

    REQUIRES_TIME_WINDOW = False

    AGGREGATE_REFUSAL_REASON = (
        "HANZE serves observed flood event / impact records (and, with "
        "with_geometry, their NUTS-3 region polygons), not gridded rasters, so "
        "there is no meaningful gridded reduction. Call download() without "
        "aggregate= and post-process the returned DataFrame / FeatureCollection "
        "directly"
    )

    #: Whether the transport should draw a progress bar, set from
    #: `download(progress_bar=...)` so the flag reaches the fetch.
    _progress: bool = True

    def __init__(
        self,
        start: str | None = None,
        end: str | None = None,
        lat_lim: list[float] | None = None,
        lon_lim: list[float] | None = None,
        temporal_resolution: str = "all",
        path: Path | str | None = None,
        fmt: str = "%Y-%m-%d",
        country: str | list[str] | None = None,
        region: str | list[str] | None = None,
        flood_type: str | list[str] | None = None,
        with_geometry: bool = False,
        timeout: float = 120.0,
    ):
        """Initialise a HANZE backend instance.

        Args:
            start: Inclusive start of an optional window, parsed with `fmt`. Only
                its year is significant — HANZE indexes events by year. `None`
                means "from the beginning of the record".
            end: Inclusive end of the optional window; `None` means "to the end
                of the record".
            lat_lim: `[lat_min, lat_max]` bounding-box latitudes in degrees. A
                non-global box selects the affected regions (and, on the tabular
                path, the events touching them) that intersect it, which loads
                the region geometry.
            lon_lim: `[lon_min, lon_max]` bounding-box longitudes in degrees.
            temporal_resolution: HANZE issues one query over the whole window, so
                this is the sentinel `"all"`, not a pandas frequency alias.
            path: Output directory for the cached source files and the written
                table / vector file. Created by the parent class if absent.
            fmt: `strptime` format for `start` / `end`.
            country: One ISO2 country code or a list of them (`"DE"`,
                `["DE", "NL"]`). `None` keeps every country.
            region: One NUTS-3 code or a list of them (`"DE300"`), matched
                against each event's affected-region list. `None` keeps every
                region.
            flood_type: One flood type or a list of them — any of `"River"`,
                `"Flash"`, `"Coastal"`, `"River/Coastal"`. `None` keeps every
                type.
            with_geometry: When `True`, additionally download the NUTS-3 region
                shapefile and return a `FeatureCollection` of the affected
                regions instead of the events `DataFrame` (sets
                `OUTPUT_KIND="vector"` for this instance).
            timeout: Per-request timeout in seconds for the Zenodo downloads.

        Raises:
            ValueError: If a `country` value is not a 2-letter ISO2 code, a
                `region` value is not a 5-character NUTS-3 code, or a
                `flood_type` value is not a registered HANZE flood type.
        """
        self._catalog = Catalog()
        # Resolve the always-loaded record / geometry blocks into non-optional
        # attributes once, so the rest of the backend (and the type checker) can
        # read them without re-guarding the catalog's `... | None` fields.
        if self._catalog.record is None or self._catalog.geometry is None:
            raise ValueError(
                "the HANZE catalog failed to load its 'record:'/'geometry:' "
                "block; the bundled hanze_data_catalog.yaml is malformed."
            )
        self._record = self._catalog.record
        self._geo = self._catalog.geometry
        self._country = _normalize_country(country)
        self._region = _normalize_region(region)
        # Validate flood types against the catalog (did-you-mean hint on a typo).
        self._flood_types = [
            self._resolve_flood_type(name) for name in _as_list(flood_type)
        ]
        self._with_geometry = with_geometry
        self._timeout = timeout
        self._http: HttpClient | None = None
        self._regions_fc: FeatureCollection | None = None

        self.OUTPUT_KIND = "vector" if with_geometry else "tabular"

        super().__init__(
            start=cast("str", start),
            end=cast("str", end),
            # HANZE is facet-only: it is a single product selected by
            # country=/region=/flood_type=, so it declares no `variables`
            # parameter (the facade neither requires nor forwards one). The base
            # class still wants the argument, so an empty list is passed here.
            variables=[],
            temporal_resolution=temporal_resolution,
            lat_lim=lat_lim if lat_lim is not None else _GLOBAL_LAT,
            lon_lim=lon_lim if lon_lim is not None else _GLOBAL_LON,
            fmt=fmt,
            path=path,
        )

    def _resolve_flood_type(self, name: str) -> str:
        """Resolve one requested flood type against the catalog vocabulary.

        Args:
            name: A flood-type string in any casing (`"river"`).

        Returns:
            str: The canonical HANZE flood type (`"River"`).

        Raises:
            ValueError: If `name` is not a registered flood type.
        """
        # Case-insensitive match against the catalog vocabulary, so "river"
        # resolves to "River" rather than failing the did-you-mean.
        wanted = name.strip().lower()
        for canonical in self._catalog.flood_types():
            if canonical.lower() == wanted:
                return canonical
        # Fall through to the catalog's did-you-mean error.
        self._catalog.get_flood_type(name.strip())
        raise AssertionError("unreachable")  # pragma: no cover

    def _create_grid(self, lat_lim: list[float], lon_lim: list[float]) -> SpatialExtent:
        """Capture the requested bounds as a :class:`SpatialExtent`.

        Args:
            lat_lim: `[min_lat, max_lat]`.
            lon_lim: `[min_lon, max_lon]`.

        Returns:
            SpatialExtent: The requested extent.
        """
        return SpatialExtent.from_pairs(lat_lim=lat_lim, lon_lim=lon_lim)

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

        HANZE covers a whole record and is indexed by event year, so `None`
        bounds are legal and yield a `None`-dated extent.

        Args:
            start: Inclusive start date string, or `None`.
            end: Inclusive end date string, or `None`.
            temporal_resolution: Recorded as the resolution label.
            fmt: `strptime` format tried first for a string bound; a non-matching
                string falls back to an ISO-8601 parse.

        Returns:
            TemporalExtent: Frozen model with the parsed (or `None`) endpoints.

        Raises:
            ValueError: If `start` parses to a date later than `end`.
        """
        start_dt = to_datetime(start, fmt) if start else None
        end_dt = to_datetime(end, fmt) if end else None
        dates = (
            pd.DatetimeIndex([start_dt, end_dt])
            if start_dt is not None and end_dt is not None
            else pd.DatetimeIndex([])
        )
        return TemporalExtent(
            start_date=start_dt,
            end_date=end_dt,
            resolution=temporal_resolution,
            dates=dates,
        )

    @property
    def _year_range(self) -> tuple[int | None, int | None]:
        """Return the requested window as inclusive year bounds.

        Returns:
            tuple[int | None, int | None]: `(first_year, last_year)`, either
                `None` when that end of the window was not given.
        """
        start = self.time.start_date
        end = self.time.end_date
        return (
            start.year if start is not None else None,
            end.year if end is not None else None,
        )

    @property
    def _bbox(self) -> tuple[float, float, float, float] | None:
        """Return the request bbox, or `None` for a Europe-wide request.

        Returns:
            tuple[float, float, float, float] | None:
                `(min_lon, min_lat, max_lon, max_lat)`, or `None` when the
                request covers the whole globe (so no region is dropped).
        """
        space = self.space
        whole_globe = (
            space.latitude_min <= _GLOBAL_LAT[0]
            and space.latitude_max >= _GLOBAL_LAT[1]
            and space.longitude_min <= _GLOBAL_LON[0]
            and space.longitude_max >= _GLOBAL_LON[1]
        )
        if whole_globe:
            return None
        return (
            space.longitude_min,
            space.latitude_min,
            space.longitude_max,
            space.latitude_max,
        )

    def _client(self) -> HttpClient:
        """Return this instance's pooled client, building it on first use.

        Zenodo is a single origin; a dropped connection there is a normal event,
        so connection and timeout errors are retried too (matching `emdat`).

        Returns:
            HttpClient: The same instance on every later call.
        """
        if self._http is None:
            self._http = HttpClient(
                timeout=self._timeout,
                retry_on_exceptions=(requests.ConnectionError, requests.Timeout),
            )
        return self._http

    def _download_file(
        self, key: str, *, expect_magic: bytes | tuple[bytes, ...] | None = None
    ) -> Path:
        """Download one catalog file into `root_dir`, reusing a cached copy.

        Args:
            key: A logical file key (`"events"`, `"regions"`, or `"region_names"`).
            expect_magic: Optional leading-byte guard (one prefix or a tuple of
                acceptable prefixes) rejecting an error page served with a 200
                status.

        Returns:
            Path: The local file in `root_dir`.
        """
        record = self._record.record
        entry = self._catalog.file(key)
        local = self.root_dir / entry.name
        if not local.exists():
            logger.info(f"HANZE: downloading {entry.name} (record {record}).")
            self._client().download(
                entry.content_url(record),
                local,
                expect_magic=expect_magic,
                progress=self._progress,
            )
        return local

    def _load_events(self) -> pd.DataFrame:
        """Download and parse the HANZE events / impacts CSV.

        The download is guarded by the events header's leading bytes, so an HTML
        error page served with a `200` status (a proxy / CDN hiccup) is rejected
        at the download site rather than cached under the CSV name and failing
        confusingly at `read_csv` on every later call — the same guard the region
        zip (`PK`) and the sibling `emdat` xlsx (`PK`) use. The live file is
        published with a UTF-8 BOM, so both the BOM-prefixed and the plain header
        are accepted; `utf-8-sig` then strips the BOM so the `ID` column name is
        clean.

        Returns:
            pandas.DataFrame: The full events table, HANZE's documented headers.
        """
        local = self._download_file("events", expect_magic=(b"\xef\xbb\xbfID,", b"ID,"))
        return pd.read_csv(local, encoding="utf-8-sig")

    def _load_regions(self) -> FeatureCollection:
        """Download, extract and read the NUTS-3 region shapefile (cached).

        Returns:
            FeatureCollection: The region polygons in the shapefile's stored CRS
                (`EPSG:3035`).
        """
        if self._regions_fc is not None:
            return self._regions_fc
        from pyramids.feature.collection import FeatureCollection

        archive = self._download_file("regions", expect_magic=b"PK")
        stem = self._geo.member_stem
        members = extract_members(
            archive,
            self.root_dir / "hanze_regions",
            include=(".shp", ".shx", ".dbf", ".prj", ".cpg"),
        )
        shp = next(
            (m for m in members if m.stem == stem and m.suffix.lower() == ".shp"),
            None,
        )
        if shp is None:
            raise ValueError(
                f"the HANZE region archive {archive.name} has no "
                f"{stem}.shp member (found {[m.name for m in members]})."
            )
        self._regions_fc = FeatureCollection.read_file(str(shp))
        return self._regions_fc

    def _bbox_region_codes(self) -> set[str] | None:
        """Return the NUTS-3 codes whose region intersects the request bbox.

        Loads the region geometry (in `EPSG:3035`), reprojects to WGS84, and
        selects the polygons intersecting the bbox. `None` when the request is
        Europe-wide (no bbox restriction).

        Returns:
            set[str] | None: The in-bbox NUTS-3 codes, or `None` for a whole-globe
                request.
        """
        bbox = self._bbox
        if bbox is None:
            return None
        regions = self._load_regions().to_crs(geometry_module.OUTPUT_CRS)
        join_field = self._geo.join_field
        min_lon, min_lat, max_lon, max_lat = bbox
        within = regions.cx[min_lon:max_lon, min_lat:max_lat]
        # Upper-cased here (like `self._region`) so `_row_matches_codes` compares
        # already-normalised sets rather than re-casing them per event row.
        return set(within[join_field].astype(str).str.upper())

    def _filter_events(self, events: pd.DataFrame) -> pd.DataFrame:
        """Apply the request's country / region / type / date / bbox filters.

        Args:
            events: The full events table.

        Returns:
            pandas.DataFrame: The matching rows, index reset.
        """
        columns = self._catalog.columns
        mask = pd.Series(True, index=events.index)

        if self._country:
            mask &= (
                events[columns["country_code"]]
                .astype(str)
                .str.upper()
                .isin(self._country)
            )
        if self._flood_types:
            mask &= events[columns["type"]].isin(self._flood_types)

        first_year, last_year = self._year_range
        if first_year is not None:
            mask &= events[columns["year"]] >= first_year
        if last_year is not None:
            mask &= events[columns["year"]] <= last_year

        # An explicit `region=` restriction only counts when non-empty; the
        # bbox-derived set counts whenever a bbox is set, even if it resolves to
        # no regions (a bbox over open water legitimately drops every event).
        code_filters: list[set[str]] = []
        if self._region:
            code_filters.append(self._region)
        bbox_codes = self._bbox_region_codes()
        if bbox_codes is not None:
            code_filters.append(bbox_codes)
        if code_filters:
            mask &= events[columns["regions_nuts3"]].apply(
                lambda cell: self._row_matches_codes(cell, code_filters)
            )

        return events[mask].reset_index(drop=True)

    @staticmethod
    def _row_matches_codes(cell: object, code_filters: list[set[str]]) -> bool:
        """Whether an event's affected-region list satisfies every code filter.

        Args:
            cell: One `Regions affected (NUTS 3)` cell.
            code_filters: One set per active restriction (explicit `region=`, and
                the bbox-derived codes); the event must intersect **each**.

        Returns:
            bool: `True` when the event's codes intersect every filter set.
        """
        # The filter sets are already upper-cased at construction (`self._region`)
        # and in `_bbox_region_codes`, so only the row's codes need normalising.
        row_codes = {code.upper() for code in geometry_module.split_nuts3(cell)}
        return all(bool(row_codes & codes) for codes in code_filters)

    def _search(self) -> list[RemoteProduct]:
        """Pin the one product to fetch (the HANZE events table).

        Returns:
            list[RemoteProduct]: A single product carrying the record id.
        """
        return [
            RemoteProduct(
                id="hanze:events",
                metadata={"record": self._record.record},
            )
        ]

    def _fetch(self, products: list[RemoteProduct]) -> list[Any]:
        """Download, filter, and shape the one product to the instance's kind.

        Args:
            products: The single-element list from :meth:`_search`.

        Returns:
            list[Any]: One element — a filtered :class:`pandas.DataFrame`
                (tabular), or a
                :class:`~pyramids.feature.collection.FeatureCollection` of the
                affected regions (vector).
        """
        events = self._filter_events(self._load_events())
        if self._with_geometry:
            return [self._build_region_collection(events)]
        return [events]

    def _build_region_collection(self, events: pd.DataFrame) -> FeatureCollection:
        """Join the filtered events to their affected NUTS-3 region polygons.

        When the request carries a bbox, the joined regions are additionally
        restricted to it by bounding-box intersection (`GeoDataFrame.cx`): an
        event that touched an in-bbox region also lists regions outside the box,
        and returning those would put polygons well outside a spatial query on
        the map. A region whose extent intersects the box is kept **whole** (it
        is selected, not geometrically trimmed), so the vector answer stays
        "affected regions within the box", matching the tabular bbox path.

        Args:
            events: The filtered events table.

        Returns:
            FeatureCollection: One polygon per affected region (restricted to the
                regions intersecting the bbox when one is set), CRS `EPSG:4326`.
        """
        regions = self._load_regions()
        geometry = self._geo
        collection = geometry_module.join_events_to_regions(
            events,
            regions,
            regions_column=self._catalog.columns["regions_nuts3"],
            join_field=geometry.join_field,
            name_field=geometry.name_field,
        )
        bbox = self._bbox
        if bbox is None or not len(collection):
            return collection
        from pyramids.feature.collection import FeatureCollection

        min_lon, min_lat, max_lon, max_lat = bbox
        within = collection.cx[min_lon:max_lon, min_lat:max_lat]
        return FeatureCollection(within.reset_index(drop=True))

    def _api(self) -> list[Any]:
        """Compose :meth:`_search` and :meth:`_fetch`.

        Returns:
            list[Any]: The fetched result (one element).
        """
        return self._api_via_search_fetch()

    def download(self, progress_bar: bool = True) -> pd.DataFrame | FeatureCollection:
        """Fetch HANZE and return the per-instance shape.

        Runs the download + filter, writes the result to `path` (a CSV for the
        tabular default, a GeoPackage for `with_geometry`), and returns it.

        Args:
            progress_bar: Whether to draw a download progress bar. Passed through
                to the transport, so `False` really does silence it.

        Returns:
            A :class:`pandas.DataFrame` of events + impacts (the default), or a
            :class:`~pyramids.feature.collection.FeatureCollection` of the
            affected NUTS-3 regions (`with_geometry=True`). Both are also written
            under `root_dir`.

        Raises:
            requests.HTTPError: If a Zenodo download returns a non-2xx status.
            ValueError: If a download's body fails its content guard (an HTML
                error page served with a 200 status), or `with_geometry=True` and
                the region archive has no `<member_stem>.shp` member.
        """
        self._progress = progress_bar
        results = self._api()
        # `_search` always yields one product, so `_fetch` returns a single
        # element (a 0-row DataFrame / empty FC is still one element); the
        # `_empty_result()` fallback is a defensive guard for a future `_search`
        # that could return nothing, not a path this backend reaches today.
        result = results[0] if results else self._empty_result()
        self._log_citation()

        if self.OUTPUT_KIND == "vector":
            out_path = self.root_dir / (self._result_stem("hanze_regions") + ".gpkg")
            # Written unconditionally — an empty result still writes a schema-only
            # GeoPackage, so the vector path matches the tabular one (which always
            # writes a header-only CSV) and a caller globbing `path` finds a file.
            result.to_file(str(out_path), driver="GPKG")
            logger.info(
                f"HANZE: {len(result)} affected region(s) written to {out_path}."
            )
            return result

        out_path = self.root_dir / (self._result_stem("hanze_events") + ".csv")
        result.to_csv(out_path, index=False)
        logger.info(f"HANZE: {len(result)} event(s) written to {out_path}.")
        return result

    def _empty_result(self) -> pd.DataFrame | FeatureCollection:
        """Return the empty result matching the instance's output kind."""
        if self._with_geometry:
            return geometry_module.empty_region_fc()
        return pd.DataFrame()

    def _result_stem(self, base: str) -> str:
        """Compose an output file stem that encodes the request's filters.

        A plain `base` for an unfiltered request; otherwise `base-<digest>` so
        two differently-filtered queries into one `path=` do not overwrite each
        other. The digest is order-insensitive in the multi-value filters.

        Args:
            base: The stem prefix (`"hanze_events"` / `"hanze_regions"`).

        Returns:
            str: `base`, or `base-<8-hex-digest>` when any filter is active.
        """
        first_year, last_year = self._year_range
        applied = (
            bool(self._country),
            bool(self._region),
            bool(self._flood_types),
            first_year is not None,
            last_year is not None,
            self._bbox is not None,
        )
        if not any(applied):
            return base
        request = (
            tuple(sorted(self._country)),
            tuple(sorted(self._region)),
            tuple(sorted(self._flood_types)),
            self._year_range,
            self._bbox,
        )
        digest = hashlib.sha1(
            repr(request).encode(), usedforsecurity=False
        ).hexdigest()[:8]
        return f"{base}-{digest}"

    def _log_citation(self) -> None:
        """Log the CC-BY attribution once (info, not a warning)."""
        record = self._record
        if record.attribution:
            logger.info(f"HANZE source citation: {record.attribution}")

__init__(start=None, end=None, lat_lim=None, lon_lim=None, temporal_resolution='all', path=None, fmt='%Y-%m-%d', country=None, region=None, flood_type=None, with_geometry=False, timeout=120.0)

Initialise a HANZE backend instance.

Parameters:

Name	Type	Description	Default
start	str | None	Inclusive start of an optional window, parsed with fmt. Only its year is significant — HANZE indexes events by year. None means "from the beginning of the record".	None
end	str | None	Inclusive end of the optional window; None means "to the end of the record".	None
lat_lim	list[float] | None	[lat_min, lat_max] bounding-box latitudes in degrees. A non-global box selects the affected regions (and, on the tabular path, the events touching them) that intersect it, which loads the region geometry.	None
lon_lim	list[float] | None	[lon_min, lon_max] bounding-box longitudes in degrees.	None
temporal_resolution	str	HANZE issues one query over the whole window, so this is the sentinel "all", not a pandas frequency alias.	'all'
path	Path | str | None	Output directory for the cached source files and the written table / vector file. Created by the parent class if absent.	None
fmt	str	strptime format for start / end.	'%Y-%m-%d'
country	str | list[str] | None	One ISO2 country code or a list of them ("DE", ["DE", "NL"]). None keeps every country.	None
region	str | list[str] | None	One NUTS-3 code or a list of them ("DE300"), matched against each event's affected-region list. None keeps every region.	None
flood_type	str | list[str] | None	One flood type or a list of them — any of "River", "Flash", "Coastal", "River/Coastal". None keeps every type.	None
with_geometry	bool	When True, additionally download the NUTS-3 region shapefile and return a FeatureCollection of the affected regions instead of the events DataFrame (sets OUTPUT_KIND="vector" for this instance).	False
timeout	float	Per-request timeout in seconds for the Zenodo downloads.	120.0

Raises:

Type	Description
ValueError	If a country value is not a 2-letter ISO2 code, a region value is not a 5-character NUTS-3 code, or a flood_type value is not a registered HANZE flood type.

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

def __init__(
    self,
    start: str | None = None,
    end: str | None = None,
    lat_lim: list[float] | None = None,
    lon_lim: list[float] | None = None,
    temporal_resolution: str = "all",
    path: Path | str | None = None,
    fmt: str = "%Y-%m-%d",
    country: str | list[str] | None = None,
    region: str | list[str] | None = None,
    flood_type: str | list[str] | None = None,
    with_geometry: bool = False,
    timeout: float = 120.0,
):
    """Initialise a HANZE backend instance.

    Args:
        start: Inclusive start of an optional window, parsed with `fmt`. Only
            its year is significant — HANZE indexes events by year. `None`
            means "from the beginning of the record".
        end: Inclusive end of the optional window; `None` means "to the end
            of the record".
        lat_lim: `[lat_min, lat_max]` bounding-box latitudes in degrees. A
            non-global box selects the affected regions (and, on the tabular
            path, the events touching them) that intersect it, which loads
            the region geometry.
        lon_lim: `[lon_min, lon_max]` bounding-box longitudes in degrees.
        temporal_resolution: HANZE issues one query over the whole window, so
            this is the sentinel `"all"`, not a pandas frequency alias.
        path: Output directory for the cached source files and the written
            table / vector file. Created by the parent class if absent.
        fmt: `strptime` format for `start` / `end`.
        country: One ISO2 country code or a list of them (`"DE"`,
            `["DE", "NL"]`). `None` keeps every country.
        region: One NUTS-3 code or a list of them (`"DE300"`), matched
            against each event's affected-region list. `None` keeps every
            region.
        flood_type: One flood type or a list of them — any of `"River"`,
            `"Flash"`, `"Coastal"`, `"River/Coastal"`. `None` keeps every
            type.
        with_geometry: When `True`, additionally download the NUTS-3 region
            shapefile and return a `FeatureCollection` of the affected
            regions instead of the events `DataFrame` (sets
            `OUTPUT_KIND="vector"` for this instance).
        timeout: Per-request timeout in seconds for the Zenodo downloads.

    Raises:
        ValueError: If a `country` value is not a 2-letter ISO2 code, a
            `region` value is not a 5-character NUTS-3 code, or a
            `flood_type` value is not a registered HANZE flood type.
    """
    self._catalog = Catalog()
    # Resolve the always-loaded record / geometry blocks into non-optional
    # attributes once, so the rest of the backend (and the type checker) can
    # read them without re-guarding the catalog's `... | None` fields.
    if self._catalog.record is None or self._catalog.geometry is None:
        raise ValueError(
            "the HANZE catalog failed to load its 'record:'/'geometry:' "
            "block; the bundled hanze_data_catalog.yaml is malformed."
        )
    self._record = self._catalog.record
    self._geo = self._catalog.geometry
    self._country = _normalize_country(country)
    self._region = _normalize_region(region)
    # Validate flood types against the catalog (did-you-mean hint on a typo).
    self._flood_types = [
        self._resolve_flood_type(name) for name in _as_list(flood_type)
    ]
    self._with_geometry = with_geometry
    self._timeout = timeout
    self._http: HttpClient | None = None
    self._regions_fc: FeatureCollection | None = None

    self.OUTPUT_KIND = "vector" if with_geometry else "tabular"

    super().__init__(
        start=cast("str", start),
        end=cast("str", end),
        # HANZE is facet-only: it is a single product selected by
        # country=/region=/flood_type=, so it declares no `variables`
        # parameter (the facade neither requires nor forwards one). The base
        # class still wants the argument, so an empty list is passed here.
        variables=[],
        temporal_resolution=temporal_resolution,
        lat_lim=lat_lim if lat_lim is not None else _GLOBAL_LAT,
        lon_lim=lon_lim if lon_lim is not None else _GLOBAL_LON,
        fmt=fmt,
        path=path,
    )

download(progress_bar=True)

Fetch HANZE and return the per-instance shape.

Runs the download + filter, writes the result to path (a CSV for the tabular default, a GeoPackage for with_geometry), and returns it.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Whether to draw a download progress bar. Passed through to the transport, so False really does silence it.	True

Returns:

Name	Type	Description
A	DataFrame | FeatureCollection	class:pandas.DataFrame of events + impacts (the default), or a
	DataFrame | FeatureCollection	class:~pyramids.feature.collection.FeatureCollection of the
	DataFrame | FeatureCollection	affected NUTS-3 regions (with_geometry=True). Both are also written
	DataFrame | FeatureCollection	under root_dir.

Raises:

Type	Description
HTTPError	If a Zenodo download returns a non-2xx status.
ValueError	If a download's body fails its content guard (an HTML error page served with a 200 status), or with_geometry=True and the region archive has no <member_stem>.shp member.

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

def download(self, progress_bar: bool = True) -> pd.DataFrame | FeatureCollection:
    """Fetch HANZE and return the per-instance shape.

    Runs the download + filter, writes the result to `path` (a CSV for the
    tabular default, a GeoPackage for `with_geometry`), and returns it.

    Args:
        progress_bar: Whether to draw a download progress bar. Passed through
            to the transport, so `False` really does silence it.

    Returns:
        A :class:`pandas.DataFrame` of events + impacts (the default), or a
        :class:`~pyramids.feature.collection.FeatureCollection` of the
        affected NUTS-3 regions (`with_geometry=True`). Both are also written
        under `root_dir`.

    Raises:
        requests.HTTPError: If a Zenodo download returns a non-2xx status.
        ValueError: If a download's body fails its content guard (an HTML
            error page served with a 200 status), or `with_geometry=True` and
            the region archive has no `<member_stem>.shp` member.
    """
    self._progress = progress_bar
    results = self._api()
    # `_search` always yields one product, so `_fetch` returns a single
    # element (a 0-row DataFrame / empty FC is still one element); the
    # `_empty_result()` fallback is a defensive guard for a future `_search`
    # that could return nothing, not a path this backend reaches today.
    result = results[0] if results else self._empty_result()
    self._log_citation()

    if self.OUTPUT_KIND == "vector":
        out_path = self.root_dir / (self._result_stem("hanze_regions") + ".gpkg")
        # Written unconditionally — an empty result still writes a schema-only
        # GeoPackage, so the vector path matches the tabular one (which always
        # writes a header-only CSV) and a caller globbing `path` finds a file.
        result.to_file(str(out_path), driver="GPKG")
        logger.info(
            f"HANZE: {len(result)} affected region(s) written to {out_path}."
        )
        return result

    out_path = self.root_dir / (self._result_stem("hanze_events") + ".csv")
    result.to_csv(out_path, index=False)
    logger.info(f"HANZE: {len(result)} event(s) written to {out_path}.")
    return result

HanzeFile

Bases: BaseModel

One downloadable Zenodo object of the pinned HANZE record.

Attributes:

Name	Type	Description
name	str	The file name on the record.
description	str	One-line human-readable summary.

Examples:

• The content URL is composed from the pinned record and file name:

  >>> from earthlens.hanze import Catalog
  >>> events = Catalog().file("events")
  >>> events.content_url(20478847)
  'https://zenodo.org/api/records/20478847/files/HANZE_events_v3_0_1b.csv/content'
  

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

class HanzeFile(BaseModel):
    """One downloadable Zenodo object of the pinned HANZE record.

    Attributes:
        name: The file name on the record.
        description: One-line human-readable summary.

    Examples:
        - The content URL is composed from the pinned record and file name:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> events = Catalog().file("events")
            >>> events.content_url(20478847)
            'https://zenodo.org/api/records/20478847/files/HANZE_events_v3_0_1b.csv/content'

            ```
    """

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

    name: str
    description: str = ""

    def content_url(self, record: int) -> str:
        """Return the Zenodo REST content URL this file is served from.

        Args:
            record: The pinned version record id the file belongs to.

        Returns:
            str: `https://zenodo.org/api/records/<record>/files/<name>/content`.

        Examples:
            - Compose the REST content URL for a file on a record:
                ```python
                >>> from earthlens.hanze import HanzeFile
                >>> HanzeFile(name="events.csv").content_url(20478847)
                'https://zenodo.org/api/records/20478847/files/events.csv/content'

                ```
        """
        return _CONTENT_URL.format(record=record, name=self.name)

content_url(record)

Return the Zenodo REST content URL this file is served from.

Parameters:

Name	Type	Description	Default
record	int	The pinned version record id the file belongs to.	required

Returns:

Name	Type	Description
str	str	https://zenodo.org/api/records/<record>/files/<name>/content.

Examples:

• Compose the REST content URL for a file on a record:

  >>> from earthlens.hanze import HanzeFile
  >>> HanzeFile(name="events.csv").content_url(20478847)
  'https://zenodo.org/api/records/20478847/files/events.csv/content'
  

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

def content_url(self, record: int) -> str:
    """Return the Zenodo REST content URL this file is served from.

    Args:
        record: The pinned version record id the file belongs to.

    Returns:
        str: `https://zenodo.org/api/records/<record>/files/<name>/content`.

    Examples:
        - Compose the REST content URL for a file on a record:
            ```python
            >>> from earthlens.hanze import HanzeFile
            >>> HanzeFile(name="events.csv").content_url(20478847)
            'https://zenodo.org/api/records/20478847/files/events.csv/content'

            ```
    """
    return _CONTENT_URL.format(record=record, name=self.name)

ZenodoRecord

Bases: BaseModel

The pinned Zenodo version record HANZE is fetched from.

Attributes:

Name	Type	Description
record	int	The pinned Zenodo version record id (20478847). Every file URL is composed from it, so a request is reproducible.
concept_doi	str	The moving concept DOI. Recorded so a refresh check can discover a newer version; never used to fetch.
version	str	The dataset version (v3.0.1-beta). Flagged as beta in the docs and logs.
data_period	str	The first-last year span the record covers ("1870-2025"), for documentation and the drift check.
license	str	SPDX-ish licence id (CC-BY-4.0).
attribution	str	The citation obligation the licence carries.

Examples:

• The record is the pinned version, not the concept DOI:

  >>> from earthlens.hanze import Catalog
  >>> Catalog().record.record
  20478847
  >>> Catalog().record.version
  'v3.0.1-beta'
  

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

class ZenodoRecord(BaseModel):
    """The pinned Zenodo version record HANZE is fetched from.

    Attributes:
        record: The pinned Zenodo *version* record id (`20478847`). Every file
            URL is composed from it, so a request is reproducible.
        concept_doi: The moving concept DOI. Recorded so a refresh check can
            discover a newer version; never used to fetch.
        version: The dataset version (`v3.0.1-beta`). Flagged as beta in the
            docs and logs.
        data_period: The `first-last` year span the record covers
            (`"1870-2025"`), for documentation and the drift check.
        license: SPDX-ish licence id (`CC-BY-4.0`).
        attribution: The citation obligation the licence carries.

    Examples:
        - The record is the pinned version, not the concept DOI:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> Catalog().record.record
            20478847
            >>> Catalog().record.version
            'v3.0.1-beta'

            ```
    """

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

    record: int
    concept_doi: str = ""
    version: str = ""
    data_period: str = ""
    license: str = ""
    attribution: str = ""

empty_region_fc()

Return an empty region FeatureCollection with the canonical schema.

Used when the filtered events reference no region present in the boundary file, so callers always get the same columns / dtypes back regardless of hit count.

Returns:

Name	Type	Description
FeatureCollection	FeatureCollection	Zero rows, the :data:REGION_COLUMNS columns with their dtypes, an empty geometry column, CRS EPSG:4326.

Examples:

• The schema is present even with no rows:

  >>> from earthlens.hanze.geometry import empty_region_fc, REGION_COLUMNS
  >>> fc = empty_region_fc()
  >>> len(fc)
  0
  >>> set(REGION_COLUMNS).issubset(fc.columns)
  True
  >>> fc.crs.to_epsg()
  4326
  

Source code in libs/providers/hazards/src/earthlens/hanze/geometry.py

def empty_region_fc() -> FeatureCollection:
    """Return an empty region `FeatureCollection` with the canonical schema.

    Used when the filtered events reference no region present in the boundary
    file, so callers always get the same columns / dtypes back regardless of hit
    count.

    Returns:
        FeatureCollection: Zero rows, the :data:`REGION_COLUMNS` columns with
            their dtypes, an empty `geometry` column, CRS `EPSG:4326`.

    Examples:
        - The schema is present even with no rows:
            ```python
            >>> from earthlens.hanze.geometry import empty_region_fc, REGION_COLUMNS
            >>> fc = empty_region_fc()
            >>> len(fc)
            0
            >>> set(REGION_COLUMNS).issubset(fc.columns)
            True
            >>> fc.crs.to_epsg()
            4326

            ```
    """
    frame = pd.DataFrame(
        {column: pd.Series([], dtype=dtype) for column, dtype in REGION_COLUMNS.items()}
    )
    gdf = gpd.GeoDataFrame(
        frame, geometry=gpd.GeoSeries([], crs=OUTPUT_CRS), crs=OUTPUT_CRS
    )
    return FeatureCollection(gdf)

join_events_to_regions(events, regions, *, regions_column, join_field, name_field)

Join filtered events to their affected NUTS-3 region polygons.

Splits each event's Regions affected (NUTS 3) list, counts the events per region, selects the boundary polygons whose join_field is among the affected codes, reprojects them to WGS84, and returns one feature per affected region carrying its code, name and event count.

Parameters:

Name	Type	Description	Default
events	DataFrame	The filtered events table.	required
regions	FeatureCollection	The NUTS-3 boundary polygons, in the shapefile's stored CRS (EPSG:3035), carrying join_field and name_field attributes.	required
regions_column	str	The events column holding the semicolon-separated NUTS-3 code list ("Regions affected (NUTS 3)").	required
join_field	str	The boundary attribute holding the NUTS-3 code ("Code").	required
name_field	str	The boundary attribute holding the region name ("Name").	required

Returns:

Name	Type	Description
FeatureCollection	FeatureCollection	One polygon per affected region, columns nuts3_code / region_name / n_events / geometry, CRS EPSG:4326. Empty (schema-only) when no affected code is present in the boundary file.

Examples:

• Join two events to their affected region polygons and read the counts:

  >>> import geopandas as gpd
  >>> import pandas as pd
  >>> from shapely.geometry import box
  >>> from earthlens.hanze.geometry import join_events_to_regions
  >>> regions = gpd.GeoDataFrame(
  ...     {"Code": ["DE300", "NL414"], "Name": ["Berlin", "Zuidoost"]},
  ...     geometry=[box(13, 52, 14, 53), box(5, 51, 6, 52)],
  ...     crs="EPSG:4326",
  ... )
  >>> events = pd.DataFrame({"regions": ["DE300;NL414", "DE300"]})
  >>> fc = join_events_to_regions(
  ...     events, regions, regions_column="regions",
  ...     join_field="Code", name_field="Name",
  ... )
  >>> dict(zip(fc["nuts3_code"], fc["n_events"]))
  {'DE300': 2, 'NL414': 1}
  >>> fc.crs.to_epsg()
  4326
  

Source code in libs/providers/hazards/src/earthlens/hanze/geometry.py

def join_events_to_regions(
    events: pd.DataFrame,
    regions: FeatureCollection,
    *,
    regions_column: str,
    join_field: str,
    name_field: str,
) -> FeatureCollection:
    """Join filtered events to their affected NUTS-3 region polygons.

    Splits each event's `Regions affected (NUTS 3)` list, counts the events per
    region, selects the boundary polygons whose `join_field` is among the
    affected codes, reprojects them to WGS84, and returns one feature per
    affected region carrying its code, name and event count.

    Args:
        events: The filtered events table.
        regions: The NUTS-3 boundary polygons, in the shapefile's stored CRS
            (`EPSG:3035`), carrying `join_field` and `name_field` attributes.
        regions_column: The events column holding the semicolon-separated NUTS-3
            code list (`"Regions affected (NUTS 3)"`).
        join_field: The boundary attribute holding the NUTS-3 code (`"Code"`).
        name_field: The boundary attribute holding the region name (`"Name"`).

    Returns:
        FeatureCollection: One polygon per affected region, columns
            `nuts3_code` / `region_name` / `n_events` / `geometry`, CRS
            `EPSG:4326`. Empty (schema-only) when no affected code is present in
            the boundary file.

    Examples:
        - Join two events to their affected region polygons and read the counts:
            ```python
            >>> import geopandas as gpd
            >>> import pandas as pd
            >>> from shapely.geometry import box
            >>> from earthlens.hanze.geometry import join_events_to_regions
            >>> regions = gpd.GeoDataFrame(
            ...     {"Code": ["DE300", "NL414"], "Name": ["Berlin", "Zuidoost"]},
            ...     geometry=[box(13, 52, 14, 53), box(5, 51, 6, 52)],
            ...     crs="EPSG:4326",
            ... )
            >>> events = pd.DataFrame({"regions": ["DE300;NL414", "DE300"]})
            >>> fc = join_events_to_regions(
            ...     events, regions, regions_column="regions",
            ...     join_field="Code", name_field="Name",
            ... )
            >>> dict(zip(fc["nuts3_code"], fc["n_events"]))
            {'DE300': 2, 'NL414': 1}
            >>> fc.crs.to_epsg()
            4326

            ```
    """
    counts = event_region_counts(events, regions_column)
    if not counts or join_field not in regions.columns:
        return empty_region_fc()

    # Compare upper-cased on both sides so the selection tolerates any case skew
    # between the events column and the boundary file's `Code` field, matching
    # the tabular path's normalisation.
    selected = regions[regions[join_field].astype(str).str.upper().isin(counts.keys())]
    if not len(selected):
        return empty_region_fc()

    # Reproject to WGS84 first; the shapefile is ETRS89-LAEA (EPSG:3035). Reset
    # the index so the geometry `GeoSeries` (built from a positional numpy array
    # below) aligns with `frame` row-for-row: `selected` keeps the boundary
    # file's original scattered indices, and geopandas aligns a `GeoSeries` to
    # the frame *by index*, so without this a region at a high original index
    # would be paired with a missing (null) geometry.
    reprojected = selected.to_crs(OUTPUT_CRS).reset_index(drop=True)
    frame = pd.DataFrame(
        {
            "nuts3_code": reprojected[join_field].astype("string"),
            "region_name": reprojected[name_field].astype("string")
            if name_field in reprojected.columns
            else pd.Series([pd.NA] * len(reprojected), dtype="string"),
            "n_events": [counts[str(code).upper()] for code in reprojected[join_field]],
        }
    ).astype(REGION_COLUMNS)
    gdf = gpd.GeoDataFrame(
        frame,
        geometry=gpd.GeoSeries(reprojected.geometry.to_numpy(), crs=OUTPUT_CRS),
        crs=OUTPUT_CRS,
    )
    return FeatureCollection(gdf)

earthlens.hanze.backend

Backend that fetches HANZE historical European flood events and impacts.

HANZE(AbstractDataSource) downloads the HANZE (Historical Analysis of Natural Hazards in Europe) database of observed European flood events and their impacts (Paprotny et al.) from its pinned static Zenodo release, filters it, and returns the event / impact records as a :class:pandas.DataFrame. It is the observed hazard -> loss record — real floods with fatalities, persons affected, area flooded and economic losses — so a modelled event set can be validated against the observed loss distribution. Companion to the global emdat backend.

Three design points carry this backend:

• Per-instance OUTPUT_KIND. The default is tabular, returning a :class:pandas.DataFrame of events + impacts. Passing with_geometry=True makes the instance vector: it additionally downloads the NUTS-3 region boundary shapefile and returns a pyramids :class:~pyramids.feature.collection.FeatureCollection of the affected regions (the emdat / eumetsat per-instance pattern). The facade reads the instance attribute to know the return shape and to gate aggregate=.
• Direct file download, not range-read. HANZE ships small individual Zenodo objects (a 618 KB events CSV, a 2.4 MB region zip), so each is fetched whole with :class:~earthlens.base.http.HttpClient and cached under path — none of caravan's multi-GB range-read machinery applies.
• No new dependency. HttpClient + pandas + base/archive + pyramids are all core, and the Zenodo record is public (CC-BY-4.0), so there is no auth and no [hanze] extra.

These are event records, not gridded rasters, so aggregate= is refused and nothing here imports a gridded-array library (no xarray).

HANZE

Bases: AbstractDataSource

HANZE historical-flood-impacts backend (per-instance output kind).

Downloads the HANZE events / impacts table from its pinned Zenodo release, filters it by country / region / flood type / date window, and returns a :class:pandas.DataFrame. With with_geometry=True it instead returns a :class:~pyramids.feature.collection.FeatureCollection of the affected NUTS-3 regions.

The record is public (CC-BY-4.0); no credentials are needed.

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	Set per instance in :meth:__init__ — "tabular" by default, "vector" when with_geometry=True. The facade reads it to know the return shape and rejects aggregate= for both.
REQUIRES_TIME_WINDOW		False — a request without a window returns every year the record covers.

Examples:

• Pull DE + NL flood events, or the affected-region geometry, through the facade (both fetch from Zenodo, so this is illustrative, not a doctest):

  from earthlens.core import EarthLens
  
  events = EarthLens(
      "hanze", start="1950", end="2020", country=["DE", "NL"]
  ).download()  # a pandas.DataFrame of events + impacts
  
  regions = EarthLens(
      "hanze", start="1990", end="2020", country="DE", with_geometry=True
  ).download()  # a FeatureCollection of the affected NUTS-3 regions
  

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

class HANZE(AbstractDataSource):
    """HANZE historical-flood-impacts backend (per-instance output kind).

    Downloads the HANZE events / impacts table from its pinned Zenodo release,
    filters it by country / region / flood type / date window, and returns a
    :class:`pandas.DataFrame`. With `with_geometry=True` it instead returns a
    :class:`~pyramids.feature.collection.FeatureCollection` of the affected
    NUTS-3 regions.

    The record is public (CC-BY-4.0); no credentials are needed.

    Attributes:
        OUTPUT_KIND: Set **per instance** in :meth:`__init__` — `"tabular"` by
            default, `"vector"` when `with_geometry=True`. The facade reads it
            to know the return shape and rejects `aggregate=` for both.
        REQUIRES_TIME_WINDOW: `False` — a request without a window returns every
            year the record covers.

    Examples:
        - Pull DE + NL flood events, or the affected-region geometry, through the
          facade (both fetch from Zenodo, so this is illustrative, not a
          doctest):

            ```python
            from earthlens.core import EarthLens

            events = EarthLens(
                "hanze", start="1950", end="2020", country=["DE", "NL"]
            ).download()  # a pandas.DataFrame of events + impacts

            regions = EarthLens(
                "hanze", start="1990", end="2020", country="DE", with_geometry=True
            ).download()  # a FeatureCollection of the affected NUTS-3 regions
            ```
    """

    OUTPUT_KIND: OutputKind = "tabular"

    REQUIRES_TIME_WINDOW = False

    AGGREGATE_REFUSAL_REASON = (
        "HANZE serves observed flood event / impact records (and, with "
        "with_geometry, their NUTS-3 region polygons), not gridded rasters, so "
        "there is no meaningful gridded reduction. Call download() without "
        "aggregate= and post-process the returned DataFrame / FeatureCollection "
        "directly"
    )

    #: Whether the transport should draw a progress bar, set from
    #: `download(progress_bar=...)` so the flag reaches the fetch.
    _progress: bool = True

    def __init__(
        self,
        start: str | None = None,
        end: str | None = None,
        lat_lim: list[float] | None = None,
        lon_lim: list[float] | None = None,
        temporal_resolution: str = "all",
        path: Path | str | None = None,
        fmt: str = "%Y-%m-%d",
        country: str | list[str] | None = None,
        region: str | list[str] | None = None,
        flood_type: str | list[str] | None = None,
        with_geometry: bool = False,
        timeout: float = 120.0,
    ):
        """Initialise a HANZE backend instance.

        Args:
            start: Inclusive start of an optional window, parsed with `fmt`. Only
                its year is significant — HANZE indexes events by year. `None`
                means "from the beginning of the record".
            end: Inclusive end of the optional window; `None` means "to the end
                of the record".
            lat_lim: `[lat_min, lat_max]` bounding-box latitudes in degrees. A
                non-global box selects the affected regions (and, on the tabular
                path, the events touching them) that intersect it, which loads
                the region geometry.
            lon_lim: `[lon_min, lon_max]` bounding-box longitudes in degrees.
            temporal_resolution: HANZE issues one query over the whole window, so
                this is the sentinel `"all"`, not a pandas frequency alias.
            path: Output directory for the cached source files and the written
                table / vector file. Created by the parent class if absent.
            fmt: `strptime` format for `start` / `end`.
            country: One ISO2 country code or a list of them (`"DE"`,
                `["DE", "NL"]`). `None` keeps every country.
            region: One NUTS-3 code or a list of them (`"DE300"`), matched
                against each event's affected-region list. `None` keeps every
                region.
            flood_type: One flood type or a list of them — any of `"River"`,
                `"Flash"`, `"Coastal"`, `"River/Coastal"`. `None` keeps every
                type.
            with_geometry: When `True`, additionally download the NUTS-3 region
                shapefile and return a `FeatureCollection` of the affected
                regions instead of the events `DataFrame` (sets
                `OUTPUT_KIND="vector"` for this instance).
            timeout: Per-request timeout in seconds for the Zenodo downloads.

        Raises:
            ValueError: If a `country` value is not a 2-letter ISO2 code, a
                `region` value is not a 5-character NUTS-3 code, or a
                `flood_type` value is not a registered HANZE flood type.
        """
        self._catalog = Catalog()
        # Resolve the always-loaded record / geometry blocks into non-optional
        # attributes once, so the rest of the backend (and the type checker) can
        # read them without re-guarding the catalog's `... | None` fields.
        if self._catalog.record is None or self._catalog.geometry is None:
            raise ValueError(
                "the HANZE catalog failed to load its 'record:'/'geometry:' "
                "block; the bundled hanze_data_catalog.yaml is malformed."
            )
        self._record = self._catalog.record
        self._geo = self._catalog.geometry
        self._country = _normalize_country(country)
        self._region = _normalize_region(region)
        # Validate flood types against the catalog (did-you-mean hint on a typo).
        self._flood_types = [
            self._resolve_flood_type(name) for name in _as_list(flood_type)
        ]
        self._with_geometry = with_geometry
        self._timeout = timeout
        self._http: HttpClient | None = None
        self._regions_fc: FeatureCollection | None = None

        self.OUTPUT_KIND = "vector" if with_geometry else "tabular"

        super().__init__(
            start=cast("str", start),
            end=cast("str", end),
            # HANZE is facet-only: it is a single product selected by
            # country=/region=/flood_type=, so it declares no `variables`
            # parameter (the facade neither requires nor forwards one). The base
            # class still wants the argument, so an empty list is passed here.
            variables=[],
            temporal_resolution=temporal_resolution,
            lat_lim=lat_lim if lat_lim is not None else _GLOBAL_LAT,
            lon_lim=lon_lim if lon_lim is not None else _GLOBAL_LON,
            fmt=fmt,
            path=path,
        )

    def _resolve_flood_type(self, name: str) -> str:
        """Resolve one requested flood type against the catalog vocabulary.

        Args:
            name: A flood-type string in any casing (`"river"`).

        Returns:
            str: The canonical HANZE flood type (`"River"`).

        Raises:
            ValueError: If `name` is not a registered flood type.
        """
        # Case-insensitive match against the catalog vocabulary, so "river"
        # resolves to "River" rather than failing the did-you-mean.
        wanted = name.strip().lower()
        for canonical in self._catalog.flood_types():
            if canonical.lower() == wanted:
                return canonical
        # Fall through to the catalog's did-you-mean error.
        self._catalog.get_flood_type(name.strip())
        raise AssertionError("unreachable")  # pragma: no cover

    def _create_grid(self, lat_lim: list[float], lon_lim: list[float]) -> SpatialExtent:
        """Capture the requested bounds as a :class:`SpatialExtent`.

        Args:
            lat_lim: `[min_lat, max_lat]`.
            lon_lim: `[min_lon, max_lon]`.

        Returns:
            SpatialExtent: The requested extent.
        """
        return SpatialExtent.from_pairs(lat_lim=lat_lim, lon_lim=lon_lim)

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

        HANZE covers a whole record and is indexed by event year, so `None`
        bounds are legal and yield a `None`-dated extent.

        Args:
            start: Inclusive start date string, or `None`.
            end: Inclusive end date string, or `None`.
            temporal_resolution: Recorded as the resolution label.
            fmt: `strptime` format tried first for a string bound; a non-matching
                string falls back to an ISO-8601 parse.

        Returns:
            TemporalExtent: Frozen model with the parsed (or `None`) endpoints.

        Raises:
            ValueError: If `start` parses to a date later than `end`.
        """
        start_dt = to_datetime(start, fmt) if start else None
        end_dt = to_datetime(end, fmt) if end else None
        dates = (
            pd.DatetimeIndex([start_dt, end_dt])
            if start_dt is not None and end_dt is not None
            else pd.DatetimeIndex([])
        )
        return TemporalExtent(
            start_date=start_dt,
            end_date=end_dt,
            resolution=temporal_resolution,
            dates=dates,
        )

    @property
    def _year_range(self) -> tuple[int | None, int | None]:
        """Return the requested window as inclusive year bounds.

        Returns:
            tuple[int | None, int | None]: `(first_year, last_year)`, either
                `None` when that end of the window was not given.
        """
        start = self.time.start_date
        end = self.time.end_date
        return (
            start.year if start is not None else None,
            end.year if end is not None else None,
        )

    @property
    def _bbox(self) -> tuple[float, float, float, float] | None:
        """Return the request bbox, or `None` for a Europe-wide request.

        Returns:
            tuple[float, float, float, float] | None:
                `(min_lon, min_lat, max_lon, max_lat)`, or `None` when the
                request covers the whole globe (so no region is dropped).
        """
        space = self.space
        whole_globe = (
            space.latitude_min <= _GLOBAL_LAT[0]
            and space.latitude_max >= _GLOBAL_LAT[1]
            and space.longitude_min <= _GLOBAL_LON[0]
            and space.longitude_max >= _GLOBAL_LON[1]
        )
        if whole_globe:
            return None
        return (
            space.longitude_min,
            space.latitude_min,
            space.longitude_max,
            space.latitude_max,
        )

    def _client(self) -> HttpClient:
        """Return this instance's pooled client, building it on first use.

        Zenodo is a single origin; a dropped connection there is a normal event,
        so connection and timeout errors are retried too (matching `emdat`).

        Returns:
            HttpClient: The same instance on every later call.
        """
        if self._http is None:
            self._http = HttpClient(
                timeout=self._timeout,
                retry_on_exceptions=(requests.ConnectionError, requests.Timeout),
            )
        return self._http

    def _download_file(
        self, key: str, *, expect_magic: bytes | tuple[bytes, ...] | None = None
    ) -> Path:
        """Download one catalog file into `root_dir`, reusing a cached copy.

        Args:
            key: A logical file key (`"events"`, `"regions"`, or `"region_names"`).
            expect_magic: Optional leading-byte guard (one prefix or a tuple of
                acceptable prefixes) rejecting an error page served with a 200
                status.

        Returns:
            Path: The local file in `root_dir`.
        """
        record = self._record.record
        entry = self._catalog.file(key)
        local = self.root_dir / entry.name
        if not local.exists():
            logger.info(f"HANZE: downloading {entry.name} (record {record}).")
            self._client().download(
                entry.content_url(record),
                local,
                expect_magic=expect_magic,
                progress=self._progress,
            )
        return local

    def _load_events(self) -> pd.DataFrame:
        """Download and parse the HANZE events / impacts CSV.

        The download is guarded by the events header's leading bytes, so an HTML
        error page served with a `200` status (a proxy / CDN hiccup) is rejected
        at the download site rather than cached under the CSV name and failing
        confusingly at `read_csv` on every later call — the same guard the region
        zip (`PK`) and the sibling `emdat` xlsx (`PK`) use. The live file is
        published with a UTF-8 BOM, so both the BOM-prefixed and the plain header
        are accepted; `utf-8-sig` then strips the BOM so the `ID` column name is
        clean.

        Returns:
            pandas.DataFrame: The full events table, HANZE's documented headers.
        """
        local = self._download_file("events", expect_magic=(b"\xef\xbb\xbfID,", b"ID,"))
        return pd.read_csv(local, encoding="utf-8-sig")

    def _load_regions(self) -> FeatureCollection:
        """Download, extract and read the NUTS-3 region shapefile (cached).

        Returns:
            FeatureCollection: The region polygons in the shapefile's stored CRS
                (`EPSG:3035`).
        """
        if self._regions_fc is not None:
            return self._regions_fc
        from pyramids.feature.collection import FeatureCollection

        archive = self._download_file("regions", expect_magic=b"PK")
        stem = self._geo.member_stem
        members = extract_members(
            archive,
            self.root_dir / "hanze_regions",
            include=(".shp", ".shx", ".dbf", ".prj", ".cpg"),
        )
        shp = next(
            (m for m in members if m.stem == stem and m.suffix.lower() == ".shp"),
            None,
        )
        if shp is None:
            raise ValueError(
                f"the HANZE region archive {archive.name} has no "
                f"{stem}.shp member (found {[m.name for m in members]})."
            )
        self._regions_fc = FeatureCollection.read_file(str(shp))
        return self._regions_fc

    def _bbox_region_codes(self) -> set[str] | None:
        """Return the NUTS-3 codes whose region intersects the request bbox.

        Loads the region geometry (in `EPSG:3035`), reprojects to WGS84, and
        selects the polygons intersecting the bbox. `None` when the request is
        Europe-wide (no bbox restriction).

        Returns:
            set[str] | None: The in-bbox NUTS-3 codes, or `None` for a whole-globe
                request.
        """
        bbox = self._bbox
        if bbox is None:
            return None
        regions = self._load_regions().to_crs(geometry_module.OUTPUT_CRS)
        join_field = self._geo.join_field
        min_lon, min_lat, max_lon, max_lat = bbox
        within = regions.cx[min_lon:max_lon, min_lat:max_lat]
        # Upper-cased here (like `self._region`) so `_row_matches_codes` compares
        # already-normalised sets rather than re-casing them per event row.
        return set(within[join_field].astype(str).str.upper())

    def _filter_events(self, events: pd.DataFrame) -> pd.DataFrame:
        """Apply the request's country / region / type / date / bbox filters.

        Args:
            events: The full events table.

        Returns:
            pandas.DataFrame: The matching rows, index reset.
        """
        columns = self._catalog.columns
        mask = pd.Series(True, index=events.index)

        if self._country:
            mask &= (
                events[columns["country_code"]]
                .astype(str)
                .str.upper()
                .isin(self._country)
            )
        if self._flood_types:
            mask &= events[columns["type"]].isin(self._flood_types)

        first_year, last_year = self._year_range
        if first_year is not None:
            mask &= events[columns["year"]] >= first_year
        if last_year is not None:
            mask &= events[columns["year"]] <= last_year

        # An explicit `region=` restriction only counts when non-empty; the
        # bbox-derived set counts whenever a bbox is set, even if it resolves to
        # no regions (a bbox over open water legitimately drops every event).
        code_filters: list[set[str]] = []
        if self._region:
            code_filters.append(self._region)
        bbox_codes = self._bbox_region_codes()
        if bbox_codes is not None:
            code_filters.append(bbox_codes)
        if code_filters:
            mask &= events[columns["regions_nuts3"]].apply(
                lambda cell: self._row_matches_codes(cell, code_filters)
            )

        return events[mask].reset_index(drop=True)

    @staticmethod
    def _row_matches_codes(cell: object, code_filters: list[set[str]]) -> bool:
        """Whether an event's affected-region list satisfies every code filter.

        Args:
            cell: One `Regions affected (NUTS 3)` cell.
            code_filters: One set per active restriction (explicit `region=`, and
                the bbox-derived codes); the event must intersect **each**.

        Returns:
            bool: `True` when the event's codes intersect every filter set.
        """
        # The filter sets are already upper-cased at construction (`self._region`)
        # and in `_bbox_region_codes`, so only the row's codes need normalising.
        row_codes = {code.upper() for code in geometry_module.split_nuts3(cell)}
        return all(bool(row_codes & codes) for codes in code_filters)

    def _search(self) -> list[RemoteProduct]:
        """Pin the one product to fetch (the HANZE events table).

        Returns:
            list[RemoteProduct]: A single product carrying the record id.
        """
        return [
            RemoteProduct(
                id="hanze:events",
                metadata={"record": self._record.record},
            )
        ]

    def _fetch(self, products: list[RemoteProduct]) -> list[Any]:
        """Download, filter, and shape the one product to the instance's kind.

        Args:
            products: The single-element list from :meth:`_search`.

        Returns:
            list[Any]: One element — a filtered :class:`pandas.DataFrame`
                (tabular), or a
                :class:`~pyramids.feature.collection.FeatureCollection` of the
                affected regions (vector).
        """
        events = self._filter_events(self._load_events())
        if self._with_geometry:
            return [self._build_region_collection(events)]
        return [events]

    def _build_region_collection(self, events: pd.DataFrame) -> FeatureCollection:
        """Join the filtered events to their affected NUTS-3 region polygons.

        When the request carries a bbox, the joined regions are additionally
        restricted to it by bounding-box intersection (`GeoDataFrame.cx`): an
        event that touched an in-bbox region also lists regions outside the box,
        and returning those would put polygons well outside a spatial query on
        the map. A region whose extent intersects the box is kept **whole** (it
        is selected, not geometrically trimmed), so the vector answer stays
        "affected regions within the box", matching the tabular bbox path.

        Args:
            events: The filtered events table.

        Returns:
            FeatureCollection: One polygon per affected region (restricted to the
                regions intersecting the bbox when one is set), CRS `EPSG:4326`.
        """
        regions = self._load_regions()
        geometry = self._geo
        collection = geometry_module.join_events_to_regions(
            events,
            regions,
            regions_column=self._catalog.columns["regions_nuts3"],
            join_field=geometry.join_field,
            name_field=geometry.name_field,
        )
        bbox = self._bbox
        if bbox is None or not len(collection):
            return collection
        from pyramids.feature.collection import FeatureCollection

        min_lon, min_lat, max_lon, max_lat = bbox
        within = collection.cx[min_lon:max_lon, min_lat:max_lat]
        return FeatureCollection(within.reset_index(drop=True))

    def _api(self) -> list[Any]:
        """Compose :meth:`_search` and :meth:`_fetch`.

        Returns:
            list[Any]: The fetched result (one element).
        """
        return self._api_via_search_fetch()

    def download(self, progress_bar: bool = True) -> pd.DataFrame | FeatureCollection:
        """Fetch HANZE and return the per-instance shape.

        Runs the download + filter, writes the result to `path` (a CSV for the
        tabular default, a GeoPackage for `with_geometry`), and returns it.

        Args:
            progress_bar: Whether to draw a download progress bar. Passed through
                to the transport, so `False` really does silence it.

        Returns:
            A :class:`pandas.DataFrame` of events + impacts (the default), or a
            :class:`~pyramids.feature.collection.FeatureCollection` of the
            affected NUTS-3 regions (`with_geometry=True`). Both are also written
            under `root_dir`.

        Raises:
            requests.HTTPError: If a Zenodo download returns a non-2xx status.
            ValueError: If a download's body fails its content guard (an HTML
                error page served with a 200 status), or `with_geometry=True` and
                the region archive has no `<member_stem>.shp` member.
        """
        self._progress = progress_bar
        results = self._api()
        # `_search` always yields one product, so `_fetch` returns a single
        # element (a 0-row DataFrame / empty FC is still one element); the
        # `_empty_result()` fallback is a defensive guard for a future `_search`
        # that could return nothing, not a path this backend reaches today.
        result = results[0] if results else self._empty_result()
        self._log_citation()

        if self.OUTPUT_KIND == "vector":
            out_path = self.root_dir / (self._result_stem("hanze_regions") + ".gpkg")
            # Written unconditionally — an empty result still writes a schema-only
            # GeoPackage, so the vector path matches the tabular one (which always
            # writes a header-only CSV) and a caller globbing `path` finds a file.
            result.to_file(str(out_path), driver="GPKG")
            logger.info(
                f"HANZE: {len(result)} affected region(s) written to {out_path}."
            )
            return result

        out_path = self.root_dir / (self._result_stem("hanze_events") + ".csv")
        result.to_csv(out_path, index=False)
        logger.info(f"HANZE: {len(result)} event(s) written to {out_path}.")
        return result

    def _empty_result(self) -> pd.DataFrame | FeatureCollection:
        """Return the empty result matching the instance's output kind."""
        if self._with_geometry:
            return geometry_module.empty_region_fc()
        return pd.DataFrame()

    def _result_stem(self, base: str) -> str:
        """Compose an output file stem that encodes the request's filters.

        A plain `base` for an unfiltered request; otherwise `base-<digest>` so
        two differently-filtered queries into one `path=` do not overwrite each
        other. The digest is order-insensitive in the multi-value filters.

        Args:
            base: The stem prefix (`"hanze_events"` / `"hanze_regions"`).

        Returns:
            str: `base`, or `base-<8-hex-digest>` when any filter is active.
        """
        first_year, last_year = self._year_range
        applied = (
            bool(self._country),
            bool(self._region),
            bool(self._flood_types),
            first_year is not None,
            last_year is not None,
            self._bbox is not None,
        )
        if not any(applied):
            return base
        request = (
            tuple(sorted(self._country)),
            tuple(sorted(self._region)),
            tuple(sorted(self._flood_types)),
            self._year_range,
            self._bbox,
        )
        digest = hashlib.sha1(
            repr(request).encode(), usedforsecurity=False
        ).hexdigest()[:8]
        return f"{base}-{digest}"

    def _log_citation(self) -> None:
        """Log the CC-BY attribution once (info, not a warning)."""
        record = self._record
        if record.attribution:
            logger.info(f"HANZE source citation: {record.attribution}")

__init__(start=None, end=None, lat_lim=None, lon_lim=None, temporal_resolution='all', path=None, fmt='%Y-%m-%d', country=None, region=None, flood_type=None, with_geometry=False, timeout=120.0)

Initialise a HANZE backend instance.

Parameters:

Name	Type	Description	Default
start	str | None	Inclusive start of an optional window, parsed with fmt. Only its year is significant — HANZE indexes events by year. None means "from the beginning of the record".	None
end	str | None	Inclusive end of the optional window; None means "to the end of the record".	None
lat_lim	list[float] | None	[lat_min, lat_max] bounding-box latitudes in degrees. A non-global box selects the affected regions (and, on the tabular path, the events touching them) that intersect it, which loads the region geometry.	None
lon_lim	list[float] | None	[lon_min, lon_max] bounding-box longitudes in degrees.	None
temporal_resolution	str	HANZE issues one query over the whole window, so this is the sentinel "all", not a pandas frequency alias.	'all'
path	Path | str | None	Output directory for the cached source files and the written table / vector file. Created by the parent class if absent.	None
fmt	str	strptime format for start / end.	'%Y-%m-%d'
country	str | list[str] | None	One ISO2 country code or a list of them ("DE", ["DE", "NL"]). None keeps every country.	None
region	str | list[str] | None	One NUTS-3 code or a list of them ("DE300"), matched against each event's affected-region list. None keeps every region.	None
flood_type	str | list[str] | None	One flood type or a list of them — any of "River", "Flash", "Coastal", "River/Coastal". None keeps every type.	None
with_geometry	bool	When True, additionally download the NUTS-3 region shapefile and return a FeatureCollection of the affected regions instead of the events DataFrame (sets OUTPUT_KIND="vector" for this instance).	False
timeout	float	Per-request timeout in seconds for the Zenodo downloads.	120.0

Raises:

Type	Description
ValueError	If a country value is not a 2-letter ISO2 code, a region value is not a 5-character NUTS-3 code, or a flood_type value is not a registered HANZE flood type.

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

def __init__(
    self,
    start: str | None = None,
    end: str | None = None,
    lat_lim: list[float] | None = None,
    lon_lim: list[float] | None = None,
    temporal_resolution: str = "all",
    path: Path | str | None = None,
    fmt: str = "%Y-%m-%d",
    country: str | list[str] | None = None,
    region: str | list[str] | None = None,
    flood_type: str | list[str] | None = None,
    with_geometry: bool = False,
    timeout: float = 120.0,
):
    """Initialise a HANZE backend instance.

    Args:
        start: Inclusive start of an optional window, parsed with `fmt`. Only
            its year is significant — HANZE indexes events by year. `None`
            means "from the beginning of the record".
        end: Inclusive end of the optional window; `None` means "to the end
            of the record".
        lat_lim: `[lat_min, lat_max]` bounding-box latitudes in degrees. A
            non-global box selects the affected regions (and, on the tabular
            path, the events touching them) that intersect it, which loads
            the region geometry.
        lon_lim: `[lon_min, lon_max]` bounding-box longitudes in degrees.
        temporal_resolution: HANZE issues one query over the whole window, so
            this is the sentinel `"all"`, not a pandas frequency alias.
        path: Output directory for the cached source files and the written
            table / vector file. Created by the parent class if absent.
        fmt: `strptime` format for `start` / `end`.
        country: One ISO2 country code or a list of them (`"DE"`,
            `["DE", "NL"]`). `None` keeps every country.
        region: One NUTS-3 code or a list of them (`"DE300"`), matched
            against each event's affected-region list. `None` keeps every
            region.
        flood_type: One flood type or a list of them — any of `"River"`,
            `"Flash"`, `"Coastal"`, `"River/Coastal"`. `None` keeps every
            type.
        with_geometry: When `True`, additionally download the NUTS-3 region
            shapefile and return a `FeatureCollection` of the affected
            regions instead of the events `DataFrame` (sets
            `OUTPUT_KIND="vector"` for this instance).
        timeout: Per-request timeout in seconds for the Zenodo downloads.

    Raises:
        ValueError: If a `country` value is not a 2-letter ISO2 code, a
            `region` value is not a 5-character NUTS-3 code, or a
            `flood_type` value is not a registered HANZE flood type.
    """
    self._catalog = Catalog()
    # Resolve the always-loaded record / geometry blocks into non-optional
    # attributes once, so the rest of the backend (and the type checker) can
    # read them without re-guarding the catalog's `... | None` fields.
    if self._catalog.record is None or self._catalog.geometry is None:
        raise ValueError(
            "the HANZE catalog failed to load its 'record:'/'geometry:' "
            "block; the bundled hanze_data_catalog.yaml is malformed."
        )
    self._record = self._catalog.record
    self._geo = self._catalog.geometry
    self._country = _normalize_country(country)
    self._region = _normalize_region(region)
    # Validate flood types against the catalog (did-you-mean hint on a typo).
    self._flood_types = [
        self._resolve_flood_type(name) for name in _as_list(flood_type)
    ]
    self._with_geometry = with_geometry
    self._timeout = timeout
    self._http: HttpClient | None = None
    self._regions_fc: FeatureCollection | None = None

    self.OUTPUT_KIND = "vector" if with_geometry else "tabular"

    super().__init__(
        start=cast("str", start),
        end=cast("str", end),
        # HANZE is facet-only: it is a single product selected by
        # country=/region=/flood_type=, so it declares no `variables`
        # parameter (the facade neither requires nor forwards one). The base
        # class still wants the argument, so an empty list is passed here.
        variables=[],
        temporal_resolution=temporal_resolution,
        lat_lim=lat_lim if lat_lim is not None else _GLOBAL_LAT,
        lon_lim=lon_lim if lon_lim is not None else _GLOBAL_LON,
        fmt=fmt,
        path=path,
    )

download(progress_bar=True)

Fetch HANZE and return the per-instance shape.

Runs the download + filter, writes the result to path (a CSV for the tabular default, a GeoPackage for with_geometry), and returns it.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Whether to draw a download progress bar. Passed through to the transport, so False really does silence it.	True

Returns:

Name	Type	Description
A	DataFrame | FeatureCollection	class:pandas.DataFrame of events + impacts (the default), or a
	DataFrame | FeatureCollection	class:~pyramids.feature.collection.FeatureCollection of the
	DataFrame | FeatureCollection	affected NUTS-3 regions (with_geometry=True). Both are also written
	DataFrame | FeatureCollection	under root_dir.

Raises:

Type	Description
HTTPError	If a Zenodo download returns a non-2xx status.
ValueError	If a download's body fails its content guard (an HTML error page served with a 200 status), or with_geometry=True and the region archive has no <member_stem>.shp member.

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

def download(self, progress_bar: bool = True) -> pd.DataFrame | FeatureCollection:
    """Fetch HANZE and return the per-instance shape.

    Runs the download + filter, writes the result to `path` (a CSV for the
    tabular default, a GeoPackage for `with_geometry`), and returns it.

    Args:
        progress_bar: Whether to draw a download progress bar. Passed through
            to the transport, so `False` really does silence it.

    Returns:
        A :class:`pandas.DataFrame` of events + impacts (the default), or a
        :class:`~pyramids.feature.collection.FeatureCollection` of the
        affected NUTS-3 regions (`with_geometry=True`). Both are also written
        under `root_dir`.

    Raises:
        requests.HTTPError: If a Zenodo download returns a non-2xx status.
        ValueError: If a download's body fails its content guard (an HTML
            error page served with a 200 status), or `with_geometry=True` and
            the region archive has no `<member_stem>.shp` member.
    """
    self._progress = progress_bar
    results = self._api()
    # `_search` always yields one product, so `_fetch` returns a single
    # element (a 0-row DataFrame / empty FC is still one element); the
    # `_empty_result()` fallback is a defensive guard for a future `_search`
    # that could return nothing, not a path this backend reaches today.
    result = results[0] if results else self._empty_result()
    self._log_citation()

    if self.OUTPUT_KIND == "vector":
        out_path = self.root_dir / (self._result_stem("hanze_regions") + ".gpkg")
        # Written unconditionally — an empty result still writes a schema-only
        # GeoPackage, so the vector path matches the tabular one (which always
        # writes a header-only CSV) and a caller globbing `path` finds a file.
        result.to_file(str(out_path), driver="GPKG")
        logger.info(
            f"HANZE: {len(result)} affected region(s) written to {out_path}."
        )
        return result

    out_path = self.root_dir / (self._result_stem("hanze_events") + ".csv")
    result.to_csv(out_path, index=False)
    logger.info(f"HANZE: {len(result)} event(s) written to {out_path}.")
    return result

earthlens.hanze.catalog

Catalog for the HANZE historical-flood-impacts backend.

HANZE is a single tabular product — the database of observed European flood events and their impacts (Paprotny et al.) — published as individual small files on a pinned Zenodo version record. This module is the bridge between the friendly request vocabulary (type="River", country="DE") and what the release actually ships: the pinned record, the per-file names, the flood-type vocabulary, the friendly-name -> CSV-header map, and the region-geometry join configuration.

Four shapes are modelled, all frozen:

• :class:ZenodoRecord — the pinned version record, its concept DOI, version, data_period, licence and attribution. Pinning a version rather than the moving concept DOI is what makes a request reproducible.
• :class:HanzeFile — one downloadable Zenodo object (its name, and the REST content url composed from the pinned record). HANZE ships small individual files, so each is a direct download, never a range-read.
• :class:FloodType — one row of the Type vocabulary (River, Flash, Coastal, River/Coastal). These are the catalog's dict-surface rows, keyed by type under the inherited :attr:datasets field.
• :class:GeometryJoin — the region-shapefile join: its member stem, the join field (Code), the name field, and the shapefile CRS (EPSG:3035).

:class:Catalog is a thin :class:earthlens.base.AbstractCatalog subclass that loads the bundled hanze_data_catalog.yaml through the shared :func:~earthlens.base.catalog_source.load_catalog (with a CatalogParseCache), mirroring gdacs/catalog.py. :data:CATALOG_PATH is the path to the bundled YAML and is monkey-patchable in tests.

Catalog

Bases: AbstractCatalog

Catalog for the HANZE backend.

Reads the bundled hanze_data_catalog.yaml (shipped as package data) and exposes the pinned Zenodo record, the per-file names, the flood-Type vocabulary (as :class:FloodType rows keyed by type under the inherited :attr:datasets field — the cat["River"] / "River" in cat / len(cat) dict surface), the friendly-name -> CSV-header map, and the region-geometry join configuration. Instantiate with no arguments (Catalog()).

Attributes:

Name	Type	Description
datasets	dict[str, FloodType]	Map from a flood-Type string to its :class:FloodType row.
record	ZenodoRecord | None	The pinned :class:ZenodoRecord.
files	dict[str, HanzeFile]	Map from a logical key ("events", "regions", "region_names") to its :class:HanzeFile.
geometry	GeometryJoin | None	The :class:GeometryJoin for the region attach.
columns	dict[str, str]	Friendly name -> exact HANZE CSV header.

Examples:

• List the flood types and resolve one, and read the pinned record:

  >>> from earthlens.hanze import Catalog
  >>> cat = Catalog()
  >>> cat.flood_types()
  ['Coastal', 'Flash', 'River', 'River/Coastal']
  >>> cat.get_flood_type("River").description
  'Riverine (fluvial) floods.'
  >>> "Coastal" in cat
  True
  >>> cat.column("country_code")
  'Country code'
  

• An unknown flood type raises with a did-you-mean hint:

  >>> from earthlens.hanze import Catalog
  >>> Catalog().get_flood_type("Rivers")
  Traceback (most recent call last):
      ...
  ValueError: 'Rivers' is not in the HANZE catalog. Known flood types: ['Coastal', 'Flash', 'River', 'River/Coastal']. Did you mean 'River'?
  

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

class Catalog(AbstractCatalog):
    """Catalog for the HANZE backend.

    Reads the bundled `hanze_data_catalog.yaml` (shipped as package data) and
    exposes the pinned Zenodo record, the per-file names, the flood-`Type`
    vocabulary (as :class:`FloodType` rows keyed by type under the inherited
    :attr:`datasets` field — the `cat["River"]` / `"River" in cat` / `len(cat)`
    dict surface), the friendly-name -> CSV-header map, and the region-geometry
    join configuration. Instantiate with no arguments (`Catalog()`).

    Attributes:
        datasets: Map from a flood-`Type` string to its :class:`FloodType` row.
        record: The pinned :class:`ZenodoRecord`.
        files: Map from a logical key (`"events"`, `"regions"`,
            `"region_names"`) to its :class:`HanzeFile`.
        geometry: The :class:`GeometryJoin` for the region attach.
        columns: Friendly name -> exact HANZE CSV header.

    Examples:
        - List the flood types and resolve one, and read the pinned record:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> cat = Catalog()
            >>> cat.flood_types()
            ['Coastal', 'Flash', 'River', 'River/Coastal']
            >>> cat.get_flood_type("River").description
            'Riverine (fluvial) floods.'
            >>> "Coastal" in cat
            True
            >>> cat.column("country_code")
            'Country code'

            ```
        - An unknown flood type raises with a did-you-mean hint:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> Catalog().get_flood_type("Rivers")
            Traceback (most recent call last):
                ...
            ValueError: 'Rivers' is not in the HANZE catalog. Known flood types: ['Coastal', 'Flash', 'River', 'River/Coastal']. Did you mean 'River'?

            ```
    """

    _catalog_kind: str = "HANZE catalog"
    _entry_noun: str = "flood types"

    datasets: dict[str, FloodType] = Field(default_factory=dict)
    #: `record` / `geometry` default to `None` rather than a placeholder model:
    #: the base :meth:`model_post_init` autoload fills a field only when its
    #: current value is falsy, and a placeholder model instance is truthy, so it
    #: would be skipped and `Catalog()` would keep the placeholder. `None` is
    #: falsy, so the bundled record / geometry load as intended. `load()` and a
    #: test passing `datasets=` + these fields populate them directly.
    record: ZenodoRecord | None = Field(default=None, repr=False)
    files: dict[str, HanzeFile] = Field(default_factory=dict, repr=False)
    geometry: GeometryJoin | None = Field(default=None, repr=False)
    columns: dict[str, str] = Field(default_factory=dict, repr=False)

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

        Returns:
            dict[str, Any]: The full field payload read from the bundled catalog.
        """
        return dict(_parse_catalog([CATALOG_PATH]))

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

        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 `catalog_path` does not exist, a required block is
                missing, or a row fails validation.

        Examples:
            - Loading the bundled catalog yields the pinned record and types:
                ```python
                >>> from earthlens.hanze import Catalog
                >>> cat = Catalog.load()
                >>> cat.record.record
                20478847
                >>> cat.flood_types()
                ['Coastal', 'Flash', 'River', 'River/Coastal']

                ```
        """
        catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
        parsed = load_catalog(
            catalog_path, _CATALOG_CACHE, _parse_catalog, provider="HANZE"
        )
        return cls(**parsed)

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

        Returns:
            dict[str, FloodType]: Same object as :attr:`datasets`.

        Examples:
            - The flood-type map is keyed by the `Type` string:
                ```python
                >>> from earthlens.hanze import Catalog
                >>> sorted(Catalog().get_catalog())
                ['Coastal', 'Flash', 'River', 'River/Coastal']

                ```
        """
        return self.datasets

    def get_flood_type(self, flood_type: str) -> FloodType:
        """Return the :class:`FloodType` for `flood_type`, with a did-you-mean hint.

        Thin alias over :meth:`~earthlens.base.AbstractCatalog.get_dataset`.

        Args:
            flood_type: A HANZE flood-`Type` string (`"River"`, `"Coastal"`,
                `"Flash"`, `"River/Coastal"`).

        Returns:
            FloodType: The matching row.

        Raises:
            ValueError: If `flood_type` is not a registered flood type.

        Examples:
            - Resolve a type and read its description:
                ```python
                >>> from earthlens.hanze import Catalog
                >>> Catalog().get_flood_type("Coastal").description
                'Coastal (storm-surge) floods.'

                ```
        """
        return cast("FloodType", self.get_dataset(flood_type))

    def flood_types(self) -> list[str]:
        """Return the registered flood-`Type` strings, sorted.

        Returns:
            list[str]: The flood types
                (`["Coastal", "Flash", "River", "River/Coastal"]`).

        Examples:
            - The registered types come back sorted:
                ```python
                >>> from earthlens.hanze import Catalog
                >>> Catalog().flood_types()
                ['Coastal', 'Flash', 'River', 'River/Coastal']

                ```
        """
        return sorted(self.datasets)

    def file(self, key: str) -> HanzeFile:
        """Return the :class:`HanzeFile` for a logical key.

        Args:
            key: `"events"`, `"regions"`, or `"region_names"`.

        Returns:
            HanzeFile: The matching file descriptor.

        Raises:
            KeyError: If `key` is not a known file.

        Examples:
            - Resolve the events and region file names:
                ```python
                >>> from earthlens.hanze import Catalog
                >>> cat = Catalog()
                >>> cat.file("events").name
                'HANZE_events_v3_0_1b.csv'
                >>> cat.file("regions").name
                'Regions_v2024_simplified.zip'

                ```
        """
        return self.files[key]

    def column(self, friendly: str) -> str:
        """Return the exact HANZE CSV header for a friendly column name.

        Args:
            friendly: A friendly key from the catalog's `columns:` map
                (`"country_code"`, `"type"`, `"regions_nuts3"`, ...).

        Returns:
            str: The exact CSV header (`"Country code"`).

        Raises:
            KeyError: If `friendly` is not a mapped column.

        Examples:
            - Map friendly keys to their exact HANZE headers:
                ```python
                >>> from earthlens.hanze import Catalog
                >>> cat = Catalog()
                >>> cat.column("country_code")
                'Country code'
                >>> cat.column("regions_nuts3")
                'Regions affected (NUTS 3)'

                ```
        """
        return self.columns[friendly]

column(friendly)

Return the exact HANZE CSV header for a friendly column name.

Parameters:

Name	Type	Description	Default
friendly	str	A friendly key from the catalog's columns: map ("country_code", "type", "regions_nuts3", ...).	required

Returns:

Name	Type	Description
str	str	The exact CSV header ("Country code").

Raises:

Type	Description
KeyError	If friendly is not a mapped column.

Examples:

• Map friendly keys to their exact HANZE headers:

  >>> from earthlens.hanze import Catalog
  >>> cat = Catalog()
  >>> cat.column("country_code")
  'Country code'
  >>> cat.column("regions_nuts3")
  'Regions affected (NUTS 3)'
  

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

def column(self, friendly: str) -> str:
    """Return the exact HANZE CSV header for a friendly column name.

    Args:
        friendly: A friendly key from the catalog's `columns:` map
            (`"country_code"`, `"type"`, `"regions_nuts3"`, ...).

    Returns:
        str: The exact CSV header (`"Country code"`).

    Raises:
        KeyError: If `friendly` is not a mapped column.

    Examples:
        - Map friendly keys to their exact HANZE headers:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> cat = Catalog()
            >>> cat.column("country_code")
            'Country code'
            >>> cat.column("regions_nuts3")
            'Regions affected (NUTS 3)'

            ```
    """
    return self.columns[friendly]

file(key)

Return the :class:HanzeFile for a logical key.

Parameters:

Name	Type	Description	Default
key	str	"events", "regions", or "region_names".	required

Returns:

Name	Type	Description
HanzeFile	HanzeFile	The matching file descriptor.

Raises:

Type	Description
KeyError	If key is not a known file.

Examples:

• Resolve the events and region file names:

  >>> from earthlens.hanze import Catalog
  >>> cat = Catalog()
  >>> cat.file("events").name
  'HANZE_events_v3_0_1b.csv'
  >>> cat.file("regions").name
  'Regions_v2024_simplified.zip'
  

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

def file(self, key: str) -> HanzeFile:
    """Return the :class:`HanzeFile` for a logical key.

    Args:
        key: `"events"`, `"regions"`, or `"region_names"`.

    Returns:
        HanzeFile: The matching file descriptor.

    Raises:
        KeyError: If `key` is not a known file.

    Examples:
        - Resolve the events and region file names:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> cat = Catalog()
            >>> cat.file("events").name
            'HANZE_events_v3_0_1b.csv'
            >>> cat.file("regions").name
            'Regions_v2024_simplified.zip'

            ```
    """
    return self.files[key]

flood_types()

Return the registered flood-Type strings, sorted.

Returns:

Type	Description
list[str]	list[str]: The flood types (["Coastal", "Flash", "River", "River/Coastal"]).

Examples:

• The registered types come back sorted:

  >>> from earthlens.hanze import Catalog
  >>> Catalog().flood_types()
  ['Coastal', 'Flash', 'River', 'River/Coastal']
  

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

def flood_types(self) -> list[str]:
    """Return the registered flood-`Type` strings, sorted.

    Returns:
        list[str]: The flood types
            (`["Coastal", "Flash", "River", "River/Coastal"]`).

    Examples:
        - The registered types come back sorted:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> Catalog().flood_types()
            ['Coastal', 'Flash', 'River', 'River/Coastal']

            ```
    """
    return sorted(self.datasets)

get_catalog()

Return the flood-type map (satisfies the abstract contract).

Returns:

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

Examples:

• The flood-type map is keyed by the Type string:

  >>> from earthlens.hanze import Catalog
  >>> sorted(Catalog().get_catalog())
  ['Coastal', 'Flash', 'River', 'River/Coastal']
  

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

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

    Returns:
        dict[str, FloodType]: Same object as :attr:`datasets`.

    Examples:
        - The flood-type map is keyed by the `Type` string:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> sorted(Catalog().get_catalog())
            ['Coastal', 'Flash', 'River', 'River/Coastal']

            ```
    """
    return self.datasets

get_flood_type(flood_type)

Return the :class:FloodType for flood_type, with a did-you-mean hint.

Thin alias over :meth:~earthlens.base.AbstractCatalog.get_dataset.

Parameters:

Name	Type	Description	Default
flood_type	str	A HANZE flood-Type string ("River", "Coastal", "Flash", "River/Coastal").	required

Returns:

Name	Type	Description
FloodType	FloodType	The matching row.

Raises:

Type	Description
ValueError	If flood_type is not a registered flood type.

Examples:

• Resolve a type and read its description:

  >>> from earthlens.hanze import Catalog
  >>> Catalog().get_flood_type("Coastal").description
  'Coastal (storm-surge) floods.'
  

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

def get_flood_type(self, flood_type: str) -> FloodType:
    """Return the :class:`FloodType` for `flood_type`, with a did-you-mean hint.

    Thin alias over :meth:`~earthlens.base.AbstractCatalog.get_dataset`.

    Args:
        flood_type: A HANZE flood-`Type` string (`"River"`, `"Coastal"`,
            `"Flash"`, `"River/Coastal"`).

    Returns:
        FloodType: The matching row.

    Raises:
        ValueError: If `flood_type` is not a registered flood type.

    Examples:
        - Resolve a type and read its description:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> Catalog().get_flood_type("Coastal").description
            'Coastal (storm-surge) floods.'

            ```
    """
    return cast("FloodType", self.get_dataset(flood_type))

load(catalog_path=None) classmethod

Read the HANZE catalog from disk.

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 catalog_path does not exist, a required block is missing, or a row fails validation.

Examples:

• Loading the bundled catalog yields the pinned record and types:

  >>> from earthlens.hanze import Catalog
  >>> cat = Catalog.load()
  >>> cat.record.record
  20478847
  >>> cat.flood_types()
  ['Coastal', 'Flash', 'River', 'River/Coastal']
  

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

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

    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 `catalog_path` does not exist, a required block is
            missing, or a row fails validation.

    Examples:
        - Loading the bundled catalog yields the pinned record and types:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> cat = Catalog.load()
            >>> cat.record.record
            20478847
            >>> cat.flood_types()
            ['Coastal', 'Flash', 'River', 'River/Coastal']

            ```
    """
    catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
    parsed = load_catalog(
        catalog_path, _CATALOG_CACHE, _parse_catalog, provider="HANZE"
    )
    return cls(**parsed)

FloodType

Bases: BaseModel

One entry of the HANZE flood-Type vocabulary.

The type string ("River", "River/Coastal") is the parent key in :attr:Catalog.datasets and is not stored on the row.

Attributes:

Name	Type	Description
description	str	Short note on what the flood type covers.

Examples:

• Build a row directly:

  >>> from earthlens.hanze import FloodType
  >>> FloodType(description="Riverine (fluvial) floods.").description
  'Riverine (fluvial) floods.'
  

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

class FloodType(BaseModel):
    """One entry of the HANZE flood-`Type` vocabulary.

    The type string (`"River"`, `"River/Coastal"`) is the parent key in
    :attr:`Catalog.datasets` and is not stored on the row.

    Attributes:
        description: Short note on what the flood type covers.

    Examples:
        - Build a row directly:
            ```python
            >>> from earthlens.hanze import FloodType
            >>> FloodType(description="Riverine (fluvial) floods.").description
            'Riverine (fluvial) floods.'

            ```
    """

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

    description: str = ""

GeometryJoin

Bases: BaseModel

The region-shapefile join configuration for with_geometry.

Attributes:

Name	Type	Description
member_stem	str	The shapefile member stem inside the region zip ("NUTS3_regions_v2024_simplified"); the .shp and its sidecars share it.
join_field	str	The shapefile attribute holding the NUTS-3 code ("Code"), joined to the semicolon-split Regions affected (NUTS 3) list.
name_field	str	The shapefile attribute holding the region name ("Name").
crs	str	The shapefile's stored CRS ("EPSG:3035", ETRS89-LAEA Europe). The backend reprojects to WGS84 for a degree bbox filter and for parity with the other vector backends.

Examples:

• The join field and CRS are what the geometry attach reads:

  >>> from earthlens.hanze import Catalog
  >>> geometry = Catalog().geometry
  >>> geometry.join_field, geometry.crs
  ('Code', 'EPSG:3035')
  

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

class GeometryJoin(BaseModel):
    """The region-shapefile join configuration for `with_geometry`.

    Attributes:
        member_stem: The shapefile member stem inside the region zip
            (`"NUTS3_regions_v2024_simplified"`); the `.shp` and its sidecars
            share it.
        join_field: The shapefile attribute holding the NUTS-3 code (`"Code"`),
            joined to the semicolon-split `Regions affected (NUTS 3)` list.
        name_field: The shapefile attribute holding the region name (`"Name"`).
        crs: The shapefile's stored CRS (`"EPSG:3035"`, ETRS89-LAEA Europe). The
            backend reprojects to WGS84 for a degree bbox filter and for parity
            with the other vector backends.

    Examples:
        - The join field and CRS are what the geometry attach reads:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> geometry = Catalog().geometry
            >>> geometry.join_field, geometry.crs
            ('Code', 'EPSG:3035')

            ```
    """

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

    member_stem: str
    join_field: str = "Code"
    name_field: str = "Name"
    crs: str = "EPSG:3035"

HanzeFile

Bases: BaseModel

One downloadable Zenodo object of the pinned HANZE record.

Attributes:

Name	Type	Description
name	str	The file name on the record.
description	str	One-line human-readable summary.

Examples:

• The content URL is composed from the pinned record and file name:

  >>> from earthlens.hanze import Catalog
  >>> events = Catalog().file("events")
  >>> events.content_url(20478847)
  'https://zenodo.org/api/records/20478847/files/HANZE_events_v3_0_1b.csv/content'
  

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

class HanzeFile(BaseModel):
    """One downloadable Zenodo object of the pinned HANZE record.

    Attributes:
        name: The file name on the record.
        description: One-line human-readable summary.

    Examples:
        - The content URL is composed from the pinned record and file name:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> events = Catalog().file("events")
            >>> events.content_url(20478847)
            'https://zenodo.org/api/records/20478847/files/HANZE_events_v3_0_1b.csv/content'

            ```
    """

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

    name: str
    description: str = ""

    def content_url(self, record: int) -> str:
        """Return the Zenodo REST content URL this file is served from.

        Args:
            record: The pinned version record id the file belongs to.

        Returns:
            str: `https://zenodo.org/api/records/<record>/files/<name>/content`.

        Examples:
            - Compose the REST content URL for a file on a record:
                ```python
                >>> from earthlens.hanze import HanzeFile
                >>> HanzeFile(name="events.csv").content_url(20478847)
                'https://zenodo.org/api/records/20478847/files/events.csv/content'

                ```
        """
        return _CONTENT_URL.format(record=record, name=self.name)

content_url(record)

Return the Zenodo REST content URL this file is served from.

Parameters:

Name	Type	Description	Default
record	int	The pinned version record id the file belongs to.	required

Returns:

Name	Type	Description
str	str	https://zenodo.org/api/records/<record>/files/<name>/content.

Examples:

• Compose the REST content URL for a file on a record:

  >>> from earthlens.hanze import HanzeFile
  >>> HanzeFile(name="events.csv").content_url(20478847)
  'https://zenodo.org/api/records/20478847/files/events.csv/content'
  

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

def content_url(self, record: int) -> str:
    """Return the Zenodo REST content URL this file is served from.

    Args:
        record: The pinned version record id the file belongs to.

    Returns:
        str: `https://zenodo.org/api/records/<record>/files/<name>/content`.

    Examples:
        - Compose the REST content URL for a file on a record:
            ```python
            >>> from earthlens.hanze import HanzeFile
            >>> HanzeFile(name="events.csv").content_url(20478847)
            'https://zenodo.org/api/records/20478847/files/events.csv/content'

            ```
    """
    return _CONTENT_URL.format(record=record, name=self.name)

ZenodoRecord

Bases: BaseModel

The pinned Zenodo version record HANZE is fetched from.

Attributes:

Name	Type	Description
record	int	The pinned Zenodo version record id (20478847). Every file URL is composed from it, so a request is reproducible.
concept_doi	str	The moving concept DOI. Recorded so a refresh check can discover a newer version; never used to fetch.
version	str	The dataset version (v3.0.1-beta). Flagged as beta in the docs and logs.
data_period	str	The first-last year span the record covers ("1870-2025"), for documentation and the drift check.
license	str	SPDX-ish licence id (CC-BY-4.0).
attribution	str	The citation obligation the licence carries.

Examples:

• The record is the pinned version, not the concept DOI:

  >>> from earthlens.hanze import Catalog
  >>> Catalog().record.record
  20478847
  >>> Catalog().record.version
  'v3.0.1-beta'
  

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

class ZenodoRecord(BaseModel):
    """The pinned Zenodo version record HANZE is fetched from.

    Attributes:
        record: The pinned Zenodo *version* record id (`20478847`). Every file
            URL is composed from it, so a request is reproducible.
        concept_doi: The moving concept DOI. Recorded so a refresh check can
            discover a newer version; never used to fetch.
        version: The dataset version (`v3.0.1-beta`). Flagged as beta in the
            docs and logs.
        data_period: The `first-last` year span the record covers
            (`"1870-2025"`), for documentation and the drift check.
        license: SPDX-ish licence id (`CC-BY-4.0`).
        attribution: The citation obligation the licence carries.

    Examples:
        - The record is the pinned version, not the concept DOI:
            ```python
            >>> from earthlens.hanze import Catalog
            >>> Catalog().record.record
            20478847
            >>> Catalog().record.version
            'v3.0.1-beta'

            ```
    """

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

    record: int
    concept_doi: str = ""
    version: str = ""
    data_period: str = ""
    license: str = ""
    attribution: str = ""

clear_catalog_cache()

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

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

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

earthlens.hanze.geometry

Attach affected-region geometry to HANZE flood events.

This module is the only place in the HANZE backend that touches a GIS vector container, so per the pyramids policy it keeps all geometry / CRS handling inside pyramids primitives. Each HANZE event names the NUTS-3 regions it affected as a semicolon-separated code list in the Regions affected (NUTS 3) column; this module splits that list, counts how many of the (already-filtered) events touch each region, joins those codes to the NUTS-3 boundary polygons on the shapefile's Code field, reprojects the result from the shapefile's stored ETRS89-LAEA CRS (EPSG:3035) to WGS84, and returns a pyramids :class:~pyramids.feature.collection.FeatureCollection of one polygon per affected region.

The output schema is the same on the populated path (:func:join_events_to_regions) and the empty path (:func:empty_region_fc) — nuts3_code, region_name, n_events, and geometry — so a downstream to_file never chokes on a schema mismatch between a hit and a miss.

Because HANZE's impact figures (fatalities, losses) are per event national / multi-region totals rather than per-region values, they are deliberately not summed onto the regions — doing so would double-count. The one honest per-region metric is n_events, the number of the filtered events that affected each region, which is what a choropleth map should show.

empty_region_fc()

Return an empty region FeatureCollection with the canonical schema.

Used when the filtered events reference no region present in the boundary file, so callers always get the same columns / dtypes back regardless of hit count.

Returns:

Name	Type	Description
FeatureCollection	FeatureCollection	Zero rows, the :data:REGION_COLUMNS columns with their dtypes, an empty geometry column, CRS EPSG:4326.

Examples:

• The schema is present even with no rows:

  >>> from earthlens.hanze.geometry import empty_region_fc, REGION_COLUMNS
  >>> fc = empty_region_fc()
  >>> len(fc)
  0
  >>> set(REGION_COLUMNS).issubset(fc.columns)
  True
  >>> fc.crs.to_epsg()
  4326
  

Source code in libs/providers/hazards/src/earthlens/hanze/geometry.py

def empty_region_fc() -> FeatureCollection:
    """Return an empty region `FeatureCollection` with the canonical schema.

    Used when the filtered events reference no region present in the boundary
    file, so callers always get the same columns / dtypes back regardless of hit
    count.

    Returns:
        FeatureCollection: Zero rows, the :data:`REGION_COLUMNS` columns with
            their dtypes, an empty `geometry` column, CRS `EPSG:4326`.

    Examples:
        - The schema is present even with no rows:
            ```python
            >>> from earthlens.hanze.geometry import empty_region_fc, REGION_COLUMNS
            >>> fc = empty_region_fc()
            >>> len(fc)
            0
            >>> set(REGION_COLUMNS).issubset(fc.columns)
            True
            >>> fc.crs.to_epsg()
            4326

            ```
    """
    frame = pd.DataFrame(
        {column: pd.Series([], dtype=dtype) for column, dtype in REGION_COLUMNS.items()}
    )
    gdf = gpd.GeoDataFrame(
        frame, geometry=gpd.GeoSeries([], crs=OUTPUT_CRS), crs=OUTPUT_CRS
    )
    return FeatureCollection(gdf)

event_region_counts(events, regions_column)

Count how many events affect each NUTS-3 region.

Parameters:

Name	Type	Description	Default
events	DataFrame	The filtered events table.	required
regions_column	str	The column holding the semicolon-separated NUTS-3 code list ("Regions affected (NUTS 3)").	required

Returns:

Type	Description
Counter[str]	Counter[str]: NUTS-3 code -> number of events referencing it. Each code is counted at most once per event, even if it appears twice in that event's list.

Examples:

• Two events over three regions, counted per code (case-normalised):

  >>> import pandas as pd
  >>> from earthlens.hanze.geometry import event_region_counts
  >>> events = pd.DataFrame({"regions": ["DE300;DE711", "de300"]})
  >>> counts = event_region_counts(events, "regions")
  >>> counts["DE300"]
  2
  >>> counts["DE711"]
  1
  

Source code in libs/providers/hazards/src/earthlens/hanze/geometry.py

def event_region_counts(events: pd.DataFrame, regions_column: str) -> Counter[str]:
    """Count how many events affect each NUTS-3 region.

    Args:
        events: The filtered events table.
        regions_column: The column holding the semicolon-separated NUTS-3 code
            list (`"Regions affected (NUTS 3)"`).

    Returns:
        Counter[str]: NUTS-3 code -> number of events referencing it. Each code
            is counted at most once per event, even if it appears twice in that
            event's list.

    Examples:
        - Two events over three regions, counted per code (case-normalised):
            ```python
            >>> import pandas as pd
            >>> from earthlens.hanze.geometry import event_region_counts
            >>> events = pd.DataFrame({"regions": ["DE300;DE711", "de300"]})
            >>> counts = event_region_counts(events, "regions")
            >>> counts["DE300"]
            2
            >>> counts["DE711"]
            1

            ```
    """
    counts: Counter[str] = Counter()
    if regions_column not in events.columns:
        return counts
    for cell in events[regions_column]:
        # Upper-case the codes so the join stays in lockstep with the tabular
        # filter (`_row_matches_codes` / `_bbox_region_codes`), which compare
        # upper-cased sets — the two output kinds must not diverge on case skew.
        counts.update({code.upper() for code in split_nuts3(cell)})
    return counts

join_events_to_regions(events, regions, *, regions_column, join_field, name_field)

Join filtered events to their affected NUTS-3 region polygons.

Splits each event's Regions affected (NUTS 3) list, counts the events per region, selects the boundary polygons whose join_field is among the affected codes, reprojects them to WGS84, and returns one feature per affected region carrying its code, name and event count.

Parameters:

Name	Type	Description	Default
events	DataFrame	The filtered events table.	required
regions	FeatureCollection	The NUTS-3 boundary polygons, in the shapefile's stored CRS (EPSG:3035), carrying join_field and name_field attributes.	required
regions_column	str	The events column holding the semicolon-separated NUTS-3 code list ("Regions affected (NUTS 3)").	required
join_field	str	The boundary attribute holding the NUTS-3 code ("Code").	required
name_field	str	The boundary attribute holding the region name ("Name").	required

Returns:

Name	Type	Description
FeatureCollection	FeatureCollection	One polygon per affected region, columns nuts3_code / region_name / n_events / geometry, CRS EPSG:4326. Empty (schema-only) when no affected code is present in the boundary file.

Examples:

• Join two events to their affected region polygons and read the counts:

  >>> import geopandas as gpd
  >>> import pandas as pd
  >>> from shapely.geometry import box
  >>> from earthlens.hanze.geometry import join_events_to_regions
  >>> regions = gpd.GeoDataFrame(
  ...     {"Code": ["DE300", "NL414"], "Name": ["Berlin", "Zuidoost"]},
  ...     geometry=[box(13, 52, 14, 53), box(5, 51, 6, 52)],
  ...     crs="EPSG:4326",
  ... )
  >>> events = pd.DataFrame({"regions": ["DE300;NL414", "DE300"]})
  >>> fc = join_events_to_regions(
  ...     events, regions, regions_column="regions",
  ...     join_field="Code", name_field="Name",
  ... )
  >>> dict(zip(fc["nuts3_code"], fc["n_events"]))
  {'DE300': 2, 'NL414': 1}
  >>> fc.crs.to_epsg()
  4326
  

Source code in libs/providers/hazards/src/earthlens/hanze/geometry.py

def join_events_to_regions(
    events: pd.DataFrame,
    regions: FeatureCollection,
    *,
    regions_column: str,
    join_field: str,
    name_field: str,
) -> FeatureCollection:
    """Join filtered events to their affected NUTS-3 region polygons.

    Splits each event's `Regions affected (NUTS 3)` list, counts the events per
    region, selects the boundary polygons whose `join_field` is among the
    affected codes, reprojects them to WGS84, and returns one feature per
    affected region carrying its code, name and event count.

    Args:
        events: The filtered events table.
        regions: The NUTS-3 boundary polygons, in the shapefile's stored CRS
            (`EPSG:3035`), carrying `join_field` and `name_field` attributes.
        regions_column: The events column holding the semicolon-separated NUTS-3
            code list (`"Regions affected (NUTS 3)"`).
        join_field: The boundary attribute holding the NUTS-3 code (`"Code"`).
        name_field: The boundary attribute holding the region name (`"Name"`).

    Returns:
        FeatureCollection: One polygon per affected region, columns
            `nuts3_code` / `region_name` / `n_events` / `geometry`, CRS
            `EPSG:4326`. Empty (schema-only) when no affected code is present in
            the boundary file.

    Examples:
        - Join two events to their affected region polygons and read the counts:
            ```python
            >>> import geopandas as gpd
            >>> import pandas as pd
            >>> from shapely.geometry import box
            >>> from earthlens.hanze.geometry import join_events_to_regions
            >>> regions = gpd.GeoDataFrame(
            ...     {"Code": ["DE300", "NL414"], "Name": ["Berlin", "Zuidoost"]},
            ...     geometry=[box(13, 52, 14, 53), box(5, 51, 6, 52)],
            ...     crs="EPSG:4326",
            ... )
            >>> events = pd.DataFrame({"regions": ["DE300;NL414", "DE300"]})
            >>> fc = join_events_to_regions(
            ...     events, regions, regions_column="regions",
            ...     join_field="Code", name_field="Name",
            ... )
            >>> dict(zip(fc["nuts3_code"], fc["n_events"]))
            {'DE300': 2, 'NL414': 1}
            >>> fc.crs.to_epsg()
            4326

            ```
    """
    counts = event_region_counts(events, regions_column)
    if not counts or join_field not in regions.columns:
        return empty_region_fc()

    # Compare upper-cased on both sides so the selection tolerates any case skew
    # between the events column and the boundary file's `Code` field, matching
    # the tabular path's normalisation.
    selected = regions[regions[join_field].astype(str).str.upper().isin(counts.keys())]
    if not len(selected):
        return empty_region_fc()

    # Reproject to WGS84 first; the shapefile is ETRS89-LAEA (EPSG:3035). Reset
    # the index so the geometry `GeoSeries` (built from a positional numpy array
    # below) aligns with `frame` row-for-row: `selected` keeps the boundary
    # file's original scattered indices, and geopandas aligns a `GeoSeries` to
    # the frame *by index*, so without this a region at a high original index
    # would be paired with a missing (null) geometry.
    reprojected = selected.to_crs(OUTPUT_CRS).reset_index(drop=True)
    frame = pd.DataFrame(
        {
            "nuts3_code": reprojected[join_field].astype("string"),
            "region_name": reprojected[name_field].astype("string")
            if name_field in reprojected.columns
            else pd.Series([pd.NA] * len(reprojected), dtype="string"),
            "n_events": [counts[str(code).upper()] for code in reprojected[join_field]],
        }
    ).astype(REGION_COLUMNS)
    gdf = gpd.GeoDataFrame(
        frame,
        geometry=gpd.GeoSeries(reprojected.geometry.to_numpy(), crs=OUTPUT_CRS),
        crs=OUTPUT_CRS,
    )
    return FeatureCollection(gdf)

split_nuts3(value)

Split one Regions affected (NUTS 3) cell into NUTS-3 codes.

The cell is a semicolon-separated list ("AL011;AL012;AL013"); surrounding whitespace and empty segments are dropped. A missing / non-string cell yields an empty list.

Parameters:

Name	Type	Description	Default
value	object	One cell of the Regions affected (NUTS 3) column.	required

Returns:

Type	Description
list[str]	list[str]: The NUTS-3 codes, in order, with blanks removed.

Examples:

• A semicolon list splits into its codes; a blank cell yields nothing:

  >>> from earthlens.hanze.geometry import split_nuts3
  >>> split_nuts3("AL011; AL012 ;;AL013")
  ['AL011', 'AL012', 'AL013']
  >>> split_nuts3(None)
  []
  

Source code in libs/providers/hazards/src/earthlens/hanze/geometry.py

def split_nuts3(value: object) -> list[str]:
    """Split one `Regions affected (NUTS 3)` cell into NUTS-3 codes.

    The cell is a semicolon-separated list (`"AL011;AL012;AL013"`); surrounding
    whitespace and empty segments are dropped. A missing / non-string cell
    yields an empty list.

    Args:
        value: One cell of the `Regions affected (NUTS 3)` column.

    Returns:
        list[str]: The NUTS-3 codes, in order, with blanks removed.

    Examples:
        - A semicolon list splits into its codes; a blank cell yields nothing:
            ```python
            >>> from earthlens.hanze.geometry import split_nuts3
            >>> split_nuts3("AL011; AL012 ;;AL013")
            ['AL011', 'AL012', 'AL013']
            >>> split_nuts3(None)
            []

            ```
    """
    if not isinstance(value, str):
        return []
    return [code.strip() for code in value.split(";") if code.strip()]