Pluvial flood risk

This example shows a workflow to derive pluvial flood risk using the SFINCS and Delft-FIAT models. The starting point is a user defined region and data catalog. Rainfall IDF curves from the GPEX dataset are translated into design events (hyetographs) for different return periods and used to simulate the flood hazard maps. The hazard maps are combined with exposure and impact data to derive risk.

# Import packages and setup logging
from pathlib import Path

from hydroflows import Workflow, WorkflowConfig
from hydroflows.log import setuplog
from hydroflows.methods import fiat, flood_adapt, rainfall, sfincs
from hydroflows.methods.utils.example_data import fetch_data

logger = setuplog(level="INFO")

INFO - log - hydroflows version: 0.1.0

# General setup of workflow
name = "pluvial_risk"
pwd = Path().resolve()  # Get the current file location
case_root = Path(pwd, "cases", name)  # output directory
pwd_rel = "../../"  # relative path from the case directory to the current file

Workflow inputs

The example requires the following inputs which are provided via a configuration file:

• a user defined region that can be used to delineate the SFINCS model domain

• a data catalog file describing all input datasets. Here we fetch some test datasets for a region in Northern Italy.

• The GPEX IDF dataset. Note that rainfall timeseries data can also be used to derive design events.

• HydroMT configuration files for both models

• model executables (docker is also possible for SFINCS)

# Fetch the test global-data data catalog
cache_dir = fetch_data(data="global-data")

# Setup the config file
config = WorkflowConfig(
    # general settings
    region=Path(pwd_rel, "data/build/region.geojson"),
    catalog_path=Path(cache_dir, "data_catalog.yml"),
    plot_fig=True,
    # sfincs settings
    hydromt_sfincs_config=Path(pwd_rel, "hydromt_config/sfincs_config.yml"),
    sfincs_exe=Path(pwd_rel, "bin/sfincs_v2.1.1/sfincs.exe"),
    # fiat settings
    hydromt_fiat_config=Path(pwd_rel, "hydromt_config/fiat_config.yml"),
    fiat_exe=Path(pwd_rel, "bin/fiat_v0.2.0/fiat.exe"),
    risk=True,
    # rainfall and design events settings
    start_date="2000-01-01",
    end_date="2021-12-31",
    rps=[2, 5, 10, 50, 100],
    # delta temp climate scenarios
    dt_scenarios={"present": 0, "rcp85_2050": 1.5},
)

Create the workflow

# create and empty workflow
wf = Workflow(name=name, config=config, root=case_root)

Build models

In this section we build a model cascade and make sure these are configured correctly for offline coupling, i.e. Delft-FIAT uses the same ground elevation as SFINCS. Note that you can also skip these steps and use your own models instead.

First, we build a SFINCS model for the user defined region using.

• setting from the hydromt_sfincs_config, see the HydroMT-SFINCS docs for more info.

• data from the catalog_path, see the HydroMT docs for more info.

# Build a SFINCS model
sfincs_build = sfincs.SfincsBuild(
    region=wf.get_ref("$config.region"),
    sfincs_root="models/sfincs",
    config=wf.get_ref("$config.hydromt_sfincs_config"),
    catalog_path=wf.get_ref("$config.catalog_path"),
    plot_fig=wf.get_ref("$config.plot_fig"),
    subgrid_output=True,  # save subgrid output for subsequent methods
)
wf.create_rule(sfincs_build, rule_id="sfincs_build")

Rule(id=sfincs_build, method=sfincs_build, runs=1)

Next, we build a FIAT model using:

• the sfincs_build output for the model region and ground elevation

• settings from the hydromt_fiat_config, see the hydromt_fiat docs

• data from the data catalog

# Build a Delft-FIAT model
fiat_build = fiat.FIATBuild(
    region=sfincs_build.output.sfincs_region,
    ground_elevation=sfincs_build.output.sfincs_subgrid_dep,
    fiat_root="models/fiat",
    catalog_path=wf.get_ref("$config.catalog_path"),
    config=wf.get_ref("$config.hydromt_fiat_config"),
)
wf.create_rule(fiat_build, rule_id="fiat_build")

Rule(id=fiat_build, method=fiat_build, runs=1)

Derive pluvial design events

Here, we define pluvial design events from GPEX IDF data using the alternating block method. We derive the IDF curves for the centroid of the SFINCS model region. Alternatively, you can use the PluvialDesignEvents class to define pluvial events from a rainfall time series.

