Skip to content

inland_flood_analysis

inland_flood_analysis

Classes

InlandFloodAnalysis 

InlandFloodAnalysis(raster_collection: RasterCollection, buildings: Any, vulnerability: AbstractVulnerabilityFunction, calculate_aal: bool = True, aal_rate_limits: Optional[Tuple[float, float]] = None, aal_truncation: int = 0)

Vectorized inland flood analysis orchestrator.

This class expects a raster_input mapping of return_period -> raster-like object where each raster implements get_value_vectorized(geometries) -> array-like depths. The buildings object must expose a pandas Geo/DataFrame at buildings.gdf and contain a column with monetary values (building_value_col). The vulnerability object must implement calculate_vulnerability(exposure_df) and return a DataFrame of damage ratios with the same shape as exposure_df.

Damage function matching is configured via wildcard_fields on the :class:~inland_consequences.InlandFloodVulnerability instance.

Source code in src/inland_consequences/inland_flood_analysis.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def __init__(
    self,
    raster_collection: RasterCollection,
    buildings: Any,
    vulnerability: AbstractVulnerabilityFunction,
    calculate_aal: bool = True,
    aal_rate_limits: Optional[Tuple[float, float]] = None,
    aal_truncation: int = 0,
) -> None:
    # Must be a RasterCollection instance (validated by its constructor)
    if not isinstance(raster_collection, RasterCollection):
        raise TypeError("raster_collection must be a RasterCollection instance")
    self.conn = None  # type: duckdb.DuckDBPyConnection | None
    self.db_path = None  # type: str | None
    self.raster_collection = raster_collection
    self.buildings = buildings
    self.vulnerability: AbstractVulnerabilityFunction = vulnerability
    self.calculate_aal = calculate_aal
    self.aal_rate_limits = aal_rate_limits
    self.aal_truncation = aal_truncation

    # Minimal validation
    if not hasattr(self.buildings, "gdf"):
        raise ValueError("buildings must have a .gdf attribute containing building rows")
Functions
__enter__ 
__enter__()

Establishes the persistent connection.

Source code in src/inland_consequences/inland_flood_analysis.py 
 95
 96
 97
 98
 99
100
101
102
103
104
def __enter__(self):
    """Establishes the persistent connection."""
    if self.conn is not None:
        raise RuntimeError("DataProcessor context manager is not re-entrant.")

    db_id = self._get_db_identifier()
    self.db_path = db_id  # Store path for later reference
    self.conn = duckdb.connect(database=db_id, config={'storage_compatibility_version': 'latest'})
    self._setup_logging(db_id)
    return self
__exit__ 
__exit__(exc_type, exc_val, exc_tb)

Closes the persistent connection.

Source code in src/inland_consequences/inland_flood_analysis.py 
106
107
108
109
110
def __exit__(self, exc_type, exc_val, exc_tb):
    """Closes the persistent connection."""
    if self.conn:
        self.conn.close()
    self.conn = None # Reset the connection attribute