1"""
2Module contains a class for representing complex control rules as implemented in EPANET.
3"""
4from copy import deepcopy
5from typing import Any
6import numpy as np
7from epanet_plus import EpanetConstants
8
9from ...serialization import JsonSerializable, COMPLEX_CONTROL_ID, COMPLEX_CONTROL_CONDITION_ID, \
10 COMPLEX_CONTROL_ACTION_ID, serializable
11
12
13EN_R_AND = 2
14EN_R_OR = 3
15
16EN_R_DEMAND = 1
17EN_R_HEAD = 2
18EN_R_LEVEL = 3
19EN_R_PRESSURE = 4
20EN_R_FLOW = 5
21EN_R_STATUS = 6
22EN_R_SETTING = 7
23EN_R_POWER = 8
24EN_R_TIME = 9
25EN_R_CLOCKTIME = 10
26EN_R_FILLTIME = 11
27EN_R_DRAINTIME = 12
28
29EN_R_EQ = 0
30EN_R_NEQ = 1
31EN_R_LEQ = 2
32EN_R_GEQ = 3
33EN_R_LESS = 4
34EN_R_GREATER = 5
35EN_R_IS = 6
36EN_R_NOT = 7
37EN_R_BELOW = 8
38EN_R_ABOVE = 9
39
40EN_R_ACTION_SETTING = -1
41EN_R_ACTION_STATUS_OPEN = 1
42EN_R_ACTION_STATUS_CLOSED = 2
43EN_R_ACTION_STATUS_ACTIVE = 3
44
45RULESTATUS = ['OPEN', 'CLOSED', 'ACTIVE']
46
47
[docs]
48@serializable(COMPLEX_CONTROL_CONDITION_ID, ".epytflow_complex_control_condition")
49class RuleCondition(JsonSerializable):
50 """
51 Class representing a rule condition.
52
53 Parameters
54 ----------
55 object_type_id : `int`
56 ID of the object type.
57
58 Must be one of the following EPANET constants:
59
60 - EN_R_NODE = 6
61 - EN_R_LINK = 7
62 - EN_R_SYSTEM = 8
63 object_id : `str`
64 ID of the object (i.e. junction, pipe, link, tank, etc.).
65 attribute_id : `int`
66 Type ID of the object's attribute that is checked.
67
68 Must be on of the following constants:
69
70 - EN_R_DEMAND = 1
71 - EN_R_HEAD = 2
72 - EN_R_LEVEL = 3
73 - EN_R_PRESSURE = 4
74 - EN_R_FLOW = 5
75 - EN_R_STATUS = 6
76 - EN_R_SETTING = 7
77 - EN_R_TIME = 9
78 - EN_R_CLOCKTIME = 10
79 - EN_R_FILLTIME = 11
80 - EN_R_DRAINTIME = 12
81 relation_type_id : `int`
82 ID of the type of comparison.
83
84 Must be one of the following constants:
85
86 - EN_R_EQ = 0
87 - EN_R_NEQ = 1
88 - EN_R_LEQ = 2
89 - EN_R_GEQ = 3
90 - EN_R_LESS = 4
91 - EN_R_GREATER = 5
92 - EN_R_IS = 6
93 - EN_R_NOT = 7
94 - EN_R_BELOW = 8
95 - EN_R_ABOVE = 9
96 value : `Any`
97 Value that is compared against.
98 """
99 def __init__(self, object_type_id: int, object_id: str, attribute_id: int,
100 relation_type_id: int, value: Any, **kwds):
101 if not isinstance(object_type_id, int):
102 raise TypeError("'object_type_id' must be an instance of 'int' " +
103 f"but not of '{type(object_type_id)}'")
104 if object_type_id not in [EpanetConstants.EN_R_NODE, EpanetConstants.EN_R_LINK,
105 EpanetConstants.EN_R_SYSTEM]:
106 raise ValueError(f"Invalid value '{object_type_id}' for 'object_type_id'")
107 if not isinstance(object_id, str):
108 raise TypeError("'object_id' must be an instance of 'str' " +
109 f"but not of '{type(object_id)}'")
110 if not isinstance(attribute_id, int):
111 raise TypeError("'attribute_id' must be an instance of 'int' " +
112 f"but not of '{type(attribute_id)}'")
113 if attribute_id not in [EN_R_DEMAND, EN_R_HEAD, EN_R_LEVEL, EN_R_PRESSURE,
114 EN_R_FLOW, EN_R_STATUS, EN_R_SETTING, EN_R_POWER, EN_R_TIME,
115 EN_R_CLOCKTIME, EN_R_FILLTIME, EN_R_DRAINTIME]:
116 raise ValueError(f"Invalid value '{attribute_id}' for 'attribute_id'")
117 if not isinstance(relation_type_id, int):
118 raise TypeError("'relation_type_id' must be an instance of 'int' " +
119 f"but not of '{type(relation_type_id)}'")
120 if relation_type_id not in [EN_R_EQ, EN_R_NEQ, EN_R_LEQ, EN_R_GEQ, EN_R_LESS, EN_R_GREATER,
121 EN_R_IS, EN_R_NOT, EN_R_BELOW, EN_R_ABOVE]:
122 raise ValueError(f"Invalid value '{relation_type_id}' for 'relation_type_id'")
123
124 self._object_type_id = object_type_id
125 self._object_id = object_id
126 self._attribute_id = attribute_id
127 self._relation_type_id = relation_type_id
128 self._value = value
129
130 super().__init__(**kwds)
131
132 @property
133 def object_type_id(self) -> int:
134 """
135 Returns the ID of the object type.
136
137 Will be one of the following EPANET constants:
138
139 - EN_R_NODE = 6
140 - EN_R_LINK = 7
141 - EN_R_SYSTEM = 8
142
143 Returns
144 -------
145 `int`
146 ID of the object type..
147 """
148 return self._object_type_id
149
150 @property
151 def object_id(self) -> str:
152 """
153 Returns the ID of the object (i.e. junction, pipe, link, tank, etc.).
154
155 Returns
156 -------
157 `str`
158 ID of the object.
159 """
160 return self._object_id
161
162 @property
163 def attribute_id(self) -> int:
164 """
165 Returns the type ID of the object's attribute that is checked.
166
167 Will be one of the following constants:
168
169 - EN_R_DEMAND = 1
170 - EN_R_HEAD = 2
171 - EN_R_LEVEL = 3
172 - EN_R_PRESSURE = 4
173 - EN_R_FLOW = 5
174 - EN_R_STATUS = 6
175 - EN_R_SETTING = 7
176 - EN_R_TIME = 9
177 - EN_R_CLOCKTIME = 10
178 - EN_R_FILLTIME = 11
179 - EN_R_DRAINTIME = 12
180
181 Returns
182 -------
183 `int`
184 Type ID of the object's attribute that is checked.
185 """
186 return self._attribute_id
187
188 @property
189 def relation_type_id(self) -> int:
190 """
191 Returns the ID of the type of comparison.
192
193 Will be one of the following constants:
194
195 - EN_R_EQ = 0
196 - EN_R_NEQ = 1
197 - EN_R_LEQ = 2
198 - EN_R_GEQ = 3
199 - EN_R_LESS = 4
200 - EN_R_GREATER = 5
201 - EN_R_IS = 6
202 - EN_R_NOT = 7
203 - EN_R_BELOW = 8
204 - EN_R_ABOVE = 9
205
206 Returns
207 -------
208 `int`
209 ID of the type of comparison.
210 """
211 return self._relation_type_id
212
213 @property
214 def value(self) -> Any:
215 """
216 Returns the value that is compared against.
217
218 Returns
219 -------
220 `Any`
221 Value that is compared against.
222 """
223 return self._value
224
[docs]
225 def get_attributes(self) -> dict:
226 return super().get_attributes() | {"object_type_id": self._object_type_id,
227 "object_id": self._object_id,
228 "attribute_id": self._attribute_id,
229 "relation_type_id": self._relation_type_id,
230 "value": self._value}
231
232 def __eq__(self, other) -> bool:
233 return self._object_type_id == other.object_type_id and \
234 self._object_id == other.object_id and self._attribute_id == other.attribute_id and \
235 self._relation_type_id == other.relation_type_id and self._value == other.value
236
237 def __str__(self) -> str:
238 desc = ""
239
240 if self._attribute_id == EN_R_DEMAND:
241 if self._object_type_id == EpanetConstants.EN_R_NODE:
242 desc += f"JUNCTION {self._object_id} DEMAND "
243 elif self._object_type_id == EpanetConstants.EN_R_SYSTEM:
244 desc += "SYSTEM DEMAND "
245 elif self._attribute_id == EN_R_HEAD:
246 desc += f"JUNCTION {self._object_id} HEAD "
247 elif self._attribute_id == EN_R_LEVEL:
248 desc += f"TANK {self._object_id} LEVEL "
249 elif self._attribute_id == EN_R_PRESSURE:
250 desc += f"JUNCTION {self._object_id} PRESSURE "
251 elif self._attribute_id == EN_R_FLOW:
252 desc += f"LINK {self._object_id} FLOW "
253 elif self._attribute_id == EN_R_STATUS:
254 desc += f"LINK {self._object_id} STATUS "
255 elif self._attribute_id == EN_R_SETTING:
256 desc += f"LINK {self._object_id} SETTING "
257 elif self._attribute_id == EN_R_TIME:
258 desc += "SYSTEM TIME "
259 elif self._attribute_id == EN_R_CLOCKTIME:
260 desc += "SYSTEM CLOCKTIME "
261 elif self._attribute_id == EN_R_FILLTIME:
262 desc += f"TANK {self._object_id} FILLTIME "
263 elif self._attribute_id == EN_R_DRAINTIME:
264 desc += f"TANK {self._object_id} DRAINTIME "
265
266 if self._relation_type_id == EN_R_EQ:
267 desc += "= "
268 elif self._relation_type_id == EN_R_IS:
269 desc += "IS "
270 elif self._relation_type_id == EN_R_NOT:
271 desc += "IS NOT "
272 elif self._relation_type_id == EN_R_LEQ:
273 desc += "<= "
274 elif self._relation_type_id == EN_R_GEQ:
275 desc += ">= "
276 elif self._relation_type_id == EN_R_ABOVE:
277 desc += "ABOVE "
278 elif self._relation_type_id == EN_R_BELOW:
279 desc += "BELOW "
280 elif self._relation_type_id == EN_R_LESS:
281 desc += "< "
282 elif self._relation_type_id == EN_R_GREATER:
283 desc += "> "
284
285 desc += str(self._value)
286
287 return desc
288
289
[docs]
290@serializable(COMPLEX_CONTROL_ACTION_ID, ".epytflow_complex_control_action")
291class RuleAction(JsonSerializable):
292 """
293 Class representing a rule action.
294
295 Parameters
296 ----------
297 link_type_id : `int`
298 Link type ID.
299
300 Must be one of following EPANET constants:
301
302 - EN_CVPIPE = 0
303 - EN_PIPE = 1
304 - EN_PUMP = 2
305 - EN_PRV = 3
306 - EN_PSV = 4
307 - EN_PBV = 5
308 - EN_FCV = 6
309 - EN_TCV = 7
310 - EN_GPV = 8
311 link_id : `str`
312 Link ID.
313 action_type_id : `int`
314 Type ID of the action.
315
316 Must be one of the following constants:
317
318 - EN_R_ACTION_SETTING = -1
319 - EN_R_ACTION_STATUS_OPEN = 1
320 - EN_R_ACTION_STATUS_CLOSED = 2
321 - EN_R_ACTION_STATUS_ACTIVE = 3
322 action_value : `Any`
323 Value of the acton (e.g. pump speed).
324 Only relevant if action_type_id = EN_R_SETTING, will be ignored in all other cases.
325 """
326 def __init__(self, link_type_id: int, link_id: str, action_type_id: int, action_value: Any,
327 **kwds):
328 if not isinstance(link_type_id, int):
329 raise TypeError("'link_type_id' must be an istanace of 'int' " +
330 f"but not of '{type(link_type_id)}'")
331 if link_type_id not in [EpanetConstants.EN_CVPIPE, EpanetConstants.EN_PIPE,
332 EpanetConstants.EN_PUMP, EpanetConstants.EN_PRV,
333 EpanetConstants.EN_PSV, EpanetConstants.EN_PBV,
334 EpanetConstants.EN_FCV, EpanetConstants.EN_TCV,
335 EpanetConstants.EN_GPV]:
336 raise ValueError(f"Invalid value '{link_type_id}' for 'link_type_id'")
337 if not isinstance(link_id, str):
338 raise TypeError("'link_id' must be an instance of 'str' " +
339 f"but not of '{type(link_id)}'")
340 if not isinstance(action_type_id, int):
341 raise TypeError("'action_type_id' must be an instance of 'int' " +
342 f"but not of '{type(action_type_id)}'")
343 if action_type_id not in [EN_R_ACTION_SETTING, EN_R_ACTION_STATUS_OPEN,
344 EN_R_ACTION_STATUS_CLOSED, EN_R_ACTION_STATUS_ACTIVE]:
345 raise ValueError(f"Invalid value '{action_type_id}' for 'action_type_id'")
346
347 self._link_type_id = link_type_id
348 self._link_id = link_id
349 self._action_type_id = action_type_id
350 self._action_value = action_value
351
352 super().__init__(**kwds)
353
354 @property
355 def link_type_id(self) -> int:
356 """
357 Returns the link type ID.
358
359 Will be one of the following EPANET constants:
360
361 - EN_CVPIPE = 0
362 - EN_PIPE = 1
363 - EN_PUMP = 2
364 - EN_PRV = 3
365 - EN_PSV = 4
366 - EN_PBV = 5
367 - EN_FCV = 6
368 - EN_TCV = 7
369 - EN_GPV = 8
370
371 Returns
372 -------
373 `int`
374 Link type ID.
375 """
376 return self._link_type_id
377
378 @property
379 def link_id(self) -> str:
380 """
381 Returns the link ID.
382
383 Returns
384 -------
385 `str`
386 Link ID.
387 """
388 return self._link_id
389
390 @property
391 def action_type_id(self) -> int:
392 """
393 Returns the type ID of the action.
394
395 Will be one of the following constants:
396
397 - EN_R_ACTION_SETTING = -1
398 - EN_R_ACTION_STATUS_OPEN = 1
399 - EN_R_ACTION_STATUS_CLOSED = 2
400 - EN_R_ACTION_STATUS_ACTIVE = 3
401
402 Returns
403 -------
404 `int`
405 Type ID of the action.
406 """
407 return self._action_type_id
408
409 @property
410 def action_value(self) -> Any:
411 """
412 Returns the value of the acton (e.g. pump speed).
413 Only relevant if action_type_id = EN_R_SETTING.
414
415 Returns
416 -------
417 `Any`
418 Value of the action.
419 """
420 return self._action_value
421
[docs]
422 def get_attributes(self) -> dict:
423 return super().get_attributes() | {"link_type_id": self._link_type_id,
424 "link_id": self._link_id,
425 "action_type_id": self._action_type_id,
426 "action_value": self._action_value}
427
428 def __eq__(self, other) -> bool:
429 return self._link_type_id == other.link_type_id and \
430 self._link_id == other.link_id and \
431 self._action_type_id == other.action_type_id and \
432 self._action_value == other.action_value
433
434 def __str__(self) -> str:
435 desc = ""
436
437 if self._link_type_id in [EpanetConstants.EN_CVPIPE, EpanetConstants.EN_PIPE]:
438 desc += "PIPE "
439 elif self._link_type_id == EpanetConstants.EN_PUMP:
440 desc += "PUMP "
441 else:
442 desc += "VALVE "
443
444 desc += f"{self._link_id} "
445
446 if self._action_type_id == EN_R_ACTION_SETTING:
447 desc += f"SETTING IS {self._action_value}"
448 elif self._action_type_id == EN_R_ACTION_STATUS_OPEN:
449 desc += "STATUS IS OPEN"
450 elif self._action_type_id == EN_R_ACTION_STATUS_CLOSED:
451 desc += "STATUS IS CLOSED"
452 elif self.action_type_id == EN_R_ACTION_STATUS_ACTIVE:
453 desc += "STATUS IS ACTIVE"
454
455 return desc
456
457
[docs]
458@serializable(COMPLEX_CONTROL_ID, ".epytflow_complex_control")
459class ComplexControlModule(JsonSerializable):
460 """
461 Class representing a complex control module (i.e. IF-THEN-ELSE rule) as implemented in EPANET.
462
463 Parameters
464 ----------
465 rule_id : `str`
466 ID of the rule.
467 condition_1 : :class:`~epyt_flow.simulation.scada.complex_control.RuleCondition`
468 First condition of this rule.
469 additional_conditions : list[tuple[int, :class:`~epyt_flow.simulation.scada.complex_control.RuleCondition`]]
470 List of (optional) additional conditions incl. their conjunction operator
471 (must be either EN_R_AND = 2 or EN_R_OR = 3).
472
473 Empty list if there are no additional conditions.
474 actions : list[:class:`~epyt_flow.simulation.scada.complex_control.RuleAction`]
475 List of actions that are applied if the conditions are met.
476 Must contain at least one action.
477 else_actions : list[:class:`~epyt_flow.simulation.scada.complex_control.RuleAction`]
478 List of actions that are applied if the conditions are NOT met.
479 priority : `int`
480 Priority of this control rule.
481 """
482 def __init__(self, rule_id: str, condition_1: RuleCondition,
483 additional_conditions: list[tuple[int, RuleCondition]], actions: list[RuleAction],
484 else_actions: list[RuleAction], priority: int, **kwds):
485 if not isinstance(rule_id, str):
486 raise TypeError(f"'rule_id' must be an instance of 'str' but not of '{type(rule_id)}'")
487 if not isinstance(condition_1, RuleCondition):
488 raise TypeError("'condition_1' must be an instance of " +
489 "'epyt_flow.simulation.scada.RuleCondition' " +
490 f"but not of '{type(condition_1)}'")
491 if not isinstance(additional_conditions, list) or \
492 any(not isinstance(condition, tuple) for condition in additional_conditions) or \
493 any(not isinstance(condition[0], int) or condition[0] not in [EN_R_AND, EN_R_OR] or
494 not isinstance(condition[1], RuleCondition)
495 for condition in additional_conditions):
496 raise TypeError("'additional_conditions' must be a list of " +
497 "'tuple[int, epyt_flow.simulation.scada.RuleCondition]' instances")
498 if not isinstance(actions, list) or any(not isinstance(action, RuleAction)
499 for action in actions):
500 raise TypeError("'actions' must be a list of " +
501 "'epyt_flow.simulation.scada.RuleAction' instances")
502 if len(actions) == 0:
503 raise ValueError("'actions' must contain at least one action")
504 if not isinstance(else_actions, list) or any(not isinstance(action, RuleAction)
505 for action in actions):
506 raise TypeError("'else_actions' must be a list of " +
507 "'epyt_flow.simulation.scada.RuleAction' instances")
508 if not isinstance(priority, int) or priority < 0:
509 raise TypeError("'priority' must be a non-negative integer")
510
511 self._rule_id = rule_id
512 self._condition_1 = condition_1
513 self._additional_conditions = additional_conditions
514 self._actions = actions
515 self._else_actions = else_actions
516 self._priority = priority
517
518 super().__init__(**kwds)
519
520 @property
521 def rule_id(self) -> str:
522 """
523 Returns the ID of this control rule.
524
525 Returns
526 -------
527 `str`
528 ID of this control rule.
529 """
530 return self._rule_id
531
532 @property
533 def condition_1(self) -> RuleCondition:
534 """
535 Returns the first condition of this rule.
536
537 Returns
538 -------
539 :class:`~epyt_flow.simulation.scada.complex_control.RuleCondition`
540 First condition of this rule.
541 """
542 return deepcopy(self._condition_1)
543
544 @property
545 def additional_conditions(self) -> list[tuple[int, RuleCondition]]:
546 """
547 Returns the list of (optional) additional conditions incl. their conjunction operator.
548 Empty list if there are no additional conditions.
549
550 Returns
551 -------
552 list[tuple[int, :class:`~epyt_flow.simulation.scada.complex_control.RuleCondition`]]
553 List of (optional) additional conditions incl. their conjunction operator.
554 """
555 return deepcopy(self._additional_conditions)
556
557 @property
558 def actions(self) -> list[RuleAction]:
559 """
560 Returns the list of actions that are applied if the conditions are met.
561
562 Returns
563 -------
564 list[:class:`~epyt_flow.simulation.scada.complex_control.RuleAction`]
565 List of actions that are applied if the conditions are met.
566 """
567 return deepcopy(self._actions)
568
569 @property
570 def else_actions(self) -> list[RuleAction]:
571 """
572 Returns the list of actions that are applied if the conditions are NOT met.
573
574 Returns
575 -------
576 list[:class:`~epyt_flow.simulation.scada.complex_control.RuleAction`]
577 List of actions that are applied if the conditions are NOT met.
578 """
579 return deepcopy(self._else_actions)
580
581 @property
582 def priority(self) -> int:
583 """
584 Returns the priority of this control rule.
585
586 Returns
587 -------
588 `int`
589 Priority of this control rule.
590 """
591 return self._priority
592
[docs]
593 def get_attributes(self) -> dict:
594 return super().get_attributes() | {"rule_id": self._rule_id,
595 "condition_1": self._condition_1,
596 "additional_conditions": self._additional_conditions,
597 "actions": self._actions,
598 "else_actions": self._else_actions,
599 "priority": self._priority}
600
601 def __eq__(self, other) -> bool:
602 return super().__eq__(other) and self._rule_id == other.rule_id and \
603 self._priority == other.priority and self._condition_1 == other.condition_1 and \
604 np.all(self._additional_conditions == other.additional_conditions) and \
605 np.all(self._actions == other.actions) and \
606 np.all(self._else_actions == other.else_actions)
607
608 def __str__(self) -> str:
609 desc = ""
610
611 desc += f"RULE {self._rule_id}\n"
612 desc += f"IF {self._condition_1} "
613 for op, action in self._additional_conditions:
614 if op == EN_R_AND:
615 desc += "\nAND "
616 elif op == EN_R_OR:
617 desc += "\nOR "
618
619 desc += f"{action} "
620 desc += "\nTHEN "
621 desc += "\nAND ".join(str(action) for action in self._actions)
622 if len(self._else_actions) != 0:
623 desc += "\nELSE " + "\nAND ".join(str(action) for action in self._else_actions)
624
625 desc += f"\nPRIORITY {self._priority}"
626
627 return desc