1"""
2Module provides implementations of different types of actuator events.
3"""
4import warnings
5from epanet_plus import EPyT, EpanetConstants
6
7from .system_event import SystemEvent
8from ...serialization import serializable, JsonSerializable, PUMP_STATE_EVENT_ID, \
9 PUMP_SPEED_EVENT_ID, VALVE_STATE_EVENT_ID
10
11
[docs]
12class ActuatorConstants:
13 """
14 Class defining some constants related to actuator events.
15
16 Attributes
17 ----------
18 EN_CLOSED
19 Valve or pump is closed.
20 EN_OPEN
21 Valve or pump is open -- i.e. active.
22 EN_SET_CLOSED
23 Link set closed indicator
24 EN_SET_OPEN
25 Link set open indicator
26 """
27 EN_CLOSED = 0
28 EN_OPEN = 1
29 EN_SET_CLOSED = -1e10
30 EN_SET_OPEN = 1e10
31
32
[docs]
33class ActuatorEvent(SystemEvent):
34 """
35 Base class of an actuator event.
36
37 .. note::
38 Note that actuator events are one-time events -- i.e.
39 they are executed only once at a given point in time.
40
41 Parameters
42 ----------
43 time : int
44 Time (in seconds since simulation start) at which this event is executed.
45 """
46 def __init__(self, time: int, **kwds):
47 super().__init__(start_time=time, end_time=time+1, **kwds)
48
[docs]
49 def get_attributes(self) -> dict:
50 return {"time": self.start_time}
51
52
[docs]
53class PumpEvent(ActuatorEvent):
54 """
55 Base class of a pump event.
56
57 Parameters
58 ----------
59 pump_id : str
60 ID of the pump that is affected by this event.
61 """
62 def __init__(self, pump_id: str, **kwds):
63 self._pump_id = pump_id
64
65 super().__init__(**kwds)
66
[docs]
67 def init(self, epanet_api: EPyT) -> None:
68 if self._pump_id not in epanet_api.get_all_pumps_id():
69 raise ValueError(f"Invalid pump ID '{self._pump_id}'")
70
71 super().init(epanet_api)
72
[docs]
73 def get_attributes(self) -> dict:
74 return super().get_attributes() | {"pump_id": self._pump_id}
75
76 @property
77 def pump_id(self) -> str:
78 """
79 Gets the ID of the pump affected by this event.
80
81 Returns
82 -------
83 `str`
84 Pump ID.
85 """
86 return self._pump_id
87
88
[docs]
89@serializable(PUMP_STATE_EVENT_ID, ".epytflow_pump_state_event")
90class PumpStateEvent(PumpEvent, JsonSerializable):
91 """
92 Class implementing a pump state event.
93
94 Parameters
95 ----------
96 pump_state : `str`
97 New state of the pump -- i.e. the state of the pump is set to this value
98 while the event is active.
99
100 Must be one of the following constants defined in
101 :class:`~epyt_flow.simulation.events.actuator_events.ActuatorConstants`:
102
103 - EN_CLOSED = 0
104 - EN_OPEN = 1
105 """
106 def __init__(self, pump_state: int, **kwds):
107 if not isinstance(pump_state, int):
108 raise TypeError("'pump_state' must be an instace of 'int' " +
109 f"but not of {type(pump_state)}")
110 if not 0 <= pump_state <= 1:
111 raise ValueError(f"Invalid pump state '{pump_state}' -- " +
112 "must be either EN_CLOSED (0) or EN_OPEN (1)")
113
114 self._pump_state = pump_state
115
116 super().__init__(**kwds)
117
[docs]
118 def get_attributes(self) -> dict:
119 return super().get_attributes() | {"pump_state": self._pump_state}
120
121 @property
122 def pump_state(self) -> int:
123 """
124 Gets the new pump state.
125
126 Returns
127 -------
128 `int`
129 New pump state.
130
131 One of the following constants defined in
132 :class:`~epyt_flow.simulation.events.actuator_events.ActuatorConstants`:
133
134 - EN_CLOSED = 0
135 - EN_OPEN = 1
136 """
137 return self._pump_state
138
[docs]
139 def apply(self, cur_time: int) -> None:
140 pump_link_idx = self._epanet_api.getlinkindex(self.pump_id)
141
142 pattern_idx = self._epanet_api.getlinkvalue(pump_link_idx, EpanetConstants.EN_LINKPATTERN)
143 if pattern_idx != 0:
144 warnings.warn(f"Can not set pump state of pump {self.pump_id} " +
145 "because a pump pattern exists")
146 else:
147 self._epanet_api.setlinkvalue(pump_link_idx, EpanetConstants.EN_STATUS,
148 self._pump_state)
149
150
[docs]
151@serializable(PUMP_SPEED_EVENT_ID, ".epytflow_pump_speed_event")
152class PumpSpeedEvent(PumpEvent, JsonSerializable):
153 """
154 Class implementing a pump speed event.
155
156 Parameters
157 ----------
158 pump_speed : float
159 New pump speed -- i.e. the speed of the pump is set to this value while the event is active.
160 """
161 def __init__(self, pump_speed: float, **kwds):
162 if not isinstance(pump_speed, float):
163 raise TypeError("'pump_speed' must be an instance of 'float' " +
164 f"but not of {type(pump_speed)}")
165 if pump_speed <= 0:
166 raise ValueError("Pump speed must be positive")
167
168 self._pump_speed = pump_speed
169
170 super().__init__(**kwds)
171
[docs]
172 def get_attributes(self) -> dict:
173 return super().get_attributes() | {"pump_speed": self._pump_speed}
174
175 @property
176 def pump_speed(self) -> float:
177 """
178 Gets the new pump speed.
179
180 Returns
181 -------
182 `float`
183 New pump speed.
184 """
185 return self._pump_speed
186
[docs]
187 def apply(self, cur_time: int) -> None:
188 pump_idx = self._epanet_api.get_link_idx(self.pump_id)
189 pattern_idx = self._epanet_api.getlinkvalue(pump_idx, EpanetConstants.EN_LINKPATTERN)
190
191 if pattern_idx == 0:
192 warnings.warn(f"No pattern for pump '{self.pump_id}' found -- a new pattern is created")
193 pattern_id = f"pump_speed_{self.pump_id}"
194 self._epanet_api.add_pattern(pattern_id, [self._pump_speed])
195 pattern_idx = self._epanet_api.getpatternindex(pattern_id)
196 self._epanet_api.setlinkvalue(pump_idx, EpanetConstants.EN_LINKPATTERN, pattern_idx)
197
198 self._epanet_api.setpattern(pattern_idx, [self._pump_speed], 1)
199
200
[docs]
201@serializable(VALVE_STATE_EVENT_ID, ".epytflow_valve_state_event")
202class ValveStateEvent(ActuatorEvent, JsonSerializable):
203 """
204 Class implementing a valve state event.
205
206 Parameters
207 ----------
208 valve_id : `str`
209 ID of the valve that is affected by this event.
210 valve_state : `str`
211 New state of the valve -- i.e. the valve state is set to this value this event is executed.
212
213 Must be one of the following constants defined in
214 :class:`~epyt_flow.simulation.events.actuator_events.ActuatorConstants`:
215
216 - EN_CLOSED = 0
217 - EN_OPEN = 1
218 """
219 def __init__(self, valve_id: str, valve_state: int, **kwds):
220 if not isinstance(valve_state, int):
221 raise TypeError("'valve_state' must be an instance of 'int' " +
222 f"but not of {type(valve_state)}")
223 if not 0 <= valve_state <= 1:
224 raise ValueError(f"Invalid valve state '{valve_state}' -- " +
225 "must be either EN_CLOSED (0) or EN_OPEN (1)")
226
227 self._valve_id = valve_id
228 self._valve_state = valve_state
229
230 super().__init__(**kwds)
231
[docs]
232 def init(self, epanet_api: EPyT) -> None:
233 if self._valve_id not in epanet_api.get_all_valves_id():
234 raise ValueError(f"Invalid valve ID '{self._valve_id}'")
235
236 super().init(epanet_api)
237
[docs]
238 def get_attributes(self) -> dict:
239 return super().get_attributes() | {"valve_id": self._valve_id,
240 "valve_state": self._valve_state}
241
242 @property
243 def valve_id(self) -> str:
244 """
245 Gets the ID of the valve affected by this event.
246
247 Returns
248 -------
249 `str`
250 Valve ID.
251 """
252 return self._valve_id
253
254 @property
255 def valve_state(self) -> int:
256 """
257 Gets the new state of the valve.
258
259 Returns
260 -------
261 `int`
262 New valve state.
263
264 One of the following constants defined in
265 :class:`~epyt_flow.simulation.events.actuator_events.ActuatorConstants`:
266 """
267 return self._valve_state
268
[docs]
269 def apply(self, cur_time: int) -> None:
270 valve_link_idx = self._epanet_api.get_link_idx(self._valve_id)
271 self._epanet_api.setlinkvalue(valve_link_idx, EpanetConstants.EN_STATUS, self._valve_state)