FLODIS — API reference

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

earthlens.flodis

FLODIS observed-flood footprints <-> impacts backend.

Fetches the FLODIS dataset (Mester, Frieler & Schewe, PIK; Sci Data 10, 482, 2023) from its pinned static Zenodo record, and returns per-event impact records as a :class:pandas.DataFrame. FLODIS is the observed hazard-footprint -> impact bridge: it links EM-DAT fatalities + economic damages (dataset="damages") and IDMC displacements (dataset="displacement") to Global Flood Database satellite flood footprints, adding per-event affected population, GDP and critical-infrastructure counts. The global companion to the European hanze backend, and to the raw impact tables in emdat.

FLODIS carries the join keys — disasterno (EM-DAT) on the damages table, GID_1 / GID_2 (GADM) on the displacement table — but does not re-fetch the footprints: the GDIS geometry comes from the shipped emdat backend and the GFD extents from the shipped gee backend (GLOBAL_FLOOD_DB/MODIS_EVENTS/V1), joined on those keys.

This is a tabular backend: the result is a table of per-event impact rows, not a gridded array, so the :class:earthlens.earthlens.EarthLens facade rejects an aggregate= argument.

FLODIS needs no credentials — the Zenodo record is public (CC-BY-4.0) — so there is no auth class and no [flodis] extra: the only dependencies (HttpClient, pandas) are core.

Public surface (re-exported from this package):

• :class:FLODIS — the backend; instantiate with dataset= and optional country= / gid= / date filters, then call :meth:FLODIS.download.
• :class:Catalog — loader for the bundled flodis_data_catalog.yaml.
• :class:ZenodoRecord / :class:FlodisDataset — the catalog's frozen row models.
• :data:CATALOG_PATH — path to the bundled catalog YAML; monkey-patchable in tests.

Examples:

• List the selectable tables:

  >>> from earthlens.flodis import Catalog
  >>> Catalog().tables()
  ['damages', 'displacement']
  

Catalog

Bases: AbstractCatalog

Catalog for the FLODIS backend.

Reads the bundled flodis_data_catalog.yaml (shipped as package data) and exposes the pinned Zenodo record, the two selectable tables (as :class:FlodisDataset rows keyed by dataset under the inherited :attr:datasets field — the cat["damages"] / "damages" in cat / len(cat) dict surface), and the friendly-name -> CSV-header map. Instantiate with no arguments (Catalog()).

Attributes:

Name	Type	Description
datasets	dict[str, FlodisDataset]	Map from a dataset string to its :class:FlodisDataset row.
record	ZenodoRecord | None	The pinned :class:ZenodoRecord.
columns	dict[str, str]	Friendly name -> exact FLODIS CSV header.

Examples:

• List the tables, resolve one, and read the pinned record:

  >>> from earthlens.flodis import Catalog
  >>> cat = Catalog()
  >>> cat.tables()
  ['damages', 'displacement']
  >>> cat.dataset("displacement").file
  'FLODIS_displacement.csv'
  >>> "damages" in cat
  True
  >>> cat.column("disasterno")
  'disasterno'
  

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

  >>> from earthlens.flodis import Catalog
  >>> Catalog().dataset("damage")
  Traceback (most recent call last):
      ...
  ValueError: 'damage' is not in the FLODIS catalog. Known datasets: ['damages', 'displacement']. Did you mean 'damages'?
  

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

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

    Reads the bundled `flodis_data_catalog.yaml` (shipped as package data) and
    exposes the pinned Zenodo record, the two selectable tables (as
    :class:`FlodisDataset` rows keyed by `dataset` under the inherited
    :attr:`datasets` field — the `cat["damages"]` / `"damages" in cat` /
    `len(cat)` dict surface), and the friendly-name -> CSV-header map.
    Instantiate with no arguments (`Catalog()`).

    Attributes:
        datasets: Map from a `dataset` string to its :class:`FlodisDataset` row.
        record: The pinned :class:`ZenodoRecord`.
        columns: Friendly name -> exact FLODIS CSV header.

    Examples:
        - List the tables, resolve one, and read the pinned record:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> cat = Catalog()
            >>> cat.tables()
            ['damages', 'displacement']
            >>> cat.dataset("displacement").file
            'FLODIS_displacement.csv'
            >>> "damages" in cat
            True
            >>> cat.column("disasterno")
            'disasterno'

            ```
        - An unknown table raises with a did-you-mean hint:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> Catalog().dataset("damage")
            Traceback (most recent call last):
                ...
            ValueError: 'damage' is not in the FLODIS catalog. Known datasets: ['damages', 'displacement']. Did you mean 'damages'?

            ```
    """

    _catalog_kind: str = "FLODIS catalog"
    _entry_noun: str = "datasets"

    datasets: dict[str, FlodisDataset] = Field(default_factory=dict)
    #: `record` defaults 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. `None` is falsy, so the bundled record loads as intended.
    record: ZenodoRecord | 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 FLODIS 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 tables:
                ```python
                >>> from earthlens.flodis import Catalog
                >>> cat = Catalog.load()
                >>> cat.record.record
                8123096
                >>> cat.tables()
                ['damages', 'displacement']

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

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

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

        Examples:
            - The table map is keyed by the `dataset` string:
                ```python
                >>> from earthlens.flodis import Catalog
                >>> sorted(Catalog().get_catalog())
                ['damages', 'displacement']

                ```
        """
        return self.datasets

    def dataset(self, name: str) -> FlodisDataset:
        """Return the :class:`FlodisDataset` for `name`, with a did-you-mean hint.

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

        Args:
            name: A FLODIS table name (`"damages"` or `"displacement"`).

        Returns:
            FlodisDataset: The matching row.

        Raises:
            ValueError: If `name` is not a registered table.

        Examples:
            - Resolve a table and read its file name:
                ```python
                >>> from earthlens.flodis import Catalog
                >>> Catalog().dataset("damages").file
                'FLODIS_mortality_damage.csv'

                ```
        """
        return cast("FlodisDataset", self.get_dataset(name))

    def tables(self) -> list[str]:
        """Return the registered table names, sorted.

        Returns:
            list[str]: The table names (`["damages", "displacement"]`).

        Examples:
            - The registered tables come back sorted:
                ```python
                >>> from earthlens.flodis import Catalog
                >>> Catalog().tables()
                ['damages', 'displacement']

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

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

        Args:
            friendly: A friendly key from the catalog's `columns:` map
                (`"iso3"`, `"year"`, `"disasterno"`, `"gid_1"`, ...).

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

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

        Examples:
            - Map friendly keys to their exact FLODIS headers:
                ```python
                >>> from earthlens.flodis import Catalog
                >>> cat = Catalog()
                >>> cat.column("iso3")
                'ISO3'
                >>> cat.column("total_damages_000_usd")
                'total_damages_(000_USD)'

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

column(friendly)

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

Parameters:

Name	Type	Description	Default
friendly	str	A friendly key from the catalog's columns: map ("iso3", "year", "disasterno", "gid_1", ...).	required

Returns:

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

Raises:

Type	Description
KeyError	If friendly is not a mapped column.

Examples:

• Map friendly keys to their exact FLODIS headers:

  >>> from earthlens.flodis import Catalog
  >>> cat = Catalog()
  >>> cat.column("iso3")
  'ISO3'
  >>> cat.column("total_damages_000_usd")
  'total_damages_(000_USD)'
  

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

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

    Args:
        friendly: A friendly key from the catalog's `columns:` map
            (`"iso3"`, `"year"`, `"disasterno"`, `"gid_1"`, ...).

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

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

    Examples:
        - Map friendly keys to their exact FLODIS headers:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> cat = Catalog()
            >>> cat.column("iso3")
            'ISO3'
            >>> cat.column("total_damages_000_usd")
            'total_damages_(000_USD)'

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

dataset(name)

Return the :class:FlodisDataset for name, with a did-you-mean hint.

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

Parameters:

Name	Type	Description	Default
name	str	A FLODIS table name ("damages" or "displacement").	required

Returns:

Name	Type	Description
FlodisDataset	FlodisDataset	The matching row.

Raises:

Type	Description
ValueError	If name is not a registered table.

Examples:

• Resolve a table and read its file name:

  >>> from earthlens.flodis import Catalog
  >>> Catalog().dataset("damages").file
  'FLODIS_mortality_damage.csv'
  

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

def dataset(self, name: str) -> FlodisDataset:
    """Return the :class:`FlodisDataset` for `name`, with a did-you-mean hint.

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

    Args:
        name: A FLODIS table name (`"damages"` or `"displacement"`).

    Returns:
        FlodisDataset: The matching row.

    Raises:
        ValueError: If `name` is not a registered table.

    Examples:
        - Resolve a table and read its file name:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> Catalog().dataset("damages").file
            'FLODIS_mortality_damage.csv'

            ```
    """
    return cast("FlodisDataset", self.get_dataset(name))

get_catalog()

Return the table map (satisfies the abstract contract).

Returns:

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

Examples:

• The table map is keyed by the dataset string:

  >>> from earthlens.flodis import Catalog
  >>> sorted(Catalog().get_catalog())
  ['damages', 'displacement']
  

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

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

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

    Examples:
        - The table map is keyed by the `dataset` string:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> sorted(Catalog().get_catalog())
            ['damages', 'displacement']

            ```
    """
    return self.datasets

load(catalog_path=None) classmethod

Read the FLODIS 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 tables:

  >>> from earthlens.flodis import Catalog
  >>> cat = Catalog.load()
  >>> cat.record.record
  8123096
  >>> cat.tables()
  ['damages', 'displacement']
  

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

@classmethod
def load(cls, catalog_path: Path | None = None) -> Catalog:
    """Read the FLODIS 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 tables:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> cat = Catalog.load()
            >>> cat.record.record
            8123096
            >>> cat.tables()
            ['damages', 'displacement']

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

