CLI Automation

Automation can be done in any language that can call the CLI tool,
but a detailed example is provide below for Python:
FastFlood Python Example

Full Python code:

"""
fastflood_runner.py

A simple Python wrapper around the local FastFlood CLI.

This script supports:
1. Hazard maps -> uses: -autohazard
2. Forecast maps -> uses: -autoforecast
3. Specific forecast date or "latest"
4. Return period hazard maps
5. Optional climate scenario for hazard runs
6. Optional extra FastFlood flags and outputs

IMPORTANT:
- Replace FASTFLOOD_EXE with your local path to fastflood.exe
- Replace WORKDIR with a suitable working directory
- Replace "<KEY>" with your actual key when you want to run
"""

from __future__ import annotations

import os
import shlex
import subprocess
import datetime as dt
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional, List

# =============================================================================
# USER-EDITABLE DEFAULTS
# =============================================================================

FASTFLOOD_EXE = r"H:\fastflood\fastflood.exe"
WORKDIR = r"H:\fastflood\work"
API_KEY = "<KEY>" # Placeholder on purpose

# =============================================================================
# VALIDATION CONSTANTS
# =============================================================================

VALID_HAZARD_RESOLUTIONS = {"low", "medium", "high"}
VALID_FORECAST_RESOLUTIONS = {
"low", "medium", "high", "veryhigh",
"2.5m", "5m", "10m", "20m", "40m", "80m", "160m", "300m", "600m", "1.2km"
}
VALID_RETURN_PERIODS = {2, 5, 10, 20, 40, 50, 100, 200, 500, 1000}
VALID_CLIMATE_SCENARIOS = {"ssp124", "ssp245", "ssp460", "ssp585"}
VALID_CLIMATE_DURATIONS_DAYS = {1, 3, 7}

# =============================================================================
# DATA CLASSES
# =============================================================================

@dataclass
class BoundingBox:
"""
Geographic model extent.

FastFlood expects:
ulx, uly, brx, bry

Where:
- ulx = upper-left x (longitude)
- uly = upper-left y (latitude)
- brx = bottom-right x (longitude)
- bry = bottom-right y (latitude)
"""
ulx: float
uly: float
brx: float
bry: float

def as_cli_args(self) -> List[str]:
return [str(self.ulx), str(self.uly), str(self.brx), str(self.bry)]

@dataclass
class ClimateSettings:
"""
Optional climate change settings for hazard simulations.

FastFlood docs say:
--climate <scenario> <period> <quantile> <return_period> <duration_days>

Example:
--climate ssp585 2050 50 100 1
"""
scenario: str # ssp124 / ssp245 / ssp460 / ssp585
period: int # 2020..2100
quantile: int # 15 / 50 / 85
return_period: int # 2..1000
duration_days: int # 1 / 3 / 7

def validate(self) -> None:
if self.scenario not in VALID_CLIMATE_SCENARIOS:
raise ValueError(
f"Invalid climate scenario: {self.scenario}. "
f"Valid options: {sorted(VALID_CLIMATE_SCENARIOS)}"
)

if not (2020 <= self.period <= 2100):
raise ValueError("Climate period must be between 2020 and 2100.")

if self.quantile not in {15, 50, 85}:
raise ValueError("Climate quantile must be one of: 15, 50, 85.")

if not (2 <= self.return_period <= 1000):
raise ValueError("Climate return period must be between 2 and 1000.")

if self.duration_days not in VALID_CLIMATE_DURATIONS_DAYS:
raise ValueError(
f"Climate duration_days must be one of: {sorted(VALID_CLIMATE_DURATIONS_DAYS)}"
)

def as_cli_args(self) -> List[str]:
self.validate()
return [
"--climate",
self.scenario,
str(self.period),
str(self.quantile),
str(self.return_period),
str(self.duration_days),
]

