epyt_flow.gym

epyt_flow.gym.scenario_control_env

Module provides a base class for control environments.

class epyt_flow.gym.scenario_control_env.ScenarioControlEnv(scenario_config: ScenarioConfig, autoreset: bool = False, reapply_uncertainties_at_reset: bool = False, **kwds)

Bases: ABC

Base class for a control environment challenge.

Parameters:

• scenario_config (ScenarioConfig) – Scenario configuration.

• autoreset (bool, optional) –

  If True, environment is automatically reset if terminated.

  The default is False.

• reapply_uncertainties_at_reset (bool, optional) –

  If True, the uncertainties are re-applied to the original properties at each reset.

  The default is False.

_scenario_sim

Scenario simulator of the control scenario.

Type:

ScenarioSimulator, protected

_scenario_config

Scenario configuration.

Type:

ScenarioConfig

_sim_generator

Generator for running the step-wise simulation.

Type:

Generator[Union[ScadaData, dict], bool, None], protected

_hydraulic_scada_data

SCADA data from the hydraulic simulation – only used if EPANET-MSX is used in the control scenario.

Type:

ScadaData, protected

property autoreset: bool

True, if environment automatically resets after it terminated.

Returns:

True, if environment automatically resets after it terminated.

Return type:

bool

close() → None

Frees all resources.

contains_events() → bool

Check if the scenario contains any events.

Returns:

True is the scenario contains any events, False otherwise.

Return type:

bool

property reapply_uncertainties_at_reset: bool

True, if the uncertainties are re-applied to the original properties at each reset.

Returns:

True, if the uncertainties are re-applied to the original properties at each reset.

Return type:

bool

reset() → tuple[ScadaData, bool] | ScadaData

Resets the environment (i.e. simulation).

Returns:

Current SCADA data (i.e. sensor readings).

Return type:

ScadaData

set_node_quality_source_value(node_id: str, pattern_id: str, qual_value: float) → None

Sets the quality source at a particular node to a specific value – e.g. setting the chlorine concentration injection to a specified value.

Parameters:

• node_id (str) – ID of the node.

• pattern_id (str) – ID of the quality pattern at the specific node.

• qual_value (float) – New quality source value.

set_node_species_source_value(species_id: str, node_id: str, source_type: int, pattern_id: str, source_strength: float) → None

Sets the species source at a particular node to a specific value – i.e. setting the species injection amount at a particular location.

Parameters:

• species_id (str) – ID of the species.

• node_id (str) – ID of the node.

• source_type (int) –

  Type of the external species injection source – must be one of the following EPANET constants:

  • EN_CONCEN = 0

  • EN_MASS = 1

  • EN_SETPOINT = 2

  • EN_FLOWPACED = 3

  Description:

  • E_CONCEN Sets the concentration of external inflow entering a node

  • EN_MASS Injects a given mass/minute into a node

  • EN_SETPOINT Sets the concentration leaving a node to a given value

  • EN_FLOWPACED Adds a given value to the concentration leaving a node

• pattern_id (str) – ID of the source pattern.

• source_strength (float) – Amount of the injected species (source strength) – i.e. interpreation of this number depends on source_type

set_pump_speed(pump_id: str, speed: float) → None

Sets the speed of a pump.

Parameters:

• pump_id (str) – ID of the pump for which the pump speed is set.

• speed (float) – New pump speed.

set_pump_status(pump_id: str, status: int) → None

Sets the status of a pump.

Parameters:

• pump_id (str) – ID of the pump for which the status is set.

• status (int) –

  New status of the pump – either active (i.e. open) or inactive (i.e. closed).

  Must be one of the following constants defined in ActuatorConstants:

  • EN_CLOSED = 0

  • EN_OPEN = 1

set_valve_status(valve_id: str, status: int) → None

Sets the status of a valve.

Parameters:

• valve_id (str) – ID of the valve for which the status is set.

• status (int) –

  New status of the valve – either open or closed.

  Must be one of the following constants defined in ActuatorConstants:

  • EN_CLOSED = 0

  • EN_OPEN = 1

abstract step(*actions) → tuple[ScadaData, float, bool] | tuple[ScadaData, float]

Performs the next step by applying an action and observing the consequences (SCADA data, reward, terminated).

Note that terminated is only returned if autoreset=False otherwise only the current SCADA data and reward are returned.

Returns:

Triple or tuple of observations (ScadaData), reward (float), and terminated (bool).

Return type:

( ScadaData , float, bool) or ( ScadaData , float)