movici_urban_drainage_model¶

attributes¶

Attribute specifications for urban drainage (storm/sewer) simulation using pyswmm/SWMM.

Attribute naming follows the Movici documentation convention:

urban_drainage.* : urban-drainage / SWMM specific attributes
shape.length : conduit length (reused from movici core)

Enum-typed attributes store an integer index into a dataset-defined enum (the general.enum section of the dataset). The integer maps to a SWMM keyword string (e.g. "CIRCULAR") which the simulation wrapper feeds to SWMM. The expected keyword values for each enum are documented next to its declaration.

dataset¶

Entity definitions for urban drainage (storm/sewer) simulation.

Entity groups map the SWMM object model onto Movici geometry entities:

nodes (junctions, outfalls, storage units) are PointEntity
links (conduits, pumps, orifices, weirs, outlets) are LinkEntity and route between their from_node_id / to_node_id
subcatchments are PolygonEntity
rain gages are PointEntity

Static SWMM parameters (needed to build the model) are INIT; optional parameters and runtime control inputs are OPT; per-step simulation results are PUB. A few attributes are PUB | OPT: they are published every step and, when supplied, also seed the corresponding initial condition.

Note

Dividers are not modelled: SWMM treats them as ordinary junctions under dynamic-wave routing. Coordinates (geometry.x/y) are used to write the SWMM [COORDINATES] section; they do not affect the simulation results.

class ConduitEntity(name: str | None = None, optional: bool | None = None, exclude: Iterable[str] | None = None, override_exclude: Iterable[str] | None = None)¶

Bases: DrainageLinkEntity

Drainage conduits: pipes and channels carrying gravity flow.

barrels¶

cross_section_geometry¶

cross_section_shape¶

from_offset¶

length¶

roughness¶

to_offset¶

class DrainageLinkEntity(name: str | None = None, optional: bool | None = None, exclude: Iterable[str] | None = None, override_exclude: Iterable[str] | None = None)¶

Bases: LinkEntity

Base class for drainage network links with common output attributes.

Links route between from_node_id / to_node_id (inherited, INIT); the line geometry is not needed by the simulation.

current_setting¶

flow¶

flow_depth¶

flow_volume¶

froude_number¶

is_ready()¶

target_setting¶

class DrainageNodeEntity(name: str | None = None, optional: bool | None = None, exclude: Iterable[str] | None = None, override_exclude: Iterable[str] | None = None)¶

Bases: PointEntity

Base class for drainage network nodes with common output attributes.

Keeps geometry.x / geometry.y (used to write SWMM [COORDINATES]) and drops the elevation / reference attributes that the model does not need.

flooding_rate¶

generated_inflow¶

hydraulic_head¶

lateral_inflow¶

stored_volume¶

total_inflow¶

total_outflow¶

water_depth¶

class JunctionEntity(name: str | None = None, optional: bool | None = None, exclude: Iterable[str] | None = None, override_exclude: Iterable[str] | None = None)¶

Bases: DrainageNodeEntity

Drainage junctions: manholes / pipe connection nodes.

invert_elevation¶

max_depth¶

ponded_area¶

surcharge_depth¶

class OrificeEntity(name: str | None = None, optional: bool | None = None, exclude: Iterable[str] | None = None, override_exclude: Iterable[str] | None = None)¶

Bases: DrainageLinkEntity

Drainage orifices: openings (side or bottom) regulating flow.

cross_section_geometry carries the opening dimensions (Geom1 = height, Geom2 = width for rectangular orifices; Geom1 = diameter for circular).

crest_height¶

cross_section_geometry¶

discharge_coefficient¶

flap_gate¶

orifice_shape¶

orifice_type¶

class OutfallEntity(name: str | None = None, optional: bool | None = None, exclude: Iterable[str] | None = None, override_exclude: Iterable[str] | None = None)¶

Bases: DrainageNodeEntity

Drainage outfalls: terminal nodes with a boundary stage condition.

fixed_stage¶

flap_gate¶

invert_elevation¶

outfall_type¶

class OutletEntity(name: str | None = None, optional: bool | None = None, exclude: Iterable[str] | None = None, override_exclude: Iterable[str] | None = None)¶

Bases: DrainageLinkEntity

Drainage outlets: flow-control links with a head/depth-discharge rating.

Rating is either functional (flow = coefficient * x ^ exponent) or a tabular rating_curve of (head_or_depth, flow) points, selected by outlet_rating_type.

