fix(ocpp-server): ensure the CLI options help is not truncated

[e-mobility-charging-stations-simulator.git] / tests / ocpp-server / timer.py

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