epyt_flow.simulation.events

epyt_flow.simulation.events.event

Module provides a base class for events.

class epyt_flow.simulation.events.event.Event(start_time: int, end_time: int | None = None, **kwds)[source]

Bases: ABC

Base class for an event.

Parameters:

start_time (int) – Starting time (seconds since the simulation start) of this event.
end_time (int, optional) –
Time (seconds since the simulation start) when this event ends – None if it never ends.

The default is None.

property end_time: int

Gets the end time (seconds since the simulation start) of this event. float(“inf”) if it never ends.

Returns:: End time of this event.
Return type:: int

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 and value.
Return type:: dict

property start_time: int

Gets the start time (seconds since the simulation start) of this event.

Returns:: Start time of this event.
Return type:: int

epyt_flow.simulation.events.system_event

Module provides base classes for system events such as leakages, actuator events, etc.

class epyt_flow.simulation.events.system_event.SystemEvent(**kwds)[source]

Bases: Event

Base class for a system event – i.e. an event that affects the EPANET simulation.

abstract apply(cur_time: int) → None[source]

Implements the event using EPANET and EPANET-MSX.

This function is only called when the event is active.

Parameters:: cur_time (int) – Current time (seconds since the start) in the simulation.

cleanup() → None[source]: Clean up any changes/modifications made by this event.

exit(cur_time) → None[source]

Is called ONCE after the event is over – i.e. next time step after end_time.

Any “clean-up” or “exiting” logic should go here.

Parameters:: cur_time (int) – Current time (seconds since the start) in the simulation.

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

Initializes the event.

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

reset() → None[source]: Resets this event – i.e. make it ready for another simulation run.

step(cur_time) → None[source]

Is called at every iteration (time step) in the simulation. apply or exit are called if necessary.

Parameters:: cur_time (int) – Current time (seconds since the start) in the simulation.

epyt_flow.simulation.events.leakages

Module provides classes for implementing leakages.

class epyt_flow.simulation.events.leakages.AbruptLeakage(link_id: str, diameter: float | None = None, area: float | None = None, **kwds)[source]

Bases: Leakage

Class implementing an abrupt leakage event.

Parameters:

link_id (str) – ID of the link at which the leak is placed.
diameter (float, optional) –
Diameter of the leak.

Alternatively, ‘area’ can be used to specify the size of this leak – in this case, ‘diameter’ must be set to None.

The default is None.
area (float, optional) –
Area of the leakd.

Alternatively, ‘diameter’ can be used to specify the size of this leak – in this case, ‘area’ must be set to None.

The default is None.

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

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

Initializes the event.

Parameters:: epanet_api – API to EPANET and EPANET-MSX.

class epyt_flow.simulation.events.leakages.IncipientLeakage(link_id: str, peak_time: int, diameter: float | None = None, area: float | None = None, **kwds)[source]

Bases: Leakage

Class implementing an incipient leakage event.

Parameters:

link_id (str) – ID of the link at which the leak is placed.
diameter (float, optional) –
Maximum diameter of the leak – i.e. small leak diameter in the beginning, growing over time until peak time is reached.

Alternatively, ‘area’ can be used to specify the size of this leak – in this case, ‘diameter’ must be set to None.

The default is None.
area (float, optional) –
Maximum area of the leak – i.e. small leak area in the beginning, growing over time until peak time is reached.

Alternatively, ‘diameter’ can be used to specify the size of this leak – in this case, ‘area’ must be set to None.

The default is None.
peak_time (int) – Time (seconds since the simulation start) when this leak reaches its larges size (leak diameter).

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 and value.
Return type:: dict

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

Initializes the event.

Parameters:: epanet_api – API to EPANET and EPANET-MSX.

property peak_time: int

Gets the peak time (seconds since the simulation start) of the leak.

Returns:: Peak time of the leak.
Return type:: int

class epyt_flow.simulation.events.leakages.Leakage(link_id: str, profile: numpy.ndarray, diameter: float | None = None, area: float | None = None, node_id: str | None = None, **kwds)[source]