crest_height¶

flap_gate¶

outlet_rating_type¶

rating_coefficient¶

rating_curve¶

rating_exponent¶

class PumpEntity(name: str | None = None, optional: bool | None = None, exclude: Iterable[str] | None = None, override_exclude: Iterable[str] | None = None)¶

Bases: DrainageLinkEntity

Drainage pumps: lift water from the inlet node to the outlet node.

With no pump_curve (or an IDEAL pump_curve_type) the pump passes its inlet inflow directly. Otherwise the curve defines the pump behaviour.

pump_curve¶

pump_curve_type¶

shutoff_depth¶

startup_depth¶

class RainGageEntity(name: str | None = None, optional: bool | None = None, exclude: Iterable[str] | None = None, override_exclude: Iterable[str] | None = None)¶

Bases: PointEntity

Drainage rain gages: rainfall sources for subcatchments.

Rainfall is driven at runtime by another model publishing rainfall_intensity (applied via SWMM RainGage.total_precip); there is no configured time series.

rainfall¶

rainfall_intensity¶

class StorageEntity(name: str | None = None, optional: bool | None = None, exclude: Iterable[str] | None = None, override_exclude: Iterable[str] | None = None)¶

Bases: DrainageNodeEntity

Drainage storage units: ponds / basins / tanks.

The storage geometry is selected by storage_geometry (FUNCTIONAL / TABULAR / CYLINDRICAL / CONICAL / PARABOLIC / PYRAMIDAL). When it is unset the shape is inferred for backward compatibility: a storage_curve means TABULAR, otherwise FUNCTIONAL (area = constant + coefficient * depth ^ exponent). The geometric shapes read their L/W/Z dimensions from storage_geometry_parameters.

invert_elevation¶

max_depth¶

ponded_area¶

storage_coefficient¶

storage_constant¶

storage_curve¶

storage_exponent¶

storage_geometry¶

storage_geometry_parameters¶

surcharge_depth¶

class SubcatchmentEntity(name: str | None = None, optional: bool | None = None, exclude: Iterable[str] | None = None, override_exclude: Iterable[str] | None = None)¶

Bases: PolygonEntity

Drainage subcatchments: surfaces generating runoff from rainfall.

Runoff is routed to outlet_node_id and driven by the rain gage referenced through raingage_id. The polygon geometry is optional and used for display only, so readiness does not depend on it.

Only a single subarea per subcatchment is modelled (SWMM supports more).

area¶

conductivity¶

curve_number¶

decay_constant¶

dry_time¶

evaporation_loss¶

infiltration_loss¶

initial_deficit¶

is_ready()¶

max_infiltration_rate¶

min_infiltration_rate¶

n_imperv¶

n_perv¶

outlet_node_id¶

pct_zero¶

percent_impervious¶

rainfall¶

raingage_id¶

runoff¶

s_imperv¶

s_perv¶

slope¶

suction_head¶

width¶

Bases: object

conduits: ConduitEntity¶

junctions: JunctionEntity¶

orifices: OrificeEntity¶

outfalls: OutfallEntity¶

outlets: OutletEntity¶

pumps: PumpEntity¶

raingages: RainGageEntity¶

storage: StorageEntity¶

subcatchments: SubcatchmentEntity¶

weirs: WeirEntity¶

class WeirEntity(name: str | None = None, optional: bool | None = None, exclude: Iterable[str] | None = None, override_exclude: Iterable[str] | None = None)¶

Bases: DrainageLinkEntity

Drainage weirs: crested structures regulating overflow.

cross_section_geometry carries the opening dimensions (Geom1 = height, Geom2 = length, Geom3 = side slope for trapezoidal/v-notch weirs).

crest_height¶

cross_section_geometry¶

discharge_coefficient¶

flap_gate¶

weir_type¶

inp_writer¶

Helpers for synthesising a SWMM .inp input file from a Movici dataset.

pyswmm requires an .inp file to open a Simulation. Unlike WNTR (which builds its network model in memory), there is no supported way to construct a SWMM model purely in code, so the simulation wrapper writes a transient .inp describing the Movici dataset and opens the simulation on it.

InpBuilder collects rows per [SECTION] and renders them into the ordered, whitespace-delimited text that SWMM expects.

class InpBuilder¶

Bases: object

Collects .inp rows by section and renders the full input file.

add(section: str, *cells: Any) → None¶: Append a whitespace-delimited row of cells to section.

