Fluvial flood risk#

This example shows a workflow to derive fluvial flood risk using the Wflow, SFINCS and Delft-FIAT models. The starting point is a user defined region and data catalog. Wflow simulated discharge is translated into hydrographs 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.

This example also show how to work with parsing to CWL.

[1]:

# Import packages
from pathlib import Path

from hydroflows import Workflow, WorkflowConfig
from hydroflows.log import setuplog
from hydroflows.methods import discharge, fiat, 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 = "fluvial_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.

  • HydroMT configuration files for all three models

  • how to execute models. Since the CWL runner we use is not supported on Windows, we opt for docker for WFLOW, SFINCS and python for Delft-FIAT.

[3]:

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

[4]:

# Setup the configuration
config = WorkflowConfig(
    # general settings
    region=Path(pwd_rel, "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_rel, "hydromt_config/sfincs_config.yml"),
    sfincs_run_method="docker",
    depth_min=0.05,  # minimum depth for inundation map
    # fiat settings
    hydromt_fiat_config=Path(pwd_rel, "hydromt_config/fiat_config.yml"),
    fiat_run_method="python",
    # wflow settings
    hydromt_wflow_config=Path(pwd_rel, "hydromt_config/wflow_config.yml"),
    wflow_run_method="docker",
    # design event settings
    rps=[2, 5, 10, 50, 100],
)

Create the workflow#

[5]:

# 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. 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.

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.

  • Note that we need src points at the boundary of the SFINCS model (src_points_output=True) to get Wflow output at the right locations. Make sure the hydromt_sfincs configuration defines these source locations.

[6]:

# Build a 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=True,
    src_points_output=True,
)
wf.create_rule(sfincs_build, rule_id="sfincs_build")

[6]:

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

Next we build a Wflow model using:

  • the sfincs_build output for the model region

  • “gauges” based on SFINCS source points

  • settings from the hydromt_wflow_config, see HydroMT-Wflow docs

  • data from the data catalog

[7]:

# Build a Wflow model
wflow_build = wflow.WflowBuild(
    region=sfincs_build.output.sfincs_region,
    config=wf.get_ref("$config.hydromt_wflow_config"),
    wflow_root="models/wflow",
    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)

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

[8]:

# Build a 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")

[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. First, we update and run Wflow to simulate discharge for the present climate. Then, we derive design hydrographs for a number of return periods from discharge time series. By default, the magnitude and shape of these events are based on the annual maxima peaks from the timeseries. Note that we have set copy_model to True. This is required to be able to run with CWL, as the moving around of inputs done by CWL does not play nice with the relative paths in model config files otherwise used. This also holds for the update_sfincs and update_fiat steps we will encounter later in this workflow.

Note that in case of multiple discharge boundary locations this approach assumes full dependence which might be an oversimplification of the reality.

[9]:

# Update Wflow meteorological forcing for the simulation period
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",
    copy_model=True, # Necessary for CWL
)
wf.create_rule(wflow_update, rule_id="wflow_update")

[9]:

Rule(id=wflow_update, method=wflow_update_forcing, runs=1)

[10]:

# Run the wflow model for a continuous simulation setup in the wflow_update rule
wflow_run = wflow.WflowRun(
    wflow_toml=wflow_update.output.wflow_out_toml,
    run_method=wf.get_ref("$config.wflow_run_method"),
)
wf.create_rule(wflow_run, rule_id="wflow_run")

[10]:

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

[11]:

# Derive fluvial design events
# Checkout the FluvialDesignEvents parameters for many options
fluvial_events = discharge.FluvialDesignEvents(
    discharge_nc=wflow_run.output.wflow_output_timeseries,
    rps=wf.get_ref("$config.rps"),
    event_root="input/events",
    index_dim="Q_gauges_bounds",
    wildcard="events",
)

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

INFO - wildcards - Added wildcard 'events' with values: ['q_event_rp002', 'q_event_rp005', 'q_event_rp010', 'q_event_rp050', 'q_event_rp100']