@dataclass
class RunOutputs:
"""
Output files to request from FastFlood.

Only include the ones you want.
"""
whout: Optional[str] = "flooddepth.tif" # maximum water depth
qout: Optional[str] = "discharge.tif" # peak discharge
rainout: Optional[str] = None
ruout: Optional[str] = None
vout: Optional[str] = None
atout: Optional[str] = None
hydrograph: Optional[str] = None

def as_cli_args(self) -> List[str]:
args: List[str] = []

if self.whout:
args += ["-whout", self.whout]
if self.qout:
args += ["-qout", self.qout]
if self.rainout:
args += ["-rainout", self.rainout]
if self.ruout:
args += ["-ruout", self.ruout]
if self.vout:
args += ["-vout", self.vout]
if self.atout:
args += ["-atout", self.atout]
if self.hydrograph:
args += ["-hydrograph", self.hydrograph]

return args

@dataclass
class CommonOptions:
"""
Common optional switches to apply to either hazard or forecast runs.

Notes:
- d_lu / d_inf are often useful automatic data downloads
- mult_man / mult_inf let you tune roughness / infiltration
- cores lets you control CPU usage
- verbose prints more detail
"""
d_lu: bool = True
d_inf: bool = True
mult_man: Optional[float] = None
mult_inf: Optional[float] = None
mult_rain: Optional[float] = None
cores: Optional[int] = None
quality: Optional[float] = None
noautorefine: bool = False
verbose: bool = False
protectionlevels: Optional[int] = None
noautobcond: bool = False
noautoblock: bool = False

def as_cli_args(self) -> List[str]:
args: List[str] = []

if self.d_lu:
args.append("-d_lu")
if self.d_inf:
args.append("-d_inf")

if self.mult_man is not None:
args += ["-mult_man", str(self.mult_man)]
if self.mult_inf is not None:
args += ["-mult_inf", str(self.mult_inf)]
if self.mult_rain is not None:
args += ["-mult_rain", str(self.mult_rain)]

if self.cores is not None:
args += ["-cores", str(self.cores)]

if self.quality is not None:
args += ["-quality", str(self.quality)]

if self.noautorefine:
args.append("-noautorefine")

if self.verbose:
args.append("-verbose")

if self.protectionlevels is not None:
args += ["-protectionlevels", str(self.protectionlevels)]

if self.noautobcond:
args.append("-noautobcond")

if self.noautoblock:
args.append("-noautoblock")

return args

# =============================================================================
# FASTFLOOD RUNNER
# =============================================================================

class FastFloodRunner:
"""
Thin wrapper around the FastFlood CLI.

Main public methods:
- run_hazard(...)
- run_forecast(...)
"""

def __init__(self, exe_path: str, workdir: str, api_key: str) -> None:
self.exe_path = Path(exe_path)
self.workdir = Path(workdir)
self.api_key = api_key

if not self.exe_path.exists():
raise FileNotFoundError(f"FastFlood executable not found: {self.exe_path}")

self.workdir.mkdir(parents=True, exist_ok=True)

# -------------------------------------------------------------------------
# Internal helpers
# -------------------------------------------------------------------------

def _validate_resolution(self, resolution: str, mode: str) -> None:
if mode == "hazard":
if resolution not in VALID_HAZARD_RESOLUTIONS:
raise ValueError(
f"Invalid hazard resolution: {resolution}. "
f"Valid: {sorted(VALID_HAZARD_RESOLUTIONS)}"
)
elif mode == "forecast":
if resolution not in VALID_FORECAST_RESOLUTIONS:
raise ValueError(
f"Invalid forecast resolution: {resolution}. "
f"Valid: {sorted(VALID_FORECAST_RESOLUTIONS)}"
)
else:
raise ValueError(f"Unknown mode: {mode}")

def _validate_forecast_date(self, forecast_date: str) -> None:
"""
Accept either:
- 'latest'
- YYYY_MM_DD
- YYYY-MM-DD

The docs say string date or latest.
Your examples show 'latest' and dates like 2024_07_01 in other scripts.
To keep it flexible, we allow both separators here.
"""
if forecast_date == "latest":
return

