epyt_flow.simulation

epyt_flow.simulation.scenario_config

Module provides a class for specifying scenario configurations.

class epyt_flow.simulation.scenario_config.ScenarioConfig(scenario_config: Any | None = None, f_inp_in: str | None = None, f_msx_in: str | None = None, network_topology: NetworkTopology | None = None, general_params: dict | None = None, sensor_config: SensorConfig | None = None, memory_consumption_estimate: float | None = None, custom_controls: list[CustomControlModule] = [], simple_controls: list[SimpleControlModule] = [], complex_controls: list[ComplexControlModule] = [], sensor_noise: SensorNoise | None = None, model_uncertainty: ModelUncertainty | None = None, system_events: list[SystemEvent] = [], sensor_reading_events: list[SensorReadingEvent] = [], **kwds)[source]

Bases: Serializable

Configuration of a scenario.

Parameters:

  • scenario_config (ScenarioConfig, optional) –

    Uses the given scenario configuration to create this instance – other attributes passed to this constructor override the attributes in ‘scenario_config’.

    Note that if ‘scenario_config’ is None then ‘f_inp_in’ can not be None – i.e. either ‘scenario_config’ or ‘f_inp_in’ must be given.

    The default is None.

  • f_inp_in (str, optional) –

    Path to the .inp file.

    Note that if ‘f_inp_in’ is None then ‘scenario_config’ can not be None – i.e. either ‘scenario_config’ or ‘f_inp_in’ must be given.

    The default is None.

  • f_msx_in (str, optional) –

    Path to the .msx file – optional, only necessary if EPANET-MSX is used.

    The default is None

  • network_topology (NetworkTopology, optional) –

    Specification of the network topology – necessary if .inp file does not exist or is not shared.

    The default is None.

  • general_params (dict, optional) –

    General parameters such as the demand model, hydraulic time steps, etc.

    The default is None

  • sensor_config (SensorConfig, optional) –

    Specification of all sensors.

    The default is None

  • memory_consumption_estimate (float, optional) –

    Estimated memory consumption of this scenario in MB – i.e. the amount of memory that is needed on the hard disk as well as in RAM.

    The default is None.

  • sensor_noise (SensorNoise, optional) –

    Speciation of sensor noise – i.e. noise/uncertainty affecting the sensor readings.

    The default is None

  • csutom_controls (list[CustomControlModule], optional) –

    List of custom control modules that are active during the simulation.

    The default is an empty list.

  • simple_controls (list[SimpleControlModule], optional) –

    List of EPANET control rules that are active during the simulation.

    The default is an empty list.

  • complex_controls (list[ComplexControlModule], optional) –

    List of complex (i.e. IF-THEN-ELSE) EPANET control rules that are active during the simulation.

    The default is an empty list.

  • model_uncertainty (ModelUncertainty, optional) – Specification of model uncertainty.

  • system_events (list[SystemEvent], optional) –

    List of system events – i.e. events that directly affect the simulation (e.g. leakages).

    The default is an empty list.

  • sensor_reading_events (list[SensorReadingEvent], optional) –

    List of sensor reading events – i.e. events that affect the readings of sensors.

    The default is an empty list.

property complex_controls: list[ComplexControlModule]

Gets the list of all complex (i.e. IF-THEN-ELSE) EPANET control rules that are active during the simulation.

Returns:

List of all complex EPANET control rules that are active during the simulation.

Return type:

list[ComplexControlModule]

property custom_controls: list[CustomControlModule]

Returns the list of all custom control modules that are active during the simulation.

Returns:

List of all custom control modules that are active during the simulation.

Return type:

list[CustomControlModule]

property f_inp_in: str

Gets the path to the .inp file.

Returns:

Path to the .inp file.

Return type:

str

property f_msx_in: str

Gets the path to the .msx file.

Returns:

Path to the .msx file.

Return type:

str

static file_ext() → str

Returns the file extension of this class.

This function is automatically implemented by applying the serializable() decorator.

Returns:

File extension.

Return type:

str

property general_params: dict

Gets general parameters such as hydraulic time step, etc.

Returns:

All general parameters as dictionary – the parameter name serves as a key.

Return type:

dict

get_attributes() → dict[source]

Gets all attributes to be serialized – these attributes are passed to the constructor when the object is deserialized.

Returns:

Dictionary of attributes – i.e. pairs of attribute name + value.

Return type:

dict

static load_from_json(config_data: str) → Any[source]

Loads a scenario configuration from a given JSON string.

Parameters:

config_data (str) – JSON data.

Returns:

Loaded scenario configuration.

Return type:

ScenarioConfig

static load_from_json_file(f_json_in: str) → Any[source]

Loads a scenario configuration from a given JSON file.

Parameters:

f_json_in (str) – Path to JSON configuration file.

Returns:

Loaded scenario configuration.

Return type:

ScenarioConfig

property memory_consumption_estimate: float

Gets the estimated memory consumption of this scenario – i.e. the amount of memory that is needed on the hard disk as well as in RAM.

Returns:

Estimated memory consumption in MB.

Return type:

float

property model_uncertainty: ModelUncertainty

Gets the model uncertainty specification.

Returns:

Model uncertainty specification.

Return type:

ModelUncertainty

property network_topology: NetworkTopology

Returns the specification of the network topology.

Returns:

Network topology.

Return type:

NetworkTopology

property sensor_config: SensorConfig

Gets the sensor configuration.

Returns:

Sensor configuration.

Return type:

SensorConfig

property sensor_noise: SensorNoise

Gets the sensor noise/uncertainty specification.

Returns:

Sensor noise/uncertainty.

Return type:

SensorNoise

property sensor_reading_events: list[SensorReadingEvent]

Gets all sensor reading events.

Returns:

All sensor reading events.

Return type:

list[SensorReadingEvent]

property simple_controls: list[SimpleControlModule]

Gets the list of all EPANET control rules that are active during the simulation.

Returns:

List of all EPANET control rules that are active during the simulation.

Return type:

list[SimpleControlModule]

property system_events: list[SystemEvent]

Gets all system events.

Returns:

All system events.

Return type:

list[SystemEvent]

epyt_flow.simulation.simulator

Module provides a class for scenario simulations.

class epyt_flow.simulation.scenario_simulator.ScenarioSimulator(f_inp_in: str | None = None, f_msx_in: str | None = None, scenario_config: ScenarioConfig | None = None, epanet_verbose: bool = False, raise_exception_on_error: bool = False, warn_on_error: bool = True, ignore_error_codes: list[int] = [])[source]

Bases: object

Class for running a simulation of a water distribution network scenario.

Parameters:

  • f_inp_in (str) –

    Path to the .inp file.

    If this is None, then ‘scenario_config’ must be set with a valid configuration.

  • f_msx_in (str, option) –

    Path to the .msx file – optional, only necessary if EPANET-MSX is used.

    The default is None.

  • scenario_config (ScenarioConfig) –

    Configuration of the scenario – i.e. a description of the scenario to be simulated.

    If this is None, then ‘f_inp_in’ must be set with a valid path to the .inp file that is to be simulated.

  • epanet_verbose (bool, optional) –

    If True, EPyT is verbose and might print messages from time to time.

    The default is False.

  • raise_exception_on_error (bool, optional) –

    If True, an exception is raised whenever an error occurs in EPANET or EPANET-MSX.

    The default is False.

  • warn_on_error (bool, optional) –

    If True, a warning is generated whenever an error occurs in EPANET or EPANET-MSX.

    The default is True.

  • ignore_error_codes (list[int], optional) –

    List of error codes that should be ignored – i.e., no exception or warning will be generated. However, error codes will still be included in the SCADA data.

    The default is [].

epanet_api

API to EPANET and EPANET-MSX.

Type:

EPyT

_model_uncertainty

Model uncertainty.

Type:

ModelUncertainty, protected

_sensor_noise

Sensor noise.

Type:

SensorNoise, protected

_sensor_config

Sensor configuration.

Type:

SensorConfig, protected

_custom_controls

List of custom control modules.

Type:

list[CustomControlModule], protected

_simple_controls

List of simle EPANET control rules.

Type:

list[SimpleControlModule], protected

_complex_controls

List of complex (IF-THEN-ELSE) EPANET control rules.

Type:

list[ComplexControlModule], protected

_system_events

Lsit of system events such as leakages.

Type:

list[SystemEvent], protected

_sensor_reading_events

List of sensor reading events such as sensor override attacks.

Type:

list[SensorReadingEvent], protected

property actuator_events: list[ActuatorEvent]

Gets all actuator events.

Returns:

All actuator events.

Return type:

list[ActuatorEvent]

add_actuator_event(event: ActuatorEvent) → None[source]

Adds an actuator event to the scenario simulation.

Parameters:

event (ActuatorEvent) – Actuator event.

add_complex_control(control: ComplexControlModule) → None[source]

Adds an complex (IF-THEN-ELSE) EPANET control rule to the scenario simulation.

Parameters:

control (ComplexControlModule) – Complex EPANET control module.

add_custom_control(control: CustomControlModule) → None[source]

Adds a custom control module to the scenario simulation.

Parameters:

control (CustomControlModule) – Custom control module.

add_leakage(leakage_event: Leakage) → None[source]

Adds a leakage to the scenario simulation.

Parameters:

event (Leakage) – Leakage.

add_pattern(pattern_id: str, pattern: numpy.ndarray) → None[source]

Adds a pattern to the EPANET scenario.

Parameters:

  • pattern_id (str) – ID of the pattern.

  • pattern (numpy.ndarray) – Pattern of multipliers over time.

