Combined pluvial and fluvial flood risk#

This example shows a workflow to derive pluvial and fluvial flood risk using the Wflow, SFINCS and Delft-FIAT models. The goal of the example is to show how to combine different event sets in the workflow. Note that both types of flood risk are assessed independently. Additional statistical analysis are needed to derive compound flood risk estimates. For more details about each step we refer the user to the pluvial flood risk or fluvial flood risk examples.

[1]:

# Import packages
from pathlib import Path

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

logger = setuplog(level="INFO")

INFO - log - hydroflows version: 0.1.0

[2]:

# Define case name and root directory
name = "pluvial_fluvial_hazard"
pwd = Path().resolve()  # Get the current file location
case_root = Path(pwd, "cases", name)  # output directory
pwd_rel = "../../"

Workflow inputs#

[3]:

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

Downloading file 'global-data.tar.gz' from 'doi:10.5281/zenodo.14967510/global-data.tar.gz' to '/home/runner/.cache/hydroflows'.
Untarring contents of '/home/runner/.cache/hydroflows/global-data.tar.gz' to '/home/runner/.cache/hydroflows/global-data'

[4]:

config = WorkflowConfig(
    # general settings
    region=Path(pwd, "data/build/region.geojson"),
    catalog_path=Path(cache_dir, "data_catalog.yml"),
    plot_fig=True,
    start_date="2014-01-01",
    end_date="2021-12-31",
    # sfincs settings
    hydromt_sfincs_config=Path(pwd, "hydromt_config/sfincs_config.yml"),
    sfincs_exe=Path(pwd, "bin/sfincs_v2.1.1/sfincs.exe"),
    subgrid_output=True,
    src_points_output=True,
    # wflow settings
    hydromt_wflow_config=Path(pwd, "hydromt_config/wflow_config.yml"),
    wflow_exe=Path(pwd, "bin/wflow_v0.8.1/bin/wflow_cli.exe"),
    # fiat settings
    hydromt_fiat_config=Path(pwd, "hydromt_config/fiat_config.yml"),
    fiat_exe=Path(pwd, "bin/fiat_v0.2.1/fiat.exe"),
    risk=True,
    # design events settings
    rps=[5, 10, 25],
)

Create the workflow#

[5]:

# create and empty workflow
wf = Workflow(config=config, name=name, 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. Wflow exports discharge at the right locations and Delft-FIAT uses the same ground elevation as SFINCS. Note that you can also skip these steps and use your own models instead.

[6]:

# Build SFINCS model
sfincs_build = sfincs.SfincsBuild(
    region=wf.get_ref("$config.region"),
    config=wf.get_ref("$config.hydromt_sfincs_config"),
    sfincs_root="models/sfincs",
    catalog_path=wf.get_ref("$config.catalog_path"),
    plot_fig=wf.get_ref("$config.plot_fig"),
    subgrid_output=wf.get_ref("$config.subgrid_output"),
    src_points_output=wf.get_ref("$config.src_points_output"),
)
wf.create_rule(sfincs_build, rule_id="sfincs_build")

[6]:

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

[7]:

# Wflow build
wflow_build = wflow.WflowBuild(
    region=sfincs_build.output.sfincs_region,
    wflow_root="models/wflow",
    config=wf.get_ref("$config.hydromt_wflow_config"),
    catalog_path=wf.get_ref("$config.catalog_path"),
    gauges=sfincs_build.output.sfincs_src_points,
    plot_fig=wf.get_ref("$config.plot_fig"),
)
wf.create_rule(wflow_build, rule_id="wflow_build")

[7]:

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

[8]:

# Fiat build
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")

[8]:

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

Derive fluvial design events from Wflow#

In this section we derive fluvial design events for each Wflow output gauge location.

[9]:

# Update forcing
wflow_update = wflow.WflowUpdateForcing(
    wflow_toml=wflow_build.output.wflow_toml,
    catalog_path=wf.get_ref("$config.catalog_path"),
    start_time=wf.get_ref("$config.start_date"),
    end_time=wf.get_ref("$config.end_date"),
    output_dir=wflow_build.output.wflow_toml.parent/"simulations"/"default"
)
wf.create_rule(wflow_update, rule_id="wflow_update")

# Run wflow
wflow_run = wflow.WflowRun(
    wflow_toml=wflow_update.output.wflow_out_toml,
    wflow_bin=wf.get_ref("$config.wflow_exe"),
)
wf.create_rule(wflow_run, rule_id="wflow_run")

# Generate fluvial events
fluvial_events = discharge.FluvialDesignEvents(
    discharge_nc=wflow_run.output.wflow_output_timeseries,
    rps=wf.get_ref("$config.rps"),
    wildcard="fluvial_events",
    event_root="data/events",
    index_dim="Q_gauges_bounds",
)
wf.create_rule(fluvial_events, rule_id="fluvial_events")

INFO - wildcards - Added wildcard 'fluvial_events' with values: ['q_event_rp005', 'q_event_rp010', 'q_event_rp025']

[9]:

Rule(id=fluvial_events, method=fluvial_design_events, runs=1, expand=['fluvial_events'])

Derive pluvial design events#

Here we derive pluvial design events from ERA5 rainfall timeseries data.

[10]:

# Pluvial events
pluvial_data = rainfall.GetERA5Rainfall(
    region=sfincs_build.output.sfincs_region,
    output_dir="data/era5",
    start_date=wf.get_ref("$config.start_date"),
    end_date=wf.get_ref("$config.end_date"),
)
wf.create_rule(pluvial_data, rule_id="pluvial_data")

# Actual derivation of events based on precip
pluvial_events = rainfall.PluvialDesignEvents(
    precip_nc=pluvial_data.output.precip_nc,
    rps=wf.get_ref("$config.rps"),
    wildcard="pluvial_events",
    event_root="data/events",
)
wf.create_rule(pluvial_events, rule_id="pluvial_events")

INFO - wildcards - Added wildcard 'pluvial_events' with values: ['p_event_rp005', 'p_event_rp010', 'p_event_rp025']

[10]:

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

Combine event sets#

In order to run the SFINCS model for all events (pluvial and fluvial) and the Delft-FIAT model for both event sets, we need to combine all events into a new {all_events} wildcard and create a new {event_set} wildcard for both event sets.

[11]:

# Combine fluvial and pluvial events into one set in order to run SFINCS for all in parallel
all_events = wf.wildcards.get("pluvial_events") + wf.wildcards.get("fluvial_events")
wf.wildcards.set("all_events", all_events)

# Set the combined event_set of both fluvial and pluvial
wf.wildcards.set("event_set", ["fluvial_design_events", "pluvial_design_events"])

INFO - wildcards - Added wildcard 'all_events' with values: ['p_event_rp005', 'p_event_rp010', 'p_event_rp025', 'q_event_rp005', 'q_event_rp010', 'q_event_rp025']
INFO - wildcards - Added wildcard 'event_set' with values: ['fluvial_design_events', 'pluvial_design_events']

Derive flood hazard#

Here we update, run, and postprocess SFINCS for all events (combined pluvial and fluvial)

[12]:

# Updating, running and postprocessing SFINCS model for combined events
sfincs_update = sfincs.SfincsUpdateForcing(
    sfincs_inp=sfincs_build.output.sfincs_inp,
    event_yaml="data/events/{all_events}.yml",
    output_dir=sfincs_build.output.sfincs_inp.parent/"simulations"/"{all_events}"
)
wf.create_rule(sfincs_update, rule_id="sfincs_update")

# Run SFINCS model
sfincs_run = sfincs.SfincsRun(
    sfincs_inp=sfincs_update.output.sfincs_out_inp,
    sfincs_exe=wf.get_ref("$config.sfincs_exe"),
)
wf.create_rule(sfincs_run, rule_id="sfincs_run")

# Postprocesses SFINCS results
sfincs_post = sfincs.SfincsPostprocess(
    sfincs_map=sfincs_run.output.sfincs_map,
    event_name="{all_events}",
)
wf.create_rule(sfincs_post, rule_id="sfincs_post")

[12]:

Rule(id=sfincs_post, method=sfincs_postprocess, runs=6, repeat=['all_events'])

Derive flood risk#

Here we calculate flood risk separately for each event set (fluvial and pluvial).

[13]:

# Update hazard
fiat_update = fiat.FIATUpdateHazard(
    fiat_cfg=fiat_build.output.fiat_cfg,
    event_set_yaml="data/events/{event_set}.yml",
    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/"simulations"/"{event_set}"
)
wf.create_rule(fiat_update, rule_id="fiat_update")

# Run FIAT
fiat_run = fiat.FIATRun(
    fiat_cfg=fiat_update.output.fiat_out_cfg,
    fiat_exe=wf.get_ref("$config.fiat_exe"),
)
wf.create_rule(fiat_run, rule_id="fiat_run")

# 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"
)
wf.create_rule(fiat_visualize, rule_id="fiat_visualize")

[13]:

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

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.

[14]:

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

[14]:

../_images/_examples_pluvial_fluvial_risk_24_0.svg

[15]:

# Test the workflow
wf.dryrun()

INFO - workflow - Dryrun rule 1/14: sfincs_build (1 runs)
INFO - workflow - Dryrun rule 2/14: pluvial_data (1 runs)
INFO - workflow - Dryrun rule 3/14: pluvial_events (1 runs)
INFO - workflow - Dryrun rule 4/14: fiat_build (1 runs)
INFO - workflow - Dryrun rule 5/14: wflow_build (1 runs)
INFO - workflow - Dryrun rule 6/14: wflow_update (1 runs)
INFO - workflow - Dryrun rule 7/14: wflow_run (1 runs)
INFO - workflow - Dryrun rule 8/14: fluvial_events (1 runs)
INFO - workflow - Dryrun rule 9/14: sfincs_update (6 runs)
INFO - workflow - Dryrun rule 10/14: sfincs_run (6 runs)
INFO - workflow - Dryrun rule 11/14: sfincs_post (6 runs)
INFO - workflow - Dryrun rule 12/14: fiat_update (2 runs)
INFO - workflow - Dryrun rule 13/14: fiat_run (2 runs)
INFO - workflow - Dryrun rule 14/14: 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

[16]:

# 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():
    print(f"- {f.name}")

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

[17]:

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