for fmt in ("%Y_%m_%d", "%Y-%m-%d"):
try:
dt.datetime.strptime(forecast_date, fmt)
return
except ValueError:
pass

raise ValueError(
f"Invalid forecast date: {forecast_date}. "
f"Use 'latest', 'YYYY_MM_DD', or 'YYYY-MM-DD'."
)

def _base_command(self) -> List[str]:
"""
Returns the required starting part of every command.
"""
return [
str(self.exe_path),
"-key", self.api_key,
"-sim",
]

def _run_command(
self,
cmd: List[str],
log_name: Optional[str] = None,
print_command: bool = True,
check: bool = True,
) -> subprocess.CompletedProcess:
"""
Run the command in the configured working directory.

Also writes stdout/stderr to a log file if log_name is provided.
"""
if print_command:
# shlex.join makes the command much easier to read in logs / console
print("\nRunning command:")
print(shlex.join(cmd))

proc = subprocess.run(
cmd,
cwd=self.workdir,
capture_output=True,
text=True
)

if log_name:
log_path = self.workdir / log_name
with open(log_path, "w", encoding="utf-8") as f:
f.write("COMMAND:\n")
f.write(shlex.join(cmd))
f.write("\n\nSTDOUT:\n")
f.write(proc.stdout or "")
f.write("\n\nSTDERR:\n")
f.write(proc.stderr or "")

print(f"Log written to: {log_path}")

if check and proc.returncode != 0:
raise RuntimeError(
"FastFlood command failed.\n"
f"Return code: {proc.returncode}\n"
f"STDOUT:\n{proc.stdout}\n"
f"STDERR:\n{proc.stderr}"
)

return proc

# -------------------------------------------------------------------------
# Public method: hazard run
# -------------------------------------------------------------------------

def run_hazard(
self,
bbox: BoundingBox,
scenario: str,
return_period: int,
resolution: str = "high",
outputs: Optional[RunOutputs] = None,
common: Optional[CommonOptions] = None,
climate: Optional[ClimateSettings] = None,
extra_args: Optional[List[str]] = None,
log_name: Optional[str] = None,
) -> subprocess.CompletedProcess:
"""
Run an automatic hazard simulation.

FastFlood format:
fastflood.exe -key <KEY> -sim -autohazard ulx uly brx bry scenario return_period resolution ...

Example:
fastflood.exe -key <KEY> -sim -autohazard -74.18 40.75 -74.14 40.72 flashfloodfluvialflood 50 high -d_lu -d_inf -whout flooddepth.tif -qout discharge.tif

Parameters
----------
bbox : BoundingBox
Model domain.
scenario : str
FastFlood scenario string, e.g.:
- "flashflood"
- "fluvialflood"
- "flashfloodfluvialflood"
- "coastal"
- combinations as supported by FastFlood
return_period : int
Return period in years.
resolution : str
low / medium / high for autohazard.
outputs : RunOutputs
Output file requests.
common : CommonOptions
Common modifiers and switches.
climate : ClimateSettings | None
Optional climate change modifier for hazard runs.
extra_args : list[str] | None
Any additional raw FastFlood CLI arguments.
log_name : str | None
Optional log file name.
"""
if outputs is None:
outputs = RunOutputs()

if common is None:
common = CommonOptions()

if return_period < 2 or return_period > 1000:
raise ValueError("return_period must be between 2 and 1000.")

self._validate_resolution(resolution, mode="hazard")

cmd = self._base_command()
cmd += [
"-autohazard",
*bbox.as_cli_args(),
scenario,
str(return_period),
resolution,
]

# Optional climate settings
if climate is not None:
cmd += climate.as_cli_args()

# Common options
cmd += common.as_cli_args()

# Output options
cmd += outputs.as_cli_args()

# User-provided raw extra arguments
if extra_args:
cmd += extra_args

return self._run_command(
cmd=cmd,
log_name=log_name,
print_command=True,
check=True,
)

# -------------------------------------------------------------------------
# Public method: forecast run
# -------------------------------------------------------------------------