pluvial_events = rainfall.PluvialDesignEventsGPEX(
    gpex_nc=Path(cache_dir, "gpex.nc"),
    region=sfincs_build.output.sfincs_region,
    event_root="data/events/default",
    rps=wf.get_ref("$config.rps"),
    wildcard="events", # wildcard to use for the pluvial events
)

# Note that a new "events" wildcard is created for the events
wf.create_rule(pluvial_events, rule_id="pluvial_events")

INFO - wildcards - Added wildcard 'events' with values: ['p_event_rp002', 'p_event_rp005', 'p_event_rp010', 'p_event_rp050', 'p_event_rp100']

Rule(id=pluvial_events, method=pluvial_design_events_GPEX, runs=1, expand=['events'])

The events are scaled for future climate predictions based on the Clausius-Clapeyron relation. A new wildcard for scenarios is introduced develop hazard maps for each event per scenario.

future_pluvial_events = rainfall.FutureClimateRainfall(
    scenarios=wf.get_ref("$config.dt_scenarios"),
    event_set_yaml=pluvial_events.output.event_set_yaml,
    event_names=pluvial_events.params.event_names,
    event_root="data/events",
    event_wildcard="event",
    scenario_wildcard="scenario",
)
wf.create_rule(future_pluvial_events, rule_id="future_pluvial_events")

INFO - wildcards - Added wildcard 'event' with values: ['p_event_rp002', 'p_event_rp005', 'p_event_rp010', 'p_event_rp050', 'p_event_rp100']
INFO - wildcards - Added wildcard 'scenario' with values: ['present', 'rcp85_2050']

Rule(id=future_pluvial_events, method=future_climate_rainfall, runs=1, expand=['scenario', 'event'])

Derive flood hazard

To derive flood hazard maps for each event, we

1. Update the SFINCS model using the rainfall event timeseries. This will create new SFINCS instances for each event.

2. Run the SFINCS model. This will create simulated water levels for each event.

3. Postprocess the SFINCS output. This will postprocess the SFINCS results to a regular grid of maximum water levels.

4. Optionally, downscale the SFINCS output. This will downscale the max simulated SFINCS water levels to a high-res flood depth map.

# Update the SFINCS model with pluvial events
sfincs_update = sfincs.SfincsUpdateForcing(
    sfincs_inp=sfincs_build.output.sfincs_inp,
    event_yaml=future_pluvial_events.output.future_event_yaml,
    output_dir=sfincs_build.output.sfincs_inp.parent/"sim_{scenario}"/"{event}"
)
wf.create_rule(sfincs_update, rule_id="sfincs_update")

Rule(id=sfincs_update, method=sfincs_update_forcing, runs=10, repeat=['scenario', 'event'])

# Run the SFINCS model for each pluvial event
sfincs_run = sfincs.SfincsRun(
    sfincs_inp=sfincs_update.output.sfincs_out_inp,
    sfincs_exe=wf.get_ref("$config.sfincs_exe"),
    run_method="exe", # alternatively use "docker" to run in a docker container
)
wf.create_rule(sfincs_run, rule_id="sfincs_run")

Rule(id=sfincs_run, method=sfincs_run, runs=10, repeat=['scenario', 'event'])

# Postprocesses SFINCS results
sfincs_post = sfincs.SfincsPostprocess(
    sfincs_map=sfincs_run.output.sfincs_map,
    output_root=sfincs_run.output.sfincs_map.parent
)
wf.create_rule(sfincs_post, rule_id="sfincs_post")

Rule(id=sfincs_post, method=sfincs_postprocess, runs=10, repeat=['scenario', 'event'])

# Downscale the SFINCS waterlevels to high-resolution water
sfincs_downscale = sfincs.SfincsDownscale(
    sfincs_map=sfincs_run.output.sfincs_map,
    sfincs_subgrid_dep=sfincs_build.output.sfincs_subgrid_dep,
    output_root="output/hazard/{scenario}/{event}",
)

Derive flood risk

To calculate flood risk, we

• Update Delft-FIAT with all pluvial events which are combined in an event set. This will create a new Delft-FIAT instance for the event set.

• Run Delft-FIAT to calculate flood impact and risk. This will create impact and risk data at the individual and aggregated asset level.

• Visualize the risk results at the aggregated asset level.

# Update FIAT hazard forcing with the pluvial eventset to compute pluvial flood risk
fiat_update = fiat.FIATUpdateHazard(
    fiat_cfg=fiat_build.output.fiat_cfg,
    event_set_yaml=future_pluvial_events.output.future_event_set_yaml,
    map_type="water_level",
    hazard_maps=sfincs_post.output.sfincs_zsmax,
    risk=wf.get_ref("$config.risk"),
    output_dir=fiat_build.output.fiat_cfg.parent/"sim_{scenario}"
)
wf.create_rule(fiat_update, rule_id="fiat_update")

