tests/ocpp-server/timer.py

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