Coastal flood risk

This example shows a workflow to derive coastal flood risk using the SFINCS and Delft-FIAT models. The starting point is a user defined region and data catalog. Coastal storm tide hydrographs for different return periods are derived from the GTSM reanalysis dataset and used to simulate the flood hazard maps. Note that wave setup / runup are not considered. The hazard maps are combined with exposure and impact data to derive risk.

# Import modules
from pathlib import Path

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

logger = setuplog(level="INFO")

INFO - log - hydroflows version: 0.1.0

# Define case name and root directory
name = "coastal_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 including the GTSM reanalysis dataset. Here we fetch some test datasets for a region in Northern Italy.

• HydroMT configuration files for both models.

• model executables (docker is also possible for SFINCS)

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

# 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,
    # 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,
    # coastal time series and design events
    start_time="2014-01-01",
    end_time="2021-12-31",
    rps=[2, 5, 10, 50, 100],
    # future climate SLR scenarios
    slr_scenarios={"present": 0, "rcp85_2050": 0.2},
)

Create the workflow

wf = Workflow(config=config, name=name, root=case_root)

Build models

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"),
    # save subgrid output to use in subsequent rules:
    subgrid_output=True,
)
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 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 coastal design events

Here, we define coastal design events for different return periods from the GTSM surge and tide timeseries data. These time series can also be replaced with observed timeseries if available. The total water level (combined tide and surge) of each event is derived by fitting an extreme value distribution to total water level peaks, using a peak-over-threshold approach. The shape of each event is derived by combining the spring tide signal with an average surge signal which is scaled to match the event total water level.

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

# Get the GTSM data from reanalysis data for region and time period
get_gtsm_data = coastal.GetGTSMData(
    gtsm_catalog=wf.get_ref("$config.catalog_path"),
    start_time=wf.get_ref("$config.start_time"),
    end_time=wf.get_ref("$config.end_time"),
    region=sfincs_build.output.sfincs_region,
    data_root="data/gtsm",
)
wf.create_rule(get_gtsm_data, rule_id="get_gtsm_data")

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

# Generate coastal design events
coastal_events = coastal.CoastalDesignEvents(
    surge_timeseries=get_gtsm_data.output.surge_nc,
    tide_timeseries=get_gtsm_data.output.tide_nc,
    bnd_locations=get_gtsm_data.output.bnd_locations,
    rps=wf.get_ref("$config.rps"),
    event_root="data/events/default",
    wildcard="event", # wildcard to use for the pluvial events

)

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

INFO - wildcards - Added wildcard 'event' with values: ['h_event_rp002', 'h_event_rp005', 'h_event_rp010', 'h_event_rp050', 'h_event_rp100']

Rule(id=coastal_events, method=coastal_design_events, runs=1, expand=['event'])

The events are scaled for future climate predictions based on the a sea level rise offset. A new wildcard for scenarios is introduced develop hazard maps for each event per scenario.

future_coastal_events = coastal.FutureSLR(
    scenarios=wf.get_ref("$config.slr_scenarios"),
    event_set_yaml=coastal_events.output.event_set_yaml,
    event_names=coastal_events.params.event_names,
    event_root="data/events",
    event_wildcard="event",
    scenario_wildcard="scenario",
)
wf.create_rule(future_coastal_events, rule_id="future_coastal_events")

INFO - wildcards - Added wildcard 'scenario' with values: ['present', 'rcp85_2050']

Rule(id=future_coastal_events, method=future_slr, runs=1, expand=['event', 'scenario'])

Derive flood hazard

To derive flood hazard maps for each event, we

1. Update the SFINCS model using the water level 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_coastal_events.output.future_event_csv,
    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=['event', 'scenario'])

# 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=['event', 'scenario'])

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

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

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

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

Derive flood risk

To calculate flood risk, we

• Update Delft-FIAT with all coastal 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_coastal_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"),
)
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,
    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'])

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
# to run the workflow in HydroFlows use wf.run(), or (preferred) use wf.to_snakemake() to create a snakemake file
wf.dryrun()

INFO - workflow - Dryrun rule 1/12: sfincs_build (1 runs)
INFO - workflow - Dryrun rule 2/12: get_gtsm_data (1 runs)
INFO - workflow - Dryrun rule 3/12: coastal_events (1 runs)
INFO - workflow - Dryrun rule 4/12: future_coastal_events (1 runs)
INFO - workflow - Dryrun rule 5/12: fiat_build (1 runs)
INFO - workflow - Dryrun rule 6/12: sfincs_update (10 runs)
INFO - workflow - Dryrun rule 7/12: sfincs_run (10 runs)
INFO - workflow - Dryrun rule 8/12: sfincs_downscale (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():
    print(f"- {f.name}")

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

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