def run_forecast(
self,
bbox: BoundingBox,
scenario: str,
forecast_date: str = "latest",
start_offset_hours: int = 0,
duration_hours: int = 24,
resolution: str = "high",
outputs: Optional[RunOutputs] = None,
common: Optional[CommonOptions] = None,
extra_args: Optional[List[str]] = None,
log_name: Optional[str] = None,
) -> subprocess.CompletedProcess:
"""
Run an automatic forecast simulation.

FastFlood format:
fastflood.exe -key <KEY> -sim -autoforecast ulx uly brx bry scenario date_or_latest start_offset duration resolution ...

Example:
fastflood.exe -key <KEY> -sim -autoforecast 4.84 52.36 4.89 52.32 flashfloodfluvialflood latest 0 150 high -whout flooddepth.tif -qout discharge.tif

Parameters
----------
bbox : BoundingBox
Model domain.
scenario : str
FastFlood scenario string.
forecast_date : str
"latest" or a specific day string like "2025_07_03" or "2025-07-03"
start_offset_hours : int
Hours relative to forecast start.
duration_hours : int
Simulation rainfall/event duration in hours.
resolution : str
low/medium/high/veryhigh or explicit resolutions supported by autoforecast.
outputs : RunOutputs
Output file requests.
common : CommonOptions
Common modifiers and switches.
extra_args : list[str] | None
Extra raw CLI args.
log_name : str | None
Optional log file name.
"""
if outputs is None:
outputs = RunOutputs()

if common is None:
common = CommonOptions()

self._validate_resolution(resolution, mode="forecast")
self._validate_forecast_date(forecast_date)

if duration_hours <= 0:
raise ValueError("duration_hours must be > 0.")

cmd = self._base_command()
cmd += [
"-autoforecast",
*bbox.as_cli_args(),
scenario,
forecast_date,
str(start_offset_hours),
str(duration_hours),
resolution,
]

cmd += common.as_cli_args()
cmd += outputs.as_cli_args()

if extra_args:
cmd += extra_args

return self._run_command(
cmd=cmd,
log_name=log_name,
print_command=True,
check=True,
)

# =============================================================================
# EXAMPLE USAGE FUNCTIONS
# =============================================================================

def example_hazard_simple(runner: FastFloodRunner) -> None:
"""
Example 1:
Simple hazard map for a 50-year event.

Equivalent in spirit to:
fastflood.exe -key <KEY> -sim -autohazard <bbox> flashfloodfluvialflood 50 high -d_lu -d_inf -whout flooddepth.tif -qout discharge.tif
"""
bbox = BoundingBox(
ulx=-74.18627167553016,
uly=40.75974044414834,
brx=-74.14477284088257,
bry=40.72829955302365,
)

runner.run_hazard(
bbox=bbox,
scenario="flashfloodfluvialflood",
return_period=50,
resolution="high",
outputs=RunOutputs(
whout="hazard_flooddepth_50y.tif",
qout="hazard_discharge_50y.tif",
),
common=CommonOptions(
d_lu=True,
d_inf=True,
),
log_name="hazard_simple.log",
)

def example_hazard_with_climate(runner: FastFloodRunner) -> None:
"""
Example 2:
Hazard map for a 100-year event with climate change settings.

This adds:
--climate ssp585 2050 50 100 1

Meaning roughly:
- SSP585 scenario
- period 2050
- quantile 50
- return period 100
- duration 1 day
"""
bbox = BoundingBox(
ulx=4.841960147268389,
uly=52.36022933081758,
brx=4.893425628714431,
bry=52.32878843969289,
)

climate = ClimateSettings(
scenario="ssp585",
period=2050,
quantile=50,
return_period=100,
duration_days=1,
)

runner.run_hazard(
bbox=bbox,
scenario="flashfloodfluvialflood",
return_period=100,
resolution="high",
climate=climate,
outputs=RunOutputs(
whout="hazard_cc_flooddepth_100y_ssp585_2050.tif",
qout="hazard_cc_discharge_100y_ssp585_2050.tif",
rainout="hazard_cc_rain_100y_ssp585_2050.tif",
),
common=CommonOptions(
d_lu=True,
d_inf=True,
mult_man=1.0,
mult_inf=1.0,
cores=4,
),
log_name="hazard_climate.log",
)