[11]:

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

Derive flood hazard#

To derive flood hazard maps for each event, we

  1. Update the SFINCS model using the discharge 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.

[12]:

# Update the SFINCS model with fluvial events
sfincs_update = sfincs.SfincsUpdateForcing(
    sfincs_inp=sfincs_build.output.sfincs_inp,
    event_yaml=fluvial_events.output.event_yaml,
    output_dir=sfincs_build.output.sfincs_inp.parent/"simulations"/"{events}",
    copy_model=True, # Necessary for CWL
)
wf.create_rule(sfincs_update, rule_id="sfincs_update")

[12]:

Rule(id=sfincs_update, method=sfincs_update_forcing, runs=5, repeat=['events'])

[13]:

# Run the SFINCS model for each fluvial event
sfincs_run = sfincs.SfincsRun(
    sfincs_inp=sfincs_update.output.sfincs_out_inp,
    run_method=wf.get_ref("$config.sfincs_run_method")
)
wf.create_rule(sfincs_run, rule_id="sfincs_run")

[13]:

Rule(id=sfincs_run, method=sfincs_run, runs=5, repeat=['events'])

[14]:

# Postprocesses SFINCS results to a regular grid of maximum water levels
sfincs_post = sfincs.SfincsPostprocess(
    sfincs_map=sfincs_run.output.sfincs_map,
)
wf.create_rule(sfincs_post, rule_id="sfincs_post")

[14]:

Rule(id=sfincs_post, method=sfincs_postprocess, runs=5, repeat=['events'])

[15]:

# Optionally, downscale the SFINCS output to derive high-res flood hazard maps
sfincs_downscale = sfincs.SfincsDownscale(
    sfincs_map=sfincs_run.output.sfincs_map,
    sfincs_subgrid_dep=sfincs_build.output.sfincs_subgrid_dep,
    depth_min=wf.get_ref("$config.depth_min"),
    output_root="output/hazard",
)
wf.create_rule(sfincs_downscale, rule_id="sfincs_downscale")

[15]:

Rule(id=sfincs_downscale, method=sfincs_downscale, runs=5, repeat=['events'])

Derive flood risk#

To calculate flood risk, we

  • Update Delft-FIAT with all fluvial 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.

[16]:

# Update FIAT hazard forcing with the fluvial eventset to compute fluvial flood risk
fiat_update = fiat.FIATUpdateHazard(
    fiat_cfg=fiat_build.output.fiat_cfg,
    event_set_yaml=fluvial_events.output.event_set_yaml,
    map_type="water_level",
    hazard_maps=sfincs_post.output.sfincs_zsmax,
    risk=True,
    output_dir=fiat_build.output.fiat_cfg.parent/"simulations",
    copy_model=True, # Necessary for CWL
)
wf.create_rule(fiat_update, rule_id="fiat_update")

[16]:

Rule(id=fiat_update, method=fiat_update_hazard, runs=1, reduce=['events'])

[17]:

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

[17]:

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

[18]:

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

[18]:

Rule(id=fiat_visualize, method=fiat_visualize, 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.

[19]:

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

[19]:
../_images/_examples_fluvial_risk_31_0.svg 
[20]:

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

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

[21]:

# Write the workflow to a Snakefile and snakefile.config.yml
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/fluvial_risk:
- Snakefile
- rulegraph.svg
- Snakefile.config.yml

[22]:

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

[23]:

# Write the workflow to a cwl file and cwl config file
wf.to_cwl()

[24]:

# uncomment to run the workflow with cwll
# cwltool does not by default preserve environment variables. This causes issues when running Delft-FIAT.
# Hence the extra flag to explicitly tell cwltool to preserve the PROJ_DATA environment variable
# import subprocess
# subprocess.run(["cwltool", "--preserve-environment", "PROJ_DATA", f"{wf.name}.cwl", f"{wf.name}.config.yml"], cwd=wf.root)

[ ]: