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