epyt_flow.simulation.scada<a class="headerlink" href="#epyt-flow-simulation-scada" title="Link to this heading">

property bulk_species_node_concentration_raw: numpy.ndarray

Gets the raw bulk species concentrations at nodes.

Returns:

Raw species concentrations.

Return type:

change_sensor_config(sensor_config: SensorConfig) → None[source]

Changes the sensor configuration.

Parameters:: sensor_config (SensorConfig) – New sensor configuration.

change_sensor_faults(sensor_faults: list[SensorFault]) → None[source]

Changes the sensor faults – overrides all previous sensor faults!

sensor_faultslist[SensorFault]: List of new sensor faults.

change_sensor_noise(sensor_noise: SensorNoise) → None[source]

Changes the sensor noise/uncertainty.

Parameters:: sensor_noise (SensorNoise) – New sensor noise/uncertainty specification.

change_sensor_reading_attacks(sensor_reading_attacks: list[SensorReadingAttack]) → None[source]

Changes the sensor reading attacks – overrides all previous sensor reading attacks!

sensor_reading_attackslist[SensorReadingAttack]: List of new sensor reading attacks.

change_sensor_reading_events(sensor_reading_events: list[SensorReadingEvent]) → None[source]

Changes the sensor reading events – overrides all previous sensor reading events (incl. sensor faults)!

sensor_reading_eventslist[SensorReadingEvent]: List of new sensor reading events.

concatenate(other) → None[source]

Concatenates two ScadaData instances – i.e. add SCADA data from another given ScadaData instance to this one.

Note that the two ScadaData instances must be the same in all other attributs (e.g. sensor configuration, etc.).

Parameters:: other (ScadaData) – Other scada data to be concatenated to this data.

Changes the units of some measurement units.

Note

Beaware of potential rounding errors.

Parameters:

flow_unit (int, optional) –
New (flow) units of hydraulic measurements – note that the flow unit specifies all other hydraulic measurement units, except pressure.

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)
If None, units of dependent hydraulic measurement are not changed.

The default is None.
pressure_unit (int, optional) –
New pressure units of hydraulic measurementsO

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_unit (int, optional) –
New unit of quality measurements – i.e. chemical concentration. Only relevant if basic quality analysis was performed.

Must be one of the following constants:
- MASS_UNIT_MG = 4 (mg/L)
- MASS_UNIT_UG = 5 (ug/L)
If None, units of quality measurements are not changed.

The default is None.
bulk_species_mass_unit (list[int], optional) –
New units of all bulk species measurements – i.e. for each bulk species the measurement unit is specified. Note that the assumed ordering is the same as given in ‘bulk_species’ in the sensor configuration – 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)
If None, measurement units of bulk species are not changed.

The default is None.
surface_species_mass_unit (list[int], optional) –
New units of all surface species measurements – i.e. for each surface species the measurement unit is specified. Note that the assumed ordering is the same as given in ‘surface_species’ in the sensor configuration – 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)
If None, measurement units of surface species are not changed.

The default is None.
surface_species_area_unit (int, optional) –
New 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)
If None, are units of surface species are not changed.

The default is None.

Returns:

SCADA data instance with the new units.

Return type:

ScadaData

property demand_data_raw: numpy.ndarray

Gets the raw demand readings.

Returns:

Raw demand readings.

Return type:

extract_time_window(start_time: int, end_time: int | None = None)[source]

Extracts a time window of SCADA data from this SCADA data instance – i.e. creating a new SCADA data instance containing data from the requested time period only.

Parameters:

start_time (int) – Start time – i.e. beginning of the time window.
end_time (int, optional) –
End time – i.e. end of the time window. If None, all sensor readings from the start time onward are included

The default is None.

Returns:

New SCADA data instance containing data from the requested time period only.

Return type:

ScadaData

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_data_raw: numpy.ndarray

Gets the raw flow readings.

Returns:

Raw flow readings.

Return type:

property frozen_sensor_config: bool

Checks if the sensor configuration is frozen or not.

Returns:: True if the sensor configuration is frozen, False otherwise.
Return type:: bool

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_data() → numpy.ndarray[source]

Computes the final sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

If the ordering has not been changed in the sensor config, the columns (i.e. sensor readings) are ordered as follows:

Pressures

Flows

Demands

Nodes quality

Links quality

Valve state

Pumps state

Pumps efficiency

Pumps energy consumption

Tanks volume

Surface species concentrations

Bulk species nodes concentrations

Bulk species links concentrations

Otherwise, the ordering follows the one specified in the sensor config

Returns:

Final sensor readings.

Return type:

get_data_bulk_species_concentrations_as_edge_features(default_missing_value: float = 0.0) → tuple[numpy.ndarray, numpy.ndarray][source]

Returns the concentrations of bulk species as edge features together with a boolean mask indicating the presence of a sensor.

Note that only bulk species with at least one sensor are considered.

Parameters:

default_missing_value (float, optional) –

Default value (i.e. missing value) for links where no bulk species sensor is installed.

The default is 0.

Returns:

Concentrations of bulk species as edge features of shape [num_time_steps, num_links * 2, num_species], and mask of shape [num_links * 2, num_species].

Return type:

get_data_bulk_species_concentrations_as_node_features(default_missing_value: float = 0.0) → tuple[numpy.ndarray, numpy.ndarray][source]

Returns the concentrations of bulk species as node features together with a boolean mask indicating the presence of a sensor.

Note that only bulk species with at least one sensor are considered.

Parameters:

default_missing_value (float, optional) –

Default value (i.e. missing value) for nodes where no bulk species sensor is installed.

The default is 0.

Returns:

Concentrations of bulk species as node features of shape [num_time_steps, num_nodes, num_species], and mask of shape [num_nodes, num_species].

