epyt_control.envs

epyt_control.envs.rl_env

This module contains a base class for reinforcement learning (RL) environments.

class epyt_control.envs.rl_env.RlEnv(*args: Any, **kwargs: Any)[source]

Bases: ScenarioControlEnv, Env

Base class for reinforcement learning environments.

Parameters:

  • scenario_config (epyt_flow.simulation.ScenarioConfig) – Config of the scenario.

  • gym_action_space (gymnasium.spaces.Space) – Gymnasium action space.

  • action_space (list[Action]) – List of all action spaces – one space for each element that can be controlled by the agent.

  • reload_scenario_when_reset (bool, optional) –

    If True, the scenario (incl. the .inp and .msx file) is reloaded from the hard disk. If False, only the simulation is reset.

    The default is True.

  • hyd_file_in (str, optional) –

    Path to an EPANET .hyd file containing the simulated hydraulics. Can only be used in conjunction with ‘hyd_scada_in’ in the case of an EPANET-MSX scenario. If set, hydraulics will not be simulated but taken from the specified file.

    The default is None.

  • hyd_scada_in (epyt_flow.simulation.ScadaData, optional) –

    ScadaData instance containing the simulated hydraulics – must match the hydraulics from ‘hyd_file_in’. Can only be used in conjunction with ‘hyd_file_in’.

    The default is None.

  • frozen_sensor_config (bool, optional) –

    If True, only the sensor readings from the observation space will be stored when running the simulation – note that this implies that the reward function can only use the observations. This can lead to a significant speed-up of the simulation.

    The default is False.

property action_space: gymnasium.spaces.Space

Returns the action space of this environment.

Returns:

Gymnasium (action) space instance.

Return type:

gymnasium.spaces.Space

property observation_space: gymnasium.spaces.Space

Returns the observation space of this environment.

Returns:

Gymnasium (observation) space instance.

Return type:

gymnasium.spaces.Space

reset(seed: int | None = None, options: dict[str, Any] | None = None) → tuple[numpy.ndarray, dict][source]

Resets this environment to an initial internal state, returning an initial observation and info.

Parameters:

  • seed (int, optional) –

    The seed that is used to initialize the environment’s PRNG.

    The default is None.

  • options (dict[str, Any], optional) –

    Additional information to specify how the environment is reset (optional, depending on the specific environment).

    The default is None.

Returns:

Observation (numpy.ndarray), {“scada_data”: ScadaData} (epyt_flow.simulation.ScadaData as additional info).

Return type:

tuple[numpy.ndarray, dict]

step(action: numpy.ndarray) → tuple[numpy.ndarray, float, bool, bool, dict][source]

Performs the next step by applying an action and observing the next state together with a reward.

Parameters:

action – Actions to be executed.

Returns:

Observation (numpy.ndarray), reward, terminated, False (truncated), {“scada_data”: ScadaData} (epyt_flow.simulation.ScadaData as additional info).

Return type:

tuple[numpy.ndarray, float, bool, bool, dict]

epyt_control.envs.hydraulic_control_env

This module contains a base class for EPANET control environments – i.e. controlling hydraulic actuators such as pumps and valves or single chemical (no EPANET-MSX support!).

class epyt_control.envs.hydraulic_control_env.EpanetControlEnv(*args: Any, **kwargs: Any)[source]

Bases: RlEnv

Base class for hydraulic control environments (incl. basic quality that can be simulated with EPANET only).

Parameters:

  • scenario_config – Configuration of the scenario.

  • pumps_speed_actions (list[PumpSpeedAction], optional) –

    List of pumps where the speed has to be controlled.

    The default is None.

  • pumps_state_actions (list[PumpStateAction], optional) –

    Lisst of pumps where the state has to be controlled.

    The default is None.

  • valves_state_actions (list[ValveStateAction], optional) –

    List of valves that have to be controlled.

    The default is None.

  • chemical_injection_actions (list[ChemicalInjectionAction], optional) –

    List chemical injection actions – i.e. places in the network where the injection of the chemical has to be controlled.

    The default is None.

epyt_control.envs.hydraulic_control_env.HydraulicControlEnv

alias of EpanetControlEnv

class epyt_control.envs.hydraulic_control_env.MultiConfigEpanetControlEnv(*args: Any, **kwargs: Any)[source]

Bases: EpanetControlEnv

Base class for hydraulic control environments (incl. basic quality that can be simulated with EPANET only) that can handle multiple scenario configurations – those scenarios are utilized in a round-robin scheduling scheme (i.e. autorest=True).

Parameters:

  • scenario_configs – List of all scenario configurations that are used in this environment.

  • pumps_speed_actions (list[PumpSpeedAction], optional) –

    List of pumps where the speed has to be controlled.

    The default is None.

  • pumps_state_actions (list[PumpStateAction], optional) –

    Lisst of pumps where the state has to be controlled.

    The default is None.

  • valves_state_actions (list[ValveStateAction], optional) –

    List of valves that have to be controlled.

    The default is None.

  • chemical_injection_actions (list[ChemicalInjectionAction], optional) –

    List chemical injection actions – i.e. places in the network where the injection of the chemical has to be controlled.

    The default is None.

  • autoreset (bool, optional) – If True, the environment is automatically reset when the episode ends. In this case, terminated will always be False, so the environment can’t be wrapped in a vectorized environment from Stable Baselines 3. If False, the environment’s step method returns terminated=True at the end of an episode, and reset has to be called to start the next scenario. The default is True.

  • reload_scenario_when_reset (bool, optional) –

    If True, the scenario (incl. the .inp and .msx file) is reloaded from the hard disk. If False, only the simulation is reset.

    The default is True.