Bases: SystemEvent, JsonSerializable

Base class for a leakage.

Parameters:

link_id (str) – ID of the link at which the leak is placed. Note that if the leak is placed at a node, then ‘link_id’ must be None and the ID of the node must be set in ‘node_id’
diameter (float, optional) –
Diameter of this leak in either foot or meter (depending on the used flow units).

Alternatively, ‘area’ can be used for specifying the size of this leakage – in this case, ‘diameter’ must be set to ‘None’.

The default is None.
area (float, optional) –
Area of this leak in either foot^2 or meter^2 (depending on the used flow units).

Alternatively, ‘diameter’ can be used for specifying the size of this leakage – in this case, ‘area’ must be set to ‘None’.

The default is None.
profile (numpy.ndarray) – Pattern of this leak.
node_id (str, optional) –
ID of the node at which the leak is placed. This parameter must only be set if the leak is placed at a node instead of a link. In this case, ‘link_id’ must be None.

The default is None.

apply(cur_time: int) → None[source]

Implements the event using EPANET and EPANET-MSX.

This function is only called when the event is active.

Parameters:: cur_time (int) – Current time (seconds since the start) in the simulation.

property area: float

Gets the area of the leak in either foot^2 or meter^2 (depending on the sued flow units).

Returns:: Area of the leak.
Return type:: float

cleanup() → None[source]: Clean up any changes/modifications made by this event.

compute_leak_area(diameter: float) → float[source]

Computes the leak area given the diameter.

leak_area = pi * (diameter * .5)^2

Parameters:: diameter (float) – Diameter (foot or meter) of the leak.
Returns:: Leak area in foot^2 or meter^2.
Return type:: float

compute_leak_emitter_coefficient(area: float, discharge_coef: float = 0.75) → float[source]

Computes the leak emitter coefficient.

emitter_coef = discharge_coef * area * sqrt(2*g) where g is the gravitational constant, and discharge_coef = .75

leak_demand = emitter_coef * pressure^alpha where alpha = .5

Parameters:

area (float) – Leak area (foot^2 or meter^2) as computed in compute_leak_area().
discharge_coef (float, optional) –
Discharge coefficient.

The default is set to 0.75

Returns:

Leak emitter coefficient.

Return type:

float

property diameter: float

Gets the diameter of the leak in either foot or meter (depending on the sued flow units).

Returns:: Diameter (foot or meter) of the leak.
Return type:: float

exit(cur_time) → None[source]

Is called ONCE after the event is over – i.e. next time step after end_time.

Any “clean-up” or “exiting” logic should go here.

Parameters:: cur_time (int) – Current time (seconds since the start) in the simulation.

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 and value.
Return type:: dict

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

Initializes the event.

Parameters:: epanet_api – API to EPANET and EPANET-MSX.

property link_id: str

Gets the ID of the link at which the leak is placed.

Returns:: ID of the link at which the leak is placed.
Return type:: str

property node_id: str

Gets the ID of the node at which the leak is placed.

Returns:: ID of the node at which the leak is placed.
Return type:: str

property profile: numpy.ndarray

Gets the pattern of the leak.

Returns:: Pattern of the leak.
Return type:: numpy.ndarray

reset() → None[source]: Resets this event – i.e. make it ready for another simulation run.

epyt_flow.simulation.events.quality_events

Module provides a class for implementing species injection (e.g. contamination) events.

class epyt_flow.simulation.events.quality_events.SpeciesInjectionEvent(species_id: str, node_id: str, profile: numpy.ndarray, source_type: int, **kwds)[source]

Bases: SystemEvent, JsonSerializable

Class implementing a (bulk) species injection event – e.g. modeling a contamination event.

Parameters:

species_id (str) – ID of the bulk species that is going to be injected.
node_id (str) – ID of the node at which the injection is palced.
profile –
Injection strength profile – i.e. every entry corresponds to the strength of the injection at a point in time. Pattern will repeat if it is shorter than the total injection time.

