Pipelines

The pipelines split into consumer (query the index) and producer (build the index). Most users only touch the consumer path indirectly, via floods() / hazard().

Consumer

The query pipelines behind the public API.

DiscoveryPipeline

DiscoveryPipeline(settings: Settings | None = None)

Resolve a region and query the index into a FloodFrame catalogue (cheap).

Build the resolver + dictionary repository; validate the cache manifest.

Methods:

Name	Description
`query`	Query the index for a region/time -> a FloodFrame (no rasters fetched).

Methods:

query

query(
    region: Any = None,
    *,
    point: tuple[float, float] | None = None,
    radius_m: float = 0.0,
    bbox: tuple[float, float, float, float] | None = None,
    shapefile: str | Path | None = None,
    buffer_m: float = 0.0,
    year: int | None = None,
    start: str | int | None = None,
    end: str | int | None = None,
    level: int | None = None,
    shape: str = "exact",
    output_dir: str | Path | None = None
) -> FloodFrame

Query the index for a region/time -> a FloodFrame (no rasters fetched).

Backs floods — see it for the argument reference. output_dir is the directory the cached-download auto-detect scans (defaults to settings.output_dir); pass the same dir you download to so a re-query of a previously-downloaded area is immediately actionable.

Returns:

Type	Description
`FloodFrame`	A historic `FloodFrame` (one row per flood event); empty if the
`FloodFrame`	ROI is outside the index or contains no floods.

Raises:

Type	Description
`FileNotFoundError`	If no local index exists and no remote index is configured.

HazardPipeline

HazardPipeline(settings: Settings | None = None)

Resolve a region and query the GLOFAS tile index into a FloodFrame (cheap).

Build the resolver and tile index.

Methods:

Name	Description
`query`	Query GLOFAS hazard tiles for a region -> a FloodFrame (no rasters fetched).

Methods:

query

query(
    region: Any = None,
    *,
    point: tuple[float, float] | None = None,
    radius_m: float = 0.0,
    bbox: tuple[float, float, float, float] | None = None,
    shapefile: str | Path | None = None,
    buffer_m: float = 0.0,
    return_period: int | list[int] | None = None,
    level: int | None = None,
    shape: str = "exact",
    output_dir: str | Path | None = None
) -> FloodFrame

Query GLOFAS hazard tiles for a region -> a FloodFrame (no rasters fetched).

Backs hazard — see it for the argument reference. output_dir is the directory the cached-download auto-detect scans (defaults to settings.output_dir).

Returns:

Type	Description
`FloodFrame`	A hazard `FloodFrame` (one row per
`FloodFrame`	return period).

Raises:

Type	Description
`GeocodingError`	If a place name cannot be resolved.
`HazardError`	If the hazard tile index cannot be located or read.

Producer

The build pipelines (see the HPC Runbook).

MirrorPipeline

MirrorPipeline(settings: Settings | None = None)

Download all source tiles to the cache (download-only, resumable).

Build the scraper, inventory, and downloader; ensure cache dirs exist.

Methods:

Name	Description
`plan`	Return a side-effect-free summary of what `run()` would do (for --dry-run).
`run`	Download all (optionally year-filtered) source tiles to the cache.

Methods:

plan

plan(
    year: int | None = None, *, limit: int | None = None
) -> dict[str, Any]

Return a side-effect-free summary of what run() would do (for --dry-run).

run

run(
    year: int | None = None,
    *,
    update: bool = False,
    verify: bool = False,
    retry_failed: bool = False,
    limit: int | None = None
) -> int

Download all (optionally year-filtered) source tiles to the cache.

Parameters:

Name	Type	Description	Default
`year`	`int \| None`	Mirror only this year; `None` mirrors all years.	`None`
`update`	`bool`	Force a fresh inventory scrape first.	`False`
`verify`	`bool`	Re-download any cached file whose on-disk size differs from the size JRC lists for it (catches truncated/corrupt cache files).	`False`
`retry_failed`	`bool`	Only re-attempt files in the dead-letter ledger.	`False`
`limit`	`int \| None`	Only handle the first `limit` tiles (stable; for small test runs).	`None`

Returns:

Name	Type	Description
`int`	`int`	The number of tiles available in the cache after the run.

IngestionPipeline

IngestionPipeline(settings: Settings | None = None)

Orchestrates the massive Download -> Process workflow.

This class serves as the controller, instantiating necessary services and managing the flow of data between them. It ensures that the cache directories are prepared and handles the parallel execution logic.

Initiate the Ingestion Pipeline.

Initializes the following internal services: - Scraper: For finding files on the web. - Inventory: For tracking what is available. - Downloader: For reliable file transfer. - Processor: For converting TIFs to Parquet.

It also ensures that settings.cache_dir and subdirectories exist.

Methods:

Name	Description
`update_inventory`	Scrapes the remote website and updates the local inventory CSV.
`plan`	Side-effect-free summary of what `run()` would process (for --dry-run).
`run`	Execute the ingestion pipeline with optional filters.

Methods:

update_inventory

update_inventory() -> None

Scrapes the remote website and updates the local inventory CSV.

This is a blocking network operation. It fetches the directory listing for every year available on the JRC server and saves the metadata (filenames, dates, URLs) to inventory.csv.

Logs

Info: When scraping starts and finishes.
Error: If specific year directories cannot be accessed.

plan

plan(
    year: int | None = None,
    month: str | None = None,
    *,
    limit: int | None = None
) -> dict[str, Any]

Side-effect-free summary of what run() would process (for --dry-run).

run

run(
    year: int | None = None,
    month: str | None = None,
    update: bool = False,
    limit: int | None = None,
) -> None

Execute the ingestion pipeline with optional filters.

This is the main entry point. It checks the local inventory, filters it based on the provided criteria, and queues up the work.

Parameters:

Name	Type	Description	Default
`year`	`Optional[int]`	The year to process (e.g., 2020). If None, ALL years in the inventory are processed (use with caution).	`None`
`month`	`Optional[str]`	The month to filter by (e.g., "05" or "5"). Effective only if `year` is also provided.	`None`
`update`	`bool`	If True, forces a fresh scrape of the JRC website to update `inventory.csv` before processing starts. Defaults to False.	`False`
`limit`	`Optional[int]`	If set, only the first N tiles (stable order) are handled — for small test runs. Defaults to None (all).	`None`

Process

Inventory Check: Loads the CSV. If empty or update=True, runs scraping.
Filtering: Narrows down the list of files to process.
Downloading: Runs DownloadService.download_file in threads.
Processing: Runs RasterProcessor.process in separate processes.

Notes

The pipeline is robust to individual file failures. If a single TIF is corrupt, it logs the error and continues processing the rest of the batch.
Intermediate files (downloads) are kept in the cache to avoid re-downloading if the pipeline is restarted.

ExportPipeline

ExportPipeline(settings: Settings | None = None)

Aggregate flood occurrences into the Index COG + Parquet dictionary.

Set up the disk-backed DuckDB connection (spills if over RAM).

Methods:

Name	Description
`run`	Materialize the aggregation, write the COG, dictionary, and manifest.

Methods:

run

run() -> None

Materialize the aggregation, write the COG, dictionary, and manifest.

IndexDoctor

IndexDoctor(settings: Settings | None = None)

Validate + report on a built index bundle.

Bind to a settings object (defaults to get_settings).

Methods:

Name	Description
`run`	Run all checks; raise on failure, else return a report dict.

Methods:

run

run() -> dict[str, Any]

Run all checks; raise on failure, else return a report dict.