Исходный код EJIO.utils.component.clock.timer

## /utils/component/clock/timer.py

from .time_class import Time
from .err import ClockError
from typing import Callable
from datetime import datetime
from threading import Lock, Thread
from time import sleep

__all__ = [
    'Timer',
]




class Timer:
    """
    Таймер, отслеживающий заданную по времени продолжительность
    и выполняющий указанную функцию по окончании отчета.
    """
    __slots__ = ('duration', '_func', '_elapsed_time', '_running', '_lock', '_thread')

    def __init__(self, duration: Time, func: Callable) -> None:
        """
        Создает таймер с указанной продолжительностьюи и функцией для выполнения по истечении таймера.

        :param duration: Продолжительность таймера типа Time.
        :param func: Функция, которую нужно выполнить по завершении таймера.
        :raises TypeError: Если продолжительность не является объектом Time.
        :raises ValueError: Если функция не callable.
        """
        if not isinstance(duration, Time):
            raise TypeError("duration must be of type Time")
        if not callable(func):
            raise ValueError("func must be of type Callable")

        self.duration: Time = duration
        self._func: Callable = func
        self._elapsed_time: int = 0
        self._running: bool = False
        self._lock = Lock()
        self._thread: Thread | None = None

    def _run(self) -> None:
        """Внутренняя функция, которая выполняет отсчет времени."""
        start_time = datetime.now()
        while self._running and self._elapsed_time < self.duration.as_nanoseconds():
            sleep(0.01)
            with self._lock:
                self._elapsed_time = int((datetime.now() - start_time).total_seconds() * int(1e9))

        with self._lock:
            if self._elapsed_time >= self.duration.as_nanoseconds():
                self._func()
                self._running = False



    def start(self) -> 'Timer':
        """
        Запускает таймер.

        :return: Объект таймера.
        """
        with self._lock:
            if self._running:
                raise ClockError("Timer already started")
            self._running = True
            self._thread = Thread(target=self._run, daemon=True)
            self._thread.start()
            return self




    def stop(self) -> 'Timer':
        """
        Останавливает таймер.

        :return: Объект таймера.
        """
        with self._lock:
            if not self._running:
                raise ClockError("Timer is not started")
            self._running = False
            self._thread = None
            return self




    def prolong(self, additional_time: Time) -> 'Timer':
        """
        Увеличивает длительность таймера.

        :param additional_time: Время, на которое нужно увеличить таймер.
        :return: Объект таймера.
        :raises TypeError: Если additional_time не является объектом Time.
        """
        if not isinstance(additional_time, Time):
            raise TypeError("additional_time must be of type Time")

        with self._lock:
            self.duration += additional_time
            return self


    @property
    def remaining_time(self) -> Time:
        """
        Возвращает оставшееся время.

        :return: Оставшееся время в формате Time.
        """
        with self._lock:
            remaining_ns = max(self.duration.as_nanoseconds() - self._elapsed_time, 0)
            return Time(nanoseconds=remaining_ns)

    @property
    def completed(self) -> bool:
        """
        Проверяет, завершился ли таймер.

        :return: True, если таймер завершен, иначе False.
        """
        with self._lock:
            return self._elapsed_time >= self.duration.as_nanoseconds()