Large scale flood hazard (multiple regions)

This example shows a workflow to derive pluvial flood hazard using the SFINCS model. The goal of the example is to show how to scale the flood hazard to multiple model domains. For more details about each step we refer the user to the pluvial flood risk example.

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

# Import modules
from pathlib import Path

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

logger = setuplog(level="INFO")

INFO - log - hydroflows version: 0.1.0

# Define case name and root directory
name = "pluvial_multiple_regions"
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

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

Since the CWL runner we will be using is not supported on Windows, we will opt to run SFINCS using a docker container.

# Setup the configuration
config = WorkflowConfig(
    config=Path(pwd_rel, "hydromt_config/sfincs_config.yml"),
    catalog_path=Path(cache_dir, "data_catalog.yml"),
    sfincs_run_method="docker",
    start_date="2014-01-01",
    end_date="2021-12-31",
    # sfincs settings
    hydromt_sfincs_config=Path(pwd_rel, "hydromt_config/sfincs_config.yml"),
    subgrid_output=True,
    # design event settings
    rps=[2, 5, 10],
)

Create the workflow

Note that we initialize the workflow with a region wildcard to create flood hazard for multiple regions.

# Setup the workflow
wf = Workflow(
    config=config,
    wildcards={"region": ["region", "region2"]},
    name=name,
    root=case_root,
)

Build models

In this section we build SFINCS models for multiple regions. Note that we use the {region} wildcard on the in- and outputs of each method. The method will be executed for each input region.

# Build the SFINCS models
sfincs_build = sfincs.SfincsBuild(
    region=Path(pwd_rel, "data/build/{region}.geojson"),  # input region
    sfincs_root="models/sfincs/{region}",  # output model directory
    config=wf.get_ref("$config.hydromt_sfincs_config"),
    catalog_path=wf.get_ref("$config.catalog_path"),
    subgrid_output=wf.get_ref("$config.subgrid_output"),
    )
wf.create_rule(sfincs_build, "sfincs_build")

Rule(id=sfincs_build, method=sfincs_build, runs=2, repeat=['region'])

Derive pluvial design events

In contrast to the pluvial risk workflow, we will name the the event_root param “events/{region}_events”. This way we avoid the following conflict when parsing to CWL. The outputs of both the pluvial_events rule and the sfincs_build rule will be used as inputs to the sfincs_update rule. If we were to follow the naming as the pluvial risk example, the pluvial_events and sfincs_build rules will create output directory with “{region}” as stem. Since CWL will flatten the rest of those output directory paths, the sfincs_update rule will receive two input directories both named “{region}”.

pluvial_events = rainfall.PluvialDesignEventsGPEX(
    gpex_nc=Path(cache_dir, "gpex.nc"),
    region=sfincs_build.output.sfincs_region,
    event_root="events/{region}_events",
    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']

Rule(id=pluvial_events, method=pluvial_design_events_GPEX, runs=2, repeat=['region'], expand=['events'])

Derive flood hazard

Here we have to set the copy_model param of the sfincs_update rule to True. The flattening of input file paths by CWL we encountered earlier does not work well with the relative paths the model config files will be using when copy_model is set to False.

Note also that in the sfincs_run rule we use the run_method param instead of the sfincs_exe param to indicate that we will be using the docker container.

# Update the SFINCS models
sfincs_update = sfincs.SfincsUpdateForcing(
    sfincs_inp=sfincs_build.output.sfincs_inp,
    event_yaml=pluvial_events.output.event_yaml,
    output_dir=sfincs_build.output.sfincs_inp.parent/"simulations"/"{events}",
    copy_model=True,
)
wf.create_rule(sfincs_update, rule_id="sfincs_update")

# Run SFINCS model
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")

# 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/{region}",
)
wf.create_rule(sfincs_downscale, "sfincs_downscale")

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

Visualize and execute the workflow

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

# Do a dry run of the workflow
wf.dryrun()

INFO - workflow - Dryrun rule 1/5: sfincs_build (2 runs)
INFO - workflow - Dryrun rule 2/5: pluvial_events (2 runs)
INFO - workflow - Dryrun rule 3/5: sfincs_update (6 runs)
INFO - workflow - Dryrun rule 4/5: sfincs_run (6 runs)
INFO - workflow - Dryrun rule 5/5: sfincs_downscale (6 runs)

# 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/pluvial_multiple_regions:
- Snakefile
- rulegraph.svg
- Snakefile.config.yml

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

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

# uncomment to run the workflow with cwll
# import subprocess
# subprocess.run(["cwltool", "pluvial_multiple_regions.cwl", "pluvial_multiple_regions.config.yml"], cwd=wf.root)