tables()

Return the registered table names, sorted.

Returns:

Type	Description
list[str]	list[str]: The table names (["damages", "displacement"]).

Examples:

• The registered tables come back sorted:

  >>> from earthlens.flodis import Catalog
  >>> Catalog().tables()
  ['damages', 'displacement']
  

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

def tables(self) -> list[str]:
    """Return the registered table names, sorted.

    Returns:
        list[str]: The table names (`["damages", "displacement"]`).

    Examples:
        - The registered tables come back sorted:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> Catalog().tables()
            ['damages', 'displacement']

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

FLODIS

Bases: AbstractDataSource

FLODIS observed-flood impacts backend (tabular).

Downloads the selected FLODIS table from its pinned Zenodo release, filters it by country / GADM code / date window, and returns a :class:pandas.DataFrame carrying the join keys (disasterno for damages, GID_1 / GID_2 for displacement) so a caller can join to the shipped emdat (GDIS) footprints and gee (Global Flood Database) extents.

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

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	"tabular" — the result is a table of per-event impact rows, so the facade rejects aggregate=.
REQUIRES_TIME_WINDOW		False — a request without a window returns every year the record covers (2000-2018).

Examples:

• Pull Mozambique flood-damage events, or the displacement table, through the facade (both fetch from Zenodo, so this is illustrative, not a doctest):

  from earthlens.core import EarthLens
  
  damages = EarthLens(
      "flodis", dataset="damages", country="MOZ", start="2000", end="2018"
  ).download()  # a pandas.DataFrame keyed on disasterno
  
  displacement = EarthLens(
      "flodis", dataset="displacement", country="MOZ"
  ).download()  # a pandas.DataFrame keyed on GID_1 / GID_2
  

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

class FLODIS(AbstractDataSource):
    """FLODIS observed-flood impacts backend (tabular).

    Downloads the selected FLODIS table from its pinned Zenodo release, filters
    it by country / GADM code / date window, and returns a
    :class:`pandas.DataFrame` carrying the join keys (`disasterno` for `damages`,
    `GID_1` / `GID_2` for `displacement`) so a caller can join to the shipped
    `emdat` (GDIS) footprints and `gee` (Global Flood Database) extents.

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

    Attributes:
        OUTPUT_KIND: `"tabular"` — the result is a table of per-event impact
            rows, so the facade rejects `aggregate=`.
        REQUIRES_TIME_WINDOW: `False` — a request without a window returns every
            year the record covers (2000-2018).

    Examples:
        - Pull Mozambique flood-damage events, or the displacement table, through
          the facade (both fetch from Zenodo, so this is illustrative, not a
          doctest):

            ```python
            from earthlens.core import EarthLens

            damages = EarthLens(
                "flodis", dataset="damages", country="MOZ", start="2000", end="2018"
            ).download()  # a pandas.DataFrame keyed on disasterno

            displacement = EarthLens(
                "flodis", dataset="displacement", country="MOZ"
            ).download()  # a pandas.DataFrame keyed on GID_1 / GID_2
            ```
    """

    OUTPUT_KIND: OutputKind = "tabular"

    REQUIRES_TIME_WINDOW = False

    AGGREGATE_REFUSAL_REASON = (
        "FLODIS serves per-event flood impact records (deaths / damages / "
        "displacements matched to observed footprints), not gridded rasters, so "
        "there is no meaningful gridded reduction. Call download() without "
        "aggregate= and post-process the returned DataFrame 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",
        dataset: str = "damages",
        variables: list[str] | None = None,
        country: str | list[str] | None = None,
        gid: str | list[str] | None = None,
        timeout: float = 120.0,
    ):
        """Initialise a FLODIS backend instance.

        Args:
            start: Inclusive start of an optional window, parsed with `fmt`. Only
                its year is significant — FLODIS indexes events by year. `None`
                means "from the beginning of the record" (2000).
            end: Inclusive end of the optional window; `None` means "to the end
                of the record" (2018).
            lat_lim: Accepted for facade parity but not a filter axis — FLODIS
                tables carry no per-row coordinates (footprints come from the
                `emdat` / `gee` backends).
            lon_lim: Accepted for facade parity; see `lat_lim`.
            temporal_resolution: FLODIS issues one query over the whole window,
                so this is the sentinel `"all"`, not a pandas frequency alias.
            path: Output directory for the cached CSV and the written table.
                Created by the parent class if absent.
            fmt: `strptime` format for `start` / `end`.
            dataset: Which table to fetch — `"damages"` (EM-DAT deaths/damages,
                the default) or `"displacement"` (IDMC displacements).
            variables: FLODIS has no variable axis (`dataset=` selects a whole
                table). Accepted only so the facade can route `dataset=`; a
                non-empty value is rejected.
            country: One ISO3 country code or a list of them (`"MOZ"`,
                `["MOZ", "BGD"]`). `None` keeps every country.
            gid: One GADM code or a list of them, matched against the
                displacement table's `GID_1` / `GID_2`. Only valid with
                `dataset="displacement"` (the damages table is not GADM-keyed).
                `None` keeps every region.
            timeout: Per-request timeout in seconds for the Zenodo download.

        Raises:
            ValueError: If `dataset` is not a registered table, `variables=` is
                non-empty, a `country` value is not a 3-letter ISO3 code, or
                `gid=` is given for the non-GADM `damages` table.
        """
        self._catalog = Catalog()
        if self._catalog.record is None:
            raise ValueError(
                "the FLODIS catalog failed to load its 'record:' block; the "
                "bundled flodis_data_catalog.yaml is malformed."
            )
        self._record = self._catalog.record
        if variables:
            raise ValueError(
                "FLODIS selects a whole table with dataset= and has no variable "
                f"axis; got variables={variables!r}. Use "
                "dataset='damages' | 'displacement'."
            )
        self._dataset_name = dataset
        # Resolve against the catalog (did-you-mean hint on a typo).
        self._dataset: FlodisDataset = self._catalog.dataset(dataset)
        self._country = _normalize_iso3(country)
        self._gid = _normalize_gid(gid, dataset, self._dataset)
        self._timeout = timeout
        self._http: HttpClient | None = None

        super().__init__(
            start=cast("str", start),
            end=cast("str", end),
            # FLODIS is facet-only over the wire: the table is chosen by
            # `dataset=`, so no `variables` list is threaded to the base class.
            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 _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 (recorded, not a filter axis).
        """
        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`.

        FLODIS 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,
        )

    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 `hanze`).

        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 _load_table(self) -> pd.DataFrame:
        """Download and parse the selected FLODIS table (cached).

        The download is guarded by the shared CSV magic (`,ISO3,`), so an HTML
        error page served with a `200` status is rejected at the download site
        rather than cached under the CSV name and failing confusingly at
        `read_csv` on every later call. The leading unnamed index column FLODIS
        ships (a bare pandas write index) is dropped with `index_col=0`.

        Returns:
            pandas.DataFrame: The full selected table, FLODIS's documented
                headers, with a clean `RangeIndex`.
        """
        record = self._record.record
        entry = self._dataset
        # The pristine download lives in a dedicated sub-directory, never in
        # `root_dir` beside the written output. `download()` writes its filtered
        # result as `flodis_<table>.csv`, which on a case-insensitive filesystem
        # (Windows, default macOS) would be the *same path* as the raw
        # `FLODIS_<table>.csv` for the displacement table — overwriting the
        # pristine cache with an index-stripped copy and corrupting every later
        # read. Separating the two directories makes that collision impossible.
        local = self._source_path()
        if not local.exists():
            logger.info(f"FLODIS: downloading {entry.file} (record {record}).")
            local.parent.mkdir(parents=True, exist_ok=True)
            self._client().download(
                entry.content_url(record),
                local,
                expect_magic=_CSV_MAGIC,
                progress=self._progress,
            )
        return pd.read_csv(local, index_col=0).reset_index(drop=True)

    def _source_path(self) -> Path:
        """Return the cache path for the pristine download of the selected table.

        Kept in a dedicated `flodis_source/` sub-directory so it cannot collide
        with the filtered CSV `download()` writes into `root_dir` (see
        :meth:`_load_table`).

        Returns:
            Path: `root_dir/flodis_source/<file>`.
        """
        return self.root_dir / "flodis_source" / self._dataset.file

    def _filter_table(self, table: pd.DataFrame) -> pd.DataFrame:
        """Apply the request's country / GADM / date filters.

        Args:
            table: The full selected table.

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

        if self._country:
            mask &= table[columns["iso3"]].astype(str).str.upper().isin(self._country)

        if self._gid:
            # Filter against the dataset's own join-key columns (`GID_1`/`GID_2`),
            # the same source of truth `_normalize_gid` validated against — so a
            # `gid=` that was accepted always has real columns to match, rather
            # than silently dropping every row if the `columns:` map drifted.
            gid_mask = pd.Series(False, index=table.index)
            for col in self._dataset.key_columns:
                if col in table.columns:
                    gid_mask |= table[col].astype(str).str.upper().isin(self._gid)
            mask &= gid_mask

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

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

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

        Returns:
            list[RemoteProduct]: A single product carrying the dataset name and
                record id.
        """
        return [
            RemoteProduct(
                id=f"flodis:{self._dataset_name}",
                metadata={"record": self._record.record, "file": self._dataset.file},
            )
        ]

    def _fetch(self, products: list[RemoteProduct]) -> list[Any]:
        """Download and filter the one product.

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

        Returns:
            list[Any]: One element — the filtered :class:`pandas.DataFrame`.
        """
        # Single-product backend: the one table is re-derived from instance state
        # (`dataset` + filters), so `products` is accepted for the base
        # search -> fetch contract but carries nothing this method needs to read.
        return [self._filter_table(self._load_table())]

    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:
        """Fetch the selected FLODIS table and return it.

        Runs the download + filter, writes the result to `path` as a CSV, 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:
            pandas.DataFrame: The filtered per-event impact rows, carrying the
                join keys (`disasterno` for `damages`, `GID_1` / `GID_2` for
                `displacement`). Also written under `root_dir`.

        Raises:
            requests.HTTPError: If the Zenodo download returns a non-2xx status.
            ValueError: If the download's body fails its content guard (an HTML
                error page served with a 200 status).
        """
        self._progress = progress_bar
        results = self._api()
        result = cast("pd.DataFrame", results[0])
        self._log_citation()
        out_path = self.root_dir / (self._result_stem() + ".csv")
        result.to_csv(out_path, index=False)
        logger.info(
            f"FLODIS {self._dataset_name}: {len(result)} row(s) written to {out_path}."
        )
        return result

    def _result_stem(self) -> str:
        """Compose an output file stem that encodes the table and its filters.

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

        Returns:
            str: `flodis_<table>`, or `flodis_<table>-<8-hex-digest>` when any
                filter is active.
        """
        base = f"flodis_{self._dataset_name}"
        first_year, last_year = self._year_range
        applied = (
            bool(self._country),
            bool(self._gid),
            first_year is not None,
            last_year is not None,
        )
        if not any(applied):
            return base
        request = (
            tuple(sorted(self._country)),
            tuple(sorted(self._gid)),
            self._year_range,
        )
        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"FLODIS source citation: {record.attribution}")

__init__(start=None, end=None, lat_lim=None, lon_lim=None, temporal_resolution='all', path=None, fmt='%Y-%m-%d', dataset='damages', variables=None, country=None, gid=None, timeout=120.0)

Initialise a FLODIS backend instance.

Parameters:

Name	Type	Description	Default
start	str | None	Inclusive start of an optional window, parsed with fmt. Only its year is significant — FLODIS indexes events by year. None means "from the beginning of the record" (2000).	None
end	str | None	Inclusive end of the optional window; None means "to the end of the record" (2018).	None
lat_lim	list[float] | None	Accepted for facade parity but not a filter axis — FLODIS tables carry no per-row coordinates (footprints come from the emdat / gee backends).	None
lon_lim	list[float] | None	Accepted for facade parity; see lat_lim.	None
temporal_resolution	str	FLODIS 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 CSV and the written table. Created by the parent class if absent.	None
fmt	str	strptime format for start / end.	'%Y-%m-%d'
dataset	str	Which table to fetch — "damages" (EM-DAT deaths/damages, the default) or "displacement" (IDMC displacements).	'damages'
variables	list[str] | None	FLODIS has no variable axis (dataset= selects a whole table). Accepted only so the facade can route dataset=; a non-empty value is rejected.	None
country	str | list[str] | None	One ISO3 country code or a list of them ("MOZ", ["MOZ", "BGD"]). None keeps every country.	None
gid	str | list[str] | None	One GADM code or a list of them, matched against the displacement table's GID_1 / GID_2. Only valid with dataset="displacement" (the damages table is not GADM-keyed). None keeps every region.	None
timeout	float	Per-request timeout in seconds for the Zenodo download.	120.0

Raises:

Type	Description
ValueError	If dataset is not a registered table, variables= is non-empty, a country value is not a 3-letter ISO3 code, or gid= is given for the non-GADM damages table.

Source code in libs/providers/hazards/src/earthlens/flodis/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",
    dataset: str = "damages",
    variables: list[str] | None = None,
    country: str | list[str] | None = None,
    gid: str | list[str] | None = None,
    timeout: float = 120.0,
):
    """Initialise a FLODIS backend instance.

    Args:
        start: Inclusive start of an optional window, parsed with `fmt`. Only
            its year is significant — FLODIS indexes events by year. `None`
            means "from the beginning of the record" (2000).
        end: Inclusive end of the optional window; `None` means "to the end
            of the record" (2018).
        lat_lim: Accepted for facade parity but not a filter axis — FLODIS
            tables carry no per-row coordinates (footprints come from the
            `emdat` / `gee` backends).
        lon_lim: Accepted for facade parity; see `lat_lim`.
        temporal_resolution: FLODIS issues one query over the whole window,
            so this is the sentinel `"all"`, not a pandas frequency alias.
        path: Output directory for the cached CSV and the written table.
            Created by the parent class if absent.
        fmt: `strptime` format for `start` / `end`.
        dataset: Which table to fetch — `"damages"` (EM-DAT deaths/damages,
            the default) or `"displacement"` (IDMC displacements).
        variables: FLODIS has no variable axis (`dataset=` selects a whole
            table). Accepted only so the facade can route `dataset=`; a
            non-empty value is rejected.
        country: One ISO3 country code or a list of them (`"MOZ"`,
            `["MOZ", "BGD"]`). `None` keeps every country.
        gid: One GADM code or a list of them, matched against the
            displacement table's `GID_1` / `GID_2`. Only valid with
            `dataset="displacement"` (the damages table is not GADM-keyed).
            `None` keeps every region.
        timeout: Per-request timeout in seconds for the Zenodo download.

    Raises:
        ValueError: If `dataset` is not a registered table, `variables=` is
            non-empty, a `country` value is not a 3-letter ISO3 code, or
            `gid=` is given for the non-GADM `damages` table.
    """
    self._catalog = Catalog()
    if self._catalog.record is None:
        raise ValueError(
            "the FLODIS catalog failed to load its 'record:' block; the "
            "bundled flodis_data_catalog.yaml is malformed."
        )
    self._record = self._catalog.record
    if variables:
        raise ValueError(
            "FLODIS selects a whole table with dataset= and has no variable "
            f"axis; got variables={variables!r}. Use "
            "dataset='damages' | 'displacement'."
        )
    self._dataset_name = dataset
    # Resolve against the catalog (did-you-mean hint on a typo).
    self._dataset: FlodisDataset = self._catalog.dataset(dataset)
    self._country = _normalize_iso3(country)
    self._gid = _normalize_gid(gid, dataset, self._dataset)
    self._timeout = timeout
    self._http: HttpClient | None = None

    super().__init__(
        start=cast("str", start),
        end=cast("str", end),
        # FLODIS is facet-only over the wire: the table is chosen by
        # `dataset=`, so no `variables` list is threaded to the base class.
        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 the selected FLODIS table and return it.

Runs the download + filter, writes the result to path as a CSV, 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:

Type	Description
DataFrame	pandas.DataFrame: The filtered per-event impact rows, carrying the join keys (disasterno for damages, GID_1 / GID_2 for displacement). Also written under root_dir.

Raises:

Type	Description
HTTPError	If the Zenodo download returns a non-2xx status.
ValueError	If the download's body fails its content guard (an HTML error page served with a 200 status).

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

def download(self, progress_bar: bool = True) -> pd.DataFrame:
    """Fetch the selected FLODIS table and return it.

    Runs the download + filter, writes the result to `path` as a CSV, 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:
        pandas.DataFrame: The filtered per-event impact rows, carrying the
            join keys (`disasterno` for `damages`, `GID_1` / `GID_2` for
            `displacement`). Also written under `root_dir`.

    Raises:
        requests.HTTPError: If the Zenodo download returns a non-2xx status.
        ValueError: If the download's body fails its content guard (an HTML
            error page served with a 200 status).
    """
    self._progress = progress_bar
    results = self._api()
    result = cast("pd.DataFrame", results[0])
    self._log_citation()
    out_path = self.root_dir / (self._result_stem() + ".csv")
    result.to_csv(out_path, index=False)
    logger.info(
        f"FLODIS {self._dataset_name}: {len(result)} row(s) written to {out_path}."
    )
    return result

FlodisDataset

Bases: BaseModel

One selectable FLODIS table (a row of the catalog's dict surface).

The dataset string ("damages", "displacement") is the parent key in :attr:Catalog.datasets and is not stored on the row.

Attributes:

Name	Type	Description
file	str	The file name on the Zenodo record ("FLODIS_mortality_damage.csv").
description	str	One-line human-readable summary.
key_columns	tuple[str, ...]	The join-key column(s) the table is keyed on — disasterno (EM-DAT) for damages, GID_1 / GID_2 (GADM) for displacement. A caller joins on these to the emdat (GDIS) footprints and gee (Global Flood Database) extents.

Examples:

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

  >>> from earthlens.flodis import Catalog
  >>> damages = Catalog().dataset("damages")
  >>> damages.key_columns
  ('disasterno',)
  >>> damages.content_url(8123096)
  'https://zenodo.org/api/records/8123096/files/FLODIS_mortality_damage.csv/content'
  

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

class FlodisDataset(BaseModel):
    """One selectable FLODIS table (a row of the catalog's dict surface).

    The `dataset` string (`"damages"`, `"displacement"`) is the parent key in
    :attr:`Catalog.datasets` and is not stored on the row.

    Attributes:
        file: The file name on the Zenodo record
            (`"FLODIS_mortality_damage.csv"`).
        description: One-line human-readable summary.
        key_columns: The join-key column(s) the table is keyed on — `disasterno`
            (EM-DAT) for `damages`, `GID_1` / `GID_2` (GADM) for `displacement`.
            A caller joins on these to the `emdat` (GDIS) footprints and `gee`
            (Global Flood Database) extents.

    Examples:
        - The content URL is composed from the pinned record and file name:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> damages = Catalog().dataset("damages")
            >>> damages.key_columns
            ('disasterno',)
            >>> damages.content_url(8123096)
            'https://zenodo.org/api/records/8123096/files/FLODIS_mortality_damage.csv/content'

            ```
    """

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

    file: str
    description: str = ""
    key_columns: tuple[str, ...] = ()

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

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

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

        Examples:
            - Compose the REST content URL for a table on a record:
                ```python
                >>> from earthlens.flodis import FlodisDataset
                >>> FlodisDataset(file="FLODIS_displacement.csv").content_url(8123096)
                'https://zenodo.org/api/records/8123096/files/FLODIS_displacement.csv/content'

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

content_url(record)

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

Parameters:

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

Returns:

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

Examples:

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

  >>> from earthlens.flodis import FlodisDataset
  >>> FlodisDataset(file="FLODIS_displacement.csv").content_url(8123096)
  'https://zenodo.org/api/records/8123096/files/FLODIS_displacement.csv/content'
  

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

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

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

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

    Examples:
        - Compose the REST content URL for a table on a record:
            ```python
            >>> from earthlens.flodis import FlodisDataset
            >>> FlodisDataset(file="FLODIS_displacement.csv").content_url(8123096)
            'https://zenodo.org/api/records/8123096/files/FLODIS_displacement.csv/content'

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

ZenodoRecord

Bases: BaseModel

The pinned Zenodo record FLODIS is fetched from.

Attributes:

Name	Type	Description
record	int	The pinned Zenodo record id (8123096). Every file URL is composed from it, so a request is reproducible.
concept_doi	str	The dataset DOI as cited in the paper. Recorded for the citation and a future refresh check; never used to fetch.
version	str	The dataset version label, if the record carries one.
data_period	str	The first-last year span the record covers ("2000-2018").
license	str	SPDX-ish licence id (CC-BY-4.0).
attribution	str	The citation obligation the licence carries.

Examples:

• The record is the pinned id, and the licence is CC-BY-4.0:

  >>> from earthlens.flodis import Catalog
  >>> Catalog().record.record
  8123096
  >>> Catalog().record.license
  'CC-BY-4.0'
  

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

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

    Attributes:
        record: The pinned Zenodo record id (`8123096`). Every file URL is
            composed from it, so a request is reproducible.
        concept_doi: The dataset DOI as cited in the paper. Recorded for the
            citation and a future refresh check; never used to fetch.
        version: The dataset version label, if the record carries one.
        data_period: The `first-last` year span the record covers (`"2000-2018"`).
        license: SPDX-ish licence id (`CC-BY-4.0`).
        attribution: The citation obligation the licence carries.

    Examples:
        - The record is the pinned id, and the licence is CC-BY-4.0:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> Catalog().record.record
            8123096
            >>> Catalog().record.license
            'CC-BY-4.0'

            ```
    """

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

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

