Added timestamping to incoming values and timeout to signal descriptor
Added timestamping to ValueDescriptor to be able to record on a timebase, timebase uses monotonic_ns() which is a system wide timebase that can be called by connectors written in a host of languages Added timeout to signal descriptors, live values now return None when they are timed out
This commit is contained in:
@@ -32,6 +32,7 @@ class Core:
|
|||||||
self._state: Literal["uninitd", "initd", "started", "stopping", "stopped"] = (
|
self._state: Literal["uninitd", "initd", "started", "stopping", "stopped"] = (
|
||||||
"unintid"
|
"unintid"
|
||||||
)
|
)
|
||||||
|
self._recording = threading.Event()
|
||||||
self._core_version: VersionDescriptor = CORE_VERSION
|
self._core_version: VersionDescriptor = CORE_VERSION
|
||||||
self._stop_event: threading.Event = threading.Event()
|
self._stop_event: threading.Event = threading.Event()
|
||||||
self._core_config: CoreConfig = config
|
self._core_config: CoreConfig = config
|
||||||
@@ -103,6 +104,12 @@ class Core:
|
|||||||
extra={"event": "core.stopped", "core_state": self._state},
|
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:
|
def get_signal_descriptor(self, signal_id: UUID) -> SignalDescriptor | None:
|
||||||
return self._connector_registry.get_signal_descriptor(signal_id)
|
return self._connector_registry.get_signal_descriptor(signal_id)
|
||||||
|
|
||||||
@@ -136,7 +143,15 @@ class Core:
|
|||||||
live_value = self._live_values.get(message.signal_id)
|
live_value = self._live_values.get(message.signal_id)
|
||||||
|
|
||||||
if live_value is None:
|
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
|
self._live_values[message.signal_id] = live_value
|
||||||
|
|
||||||
live_value.update(message.value)
|
live_value.update(message.value, message.timestamp)
|
||||||
|
|||||||
@@ -11,6 +11,7 @@ class ValueDescriptor(BaseModel):
|
|||||||
type: Literal["value_descriptor"] = "value_descriptor"
|
type: Literal["value_descriptor"] = "value_descriptor"
|
||||||
signal_id: UUID
|
signal_id: UUID
|
||||||
value: float
|
value: float
|
||||||
|
timestamp: int
|
||||||
|
|
||||||
|
|
||||||
class ValueBatch(BaseModel):
|
class ValueBatch(BaseModel):
|
||||||
|
|||||||
@@ -17,6 +17,7 @@ class SignalDescriptor(BaseModel):
|
|||||||
min_value: float | None = None
|
min_value: float | None = None
|
||||||
max_value: float | None = None
|
max_value: float | None = None
|
||||||
unit: str | None = None
|
unit: str | None = None
|
||||||
|
timeout_ms: int = 2000
|
||||||
|
|
||||||
|
|
||||||
class DynaLabHello(BaseModel):
|
class DynaLabHello(BaseModel):
|
||||||
|
|||||||
@@ -4,17 +4,27 @@
|
|||||||
|
|
||||||
|
|
||||||
from threading import Lock
|
from threading import Lock
|
||||||
|
from time import monotonic, monotonic_ns
|
||||||
|
|
||||||
|
|
||||||
class Value:
|
class Value:
|
||||||
_value: float = 0
|
def __init__(self, timeout_ms: int) -> None:
|
||||||
_last_updated: int = 0
|
self._value: float = 0
|
||||||
_lock: Lock = Lock()
|
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:
|
with self._lock:
|
||||||
self._value = value
|
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:
|
with self._lock:
|
||||||
return self._value
|
return self._value
|
||||||
|
|||||||
+3
-6
@@ -4,7 +4,7 @@ import threading
|
|||||||
from concurrent.futures import CancelledError as FutureCancelledError
|
from concurrent.futures import CancelledError as FutureCancelledError
|
||||||
from concurrent.futures import TimeoutError as FutureTimeoutError
|
from concurrent.futures import TimeoutError as FutureTimeoutError
|
||||||
from statistics import mean
|
from statistics import mean
|
||||||
from time import monotonic, perf_counter
|
from time import monotonic, monotonic_ns, perf_counter
|
||||||
from uuid import UUID, uuid4
|
from uuid import UUID, uuid4
|
||||||
|
|
||||||
from rich.logging import RichHandler
|
from rich.logging import RichHandler
|
||||||
@@ -117,8 +117,7 @@ def value_sender_thread(
|
|||||||
next_send_time += period
|
next_send_time += period
|
||||||
|
|
||||||
value = ValueDescriptor(
|
value = ValueDescriptor(
|
||||||
signal_id=signal_id,
|
signal_id=signal_id, value=2.0, timestamp=monotonic_ns()
|
||||||
value=2.0,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
message = ValueBatch(
|
message = ValueBatch(
|
||||||
@@ -216,9 +215,7 @@ async def perform_handshake(
|
|||||||
connector_version=CONNECTOR_VERSION.get_version(),
|
connector_version=CONNECTOR_VERSION.get_version(),
|
||||||
signals=[
|
signals=[
|
||||||
SignalDescriptor(
|
SignalDescriptor(
|
||||||
id=signal_id,
|
id=signal_id, name="Dummy signal", type="number", timeout_ms=5000
|
||||||
name="Dummy signal",
|
|
||||||
type="number",
|
|
||||||
)
|
)
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user