add_quality_source(node_id: str, source_type: int, pattern: numpy.ndarray | None = None, pattern_id: str | None = None, source_strength: int = 1.0) → None[source]

Adds a new external water quality source at a particular node.

Parameters:

  • node_id (str) – ID of the node at which this external water quality source is placed.

  • source_type (int,) –

    Types of the external water quality source – must be 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 –

    1d source pattern multipiers over time – i.e. quality-source = source_strength * pattern.

    If None, the pattern pattern_id is assume to already exist.

    The default is None.

  • pattern_id (str, optional) –

    ID of the source pattern.

    If None, a pattern_id will be generated automatically – be aware that this could conflict with existing pattern IDs (in this case, an exception is raised).

    The default is None.

  • source_strength (int, optional) –

    Quality source strength – i.e. quality-source = source_strength * pattern.

    The default is 1.

add_sensor_fault(sensor_fault_event: SensorFault) → None[source]

Adds a sensor fault to the scenario simulation.

Parameters:

sensor_fault_event (SensorFault) – Sensor fault specifications.

add_sensor_reading_attack(sensor_reading_attack: SensorReadingAttack) → None[source]

Adds a sensor reading attack to the scenario simulation.

Parameters:

sensor_reading_attack (SensorReadingAttack) – Sensor fault specifications.

add_sensor_reading_event(event: SensorReadingEvent) → None[source]

Adds a sensor reading event to the scenario simulation.

Parameters:

event (SensorReadingEvent) – Sensor reading event.

add_simple_control(control: SimpleControlModule) → None[source]

Adds a simple EPANET control rule to the scenario simulation.

Parameters:

control (SimpleControlModule) – Simple EPANET control module.

add_species_injection_source(species_id: str, node_id: str, pattern: numpy.ndarray, source_type: int, pattern_id: str | None = None, source_strength: float = 1.0) → None[source]

Adds a new external bulk species injection source at a particular node.

Only for EPANET-MSX scenarios.

Parameters:

  • species_id (str) – ID of the (bulk or surface) species.

  • node_id (str) – ID of the node at which this external (bulk or surface) species injection source is placed.

  • pattern –

    1d source pattern.

    Note that the pattern time step is equivalent to the EPANET pattern time step.

  • source_type (int,) –

    Type of the external (bulk or surface) 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, optional) –

    ID of the source pattern.

    If None, a pattern_id will be generated automatically – be aware that this could conflict with existing pattern IDs (in this case, an exception is raised).

    The default is None.

  • source_strength (float, optional) –

    Injection source strength – i.e. injection = source_strength * pattern.

    The default is 1.

add_system_event(event: SystemEvent) → None[source]

Adds a system event to the scenario simulation – i.e. an event directly affecting the EPANET simulation.

Parameters:

event (SystemEvent) – System event.

close()[source]

Closes & unloads all resources and libraries.

Call this function after the simulation is done – do not call this function before!

property complex_controls: list[SimpleControlModule]

Gets all complex (IF-THEN-ELSE) EPANET control rules.

Returns:

All complex EPANET control rules.

Return type:

list[ComplexControlModule]

property custom_controls: list[CustomControlModule]

Returns all custom control modules.

Returns:

All custom control modules.

Return type:

list[CustomControlModule]

enable_chemical_analysis(chemical_name: str = 'Chlorine', chemical_units: int = 4) → None[source]

Sets chemical analysis.

ATTENTION: Do not forget to inject this chemical into the WDN.

Parameters:

  • chemical_name (str, optional) –

    Name of the chemical being analyzed.

    The default is “Chlorine”.

  • chemical_units (str, optional) –

    Units that the chemical is measured in.

    Must be one of the following constants:

    • MASS_UNIT_MG = 4 (mg/L)

    • MASS_UNIT_UG = 5 (ug/L)

    The default is MASS_UNIT_MG.

enable_sourcetracing_analysis(trace_node_id: str) → None[source]

Set source tracing analysis – i.e. tracks the percentage of flow from a given node reaching all other nodes over time.

Parameters:

trace_node_id (str) – ID of the node traced in the source tracing analysis.

enable_waterage_analysis() → None[source]

Sets water age analysis – i.e. estimates the water age (in hours) at all places in the network.

estimate_memory_consumption() → float[source]

Estimates the memory consumption of the simulation – i.e. the amount of memory that is needed on the hard disk as well as in RAM.

Returns:

Estimated memory consumption in MB.

Return type:

float

property f_inp_in: str

Gets the path to the .inp file.

Returns:

Path to the .inp file.

Return type:

str

property f_msx_in: str

Gets the path to the .msx file.

Returns:

Path to the .msx file.

Return type:

str

get_demand_model() → dict[source]

Gets the demand model and its parameters.

Returns:

Demand model.

Return type:

dict

get_events_active_time_points() → list[int][source]

Gets a list of time points (i.e. seconds since simulation start) at which at least one event (system or sensor readinge event) is active.

Returns:

List of time points at which at least one event is active.

Return type:

list[int]

get_flow_units() → int[source]

Gets the flow units.

Will be one of the following EPANET constants:

  • EN_CFS = 0 (cu foot/sec)

  • EN_GPM = 1 (gal/min)

  • EN_MGD = 2 (Million gal/day)

  • EN_IMGD = 3 (Imperial MGD)

  • EN_AFD = 4 (ac-foot/day)

  • EN_LPS = 5 (liter/sec)

  • EN_LPM = 6 (liter/min)

  • EN_MLD = 7 (Megaliter/day)

  • EN_CMH = 8 (cubic meter/hr)

  • EN_CMD = 9 (cubic meter/day)

Returns:

Flow units.

Return type:

int

get_hydraulic_time_step() → int[source]

Gets the hydraulic time step – i.e. time step in the hydraulic simulation.

Returns:

Hydraulic time step in seconds.

Return type:

int

get_node_base_demand(node_id: str) → float[source]

Returns the base demand of a given node. None, if there does not exist any base demand.

Note that base demands are summed up in the case of different demand categories.

Parameters:

node_id (str) – ID of the node.

Returns:

Base demand.

Return type:

float

get_node_demand_pattern(node_id: str) → numpy.ndarray[source]

Returns the values of the primary demand pattern of a given node – i.e. multiplier factors that are applied to the base demand.

Parameters:

node_id (str) – ID of the node.

Returns:

The demand pattern – i.e. multiplier factors over time.

Return type:

numpy.ndarray

get_pattern(pattern_id: str) → numpy.ndarray[source]

Returns the EPANET pattern (i.e. all multiplier factors over time) given its ID.

Parameters:

pattern_id (str) – ID of the pattern.

Returns:

The pattern – i.e. multiplier factors over time.

Return type:

numpy.ndarray

get_pressure_units() → int[source]

Returns the current pressure units.

Will be one of the following EPANET constants:

  • EN_PSI = 0 (Pounds per square inch)

  • EN_KPA = 1 (Kilopascals)

  • EN_METERS = 2 (Meters)

  • EN_BAR = 3 (Bar)

  • EN_FEET = 4 (Feet)

Returns:

Pressure units.

Return type:

int

get_pump_energy_price(pump_id: str) → float[source]

Returns the energy price of a given pump.

Parameters:

pump_id (str) – ID of the pump.

Returns:

Energy price.

Return type:

float

get_pump_energy_price_pattern(pump_id: str) → numpy.ndarray[source]

Returns the energy price pattern of a given pump.

Parameters:

pump_id (str) – ID of the pump.

Returns:

Energy price pattern. None, if none exists.

Return type:

numpy.ndarray

get_quality_model() → dict[source]

Gets the quality model and its parameters.

Note that this quality model refers to the basic quality analysis as implemented in EPANET.

Returns:

Quality model.

Return type:

dict

get_quality_time_step() → int[source]

Gets the quality time step – i.e. time step in the simple quality simulation.

Returns:

Quality time step in seconds.

Return type:

int

get_reporting_time_step() → int[source]

Gets the reporting time steps – i.e. time steps at which sensor readings are provided.

Is always a multiple of the hydraulic time step.

Returns:

Reporting time steps in seconds.

Return type:

int

get_scenario_config(include_network_topology: bool = True) → ScenarioConfig[source]

Gets the configuration of this scenario – i.e. all information & elements that completely describe this scenario.

Parameters:

include_network_topology (bool, optional) –

If True, the full specification of the network topology (incl. demand patterns) will be included in the scenario configuration.

The default is True.

Returns:

Complete scenario specification.

Return type:

ScenarioConfig

get_simulation_duration() → int[source]

Gets the simulation duration – i.e. time length to be simulated.

Returns:

Simulation duration in seconds.

Return type:

int

get_topology(include_demand_patterns: bool = False) → NetworkTopology[source]

Gets the topology (incl. information such as elevations, pipe diameters, etc.) of this WDN.

Parameters:

include_demand_patterns (bool, optional) –

If True, demand patterns will be included – be aware that this will increase the object’s memory footprint.

The default is False.

Returns:

Topology of this WDN as a graph.

Return type:

NetworkTopology

get_units_category() → int[source]

Gets the category of units – i.e. US Customary or SI Metric units.

Will be one of the following constants:

  • UNITS_USCUSTOM = 0 (US Customary)

  • UNITS_SIMETRIC = 1 (SI Metric)

Returns:

Units category.

Return type:

int

property leakages: list[Leakage]

Gets all leakages.

Returns:

All leakages.

Return type:

list[Leakage]