earthlens.flodis.backend

Backend that fetches the FLODIS observed-flood impact tables.

FLODIS(AbstractDataSource) downloads one of the two FLODIS tables — damages (EM-DAT fatalities + economic damages) or displacement (IDMC displacements), each matched to Global Flood Database satellite footprints — from its pinned static Zenodo record, filters it, and returns the per-event impact records as a :class:pandas.DataFrame. It is the observed hazard-footprint -> impact bridge: the global companion to the European hanze backend, and to the raw impact tables in emdat.

Three design points carry this backend:

• dataset= selects the table. dataset="damages" (default) returns the EM-DAT deaths/damages table keyed on disasterno; dataset="displacement" returns the IDMC table keyed on GID_1 / GID_2. The selector rides the facade's native-dataset path (the s3 precedent), so EarthLens("flodis", dataset="displacement", ...) reaches the backend as a dataset= kwarg. FLODIS has no variable axis: a non-empty variables= is rejected.
• Fetch the tables; join footprints via shipped backends. FLODIS carries the join keys (disasterno, GID_1 / GID_2) but does not re-fetch the footprints — the GDIS geometry comes from the shipped emdat backend and the GFD extents from the shipped gee backend, joined on those keys. The tables have no per-row coordinates, so lat_lim / lon_lim are accepted (the facade always supplies them) but are not a filter axis; filter by country= (ISO3), gid= (GADM, displacement only) and the date window instead.
• No new dependency. HttpClient + pandas are core and the Zenodo record is public (CC-BY-4.0), so there is no auth and no [flodis] extra.

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

