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         """
  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         """Cancels the timer. The callback will not be called."""
  65         self._task.cancel()