Return type:

get_data_bulk_species_link_concentration(bulk_species_sensor_locations: dict | None = None) → numpy.ndarray[source]

Gets the final bulk species link/pipe concentration sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

bulk_species_sensor_locations (dict, optional) –

Existing bulk species concentration sensors (species ID and link/pipe IDs) for which the sensor readings are requested. If None, the readings from all bulk species concentration link/pipe sensors are returned.

The default is None.

Returns:

Bulk species concentration sensor readings.

Return type:

get_data_bulk_species_node_concentration(bulk_species_sensor_locations: dict | None = None) → numpy.ndarray[source]

Gets the final bulk species node concentration sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

bulk_species_sensor_locations (dict, optional) –

Existing bulk species concentration sensors (species ID and node IDs) for which the sensor readings are requested. If None, the readings from all bulk species node concentration sensors are returned.

The default is None.

Returns:

Bulk species concentration sensor readings.

Return type:

get_data_demands(sensor_locations: list[str] | None = None) → numpy.ndarray[source]

Gets the final demand sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –

Existing demand sensor locations for which the sensor readings are requested. If None, the readings from all demand sensors are returned.

The default is None.

Returns:

Demand sensor readings.

Return type:

get_data_demands_as_node_features(default_missing_value: float = 0.0) → tuple[numpy.ndarray, numpy.ndarray][source]

Returns the demands as node features together with a boolean mask indicating the presence of a sensor.

Parameters:

default_missing_value (float, optional) –

Default value (i.e. missing value) for nodes where no demand sensor is installed.

The default is 0.

Returns:

Demands as node features of shape [num_time_steps, num_nodes], and mask of shape [num_nodes].

Return type:

get_data_edge_features(default_missing_value: float = 0.0) → tuple[numpy.ndarray, numpy.ndarray][source]

Returns the sensor readings as edge features together with a boolean mask indicating the presence of a sensor – i.e. flow, quality, surface species concentration, bulk species concentration at each link.

Note that only quantities with at least one sensor are considered.

Parameters:

default_missing_value (float, optional) –

Default value (i.e. missing value) for links where no sensor is installed.

The default is 0.

Returns:

Edge features of shape [num_links, num_time_steps, num_edge_features] and mask of shape [num_links, num_edge_features].

Return type:

get_data_flows(sensor_locations: list[str] | None = None) → numpy.ndarray[source]

Gets the final flow sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –

Existing flow sensor locations for which the sensor readings are requested. If None, the readings from all flow sensors are returned.

The default is None.

Returns:

Flow sensor readings.

Return type:

get_data_flows_as_edge_features(default_missing_value: float = 0.0) → tuple[numpy.ndarray, numpy.ndarray][source]

Returns the flows as edge features together with a boolean mask indicating the presence of a sensor.

Note that the second link has the opposite flow direction of the flow at the first link – recall that we have an undirected graph, i.e. two edges per link.

Parameters:

default_missing_value (float, optional) –

Default value (i.e. missing value) for links where no flow sensor is installed.

The default is 0.

Returns:

Flows as edge features of shape [num_time_steps, num_links * 2] and mask of shape [num_links * 2].

Return type:

get_data_links_quality(sensor_locations: list[str] | None = None) → numpy.ndarray[source]

Gets the final link quality sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –

Existing link quality sensor locations for which the sensor readings are requested. If None, the readings from all link quality sensors are returned.

The default is None.

Returns:

Link quality sensor readings.

Return type:

get_data_links_quality_as_edge_features(default_missing_value: float = 0) → tuple[numpy.ndarray, numpy.ndarray][source]

Returns the links’ quality as edge features together with a boolean mask indicating the presence of a sensor.

Parameters:

default_missing_value (float, optional) –

Default value (i.e. missing value) for links where no quality sensor is installed.

The default is 0.

Returns:

Links’ quality as edge features of shape [num_time_steps, num_links * 2], and mask of shape [num_links * 2].

Return type:

get_data_node_features(default_missing_value: float = 0.0) → tuple[numpy.ndarray, numpy.ndarray][source]

Returns the sensor readings as node features together with a boolean mask indicating the presence of a sensor – i.e. pressure, demand, quality, bulk species concentration at each node.

Note that only quantities with at least one sensor are considered.

Parameters:

default_missing_value (float, optional) –

Default value (i.e. missing value) for nodes where no sensor is installed.

The default is 0.

Returns:

Node features of shape [num_nodes, num_time_steps, num_node_features], and mask of shape [num_nodes, num_node_features].

Return type:

get_data_nodes_quality(sensor_locations: list[str] | None = None) → numpy.ndarray[source]

Gets the final node quality sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –

Existing node quality sensor locations for which the sensor readings are requested. If None, the readings from all node quality sensors are returned.

The default is None.

Returns:

Node quality sensor readings.

Return type:

get_data_nodes_quality_as_node_features(default_missing_value: float = 0) → tuple[numpy.ndarray, numpy.ndarray][source]

Returns the nodes’ quality as node features together with a boolean mask indicating the presence of a sensor.

Parameters:

default_missing_value (float, optional) –

Default value (i.e. missing value) for nodes where no quality sensor is installed.

The default is 0.

Returns:

Nodes’ quality as node features of shape [num_time_steps, num_nodes], and mask of shape [num_nodes].

Return type:

get_data_pressures(sensor_locations: list[str] | None = None) → numpy.ndarray[source]

Gets the final pressure sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –

Existing pressure sensor locations for which the sensor readings are requested. If None, the readings from all pressure sensors are returned.

The default is None.

Returns:

Pressure sensor readings.

Return type:

get_data_pressures_as_node_features(default_missing_value: float = 0.0) → tuple[numpy.ndarray, numpy.ndarray][source]