FLODIS

Bases: AbstractDataSource

FLODIS observed-flood impacts backend (tabular).

Downloads the selected FLODIS table from its pinned Zenodo release, filters it by country / GADM code / date window, and returns a :class:pandas.DataFrame carrying the join keys (disasterno for damages, GID_1 / GID_2 for displacement) so a caller can join to the shipped emdat (GDIS) footprints and gee (Global Flood Database) extents.

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

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	"tabular" — the result is a table of per-event impact rows, so the facade rejects aggregate=.
REQUIRES_TIME_WINDOW		False — a request without a window returns every year the record covers (2000-2018).

Examples:

• Pull Mozambique flood-damage events, or the displacement table, through the facade (both fetch from Zenodo, so this is illustrative, not a doctest):

  from earthlens.core import EarthLens
  
  damages = EarthLens(
      "flodis", dataset="damages", country="MOZ", start="2000", end="2018"
  ).download()  # a pandas.DataFrame keyed on disasterno
  
  displacement = EarthLens(
      "flodis", dataset="displacement", country="MOZ"
  ).download()  # a pandas.DataFrame keyed on GID_1 / GID_2
  

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

class FLODIS(AbstractDataSource):
    """FLODIS observed-flood impacts backend (tabular).

    Downloads the selected FLODIS table from its pinned Zenodo release, filters
    it by country / GADM code / date window, and returns a
    :class:`pandas.DataFrame` carrying the join keys (`disasterno` for `damages`,
    `GID_1` / `GID_2` for `displacement`) so a caller can join to the shipped
    `emdat` (GDIS) footprints and `gee` (Global Flood Database) extents.

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

    Attributes:
        OUTPUT_KIND: `"tabular"` — the result is a table of per-event impact
            rows, so the facade rejects `aggregate=`.
        REQUIRES_TIME_WINDOW: `False` — a request without a window returns every
            year the record covers (2000-2018).

    Examples:
        - Pull Mozambique flood-damage events, or the displacement table, through
          the facade (both fetch from Zenodo, so this is illustrative, not a
          doctest):

            ```python
            from earthlens.core import EarthLens

            damages = EarthLens(
                "flodis", dataset="damages", country="MOZ", start="2000", end="2018"
            ).download()  # a pandas.DataFrame keyed on disasterno

            displacement = EarthLens(
                "flodis", dataset="displacement", country="MOZ"
            ).download()  # a pandas.DataFrame keyed on GID_1 / GID_2
            ```
    """

    OUTPUT_KIND: OutputKind = "tabular"

    REQUIRES_TIME_WINDOW = False

    AGGREGATE_REFUSAL_REASON = (
        "FLODIS serves per-event flood impact records (deaths / damages / "
        "displacements matched to observed footprints), not gridded rasters, so "
        "there is no meaningful gridded reduction. Call download() without "
        "aggregate= and post-process the returned DataFrame 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",
        dataset: str = "damages",
        variables: list[str] | None = None,
        country: str | list[str] | None = None,
        gid: str | list[str] | None = None,
        timeout: float = 120.0,
    ):
        """Initialise a FLODIS backend instance.

        Args:
            start: Inclusive start of an optional window, parsed with `fmt`. Only
                its year is significant — FLODIS indexes events by year. `None`
                means "from the beginning of the record" (2000).
            end: Inclusive end of the optional window; `None` means "to the end
                of the record" (2018).
            lat_lim: Accepted for facade parity but not a filter axis — FLODIS
                tables carry no per-row coordinates (footprints come from the
                `emdat` / `gee` backends).
            lon_lim: Accepted for facade parity; see `lat_lim`.
            temporal_resolution: FLODIS issues one query over the whole window,
                so this is the sentinel `"all"`, not a pandas frequency alias.
            path: Output directory for the cached CSV and the written table.
                Created by the parent class if absent.
            fmt: `strptime` format for `start` / `end`.
            dataset: Which table to fetch — `"damages"` (EM-DAT deaths/damages,
                the default) or `"displacement"` (IDMC displacements).
            variables: FLODIS has no variable axis (`dataset=` selects a whole
                table). Accepted only so the facade can route `dataset=`; a
                non-empty value is rejected.
            country: One ISO3 country code or a list of them (`"MOZ"`,
                `["MOZ", "BGD"]`). `None` keeps every country.
            gid: One GADM code or a list of them, matched against the
                displacement table's `GID_1` / `GID_2`. Only valid with
                `dataset="displacement"` (the damages table is not GADM-keyed).
                `None` keeps every region.
            timeout: Per-request timeout in seconds for the Zenodo download.

        Raises:
            ValueError: If `dataset` is not a registered table, `variables=` is
                non-empty, a `country` value is not a 3-letter ISO3 code, or
                `gid=` is given for the non-GADM `damages` table.
        """
        self._catalog = Catalog()
        if self._catalog.record is None:
            raise ValueError(
                "the FLODIS catalog failed to load its 'record:' block; the "
                "bundled flodis_data_catalog.yaml is malformed."
            )
        self._record = self._catalog.record
        if variables:
            raise ValueError(
                "FLODIS selects a whole table with dataset= and has no variable "
                f"axis; got variables={variables!r}. Use "
                "dataset='damages' | 'displacement'."
            )
        self._dataset_name = dataset
        # Resolve against the catalog (did-you-mean hint on a typo).
        self._dataset: FlodisDataset = self._catalog.dataset(dataset)
        self._country = _normalize_iso3(country)
        self._gid = _normalize_gid(gid, dataset, self._dataset)
        self._timeout = timeout
        self._http: HttpClient | None = None

        super().__init__(
            start=cast("str", start),
            end=cast("str", end),
            # FLODIS is facet-only over the wire: the table is chosen by
            # `dataset=`, so no `variables` list is threaded to the base class.
            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 _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 (recorded, not a filter axis).
        """
        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`.

        FLODIS 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,
        )

    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 `hanze`).

        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 _load_table(self) -> pd.DataFrame:
        """Download and parse the selected FLODIS table (cached).

        The download is guarded by the shared CSV magic (`,ISO3,`), so an HTML
        error page served with a `200` status is rejected at the download site
        rather than cached under the CSV name and failing confusingly at
        `read_csv` on every later call. The leading unnamed index column FLODIS
        ships (a bare pandas write index) is dropped with `index_col=0`.

        Returns:
            pandas.DataFrame: The full selected table, FLODIS's documented
                headers, with a clean `RangeIndex`.
        """
        record = self._record.record
        entry = self._dataset
        # The pristine download lives in a dedicated sub-directory, never in
        # `root_dir` beside the written output. `download()` writes its filtered
        # result as `flodis_<table>.csv`, which on a case-insensitive filesystem
        # (Windows, default macOS) would be the *same path* as the raw
        # `FLODIS_<table>.csv` for the displacement table — overwriting the
        # pristine cache with an index-stripped copy and corrupting every later
        # read. Separating the two directories makes that collision impossible.
        local = self._source_path()
        if not local.exists():
            logger.info(f"FLODIS: downloading {entry.file} (record {record}).")
            local.parent.mkdir(parents=True, exist_ok=True)
            self._client().download(
                entry.content_url(record),
                local,
                expect_magic=_CSV_MAGIC,
                progress=self._progress,
            )
        return pd.read_csv(local, index_col=0).reset_index(drop=True)

    def _source_path(self) -> Path:
        """Return the cache path for the pristine download of the selected table.

        Kept in a dedicated `flodis_source/` sub-directory so it cannot collide
        with the filtered CSV `download()` writes into `root_dir` (see
        :meth:`_load_table`).

        Returns:
            Path: `root_dir/flodis_source/<file>`.
        """
        return self.root_dir / "flodis_source" / self._dataset.file

    def _filter_table(self, table: pd.DataFrame) -> pd.DataFrame:
        """Apply the request's country / GADM / date filters.

        Args:
            table: The full selected table.

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

        if self._country:
            mask &= table[columns["iso3"]].astype(str).str.upper().isin(self._country)

        if self._gid:
            # Filter against the dataset's own join-key columns (`GID_1`/`GID_2`),
            # the same source of truth `_normalize_gid` validated against — so a
            # `gid=` that was accepted always has real columns to match, rather
            # than silently dropping every row if the `columns:` map drifted.
            gid_mask = pd.Series(False, index=table.index)
            for col in self._dataset.key_columns:
                if col in table.columns:
                    gid_mask |= table[col].astype(str).str.upper().isin(self._gid)
            mask &= gid_mask

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

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

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

        Returns:
            list[RemoteProduct]: A single product carrying the dataset name and
                record id.
        """
        return [
            RemoteProduct(
                id=f"flodis:{self._dataset_name}",
                metadata={"record": self._record.record, "file": self._dataset.file},
            )
        ]

    def _fetch(self, products: list[RemoteProduct]) -> list[Any]:
        """Download and filter the one product.

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

        Returns:
            list[Any]: One element — the filtered :class:`pandas.DataFrame`.
        """
        # Single-product backend: the one table is re-derived from instance state
        # (`dataset` + filters), so `products` is accepted for the base
        # search -> fetch contract but carries nothing this method needs to read.
        return [self._filter_table(self._load_table())]

    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:
        """Fetch the selected FLODIS table and return it.

        Runs the download + filter, writes the result to `path` as a CSV, 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:
            pandas.DataFrame: The filtered per-event impact rows, carrying the
                join keys (`disasterno` for `damages`, `GID_1` / `GID_2` for
                `displacement`). Also written under `root_dir`.

        Raises:
            requests.HTTPError: If the Zenodo download returns a non-2xx status.
            ValueError: If the download's body fails its content guard (an HTML
                error page served with a 200 status).
        """
        self._progress = progress_bar
        results = self._api()
        result = cast("pd.DataFrame", results[0])
        self._log_citation()
        out_path = self.root_dir / (self._result_stem() + ".csv")
        result.to_csv(out_path, index=False)
        logger.info(
            f"FLODIS {self._dataset_name}: {len(result)} row(s) written to {out_path}."
        )
        return result

    def _result_stem(self) -> str:
        """Compose an output file stem that encodes the table and its filters.

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

        Returns:
            str: `flodis_<table>`, or `flodis_<table>-<8-hex-digest>` when any
                filter is active.
        """
        base = f"flodis_{self._dataset_name}"
        first_year, last_year = self._year_range
        applied = (
            bool(self._country),
            bool(self._gid),
            first_year is not None,
            last_year is not None,
        )
        if not any(applied):
            return base
        request = (
            tuple(sorted(self._country)),
            tuple(sorted(self._gid)),
            self._year_range,
        )
        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"FLODIS source citation: {record.attribution}")

__init__(start=None, end=None, lat_lim=None, lon_lim=None, temporal_resolution='all', path=None, fmt='%Y-%m-%d', dataset='damages', variables=None, country=None, gid=None, timeout=120.0)

Initialise a FLODIS backend instance.

Parameters:

Name	Type	Description	Default
start	str | None	Inclusive start of an optional window, parsed with fmt. Only its year is significant — FLODIS indexes events by year. None means "from the beginning of the record" (2000).	None
end	str | None	Inclusive end of the optional window; None means "to the end of the record" (2018).	None
lat_lim	list[float] | None	Accepted for facade parity but not a filter axis — FLODIS tables carry no per-row coordinates (footprints come from the emdat / gee backends).	None
lon_lim	list[float] | None	Accepted for facade parity; see lat_lim.	None
temporal_resolution	str	FLODIS 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 CSV and the written table. Created by the parent class if absent.	None
fmt	str	strptime format for start / end.	'%Y-%m-%d'
dataset	str	Which table to fetch — "damages" (EM-DAT deaths/damages, the default) or "displacement" (IDMC displacements).	'damages'
variables	list[str] | None	FLODIS has no variable axis (dataset= selects a whole table). Accepted only so the facade can route dataset=; a non-empty value is rejected.	None
country	str | list[str] | None	One ISO3 country code or a list of them ("MOZ", ["MOZ", "BGD"]). None keeps every country.	None
gid	str | list[str] | None	One GADM code or a list of them, matched against the displacement table's GID_1 / GID_2. Only valid with dataset="displacement" (the damages table is not GADM-keyed). None keeps every region.	None
timeout	float	Per-request timeout in seconds for the Zenodo download.	120.0

Raises:

Type	Description
ValueError	If dataset is not a registered table, variables= is non-empty, a country value is not a 3-letter ISO3 code, or gid= is given for the non-GADM damages table.

Source code in libs/providers/hazards/src/earthlens/flodis/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",
    dataset: str = "damages",
    variables: list[str] | None = None,
    country: str | list[str] | None = None,
    gid: str | list[str] | None = None,
    timeout: float = 120.0,
):
    """Initialise a FLODIS backend instance.

    Args:
        start: Inclusive start of an optional window, parsed with `fmt`. Only
            its year is significant — FLODIS indexes events by year. `None`
            means "from the beginning of the record" (2000).
        end: Inclusive end of the optional window; `None` means "to the end
            of the record" (2018).
        lat_lim: Accepted for facade parity but not a filter axis — FLODIS
            tables carry no per-row coordinates (footprints come from the
            `emdat` / `gee` backends).
        lon_lim: Accepted for facade parity; see `lat_lim`.
        temporal_resolution: FLODIS issues one query over the whole window,
            so this is the sentinel `"all"`, not a pandas frequency alias.
        path: Output directory for the cached CSV and the written table.
            Created by the parent class if absent.
        fmt: `strptime` format for `start` / `end`.
        dataset: Which table to fetch — `"damages"` (EM-DAT deaths/damages,
            the default) or `"displacement"` (IDMC displacements).
        variables: FLODIS has no variable axis (`dataset=` selects a whole
            table). Accepted only so the facade can route `dataset=`; a
            non-empty value is rejected.
        country: One ISO3 country code or a list of them (`"MOZ"`,
            `["MOZ", "BGD"]`). `None` keeps every country.
        gid: One GADM code or a list of them, matched against the
            displacement table's `GID_1` / `GID_2`. Only valid with
            `dataset="displacement"` (the damages table is not GADM-keyed).
            `None` keeps every region.
        timeout: Per-request timeout in seconds for the Zenodo download.

    Raises:
        ValueError: If `dataset` is not a registered table, `variables=` is
            non-empty, a `country` value is not a 3-letter ISO3 code, or
            `gid=` is given for the non-GADM `damages` table.
    """
    self._catalog = Catalog()
    if self._catalog.record is None:
        raise ValueError(
            "the FLODIS catalog failed to load its 'record:' block; the "
            "bundled flodis_data_catalog.yaml is malformed."
        )
    self._record = self._catalog.record
    if variables:
        raise ValueError(
            "FLODIS selects a whole table with dataset= and has no variable "
            f"axis; got variables={variables!r}. Use "
            "dataset='damages' | 'displacement'."
        )
    self._dataset_name = dataset
    # Resolve against the catalog (did-you-mean hint on a typo).
    self._dataset: FlodisDataset = self._catalog.dataset(dataset)
    self._country = _normalize_iso3(country)
    self._gid = _normalize_gid(gid, dataset, self._dataset)
    self._timeout = timeout
    self._http: HttpClient | None = None

    super().__init__(
        start=cast("str", start),
        end=cast("str", end),
        # FLODIS is facet-only over the wire: the table is chosen by
        # `dataset=`, so no `variables` list is threaded to the base class.
        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 the selected FLODIS table and return it.

Runs the download + filter, writes the result to path as a CSV, 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:

Type	Description
DataFrame	pandas.DataFrame: The filtered per-event impact rows, carrying the join keys (disasterno for damages, GID_1 / GID_2 for displacement). Also written under root_dir.

Raises:

Type	Description
HTTPError	If the Zenodo download returns a non-2xx status.
ValueError	If the download's body fails its content guard (an HTML error page served with a 200 status).

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

def download(self, progress_bar: bool = True) -> pd.DataFrame:
    """Fetch the selected FLODIS table and return it.

    Runs the download + filter, writes the result to `path` as a CSV, 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:
        pandas.DataFrame: The filtered per-event impact rows, carrying the
            join keys (`disasterno` for `damages`, `GID_1` / `GID_2` for
            `displacement`). Also written under `root_dir`.

    Raises:
        requests.HTTPError: If the Zenodo download returns a non-2xx status.
        ValueError: If the download's body fails its content guard (an HTML
            error page served with a 200 status).
    """
    self._progress = progress_bar
    results = self._api()
    result = cast("pd.DataFrame", results[0])
    self._log_citation()
    out_path = self.root_dir / (self._result_stem() + ".csv")
    result.to_csv(out_path, index=False)
    logger.info(
        f"FLODIS {self._dataset_name}: {len(result)} row(s) written to {out_path}."
    )
    return result

earthlens.flodis.catalog

Catalog for the FLODIS observed-flood impacts backend.

FLODIS ships two small tabular products — damages (EM-DAT fatalities and economic damages matched to Global Flood Database footprints) and displacement (IDMC displacements matched to the same) — published on a pinned Zenodo record. This module is the bridge between the friendly request vocabulary (dataset="damages", country="MOZ") and what the release actually ships: the pinned record, the per-dataset file names and join keys, and the friendly-name -> CSV-header map.

Three shapes are modelled, all frozen:

• :class:ZenodoRecord — the pinned record, its DOI, data_period, licence and attribution. Pinning a record id rather than a moving branch is what makes a request reproducible.
• :class:FlodisDataset — one selectable table (its Zenodo file name, a description, and its join-key columns). These are the catalog's dict-surface rows, keyed by dataset name ("damages" / "displacement") under the inherited :attr:datasets field.
• the columns map — friendly name -> exact FLODIS CSV header, so the backend locates the selector / join-key / impact columns without hard-coding spellings.

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

Catalog

Bases: AbstractCatalog

Catalog for the FLODIS backend.

Reads the bundled flodis_data_catalog.yaml (shipped as package data) and exposes the pinned Zenodo record, the two selectable tables (as :class:FlodisDataset rows keyed by dataset under the inherited :attr:datasets field — the cat["damages"] / "damages" in cat / len(cat) dict surface), and the friendly-name -> CSV-header map. Instantiate with no arguments (Catalog()).

Attributes:

Name	Type	Description
datasets	dict[str, FlodisDataset]	Map from a dataset string to its :class:FlodisDataset row.
record	ZenodoRecord | None	The pinned :class:ZenodoRecord.
columns	dict[str, str]	Friendly name -> exact FLODIS CSV header.

Examples:

• List the tables, resolve one, and read the pinned record:

  >>> from earthlens.flodis import Catalog
  >>> cat = Catalog()
  >>> cat.tables()
  ['damages', 'displacement']
  >>> cat.dataset("displacement").file
  'FLODIS_displacement.csv'
  >>> "damages" in cat
  True
  >>> cat.column("disasterno")
  'disasterno'
  

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

  >>> from earthlens.flodis import Catalog
  >>> Catalog().dataset("damage")
  Traceback (most recent call last):
      ...
  ValueError: 'damage' is not in the FLODIS catalog. Known datasets: ['damages', 'displacement']. Did you mean 'damages'?
  

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

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

    Reads the bundled `flodis_data_catalog.yaml` (shipped as package data) and
    exposes the pinned Zenodo record, the two selectable tables (as
    :class:`FlodisDataset` rows keyed by `dataset` under the inherited
    :attr:`datasets` field — the `cat["damages"]` / `"damages" in cat` /
    `len(cat)` dict surface), and the friendly-name -> CSV-header map.
    Instantiate with no arguments (`Catalog()`).

    Attributes:
        datasets: Map from a `dataset` string to its :class:`FlodisDataset` row.
        record: The pinned :class:`ZenodoRecord`.
        columns: Friendly name -> exact FLODIS CSV header.

    Examples:
        - List the tables, resolve one, and read the pinned record:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> cat = Catalog()
            >>> cat.tables()
            ['damages', 'displacement']
            >>> cat.dataset("displacement").file
            'FLODIS_displacement.csv'
            >>> "damages" in cat
            True
            >>> cat.column("disasterno")
            'disasterno'

            ```
        - An unknown table raises with a did-you-mean hint:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> Catalog().dataset("damage")
            Traceback (most recent call last):
                ...
            ValueError: 'damage' is not in the FLODIS catalog. Known datasets: ['damages', 'displacement']. Did you mean 'damages'?

            ```
    """

    _catalog_kind: str = "FLODIS catalog"
    _entry_noun: str = "datasets"

    datasets: dict[str, FlodisDataset] = Field(default_factory=dict)
    #: `record` defaults 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. `None` is falsy, so the bundled record loads as intended.
    record: ZenodoRecord | 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 FLODIS 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 tables:
                ```python
                >>> from earthlens.flodis import Catalog
                >>> cat = Catalog.load()
                >>> cat.record.record
                8123096
                >>> cat.tables()
                ['damages', 'displacement']

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

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

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

        Examples:
            - The table map is keyed by the `dataset` string:
                ```python
                >>> from earthlens.flodis import Catalog
                >>> sorted(Catalog().get_catalog())
                ['damages', 'displacement']

                ```
        """
        return self.datasets

    def dataset(self, name: str) -> FlodisDataset:
        """Return the :class:`FlodisDataset` for `name`, with a did-you-mean hint.

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

        Args:
            name: A FLODIS table name (`"damages"` or `"displacement"`).

        Returns:
            FlodisDataset: The matching row.

        Raises:
            ValueError: If `name` is not a registered table.

        Examples:
            - Resolve a table and read its file name:
                ```python
                >>> from earthlens.flodis import Catalog
                >>> Catalog().dataset("damages").file
                'FLODIS_mortality_damage.csv'

                ```
        """
        return cast("FlodisDataset", self.get_dataset(name))

    def tables(self) -> list[str]:
        """Return the registered table names, sorted.

        Returns:
            list[str]: The table names (`["damages", "displacement"]`).

        Examples:
            - The registered tables come back sorted:
                ```python
                >>> from earthlens.flodis import Catalog
                >>> Catalog().tables()
                ['damages', 'displacement']

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

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

        Args:
            friendly: A friendly key from the catalog's `columns:` map
                (`"iso3"`, `"year"`, `"disasterno"`, `"gid_1"`, ...).

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

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

        Examples:
            - Map friendly keys to their exact FLODIS headers:
                ```python
                >>> from earthlens.flodis import Catalog
                >>> cat = Catalog()
                >>> cat.column("iso3")
                'ISO3'
                >>> cat.column("total_damages_000_usd")
                'total_damages_(000_USD)'

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

column(friendly)

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

Parameters:

Name	Type	Description	Default
friendly	str	A friendly key from the catalog's columns: map ("iso3", "year", "disasterno", "gid_1", ...).	required

Returns:

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

Raises:

Type	Description
KeyError	If friendly is not a mapped column.

Examples:

• Map friendly keys to their exact FLODIS headers:

  >>> from earthlens.flodis import Catalog
  >>> cat = Catalog()
  >>> cat.column("iso3")
  'ISO3'
  >>> cat.column("total_damages_000_usd")
  'total_damages_(000_USD)'
  

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

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

    Args:
        friendly: A friendly key from the catalog's `columns:` map
            (`"iso3"`, `"year"`, `"disasterno"`, `"gid_1"`, ...).

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

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

    Examples:
        - Map friendly keys to their exact FLODIS headers:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> cat = Catalog()
            >>> cat.column("iso3")
            'ISO3'
            >>> cat.column("total_damages_000_usd")
            'total_damages_(000_USD)'

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

dataset(name)

Return the :class:FlodisDataset for name, with a did-you-mean hint.

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

Parameters:

Name	Type	Description	Default
name	str	A FLODIS table name ("damages" or "displacement").	required

Returns:

Name	Type	Description
FlodisDataset	FlodisDataset	The matching row.

Raises:

Type	Description
ValueError	If name is not a registered table.

Examples:

• Resolve a table and read its file name:

  >>> from earthlens.flodis import Catalog
  >>> Catalog().dataset("damages").file
  'FLODIS_mortality_damage.csv'
  

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

def dataset(self, name: str) -> FlodisDataset:
    """Return the :class:`FlodisDataset` for `name`, with a did-you-mean hint.

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

    Args:
        name: A FLODIS table name (`"damages"` or `"displacement"`).

    Returns:
        FlodisDataset: The matching row.

    Raises:
        ValueError: If `name` is not a registered table.

    Examples:
        - Resolve a table and read its file name:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> Catalog().dataset("damages").file
            'FLODIS_mortality_damage.csv'

            ```
    """
    return cast("FlodisDataset", self.get_dataset(name))

get_catalog()

Return the table map (satisfies the abstract contract).

Returns:

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

Examples:

• The table map is keyed by the dataset string:

  >>> from earthlens.flodis import Catalog
  >>> sorted(Catalog().get_catalog())
  ['damages', 'displacement']
  

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

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

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

    Examples:
        - The table map is keyed by the `dataset` string:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> sorted(Catalog().get_catalog())
            ['damages', 'displacement']

            ```
    """
    return self.datasets

