"""Package include method for simulate subcatchment with different
features values from Storm Water Management Model"""
from __future__ import annotations
import os
from typing import Any
import numpy as np
import pandas as pd
import swmmio
from pyswmm import Simulation, Subcatchments
from .schemas import SimulationParams
[docs]
class FeaturesSimulation:
"""
A class to simulate subcatchments with different features using the Storm Water Management Model (SWMM).
Parameters
----------
subcatchment_id : str
The identifier of the subcatchment being simulated.
raw_file : str
The path to the raw SWMM input file.
Raises
------
FileNotFoundError
If the raw_file does not exist.
ValueError
If the subcatchment_id does not exist in the SWMM model.
"""
SWMM_SIDE_EXTENSIONS: tuple[str, ...] = (".inp", ".rpt", ".out")
RESULT_KEYS: tuple[str, ...] = ("runoff", "peak_runoff_rate", "infiltration", "evaporation")
TIMESERIES_KEYS: tuple[str, ...] = (
"rainfall",
"runoff",
"infiltration_loss",
"evaporation_loss",
"runon",
)
# Source: McCuen, R. et al. (1996), Hydrology, FHWA-SA-96-067, Federal Highway Administration, Washington, DC.
MANNING_N_VALUES: tuple[float, ...] = tuple(
sorted(
[
0.011,
0.012,
0.013,
0.014,
0.015,
0.024,
0.05,
0.06,
0.17,
0.13,
0.15,
0.24,
0.41,
0.4,
0.8,
]
)
)
# Source: ASCE (1992), Design & Construction of Urban Stormwater Management Systems, New York, NY.
# Values in mm (converted from inches: 0.05, 0.1, 0.2, 0.3)
DEPRESSION_STORAGE_VALUES: tuple[float, ...] = tuple(
val * 25.4 for val in [0.05, 0.1, 0.2, 0.3]
)
@staticmethod
def _create_result_dict() -> dict[str, list[Any]]:
"""Create an empty result dictionary with standard keys."""
return {key: [] for key in FeaturesSimulation.RESULT_KEYS}
def __init__(self, subcatchment_id: str, raw_file: str) -> None:
self.raw_file = raw_file
self.subcatchment_id = subcatchment_id
self._temp_files: list[str] = []
try:
self.file = self.copy_file(copy=self.raw_file)
self.model = swmmio.Model(self.file)
available_ids = list(self.model.inp.subcatchments.index)
if subcatchment_id not in available_ids:
raise ValueError(
f"Subcatchment '{subcatchment_id}' not found in model. "
f"Available subcatchments: {available_ids}"
)
except Exception:
self._cleanup_temp_files()
raise
def __enter__(self) -> FeaturesSimulation:
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: Any,
) -> None:
self._cleanup_temp_files()
def _cleanup_temp_files(self) -> None:
"""Remove all temporary files created during simulation."""
for temp_file in self._temp_files:
base = os.path.splitext(temp_file)[0]
for ext in self.SWMM_SIDE_EXTENSIONS:
try:
os.remove(base + ext)
except OSError:
pass
self._temp_files.clear()
@staticmethod
def _validate_simulation_params(start: float, stop: float, step: float) -> SimulationParams:
"""Validate simulation parameters using Pydantic schema.
Returns the validated ``SimulationParams`` instance so callers can
use the coerced values if needed.
Raises
------
pydantic.ValidationError
If any parameter violates its constraints.
"""
return SimulationParams(start=start, stop=stop, step=step)
[docs]
def copy_file(self, copy: str | None = None, suffix: str = "copy") -> str:
"""
Create a copy of a SWMM input file with a suffix added to the end of the file name.
Parameters
----------
copy : str, optional
The path to the file you want to copy. If not specified, it will use the raw_file.
suffix : str, optional
The suffix to add to the end of the file name, defaults to 'copy'.
Returns
-------
str
The new path of the copied file.
"""
if copy is None:
copy = self.raw_file
baseline = swmmio.Model(copy)
new_path: str = os.path.join(baseline.inp.name + "_" + suffix + ".inp")
baseline.inp.save(new_path)
self._temp_files.append(new_path)
return new_path
[docs]
def get_section(self, section: str = "subcatchments") -> pd.DataFrame:
"""
Get a specified section from a SWMM input file as a pandas DataFrame.
Parameters
----------
section : str, optional
The name of the section you want to get, defaults to 'subcatchments'.
Returns
-------
pd.DataFrame
The section of the inp file as a pandas DataFrame.
"""
return getattr(swmmio.Model(self.file).inp, section)
[docs]
def calculate(self) -> dict[str, Any]:
"""
Run a simulation using the SWMM model and return the statistics of the subcatchment with the
ID `self.subcatchment_id`.
Returns
-------
dict
The statistics of the subcatchment.
"""
with Simulation(self.file) as sim:
subcatchment = Subcatchments(sim)[self.subcatchment_id]
for _ in sim:
pass
result: dict[str, Any] = subcatchment.statistics
return result
[docs]
def calculate_timeseries(self) -> pd.DataFrame:
"""Run a simulation and collect per-timestep data for the subcatchment.
Returns
-------
pd.DataFrame
DataFrame with a ``DatetimeIndex`` (name ``"datetime"``) and columns
corresponding to ``TIMESERIES_KEYS``.
"""
records: list[dict[str, Any]] = []
with Simulation(self.file) as sim:
subcatchment = Subcatchments(sim)[self.subcatchment_id]
for step in sim: # noqa: B007
records.append(
{
"datetime": sim.current_time,
**{key: getattr(subcatchment, key) for key in self.TIMESERIES_KEYS},
}
)
if not records:
df = pd.DataFrame(columns=list(self.TIMESERIES_KEYS))
df.index.name = "datetime"
return df
df = pd.DataFrame(records)
df = df.set_index("datetime")
return df
[docs]
def simulate_subcatchment(
self, feature: str, start: float = 0, stop: float = 100, step: float = 10
) -> pd.DataFrame:
"""
Simulate a subcatchment with varying feature values using the SWMM model.
Parameters
----------
feature : str
The name of the parameter to be varied in the simulation.
start : float, optional
The starting value of the parameter, defaults to 0.
stop : float, optional
The maximum value of the parameter to be simulated, defaults to 100.
step : float, optional
The step size for the simulation, defaults to 10.
Returns
-------
pd.DataFrame
A DataFrame containing the results of the simulation.
"""
self._validate_simulation_params(start, stop, step)
self.file = self.copy_file(self.raw_file)
catchment_data = self._create_result_dict()
scope = np.arange(start, stop + step / 2, step)
for percent in scope:
subcatchment = self.model.inp.subcatchments
subcatchment[feature] = subcatchment[feature].astype(float)
subcatchment.loc[self.subcatchment_id, feature] = percent
swmmio.utils.modify_model.replace_inp_section(
self.file, "[SUBCATCHMENTS]", subcatchment
)
subcatchment_stats = self.calculate()
for key in catchment_data:
catchment_data[key].append(subcatchment_stats[key])
catchment_data[feature] = scope.tolist()
return pd.DataFrame(data=catchment_data)
[docs]
def simulate_subcatchment_timeseries(
self, feature: str, start: float = 0, stop: float = 100, step: float = 10
) -> dict[float, pd.DataFrame]:
"""Simulate a subcatchment with varying feature values, collecting timeseries.
Works like :meth:`simulate_subcatchment` but returns per-timestep data
instead of summary statistics.
Parameters
----------
feature : str
The name of the parameter to be varied in the simulation.
start : float, optional
The starting value of the parameter, defaults to 0.
stop : float, optional
The maximum value of the parameter to be simulated, defaults to 100.
step : float, optional
The step size for the simulation, defaults to 10.
Returns
-------
dict[float, pd.DataFrame]
Mapping of parameter values to DataFrames with timeseries data.
"""
self._validate_simulation_params(start, stop, step)
self.file = self.copy_file(self.raw_file)
scope = np.arange(start, stop + step / 2, step)
results: dict[float, pd.DataFrame] = {}
for percent in scope:
subcatchment = self.model.inp.subcatchments
subcatchment[feature] = subcatchment[feature].astype(float)
subcatchment.loc[self.subcatchment_id, feature] = percent
swmmio.utils.modify_model.replace_inp_section(
self.file, "[SUBCATCHMENTS]", subcatchment
)
results[float(percent)] = self.calculate_timeseries()
return results
[docs]
def simulate_area(self, start: float = 1, stop: float = 10, step: float = 1) -> pd.DataFrame:
"""
Simulate the area of the subcatchment within a specified range of values.
Parameters
----------
start : float, optional
The starting value of the area to be varied, defaults to 1.
stop : float, optional
The maximum value of the area to be simulated, defaults to 10.
step : float, optional
The step size for the simulation, defaults to 1.
Returns
-------
pd.DataFrame
A DataFrame containing the results of the simulation.
"""
return self.simulate_subcatchment(feature="Area", start=start, stop=stop, step=step)
[docs]
def simulate_percent_impervious(
self, start: float = 0, stop: float = 100, step: float = 10
) -> pd.DataFrame:
"""
Simulate the percent impervious of a subcatchment within a specified range of values.
Parameters
----------
start : float, optional
The starting value of the percent impervious to be varied, defaults to 0.
stop : float, optional
The maximum value of the percent impervious to be simulated, defaults to 100.
step : float, optional
The step size for the simulation, defaults to 10.
Returns
-------
pd.DataFrame
A DataFrame containing the results of the simulation.
"""
return self.simulate_subcatchment(feature="PercImperv", start=start, stop=stop, step=step)
[docs]
def simulate_percent_slope(
self, start: float = 0, stop: float = 100, step: float = 10
) -> pd.DataFrame:
"""
Simulate the percent slope of a subcatchment within a specified range of values.
Parameters
----------
start : float, optional
The starting value of the percent slope to be varied, defaults to 0.
stop : float, optional
The maximum value of the percent slope to be simulated, defaults to 100.
step : float, optional
The step size for the simulation, defaults to 10.
Returns
-------
pd.DataFrame
A DataFrame containing the results of the simulation.
"""
return self.simulate_subcatchment(feature="PercSlope", start=start, stop=stop, step=step)
[docs]
def simulate_width(self, start: float = 0, stop: float = 100, step: float = 10) -> pd.DataFrame:
"""
Simulate the width of a subcatchment within a specified range of values.
Parameters
----------
start : float, optional
The starting value of the width to be varied, defaults to 0.
stop : float, optional
The maximum value of the width to be simulated, defaults to 100.
step : float, optional
The step size for the simulation, defaults to 10.
Returns
-------
pd.DataFrame
A DataFrame containing the results of the simulation.
"""
return self.simulate_subcatchment(feature="Width", start=start, stop=stop, step=step)
[docs]
def simulate_curb_length(
self, start: float = 0, stop: float = 100, step: float = 10
) -> pd.DataFrame:
"""
Simulate the curb length of a subcatchment within a specified range of values.
Parameters
----------
start : float, optional
The starting value of the curb length to be varied, defaults to 0.
stop : float, optional
The maximum value of the curb length to be simulated, defaults to 100.
step : float, optional
The step size for the simulation, defaults to 10.
Returns
-------
pd.DataFrame
A DataFrame containing the results of the simulation.
"""
return self.simulate_subcatchment(feature="CurbLength", start=start, stop=stop, step=step)
[docs]
def simulate_manning_n(self, param: str) -> pd.DataFrame:
"""
Simulate a subcatchment using various Manning's n values.
Parameters
----------
param : str
The name of the feature for which Manning's n values should be varied.
Returns
-------
pd.DataFrame
A DataFrame containing the results of the simulation.
"""
self.file = self.copy_file(self.raw_file)
catchment_data = self._create_result_dict()
for n in self.MANNING_N_VALUES:
subareas = self.model.inp.subareas
col = "N-" + param
subareas[col] = subareas[col].astype(float)
subareas.loc[self.subcatchment_id, col] = n
self.model.inp.subareas = subareas
swmmio.utils.modify_model.replace_inp_section(self.file, "[SUBAREAS]", subareas)
catchment_stats = self.calculate()
for key in catchment_data:
catchment_data[key].append(catchment_stats[key])
catchment_data["N-" + param] = list(self.MANNING_N_VALUES)
return pd.DataFrame(data=catchment_data)
[docs]
def simulate_n_imperv(self) -> pd.DataFrame:
"""
Simulate Manning's n for impervious area.
Returns
-------
pd.DataFrame
A DataFrame with the simulated values of Manning's n for the impervious area.
"""
return self.simulate_manning_n(param="Imperv")
[docs]
def simulate_n_perv(self) -> pd.DataFrame:
"""
Simulate Manning's n for pervious area.
Returns
-------
pd.DataFrame
A DataFrame with the simulated values of Manning's n for the pervious area.
"""
return self.simulate_manning_n(param="Perv")
[docs]
def simulate_destore(self, param: str) -> pd.DataFrame:
"""
Simulate the model for various depths of depression storage on the given area.
Parameters
----------
param : str
The name of the feature for which the depth of depression storage should be varied.
Returns
-------
pd.DataFrame
A DataFrame with the following columns:
- runoff
- peak_runoff_rate
- infiltration
- evaporation
- Destore-param
"""
self.file = self.copy_file(self.raw_file)
catchment_data = self._create_result_dict()
for n in self.DEPRESSION_STORAGE_VALUES:
subareas = self.model.inp.subareas
subareas.loc[self.subcatchment_id, "S-" + param] = n
self.model.inp.subareas = subareas
swmmio.utils.modify_model.replace_inp_section(self.file, "[SUBAREAS]", subareas)
catchment_stats = self.calculate()
for key in catchment_data:
catchment_data[key].append(catchment_stats[key])
df = pd.DataFrame(data=catchment_data)
df["Destore-" + param] = self.DEPRESSION_STORAGE_VALUES
return df
[docs]
def simulate_s_imperv(self) -> pd.DataFrame:
"""
Simulate the impervious depth of depression storage on area.
Returns
-------
pd.DataFrame
A DataFrame with the simulated values of the impervious surface area.
"""
return self.simulate_destore(param="Imperv")
[docs]
def simulate_s_perv(self) -> pd.DataFrame:
"""
Simulate the pervious depth of depression storage on area.
Returns
-------
pd.DataFrame
A DataFrame with the simulated values of the S_perv variable.
"""
return self.simulate_destore(param="Perv")
[docs]
def simulate_percent_zero_imperv(
self, start: float = 0, stop: float = 100, step: float = 10
) -> pd.DataFrame:
"""
Run a series of simulations with different percentages of impervious area with no depression storage.
Parameters
----------
start : float, optional
The starting value for the percent impervious, defaults to 0.
stop : float, optional
The maximum percent impervious to test, defaults to 100.
step : float, optional
The step size for the percent impervious, defaults to 10.
Returns
-------
pd.DataFrame
A DataFrame with the results of the simulation.
"""
self._validate_simulation_params(start, stop, step)
self.file = self.copy_file(self.raw_file)
percent_impervious = np.arange(start, stop + step / 2, step)
catchment_data = self._create_result_dict()
for percent in percent_impervious:
subareas = self.model.inp.subareas
subareas.loc[self.subcatchment_id, "PctZero"] = percent
self.model.inp.subareas = subareas
swmmio.utils.modify_model.replace_inp_section(self.file, "[SUBAREAS]", subareas)
catchment_stats = self.calculate()
for key in catchment_data:
catchment_data[key].append(catchment_stats[key])
catchment_data["Zero-Imperv"] = percent_impervious.tolist()
return pd.DataFrame(data=catchment_data)