Returns the pressures as node features together with a boolean mask indicating the presence of a sensor.

Parameters:

default_missing_value (float, optional) –

Default value (i.e. missing value) for nodes where no pressure sensor is installed.

The default is 0.

Returns:

Pressures as node features of shape [num_time_steps, num_nodes], and mask of shape [num_nodes].

Return type:

get_data_pumps_efficiency(sensor_locations: list[str] | None = None) → numpy.ndarray[source]

Gets the final pump efficiency sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –

Existing pump efficiency sensor locations for which the sensor readings are requested. If None, the readings from all pump efficiency sensors are returned.

The default is None.

Returns:

Pump efficiency sensor readings.

Return type:

get_data_pumps_efficiency_as_node_features(default_missing_value: float = 0.0) → tuple[numpy.ndarray, numpy.ndarray][source]

Returns the pump efficiency as node features together with a boolean mask indicating the presence of a sensor.

Parameters:

default_missing_value (float, optional) –

Default value (i.e. missing value) for nodes where no pump efficiency sensor is installed.

The default is 0.

Returns:

Pump efficiencies as node features of shape [num_time_steps, num_nodes], and mask of shape [num_nodes].

Return type:

get_data_pumps_energyconsumption(sensor_locations: list[str] | None = None) → numpy.ndarray[source]

Gets the final pump energy consumption sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –

Existing pump energy consumption sensor locations for which the sensor readings are requested. If None, the readings from all pump energy consumption sensors are returned.

The default is None.

Returns:

Pump energy consumption sensor readings.

Return type:

get_data_pumps_energyconsumption_as_node_features(default_missing_value: float = 0.0) → tuple[numpy.ndarray, numpy.ndarray][source]

Returns the pump energy consumption as node features together with a boolean mask indicating the presence of a sensor.

Parameters:

default_missing_value (float, optional) –

Default value (i.e. missing value) for nodes where no pump energy consumption sensor is installed.

The default is 0.

Returns:

Pump energy consumptions as node features of shape [num_time_steps, num_nodes], and mask of shape [num_nodes].

Return type:

get_data_pumps_state(sensor_locations: list[str] | None = None) → numpy.ndarray[source]

Gets the final pump state sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –

Existing pump state sensor locations for which the sensor readings are requested. If None, the readings from all pump state sensors are returned.

The default is None.

Returns:

Pump state sensor readings.

Return type:

get_data_pumps_state_as_node_features(default_missing_value: float = 0.0) → tuple[numpy.ndarray, numpy.ndarray][source]

Returns the pump state as node features together with a boolean mask indicating the presence of a sensor.

Parameters:

default_missing_value (float, optional) –

Default value (i.e. missing value) for nodes where no pump state sensor is installed.

The default is 0.

Returns:

Pump state as node features of shape [num_time_steps, num_nodes], and mask of shape [num_nodes].

Return type:

get_data_surface_species_concentration(surface_species_sensor_locations: dict | None = None) → numpy.ndarray[source]

Gets the final surface species concentration sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

surface_species_sensor_locations (dict, optional) –

Existing surface species concentration sensors (species ID and link/pipe IDs) for which the sensor readings are requested. If None, the readings from all surface species concentration sensors are returned.

The default is None.

Returns:

Surface species concentration sensor readings.

Return type:

get_data_surface_species_concentrations_as_edge_features(default_missing_value: float = 0.0) → tuple[numpy.ndarray, numpy.ndarray][source]

Returns the concentrations of surface species as edge features together with a boolean mask indicating the presence of a sensor.

Note that only surface species with at least one sensor are considered.

Parameters:

default_missing_value (float, optional) –

Default value (i.e. missing value) for links where no surface species sensor is installed.

The default is 0.

Returns:

Concentrations of surface species as edge features of shape [num_time_steps, num_links * 2, num_species], and mask of shape [num_links * 2, num_species].

Return type:

get_data_tanks_water_volume(sensor_locations: list[str] | None = None) → numpy.ndarray[source]

Gets the final water tanks volume sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –

Existing flow sensor locations for which the sensor readings are requested. If None, the readings from all water tanks volume sensors are returned.

The default is None.

Returns:

Water tanks volume sensor readings.

Return type:

get_data_tanks_water_volume_as_node_features(default_missing_value: float = 0.0) → tuple[numpy.ndarray, numpy.ndarray][source]

Returns the tank water volume as node features together with a boolean mask indicating the presence of a sensor.

Parameters:

default_missing_value (float, optional) –

Default value (i.e. missing value) for nodes where no tank water volume sensor is installed.

The default is 0.

Returns:

Tank water volumes as node features of shape [num_time_steps, num_nodes], and mask of shape [num_nodes].

Return type:

get_data_valves_state(sensor_locations: list[str] | None = None) → numpy.ndarray[source]

Gets the final valve state sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –

Existing valve state sensor locations for which the sensor readings are requested. If None, the readings from all valve state sensors are returned.

The default is None.

Returns:

Valve state sensor readings.

Return type:

get_data_valves_state_as_node_features(default_missing_value: float = 0.0) → tuple[numpy.ndarray, numpy.ndarray][source]

Returns the valves state as node features together with a boolean mask indicating the presence of a sensor.

Parameters:

default_missing_value (float, optional) –

Default value (i.e. missing value) for nodes where no valves state sensor is installed.

The default is 0.

Returns:

Valves state as node features of shape [num_time_steps, num_nodes], and mask of shape [num_nodes].

Return type:

get_topo_edge_indices() → numpy.ndarray[source]

Returns the edge indices – i.e. a 2 dimensional array where the first dimension denotes the source node indices and the second dimension denotes the target node indices for all links in the network. Nodes are ordered according to EPANET.

