| 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() |