render() → str¶: Render all collected sections into .inp text.

fmt_hms(seconds: float) → str¶: Format a duration in seconds as a SWMM HH:MM:SS string.

fmt_num(value: float, max_decimals: int | None = None) → int | float¶

Coerce a numeric value for an .inp cell, keeping full precision by default.

Returns an int for whole values (so 10 rather than 10.0) and otherwise the float unchanged; the builder renders it with str(), which round-trips full float64 precision (SWMM parses .inp numbers into doubles, so nothing is truncated). Pass max_decimals to round for a more compact file.

Undefined (NaN) values are coerced to 0: fixed-shape geometry attributes (e.g. cross_section_geometry with 4 slots) often leave the slots a given SWMM shape does not use unspecified, which surfaces as NaN here. SWMM ignores the unused Geom cells, so emitting 0 is correct and avoids int(nan) raising.

model¶

Urban drainage simulation model using pyswmm/SWMM.

This model simulates urban drainage (storm water and sewer) networks using the EPA SWMM engine through pyswmm. It performs dynamic-wave hydraulic routing and rainfall-runoff hydrology over subcatchments.

Unlike the WNTR-backed drinking-water model (which re-runs a full simulation per update), this model keeps a single live SWMM Simulation open and advances it forward to each Movici moment. Control inputs (rainfall, node inflow and regulator settings) are applied to the live simulation before advancing, because SWMM marches forward and cannot rewind.

Note

Internal SWMM controls (rules / curves) are not used; control logic should be supplied externally through the regulator target_setting and node generated_inflow attributes (e.g. via the Movici Rules Model).

class Model(model_config: dict)¶

Bases: TrackedModel

Urban drainage (storm/sewer) network simulation model using SWMM.

Simulates:

Dynamic-wave hydraulic routing through conduits, pumps, orifices, weirs and outlets
Rainfall-runoff hydrology over subcatchments driven by rain gages
Junctions, outfalls and storage units as network nodes

Per-step results (depth, head, flooding, flow, runoff, …) are published; rainfall, external inflow and regulator settings may be supplied as inputs.

classmethod get_schema_attributes()¶: Return all AttributeSpecs used by this model.

initialize(state: TrackedState)¶: Validate the network, synthesise the .inp and open the simulation.

setup(state: TrackedState, logger: Logger, **kwargs)¶: Register entity groups and wire up the simulation wrapper.

shutdown(state: TrackedState)¶: Finalise and close the simulation.

update(state: TrackedState, moment: Moment) → Moment | None¶

Advance the live simulation to moment and publish the results.

Control inputs are applied to the live simulation before advancing, since SWMM marches forward and cannot rewind. When updated again at a timestep already simulated, there is nothing to do: the (late-arriving) control changes are read from the arrays on the next forward step.

simulation_wrapper¶

Wrapper around a live pyswmm Simulation for Movici.

This module bridges Movici’s entity-based, update(moment)-driven model with SWMM’s live, forward-only simulation engine:

the Movici dataset is synthesised into a transient .inp file (see inp_writer) and a single Simulation is opened and kept alive for the whole run;
each Movici update advances that simulation to the requested moment with step_advance + iteration (SWMM cannot rewind, so we never step backwards);
control inputs (rainfall, node inflow, regulator settings) are applied to the live objects before advancing, so they govern the step about to be taken;
per-step results are read directly off the live Node / Link / Subcatchment objects (there is no results DataFrame as in WNTR).

Per-entity-group SwmmProcessors know how to (a) contribute rows to the synthesised .inp, (b) apply control inputs and (c) read results back.

class ConduitProcessor(wrapper: SimulationWrapper, entity_group: T)¶

Bases: LinkProcessor[ConduitEntity]

PREFIX: str = 'C'¶

build_inp(builder: InpBuilder) → None¶

class IdMapper¶

Bases: object

Maps Movici integer ids to SWMM string names (unique across all types).

Each processor uses a type-specific prefix ("J" for junctions, "C" for conduits, …) so names are unique across entity groups and link endpoints / subcatchment outlets resolve unambiguously.

get_swmm_name(entity_id: int) → str¶

names_by_id: Dict[int, str]¶

register(entity_id: int, swmm_name: str) → None¶

class JunctionProcessor(wrapper: SimulationWrapper, entity_group: T)¶

Bases: NodeProcessor[JunctionEntity]

PREFIX: str = 'J'¶

