openswmm_gymnasium.observations#

Observation builder and individual feature collectors. Plan §4.

The builder concatenates per-collector contributions into a single flat gymnasium.spaces.Box (the default) or a structured gymnasium.spaces.Dict (flatten=False).

author:: Caleb Buahin
copyright:: Copyright (c) 2026 Caleb Buahin
license:: MIT

class openswmm_gymnasium.observations.ObservationBuilder[source]#

Bases: object

Fluent composer for the env’s observation space.

Returned space is a flat gymnasium.spaces.Box of dtype float32 with shape (sum(collector.size),). Bounds are [-inf, +inf]; callers wishing finite bounds should wrap with a gymnasium.wrappers.NormalizeObservation or supply their own gymnasium.spaces.Box via a subclass in a later phase.

Example:

obs = (ObservationBuilder()
       .add_node_depths(["J1"])
       .add_link_flows(["C1"])
       .add_clock())

@ivar _collectors: Ordered list of bound-and-collect units. :type _collectors: list

add_clock(features=None)[source]#

Append a clock collector.

Parameters:: features (sequence of str or None) – Subset of C{(“hour_sin”, “hour_cos”, “elapsed_frac”)}. Defaults to all three.
Return type:: ObservationBuilder
Raises:: ValueError – If features contains an unrecognised key.

add_link_depths(link_ids)[source]#

Append a link-depth collector.

Return type:: ObservationBuilder
Parameters:: link_ids (Sequence[str])

add_link_flows(link_ids)[source]#

Append a link-flow collector.

Return type:: ObservationBuilder
Parameters:: link_ids (Sequence[str])

add_link_settings(link_ids)[source]#

Append a link-control-setting collector.

Return type:: ObservationBuilder
Parameters:: link_ids (Sequence[str])

add_node_depths(node_ids)[source]#

Append a node-depth collector.

Parameters:: node_ids (sequence of str) – Node IDs whose depths to observe.
Returns:: This builder, for chaining.
Return type:: ObservationBuilder

add_node_heads(node_ids)[source]#

Append a node-head collector.

Return type:: ObservationBuilder
Parameters:: node_ids (Sequence[str])

add_node_inflows(node_ids)[source]#

Append a node-inflow collector.

Return type:: ObservationBuilder
Parameters:: node_ids (Sequence[str])

add_node_overflows(node_ids)[source]#

Append a node-overflow (flooding rate) collector.

Return type:: ObservationBuilder
Parameters:: node_ids (Sequence[str])

add_rainfall(gage_ids)[source]#

Append a rain-gage rainfall collector.

Return type:: ObservationBuilder
Parameters:: gage_ids (Sequence[str])

add_subcatch_runoff(subcatch_ids)[source]#

Append a subcatchment-runoff collector.

Return type:: ObservationBuilder
Parameters:: subcatch_ids (Sequence[str])

bind(adapter)[source]#

Resolve symbolic IDs in every collector.

Parameters:: adapter (SolverAdapter) – Adapter wrapping the open solver.
Return type:: None

collect(adapter)[source]#

Read the current observation from the engine.

Parameters:: adapter (SolverAdapter) – Adapter wrapping the running solver.
Returns:: 1-D float32 array matching space.
Return type:: numpy.ndarray

space()[source]#

Construct the env’s observation space.

Return type:: gymnasium.spaces.Box
Raises:: ValueError – If no collectors have been added.