Note that the network is consideres as a directed graph – i.e. one link corresponds to two edges in opposite directions!

Returns:

Edge indices of shape [2, num_links * 2].

Return type:

join(other) → None[source]

Joins two ScadaData instances based on the sensor reading times. Consequently, both instances must be equal in their sensor reading times. Attributes (i.e. types of sensor readings) that are NOT present in THIS instance but in others will be added to this instance – all other attributes are ignored. The sensor configuration is updated according to the sensor readings in other.

Parameters:: other (ScadaData) – Other scada data to be concatenated to this data.

property link_quality_data_raw: numpy.ndarray

Gets the raw link quality readings.

Returns:

Raw link quality readings.

Return type:

map_link_id_to_edge_idx(link_id: str) → tuple[int, int][source]

Maps a given link to the corresponding two indices in the edge indices as computed by epyt_flow.simulation.scada.scada_data.ScadaData.topo_edge_indices().

Returns:: Indices.
Return type:: tuple[int, int]

property network_topo: NetworkTopology

Returns the topology of the water distribution network.

Returns:: Topology of the network.
Return type:: epyt_flow.topology.NetworkTopology

property node_quality_data_raw: numpy.ndarray

Gets the raw node quality readings.

Returns:

Raw node quality readings.

Return type:

plot_bulk_species_link_concentration(bulk_species_link_sensors: dict | None = None, show: bool = True, save_to_file: str | None = None, ax: matplotlib.axes.Axes | None = None) → matplotlib.axes.Axes[source]

Plots the final bulk species link concentration sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

bulk_species_link_sensors (dict, optional) –
Existing bulk species link concentration sensors (species ID and link/pipe IDs) for which the sensor readings have to be plotted. If None, the readings from all bulk species link concentration sensors are plotted.

The default is None.
show (bool, optional) –
If True, the plot/figure is shown in a window.

Only considered when ‘ax’ is None.

The default is True.
save_to_file (str, optional) –
File to which the plot is saved.

If specified, ‘show’ must be set to False – i.e. a plot can not be shown and saved to a file at the same time!

The default is None.
ax (matplotlib.axes.Axes, optional) –
If not None, ‘ax’ is used for plotting.

The default is None.

Returns:

Plot.

Return type:

plot_bulk_species_node_concentration(bulk_species_node_sensors: dict | None = None, show: bool = True, save_to_file: str | None = None, ax: matplotlib.axes.Axes | None = None) → matplotlib.axes.Axes[source]

Plots the final bulk species node concentration sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

bulk_species_node_sensors (dict, optional) –
Existing bulk species concentration sensors (species ID and node IDs) for which the sensor readings are requested. If None, the readings from all bulk species node concentration sensors are returned.

The default is None.
show (bool, optional) –
If True, the plot/figure is shown in a window.

Only considered when ‘ax’ is None.

The default is True.
save_to_file (str, optional) –
File to which the plot is saved.

If specified, ‘show’ must be set to False – i.e. a plot can not be shown and saved to a file at the same time!

The default is None.
ax –
If not None, ‘ax’ is used for plotting.

The default is None.

Returns:

Plot.

Return type:

plot_demands(sensor_locations: list[str] | None = None, show: bool = True, save_to_file: str | None = None, ax: matplotlib.axes.Axes | None = None) → matplotlib.axes.Axes[source]

Plots the final demand sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –
Existing demand sensor locations for which the sensor readings have to be plotted. If None, the readings from all demand sensors are plotted.

The default is None.
show (bool, optional) –
If True, the plot/figure is shown in a window.

Only considered when ‘ax’ is None.

The default is True.
save_to_file (str, optional) –
File to which the plot is saved.

If specified, ‘show’ must be set to False – i.e. a plot can not be shown and saved to a file at the same time!

The default is None.
ax –
If not None, ‘ax’ is used for plotting.

The default is None.

Returns:

Plot.

Return type:

plot_flows(sensor_locations: list[str] | None = None, show: bool = True, save_to_file: str | None = None, ax: matplotlib.axes.Axes | None = None) → matplotlib.axes.Axes[source]

Plots the final flow sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –
Existing flow sensor locations for which the sensor readings have to be plotted. If None, the readings from all flow sensors are plotted.

The default is None.
show (bool, optional) –
If True, the plot/figure is shown in a window.

Only considered when ‘ax’ is None.

The default is True.
save_to_file (str, optional) –
File to which the plot is saved.

If specified, ‘show’ must be set to False – i.e. a plot can not be shown and saved to a file at the same time!

The default is None.
ax –
If not None, ‘ax’ is used for plotting.

The default is None.

Returns:

Plot.

Return type:

plot_links_quality(sensor_locations: list[str] | None = None, show: bool = True, save_to_file: str | None = None, ax: matplotlib.axes.Axes | None = None) → matplotlib.axes.Axes[source]

Plots the final link/pipe quality sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –
Existing link quality sensor locations for which the sensor readings have to be plotted. If None, the readings from all link quality sensors are plotted.

The default is None.
show (bool, optional) –
If True, the plot/figure is shown in a window.

Only considered when ‘ax’ is None.

The default is True.
save_to_file (str, optional) –
File to which the plot is saved.

If specified, ‘show’ must be set to False – i.e. a plot can not be shown and saved to a file at the same time!

The default is None.
ax –
If not None, ‘ax’ is used for plotting.

The default is None.

Returns:

Plot.

Return type:

plot_nodes_quality(sensor_locations: list[str] | None = None, show: bool = True, save_to_file: str | None = None, ax: matplotlib.axes.Axes | None = None) → matplotlib.axes.Axes[source]

Plots the final node quality sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –
Existing node quality sensor locations for which the sensor readings have to be plotted. If None, the readings from all node quality sensors are plotted.

