| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465 |
- """Timer for asyncio."""
- import asyncio
- class Timer:
- def __init__(
- self,
- timeout: float,
- repeat: bool,
- callback,
- callback_args=(),
- callback_kwargs=None,
- ):
- """
- An asynchronous Timer object.
- Parameters
- ----------
- timeout: :class:`float`:
- The duration for which the timer should last.
- repeat: :class:`bool`:
- Whether the timer should repeat.
- callback: :class:`Coroutine` or `Method` or `Function`:
- An `asyncio` coroutine or a regular method that will be called as soon as
- the timer ends.
- callback_args: Optional[:class:`tuple`]:
- The args to be passed to the callback.
- callback_kwargs: Optional[:class:`dict`]:
- The kwargs to be passed to the callback.
- """
- self._timeout = timeout
- self._repeat = repeat
- self._callback = callback
- self._task = asyncio.create_task(self._job())
- self._callback_args = callback_args
- if callback_kwargs is None:
- callback_kwargs = {}
- self._callback_kwargs = callback_kwargs
- async def _job(self):
- if self._repeat:
- while self._task.cancelled() is False:
- await asyncio.sleep(self._timeout)
- await self._call_callback()
- else:
- await asyncio.sleep(self._timeout)
- await self._call_callback()
- async def _call_callback(self):
- if asyncio.iscoroutine(self._callback) or asyncio.iscoroutinefunction(
- self._callback
- ):
- await self._callback(*self._callback_args, **self._callback_kwargs)
- else:
- self._callback(*self._callback_args, **self._callback_kwargs)
- def cancel(self):
- """Cancels the timer. The callback will not be called."""
- self._task.cancel()
|