Note that the pattern time step is equivalent to the EPANET pattern time step.
source_type (int) –
Type of the bulk 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

apply(cur_time: int) → None[source]

Implements the event using EPANET and EPANET-MSX.

This function is only called when the event is active.

Parameters:: cur_time (int) – Current time (seconds since the start) in the simulation.

cleanup() → None[source]: Clean up any changes/modifications made by this event.

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 and value.
Return type:: dict

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

Initializes the event.

Parameters:: epanet_api – API to EPANET and EPANET-MSX.

property node_id: str

Gets the ID of the node at which the injection is palced.

Returns:: Node ID.
Return type:: str

property profile: numpy.ndarray

Gets the injection strength profile.

Returns:: Pattern of the injection.
Return type:: numpy.ndarray

property source_type: int

Type of the bulk species injection source – will be one of the following EPANET toolkit constants:

EN_CONCEN = 0

EN_MASS = 1

EN_SETPOINT = 2

EN_FLOWPACED = 3

Returns:: Type of the injection source.
Return type:: int

property species_id: str

Gets the ID of the bulk species that is going to be injected.

Returns:: Bulk species ID.
Return type:: str

epyt_flow.simulation.events.actuator_events

Module provides implementations of different types of actuator events.

class epyt_flow.simulation.events.actuator_events.ActuatorConstants[source]

Bases: object

Class defining some constants related to actuator events.

EN_CLOSED: Valve or pump is closed.

EN_OPEN: Valve or pump is open – i.e. active.

EN_SET_CLOSED: Link set closed indicator

EN_SET_OPEN: Link set open indicator

class epyt_flow.simulation.events.actuator_events.ActuatorEvent(time: int, **kwds)[source]

Bases: SystemEvent

Base class of an actuator event.

Note

Note that actuator events are one-time events – i.e. they are executed only once at a given point in time.

Parameters:: time (int) – Time (in seconds since simulation start) at which this event is executed.

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 and value.
Return type:: dict

class epyt_flow.simulation.events.actuator_events.PumpEvent(pump_id: str, **kwds)[source]

Bases: ActuatorEvent

Base class of a pump event.

Parameters:: pump_id (str) – ID of the pump that is affected by this event.

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 and value.
Return type:: dict

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

Initializes the event.

Parameters:: epanet_api – API to EPANET and EPANET-MSX.

property pump_id: str

Gets the ID of the pump affected by this event.

Returns:: Pump ID.
Return type:: str

class epyt_flow.simulation.events.actuator_events.PumpSpeedEvent(pump_speed: float, **kwds)[source]

Bases: PumpEvent, JsonSerializable

Class implementing a pump speed event.

Parameters:: pump_speed (float) – New pump speed – i.e. the speed of the pump is set to this value while the event is active.

apply(cur_time: int) → None[source]

Implements the event using EPANET and EPANET-MSX.

This function is only called when the event is active.

Parameters:: cur_time (int) – Current time (seconds since the start) in the simulation.

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 and value.
Return type:: dict

property pump_speed: float

Gets the new pump speed.

Returns:: New pump speed.
Return type:: float

class epyt_flow.simulation.events.actuator_events.PumpStateEvent(pump_state: int, **kwds)[source]

Bases: PumpEvent, JsonSerializable

Class implementing a pump state event.

Parameters:

pump_state (str) –

New state of the pump – i.e. the state of the pump is set to this value while the event is active.

Must be one of the following constants defined in ActuatorConstants:

EN_CLOSED = 0

EN_OPEN = 1

apply(cur_time: int) → None[source]

Implements the event using EPANET and EPANET-MSX.

This function is only called when the event is active.

Parameters:: cur_time (int) – Current time (seconds since the start) in the simulation.

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 and value.
Return type:: dict

property pump_state: int

Gets the new pump state.

Returns:

New pump state.

One of the following constants defined in ActuatorConstants:

EN_CLOSED = 0

EN_OPEN = 1