property model_uncertainty: ModelUncertainty

Gets the model uncertainty specification.

Returns:

Model uncertainty.

Return type:

ModelUncertainty

Places bulk species concentration sensors at every link/pipe in the network for every bulk species.

Parameters:

bulk_species (list[str], optional) –

List of bulk species IDs which we want to monitor at every link/pipe. If None, every bulk species will be monitored at every link/pipe.

The default is None.

place_bulk_species_node_sensors_everywhere(bulk_species: list[str] | None = None) → None[source]

Places bulk species concentration sensors at every node in the network for every bulk species.

Parameters:

bulk_species (list[str], optional) –

List of bulk species IDs which we want to monitor at every node. If None, every bulk species will be monitored at every node.

The default is None.

place_demand_sensors_everywhere() → None[source]

Places a demand sensor at every node in the network.

place_flow_sensors_everywhere() → None[source]

Places a flow sensors at every link/pipe in the network.

Places a water quality sensor at every link/pipe in the network.

place_node_quality_sensors_everywhere() → None[source]

Places a water quality sensor at every node in the network.

place_pressure_sensors_everywhere(junctions_only: bool = False) → None[source]

Places a pressure sensor at every node in the network.

Parameters:

junctions_only (bool, optional) –

If True, pressure sensors are only placed at junctions but not at tanks and reservoirs.

The default is False.

place_pump_efficiency_sensors_everywhere() → None[source]

Places a pump efficiency sensor at every pump in the network.

place_pump_energyconsumption_sensors_everywhere() → None[source]

Places a pump energy consumption sensor at every pump in the network.

place_pump_sensors_everywhere() → None[source]

Palces pump sensors at every pump in the network – i.e. retrieving the state, efficiency, and energy consumption of all pumps in the network.

place_pump_state_sensors_everywhere() → None[source]

Places a pump state sensor at every pump in the network.

place_sensors_everywhere() → None[source]

Places sensors everywhere – i.e. every possible quantity is monitored at every position in the network.

place_surface_species_sensors_everywhere(surface_species_id: list[str] | None = None) → None[source]

Places surface species concentration sensors at every link/pipe in the network for every surface species.

Parameters:

surface_species_id (list[str], optional) –

List of surface species IDs which we want to monitor at every link/pipe. If None, every surface species will be monitored at every link/pipe.

The default is None.

place_tank_sensors_everywhere() → None[source]

Places a water tank volume sensor at every tank in the network.

place_valve_sensors_everywhere() → None[source]

Places a valve state sensor at every valve in the network.

plot_topology(export_to_file: str | None = None) → None[source]

Plots the topology of the water distribution network.

Parameters:

export_to_file (str, optional) –

Path to the file where the visualization will be stored. If None, visualization will be just shown but NOT be stored anywhere.

The default is None.

randomize_demands() → None[source]

Randomizes all demand patterns.

remove_all_complex_controls() → None[source]

Removes all complex EPANET controls from the scenario.

remove_all_simple_controls() → None[source]

Removes all simple EPANET controls from the scenario.

remove_complex_control(control: ComplexControlModule) → None[source]

Removes a given complex (IF-THEN-ELSE) EPANET control rule from the scenario.

Parameters:

control (ComplexControlModule) – Complex EPANET control module to be removed.

remove_simple_control(control: SimpleControlModule) → None[source]

Removes a given simple EPANET control rule from the scenario.

Parameters:

control (SimpleControlModule) – Simple EPANET control module to be removed.

run_advanced_quality_simulation(hyd_file_in: str, verbose: bool = False, frozen_sensor_config: bool = False, use_quality_time_step_as_reporting_time_step: bool = False, reapply_uncertainties: bool = False, float_type: type = numpy.float32) → ScadaData[source]

Runs an advanced quality analysis using EPANET-MSX.

Parameters:

  • hyd_file_in (str) – Path to an EPANET .hyd file for storing the simulated hydraulics – the quality analysis is computed using those hydraulics.

  • verbose (bool, optional) –

    If True, method will be verbose (e.g. showing a progress bar).

    The default is False.

  • frozen_sensor_config (bool, optional) –

    If True, the sensor config can not be changed and only the required sensor nodes/links will be stored – this usually leads to a significant reduction in memory consumption.

    The default is False.

  • use_quality_time_step_as_reporting_time_step (bool, optional) –

    If True, the water quality time step will be used as the reporting time step.

    As a consequence, the simualtion results can not be merged with the hydraulic simulation.

    The default is False.

  • reapply_uncertainties (bool, optional) –

    If True, the uncertainties are re-applied on the original properties.

    The default is False.

  • float_type (type, optional) –

    Floating point type (precision).

    The default is 32bit – i.e., numpy.float32

Returns:

Quality simulation results as SCADA data.

Return type:

ScadaData

run_advanced_quality_simulation_as_generator(hyd_file_in: str, verbose: bool = False, support_abort: bool = False, return_as_dict: bool = False, frozen_sensor_config: bool = False, use_quality_time_step_as_reporting_time_step: bool = False, reapply_uncertainties: bool = False, float_type: type = numpy.float32) → Generator[tuple[ScadaData, bool] | tuple[dict, bool], bool, None][source]

Runs an advanced quality analysis using EPANET-MSX.

Parameters:

  • support_abort (bool, optional)

  • hyd_file_in (str) – Path to an EPANET .hyd file for storing the simulated hydraulics – the quality analysis is computed using those hydraulics.

  • verbose (bool) – If True, method will be verbose (e.g. showing a progress bar).

  • return_as_dict (bool, optional) –

    If True, simulation results/states are returned as a dictionary instead of a ScadaData instance.

    The default is False.

  • frozen_sensor_config (bool, optional) –

    If True, the sensor config can not be changed and only the required sensor nodes/links will be stored – this usually leads to a significant reduction in memory consumption.

    The default is False.

  • use_quality_time_step_as_reporting_time_step (bool, optional) –

    If True, the water quality time step will be used as the reporting time step.

    As a consequence, the simualtion results can not be merged with the hydraulic simulation.

    The default is False.

  • reapply_uncertainties (bool, optional) –

    If True, the uncertainties are re-applied on the original properties.

    The default is False.

  • float_type (type, optional) –

    Floating point type (precision).

    The default is 32bit – i.e., numpy.float32

Returns:

Generator containing the current EPANET-MSX simulation results as SCADA data (i.e. species concentrations) and a boolean indicating whether the simulation terminated or not.

Return type:

ScadaData

run_basic_quality_simulation(hyd_file_in: str, verbose: bool = False, frozen_sensor_config: bool = False, use_quality_time_step_as_reporting_time_step: bool = False, float_type: type = numpy.float32) → ScadaData[source]

Runs a basic quality analysis using EPANET.

Parameters:

  • hyd_file_in (str) – Path to an EPANET .hyd file for storing the simulated hydraulics – the quality analysis is computed using those hydraulics.

  • verbose (bool, optional) –

    If True, method will be verbose (e.g. showing a progress bar).

    The default is False.

  • frozen_sensor_config (bool, optional) –

    If True, the sensor config can not be changed and only the required sensor nodes/links will be stored – this usually leads to a significant reduction in memory consumption.

    The default is False.

  • use_quality_time_step_as_reporting_time_step (bool, optional) –

    If True, the water quality time step will be used as the reporting time step.

    As a consequence, the simualtion results can not be merged with the hydraulic simulation.

    The default is False.

  • float_type (type, optional) –

    Floating point type (precision).

    The default is 32bit – i.e., numpy.float32

Returns:

Quality simulation results as SCADA data.

Return type:

ScadaData

run_basic_quality_simulation_as_generator(hyd_file_in: str, verbose: bool = False, support_abort: bool = False, return_as_dict: bool = False, frozen_sensor_config: bool = False, use_quality_time_step_as_reporting_time_step: bool = False, float_type: type = numpy.float32) → Generator[tuple[ScadaData, bool] | tuple[dict, bool], bool, None][source]

Runs a basic quality analysis using EPANET.

Parameters:

  • support_abort (bool, optional)

  • hyd_file_in (str) – Path to an EPANET .hyd file for storing the simulated hydraulics – the quality analysis is computed using those hydraulics.

  • verbose (bool, optional) –

    If True, method will be verbose (e.g. showing a progress bar).

    The default is False.

  • return_as_dict (bool, optional) –

    If True, simulation results/states are returned as a dictionary instead of a ScadaData instance.

    The default is False.

  • frozen_sensor_config (bool, optional) –

    If True, the sensor config can not be changed and only the required sensor nodes/links will be stored – this usually leads to a significant reduction in memory consumption.

    The default is False.

  • use_quality_time_step_as_reporting_time_step (bool, optional) –

    If True, the water quality time step will be used as the reporting time step.

    As a consequence, the simualtion results can not be merged with the hydraulic simulation.

    The default is False.

  • float_type (type, optional) –

    Floating point type (precision).

    The default is 32bit – i.e., numpy.float32

Returns:

Generator with the current simulation results/states as SCADA data and a boolean indicating whether the simulation terminated or not.

Return type:

ScadaData

run_hydraulic_simulation(hyd_export: str | None = None, verbose: bool = False, frozen_sensor_config: bool = False, reapply_uncertainties: bool = False, float_type: type = numpy.float32) → ScadaData[source]

Runs the hydraulic simulation of this scenario (incl. basic quality if set).

Note that this function does not call EPANET-MSX even if an .msx file was provided.

