Исходный код EJIO.utils.component.clock.virtual_clock
## /utils/component/clock/virtual_clock.py
from .err import ClockError
from .time_class import Time
from .base_clock import Clock
from datetime import datetime
from typing import Self
__all__ = [
'VirtualClock',
]
[документация]
class VirtualClock(Clock):
"""
Виртуальные часы, отслеживают прошедшее время без привязки к системному.
"""
__slots__ = ('_elapsed_ns', '_initial_time', '_running', '_start_timestamp')
def __init__(self, start_time: Time | None = None) -> None:
"""
Инициализация виртуальных часов.
:param start_time: Начальное время. Если не указано, стартует с 0.
"""
super().__init__()
self._start_timestamp: datetime | None = None
self._elapsed_ns: int = 0
self._initial_time: Time = start_time if start_time else Time()
[документация]
def start(self) -> Self:
"""
Запускает виртуальные часы.
:return: Сам объект для цепочки вызовов.
:raises ClockError: Если часы уже запущены.
"""
with self._lock:
if self._running:
raise ClockError("Clock already started")
self._start_timestamp = datetime.now()
self._running = True
return self
[документация]
def stop(self) -> Self:
"""
Останавливает виртуальные часы.
:return: Сам объект для цепочки вызовов.
:raises ClockError: Если часы не запущены.
"""
with self._lock:
if not self._running:
raise ClockError("Clock is not started")
delta_ns = int((datetime.now() - self._start_timestamp).total_seconds() * int(1e9)) # type: ignore
self._elapsed_ns += delta_ns
self._start_timestamp = None
self._running = False
return self
[документация]
def get_time(self) -> Time:
"""
Возвращает текущее виртуальное время.
:return: Объект Time, представляющий текущее время.
"""
with self._lock:
total_ns = self._elapsed_ns
if self._running:
delta_ns = int((datetime.now() - self._start_timestamp).total_seconds() * int(1e9)) # type: ignore
total_ns += delta_ns
return Time(nanoseconds=total_ns) + self._initial_time