build_inp(builder: InpBuilder) → None¶

class LinkProcessor(wrapper: SimulationWrapper, entity_group: T)¶

Bases: SwmmProcessor[L]

Base for link processors: resolves endpoints, reads shared link outputs.

process_changes() → None¶

write_results() → None¶

class NodeProcessor(wrapper: SimulationWrapper, entity_group: T)¶

Bases: SwmmProcessor[N]

Base for node processors: writes coordinates, reads shared node outputs.

process_changes() → None¶

write_results() → None¶

class OrificeProcessor(wrapper: SimulationWrapper, entity_group: T)¶

Bases: LinkProcessor[OrificeEntity]

PREFIX: str = 'OR'¶

build_inp(builder: InpBuilder) → None¶

class OutfallProcessor(wrapper: SimulationWrapper, entity_group: T)¶

Bases: NodeProcessor[OutfallEntity]

PREFIX: str = 'OF'¶

build_inp(builder: InpBuilder) → None¶

class OutletProcessor(wrapper: SimulationWrapper, entity_group: T)¶

Bases: LinkProcessor[OutletEntity]

PREFIX: str = 'OU'¶

build_inp(builder: InpBuilder) → None¶

class PumpProcessor(wrapper: SimulationWrapper, entity_group: T)¶

Bases: LinkProcessor[PumpEntity]

PREFIX: str = 'PU'¶

build_inp(builder: InpBuilder) → None¶

class RainGageProcessor(wrapper: SimulationWrapper, entity_group: T)¶

Bases: SwmmProcessor[RainGageEntity]

PREFIX: str = 'RG'¶

build_inp(builder: InpBuilder) → None¶

process_changes() → None¶

write_results() → None¶

class SimulationWrapper(logger: Logger | None = None)¶

Bases: object

Owns the live pyswmm Simulation and its processors.

add_curve(points: ndarray, curve_type: str) → str¶: Register a SWMM curve and return its generated name.

advance_to(target_seconds: int) → None¶

Step the live simulation forward until at least target_seconds.

SWMM marches forward only, so this never steps backwards. Control inputs must already have been applied (see process_changes()).

checkpoint(path: str | None = None) → str¶

Snapshot the current simulation state to a hotstart file.

Pass the returned path to rollback_to() to restore this exact state (and reported time) later. May be called at any point during the run.

Parameters:: path – target .hsf path; a temporary file (cleaned up in close()) is created when omitted.
Returns:: the hotstart file path.

close() → None¶

Finalise and close the simulation, removing the transient .inp.

EPA-SWMM only permits one open simulation per process, so releasing the simulation (and the object collections that reference it) here is what lets a subsequent model run open its own simulation.

configure_options(config_options: dict, general_options: dict | None = None, start_datetime: datetime | None = None) → None¶

Store the model-config and dataset-general options for synthesising [OPTIONS].

Solver options (timesteps, routing) come from the model config; data options (flow_units, infiltration_model_default) come from the dataset general section; infiltration_model_override comes from the model config. _options is the merged view (general wins) used for the solver timesteps and routing.

elapsed_seconds() → float¶

property global_infiltration_model: str¶: The [OPTIONS] INFILTRATION keyword (per-row methods may override it).

property infiltration_default: str¶: Dataset-default infiltration model (general section), else HORTON.

property infiltration_override: str | None¶: Model-config infiltration model that forces all subcatchments, if any.

initialize(dataset: UrbanDrainageNetwork) → None¶: Build the processors, synthesise the .inp and open the simulation.

property is_us_units: bool¶

True when flow_units selects SWMM’s US (inch) unit system.

flow_units describes the data, so it is read from the dataset general section only (not overridable from the model config).

links: Links | None¶

nodes: Nodes | None¶

process_changes() → None¶: Apply all runtime control inputs to the live simulation objects.

processors: Dict[str, SwmmProcessor]¶

raingages: RainGages | None¶

rollback_to(checkpoint: str) → None¶

Restore the simulation to a state captured by checkpoint().

Closes the current engine (keeping the synthesised .inp, processors and id-map) and reopens it from the checkpoint, restoring both the engine state and the reported Movici time. The interval after the checkpoint can then be re-run - e.g. with corrected control inputs.

sim: Simulation | None¶

property start_date: str¶: SWMM MM/DD/YYYY start date (from the Movici timeline, else default).

property start_time: str¶: SWMM HH:MM:SS start time (from the Movici timeline, else default).