Parameters:

  • hyd_export (str, optional) –

    Path to an EPANET .hyd file for storing the simulated hydraulics – these hydraulics can be used later for an advanced quality analysis using EPANET-MSX.

    If None, the simulated hydraulics will NOT be exported to an EPANET .hyd file.

    The default is None.

  • verbose (bool, optional) –

    If True, method will be verbose (e.g. showing a progress bar).

    The default is False.

  • frozen_sensor_config (bool, optional) –

    If True, the sensor config can not be changed and only the required sensor nodes/links will be stored – this usually leads to a significant reduction in memory consumption.

    The default is False.

  • reapply_uncertainties (bool, optional) –

    If True, the uncertainties are re-applied on the original properties.

    The default is False.

  • float_type (type, optional) –

    Floating point type (precision).

    The default is 32bit – i.e., numpy.float32

Returns:

Simulation results as SCADA data (i.e. sensor readings).

Return type:

ScadaData

run_hydraulic_simulation_as_generator(hyd_export: str | None = None, verbose: bool = False, support_abort: bool = False, return_as_dict: bool = False, frozen_sensor_config: bool = False, reapply_uncertainties: bool = False, float_type: type = numpy.float32) → Generator[tuple[ScadaData, bool] | tuple[dict, bool], bool, None][source]

Runs the hydraulic simulation of this scenario (incl. basic quality if set) and provides the results as a generator.

Note that this function does not run EPANET-MSX, even if an .msx file was provided.

Parameters:

  • hyd_export (str, optional) –

    Path to an EPANET .hyd file for storing the simulated hydraulics – these hydraulics can be used later for an advanced quality analysis using EPANET-MSX.

    If None, the simulated hydraulics will NOT be exported to an EPANET .hyd file.

    The default is None.

  • verbose (bool, optional) –

    If True, method will be verbose (e.g. showing a progress bar).

    The default is False.

  • support_abort (bool, optional) –

    If True, the simulation can be aborted after every time step – i.e. the generator takes a boolean as an input (send) to indicate whether the simulation is to be aborted or not.

    The default is False.

  • return_as_dict (bool, optional) –

    If True, simulation results/states are returned as a dictionary instead of a ScadaData instance.

    The default is False.

  • frozen_sensor_config (bool, optional) –

    If True, the sensor config can not be changed and only the required sensor nodes/links will be stored – this usually leads to a significant reduction in memory consumption.

    The default is False.

  • reapply_uncertainties (bool, optional) –

    If True, the uncertainties are re-applied on the original properties.

    The default is False.

  • float_type (type, optional) –

    Floating point type (precision).

    The default is 32bit – i.e., numpy.float32

Returns:

Generator with the current simulation results/states as SCADA data (i.e. sensor readings) and a boolean indicating whether the simulation terminated or not.

Return type:

ScadaData

run_simulation(hyd_export: str | None = None, verbose: bool = False, frozen_sensor_config: bool = False, reapply_uncertainties: bool = False, float_type: type = numpy.float32) → ScadaData[source]

Runs the simulation of this scenario.

Parameters:

  • hyd_export (str, optional) –

    Path to an EPANET .hyd file for storing the simulated hydraulics – these hydraulics can be used later for an advanced quality analysis using EPANET-MSX.

    If None, the simulated hydraulics will NOT be exported to an EPANET .hyd file.

    The default is None.

  • verbose (bool, optional) –

    If True, method will be verbose (e.g. showing a progress bar).

    The default is False.

  • frozen_sensor_config (bool, optional) –

    If True, the sensor config can not be changed and only the required sensor nodes/links will be stored – this usually leads to a significant reduction in memory consumption.

    The default is False.

  • reapply_uncertainties (bool, optional) –

    If True, the uncertainties are re-applied on the original properties.

    The default is False.

  • float_type (type, optional) –

    Floating point type (precision).

    The default is 32bit – i.e., numpy.float32

Returns:

Simulation results as SCADA data (i.e. sensor readings).

Return type:

ScadaData

save_to_epanet_file(inp_file_path: str, msx_file_path: str | None = None, export_sensor_config: bool = True, undo_system_events: bool = True) → None[source]

Exports this scenario to EPANET files – i.e. an .inp file and (optionally) a .msx file if EPANET-MSX was loaded.

Parameters:

  • inp_file_path (str) –

    Path to the .inp file where this scenario will be stored.

    If ‘inp_file_path’ is None, ‘msx_file_path’ must not be None!

  • msx_file_path (str, optional) –

    Path to the .msx file where this MSX component of this scneario will be stored.

    Note that this is only applicable if EPANET-MSX was loaded.

    The default is None.

  • export_sensor_config (bool, optional) –

    If True, the current sensor placement is exported as well.

    The default is True.

property sensor_config: SensorConfig

Gets the sensor configuration.

Returns:

Sensor configuration.

Return type:

SensorConfig

property sensor_faults: list[SensorFault]

Gets all sensor faults.

Returns:

All sensor faults.

Return type:

list[SensorFault]

property sensor_noise: SensorNoise

Gets the sensor noise/uncertainty.

Returns:

Sensor noise.

Return type:

SensorNoise

property sensor_reading_attacks: list[SensorReadingAttack]

Gets all sensor reading attacks.

Returns:

All sensor reading attacks.

Return type:

list[SensorReadingAttack]

property sensor_reading_events: list[SensorReadingEvent]

Gets all sensor reading events (e.g. sensor faults, etc.).

Returns:

All sensor reading events.

Return type:

list[SensorReadingEvent]

Sets the bulk species link/pipe sensors – i.e. measuring bulk species concentrations at links/pipes in the network.

Parameters:

sensor_info (dict) – Bulk species sensors – keys: bulk species IDs, values: node IDs.

set_bulk_species_node_initial_concentrations(inital_conc: dict[str, list[tuple[str, float]]]) → None[source]

Species the initial bulk species concentration at nodes.

Only for EPANET-MSX scenarios.

Parameters:

inital_conc (dict[str, list[tuple[str, float]]]) – Initial concentration of species (key) at nodes – i.e. value: list of node ID and initial concentration.

set_bulk_species_node_sensors(sensor_info: dict) → None[source]

Sets the bulk species node sensors – i.e. measuring bulk species concentrations at nodes in the network.

Parameters:

sensor_info (dict) – Bulk species sensors – keys: bulk species IDs, values: node IDs.

set_demand_sensors(sensor_locations: list[str]) → None[source]

Sets the demand sensors – i.e. measuring demands at some nodes in the network.

Parameters:

sensor_locations (list[str]) – Locations (IDs) of sensors.

set_flow_sensors(sensor_locations: list[str]) → None[source]

Sets the flow sensors – i.e. measuring flows at some links/pipes in the network.

Parameters:

sensor_locations (list[str]) – Locations (IDs) of sensors.

set_general_parameters(demand_model: dict | None = None, simulation_duration: int | None = None, hydraulic_time_step: int | None = None, pattern_time_step: int | None = None, pattern_time_start: int | None = None, quality_time_step: int | None = None, advanced_quality_time_step: int | None = None, reporting_time_step: int | None = None, reporting_time_start: int | None = None, flow_units_id: int | None = None, pressure_units_id: int | None = None, quality_model: dict | None = None) → None[source]

Sets some general parameters.

Note that all these parameters can be stated in the .inp file as well.

You only have to specify the parameters that are to be changed – all others can be left as None and will not be changed.

Parameters:

  • demand_model (dict, optional) –

    Specifies the demand model (e.g. pressure-driven or demand-driven) – the dictionary must contain the “type”, the minimal pressure (“pressure_min”), the required pressure (“pressure_required”), and the pressure exponent (“pressure_exponent”).

    The default is None.

  • simulation_duration (int, optional) –

    Number of seconds to be simulated.

    The default is None.

  • hydraulic_time_step (int, optional) –

    Hydraulic time step – i.e. the interval at which hydraulics are computed.

    The default is None.

  • pattern_time_step (int, optional) –

    Pattern time step – i.e. the interval at which patterns are applied.

    Must be a multiple of hydraulic_time_step.

    The default is None.

  • pattern_time_start (int, optional) –

    Pattern time start – i.e. the offset with which patterns are applied.

    The default is None.

  • quality_time_step (int, optional) –

    Quality time step – i.e. the interval at which qualities are computed. Should be much smaller than the hydraulic time step!

    The default is None.

  • advanced_quality_time_step (ìnt, optional) –

    Time step in the advanced quality simulation – i.e. EPANET-MSX simulation. This number specifies the interval at which all species concentrations are. Should be much smaller than the hydraulic time step!

    The default is None.

  • reporting_time_step (int, optional) –

    Report time step – i.e. the interval at which hydraulics and quality states are reported.

    Must be a multiple of hydraulic_time_step.

    If None, it will be set equal to hydraulic_time_step

    The default is None.

  • reporting_time_start (int, optional) –

    Start time (in seconds) at which reporting of hydraulic and quality states starts.

    The default is None.

  • flow_units_id (int, optional) –

    Specifies the flow units – i.e. all flows will be reported in these units. If None, the units from the .inp file will be used.

    Must be one of the following EPANET constants:

    • EN_CFS = 0 (cubic foot/sec)

    • EN_GPM = 1 (gal/min)

    • EN_MGD = 2 (Million gal/day)

    • EN_IMGD = 3 (Imperial MGD)

    • EN_AFD = 4 (ac-foot/day)

    • EN_LPS = 5 (liter/sec)

    • EN_LPM = 6 (liter/min)

    • EN_MLD = 7 (Megaliter/day)

    • EN_CMH = 8 (cubic meter/hr)

    • EN_CMD = 9 (cubic meter/day)

    • EN_CMS = 10 (cubic meters per second)

    The default is None.

  • pressure_units_id (int, optional) –

    Specifies the pressure units – i.e. all pressures will be reported in these units. If None, the units from the .inp file will be used.

    Must be one of the following EPANET constants:

    • EN_PSI = 0 (Pounds per square inch)

    • EN_KPA = 1 (Kilopascals)

    • EN_METERS = 2 (Meters)

    • EN_BAR = 3 (Bar)

    • EN_FEET = 4 (Feet)

    The default is None.

  • quality_model (dict, optional) –

    Specifies the quality model – the dictionary must contain, “type”, “chemical_name”, “chemical_units”, and “trace_node_id”, of the requested quality model.

    The default is None.