load(catalog_path=None) classmethod

Read the FLODIS 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 tables:

  >>> from earthlens.flodis import Catalog
  >>> cat = Catalog.load()
  >>> cat.record.record
  8123096
  >>> cat.tables()
  ['damages', 'displacement']
  

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

@classmethod
def load(cls, catalog_path: Path | None = None) -> Catalog:
    """Read the FLODIS 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 tables:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> cat = Catalog.load()
            >>> cat.record.record
            8123096
            >>> cat.tables()
            ['damages', 'displacement']

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

tables()

Return the registered table names, sorted.

Returns:

Type	Description
list[str]	list[str]: The table names (["damages", "displacement"]).

Examples:

• The registered tables come back sorted:

  >>> from earthlens.flodis import Catalog
  >>> Catalog().tables()
  ['damages', 'displacement']
  

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

def tables(self) -> list[str]:
    """Return the registered table names, sorted.

    Returns:
        list[str]: The table names (`["damages", "displacement"]`).

    Examples:
        - The registered tables come back sorted:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> Catalog().tables()
            ['damages', 'displacement']

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

FlodisDataset

Bases: BaseModel

One selectable FLODIS table (a row of the catalog's dict surface).

The dataset string ("damages", "displacement") is the parent key in :attr:Catalog.datasets and is not stored on the row.

Attributes:

Name	Type	Description
file	str	The file name on the Zenodo record ("FLODIS_mortality_damage.csv").
description	str	One-line human-readable summary.
key_columns	tuple[str, ...]	The join-key column(s) the table is keyed on — disasterno (EM-DAT) for damages, GID_1 / GID_2 (GADM) for displacement. A caller joins on these to the emdat (GDIS) footprints and gee (Global Flood Database) extents.

Examples:

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

  >>> from earthlens.flodis import Catalog
  >>> damages = Catalog().dataset("damages")
  >>> damages.key_columns
  ('disasterno',)
  >>> damages.content_url(8123096)
  'https://zenodo.org/api/records/8123096/files/FLODIS_mortality_damage.csv/content'
  

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

class FlodisDataset(BaseModel):
    """One selectable FLODIS table (a row of the catalog's dict surface).

    The `dataset` string (`"damages"`, `"displacement"`) is the parent key in
    :attr:`Catalog.datasets` and is not stored on the row.

    Attributes:
        file: The file name on the Zenodo record
            (`"FLODIS_mortality_damage.csv"`).
        description: One-line human-readable summary.
        key_columns: The join-key column(s) the table is keyed on — `disasterno`
            (EM-DAT) for `damages`, `GID_1` / `GID_2` (GADM) for `displacement`.
            A caller joins on these to the `emdat` (GDIS) footprints and `gee`
            (Global Flood Database) extents.

    Examples:
        - The content URL is composed from the pinned record and file name:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> damages = Catalog().dataset("damages")
            >>> damages.key_columns
            ('disasterno',)
            >>> damages.content_url(8123096)
            'https://zenodo.org/api/records/8123096/files/FLODIS_mortality_damage.csv/content'

            ```
    """

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

    file: str
    description: str = ""
    key_columns: tuple[str, ...] = ()

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

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

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

        Examples:
            - Compose the REST content URL for a table on a record:
                ```python
                >>> from earthlens.flodis import FlodisDataset
                >>> FlodisDataset(file="FLODIS_displacement.csv").content_url(8123096)
                'https://zenodo.org/api/records/8123096/files/FLODIS_displacement.csv/content'

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

content_url(record)

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

Parameters:

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

Returns:

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

Examples:

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

  >>> from earthlens.flodis import FlodisDataset
  >>> FlodisDataset(file="FLODIS_displacement.csv").content_url(8123096)
  'https://zenodo.org/api/records/8123096/files/FLODIS_displacement.csv/content'
  

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

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

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

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

    Examples:
        - Compose the REST content URL for a table on a record:
            ```python
            >>> from earthlens.flodis import FlodisDataset
            >>> FlodisDataset(file="FLODIS_displacement.csv").content_url(8123096)
            'https://zenodo.org/api/records/8123096/files/FLODIS_displacement.csv/content'

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

ZenodoRecord

Bases: BaseModel

The pinned Zenodo record FLODIS is fetched from.

Attributes:

Name	Type	Description
record	int	The pinned Zenodo record id (8123096). Every file URL is composed from it, so a request is reproducible.
concept_doi	str	The dataset DOI as cited in the paper. Recorded for the citation and a future refresh check; never used to fetch.
version	str	The dataset version label, if the record carries one.
data_period	str	The first-last year span the record covers ("2000-2018").
license	str	SPDX-ish licence id (CC-BY-4.0).
attribution	str	The citation obligation the licence carries.

Examples:

• The record is the pinned id, and the licence is CC-BY-4.0:

  >>> from earthlens.flodis import Catalog
  >>> Catalog().record.record
  8123096
  >>> Catalog().record.license
  'CC-BY-4.0'
  

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

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

    Attributes:
        record: The pinned Zenodo record id (`8123096`). Every file URL is
            composed from it, so a request is reproducible.
        concept_doi: The dataset DOI as cited in the paper. Recorded for the
            citation and a future refresh check; never used to fetch.
        version: The dataset version label, if the record carries one.
        data_period: The `first-last` year span the record covers (`"2000-2018"`).
        license: SPDX-ish licence id (`CC-BY-4.0`).
        attribution: The citation obligation the licence carries.

    Examples:
        - The record is the pinned id, and the licence is CC-BY-4.0:
            ```python
            >>> from earthlens.flodis import Catalog
            >>> Catalog().record.record
            8123096
            >>> Catalog().record.license
            'CC-BY-4.0'

            ```
    """

    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/flodis/catalog.py

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