1"""
2Module provides a class for implementing species injection (e.g. contamination) events.
3"""
4from copy import deepcopy
5import warnings
6import math
7import numpy as np
8from epanet_plus import EPyT, EpanetConstants
9
10from .system_event import SystemEvent
11from ...serialization import serializable, JsonSerializable, \
12 SPECIESINJECTION_EVENT_ID
13
14
[docs]
15@serializable(SPECIESINJECTION_EVENT_ID, ".epytflow_speciesinjection_event")
16class SpeciesInjectionEvent(SystemEvent, JsonSerializable):
17 """
18 Class implementing a (bulk) species injection event -- e.g. modeling a contamination event.
19
20 Parameters
21 ----------
22 species_id : `str`
23 ID of the bulk species that is going to be injected.
24 node_id : `str`
25 ID of the node at which the injection is palced.
26 profile : `numpy.ndarray <https://numpy.org/doc/stable/reference/generated/numpy.ndarray.html>`_
27 Injection strength profile -- i.e. every entry corresponds to the strength of the injection
28 at a point in time. Pattern will repeat if it is shorter than the total injection time.
29
30 Note that the pattern time step is equivalent to the EPANET pattern time step.
31 source_type : `int`
32 Type of the bulk species injection source -- must be one of
33 the following EPANET constants:
34
35 - EN_CONCEN = 0
36 - EN_MASS = 1
37 - EN_SETPOINT = 2
38 - EN_FLOWPACED = 3
39
40 Description:
41
42 - E_CONCEN Sets the concentration of external inflow entering a node
43 - EN_MASS Injects a given mass/minute into a node
44 - EN_SETPOINT Sets the concentration leaving a node to a given value
45 - EN_FLOWPACED Adds a given value to the concentration leaving a node
46 """
47 def __init__(self, species_id: str, node_id: str, profile: np.ndarray, source_type: int,
48 **kwds):
49 if not isinstance(species_id, str):
50 raise TypeError("'species_id' must be an instance of 'str' but not of " +
51 f"'{type(species_id)}'")
52 if not isinstance(node_id, str):
53 raise TypeError("'node_id' must be an instance of 'str' but not of " +
54 f"'{type(node_id)}'")
55 if not isinstance(profile, np.ndarray):
56 raise TypeError("'profile' must be an instance of 'numpy.ndarray' but not of " +
57 f"'{type(profile)}'")
58 if not isinstance(source_type, int):
59 raise TypeError("'source_type' must be an instance of 'int' but not of " +
60 f"'{type(source_type)}'")
61 if not 0 <= source_type <= 3:
62 raise ValueError("'source_tye' must be in [0, 3]")
63
64 self._species_id = species_id
65 self._node_id = node_id
66 self._profile = profile
67 self._source_type = source_type
68
69 super().__init__(**kwds)
70
71 @property
72 def species_id(self) -> str:
73 """
74 Gets the ID of the bulk species that is going to be injected.
75
76 Returns
77 -------
78 `str`
79 Bulk species ID.
80 """
81 return self._species_id
82
83 @property
84 def node_id(self) -> str:
85 """
86 Gets the ID of the node at which the injection is palced.
87
88 Returns
89 -------
90 `str`
91 Node ID.
92 """
93 return self._node_id
94
95 @property
96 def profile(self) -> np.ndarray:
97 """
98 Gets the injection strength profile.
99
100 Returns
101 -------
102 `numpy.ndarray`
103 Pattern of the injection.
104 """
105 return deepcopy(self._profile)
106
107 @property
108 def source_type(self) -> int:
109 """
110 Type of the bulk species injection source -- will be one of
111 the following EPANET toolkit constants:
112
113 - EN_CONCEN = 0
114 - EN_MASS = 1
115 - EN_SETPOINT = 2
116 - EN_FLOWPACED = 3
117
118 Returns
119 -------
120 `int`
121 Type of the injection source.
122 """
123 return self._source_type
124
[docs]
125 def get_attributes(self) -> dict:
126 return super().get_attributes() | {"species_id": self._species_id,
127 "node_id": self._node_id, "profile": self._profile,
128 "source_type": self._source_type}
129
130 def __eq__(self, other) -> bool:
131 return super().__eq__(other) and self._species_id == other.species_id and \
132 self._node_id == other.node_id and np.all(self._profile == other.profile) and \
133 self._source_type == other.source_type
134
135 def __str__(self) -> str:
136 return f"{super().__str__()} species_id: {self._species_id} " +\
137 f"node_id: {self._node_id} profile: {self._profile} source_type: {self._source_type}"
138
139 def _get_pattern_id(self) -> str:
140 return f"{self._species_id}_{self._node_id}"
141
[docs]
142 def init(self, epanet_api: EPyT) -> None:
143 super().init(epanet_api)
144
145 # Check parameters
146 if self._species_id not in self._epanet_api.get_all_msx_species_id():
147 raise ValueError(f"Unknown species '{self._species_id}'")
148 if self._node_id not in self._epanet_api.get_all_nodes_id():
149 raise ValueError(f"Unknown node '{self._node_id}'")
150
151 # Create final injection strength pattern
152 total_sim_duration = self._epanet_api.get_simulation_duration()
153 time_step = self._epanet_api.gettimeparam(EpanetConstants.EN_PATTERNSTEP)
154
155 pattern = np.zeros(math.ceil(total_sim_duration / time_step))
156
157 end_time = self.end_time if self.end_time is not None else total_sim_duration
158 injection_pattern_length = math.ceil((end_time - self.start_time) / time_step)
159 injection_time_start_idx = int(self.start_time / time_step)
160
161 injection_pattern = None
162 if len(self._profile) == injection_pattern_length:
163 injection_pattern = self.profile
164 else:
165 injection_pattern = np.tile(self.profile,
166 math.ceil(injection_pattern_length / len(self.profile)))
167
168 pattern[injection_time_start_idx:
169 injection_time_start_idx + injection_pattern_length] = injection_pattern
170
171 # Create injection
172 pattern_id = self._get_pattern_id()
173 if pattern_id in [self._epanet_api.MSXgetID(EpanetConstants.MSX_PATTERN, pattern_idx + 1)
174 for pattern_idx in
175 range(self._epanet_api.MSXgetcount(EpanetConstants.MSX_PATTERN))]:
176 node_idx = self._epanet_api.get_node_idx(self._node_id)
177 species_idx = self._epanet_api.get_msx_species_idx(self._species_id)
178 cur_source_type = self._epanet_api.MSXgetsource(node_idx, species_idx)
179 if cur_source_type[0] != self._source_type:
180 raise ValueError("Source type does not match existing source type")
181
182 # Add new injection amount to existing injection --
183 # i.e. two injection events at the same node
184 pattern_idx = self._epanet_api.MSXgetindex(EpanetConstants.MSX_PATTERN, pattern_id)
185 cur_pattern = self._epanet_api.get_msx_pattern(pattern_idx)
186 cur_pattern = np.array(cur_pattern) + pattern
187 self._epanet_api.MSXsetpattern(pattern_idx, cur_pattern.tolist(), len(cur_pattern))
188 else:
189 self._epanet_api.add_msx_pattern(pattern_id, pattern.tolist())
190 self._epanet_api.set_msx_source(self._node_id, self._species_id, self._source_type, 1,
191 pattern_id)
192
[docs]
193 def cleanup(self) -> None:
194 warnings.warn("Can not undo SpeciesInjectionEvent -- " +
195 "EPANET-MSX does not support removing patterns")
196
[docs]
197 def apply(self, cur_time: int) -> None:
198 pass