Sets the initial status (open or closed) of a given link.

Parameters:

  • link_id (str) – ID of the link.

  • status (int) –

    Initial status of the link. Must be one of the following EPANET constants:

    • EN_CLOSED = 0

    • EN_OPEN = 1

set_initial_node_quality(node_id: str, initial_quality: float) → None[source]

Specifies the initial quality at a given node. Quality represents concentration for chemicals, hours for water age, or percent for source tracing.

Parameters:

  • node_id (str) – ID of the node.

  • initial_quality (float) – Initial node quality.

set_initial_pump_speed(pump_id: str, speed: float) → None[source]

Sets the initial pump speed of a given pump.

Parameters:

  • pump_id (str) – ID of the pump.

  • speed (float) – Initial speed of the pump.

set_initial_tank_level(tank_id, level: float) → None[source]

Sets the initial water level of a given tank.

Parameters:

  • tank_id (str) – ID of the tank.

  • level (float) – Initial water level in the tank.

Sets the link quality sensors – i.e. measuring the water quality (e.g. age, chlorine concentration, etc.) at some links/pipes in the network.

Parameters:

sensor_locations (list[str]) – Locations (IDs) of sensors.

set_model_uncertainty(model_uncertainty: ModelUncertainty) → None[source]

Specifies the model uncertainties.

Parameters:

model_uncertainty (ModelUncertainty) – Model uncertainty specifications.

set_node_demand_pattern(node_id: str, base_demand: float, demand_pattern_id: str, demand_pattern: numpy.ndarray | None = None) → None[source]

Sets the demand pattern (incl. base demand) at a given node.

Parameters:

  • node_id (str) – ID of the node for which the demand pattern is set.

  • base_demand (float) – Base demand.

  • demand_pattern_id (str) – ID of the (new) demand pattern. Existing demand pattern will be overriden if it already exisits.

  • demand_pattern –

    Demand pattern over time. Final demand over time = base_demand * demand_pattern If None, the pattern demand_pattern_id is assumed to already exist.

    The default is None.

set_node_quality_sensors(sensor_locations: list[str]) → None[source]

Sets the node quality sensors – i.e. measuring the water quality (e.g. age, chlorine concentration, etc.) at some nodes in the network.

Parameters:

sensor_locations (list[str]) – Locations (IDs) of sensors.

set_pressure_sensors(sensor_locations: list[str]) → None[source]

Sets the pressure sensors – i.e. measuring pressure at some nodes in the network.

Parameters:

sensor_locations (list[str]) – Locations (IDs) of sensors.

set_pump_efficiency_sensors(sensor_locations: list[str]) → None[source]

Sets the pump efficiency sensors – i.e. retrieving the efficiency of some pumps in the network.

Parameters:

sensor_locations (list[str]) – Locations (IDs) of sensors.

set_pump_energy_price(pump_id, price: float) → None[source]

Sets the energy price of a given pump.

Parameters:

  • pump_id (str) – ID of the pump.

  • price (float) – Energy price.

set_pump_energy_price_pattern(pump_id: str, pattern: numpy.ndarray, pattern_id: str | None = None) → None[source]

Specifies/sets the energy price pattern of a given pump.

Overwrites any existing (energy price) patterns of the given pump.

Parameters:

  • pump_id (str) – ID of the pump.

  • pattern – Pattern of multipliers.

  • pattern_id (str, optional) –

    ID of the pattern. If not specified, ‘energy_price_{pump_id}’ will be used as the pattern ID.

    The default is None.

set_pump_energyconsumption_sensors(sensor_locations: list[str]) → None[source]

Sets the pump energy consumption sensors – i.e. retrieving the energy consumption of some pumps in the network.

Parameters:

sensor_locations (list[str]) – Locations (IDs) of sensors.

set_pump_sensors(sensor_locations: list[str]) → None[source]

Sets the pump sensors – i.e. retrieving the state, efficiency, and energy consumption of some pumps in the network.

Parameters:

sensor_locations (list[str]) – Locations (IDs) of sensors.

set_pump_state_sensors(sensor_locations: list[str]) → None[source]

Sets the pump state sensors – i.e. retrieving the state of some pumps in the network.

Parameters:

sensor_locations (list[str]) – Locations (IDs) of sensors.

set_quality_parameters(initial_quality: dict[str, float] | None = None, order_wall: int | None = None, order_tank: int | None = None, order_bulk: int | None = None, global_wall_reaction_coefficient: float | None = None, global_bulk_reaction_coefficient: float | None = None, local_wall_reaction_coefficient: dict[str, float] | None = None, local_bulk_reaction_coefficient: dict[str, float] | None = None, local_tank_reaction_coefficient: dict[str, float] | None = None, limiting_potential: float | None = None) → None[source]

Specifies some parameters of the EPANET quality analysis. Note that those parameters are only relevant for EPANET but not for EPANET-MSX.

Parameters:

  • initial_quality (dict[str, float], optional) –

    Specifies the initial quality (value in the dictionary) at nodes (key in the dictionary). Quality represents concentration for chemicals, hours for water age, or percent for source tracing.

    The default is None.

  • order_wall (int, optional) –

    Specifies the order of reactions occurring in the bulk fluid at pipe walls. Value for wall reactions must be either 0 or 1. If not specified, the default reaction order is 1.0.

    The default is None.

  • order_bulk (int, optional) –

    Specifies the order of reactions occurring in the bulk fluid in tanks. Value must be either 0 or 1. If not specified, the default reaction order is 1.0.

    The default is None.

  • global_wall_reaction_coefficient (float, optional) –

    Specifies the global value for all pipe wall reaction coefficients (pipes and tanks). If not specified, the default value is zero.

    The default is None.

  • global_bulk_reaction_coefficient (float, optional) –

    Specifies the global value for all bulk reaction coefficients (pipes and tanks). If not specified, the default value is zero.

    The default is None.

  • local_wall_reaction_coefficient (dict[str, float], optional) –

    Overrides the global reaction coefficients for specific pipes (key in dictionary).

    The default is None.

  • local_bulk_reaction_coefficient (dict[str, float], optional) –

    Overrides the global reaction coefficients for specific pipes (key in dictionary).

    The default is None.

  • local_tank_reaction_coefficient (dict[str, float], optional) –

    Overrides the global reaction coefficients for specific tanks (key in dictionary).

    The default is None.

  • limiting_potential (float, optional) –

    Specifies that reaction rates are proportional to the difference between the current concentration and some (specified) limiting potential value.

    The default is None.

set_sensor_noise(sensor_noise: SensorNoise) → None[source]

Specifies the sensor noise – i.e. uncertainties of sensor readings.

Parameters:

sensor_noise (SensorNoise) – Sensor noise specification.

set_sensors(sensor_type: int, sensor_locations: list[str] | dict) → None[source]

Specifies all sensors of a given type (e.g. pressure sensor, flow sensor, etc.)

Parameters:

  • sensor_type (int) –

    Sensor type. Must be one of the following:

    • SENSOR_TYPE_NODE_PRESSURE = 1

    • SENSOR_TYPE_NODE_QUALITY = 2

    • SENSOR_TYPE_NODE_DEMAND = 3

    • SENSOR_TYPE_LINK_FLOW = 4

    • SENSOR_TYPE_LINK_QUALITY = 5

    • SENSOR_TYPE_VALVE_STATE = 6

    • SENSOR_TYPE_PUMP_STATE = 7

    • SENSOR_TYPE_TANK_VOLUME = 8

    • SENSOR_TYPE_BULK_SPECIES = 9

    • SENSOR_TYPE_SURFACE_SPECIES = 10

    • SENSOR_TYPE_PUMP_EFFICIENCY = 12

    • SENSOR_TYPE_PUMP_ENERGYCONSUMPTION = 13

  • sensor_locations (list[str] or dict) – Locations (IDs) of sensors either as a list or as a dict in the case of bulk and surface species.

Species the initial (bulk or surface) species concentration at links.

Only for EPANET-MSX scenarios.

Parameters:

inital_conc (dict[str, list[tuple[str, float]]]) – Initial concentration of species (key) at links – i.e. value: list of link ID and initial concentration.

set_surface_species_sensors(sensor_info: dict) → None[source]

Sets the surface species sensors – i.e. measuring surface species concentrations at nodes in the network.

Parameters:

sensor_info (dict) – Surface species sensors – keys: surface species IDs, values: link/pipe IDs.

set_tank_sensors(sensor_locations: list[str]) → None[source]

Sets the tank volume sensors – i.e. measuring water volumes in some tanks in the network.

Parameters:

sensor_locations (list[str]) – Locations (IDs) of sensors.

set_valve_sensors(sensor_locations: list[str]) → None[source]

