Hydraulic Engine

Testing
Package
Meta

Python toolkit to run SWMM and EPANET simulations, work with their input and result files, and export results to a Giswater PostgreSQL database or a FROST SensorThings endpoint.

Features

• Run SWMM (pyswmm) and EPANET (WNTR) simulations, with optional progress callbacks
• Read and write SWMM / EPANET INP models
• Read SWMM RPT and OUT results, and EPANET BIN results
• Export simulation results to PostgreSQL (Giswater) or FROST
• Connect to PostgreSQL, SQLite, and GeoPackage

Installation

pip install hydraulic-engine

From source:

git clone https://github.com/bgeo-gis/hydraulic-engine.git
cd hydraulic-engine
pip install -e .

For development:

pip install -e ".[dev]"

Quick start

Run a simulation

import hydraulic_engine as he
from hydraulic_engine.utils.enums import RunStatus

# SWMM
swmm = he.swmm.SwmmRunner(inp_path="drainage.inp")
swmm_result = swmm.run()

# EPANET
epanet = he.epanet.EpanetRunner(inp_path="network.inp")
epanet_result = epanet.run()

if swmm_result.status == RunStatus.SUCCESS:
    print(f"SWMM finished in {swmm_result.duration_seconds:.2f}s")

run() uses a hybrid error contract: load/validation failures and unexpected engine crashes raise (FileLoadError, ValidationError, SimulationError); cancel via step_callback returning False raises SimulationCancelled. Finished runs still return a result whose status may be SUCCESS, WARNING, or ERROR from RPT/output checks.

Pass progress_callback=fn to either runner to receive progress updates during the run.

Work with model files

from hydraulic_engine.swmm import SwmmInpHandler
from hydraulic_engine.epanet import EpanetInpHandler

swmm_inp = SwmmInpHandler()
swmm_inp.load_file("drainage.inp")
print(swmm_inp.get_summary())

epanet_inp = EpanetInpHandler()
epanet_inp.load_file("network.inp")
print(epanet_inp.get_summary())

Result files use the matching handlers: SwmmRptHandler, SwmmOutHandler, and EpanetBinHandler.

Export results

After a successful run, push results to the database or to FROST:

import hydraulic_engine as he

runner = he.swmm.SwmmRunner(inp_path="drainage.inp")
runner.run()

dao = he.create_pg_connection(
    host="localhost",
    port=5432,
    dbname="hydraulic_db",
    user="user",
    password="pass",
    schema="my_schema",
)

runner.export_result(
    to=he.ExportDataSource.DATABASE,
    result_id="my_result",  # must already exist in rpt_cat_result
    client=dao,
)

The same export_result(..., to=he.ExportDataSource.FROST) path works for SensorThings via he.create_frost_connection(...). EPANET uses the same pattern on EpanetRunner.

Exported SWMM time series follow what the model’s [REPORT] section requests (nodes, links, subcatchments).

Database connections

import hydraulic_engine as he

pg = he.create_pg_connection(host="localhost", port=5432, dbname="db", user="u", password="p")
gpkg = he.create_gpkg_connection("project.gpkg")
sqlite = he.create_sqlite_connection("local.db")

he.close_connection()

Package layout

Development

pytest tests/
black src/
ruff check src/

License

GNU General Public License v3.0 or later — see LICENSE.

Authors

BGEO — info@bgeo.es