Rain gages

Note

Engine: OpenSWMM 6 — refactored. Documents openswmm.engine.Gages. Legacy users see Legacy SWMM 5 Solver.

The Gages class manages every rain gage in the model — the sources of rainfall data that drive subcatchment runoff. A gage may draw rainfall from a time series, an external file (NWS / NCDC formats), or live runtime overrides.

Reference: openswmm_gages.h.

---

Class signature

class Gages:
    def __init__(self, solver: Solver) -> None: ...

---

Key methods

Identity

Method	Returns
count()	Number of rain gages.
get_index(id)()	Integer index for a string id.
get_id(idx)()	String id for an integer index.
add(id)()	Append a new gage; returns its index.

Configuration

Method	Action
set_rain_type(idx, type)()	GageRainType (intensity / volume / cumulative).
set_rain_interval(idx, seconds)()	Time interval between recorded rainfall samples.
set_data_source(idx, source)()	GageDataSource (timeseries vs. file).
set_timeseries(idx, ts_id)()	Bind to an existing time-series id.
set_filename(idx, path, station_id)()	Bind to an external NWS/NCDC file.

Per-step rainfall (read & override)

Method	Action / returns
get_rainfall(idx)()	Current rainfall intensity at the gage.
set_rainfall(idx, r)()	Override the gage’s rainfall this step (one-shot).
get_rainfall_bulk()	All gage rainfall as np.ndarray[float64].

For sticky cross-step rainfall overrides, use Forcing.gage_rainfall() (see Advanced forcing).

---

End-to-end example

from openswmm.engine import Solver, Gages, EngineState

with Solver("site_drainage.inp", "site_drainage.rpt", "site_drainage.out") as s:
    gages = Gages(s)
    print(f"{gages.count()} rain gages")

    rg1 = gages.get_index("RG1")
    peak_r, t_peak = 0.0, 0.0
    while s.state == EngineState.RUNNING:
        if s.step() != 0:
            break
        r = gages.get_rainfall(rg1)
        if r > peak_r:
            peak_r, t_peak = r, s.elapsed
    print(f"RG1 peak rainfall = {peak_r:.4f} at t={t_peak*24:.2f} h")

---

Common recipes

Inject a custom hyetograph (one-shot per step)

rg1 = gages.get_index("RG1")
while s.state == EngineState.RUNNING:
    if s.step() != 0:
        break
    h = s.elapsed * 24.0
    if 1.0 <= h < 4.0:
        gages.set_rainfall(rg1, 0.8)        # heavy rain hours 1-4
    else:
        gages.set_rainfall(rg1, 0.0)

Sticky override via Forcing

from openswmm.engine import Forcing, ForcingMode

forcing = Forcing(s)
forcing.gage_rainfall("RG1", 1.2, ForcingMode.REPLACE, persist=True)
while s.state == EngineState.RUNNING:
    if s.step() != 0:
        break
forcing.clear_all()

Bulk rainfall snapshot for every gage

import numpy as np

history = []
while s.state == EngineState.RUNNING:
    if s.step() != 0:
        break
    history.append(gages.get_rainfall_bulk().copy())
R = np.stack(history)        # shape (T, n_gages)

Switch a gage from time-series to runtime control

from openswmm.engine import GageDataSource

rg1 = gages.get_index("RG1")
gages.set_data_source(rg1, int(GageDataSource.TIMESERIES))   # default
# ... at runtime, drive via set_rainfall() each step

---

Bulk arrays

Method	Returns / accepts
get_rainfall_bulk()	All gage rainfall (np.ndarray[float64], shape (n_gages,)).

The standard memory-aliasing rule applies: the array shares scratch memory; call .copy() on it if you keep it across steps.

---

EngineState requirements & exceptions

Method group	Required state	Notes
identity / count	OPENED or later	n/a
configuration setters (data source, file, etc.)	OPENED	Best done before initialize().
get_rainfall	RUNNING or ENDED	n/a
set_rainfall	RUNNING	One-shot.
get_rainfall_bulk	RUNNING or ENDED	Same memory-aliasing rules as other *_bulk methods.

---

See also

• Subcatchments — the consumers of gage rainfall.

• Advanced forcing — cross-step persistent rainfall overrides.

• Tables (time series, curves, patterns) — manage time series that gages can bind to.