openswmm_gymnasium.envs#

openswmm_gymnasium.envs#

Concrete gymnasium.Env subclasses. Plan §2.1.

  • SwmmRTCEnv — runtime-only (RTC); design action portion of the Dict action space is empty. P1.

  • SwmmCIPEnv — design-only (CIP); single-step contextual-bandit pattern. Runtime action portion is empty. P3.

  • SwmmJointCIPRTCEnv — full hybrid. Design applied at reset(), runtime applied each step(). P3.

author:

Caleb Buahin

copyright:

Copyright (c) 2026 Caleb Buahin

license:

MIT

class openswmm_gymnasium.envs.SwmmCIPEnv(*args, **kwargs)[source]#

Bases: Env

Single-step CIP environment for design-only optimisation.

@ivar metadata: Gymnasium metadata (no rendering; viz §5.5 consumes

recorded trajectories).

@ivar action_space: Dict with non-empty "design" and empty

"runtime".

@ivar observation_space: Flat Box from the supplied observation

builder.

Parameters:
close()[source]#

Close any held solver.

Return type:

None

metadata: dict[str, Any] = {'render_modes': []}#
reset(*, seed=None, options=None)[source]#

Reset the env. Returns a zero observation; the agent acts next.

Parameters:

  • seed (int or None) – Optional Gymnasium seed.

  • options (dict or None) – Reserved; currently ignored.

Returns:

(obs, info) per Gymnasium 1.x. obs is a zero vector matching observation_space.

Return type:

tuple

step(action)[source]#

Apply the design action and run a full simulation.

Parameters:

action (dict) – Dict matching action_space. Only the "design" subdict is used.

Returns:

(final_obs, reward, True, False, info) — the env is a single-step contextual bandit.

Return type:

tuple

class openswmm_gymnasium.envs.SwmmJointCIPRTCEnv(*args, **kwargs)[source]#

Bases: Env

Hybrid CIP + RTC environment.

@ivar metadata: Gymnasium metadata. @ivar action_space: Dict with non-empty "design" and "runtime". @ivar observation_space: Flat Box from the supplied observation

builder.

Parameters:

  • inp_path (PathLike)

  • design_factories (Sequence[DesignActionFactory])

  • runtime_factories (Sequence[OrificeSetting] | None)

  • observation_builder (ObservationBuilder | None)

  • reward_terms (Sequence[RewardTerm] | None)

  • control_interval_steps (int)

  • max_episode_steps (int | None)

  • rpt_path (PathLike | None)

  • out_path (PathLike | None)

close()[source]#

Close the underlying solver. Safe to call multiple times.

Return type:

None

metadata: dict[str, Any] = {'render_modes': []}#
reset(*, seed=None, options=None)[source]#

Start a new episode under a freshly-applied design.

If options["design_action"] is supplied, that design is used verbatim; otherwise the design is sampled from action_space["design"].

Parameters:

  • seed (int or None) – Optional Gymnasium seed.

  • options (dict or None) – Optional dict; may include "design_action" (a dict matching action_space["design"]).

Returns:

(obs, info) per Gymnasium 1.x. info["design_action"] records the design that was applied.

Return type:

tuple

step(action)[source]#

Advance the simulation by control_interval_steps.

action["design"] is ignored — the design is fixed at reset. action["runtime"] is applied via the runtime factories.

Parameters:

action (dict) – Dict matching action_space.

Returns:

Gymnasium 1.x 5-tuple.

Return type:

tuple

Raises:

RuntimeError – If called before reset.

class openswmm_gymnasium.envs.SwmmMORTCEnv(*args, **kwargs)[source]#

Bases: SwmmMORTCEnv, MOEnv

SwmmMORTCEnv additionally subclassing mo_gymnasium.MOEnv.

Parameters:

  • args (Any)

  • ideal_point (Sequence[float] | None)

  • reference_point (Sequence[float] | None)

  • runtime_factories (Sequence[OrificeSetting] | None)

  • observation_builder (ObservationBuilder | None)

  • reward_terms (Sequence[RewardTerm] | None)

  • kwargs (Any)

class openswmm_gymnasium.envs.SwmmRTCEnv(*args, **kwargs)[source]#

Bases: Env

Runtime-only SWMM environment for RL.

The action space is spaces.Dict({"design": Dict({), “runtime”: Dict({…})})}, conforming to the plan §3 contract that every env exposes both top-level keys. "design" is empty for this env class.

The observation space is a flat gymnasium.spaces.Box produced by the supplied ObservationBuilder.

@ivar metadata: Gymnasium metadata (no rendering for now; the

Plotly viz module §5.5 consumes recorded trajectories, not live envs).

@ivar action_space: Dict of "design" + "runtime". @ivar observation_space: Flat Box.

Parameters:

  • inp_path (PathLike)

  • runtime_factories (Sequence[OrificeSetting] | None)

  • observation_builder (ObservationBuilder | None)

  • reward_terms (Sequence[RewardTerm] | None)

  • control_interval_steps (int)

  • max_episode_steps (int | None)

  • rpt_path (PathLike | None)

  • out_path (PathLike | None)

close()[source]#

Close the underlying solver. Safe to call multiple times.

Return type:

None

metadata: dict[str, Any] = {'render_modes': []}#
reset(*, seed=None, options=None)[source]#

Start a new episode.

Closes any prior solver, opens a fresh one against inp_path, binds all factories / collectors / reward terms, and returns the initial observation.

Parameters:

  • seed (int or None) – Optional seed forwarded to gymnasium.Env.reset.

  • options (dict or None) – Reserved for future use; currently ignored.

Returns:

Tuple (observation, info) per Gymnasium 1.x.

Return type:

tuple

step(action)[source]#

Advance the simulation by control_interval_steps.

Parameters:

action (dict) – Dict matching action_space.

Returns:

Tuple C{(observation, reward, terminated, truncated, info)} per Gymnasium 1.x.

Return type:

tuple

Raises:

RuntimeError – If called before reset.