Sets the valve state sensors – i.e. retrieving the state of some valves in the network.

Parameters:

sensor_locations (list[str]) – Locations (IDs) of sensors.

property simple_controls: list[SimpleControlModule]

Gets all simple EPANET control rules.

Returns:

All simple EPANET control rules.

Return type:

list[SimpleControlModule]

property system_events: list[SystemEvent]

Gets all system events (e.g. leakages, etc.).

Returns:

All system events.

Return type:

list[SystemEvent]

epyt_flow.simulation.parallel_simulation

Module provides functions for simulating several scenarios in parallel.

class epyt_flow.simulation.parallel_simulation.ParallelScenarioSimulation[source]

Bases: object

Class providing functions to run scenario simulations in parallel.

static run(scenarios: list[~epyt_flow.simulation.scenario_config.ScenarioConfig], n_jobs: int = -1, max_working_memory_consumption: int | None = None, callback: ~typing.Callable[[~epyt_flow.simulation.scada.scada_data.ScadaData, ~epyt_flow.simulation.scenario_config.ScenarioConfig, int], ~typing.Any] = <function callback_save_to_file.<locals>.callback>) → Any[source]

Simulates multiple scenarios in parallel.

Parameters:

  • scenarios (list[ScenarioConfig]) – List of scenarios to be simulated.

  • n_jobs (int, optional) –

    Number of CPUs that can be used by the simulations – usually, this translates to the number of scenarios that are simulated in parallel.

    If -1, all CPUs are used.

    The default is -1

  • max_working_memory_consumption (int, optional) –

    Maximum amount of working memory in MB that can be used by the simulations. Note that this might limit the number of scenarios that can be simulated in parallel.

    The default is None.

  • callback (Callable[[ScadaData, ScenarioConfig, int], None], optional) –

    Callback that is called after the simulation of a scenario finished.

    The callback gets the simulation results as a ScadaData instance, the scenario configuration as a ScenarioConfig instance, and the index of the scenario in ‘scenarios’ as arguments.

    The default is callback_save_to_file().

epyt_flow.simulation.parallel_simulation.callback_save_to_file(folder_out: str = '') → Callable[[ScadaData, ScenarioConfig, int], None][source]

Creates a callback for storing the simulation results in a .epytflow_scada_data file. The returned callback can be directly passed to run().

Parameters:

folder_out (str, optional) –

Path to the folder where the simulation results will be stored.

The default is the current working directory.

Returns:

Callback storing the simulation results.

Return type:

Callable[[ScadaData, ScenarioConfig, int], None]

epyt_flow.simulation.sensor_config

Module provides a class for implementing sensor configurations.

class epyt_flow.simulation.sensor_config.SensorConfig(nodes: list[str], links: list[str], valves: list[str], pumps: list[str], tanks: list[str], bulk_species: list[str], surface_species: list[str], flow_unit: int, pressure_unit: int, sensor_ordering: list[int] | None = None, pressure_sensors: list[str] = [], flow_sensors: list[str] = [], demand_sensors: list[str] = [], quality_node_sensors: list[str] = [], quality_link_sensors: list[str] = [], valve_state_sensors: list[str] = [], pump_state_sensors: list[str] = [], pump_efficiency_sensors: list[str] = [], pump_energyconsumption_sensors: list[str] = [], tank_volume_sensors: list[str] = [], bulk_species_node_sensors: dict = {}, bulk_species_link_sensors: dict = {}, surface_species_sensors: dict = {}, node_id_to_idx: dict | None = None, link_id_to_idx: dict | None = None, valve_id_to_idx: dict | None = None, pump_id_to_idx: dict | None = None, tank_id_to_idx: dict | None = None, bulkspecies_id_to_idx: dict | None = None, surfacespecies_id_to_idx: dict | None = None, quality_unit: int | None = None, bulk_species_mass_unit: list[int] = [], surface_species_mass_unit: list[int] = [], surface_species_area_unit: int | None = None, **kwds)[source]

Bases: JsonSerializable

Class for storing a sensor configuration.

Parameters:

  • nodes (list[str]) – List of all nodes (i.e. IDs) in the network.

  • links (list[str]) – List of all links/pipes (i.e. IDs) in the network.

  • valves (list[str]) – List of all valves (i.e. IDs) in the network.

  • pumps (list[str]) – List of all pumps (i.e. IDs) in the network.

  • tanks (list[str]) – List of all tanks (i.e. IDs) in the network.

  • species (list[str]) – List of all (EPANET-MSX) species (i.e. IDs) in the network

  • sensor_ordering (list[int], optional) –

    Ordering of sensor types in this list specifies the ordering of the sensor readings. The list must contain every sensor type, no matter if a sensor of that type is placed or not!

    If None, the following default will be used:

    • 1 -> pressure sensor

    • 2 -> node quality sensor

    • 3 -> demand sensor

    • 4 -> flow sensor

    • 5 -> link quality sensor

    • 6 -> valve state sensor

    • 7 -> pump state sensor

    • 8 -> tank volume sensor

    • 9 -> node bulk species sensor

    • 10 -> link bulk species sensor

    • 11 -> surface species sensor

    • 12 -> pump efficiency sensor

    • 13 -> pump energy consumption sensor

    The default is None.

  • pressure_sensors (list[str], optional) –

    List of all nodes (i.e. IDs) at which a pressure sensor is placed.

    The default is an empty list.

  • flow_sensors (list[str], optional) –

    List of all links/pipes (i.e. IDs) at which a flow sensor is placed.

    The default is an empty list.

  • demand_sensors (list[str], optional) –

    List of all nodes (i.e. IDs) at which a demand sensor is placed.

    The default is an empty list.

  • quality_node_sensors (list[str], optional) –

    List of all nodes (i.e. IDs) at which a quality sensor is placed.

    The default is an empty list.

  • quality_link_sensors (list[str], optional) –

    List of all links/pipes (i.e. IDs) at which a flow sensor is placed.

    The default is an empty list.

  • valve_state_sensors (list[str], optional) –

    List of all valves (i.e. IDs) at which a valve state sensor is placed.

    The default is an empty list.

  • pump_state_sensors (list[str], optional) –

    List of all pumps (i.e. IDs) at which a pump state sensor is placed.

    The default is an empty list.

  • tank_volume_sensors (list[str], optional) –

    List of all tanks (i.e. IDs) at which a tank volume sensor is placed.

    The default is an empty list.

  • bulk_species_node_sensors (dict, optional) –

    Bulk species node sensors as a dictionary – i.e. bulk species ID are the keys, and the sensor locations (node IDs) are the values.

    The default is an empty list.

  • bulk_species_link_sensors (dict, optional) –

    Bulk species link/pipe sensors as a dictionary – i.e. bulk species ID are the keys, and the sensor locations (link/pipe IDs) are the values.

    The default is an empty list.

  • surface_species_sensors (dict, optional) –

    Surface species sensors as a dictionary – i.e. surface species ID are the keys, and the sensor locations (link/pipe IDs) are the values.

    The default is an empty list.

  • node_id_to_idx (dict, optional) –

    Mapping of a node ID to the EPANET index (i.e. position in the raw sensor reading data).

    If None is given, it is assumed that the nodes (in ‘nodes’) are sorted according to their EPANET index.

    The default is None.

  • link_id_to_idx (dict, optional) –

    Mapping of a link/pipe ID to the EPANET index (i.e. position in the raw sensor reading data).

    If None is given, it is assumed that the links/pipes (in ‘links’) are sorted according to their EPANET index..

    The default is None.

  • valve_id_to_idx (dict, optional) –

    Mapping of a valve ID to the EPANET index (i.e. position in the raw sensor reading data).

    If None is given, it is assumed that the valves (in ‘valves’) are sorted according to their EPANET index.

    The default is None.

  • pump_id_to_idx (dict, optional) –

    Mapping of a pump ID to the EPANET index (i.e. position in the raw sensor reading data).

    If None is given, it is assumed that the pumps (in ‘pumps’) are sorted according to their EPANET index.

    The default is None.

  • tank_id_to_idx (dict, optional) –

    Mapping of a tank ID to the EPANET index (i.e. position in the raw sensor reading data).

    If None is given, it is assumed that the tanks (in ‘tanks’) are sorted according to their EPANET index.

    The default is None.

  • bulkspecies_id_to_idx (dict, optional) –

    Mapping of a surface species ID to the EPANET index (i.e. position in the raw sensor reading data).

    If None is given, it is assumed that the surface species (in ‘surface_species’) are sorted according to their EPANET index.

    The default is None.

  • flow_unit (int) –

    Specifies the flow units and consequently many other hydraulic units (US CUSTOMARY or SI METRIC) as well, except the pressure units which must be specified separately.

    Must be one of the following EPANET constants:

    • EN_CFS = 0 (cubic foot/sec)

    • EN_GPM = 1 (gal/min)

    • EN_MGD = 2 (Million gal/day)

    • EN_IMGD = 3 (Imperial MGD)

    • EN_AFD = 4 (ac-foot/day)

    • EN_LPS = 5 (liter/sec)

    • EN_LPM = 6 (liter/min)

    • EN_MLD = 7 (Megaliter/day)

    • EN_CMH = 8 (cubic meter/hr)

    • EN_CMD = 9 (cubic meter/day)

    • EN_CMS = 10 (cubic meter/sec)

  • pressure_unit (int) –

    Specifies the pressure units.

    Must be one of the following EPANET constants:

    • EN_PSI = 0 (Pounds per square inch)

    • EN_KPA = 1 (Kilopascals)

    • EN_METERS = 2 (Meters)

    • EN_BAR = 3 (Bar)

    • EN_FEET = 4 (Feet)

  • quality_unit (str, optional) –

    Measurement unit (in a basic quality analysis) – only relevant if basic water quality is enabled.

    Must be one of the following constants:

    • MASS_UNIT_MG = 4 (mg/L)

    • MASS_UNIT_UG = 5 (ug/L)

    • TIME_UNIT_HRS = 8 (hrs)

  • bulk_species_mass_unit (list[int], optional) –

    Specifies the mass unit for each bulk species – only relevant if EPANET-MSX is used.

    Must be one of the following constants:

    • MASS_UNIT_MG = 4 (milligram)

    • MASS_UNIT_UG = 5 (microgram)

    • MASS_UNIT_MOL = 6 (mole)

    • MASS_UNIT_MMOL = 7 (millimole)

    Note that the assumed ordering is the same as given in ‘bulk_species’.

  • surface_species_mass_unit (list[int], optional) –

    Specifies the mass unit for each surface species – only relevant if EPANET-MSX is used.

    Must be one of the following constants:

    • MASS_UNIT_MG = 4 (milligram)

    • MASS_UNIT_UG = 5 (microgram)

    • MASS_UNIT_MOL = 6 (mole)

    • MASS_UNIT_MMOL = 7 (millimole)

    Note that the assumed ordering is the same as given in ‘surface_species’.

  • surface_species_area_unit (int, optional) –

    Species the area unit of all surface species – only relevant if EPANET-MSX is used. Must be one of the following constants:

    • AREA_UNIT_FT2 = 1 (square feet)

    • AREA_UNIT_M2 = 2 (square meters)

    • AREA_UNIT_CM2 = 3 (square centimeters)

