1"""
2Module provides base classes for system events such as leakages, actuator events, etc.
3"""
4from abc import abstractmethod
5from epanet_plus import EPyT
6
7from .event import Event
8
9
[docs]
10class SystemEvent(Event):
11 """
12 Base class for a system event -- i.e. an event that affects the EPANET simulation.
13 """
14 def __init__(self, **kwds):
15 self._epanet_api = None
16 self._exit_called = False
17
18 super().__init__(**kwds)
19
[docs]
20 def init(self, epanet_api: EPyT) -> None:
21 """
22 Initializes the event.
23
24 Parameters
25 ----------
26 epanet_api : `epanet_plus.EPyT <https://epanet-plus.readthedocs.io/en/stable/api.html#epanet_plus.epanet_toolkit.EPyT>`_
27 API to EPANET and EPANET-MSX.
28 """
29 self._epanet_api = epanet_api
30
31 def __getstate__(self):
32 state = self.__dict__.copy()
33 if "_epanet_api" in state:
34 # Do not serialize EPyT instance!
35 del state["_epanet_api"]
36
37 return state
38
39 def __call__(self, cur_time) -> None:
40 return self.step(cur_time)
41
[docs]
42 def step(self, cur_time) -> None:
43 """
44 Is called at every iteration (time step) in the simulation.
45 `apply` or `exit` are called if necessary.
46
47 Parameters
48 ----------
49 cur_time : `int`
50 Current time (seconds since the start) in the simulation.
51 """
52 if self.start_time <= cur_time < self.end_time:
53 self.apply(cur_time)
54 elif cur_time > self.end_time:
55 if self._exit_called is False:
56 self.exit(cur_time)
57 self._exit_called = True
58
[docs]
59 def reset(self) -> None:
60 """
61 Resets this event -- i.e. make it ready for another simulation run.
62 """
63
[docs]
64 def exit(self, cur_time) -> None:
65 """
66 Is called ONCE after the event is over -- i.e. next time step after `end_time`.
67
68 Any "clean-up" or "exiting" logic should go here.
69
70 Parameters
71 ----------
72 cur_time : `int`
73 Current time (seconds since the start) in the simulation.
74 """
75
[docs]
76 def cleanup(self) -> None:
77 """
78 Clean up any changes/modifications made by this event.
79 """
80
[docs]
81 @abstractmethod
82 def apply(self, cur_time: int) -> None:
83 """
84 Implements the event using EPANET and EPANET-MSX.
85
86 This function is only called when the event is active.
87
88 Parameters
89 ----------
90 cur_time : `int`
91 Current time (seconds since the start) in the simulation.
92 """
93 raise NotImplementedError()