Rule(id=fiat_update, method=fiat_update_hazard, runs=2, repeat=['scenario'], reduce=['event'])

# Run FIAT to compute pluvial flood risk
fiat_run = fiat.FIATRun(
    fiat_cfg=fiat_update.output.fiat_out_cfg,
    fiat_exe=wf.get_ref("$config.fiat_exe"),
    run_method="exe",
)
wf.create_rule(fiat_run, rule_id="fiat_run")

Rule(id=fiat_run, method=fiat_run, runs=2, repeat=['scenario'])

# Visualize Fiat
fiat_visualize = fiat.FIATVisualize(
    fiat_cfg=fiat_update.output.fiat_out_cfg,
    fiat_output_csv=fiat_run.output.fiat_out_csv,
    spatial_joins_cfg=fiat_build.output.spatial_joins_cfg, # aggregation level for visualization
    output_dir="output/risk/{scenario}",
)
wf.create_rule(fiat_visualize, rule_id="fiat_visualize")

Rule(id=fiat_visualize, method=fiat_visualize, runs=2, repeat=['scenario'])

Prepare FloodAdapt database input

Optionally, a FloodAdapt instance can be created using the SFINCS and Delft-FIAT models and the derived pluvial event set. The SetupFloodAdapt method prepares the models and a configuration file which can readily be used by the FloodAdapt database builder.

# %% Prepare Sfincs Models for FloodAdapt
prep_sfincs_run = flood_adapt.PrepSfincsModels(
    sfincs_inp=sfincs_build.output.sfincs_inp,
)
wf.create_rule(prep_sfincs_run, rule_id="prep_sfincs_model")

# %% Prepare FloodAdapt
setup_floodadapt = flood_adapt.SetupFloodAdapt(
    fiat_cfg=fiat_build.output.fiat_cfg,
    sfincs_inp=prep_sfincs_run.output.sfincs_out_inp,
    db_name= "Pluvial_risk",
    description= "This is a pluvial risk event set database",
    event_set_yaml=pluvial_events.output.event_set_yaml,
    output_dir=Path("output/floodadapt"),
)
wf.create_rule(setup_floodadapt, rule_id="setup_floodadapt")

Rule(id=setup_floodadapt, method=setup_flood_adapt, runs=1)

Visualize and execute the workflow

To inspect the workflow we can plot the rulegraph which shows all rules their dependencies. The nodes are colored based on the type, for instance the red nodes show the result rules.

# plot the rulegraph using graphviz
wf.plot_rulegraph(filename="rulegraph.svg", plot_rule_attrs=True)

# dryrun workflow. Make sure no warnings are raised
wf.dryrun()

INFO - workflow - Dryrun rule 1/12: sfincs_build (1 runs)
INFO - workflow - Dryrun rule 2/12: prep_sfincs_model (1 runs)
INFO - workflow - Dryrun rule 3/12: pluvial_events (1 runs)
INFO - workflow - Dryrun rule 4/12: future_pluvial_events (1 runs)
INFO - workflow - Dryrun rule 5/12: fiat_build (1 runs)
INFO - workflow - Dryrun rule 6/12: setup_floodadapt (1 runs)
INFO - workflow - Dryrun rule 7/12: sfincs_update (10 runs)
INFO - workflow - Dryrun rule 8/12: sfincs_run (10 runs)
INFO - workflow - Dryrun rule 9/12: sfincs_post (10 runs)
INFO - workflow - Dryrun rule 10/12: fiat_update (2 runs)
INFO - workflow - Dryrun rule 11/12: fiat_run (2 runs)
INFO - workflow - Dryrun rule 12/12: fiat_visualize (2 runs)

The workflow can be executed using HydroFlows or a workflow engine. To run the workflow in HydroFlows use wf.run(). To run the workflow with SnakeMake (preferred) use wf.to_snakemake() to create a snakemake file, see below. You can then use the Snakemake CLI to execute the workflow, see the snakemake documentation

# Write the workflow to a Snakefile
wf.to_snakemake()

# show the files in the case directory
print(f"{wf.root.relative_to(pwd)}:")
for f in wf.root.iterdir():
    if f.is_file():
        print(f"- {f.name}")

cases/pluvial_risk:
- Snakefile
- rulegraph.svg
- Snakefile.config.yml

# uncomment to run the workflow with snakemake
# import subprocess
# subprocess.run(["snakemake", "-c", "1"], cwd=wf.root)