property bulk_species: list[str]

Gets all bulk species IDs – i.e. species that live in the water.

Returns:

All species IDs.

Return type:

list[str]

Gets all bulk species link/pipe sensors as a dictionary – i.e. bulk species IDs as keys and link/pipe IDs as values.

Returns:

Bulk species sensors – keys: bulk species IDs, values: link/pipe IDs.

Return type:

dict

property bulk_species_mass_unit: list[int]

Gets the mass unit of each bulk species.

Will be one of the following constants:

  • MASS_UNIT_MG = 4 (milligram)

  • MASS_UNIT_UG = 5 (microgram)

  • MASS_UNIT_MOL = 6 (mole)

  • MASS_UNIT_MMOL = 7 (millimole)

Returns:

Mass unit ID.

Return type:

int

property bulk_species_node_sensors: dict

Gets all bulk species node sensors as a dictionary – i.e. bulk species IDs as keys and node IDs as values.

Returns:

Bulk species sensors – keys: bulk species IDs, values: node IDs.

Return type:

dict

property bulkspecies_id_to_idx: dict

Mapping of a bulk species ID to the EPANET index (i.e. position in the raw sensor reading data).

If None, it is assumed that the bulk species (in ‘bulk_species’) are sorted according to their EPANET index.

Returns:

Bulk species ID to index mapping.

Return type:

dict

compute_readings(pressures: numpy.ndarray, flows: numpy.ndarray, demands: numpy.ndarray, nodes_quality: numpy.ndarray, links_quality: numpy.ndarray, pumps_state: numpy.ndarray, pumps_efficiency: numpy.ndarray, pumps_energyconsumption: numpy.ndarray, valves_state: numpy.ndarray, tanks_volume: numpy.ndarray, bulk_species_node_concentrations: numpy.ndarray, bulk_species_link_concentrations: numpy.ndarray, surface_species_concentrations: numpy.ndarray) → numpy.ndarray[source]

Applies the sensor configuration to a set of raw simulation results – i.e. computes the sensor readings as an array.

Columns (i.e. sensor readings) are ordered as according to sensor_ordering. If not changed, the following default order applies:

  1. Pressures

  2. Flows

  3. Demands

  4. Nodes quality

  5. Links quality

  6. Valve state

  7. Pumps state

  8. Pumps efficiency

  9. Pumps energy consumption

  10. Tanks volume

  11. Surface species concentrations

  12. Bulk species nodes concentrations

  13. Bulk species links concentrations

Parameters:

  • pressures – Pressure values at all nodes.

  • flows – Flow values at all links/pipes.

  • demands – Demand values at all nodes.

  • nodes_quality – Quality values at all nodes.

  • links_quality – Quality values at all links/pipes.

  • pumps_state – States of all pumps.

  • pumps_efficiency – Efficiency of all pumps.

  • pumps_energyconsumption – Energy consumption of all pumps.

  • valves_state – States of all valves.

  • tanks_volume – Water volume in all tanks.

  • bulk_species_node_concentrations –

    Bulk species concentrations at all nodes.

    Expect a three-dimensional array: First dimension denotes time, second dimension corresponds to species ID, and third dimension contains the concentration.

  • bulk_species_link_concentrations –

    Bulk species concentrations at all links/pipes.

    Expect a three-dimensional array: First dimension denotes time, second dimension corresponds to species ID, and third dimension contains the concentration.

  • surface_species_concentrations –

    Surface species concentrations at all links/pipes.

    Expect a three-dimensional array: First dimension denotes time, second dimension corresponds to species ID, and third dimension contains the concentration.

Returns:

Sensor readings.

Return type:

numpy.ndarray

static create_empty_sensor_config(sensor_config)[source]

Creates an empty sensor configuration from a given sensor configuration – i.e. a clone of the given sensor configuration except that no sensors are set.

Parameters:

sensor_config (SensorConfig) – Sensor configuration used as a basis.

Returns:

Empty sensor configuration.

Return type:

epyt_flow.simulation.sensor_config.SensorConfig

property demand_sensors: list[str]

Gets all demand sensors (i.e. IDs of nodes at which a demand sensor is placed).

Returns:

All node IDs with a demand sensor.

Return type:

list[str]

static file_ext() → str

Returns the file extension of this class.

This function is automatically implemented by applying the serializable() decorator.

Returns:

File extension.

Return type:

str

property flow_sensors: list[str]

Gets all flow sensors (i.e. IDs of links at which a flow sensor is placed).

Returns:

All link IDs with a flow sensor.

Return type:

list[str]

property flow_unit: int

Returns the flow units. Note that this also specifies all other hydraulic units, except pressure.

Will be one of the following EPANET constants:

  • EN_CFS = 0 (cubic foot/sec)

  • EN_GPM = 1 (gal/min)

  • EN_MGD = 2 (Million gal/day)

  • EN_IMGD = 3 (Imperial MGD)

  • EN_AFD = 4 (ac-foot/day)

  • EN_LPS = 5 (liter/sec)

  • EN_LPM = 6 (liter/min)

  • EN_MLD = 7 (Megaliter/day)

  • EN_CMH = 8 (cubic meter/hr)

  • EN_CMD = 9 (cubic meter/day)

  • EN_CMD = 10 (cubic meter/sec)

Returns:

Flow unit ID.

Return type:

int

get_as_dict() → dict[source]

Gets the sensor configuration as a dictionary.

Returns:

Dictionary of set sensors – the keys are the sensor types.

Return type:

dict

get_attributes() → dict[source]

Gets all attributes to be serialized – these attributes are passed to the constructor when the object is deserialized.

Returns:

Dictionary of attributes – i.e. pairs of attribute name + value.

Return type:

dict

get_bulk_species_mass_unit_id(bulk_species_id: str) → int[source]

Returns the mass unit of a given bulk species.

Parameters:

bulk_species_id (str) – ID of the bulk species.

Returns:

ID of the mass unit.

Will be one of the following constant:

  • MASS_UNIT_MG = 4

  • MASS_UNIT_UG = 5

  • MASS_UNIT_MOL = 6

  • MASS_UNIT_MMOL = 7

Return type:

int

get_index_of_reading(pressure_sensor: str | None = None, flow_sensor: str | None = None, demand_sensor: str | None = None, node_quality_sensor: str | None = None, link_quality_sensor: str | None = None, valve_state_sensor: str | None = None, pump_state_sensor: str | None = None, pump_efficiency_sensor: str | None = None, pump_energyconsumption_sensor: str | None = None, tank_volume_sensor: str | None = None, bulk_species_node_sensor: tuple[str, str] | None = None, bulk_species_link_sensor: tuple[str, str] | None = None, surface_species_sensor: tuple[str, str] | None = None) → int[source]

Gets the index of a particular sensor in the final sensor readings array.

Note that only one sensor ID is converted to an index. In case of multiple sensor IDs, call this function for each sensor ID separately.

Note

This function only returns the correct results if the sensor configuraton is NOT frozen!

Parameters:

  • pressure_sensor (str) – ID of the pressure sensor.

  • flow_sensor (str) – ID of the flow sensor.

  • demand_sensor (str) – ID of the demand sensor.

  • node_quality_sensor (str) – ID of the quality sensor (at a node).

  • link_quality_sensor (str) – ID of the quality sensor (at a link/pipe).

  • valve_state_sensor (str) – ID of the state sensor (at a valve).

  • pump_state_sensor (str) – ID of the state sensor (at a pump).

  • pump_efficiency_sensor (str) – ID of the efficiency sensor (at a pump).

  • pump_energyconsumption_sensor (str) – ID of the energy consumption sensor (at a pump).

  • tank_volume_sensor (str) – ID of the water volume sensor (at a tank)

  • bulk_species_node_sensor (tuple[str, str]) – Tuple of bulk species ID and sensor node ID.

  • bulk_species_link_sensor (tuple[str, str]) – Tuple of bulk species ID and sensor link/pipe ID.

  • surface_species_sensor (tuple[str, str]) – Tuple of surface species ID and sensor link/pipe ID.

