Arsenic Contamination 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/arsenic_contamination.ipynb"><img src="https://colab.research.google.com/assets/colab-badge.svg" alt="Open In Colab"/></a>'))

This example demonstrates how to add a simple arsenic contamination event to a scenario.

EPyT-Flow is available on PyPI and can be installed via pip install epyt-flow:

%pip install epyt-flow --quiet

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

import numpy as np
from epyt_flow.data.benchmarks import load_leakdb_scenarios
from epyt_flow.simulation import ScenarioSimulator, EpanetConstants, ScenarioConfig
from epyt_flow.simulation.events import SpeciesInjectionEvent
from epyt_flow.utils import to_seconds

Create a new scenario based on the first Net1 LeakDB scenario – we also add an additional EPANET-MSX configuration file arsenic_contamination.msx:

config, = load_leakdb_scenarios(scenarios_id=["1"], use_net1=True, verbose=False)
config = ScenarioConfig(scenario_config=config,
                        f_msx_in="arsenic_contamination.msx")

/tmp/Net1.inp: 100%|##########| 5.90k/5.90k [00:00<00:00, 11.3MB/s]

sim = ScenarioSimulator(scenario_config=config)

Set simulation duration to 21 days by calling set_general_parameters() – note that all time durations have to be stated in seconds, the function to_seconds() can convert minutes/hours/days to seconds:

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

Place some chlorine sensors and also keep track of the contaminant:

cl_sensor_locations = ["10", "11", "12", "13", "21", "22", "23", "31", "32"]
all_nodes = sim.sensor_config.nodes

sim.set_bulk_species_node_sensors({"Chlorine": cl_sensor_locations,
                                   # Also: Keep track of the contaminant
                                   "AsIII": all_nodes})   # Arsenite

Create a 1-day contamination event of an Arsenite injection (100 mg/day) at node “22” by creating a new instance of the SpeciesInjectionEvent class and calling add_system_event():

contamination_event = SpeciesInjectionEvent(species_id="AsIII", node_id="22",
                                            profile=np.array([100]),
                                            source_type=EpanetConstants.EN_MASS,
                                            start_time=to_seconds(days=3),
                                            end_time=to_seconds(days=4))

sim.add_system_event(contamination_event)

Run the entire simulation by calling run_simulation():

scada_data = sim.run_simulation()

Inspect simulation results – i.e. plot sensor readings over time by calling the plot_bulk_species_node_concentration function:

scada_data.plot_bulk_species_node_concentration({"Chlorine": cl_sensor_locations})

<Axes: xlabel='Time (30min steps)', ylabel='Concentration in $MG/L$'>

scada_data.plot_bulk_species_node_concentration({"AsIII": all_nodes})

<Axes: xlabel='Time (30min steps)', ylabel='Concentration in $MG/L$'>

Do not forget to close the simulation by calling close():

sim.close()