Return type:

int

class epyt_flow.simulation.events.actuator_events.ValveStateEvent(valve_id: str, valve_state: int, **kwds)[source]

Bases: ActuatorEvent, JsonSerializable

Class implementing a valve state event.

Parameters:

valve_id (str) – ID of the valve that is affected by this event.
valve_state (str) –
New state of the valve – i.e. the valve state is set to this value this event is executed.

Must be one of the following constants defined in ActuatorConstants:
- EN_CLOSED = 0
- EN_OPEN = 1

apply(cur_time: int) → None[source]

Implements the event using EPANET and EPANET-MSX.

This function is only called when the event is active.

Parameters:: cur_time (int) – Current time (seconds since the start) in the simulation.

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 and value.
Return type:: dict

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

Initializes the event.

Parameters:: epanet_api – API to EPANET and EPANET-MSX.

property valve_id: str

Gets the ID of the valve affected by this event.

Returns:: Valve ID.
Return type:: str

property valve_state: int

Gets the new state of the valve.

Returns:

New valve state.

One of the following constants defined in ActuatorConstants:

Return type:

int

epyt_flow.simulation.events.sensor_reading_event

Module provides a base class for sensor reading events such as sensor faults.

class epyt_flow.simulation.events.sensor_reading_event.SensorReadingEvent(sensor_id: str, sensor_type: int, **kwds)[source]

Bases: Event

Base class for a sensor reading event – i.e. an event directly affecting sensor readings.

Parameters:

sensor_id (str) – ID of the sensor that is affected by this event.
sensor_type (int) –
Type of the sensor that is specified in ‘sensor_id’. Must be one of the following pre-defined constants:
- 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_NODE_BULK_SPECIES = 9
- SENSOR_TYPE_NODE_LINK_SPECIES = 10
- SENSOR_TYPE_SURFACE_SPECIES = 11

abstract apply(sensor_readings: numpy.ndarray, sensor_readings_time: numpy.ndarray) → numpy.ndarray[source]

Applies the sensor reading event to sensor reading values – i.e. modify the sensor readings.

Parameters:

sensor_readings – Original sensor readings.
sensor_readings_time – Time (seconds since simulation start) for each sensor reading row in ‘sensor_readings’.

Returns:

Modified sensor readings.

Return type:

numpy.ndarray

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 and value.
Return type:: dict

property sensor_id: str

Gets the ID of the node or link that is affected by this event.

Returns:: Node or link ID.
Return type:: str

property sensor_type: int

Gets the sensor type code.

Returns:: Sensor type code.
Return type:: int

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

Validates this sensor reading event – i.e. checks whether the affected sensor is part of the given sensor configuration.

Parameters:: sensor_config (SensorConfig) – Sensor configuration.

epyt_flow.simulation.events.sensor_faults

Module provides classes for implementing different sensor faults.

class epyt_flow.simulation.events.sensor_faults.SensorFault(sensor_id: str, sensor_type: int, **kwds)[source]

Bases: SensorReadingEvent

Base class for a sensor fault

apply(sensor_readings: numpy.ndarray, sensor_readings_time: numpy.ndarray) → numpy.ndarray[source]

Applies the sensor reading event to sensor reading values – i.e. modify the sensor readings.

Parameters:

sensor_readings – Original sensor readings.
sensor_readings_time – Time (seconds since simulation start) for each sensor reading row in ‘sensor_readings’.

Returns:

Modified sensor readings.

Return type:

numpy.ndarray

abstract apply_sensor_fault(cur_multiplier: float, sensor_reading: float, cur_time: int) → float[source]

Applies this sensor fault to a given single sensor reading value – i.e. the sensor reading value is perturbed by this fault.

Parameters:

cur_multiplierfloat: Current multiplier – i.e. controls the “strength” of the fault.
sensor_readingfloat: Sensor reading value.
cur_timeint: Current time stamp (in seconds) in the simulation.

returns:: Perturbed sensor reading value.
rtype:: float

