controlpi_plugins.wait

  1"""Provide waiting/sleeping plugins for all kinds of systems.
  2
  3- Wait waits for time defined in configuration and sends "finished" event.
  4- GenericWait waits for time defined in "wait" command and sends "finished"
  5  event with "id" string defined in "wait" command.
  6
  7>>> import controlpi
  8>>> asyncio.run(controlpi.test(
  9...     {"Test Wait": {"plugin": "Wait", "seconds": 0.01},
 10...      "Test GenericWait": {"plugin": "GenericWait"}},
 11...     [{"target": "Test GenericWait", "command": "wait",
 12...       "seconds": 0.02, "id": "Long Wait"},
 13...      {"target": "Test Wait", "command": "wait"}], 0.025))
 14... # doctest: +NORMALIZE_WHITESPACE
 15test(): {'sender': '', 'event': 'registered',
 16         'client': 'Test Wait', 'plugin': 'Wait',
 17         'sends': [{'event': {'const': 'finished'}}],
 18         'receives': [{'target': {'const': 'Test Wait'},
 19                       'command': {'const': 'wait'}}]}
 20test(): {'sender': '', 'event': 'registered',
 21         'client': 'Test GenericWait', 'plugin': 'GenericWait',
 22         'sends': [{'event': {'const': 'finished'},
 23                    'id': {'type': 'string'}}],
 24         'receives': [{'target': {'const': 'Test GenericWait'},
 25                       'command': {'const': 'wait'},
 26                       'seconds': {'type': 'number'},
 27                       'id': {'type': 'string'}}]}
 28test(): {'sender': 'test()', 'target': 'Test GenericWait',
 29         'command': 'wait', 'seconds': 0.02, 'id': 'Long Wait'}
 30test(): {'sender': 'test()', 'target': 'Test Wait', 'command': 'wait'}
 31test(): {'sender': 'Test Wait', 'event': 'finished'}
 32test(): {'sender': 'Test GenericWait', 'event': 'finished',
 33         'id': 'Long Wait'}
 34"""
 35
 36import asyncio
 37
 38from controlpi import BasePlugin, Message, MessageTemplate
 39
 40
 41class Wait(BasePlugin):
 42    """Wait for time defined in configuration.
 43
 44    The "seconds" configuration key gets the number of seconds to wait after
 45    receiving a "wait" command before sending the "finished" event:
 46    >>> import controlpi
 47    >>> asyncio.run(controlpi.test(
 48    ...     {"Long Wait": {"plugin": "Wait", "seconds": 0.02},
 49    ...      "Short Wait": {"plugin": "Wait", "seconds": 0.01}},
 50    ...     [{"target": "Long Wait", "command": "wait"},
 51    ...      {"target": "Short Wait", "command": "wait"}], 0.025))
 52    ... # doctest: +NORMALIZE_WHITESPACE
 53    test(): {'sender': '', 'event': 'registered',
 54             'client': 'Long Wait', 'plugin': 'Wait',
 55             'sends': [{'event': {'const': 'finished'}}],
 56             'receives': [{'target': {'const': 'Long Wait'},
 57                           'command': {'const': 'wait'}}]}
 58    test(): {'sender': '', 'event': 'registered',
 59             'client': 'Short Wait', 'plugin': 'Wait',
 60             'sends': [{'event': {'const': 'finished'}}],
 61             'receives': [{'target': {'const': 'Short Wait'},
 62                           'command': {'const': 'wait'}}]}
 63    test(): {'sender': 'test()', 'target': 'Long Wait', 'command': 'wait'}
 64    test(): {'sender': 'test()', 'target': 'Short Wait', 'command': 'wait'}
 65    test(): {'sender': 'Short Wait', 'event': 'finished'}
 66    test(): {'sender': 'Long Wait', 'event': 'finished'}
 67    """
 68
 69    CONF_SCHEMA = {
 70        "properties": {"seconds": {"type": "number"}},
 71        "required": ["seconds"],
 72    }
 73    """Schema for Wait plugin configuration.
 74
 75    Required configuration key:
 76
 77    - 'seconds': number of seconds to wait.
 78    """
 79
 80    def process_conf(self) -> None:
 81        """Register plugin as bus client."""
 82        self._tasks = set()
 83        self.bus.register(
 84            self.name,
 85            "Wait",
 86            [MessageTemplate({"event": {"const": "finished"}})],
 87            [
 88                (
 89                    [
 90                        MessageTemplate(
 91                            {
 92                                "target": {"const": self.name},
 93                                "command": {"const": "wait"},
 94                            }
 95                        )
 96                    ],
 97                    self._wait,
 98                )
 99            ],
100        )
101
102    async def _wait(self, message: Message) -> None:
103        async def wait_coroutine():
104            await asyncio.sleep(self.conf["seconds"])
105            await self.bus.send(Message(self.name, {"event": "finished"}))
106
107        # Done in separate task to not block queue awaiting this callback:
108        task = asyncio.create_task(wait_coroutine())
109        self._tasks.add(task)
110        task.add_done_callback(self._tasks.discard)
111
112    async def run(self) -> None:
113        """Run no code proactively."""
114        pass
115
116
117class GenericWait(BasePlugin):
118    """Wait for time defined in "wait" command.
119
120    The "wait" command has message keys "seconds" defining the seconds to
121    wait and "id" defining a string to be sent back in the "finished" event
122    after the wait:
123    >>> import controlpi
124    >>> asyncio.run(controlpi.test(
125    ...     {"Test GenericWait": {"plugin": "GenericWait"}},
126    ...     [{"target": "Test GenericWait", "command": "wait",
127    ...       "seconds": 0.02, "id": "Long Wait"},
128    ...      {"target": "Test GenericWait", "command": "wait",
129    ...       "seconds": 0.01, "id": "Short Wait"}], 0.025))
130    ... # doctest: +NORMALIZE_WHITESPACE
131    test(): {'sender': '', 'event': 'registered',
132             'client': 'Test GenericWait', 'plugin': 'GenericWait',
133             'sends': [{'event': {'const': 'finished'},
134                        'id': {'type': 'string'}}],
135             'receives': [{'target': {'const': 'Test GenericWait'},
136                           'command': {'const': 'wait'},
137                           'seconds': {'type': 'number'},
138                           'id': {'type': 'string'}}]}
139    test(): {'sender': 'test()', 'target': 'Test GenericWait',
140             'command': 'wait', 'seconds': 0.02, 'id': 'Long Wait'}
141    test(): {'sender': 'test()', 'target': 'Test GenericWait',
142             'command': 'wait', 'seconds': 0.01, 'id': 'Short Wait'}
143    test(): {'sender': 'Test GenericWait', 'event': 'finished',
144             'id': 'Short Wait'}
145    test(): {'sender': 'Test GenericWait', 'event': 'finished',
146             'id': 'Long Wait'}
147    """
148
149    CONF_SCHEMA = True
150    """Schema for GenericWait plugin configuration.
151
152    There are no required or optional configuration keys.
153    """
154
155    def process_conf(self) -> None:
156        """Register plugin as bus client."""
157        self._tasks = set()
158        self.bus.register(
159            self.name,
160            "GenericWait",
161            [
162                MessageTemplate(
163                    {"event": {"const": "finished"}, "id": {"type": "string"}}
164                )
165            ],
166            [
167                (
168                    [
169                        MessageTemplate(
170                            {
171                                "target": {"const": self.name},
172                                "command": {"const": "wait"},
173                                "seconds": {"type": "number"},
174                                "id": {"type": "string"},
175                            }
176                        )
177                    ],
178                    self._wait,
179                )
180            ],
181        )
182
183    async def _wait(self, message: Message) -> None:
184        async def wait_coroutine():
185            assert isinstance(message["seconds"], float) or isinstance(
186                message["seconds"], int
187            )
188            await asyncio.sleep(message["seconds"])
189            await self.bus.send(
190                Message(self.name, {"event": "finished", "id": message["id"]})
191            )
192
193        # Done in separate task to not block queue awaiting this callback:
194        task = asyncio.create_task(wait_coroutine())
195        self._tasks.add(task)
196        task.add_done_callback(self._tasks.discard)
197
198    async def run(self) -> None:
199        """Run no code proactively."""
200        pass
201
202
203class Timer(BasePlugin):
204    """Timer that can be started and cancelled.
205
206    The "seconds" configuration key gets the number of seconds to wait after
207    receiving a "start" command before sending the "finished" event.
208    The "cancel" command cancels all outstanding "finished" events and sends
209    a corresponding "cancelled" event:
210    >>> import controlpi
211    >>> asyncio.run(controlpi.test(
212    ...     {"Timer": {"plugin": "Timer", "seconds": 0.01}},
213    ...     [{"target": "Timer", "command": "start"},
214    ...      {"target": "Timer", "command": "start"},
215    ...      {"target": "Timer", "command": "cancel"},
216    ...      {"target": "Timer", "command": "start"},
217    ...      {"target": "Timer", "command": "start"}], 0.015))
218    ... # doctest: +NORMALIZE_WHITESPACE
219    test(): {'sender': '', 'event': 'registered',
220             'client': 'Timer', 'plugin': 'Timer',
221             'sends': [{'event': {'const': 'finished'}},
222                       {'event': {'const': 'cancelled'}}],
223             'receives': [{'target': {'const': 'Timer'},
224                           'command': {'const': 'start'}},
225                          {'target': {'const': 'Timer'},
226                           'command': {'const': 'cancel'}}]}
227    test(): {'sender': 'test()', 'target': 'Timer', 'command': 'start'}
228    test(): {'sender': 'test()', 'target': 'Timer', 'command': 'start'}
229    test(): {'sender': 'test()', 'target': 'Timer', 'command': 'cancel'}
230    test(): {'sender': 'test()', 'target': 'Timer', 'command': 'start'}
231    test(): {'sender': 'Timer', 'event': 'cancelled', 'count': 2}
232    test(): {'sender': 'test()', 'target': 'Timer', 'command': 'start'}
233    test(): {'sender': 'Timer', 'event': 'finished'}
234    test(): {'sender': 'Timer', 'event': 'finished'}
235    """
236
237    CONF_SCHEMA = {
238        "properties": {"seconds": {"type": "number"}},
239        "required": ["seconds"],
240    }
241    """Schema for Timer plugin configuration.
242
243    Required configuration key:
244
245    - 'seconds': number of seconds to wait.
246    """
247
248    def process_conf(self) -> None:
249        """Register plugin as bus client."""
250        self._tasks = set()
251        self.started = 0
252        self.cancelled = 0
253        self.bus.register(
254            self.name,
255            "Timer",
256            [
257                MessageTemplate({"event": {"const": "finished"}}),
258                MessageTemplate({"event": {"const": "cancelled"}}),
259            ],
260            [
261                (
262                    [
263                        MessageTemplate(
264                            {
265                                "target": {"const": self.name},
266                                "command": {"const": "start"},
267                            }
268                        )
269                    ],
270                    self._start,
271                ),
272                (
273                    [
274                        MessageTemplate(
275                            {
276                                "target": {"const": self.name},
277                                "command": {"const": "cancel"},
278                            }
279                        )
280                    ],
281                    self._cancel,
282                ),
283            ],
284        )
285
286    async def _start(self, message: Message) -> None:
287        self.started += 1
288
289        async def wait_coroutine():
290            await asyncio.sleep(self.conf["seconds"])
291            if self.cancelled > 0:
292                self.cancelled -= 1
293                self.started -= 1
294            elif self.started > 0:
295                self.started -= 1
296                await self.bus.send(Message(self.name, {"event": "finished"}))
297
298        # Done in separate task to not block queue awaiting this callback:
299        task = asyncio.create_task(wait_coroutine())
300        self._tasks.add(task)
301        task.add_done_callback(self._tasks.discard)
302
303    async def _cancel(self, message: Message) -> None:
304        if self.cancelled < self.started:
305            cancel = self.started - self.cancelled
306            self.cancelled = self.started
307            await self.bus.send(
308                Message(self.name, {"event": "cancelled", "count": cancel})
309            )
310
311    async def run(self) -> None:
312        """Run no code proactively."""
313        pass
314
315
316class Periodic(BasePlugin):
317    """Send message periodically.
318
319    The "seconds" configuration key is the period of the repetition:
320    receiving a "wait" command before sending the "finished" event:
321    >>> import controlpi
322    >>> asyncio.run(controlpi.test(
323    ...     {"Loop": {"plugin": "Periodic", "seconds": 0.01,
324    ...               "message": {"key": "value"}}},
325    ...     [], 0.025))
326    ... # doctest: +NORMALIZE_WHITESPACE
327    test(): {'sender': '', 'event': 'registered',
328             'client': 'Loop', 'plugin': 'Periodic',
329             'sends': [{'key': {'const': 'value'}}], 'receives': []}
330    test(): {'sender': 'Loop', 'key': 'value'}
331    test(): {'sender': 'Loop', 'key': 'value'}
332    """
333
334    CONF_SCHEMA = {
335        "properties": {"seconds": {"type": "number"}, "message": {"type": "object"}},
336        "required": ["seconds", "message"],
337    }
338    """Schema for Wait plugin configuration.
339
340    Required configuration key:
341
342    - 'seconds': period of repetition in seconds.
343    - 'message': message to send periodically.
344    """
345
346    def process_conf(self) -> None:
347        """Register plugin as bus client."""
348        self.bus.register(
349            self.name,
350            "Periodic",
351            [MessageTemplate.from_message(self.conf["message"])],
352            [],
353        )
354
355    async def run(self) -> None:
356        """Run periodic loop."""
357        while True:
358            await asyncio.sleep(self.conf["seconds"])
359            await self.bus.send(Message(self.name, self.conf["message"]))