reset(seed: int | None = None, options: dict[str, Any] | None = None) → tuple[numpy.ndarray, dict][source]

Resets this environment to an initial internal state, returning an initial observation and info.

Parameters:

  • seed (int, optional) –

    The seed that is used to initialize the environment’s PRNG.

    The default is None.

  • options (dict[str, Any], optional) –

    Additional information to specify how the environment is reset (optional, depending on the specific environment).

    The default is None.

Returns:

Observation (numpy.ndarray), {“scada_data”: ScadaData} (epyt_flow.simulation.ScadaData as additional info).

Return type:

tuple[numpy.ndarray, dict]

epyt_control.envs.hydraulic_control_env.MultiConfigHydraulicControlEnv

alias of MultiConfigEpanetControlEnv

epyt_control.envs.advanced_quality_control_env

This module contains a base class for an EPANET-MSX control environment – i.e. controlling the injection and reaction of one or multiple species in an EPANET-MSX scenario (no control over pumps, valves, etc.).

epyt_control.envs.advanced_quality_control_env.AdvancedQualityControlEnv

alias of EpanetMsxControlEnv

class epyt_control.envs.advanced_quality_control_env.EpanetMsxControlEnv(*args: Any, **kwargs: Any)[source]

Bases: RlEnv

Base class for advanced quality control scenarios – i.e. EPANET-MSX control scenarios.

Parameters:

  • scenario_config – Configuration of the scenario.

  • action_space (list[SpeciesInjectionAction]) – The action spaces (i.e. list of species injections) that have to be controlled by the agent.

  • rerun_hydraulics_when_reset (bool, optional) –

    If True, the hydraulic simulation is going to be re-run when the environment is reset, otherwise the hydraulics from the initial run are re-used and the scenario will also not be reloaded – i.e. reload_scenario_when_reset=False.

    The default is True.

  • hyd_file_in (str, optional) –

    Path to an EPANET .hyd file containing the simulated hydraulics. Can only be used in conjunction with ‘hyd_scada_in’. If set, hydraulics will not be simulated but taken from the specified file.

    The default is None.

  • hyd_scada_in –

    ScadaData instance containing the simulated hydraulics – must match the hydraulics from ‘hyd_file_in’. Can only be used in conjunction with ‘hyd_file_in’.

    The default is None.

reset(seed: int | None = None, options: dict[str, Any] | None = None) → tuple[numpy.ndarray, dict][source]

Resets this environment to an initial internal state, returning an initial observation and info.

Parameters:

  • seed (int, optional) –

    The seed that is used to initialize the environment’s PRNG.

    The default is None.

  • options (dict[str, Any], optional) –

    Additional information to specify how the environment is reset (optional, depending on the specific environment).

    The default is None.

Returns:

Observation (numpy.ndarray), {“scada_data”: ScadaData} (epyt_flow.simulation.ScadaData as additional info).

Return type:

tuple[numpy.ndarray, dict]

epyt_control.envs.advanced_quality_control_env.MultiConfigAdvancedQualityControlEnv

alias of MultiConfigEpanetMsxControlEnv

class epyt_control.envs.advanced_quality_control_env.MultiConfigEpanetMsxControlEnv(*args: Any, **kwargs: Any)[source]

Bases: EpanetMsxControlEnv

Base class for advanced quality control scenarios (i.e. EPANET-MSX control scenarios) that can handle multiple scenario configurations – those scenarios are utilized in a round-robin scheduling scheme (i.e. autorest=True).

Note that all scenarios must share the same action and observation space.

Parameters:

  • scenario_configs – Configuration of the scenario. Note that all sceanrios must share the same action and observation space.

  • action_space (list[SpeciesInjectionAction]) – The action spaces (i.e. list of species injections) that have to be controlled by the agent. Must be the same for all scenarios specified in ‘scenario_configs’.

  • rerun_hydraulics_when_reset (bool, optional) – If True, the hydraulic simulation is going to be re-run when the environment is reset, otherwise the hydraulics from the initial run are re-used and the scenario will also not be reloaded – i.e. reload_scenario_when_reset=False.

  • precomputed_hydraulics –

    Pre-computed hydraulics – i.e., for each scenario in ‘scenario_configs’, a tuple of a path to an EPANET generatd .hyd file and a ScadaData instance – that are used instead of re-running the hydraulic simulation. If used, ‘rerun_hydraulics_when_reset’ must be False.

    The default is None.

reset(seed: int | None = None, options: dict[str, Any] | None = None) → tuple[numpy.ndarray, dict][source]

Resets this environment to an initial internal state, returning an initial observation and info.

Parameters:

  • seed (int, optional) –

    The seed that is used to initialize the environment’s PRNG.

    The default is None.

  • options (dict[str, Any], optional) –

    Additional information to specify how the environment is reset (optional, depending on the specific environment).

    The default is None.

Returns:

Observation (numpy.ndarray), {“scada_data”: ScadaData} (epyt_flow.simulation.ScadaData as additional info).

Return type:

tuple[numpy.ndarray, dict]