compute_multiplier(cur_time: int) → float[source]

Computes the multiplier for a given time stamp.

Parameters:: cur_time (int) – Time in seconds.
Returns:: Multiplier.
Return type:: float

class epyt_flow.simulation.events.sensor_faults.SensorFaultConstant(constant_shift: float, **kwds)[source]

Bases: SensorFault, JsonSerializable

Class implementing a constant shift sensor fault.

Parameters:: constant_shift (float) – Constant that is added to the sensor reading.

apply_sensor_fault(cur_multiplier: float, sensor_reading: float, cur_time: int) → float[source]

Applies this sensor fault to a given single sensor reading value – i.e. the sensor reading value is perturbed by this fault.

Parameters:

cur_multiplierfloat: Current multiplier – i.e. controls the “strength” of the fault.
sensor_readingfloat: Sensor reading value.
cur_timeint: Current time stamp (in seconds) in the simulation.

returns:: Perturbed sensor reading value.
rtype:: float

property constant_shift: float

Gets the Constant that is added to the sensor reading.

Returns:: Constant that is added to the sensor reading.
Return type:: float

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 and value.
Return type:: dict

class epyt_flow.simulation.events.sensor_faults.SensorFaultDrift(coef: float, **kwds)[source]

Bases: SensorFault, JsonSerializable

Class implementing a drift sensor fault.

Parameters:: coef (float) – Coefficient of the drift.

apply_sensor_fault(cur_multiplier: float, sensor_reading: float, cur_time: int) → float[source]

Applies this sensor fault to a given single sensor reading value – i.e. the sensor reading value is perturbed by this fault.

Parameters:

cur_multiplierfloat: Current multiplier – i.e. controls the “strength” of the fault.
sensor_readingfloat: Sensor reading value.
cur_timeint: Current time stamp (in seconds) in the simulation.

returns:: Perturbed sensor reading value.
rtype:: float

property coef: float

Gets the coefficient of the drift.

Returns:: Coefficient of the drift.
Return type:: float

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 and value.
Return type:: dict

class epyt_flow.simulation.events.sensor_faults.SensorFaultGaussian(std: float, **kwds)[source]

Bases: SensorFault, JsonSerializable

Class implementing a Gaussian shift sensor fault – i.e.: adding Gaussian noise (centered at zero) to the sensor reading.

Parameters:: std (float) – Standard deviation of the Gaussian noise.

apply_sensor_fault(cur_multiplier: float, sensor_reading: float, cur_time: int) → float[source]

Applies this sensor fault to a given single sensor reading value – i.e. the sensor reading value is perturbed by this fault.

Parameters:

cur_multiplierfloat: Current multiplier – i.e. controls the “strength” of the fault.
sensor_readingfloat: Sensor reading value.
cur_timeint: Current time stamp (in seconds) in the simulation.

returns:: Perturbed sensor reading value.
rtype:: float

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 and value.
Return type:: dict

property std: float

Gets the standard deviation of the Gaussian noise.

Returns:: Standard deviation of the Gaussian noise.
Return type:: float

class epyt_flow.simulation.events.sensor_faults.SensorFaultPercentage(coef: float, **kwds)[source]

Bases: SensorFault, JsonSerializable

Class implementing a percentage shift sensor fault.

Parameters:: coef (float) – Coefficient (percentage) of the shift – i.e. coef must be in (0,].

apply_sensor_fault(cur_multiplier: float, sensor_reading: float, cur_time: int) → float[source]

Applies this sensor fault to a given single sensor reading value – i.e. the sensor reading value is perturbed by this fault.

Parameters:

cur_multiplierfloat: Current multiplier – i.e. controls the “strength” of the fault.
sensor_readingfloat: Sensor reading value.
cur_timeint: Current time stamp (in seconds) in the simulation.

returns:: Perturbed sensor reading value.
rtype:: float

property coef: float

Gets the coefficient (percentage) of the shift.

Returns:: Coefficient (percentage) of the shift.
Return type:: float

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 and value.
Return type:: dict

