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         """
  16         An asynchronous Timer object.
  17
  18         Parameters
  19         ----------
  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):
  63         """Cancels the timer. The callback will not be called."""
  64         self._task.cancel()