get_surface_species_mass_unit_id(surface_species_id: str) → int[source]

Returns the mass unit of a given surface species.

Parameters:

surface_species_id (str) – ID of the surface species.

Returns:

ID of the mass unit.

Will be one of the following constant:

  • MASS_UNIT_MG = 4

  • MASS_UNIT_UG = 5

  • MASS_UNIT_MOL = 6

  • MASS_UNIT_MMOL = 7

Return type:

int

is_empty() → bool[source]

Checks if the sensor configuration is empty – i.e. no sensors are placed.

Returns:

True if no sensors are placed, False otherwise.

Return type:

bool

property junctions: list[str]

Returns all junction IDs.

Returns:

All juncitons IDs.

Return type:

list[str]

Mapping of a link/pipe ID to the EPANET index (i.e. position in the raw sensor reading data).

If None is given, it is assumed that the links/pipes (in ‘links’) are sorted according to their EPANET index.

Returns:

Link/Pipe ID to index mapping.

Return type:

dict

Gets all link IDs.

Returns:

All link IDs.

Return type:

list[str]

map_bulkspecies_id_to_idx(bulk_species_id: str) → int[source]

Gets the index of a given bulk species ID.

Parameters:

bulk_species_id (str) – Bulk species ID.

Returns:

Index of the given bulk species.

Return type:

int

Gets the index of a given link ID.

Parameters:

link_id (str) – Link ID.

Returns:

Index of the given link.

Return type:

int

map_node_id_to_idx(node_id: str) → int[source]

Gets the index of a given node ID.

Parameters:

node_id (str) – Node ID.

Returns:

Index of the given node.

Return type:

int

map_pump_id_to_idx(pump_id: str) → int[source]

Gets the index of a given pump ID.

Parameters:

pump_id (str) – Pump ID.

Returns:

Index of the given pump.

Return type:

int

map_surfacespecies_id_to_idx(surface_species_id: str) → int[source]

Gets the index of a given surface species ID.

Parameters:

surface_species_id (str) – Surface species ID.

Returns:

Index of the given surface species.

Return type:

int

map_tank_id_to_idx(tank_id: str) → int[source]

Gets the index of a given tank ID.

Parameters:

tank_id (str) – Tank ID.

Returns:

Index of the given tank.

Return type:

int

map_valve_id_to_idx(valve_id: str) → int[source]

Gets the index of a given valve ID.

Parameters:

valve_id (str) – Valve ID.

Returns:

Index of the given valve.

Return type:

int

property node_id_to_idx: dict

Mapping of a surface node ID to the EPANET index (i.e. position in the raw sensor reading data).

If None, it is assumed that the nodes (in ‘nodes’) are sorted according to their EPANET index.

Returns:

Node ID to index mapping.

Return type:

dict

property nodes: list[str]

Gets all node IDs.

Returns:

All node IDs.

Return type:

list[str]

place_sensors_everywhere() → None[source]

Places sensors everywhere – i.e. every possible quantity is monitored at every position in the network.

property pressure_sensors: list[str]

Gets all pressure sensors (i.e. IDs of nodes at which a pressure sensor is placed).

Returns:

All node IDs with a pressure sensor.

Return type:

list[str]

property pressure_unit: int

Returns the pressure units.

Will be one of the following EPANET constants:

  • EN_PSI = 0 (Pounds per square inch)

  • EN_KPA = 1 (Kilopascals)

  • EN_METERS = 2 (Meters)

  • EN_BAR = 3 (Bar)

  • EN_FEET = 4 (Feet)

Returns:

Pressure unit ID.

Return type:

int

property pump_efficiency_sensors: list[str]

Gets all pump efficiency sensors (i.e. IDs of pumps at which the efficiency is monitored).

Returns:

All pump IDs with an efficiency sensor.

Return type:

list[str]

property pump_energyconsumption_sensors: list[str]

Gets all pump energy consumption sensors (i.e. IDs of pumps at which the energy consumption is monitored).

Returns:

All pump IDs with an energy consumption sensor.

Return type:

list[str]

property pump_id_to_idx: dict

Mapping of a pump ID to the EPANET index (i.e. position in the raw sensor reading data).

If None, it is assumed that the pumps (in ‘pumps’) are sorted according to their EPANET index.

Returns:

Pump ID to index mapping.

Return type:

dict

property pump_state_sensors: list[str]

Gets all pump state sensors (i.e. IDs of pumps at which a pump state sensor is placed).

Returns:

All link IDs with a pump state sensor.

Return type:

list[str]

property pumps: list[str]

Gets all pump IDs (subset of link IDs).

Returns:

All pump IDs.

Return type:

list[str]

Gets all link quality sensors (i.e. IDs of links at which a link quality sensor is placed).

Returns:

All link IDs with a link quality sensor.

Return type:

list[str]

property quality_node_sensors: list[str]

Gets all node quality sensors (i.e. IDs of nodes at which a node quality sensor is placed).

Returns:

All node IDs with a node quality sensor.

Return type:

list[str]

property quality_unit: int

Gets the measurement unit ID used in the basic quality analysis.

Will be one of the following constants:

  • MASS_UNIT_MG = 4 (milligram)

  • MASS_UNIT_UG = 5 (microgram)

  • TIME_UNIT_HRS = 6 (hours)

Returns:

Mass unit ID.

Return type:

int

property sensor_ordering: list[int]

Returns the order in which sensors are included in ScadaData objects i.e. if you call a ScadaData’s get_data() method, the resulting array will contain sensor readings in the order returned by this method.

Returns:

List of sensor types, specifying the ordering of sensor readings.

Constants have the following meaning:

  • 1 -> pressure sensor

  • 2 -> node quality sensor

  • 3 -> demand sensor

  • 4 -> flow sensor

  • 5 -> link quality sensor

  • 6 -> valve state sensor

  • 7 -> pump state sensor

  • 8 -> tank volume sensor

  • 9 -> node bulk species sensor

  • 10 -> link bulk species sensor

  • 11 -> surface species sensor

  • 12 -> pump efficiency sensor

  • 13 -> pump energy consumption sensor

Return type:

list[int]

property sensors_id_to_idx: dict

Gets a mapping of sensor IDs to indices in the final Numpy array returned by get_data().

Returns:

Mapping of sensor IDs to indices in the final Numpy array.

Return type:

dict

property surface_species: list[str]

Gets all surface species IDs – i.e. species that live links/pipes.

Returns:

All species IDs.

Return type:

list[str]

property surface_species_area_unit: int

Gets the surface species area unit.

Will be one of the following constants:

  • AREA_UNIT_FT2 = 1 (square feet)

  • AREA_UNIT_M2 = 2 (square meters)

  • AREA_UNIT_CM2 = 3 (square centimeters)

Returns:

Area unit ID.

Return type:

int

property surface_species_mass_unit: list[int]

Gets the mass unit of each surface species.

Will be one of the following constants:

  • MASS_UNIT_MG = 4 (milligram)

  • MASS_UNIT_UG = 5 (microgram)

  • MASS_UNIT_MOL = 6 (mole)

  • MASS_UNIT_MMOL = 7 (millimole)

Returns:

Mass unit ID.

Return type:

int

property surface_species_sensors: dict

Gets all surface species sensors as a dictionary – i.e. surface species IDs as keys and link/pipe IDs as values.

Returns:

Surface species sensors – keys: surface species IDs, values: link/pipe IDs.

Return type:

dict

property surfacespecies_id_to_idx: dict

Mapping of a surface species ID to the EPANET index (i.e. position in the raw sensor reading data).

If None, it is assumed that the surface species (in ‘surface_species’) are sorted according to their EPANET index.

Returns:

Surface species ID to index mapping.

Return type:

dict

property tank_id_to_idx: dict

Mapping of a tank ID to the EPANET index (i.e. position in the raw sensor reading data).

If None, it is assumed that the tanks (in ‘tanks’) are sorted according to their EPANET index.

Returns:

Tank ID to index mapping.

Return type:

dict

property tank_volume_sensors: list[str]

Gets all tank volume sensors (i.e. IDs of tanks at which a tank volume sensor is placed).

Returns:

All tank IDs with a tank volume sensor.

Return type:

list[str]

property tanks: list[str]

Gets all tank IDs (subset of node IDs).

Returns:

All tank IDs.

Return type:

list[str]

validate(epanet_api: epanet_plus.EPyT) → None[source]

Validates this sensor configuration – i.e. checks whether all nodes, etc. exist in the .inp file.

Parameters:

epanet_api (epanet_plus.EPyT) – EPANET and EPANET-MSX API.

property valve_id_to_idx: dict

Mapping of a valve ID to the EPANET index (i.e. position in the raw sensor reading data).

If None, it is assumed that the valves (in ‘valves’) are sorted according to their EPANET index.

Returns:

Valve ID to index mapping.

Return type:

dict

property valve_state_sensors: list[str]

Gets all valve state sensors (i.e. IDs of valves at which a valve state sensor is placed).

Returns:

All valve IDs with a valve state sensor.

Return type:

list[str]

property valves: list[str]

Gets all valve IDs (subset of link IDs).

Returns:

All valve IDs.

Return type:

list[str]

epyt_flow.simulation.sensor_config.valid_sensor_types() → str[source]

Returns a description of all valid sensor types.