class epyt_flow.simulation.events.sensor_faults.SensorFaultStuckZero(sensor_id: str, sensor_type: int, **kwds)[source]

Bases: SensorFault, JsonSerializable

Class implementing a stuck-at-zero sensor fault – i.e. sensor reading is set to zero.

apply_sensor_fault(cur_multiplier: float, sensor_reading: float, cur_time: int) → float[source]

Applies this sensor fault to a given single sensor reading value – i.e. the sensor reading value is perturbed by this fault.

Parameters:

cur_multiplierfloat: Current multiplier – i.e. controls the “strength” of the fault.
sensor_readingfloat: Sensor reading value.
cur_timeint: Current time stamp (in seconds) in the simulation.

returns:: Perturbed sensor reading value.
rtype:: float

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

epyt_flow.simulation.events.sensor_reading_attack

This module provides classes for implementing different types of sensor reading attacks.

class epyt_flow.simulation.events.sensor_reading_attack.SensorOverrideAttack(new_sensor_values: numpy.ndarray, **kwds)[source]

Bases: SensorReadingAttack, JsonSerializable

Class implementing a sensor override attack – i.e. sensor reading values are overwritten by pre-defined values.

If the override attack is running out of pre-defined sensor reading values, it repeats the given values from the beginning onwards.

Parameters:: new_sensor_values – New sensor reading values – i.e. these values replace the true sensor reading values.

apply(sensor_readings: numpy.ndarray, sensor_readings_time: numpy.ndarray) → numpy.ndarray[source]

Applies the sensor reading event to sensor reading values – i.e. modify the sensor readings.

Parameters:

sensor_readings – Original sensor readings.
sensor_readings_time – Time (seconds since simulation start) for each sensor reading row in ‘sensor_readings’.

Returns:

Modified sensor readings.

Return type:

numpy.ndarray

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 and value.
Return type:: dict

property new_sensor_values: numpy.ndarray

Returns the new sensor reading values – i.e. these values replace the true sensor reading values.

Returns:: New sensor readings.
Return type:: np.ndarray

class epyt_flow.simulation.events.sensor_reading_attack.SensorReadingAttack(sensor_id: str, sensor_type: int, **kwds)[source]

Bases: SensorReadingEvent

Base class of a sensor reading attack.

class epyt_flow.simulation.events.sensor_reading_attack.SensorReplayAttack(replay_data_time_window_start: int, replay_data_time_window_end: int, **kwds)[source]

Bases: SensorReadingAttack, JsonSerializable

Class implementing a sensor replay attack – i.e. sensor readings are replaced by historical recordings.

If the provided time window of historical recordings is smaller than the time window of the attack, it repeats the historical values from the beginning onwards.

Parameters:

replay_data_time_window_start (int) – Start (seconds since simulation start) of the time window that is used in the replay of sensor readings.
replay_data_time_window_end (int) – End (seconds since simulation start) of the time window that is used in the replay of sensor readings.

apply(sensor_readings: numpy.ndarray, sensor_readings_time: numpy.ndarray) → numpy.ndarray[source]

Applies the sensor reading event to sensor reading values – i.e. modify the sensor readings.

Parameters:

sensor_readings – Original sensor readings.
sensor_readings_time – Time (seconds since simulation start) for each sensor reading row in ‘sensor_readings’.

Returns:

Modified sensor readings.

Return type:

numpy.ndarray

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 and value.
Return type:: dict

property new_sensor_values: numpy.ndarray

Returns the new sensor reading values – i.e. these values replace the true sensor reading values.

Returns:: New sensor readings.
Return type:: np.ndarray

property sensor_data_time_window_end: int

Gets the end time (seconds since simulation start) of the time window that is used in the replay of sensor readings.

Returns:: End time.
Return type:: int

property sensor_data_time_window_start: int

Gets the start time (seconds since simulation start) of the time window that is used in the replay of sensor readings.

Returns:: Start time.
Return type:: int