1"""
2The module contains classes for representing simple control rules as used in EPANET.
3"""
4from typing import Union
5from epanet_plus import EpanetConstants
6
7from ..events import ActuatorConstants
8from ...serialization import JsonSerializable, SIMPLE_CONTROL_ID, serializable
9
10
[docs]
11@serializable(SIMPLE_CONTROL_ID, ".epytflow_simple_control")
12class SimpleControlModule(JsonSerializable):
13 """
14 A class for representing a simple EPANET control rule.
15
16 Parameters
17 ----------
18 link_id : `str`
19 Link ID.
20 link_status : `int` or `float`
21 Status of the link that is set when the condition is fullfilled.
22
23 Instance of `float` if the link constitutes a pump -- in this case,
24 the argument corresponds to the pump speed.
25
26 Instance of `int` if the link constitutes a valve -- in this case,
27 must be one of the followig costants defined in
28 :class:`~epyt_flow.simulation.events.actuator_events.ActuatorConstants`:
29
30 - EN_CLOSED = 0
31 - EN_OPEN = 1
32 - EN_SET_CLOSED = -1e10
33 - EN_SET_OPEN = 1e10
34 cond_type : `int`
35 Condition/Rule type.
36
37 Must be one of the following EPANET constants:
38
39 - EN_LOWLEVEL = 0
40 - EN_HILEVEL = 1
41 - EN_TIMER = 2
42 - EN_TIMEOFDAY = 3
43 cond_var_value : `str` or `int`
44 Condition/Rule variable value.
45
46 Node ID in the cases of EN_LOWLEVEL or EN_HILEVEL.
47 Time of the day (in AM/PM format) in the case of EN_TIMEOFDAY.
48 Number of hours (as an integer) since simulation start in the case of EN_TIMER.
49 cond_comp_value : `float`
50 The condition/rule comparison value at which this control rule is triggered.
51
52 Lower or upper value on the pressure (or tank level) in the cases of
53 EN_LOWLEVEL and EN_HILEVEL.
54
55 Will be ignored in all other cases -- i.e. should be set to None.
56 """
57 def __init__(self, link_id: str, link_status: Union[int, float], cond_type: int,
58 cond_var_value: Union[str, int], cond_comp_value: float,
59 **kwds):
60 if not isinstance(link_id, str):
61 raise TypeError(f"'link_id' must be an instance of 'str' but not of '{type(link_id)}'")
62 if not isinstance(link_status, int) and not isinstance(link_status, float):
63 raise TypeError("'link_status' must be an instance of 'int' or 'float' but not " +
64 f"of '{type(link_status)}'")
65 if link_status not in [ActuatorConstants.EN_OPEN, ActuatorConstants.EN_CLOSED,
66 ActuatorConstants.EN_SET_OPEN, ActuatorConstants.EN_SET_CLOSED]:
67 if link_status < 0:
68 raise TypeError("'link_status' can not be negative")
69 if cond_type not in [EpanetConstants.EN_TIMEOFDAY, EpanetConstants.EN_TIMER,
70 EpanetConstants.EN_LOWLEVEL, EpanetConstants.EN_HILEVEL]:
71 raise ValueError(f"Invalid control type '{cond_type}' in 'cond_type'")
72
73 if cond_type == EpanetConstants.EN_TIMEOFDAY:
74 if not isinstance(cond_var_value, str):
75 raise TypeError("EN_TIMEOFDAY requires that 'cond_var_value' must be an instance " +
76 f"of 'str' but not of '{type(cond_var_value)}'")
77 if not cond_var_value.endswith("AM") and not cond_var_value.endswith("PM"):
78 raise ValueError(f"Invalid time of day format '{cond_var_value}' in " +
79 "'cond_var_value'")
80 elif cond_type == EpanetConstants.EN_TIMER:
81 if not isinstance(cond_var_value, int):
82 raise TypeError("EN_TIMER requires that 'cond_var_value' must be an instance " +
83 f"of 'int' but not of '{type(cond_var_value)}'")
84 if cond_var_value < 0:
85 raise ValueError("'cond_var_value' can not be negative")
86 else:
87 if not isinstance(cond_var_value, str):
88 raise TypeError("'cond_var_value' must be an instance of 'str' but " +
89 f"not of '{type(cond_var_value)}'")
90 if not isinstance(cond_comp_value, float):
91 raise TypeError("'cond_comp_value' must be an instance of 'float' " +
92 f"but not of '{type(cond_comp_value)}'")
93 if cond_comp_value < 0:
94 raise ValueError("'cond_comp_value' can not be negative")
95
96 self._link_id = link_id
97 self._link_status = link_status
98 self._cond_type = cond_type
99 self._cond_var_value = cond_var_value
100 self._cond_comp_value = cond_comp_value
101
102 super().__init__(**kwds)
103
104 @property
105 def link_id(self) -> str:
106 """
107 Returns the link ID.
108
109 Returns
110 -------
111 `str`
112 Link ID.
113 """
114 return self._link_id
115
116 @property
117 def link_status(self) -> Union[int, float]:
118 """
119 Returns the link status that is set when the condition is fullfilled.
120
121 Returns
122 -------
123 `int` or `float`
124 Pump speed if the link is a pump, otherwise one of the followig constants defined in
125 :class:`~epyt_flow.simulation.events.actuator_events.ActuatorConstants`:
126
127 - EN_CLOSED = 0
128 - EN_OPEN = 1
129 - EN_SET_CLOSED = -1e10
130 - EN_SET_OPEN = 1e10
131
132 """
133 return self._link_status
134
135 @property
136 def cond_type(self) -> int:
137 """
138 Returns the condition/rule type.
139
140 Returns
141 -------
142 `int`
143 Condition/Rule type -- will be one of the following EPANET constants:
144
145 - EN_LOWLEVEL = 0
146 - EN_HILEVEL = 1
147 - EN_TIMER = 2
148 - EN_TIMEOFDAY = 3
149 """
150 return self._cond_type
151
152 @property
153 def cond_var_value(self) -> Union[str, int]:
154 """
155 Return the condition/rule variable value.
156
157 Node ID in the cases of EN_LOWLEVEL or EN_HILEVEL.
158 Time of the day (in AM/PM format) in the case of EN_TIMEOFDAY.
159 Number of hours (as an integer) since simulation start in the case of EN_TIMER.
160
161 Returns
162 -------
163 `str` or `int`
164 Condition/rule variable value.
165 """
166 return self._cond_var_value
167
168 @property
169 def cond_comp_value(self) -> float:
170 """
171 Returns the condition/rule comparison value -- might be None if not needed.
172
173 Lower or upper value on the pressure (or tank level) in the cases of
174 EN_LOWLEVEL and EN_HILEVEL.
175
176 Returns
177 -------
178 `float`
179 Condition/Rule comparison value.
180 """
181 return self._cond_comp_value
182
[docs]
183 def get_attributes(self) -> dict:
184 return super().get_attributes() | {"link_id": self._link_id,
185 "link_status": self._link_status,
186 "cond_type": self._cond_type,
187 "cond_var_value": self._cond_var_value,
188 "cond_comp_value": self._cond_comp_value}
189
190 def __eq__(self, other) -> bool:
191 return super().__eq__(other) and self._link_id == other.link_id and \
192 self._link_status == other.link_status and self._cond_type == other.cond_type and \
193 self._cond_var_value == other.cond_var_value and \
194 self._cond_comp_value == other.cond_comp_value
195
196 def __str__(self) -> str:
197 control_rule_str = f"LINK {self._link_id} "
198 if isinstance(self._link_status, int):
199 control_rule_str += "OPEN " if self._link_status == ActuatorConstants.EN_OPEN or \
200 self._link_status == ActuatorConstants.EN_SET_OPEN else "CLOSED "
201 else:
202 control_rule_str += f"{self._link_status} "
203
204 if self._cond_type == EpanetConstants.EN_TIMER:
205 control_rule_str += f"AT TIME {self._cond_var_value}"
206 elif self._cond_type == EpanetConstants.EN_TIMEOFDAY:
207 control_rule_str += f"AT CLOCKTIME {self._cond_var_value}"
208 elif self._cond_type == EpanetConstants.EN_LOWLEVEL:
209 control_rule_str += f"IF NODE {self._cond_var_value} BELOW {self._cond_comp_value}"
210 elif self._cond_type == EpanetConstants.EN_HILEVEL:
211 control_rule_str += f"IF NODE {self._cond_var_value} ABOVE {self._cond_comp_value}"
212
213 return control_rule_str
214
215
[docs]
216class SimplePumpSpeedTimeControl(SimpleControlModule):
217 """
218 A class for representing a simple control rule for setting the pump speed at some point in time.
219
220 Parameters
221 ----------
222 pump_id : `str`
223 Pump ID.
224 pump_speed : `float`
225 Pump speed.
226 time : `str` or `int`
227 Time of the day (in AM/PM format) in the case or
228 number of hours (as an integer) since simulation start.
229 """
230 def __init__(self, pump_id: str, pump_speed: float, time: Union[str, int]):
231 super().__init__(link_id=pump_id, link_status=pump_speed,
232 cond_type=EpanetConstants.EN_TIMER if isinstance(time, int)
233 else EpanetConstants.EN_TIMEOFDAY,
234 cond_var_value=time, cond_comp_value=None)
235
236
[docs]
237class SimplePumpSpeedConditionControl(SimpleControlModule):
238 """
239 A class for representing a simple IF-THEN control rule for setting the pump speed.
240
241 Parameters
242 ----------
243 Parameters
244 ----------
245 pump_id : `str`
246 Pump ID.
247 pump_speed : `float`
248 Pump speed.
249 node_id : `str`
250 Node ID.
251 comp_type : `int`
252 Comparison type -- must be one of the following EPANET constants:
253
254 - EN_LOWLEVEL = 0
255 - EN_HILEVEL = 1
256 comp_value : `float`:
257 Lower or upper value on the pressure (or tank level) at which this
258 control rule is triggered.
259 """
260 def __init__(self, pump_id: str, pump_speed: float, node_id: str, comp_type: int,
261 comp_value: float):
262 super().__init__(link_id=pump_id, link_status=pump_speed, cond_type=comp_type,
263 cond_var_value=node_id, cond_comp_value=comp_value)
264
265
[docs]
266class SimpleValveTimeControl(SimpleControlModule):
267 """
268 A class for representing a simple control rule for setting the valve status
269 at some point in time.
270
271 Parameters
272 ----------
273 valve_id : `str`
274 valve ID.
275 valve_status : `int`
276 Valve status -- must be one of the followig costants defined in
277 :class:`~epyt_flow.simulation.events.actuator_events.ActuatorConstants`:
278
279 - EN_CLOSED = 0
280 - EN_OPEN = 1
281 - EN_SET_CLOSED = -1e10
282 - EN_SET_OPEN = 1e10
283 time : `str` or `int`
284 Time of the day (in AM/PM format) in the case or
285 number of hours (as an integer) since simulation start.
286 """
287 def __init__(self, valve_id: str, valve_status: int, time: Union[str, int]):
288 super().__init__(link_id=valve_id, link_status=valve_status,
289 cond_type=EpanetConstants.EN_TIMER if isinstance(time, int)
290 else EpanetConstants.EN_TIMEOFDAY,
291 cond_var_value=time, cond_comp_value=None)
292
293
[docs]
294class SimpleValveConditionControl(SimpleControlModule):
295 """
296 A class for representing a simple IF-THEN control rule for setting the valve status.
297
298 Parameters
299 ----------
300 valve_id : `str`
301 valve ID.
302 valve_status : `int`
303 Valve status -- must be one of the followig costants defined in
304 :class:`~epyt_flow.simulation.events.actuator_events.ActuatorConstants`:
305
306 - EN_CLOSED = 0
307 - EN_OPEN = 1
308 - EN_SET_CLOSED = -1e10
309 - EN_SET_OPEN = 1e10
310 node_id : `str`
311 Node ID.
312 comp_type : `int`
313 Comparison type -- must be one of the following EPANET constants:
314
315 - EN_LOWLEVEL = 0
316 - EN_HILEVEL = 1
317 comp_value : `float`:
318 Lower or upper value on the pressure (or tank level) at which this
319 control rule is triggered.
320 """
321 def __init__(self, valve_id: str, valve_status: int, node_id: str, comp_type: int,
322 comp_value: float):
323 super().__init__(link_id=valve_id, link_status=valve_status, cond_type=comp_type,
324 cond_var_value=node_id, cond_comp_value=comp_value)