The default is None.
show (bool, optional) –
If True, the plot/figure is shown in a window.

Only considered when ‘ax’ is None.

The default is True.
save_to_file (str, optional) –
File to which the plot is saved.

If specified, ‘show’ must be set to False – i.e. a plot can not be shown and saved to a file at the same time!

The default is None.
ax –
If not None, ‘ax’ is used for plotting.

The default is None.

Returns:

Plot.

Return type:

plot_pressures(sensor_locations: list[str] | None = None, show: bool = True, save_to_file: str | None = None, ax: matplotlib.axes.Axes | None = None) → matplotlib.axes.Axes[source]

Plots the final pressure sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –
Existing pressure sensor locations for which the sensor readings have to be plotted. If None, the readings from all pressure sensors are plotted.

The default is None.
show (bool, optional) –
If True, the plot/figure is shown in a window.

Only considered when ‘ax’ is None.

The default is True.
save_to_file (str, optional) –
File to which the plot is saved.

If specified, ‘show’ must be set to False – i.e. a plot can not be shown and saved to a file at the same time!

The default is None.
ax –
If not None, ‘ax’ is used for plotting.

The default is None.

Returns:

Plot.

Return type:

plot_pumps_efficiency(sensor_locations: list[str] | None = None, show: bool = True, save_to_file: str | None = None, ax: matplotlib.axes.Axes | None = None) → matplotlib.axes.Axes[source]

Plots the final pump efficiency sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –
Existing pump efficiency sensor locations for which the sensor readings have to be plotted. If None, the readings from all pump efficiency sensors are plotted.

The default is None.
show (bool, optional) –
If True, the plot/figure is shown in a window.

Only considered when ‘ax’ is None.

The default is True.
save_to_file (str, optional) –
File to which the plot is saved.

If specified, ‘show’ must be set to False – i.e. a plot can not be shown and saved to a file at the same time!

The default is None.
ax –
If not None, ‘ax’ is used for plotting.

The default is None.

Returns:

Plot.

Return type:

plot_pumps_energyconsumption(sensor_locations: list[str] | None = None, show: bool = True, save_to_file: str | None = None, ax: matplotlib.axes.Axes | None = None) → matplotlib.axes.Axes[source]

Plots the final pump energy consumption sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –
Existing pump energy consumption sensor locations for which the sensor readings have to be plotted. If None, the readings from all pump energy consumption sensors are plotted.

The default is None.
show (bool, optional) –
If True, the plot/figure is shown in a window.

Only considered when ‘ax’ is None.

The default is True.
save_to_file (str, optional) –
File to which the plot is saved.

If specified, ‘show’ must be set to False – i.e. a plot can not be shown and saved to a file at the same time!

The default is None.
ax –
If not None, ‘ax’ is used for plotting.

The default is None.

Returns:

Plot.

Return type:

plot_pumps_state(sensor_locations: list[str] | None = None, show: bool = True, save_to_file: str | None = None, ax: matplotlib.axes.Axes | None = None) → matplotlib.axes.Axes[source]

Plots the final pump state sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –
Existing pump state sensor locations for which the sensor readings have to be plotted. If None, the readings from all pump state sensors are plotted.

The default is None.
show (bool, optional) –
If True, the plot/figure is shown in a window.

Only considered when ‘ax’ is None.

The default is True.
save_to_file (str, optional) –
File to which the plot is saved.

If specified, ‘show’ must be set to False – i.e. a plot can not be shown and saved to a file at the same time!

The default is None.
ax –
If not None, ‘ax’ is used for plotting.

The default is None.

Returns:

Plot.

Return type:

plot_surface_species_concentration(surface_species_sensor_locations: dict | None = None, show: bool = True, save_to_file: str | None = None, ax: matplotlib.axes.Axes | None = None) → matplotlib.axes.Axes[source]

Plots the final surface species concentration sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

surface_species_sensor_locations (dict, optional) –
Existing surface species concentration sensors (species ID and link/pipe IDs) for which the sensor readings have to be plotted. If None, the readings from all surface species concentration sensors are plotted.

The default is None.
show (bool, optional) –
If True, the plot/figure is shown in a window.

Only considered when ‘ax’ is None.

The default is True.
save_to_file (str, optional) –
File to which the plot is saved.

If specified, ‘show’ must be set to False – i.e. a plot can not be shown and saved to a file at the same time!

The default is None.
ax –
If not None, ‘ax’ is used for plotting.

The default is None.

Returns:

Plot.

Return type:

plot_tanks_water_volume(sensor_locations: list[str] | None = None, show: bool = True, save_to_file: str | None = None, ax: matplotlib.axes.Axes | None = None) → matplotlib.axes.Axes[source]

Plots the final water tanks volume sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –
Existing flow sensor locations for which the sensor readings have to be plotted. If None, the readings from all water tanks volume sensors are plotted.

The default is None.
show (bool, optional) –
If True, the plot/figure is shown in a window.

Only considered when ‘ax’ is None.

The default is True.
save_to_file (str, optional) –
File to which the plot is saved.

If specified, ‘show’ must be set to False – i.e. a plot can not be shown and saved to a file at the same time!

The default is None.
ax –
If not None, ‘ax’ is used for plotting.

The default is None.

Returns:

Plot.

Return type:

plot_valves_state(sensor_locations: list[str] | None = None, show: bool = True, save_to_file: str | None = None, ax: matplotlib.axes.Axes | None = None) → matplotlib.axes.Axes[source]

Plots the final valve state sensor readings – note that those might be subject to given sensor faults and sensor noise/uncertainty.

Parameters:

sensor_locations (list[str], optional) –
Existing valve state sensor locations for which the sensor readings have to be plotted. If None, the readings from all valve state sensors are plotted.