property subarea_defaults: Dict[str, float]¶: Default subarea/infiltration parameters for the active unit system.

subcatchments: Subcatchments | None¶

write_results() → None¶: Read the current simulation state into the PUBLISH attribute arrays.

class StorageProcessor(wrapper: SimulationWrapper, entity_group: T)¶

Bases: NodeProcessor[StorageEntity]

GEOMETRIC_SHAPES = ('CYLINDRICAL', 'CONICAL', 'PARABOLIC', 'PYRAMIDAL')¶

KNOWN_SHAPES = ('FUNCTIONAL', 'TABULAR', 'CYLINDRICAL', 'CONICAL', 'PARABOLIC', 'PYRAMIDAL')¶

PREFIX: str = 'ST'¶

build_inp(builder: InpBuilder) → None¶

class SubcatchmentProcessor(wrapper: SimulationWrapper, entity_group: T)¶

Bases: SwmmProcessor[SubcatchmentEntity]

PREFIX: str = 'S'¶

build_inp(builder: InpBuilder) → None¶

write_results() → None¶

class SwmmProcessor(wrapper: SimulationWrapper, entity_group: T)¶

Bases: Generic[T]

Base class translating one entity group to/from SWMM.

PREFIX: str = ''¶

build_inp(builder: InpBuilder) → None¶

process_changes() → None¶

register() → None¶: Register all of this group’s ids in the shared IdMapper.

swmm_name(entity_id: int) → str¶

write_results() → None¶

class WeirProcessor(wrapper: SimulationWrapper, entity_group: T)¶

Bases: LinkProcessor[WeirEntity]

PREFIX: str = 'W'¶

build_inp(builder: InpBuilder) → None¶

swmm_source¶

SWMM .inp DataSource for the Movici dataset creator.

Reads an existing SWMM .inp file and exposes its objects as Movici dataset creator sources, mirroring movici_drinking_water_model.epanet_source. This is the authoring counterpart to the model’s runtime .inp synthesis: it lets an existing SWMM model be imported into a Movici dataset.

The parser is intentionally dependency-free (no swmmio): it tokenises the [SECTION] blocks of the .inp and exposes scalar attributes plus point / line / polygon geometry. Enum-like columns (shapes, types) are exposed as their SWMM keyword strings, which the dataset creator’s enum-conversion step maps to integer indices.

class SWMMSource(file: Path | str)¶

Bases: MultipleEntityTypeSource

Multi-entity source for reading SWMM .inp files.

Registered as the "swmm" source type for the dataset creator. Contains entity types: junctions, outfalls, storage, conduits, pumps, orifices, weirs, outlets, subcatchments and raingages. Access individual entity types with bracket notation:

source = SWMMSource("network.inp")
source["junctions"].get_attribute("invert_elevation")

The .inp file is parsed lazily on first entity-type access and shared across sub-sources of the same SWMMSource instance.

Parameters:: file – Path to the .inp file

ENTITY_TYPES = frozenset({'conduits', 'junctions', 'orifices', 'outfalls', 'outlets', 'pumps', 'raingages', 'storage', 'subcatchments', 'weirs'})¶

classmethod from_source_info(source_info)¶

Create from a source info dictionary.

If entity_type is present, returns a single-entity DataSource for that type; otherwise returns the full multi-entity source.

get_bounding_box()¶

keys() → Iterable[str]¶

class SwmmInp(path: str | Path)¶

Bases: object

A parsed SWMM .inp file: tokenised sections plus geometry maps.

coordinates: Dict[str, Tuple[float, float]]¶

curves: Dict[str, Dict[str, Any]]¶

polygons: Dict[str, List[Tuple[float, float]]]¶

sections: Dict[str, List[List[str]]]¶

symbols: Dict[str, Tuple[float, float]]¶

vertices: Dict[str, List[Tuple[float, float]]]¶

xsections: Dict[str, List[str]]¶

get_float_or_none(row: Sequence[str], idx: int) → float | None¶: Return float(row[idx]) or None if the column is absent/non-numeric.

get_gate_or_none(row: Sequence[str], idx: int) → bool | None¶

Parse a SWMM gate flag (“YES”/”NO”) into a bool for a bool attribute.

The attribute is declared DataType(bool), so the raw keyword must be converted here: bool("NO") is True, so passing the string through would silently turn an open gate into a closed one.

get_string_or_none(row: Sequence[str], idx: int) → str | None¶