Sensor Replay Attack Example

from IPython.display import display, HTML
display(HTML('<a target="_blank" href="https://colab.research.google.com/github/WaterFutures/EPyT-Flow/blob/main/docs/examples/sensor_replay_attack.ipynb"><img src="https://colab.research.google.com/assets/colab-badge.svg" alt="Open In Colab"/></a>'))

This example demonstrates how to add a sensor replay attack to a scenario – more information can be found in the documentation.

%pip install epyt-flow --quiet

Note: you may need to restart the kernel to use updated packages.

from epyt_flow.data.benchmarks import load_leakdb_scenarios
from epyt_flow.simulation import ScenarioSimulator, SENSOR_TYPE_NODE_PRESSURE
from epyt_flow.simulation.events import SensorReplayAttack
from epyt_flow.utils import to_seconds, plot_timeseries_data

Load the first LeakDB Hanoi scenario:

config = load_leakdb_scenarios(scenarios_id=["1"], use_net1=False, verbose=False)[0]

Create a new simulation:

sim = ScenarioSimulator(scenario_config=config)

Set the simulation duration to two days:

sim.set_general_parameters(simulation_duration=to_seconds(days=2))

Add a sensor replay attack – pressure readings at node “13” between 5hrs and 7hrs after simulation start (time steps 10 - 15) are replaced by the historical readings collected from the first 150min (i.e. first 5 time steps).

This is done by creating an instance of the SensorReplayAttack class and adding it to the scenarion by calling add_sensor_reading_event():

sensor_replay_attack = SensorReplayAttack(replay_data_time_window_start=0,
                                          replay_data_time_window_end=to_seconds(minutes=150),
                                          start_time=to_seconds(hours=5),
                                          end_time=to_seconds(hours=7),
                                          sensor_id="13",
                                          sensor_type=SENSOR_TYPE_NODE_PRESSURE)

sim.add_sensor_reading_event(sensor_replay_attack)

Run the entire simulation:

scada_data = sim.run_simulation()

Retrieve and plot the pressure readings:

pressure_readings = scada_data.get_data_pressures(sensor_locations=["13"])

scada_data.plot_pressures(sensor_locations=["13"])

<Axes: xlabel='Time (30min steps)', ylabel='Pressure in $meter$'>

Readings for time 10 - 15 are same as the first 5 readings!

plot_timeseries_data(pressure_readings[:20].T,    # The same as the first 5 readings!
                     x_axis_label="Time (30min steps)",
                     y_axis_label="Pressure in $m$")

<Axes: xlabel='Time (30min steps)', ylabel='Pressure in $m$'>

Sensor readings WITHOUT the sensor replay attack

# Remove sensor replay attack
scada_data.sensor_reading_events = []

# Recompute and show final sensor readings
pressure_readings = scada_data.get_data_pressures(sensor_locations=["13"])
plot_timeseries_data(pressure_readings[:20].T,
                     x_axis_label="Time (30min steps)",
                     y_axis_label="Pressure in $m$")

<Axes: xlabel='Time (30min steps)', ylabel='Pressure in $m$'>

Do not forget to close the simulation!

sim.close()