The default is None.
show (bool, optional) –
If True, the plot/figure is shown in a window.

Only considered when ‘ax’ is None.

The default is True.
save_to_file (str, optional) –
File to which the plot is saved.

If specified, ‘show’ must be set to False – i.e. a plot can not be shown and saved to a file at the same time!

The default is None.
ax –
If not None, ‘ax’ is used for plotting.

The default is None.

Returns:

Plot.

Return type:

property pressure_data_raw: numpy.ndarray

Gets the raw pressure readings.

Returns:

Raw pressure readings.

Return type:

property pumps_efficiency_data_raw: numpy.ndarray

Gets the raw efficiency of each pump.

Returns:

Pumps’ efficiency.

Return type:

property pumps_energyconsumption_data_raw: numpy.ndarray

Gets the raw energy consumption of each pump.

Returns:

Energy consumption of each pump.

Return type:

property pumps_state_data_raw: numpy.ndarray

Gets the raw pump state readings.

Returns:

Raw pump state readings.

Return type:

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.

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.

Returns:: All sensor faults.
Return type:: list[SensorReadingEvent]

property sensor_readings_time: numpy.ndarray

Gets the sensor readings time stamps.

Returns:

Sensor readings time stamps.

Return type:

property surface_species_concentration_raw: numpy.ndarray

Gets the raw surface species concentrations at links/pipes.

Returns:

Raw species concentrations.

Return type:

property tanks_volume_data_raw: numpy.ndarray

Gets the raw tank volume readings.

Returns:

Raw tank volume readings.

Return type:

to_excel_file(f_out: str, export_raw_data: bool = False) → None[source]

Exporting this SCADA data to MS Excel (.xlsx file).

Parameters:

f_out (str) – Path to the .xlsx file to which the SCADA data will be exported.
export_raw_data (bool, optional) –
If True, the raw measurements (i.e. sensor reading without any noise or faults) are exported instead of the final sensor readings.

The default is False.

to_matlab_file(f_out: str, export_raw_data: bool = False) → None[source]

Exporting this SCADA data to Matlab (.mat file).

Parameters:

f_out (str) – Path to the .mat file to which the SCADA data will be exported.
export_raw_data (bool, optional) –
If True, the raw measurements (i.e. sensor reading without any noise or faults) are exported instead of the final sensor readings.

The default is False.

to_numpy_file(f_out: str, export_raw_data: bool = False) → None[source]

Exporting this SCADA data to Numpy (.npz file).

Parameters:

f_out (str) – Path to the .npz file to which the SCADA data will be exported.
export_raw_data (bool, optional) –
If True, the raw measurements (i.e. sensor reading without any noise or faults) are exported instead of the final sensor readings.

The default is False.

to_pandas_dataframe(export_raw_data: bool = False) → pandas.DataFrame[source]

Exports this SCADA data to a Pandas dataframe.

Parameters:

export_raw_data (bool, optional) –

If True, the raw measurements (i.e. sensor reading without any noise or faults) are exported instead of the final sensor readings.

The default is False.

Returns:

Exported data.

Return type:

pandas.DataFrame

topo_adj_matrix() → scipy.sparse.bsr_array[source]

Returns the adjacency matrix of the network.

Nodes are ordered according to EPANET.

Returns:: Adjacency matrix as a sparse array of shape [num_nodes, num_nodes].
Return type:: scipy.bsr_array

property valves_state_data_raw: numpy.ndarray

Gets the raw valve state readings.

Returns:

Raw valve state readings.

Return type:

property warnings_code: numpy.ndarray

Returns the codes/IDs of EPANET errors/warnings (if any) for each time step. Note that zero denotes the absence of any error/warning.

Returns:

numpy.ndarray: Codes/IDs of EPANET errors/warnings (if any) for each time step.

epyt_flow.simulation.scada.scada_data_export

Module provides classes for exporting SCADA data stored in ScadaData.

class epyt_flow.simulation.scada.scada_data_export.ScadaDataExport(f_out: str, export_raw_data: bool = False, **kwds)[source]

Bases: object

Base class for exporting SCADA data stored in ScadaData.

Parameters:

f_out (str) – Path to the file to which the SCADA data will be exported.
export_raw_data (bool, optional) –
If True, the raw measurements (i.e. sensor reading without any noise or faults) are exported instead of the final sensor readings.

The default is False.

static create_column_desc(scada_data: ScadaData) → numpy.ndarray[source]

Creates column descriptions – i.e. sensor type and location for each column

Parameters:

scada_data (ScadaData) – SCADA data to be described.

Returns:

3-dimensional array describing all columns of the sensor readings: The first dimension describes the sensor type, the second dimension describes the sensor location, and the third one describes the measurement units.

Return type:

static create_global_sensor_config(scada_data: ScadaData) → SensorConfig[source]

Creates a global sensor configuration with sensors placed everywhere.

Parameters:: scada_data (ScadaData) – SCADA data for which the global sensor configuration is to be created.
Returns:: Global sensor configuration.
Return type:: SensorConfig

abstract export(scada_data: ScadaData) → None[source]

Exports given SCADA data.

Parameters:: scada_data (ScadaData) – SCADA data to be exported.

property export_raw_data: bool

True if the raw measurements instead of the final sensor readings are requested.

Returns:: True if the raw measurements instead of the final sensor readings are requested.
Return type:: bool

property f_out: str

Gets the path to the file to which the SCADA data will be exported.

Returns:: Path to the file to which the SCADA data will be exported.
Return type:: str

class epyt_flow.simulation.scada.scada_data_export.ScadaDataMatlabExport(f_out: str, export_raw_data: bool = False, **kwds)[source]

Bases: ScadaDataExport

