From 561f42aca224d33d75a75e800384ef6c00a56785 Mon Sep 17 00:00:00 2001 From: Hector van der Aa Date: Wed, 5 Aug 2026 20:42:19 +0100 Subject: [PATCH] Updated tests and added core live value handling Removed useless tests, will need to complete a more comprehensive test suite however this is not the priority right now Added a live value dict in the core whose values are populated by the core's input thread --- src/dynalab_core/__init__.py | 29 +++++++------ src/dynalab_core/protocols/endpoint.py | 36 +++++++++++----- src/dynalab_core/protocols/json/server.py | 13 +++++- .../protocols/packets/__init__.py | 5 ++- src/dynalab_core/protocols/packets/data.py | 5 +++ src/dynalab_core/values.py | 20 +++++++++ test/manual/peer.py | 23 +++++++++-- test/test_endpoint_logging.py | 41 ------------------- test/test_json_server.py | 11 ++--- 9 files changed, 104 insertions(+), 79 deletions(-) create mode 100644 src/dynalab_core/values.py delete mode 100644 test/test_endpoint_logging.py diff --git a/src/dynalab_core/__init__.py b/src/dynalab_core/__init__.py index ab44df6..bba7540 100644 --- a/src/dynalab_core/__init__.py +++ b/src/dynalab_core/__init__.py @@ -5,9 +5,10 @@ import logging from queue import Empty, Queue import threading -from threading import Thread +from threading import Lock, Thread from time import monotonic, monotonic_ns, sleep from typing import Literal +from uuid import UUID from dynalab_core.config import CoreConfig @@ -18,6 +19,7 @@ from dynalab_core.protocols.common import VersionDescriptor from dynalab_core.protocols.json.server import JsonServer from dynalab_core.protocols.packets import ProtocolMessage from dynalab_core.protocols.packets.data import ValueDescriptor +from dynalab_core.values import Value log = logging.getLogger(__name__) @@ -32,6 +34,9 @@ class Core: self._core_version: VersionDescriptor = CORE_VERSION self._stop_event: threading.Event = threading.Event() self._core_config: CoreConfig = config + # TODO: Implement live values dict + self._live_values: dict[UUID, Value] = {} + self._live_values_lock = Lock() self._data_input_queue: Queue[ProtocolMessage] = Queue(524288) self._input_worker_thread = Thread( @@ -98,21 +103,19 @@ class Core: ) def _input_worker(self) -> None: - last_log = 0 while not self._stop_event.is_set(): - now = monotonic_ns() - messages: list[ProtocolMessage] = [] try: - while not self._data_input_queue.empty(): - messages.append(self._data_input_queue.get_nowait()) + message = self._data_input_queue.get_nowait() except Empty: self._stop_event.wait(0.01) continue - else: - for message in messages: - if isinstance(message, ValueDescriptor): - pass - if now > last_log + 2000 * 1_000_000: - log.debug(f"Core queue size: {self._data_input_queue.qsize()}") - last_log = now + if isinstance(message, ValueDescriptor): + with self._live_values_lock: + live_value = self._live_values.get(message.signal_id) + + if live_value is None: + live_value = Value() + self._live_values[message.signal_id] = live_value + + live_value.update(message.value) diff --git a/src/dynalab_core/protocols/endpoint.py b/src/dynalab_core/protocols/endpoint.py index cde313a..d6bbdea 100644 --- a/src/dynalab_core/protocols/endpoint.py +++ b/src/dynalab_core/protocols/endpoint.py @@ -6,7 +6,7 @@ import logging from queue import Empty, Full, Queue import threading from threading import RLock, Thread -from time import monotonic, sleep +from time import monotonic, monotonic_ns, sleep from uuid import UUID from dynalab_core.constants import HELLO_PACKET @@ -15,7 +15,7 @@ from dynalab_core.protocols.errors import ( ConnectorRegistryAlreadyRegisteredError, ) from dynalab_core.protocols.packets import ProtocolMessage -from dynalab_core.protocols.packets.data import ValueDescriptor +from dynalab_core.protocols.packets.data import ValueBatch, ValueDescriptor from dynalab_core.protocols.packets.handshake import ( ConnectorHello, HandshakeAccepted, @@ -208,20 +208,34 @@ class ConnectorEndpoint: "thread_name": self._input_worker_thread.name, }, ) + last_log = 0 while not self._stop_event.is_set(): + now = monotonic_ns() + messages: list[ProtocolMessage] = [] try: - message = self._get_ingress_packet_no_wait() + while not self._packet_ingress_queue.empty(): + messages.append(self._get_ingress_packet_no_wait()) except Empty: - self._stop_event.wait(0.01) continue else: - if isinstance(message, Heartbeat): - self._heartbeat_ingress_queue.put_nowait(message) - elif isinstance(message, ValueDescriptor): - if message.signal_id in ( - signal.id for signal in self._connector_hello.signals - ): - self._core_input_queue.put_nowait(message) + for message in messages: + if isinstance(message, Heartbeat): + self._heartbeat_ingress_queue.put_nowait(message) + elif isinstance(message, ValueDescriptor): + if message.signal_id in ( + signal.id for signal in self._connector_hello.signals + ): + self._core_input_queue.put_nowait(message) + elif isinstance(message, ValueBatch): + for value in message.values: + if value.signal_id in ( + signal.id for signal in self._connector_hello.signals + ): + self._core_input_queue.put_nowait(value) + + if now > last_log + 2000 * 1_000_000: + log.debug(f"Endpoint queue size: {self._packet_ingress_queue.qsize()}") + last_log = now self._input_worker_stopped_event.set() log.debug( "Connector endpoint input worker %s stopped", diff --git a/src/dynalab_core/protocols/json/server.py b/src/dynalab_core/protocols/json/server.py index 2d6c928..a5c84b1 100644 --- a/src/dynalab_core/protocols/json/server.py +++ b/src/dynalab_core/protocols/json/server.py @@ -6,9 +6,10 @@ import asyncio from asyncio import Server import logging from queue import Empty +from statistics import mean import threading from threading import Thread -from time import monotonic +from time import monotonic, monotonic_ns from uuid import uuid4 @@ -163,9 +164,19 @@ class JsonServer: endpoint: ConnectorEndpoint, ) -> None: log.debug(f"Started input task for {endpoint.uuid()}") + intervals: list[int] = [] + last_val: int = 0 + last_print: int = 0 while not self._stop_event.is_set(): message = await read_message(reader) endpoint.put_ingress_packet(message) + now = monotonic_ns() + intervals.append(now - last_val) + last_val = now + if now > last_print + 2000 * 1_000_000: + log.debug(f"Server input frequency: {1_000_000 / mean(intervals)}kHz") + last_print = now + intervals = [] async def _output_task( self, diff --git a/src/dynalab_core/protocols/packets/__init__.py b/src/dynalab_core/protocols/packets/__init__.py index 5447b0b..ac041e8 100644 --- a/src/dynalab_core/protocols/packets/__init__.py +++ b/src/dynalab_core/protocols/packets/__init__.py @@ -6,7 +6,7 @@ from typing import Annotated from pydantic import Field -from dynalab_core.protocols.packets.data import ValueDescriptor +from dynalab_core.protocols.packets.data import ValueBatch, ValueDescriptor from dynalab_core.protocols.packets.handshake import ( ConnectorHello, DynaLabHello, @@ -22,6 +22,7 @@ ProtocolMessage = Annotated[ | HandshakeAccepted | HandshakeRejected | Heartbeat - | ValueDescriptor, + | ValueDescriptor + | ValueBatch, Field(discriminator="type"), ] diff --git a/src/dynalab_core/protocols/packets/data.py b/src/dynalab_core/protocols/packets/data.py index 8361fdf..d797996 100644 --- a/src/dynalab_core/protocols/packets/data.py +++ b/src/dynalab_core/protocols/packets/data.py @@ -11,3 +11,8 @@ class ValueDescriptor(BaseModel): type: Literal["value_descriptor"] = "value_descriptor" signal_id: UUID value: float + + +class ValueBatch(BaseModel): + type: Literal["value_batch"] = "value_batch" + values: list[ValueDescriptor] diff --git a/src/dynalab_core/values.py b/src/dynalab_core/values.py new file mode 100644 index 0000000..c975c02 --- /dev/null +++ b/src/dynalab_core/values.py @@ -0,0 +1,20 @@ +# Copyright (C) 2026 Hector van der Aa +# Copyright (C) 2026 Association Exergie +# SPDX-License-Identifier: GPL-3.0-or-later + + +from threading import Lock + + +class Value: + _value: float = 0 + _last_updated: int = 0 + _lock: Lock = Lock() + + def update(self, value: float) -> None: + with self._lock: + self._value = value + + def get(self) -> float: + with self._lock: + return self._value diff --git a/test/manual/peer.py b/test/manual/peer.py index 198582d..db0ba33 100644 --- a/test/manual/peer.py +++ b/test/manual/peer.py @@ -12,7 +12,7 @@ from rich.logging import RichHandler from dynalab_core.protocols.common import VersionDescriptor from dynalab_core.protocols.constants import PROTOCOL_VERSION from dynalab_core.protocols.json.wire import read_message, write_message -from dynalab_core.protocols.packets.data import ValueDescriptor +from dynalab_core.protocols.packets.data import ValueBatch, ValueDescriptor from dynalab_core.protocols.packets.handshake import ( ConnectorHello, DynaLabHello, @@ -26,9 +26,9 @@ PORT = 8765 # Set to None to send as quickly as possible. # For a controlled rate, use something like 5_000.0. -TARGET_FREQUENCY_HZ: float | None = None +TARGET_FREQUENCY_HZ: float | None = 1000 -FREQUENCY_SAMPLE_SIZE = 500 +FREQUENCY_SAMPLE_SIZE = 20000 CONNECTOR_VERSION = VersionDescriptor( type="alpha", @@ -116,11 +116,26 @@ def value_sender_thread( next_send_time += period - message = ValueDescriptor( + value = ValueDescriptor( signal_id=signal_id, value=2.0, ) + message = ValueBatch( + values=[ + value, + value, + value, + value, + value, + value, + value, + value, + value, + value, + ] + ) + future = asyncio.run_coroutine_threadsafe( send_message( writer, diff --git a/test/test_endpoint_logging.py b/test/test_endpoint_logging.py deleted file mode 100644 index 1e1da9d..0000000 --- a/test/test_endpoint_logging.py +++ /dev/null @@ -1,41 +0,0 @@ -import logging -from queue import Queue -from uuid import uuid4 - -import pytest - -from dynalab_core.protocols.constants import PROTOCOL_VERSION -from dynalab_core.protocols.endpoint import ConnectorEndpoint -from dynalab_core.protocols.errors import ConnectorEndpointQueueFullError -from dynalab_core.protocols.packets import ProtocolMessage -from dynalab_core.protocols.packets.handshake import ConnectorHello - - -def test_full_endpoint_queue_is_logged(caplog: pytest.LogCaptureFixture) -> None: - hello = ConnectorHello( - connector_uuid=uuid4(), - protocol_version=PROTOCOL_VERSION, - connector_name="Test connector", - connector_version="0.1.0-test", - ) - endpoint = ConnectorEndpoint(hello) - endpoint._packet_ingress_queue = Queue[ProtocolMessage](maxsize=1) - - with caplog.at_level(logging.DEBUG, logger="dynalab_core"): - try: - endpoint.put_ingress_packet(hello) - with pytest.raises(ConnectorEndpointQueueFullError): - endpoint.put_ingress_packet(hello) - finally: - endpoint.stop() - - queue_record = next( - record - for record in caplog.records - if getattr(record, "event", None) == "endpoint.queue_full" - ) - assert queue_record.levelno == logging.WARNING - assert queue_record.connector_uuid == str(hello.connector_uuid) - assert queue_record.queue_direction == "ingress" - assert queue_record.queue_size == 1 - assert queue_record.queue_capacity == 1 diff --git a/test/test_json_server.py b/test/test_json_server.py index 6fd43c2..d762583 100644 --- a/test/test_json_server.py +++ b/test/test_json_server.py @@ -11,7 +11,7 @@ from dynalab_core.protocols.json.errors import ( JsonServerStartupError, JsonServerTimeoutError, ) -from dynalab_core.protocols.packets.handshake import ConnectorHello +from dynalab_core.protocols.packets.handshake import ConnectorHello, SignalDescriptor from test.common import find_available_port @@ -93,9 +93,7 @@ def test_json_server_logs_invalid_handshake( assert server_hello peer.sendall(b"not-json\n") - rejection = _wait_for_event( - caplog, "connector.handshake_rejected" - ) + rejection = _wait_for_event(caplog, "connector.handshake_rejected") finally: core.stop() @@ -116,6 +114,7 @@ def test_json_server_waits_for_connection_handlers_on_stop( protocol_version=PROTOCOL_VERSION, connector_name="Test connector", connector_version="0.1.0-test", + signals=[SignalDescriptor(id=uuid4(), name="Dummy signal", type="number")], ) with caplog.at_level(logging.DEBUG, logger="dynalab_core"): @@ -124,9 +123,7 @@ def test_json_server_waits_for_connection_handlers_on_stop( with socket.create_connection(("127.0.0.1", port), timeout=1) as peer: peer_file = peer.makefile("rb") assert peer_file.readline() - peer.sendall( - connector_hello.model_dump_json().encode("utf-8") + b"\n" - ) + peer.sendall(connector_hello.model_dump_json().encode("utf-8") + b"\n") assert b'"type":"handshake_accepted"' in peer_file.readline() core.stop()