diff --git a/src/dynalab_core/__init__.py b/src/dynalab_core/__init__.py index f5a81dd..5e581d5 100644 --- a/src/dynalab_core/__init__.py +++ b/src/dynalab_core/__init__.py @@ -32,6 +32,7 @@ class Core: self._state: Literal["uninitd", "initd", "started", "stopping", "stopped"] = ( "unintid" ) + self._recording = threading.Event() self._core_version: VersionDescriptor = CORE_VERSION self._stop_event: threading.Event = threading.Event() self._core_config: CoreConfig = config @@ -103,6 +104,12 @@ class Core: extra={"event": "core.stopped", "core_state": self._state}, ) + def start_recording(self) -> None: + self._recording.set() + + def stop_recording(self) -> None: + self._recording.clear() + def get_signal_descriptor(self, signal_id: UUID) -> SignalDescriptor | None: return self._connector_registry.get_signal_descriptor(signal_id) @@ -136,7 +143,15 @@ class Core: live_value = self._live_values.get(message.signal_id) if live_value is None: - live_value = Value() + signal_descriptor = ( + self._connector_registry.get_signal_descriptor( + message.signal_id + ) + ) + + live_value = Value( + signal_descriptor.timeout_ms if signal_descriptor else 2000 + ) self._live_values[message.signal_id] = live_value - live_value.update(message.value) + live_value.update(message.value, message.timestamp) diff --git a/src/dynalab_core/protocols/packets/data.py b/src/dynalab_core/protocols/packets/data.py index d797996..85948d6 100644 --- a/src/dynalab_core/protocols/packets/data.py +++ b/src/dynalab_core/protocols/packets/data.py @@ -11,6 +11,7 @@ class ValueDescriptor(BaseModel): type: Literal["value_descriptor"] = "value_descriptor" signal_id: UUID value: float + timestamp: int class ValueBatch(BaseModel): diff --git a/src/dynalab_core/protocols/packets/handshake.py b/src/dynalab_core/protocols/packets/handshake.py index d9e5578..0ce2a0b 100644 --- a/src/dynalab_core/protocols/packets/handshake.py +++ b/src/dynalab_core/protocols/packets/handshake.py @@ -17,6 +17,7 @@ class SignalDescriptor(BaseModel): min_value: float | None = None max_value: float | None = None unit: str | None = None + timeout_ms: int = 2000 class DynaLabHello(BaseModel): diff --git a/src/dynalab_core/values.py b/src/dynalab_core/values.py index c975c02..d3f33b2 100644 --- a/src/dynalab_core/values.py +++ b/src/dynalab_core/values.py @@ -4,17 +4,27 @@ from threading import Lock +from time import monotonic, monotonic_ns class Value: - _value: float = 0 - _last_updated: int = 0 - _lock: Lock = Lock() + def __init__(self, timeout_ms: int) -> None: + self._value: float = 0 + self._last_updated: int = 0 + self._timeout_ms: int = timeout_ms + self._lock: Lock = Lock() - def update(self, value: float) -> None: + def update(self, value: float, timestamp: int) -> None: with self._lock: self._value = value + self._last_updated = timestamp - def get(self) -> float: + def get(self) -> float | None: + now = monotonic_ns() + if ( + now > self._last_updated + self._timeout_ms * 1_000_000 + or self._last_updated == 0 + ): + return None with self._lock: return self._value diff --git a/test/manual/peer.py b/test/manual/peer.py index db0ba33..3f0c754 100644 --- a/test/manual/peer.py +++ b/test/manual/peer.py @@ -4,7 +4,7 @@ import threading from concurrent.futures import CancelledError as FutureCancelledError from concurrent.futures import TimeoutError as FutureTimeoutError from statistics import mean -from time import monotonic, perf_counter +from time import monotonic, monotonic_ns, perf_counter from uuid import UUID, uuid4 from rich.logging import RichHandler @@ -117,8 +117,7 @@ def value_sender_thread( next_send_time += period value = ValueDescriptor( - signal_id=signal_id, - value=2.0, + signal_id=signal_id, value=2.0, timestamp=monotonic_ns() ) message = ValueBatch( @@ -216,9 +215,7 @@ async def perform_handshake( connector_version=CONNECTOR_VERSION.get_version(), signals=[ SignalDescriptor( - id=signal_id, - name="Dummy signal", - type="number", + id=signal_id, name="Dummy signal", type="number", timeout_ms=5000 ) ], )