Class for exporting SCADA data to MATLAB (.mat file).

export(scada_data: ScadaData) → None[source]

Exports given SCADA data.

Parameters:: scada_data (ScadaData) – SCADA data to be exported.

class epyt_flow.simulation.scada.scada_data_export.ScadaDataNumpyExport(f_out: str, export_raw_data: bool = False, **kwds)[source]

Bases: ScadaDataExport

Class for exporting SCADA data to numpy (.npz file).

export(scada_data: ScadaData) → None[source]

Exports given SCADA data.

Parameters:: scada_data (ScadaData) – SCADA data to be exported.

class epyt_flow.simulation.scada.scada_data_export.ScadaDataXlsxExport(f_out: str, export_raw_data: bool = False, **kwds)[source]

Bases: ScadaDataExport

Class for exporting SCADA data to Excel (.xlsx file).

export(scada_data: ScadaData) → None[source]

Exports given SCADA data.

Parameters:: scada_data (ScadaData) – SCADA data to be exported.

epyt_flow.simulation.scada.custom_control

Module provides a base class for custom control modules.

class epyt_flow.simulation.scada.custom_control.CustomControlModule(**kwds)[source]

Bases: ABC

Base class for a custom control module.

epanet_api

API to EPANET and EPANET-MSX. Is set in init().

Type:: epanet_plus.EPyT

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

Initializes the control module.

Parameters:: epanet_api – API to EPANET for implementing the control module.

set_node_quality_source_value(node_id: str, pattern_id: str, qual_value: float) → None[source]

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_pump_speed(pump_id: str, speed: float) → None[source]

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[source]

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[source]

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(scada_data: ScadaData) → None[source]

Implements the control algorithm – i.e. mapping of sensor reading to actions.

Parameters:: scada_data (ScadaData) – Sensor readings.

epyt_flow.simulation.scada.simple_control

The module contains classes for representing simple control rules as used in EPANET.

class epyt_flow.simulation.scada.simple_control.SimpleControlModule(link_id: str, link_status: int | float, cond_type: int, cond_var_value: str | int, cond_comp_value: float, **kwds)[source]

A class for representing a simple EPANET control rule.

Parameters:

link_id (str) – Link ID.
link_status (int or float) –
Status of the link that is set when the condition is fullfilled.

Instance of float if the link constitutes a pump – in this case, the argument corresponds to the pump speed.

Instance of int if the link constitutes a valve – in this case, must be one of the followig costants defined in ActuatorConstants:
- EN_CLOSED = 0
- EN_OPEN = 1
- EN_SET_CLOSED = -1e10
- EN_SET_OPEN = 1e10
cond_type (int) –
Condition/Rule type.

Must be one of the following EPANET constants:
- EN_LOWLEVEL = 0
- EN_HILEVEL = 1
- EN_TIMER = 2
- EN_TIMEOFDAY = 3
cond_var_value (str or int) –
Condition/Rule variable value.

Node ID in the cases of EN_LOWLEVEL or EN_HILEVEL. Time of the day (in AM/PM format) in the case of EN_TIMEOFDAY. Number of hours (as an integer) since simulation start in the case of EN_TIMER.
cond_comp_value (float) –
The condition/rule comparison value at which this control rule is triggered.

Lower or upper value on the pressure (or tank level) in the cases of EN_LOWLEVEL and EN_HILEVEL.

Will be ignored in all other cases – i.e. should be set to None.

property cond_comp_value: float

Returns the condition/rule comparison value – might be None if not needed.

Lower or upper value on the pressure (or tank level) in the cases of EN_LOWLEVEL and EN_HILEVEL.

Returns:: Condition/Rule comparison value.
Return type:: float

property cond_type: int

Returns the condition/rule type.

Returns:

Condition/Rule type – will be one of the following EPANET constants:

EN_LOWLEVEL = 0

EN_HILEVEL = 1

EN_TIMER = 2

EN_TIMEOFDAY = 3

Return type:

int

property cond_var_value: str | int

Return the condition/rule variable value.

Node ID in the cases of EN_LOWLEVEL or EN_HILEVEL. Time of the day (in AM/PM format) in the case of EN_TIMEOFDAY. Number of hours (as an integer) since simulation start in the case of EN_TIMER.

Returns:: Condition/rule variable value.
Return type:: str or int

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

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

property link_id: str

Returns the link ID.

Returns:: Link ID.
Return type:: str

property link_status: int | float

Returns the link status that is set when the condition is fullfilled.

Returns:

Pump speed if the link is a pump, otherwise one of the followig constants defined in ActuatorConstants:

EN_CLOSED = 0

EN_OPEN = 1

EN_SET_CLOSED = -1e10

EN_SET_OPEN = 1e10

Return type:

int or float

class epyt_flow.simulation.scada.simple_control.SimplePumpSpeedConditionControl(pump_id: str, pump_speed: float, node_id: str, comp_type: int, comp_value: float)[source]

A class for representing a simple IF-THEN control rule for setting the pump speed.

Parameters:

pump_id (str) – Pump ID.
pump_speed (float) – Pump speed.
node_id (str) – Node ID.
comp_type (int) –
Comparison type – must be one of the following EPANET constants:
- EN_LOWLEVEL = 0
- EN_HILEVEL = 1
comp_value (float:) – Lower or upper value on the pressure (or tank level) at which this control rule is triggered.

class epyt_flow.simulation.scada.simple_control.SimplePumpSpeedTimeControl(pump_id: str, pump_speed: float, time: str | int)[source]

A class for representing a simple control rule for setting the pump speed at some point in time.

Parameters:

pump_id (str) – Pump ID.
pump_speed (float) – Pump speed.
time (str or int) – Time of the day (in AM/PM format) in the case or number of hours (as an integer) since simulation start.

