| 1 | """Timer for asyncio.""" |
| 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 | ): |
| 15 | """An asynchronous Timer object. |
| 16 | |
| 17 | Parameters |
| 18 | ---------- |
| 19 | timeout: :class:`float`: |
| 20 | The duration for which the timer should last. |
| 21 | |
| 22 | repeat: :class:`bool`: |
| 23 | Whether the timer should repeat. |
| 24 | |
| 25 | callback: :class:`Coroutine` or `Method` or `Function`: |
| 26 | An `asyncio` coroutine or a regular method that will be called as soon as |
| 27 | the timer ends. |
| 28 | |
| 29 | callback_args: Optional[:class:`tuple`]: |
| 30 | The args to be passed to the callback. |
| 31 | |
| 32 | callback_kwargs: Optional[:class:`dict`]: |
| 33 | The kwargs to be passed to the callback. |
| 34 | |
| 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): |
| 63 | """Cancels the timer. The callback will not be called.""" |
| 64 | self._task.cancel() |