def example_forecast_latest(runner: FastFloodRunner) -> None:
"""
Example 3:
Latest forecast map.

Similar to:
fastflood.exe -key <KEY> -sim -autoforecast 4.84 52.36 4.89 52.32 flashfloodfluvialflood latest 0 150 high -whout flooddepth.tif -qout discharge.tif
"""
bbox = BoundingBox(
ulx=4.841960147268389,
uly=52.36022933081758,
brx=4.893425628714431,
bry=52.32878843969289,
)

runner.run_forecast(
bbox=bbox,
scenario="flashfloodfluvialflood",
forecast_date="latest",
start_offset_hours=0,
duration_hours=150,
resolution="high",
outputs=RunOutputs(
whout="forecast_latest_flooddepth.tif",
qout="forecast_latest_discharge.tif",
),
common=CommonOptions(
d_lu=True,
d_inf=True,
cores=4,
),
log_name="forecast_latest.log",
)

def example_forecast_specific_day(runner: FastFloodRunner) -> None:
"""
Example 4:
Forecast for a specific day instead of latest.

You can use either:
- YYYY_MM_DD
- YYYY-MM-DD

This is useful if you want reproducible reruns.
"""
bbox = BoundingBox(
ulx=4.841960147268389,
uly=52.36022933081758,
brx=4.893425628714431,
bry=52.32878843969289,
)

runner.run_forecast(
bbox=bbox,
scenario="flashfloodfluvialflood",
forecast_date="2025_07_03",
start_offset_hours=0,
duration_hours=72,
resolution="high",
outputs=RunOutputs(
whout="forecast_2025_07_03_flooddepth.tif",
qout="forecast_2025_07_03_discharge.tif",
rainout="forecast_2025_07_03_rain.tif",
),
common=CommonOptions(
d_lu=True,
d_inf=True,
mult_man=1.2,
mult_inf=0.9,
),
log_name="forecast_specific_day.log",
)

# =============================================================================
# MAIN
# =============================================================================

def main() -> None:
"""
Simple entry point showing how to use the wrapper.

Pick one or more example calls below.

Before running:
1. Set FASTFLOOD_EXE to your local fastflood.exe path
2. Set WORKDIR
3. Replace API_KEY = "<KEY>" with your real key when actually running
"""
runner = FastFloodRunner(
exe_path=FASTFLOOD_EXE,
workdir=WORKDIR,
api_key=API_KEY,
)

# ---------------------------------------------------------------------
# Example 1: basic hazard map
# ---------------------------------------------------------------------
# example_hazard_simple(runner)

# ---------------------------------------------------------------------
# Example 2: hazard map with climate scenario
# ---------------------------------------------------------------------
# example_hazard_with_climate(runner)

# ---------------------------------------------------------------------
# Example 3: latest forecast map
# ---------------------------------------------------------------------
# example_forecast_latest(runner)

# ---------------------------------------------------------------------
# Example 4: forecast map for a specific day
# ---------------------------------------------------------------------
# example_forecast_specific_day(runner)

# ---------------------------------------------------------------------
# Example 5: direct inline usage without helper function
# ---------------------------------------------------------------------
bbox = BoundingBox(
ulx=4.841960147268389,
uly=52.36022933081758,
brx=4.893425628714431,
bry=52.32878843969289,
)

print("\nRunning one direct inline example...\n")

runner.run_forecast(
bbox=bbox,
scenario="flashfloodfluvialflood",
forecast_date="latest",
start_offset_hours=0,
duration_hours=150,
resolution="high",
outputs=RunOutputs(
whout="inline_flooddepth.tif",
qout="inline_discharge.tif",
),
common=CommonOptions(
d_lu=True,
d_inf=True,
cores=4,
verbose=True,
),
log_name="inline_example.log",
)

if __name__ == "__main__":
main()