class epyt_flow.simulation.scada.simple_control.SimpleValveConditionControl(valve_id: str, valve_status: int, node_id: str, comp_type: int, comp_value: float)[source]

A class for representing a simple IF-THEN control rule for setting the valve status.

Parameters:

valve_id (str) – valve ID.
valve_status (int) –
Valve status – must be one of the followig costants defined in ActuatorConstants:
- EN_CLOSED = 0
- EN_OPEN = 1
- EN_SET_CLOSED = -1e10
- EN_SET_OPEN = 1e10
node_id (str) – Node ID.
comp_type (int) –
Comparison type – must be one of the following EPANET constants:
- EN_LOWLEVEL = 0
- EN_HILEVEL = 1
comp_value (float:) – Lower or upper value on the pressure (or tank level) at which this control rule is triggered.

class epyt_flow.simulation.scada.simple_control.SimpleValveTimeControl(valve_id: str, valve_status: int, time: str | int)[source]

A class for representing a simple control rule for setting the valve status at some point in time.

Parameters:

valve_id (str) – valve ID.
valve_status (int) –
Valve status – must be one of the followig costants defined in ActuatorConstants:
- EN_CLOSED = 0
- EN_OPEN = 1
- EN_SET_CLOSED = -1e10
- EN_SET_OPEN = 1e10
time (str or int) – Time of the day (in AM/PM format) in the case or number of hours (as an integer) since simulation start.

epyt_flow.simulation.scada.complex_control

Module contains a class for representing complex control rules as implemented in EPANET.

class epyt_flow.simulation.scada.complex_control.ComplexControlModule(rule_id: str, condition_1: RuleCondition, additional_conditions: list[tuple[int, RuleCondition]], actions: list[RuleAction], else_actions: list[RuleAction], priority: int, **kwds)[source]

Class representing a complex control module (i.e. IF-THEN-ELSE rule) as implemented in EPANET.

Parameters:

rule_id (str) – ID of the rule.
condition_1 (RuleCondition) – First condition of this rule.
additional_conditions (list[tuple[int, RuleCondition]]) –
List of (optional) additional conditions incl. their conjunction operator (must be either EN_R_AND = 2 or EN_R_OR = 3).

Empty list if there are no additional conditions.
actions (list[RuleAction]) – List of actions that are applied if the conditions are met. Must contain at least one action.
else_actions (list[RuleAction]) – List of actions that are applied if the conditions are NOT met.
priority (int) – Priority of this control rule.

property actions: list[RuleAction]

Returns the list of actions that are applied if the conditions are met.

Returns:: List of actions that are applied if the conditions are met.
Return type:: list[RuleAction]

property additional_conditions: list[tuple[int, RuleCondition]]

Returns the list of (optional) additional conditions incl. their conjunction operator. Empty list if there are no additional conditions.

Returns:: List of (optional) additional conditions incl. their conjunction operator.
Return type:: list[tuple[int, RuleCondition]]

property condition_1: RuleCondition

Returns the first condition of this rule.

Returns:: First condition of this rule.
Return type:: RuleCondition

property else_actions: list[RuleAction]

Returns the list of actions that are applied if the conditions are NOT met.

Returns:: List of actions that are applied if the conditions are NOT met.
Return type:: list[RuleAction]

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

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

property priority: int

Returns the priority of this control rule.

Returns:: Priority of this control rule.
Return type:: int

property rule_id: str

Returns the ID of this control rule.

Returns:: ID of this control rule.
Return type:: str

class epyt_flow.simulation.scada.complex_control.RuleAction(link_type_id: int, link_id: str, action_type_id: int, action_value: Any, **kwds)[source]

Class representing a rule action.

Parameters:

link_type_id (int) –
Link type ID.

Must be one of following EPANET constants:
- EN_CVPIPE = 0
- EN_PIPE = 1
- EN_PUMP = 2
- EN_PRV = 3
- EN_PSV = 4
- EN_PBV = 5
- EN_FCV = 6
- EN_TCV = 7
- EN_GPV = 8
link_id (str) – Link ID.
action_type_id (int) –
Type ID of the action.

Must be one of the following constants:
- EN_R_ACTION_SETTING = -1
- EN_R_ACTION_STATUS_OPEN = 1
- EN_R_ACTION_STATUS_CLOSED = 2
- EN_R_ACTION_STATUS_ACTIVE = 3
action_value (Any) – Value of the acton (e.g. pump speed). Only relevant if action_type_id = EN_R_SETTING, will be ignored in all other cases.

property action_type_id: int

Returns the type ID of the action.

Will be one of the following constants:

EN_R_ACTION_SETTING = -1

EN_R_ACTION_STATUS_OPEN = 1

EN_R_ACTION_STATUS_CLOSED = 2

EN_R_ACTION_STATUS_ACTIVE = 3

Returns:: Type ID of the action.
Return type:: int

property action_value: Any

Returns the value of the acton (e.g. pump speed). Only relevant if action_type_id = EN_R_SETTING.

Returns:: Value of the action.
Return type:: Any

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

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

property link_id: str

Returns the link ID.

Returns:: Link ID.
Return type:: str

property link_type_id: int

Returns the link type ID.

Will be one of the following EPANET constants:

EN_CVPIPE = 0

EN_PIPE = 1

EN_PUMP = 2

EN_PRV = 3

EN_PSV = 4

EN_PBV = 5

EN_FCV = 6

EN_TCV = 7

EN_GPV = 8

Returns:: Link type ID.
Return type:: int

class epyt_flow.simulation.scada.complex_control.RuleCondition(object_type_id: int, object_id: str, attribute_id: int, relation_type_id: int, value: Any, **kwds)[source]