Commit | Line | Data |
---|---|---|
cbfbfbc1 JB |
1 | """ |
2 | Timer for asyncio. | |
3 | """ | |
4 | ||
5 | import asyncio | |
6 | ||
7 | ||
8 | class Timer: | |
9 | def __init__( | |
10 | self, | |
11 | timeout: float, | |
12 | repeat: bool, | |
13 | callback, | |
14 | callback_args=(), | |
15 | callback_kwargs=None, | |
16 | ): | |
17 | """An asynchronous Timer object. | |
18 | ||
19 | Parameters | |
20 | ----------- | |
21 | timeout: :class:`float`: | |
22 | The duration for which the timer should last. | |
23 | ||
24 | repeat: :class:`bool`: | |
25 | Whether the timer should repeat. | |
26 | ||
27 | callback: :class:`Coroutine` or `Method` or `Function`: | |
28 | An `asyncio` coroutine or a regular method that will be called as soon as | |
29 | the timer ends. | |
30 | ||
31 | callback_args: Optional[:class:`tuple`]: | |
32 | The args to be passed to the callback. | |
33 | ||
34 | callback_kwargs: Optional[:class:`dict`]: | |
35 | The kwargs to be passed to the callback. | |
36 | """ | |
37 | self._timeout = timeout | |
38 | self._repeat = repeat | |
39 | self._callback = callback | |
40 | self._task = asyncio.create_task(self._job()) | |
41 | self._callback_args = callback_args | |
42 | if callback_kwargs is None: | |
43 | callback_kwargs = {} | |
44 | self._callback_kwargs = callback_kwargs | |
45 | ||
46 | async def _job(self): | |
47 | if self._repeat: | |
48 | while self._task.cancelled() is False: | |
49 | await asyncio.sleep(self._timeout) | |
50 | await self._call_callback() | |
51 | else: | |
52 | await asyncio.sleep(self._timeout) | |
53 | await self._call_callback() | |
54 | ||
55 | async def _call_callback(self): | |
56 | if asyncio.iscoroutine(self._callback) or asyncio.iscoroutinefunction( | |
57 | self._callback | |
58 | ): | |
59 | await self._callback(*self._callback_args, **self._callback_kwargs) | |
60 | else: | |
61 | self._callback(*self._callback_args, **self._callback_kwargs) | |
62 | ||
63 | def cancel(self): | |
64 | """ | |
65 | Cancels the timer. The callback will not be called. | |
66 | """ | |
67 | self._task.cancel() |