diff --git a/src/dynalab_core/constants.py b/src/dynalab_core/constants.py index 8506ffb..6981736 100644 --- a/src/dynalab_core/constants.py +++ b/src/dynalab_core/constants.py @@ -5,23 +5,19 @@ from uuid import uuid4 from dynalab_core.protocols.common import VersionDescriptor from dynalab_core.protocols.constants import PROTOCOL_VERSION -from dynalab_core.protocols.packets.handshake import ConnectorHello, DynaLabHello +from dynalab_core.protocols.packets.handshake import ( + ConnectorHello, + DynaLabHello, + JsonDynaLabHello, +) CORE_VERSION = VersionDescriptor(type="alpha", major=0, minor=0, patch=1) -HELLO_PACKET = DynaLabHello( +HELLO_PACKET = JsonDynaLabHello( instance_id=uuid4(), core_version=CORE_VERSION, protocol_version=PROTOCOL_VERSION, heartbeat_interval_ms=1000, heartbeat_timeout_ms=5000, ) - -INTERNAL_CONNECTOR_HELLO = ConnectorHello( - connector_uuid=uuid4(), - protocol_version=PROTOCOL_VERSION, - connector_name="INTERNAL_LOOPBACK", - connector_version=CORE_VERSION.get_version(), - signals=[], -) diff --git a/src/dynalab_core/protocols/endpoint.py b/src/dynalab_core/protocols/endpoint.py index 47b401b..6f3baa5 100644 --- a/src/dynalab_core/protocols/endpoint.py +++ b/src/dynalab_core/protocols/endpoint.py @@ -10,7 +10,7 @@ from time import monotonic, sleep from uuid import UUID from dynalab_core.types import CoreModeState -from dynalab_core.constants import HELLO_PACKET, INTERNAL_CONNECTOR_HELLO +from dynalab_core.constants import HELLO_PACKET from dynalab_core.errors import ConnectorError from dynalab_core.protocols.packets import ProtocolMessage from dynalab_core.protocols.packets.data import ValueBatch, ValueDescriptor @@ -486,21 +486,10 @@ class ConnectorEndpoint: class ConnectorRegistry: def __init__(self, core_input_queue: Queue, core_mode: CoreModeState) -> None: self._endpoints: dict[UUID, ConnectorEndpoint] = {} - self._internal_connector: ConnectorHello = INTERNAL_CONNECTOR_HELLO self._lock = RLock() self._core_input_queue = core_input_queue self._core_mode = core_mode - def add_internal_signal(self, signal: SignalDescriptor) -> None: - with self._lock: - if not any(s.id == signal.id for s in self._internal_connector.signals): - self._internal_connector.signals.append(signal) - - def remove_internal_signal(self, signal: SignalDescriptor) -> None: - with self._lock: - if not any(s.id == signal.id for s in self._internal_connector.signals): - self._internal_connector.signals.remove(signal) - def register( self, hello: ConnectorHello, @@ -612,14 +601,7 @@ class ConnectorRegistry: def get(self, connector_uuid: UUID) -> ConnectorEndpoint | None: with self._lock: - endpoint = self._endpoints.get(connector_uuid) - if not endpoint: - return endpoint - - if self._internal_connector.connector_uuid == connector_uuid: - return self._internal_connector - - return None + return self._endpoints.get(connector_uuid) def get_signal_descriptor(self, signal_id: UUID) -> SignalDescriptor | None: with self._lock: @@ -630,10 +612,6 @@ class ConnectorRegistry: if signal is not None: return signal - for signal in self._internal_connector.signals: - if signal.id == signal_id: - return signal - return None def get_all_signal_descriptors(self) -> list[SignalDescriptor]: @@ -646,9 +624,6 @@ class ConnectorRegistry: for signal in endpoint_signals: signals.append(signal) - for signal in self._internal_connector.signals: - signals.append(signal) - return signals def stop(self) -> bool: diff --git a/src/dynalab_core/protocols/json/server.py b/src/dynalab_core/protocols/json/server.py index 80509d5..95925fd 100644 --- a/src/dynalab_core/protocols/json/server.py +++ b/src/dynalab_core/protocols/json/server.py @@ -17,7 +17,12 @@ from dynalab_core.constants import HELLO_PACKET from dynalab_core.protocols.endpoint import ConnectorEndpoint, ConnectorRegistry from dynalab_core.errors import ConnectorError, JsonServerError from dynalab_core.protocols.json.wire import read_message, write_message -from dynalab_core.protocols.packets.handshake import ConnectorHello, HandshakeAccepted +from dynalab_core.protocols.packets.data import JsonValueBatch, JsonValueDescriptor +from dynalab_core.protocols.packets.handshake import ( + ConnectorHello, + HandshakeAccepted, + JsonConnectorHello, +) log = logging.getLogger(__name__) @@ -193,6 +198,7 @@ class JsonServer: ) while not self._stop_event.is_set(): message = await read_message(reader) + message = message.to_pure() endpoint.put_ingress_packet(message) async def _output_task( @@ -216,7 +222,8 @@ class JsonServer: except Empty: await asyncio.sleep(0.01) continue - await write_message(writer, message) + json_message = message.to_json() + await write_message(writer, json_message) if isinstance(message, HandshakeAccepted): log.info( "Connector %s connected from %s", @@ -309,7 +316,7 @@ class JsonServer: ) return - if not isinstance(message, ConnectorHello): + if not isinstance(message, JsonConnectorHello): reason = "unexpected_message" log.warning( "Connection %s sent %s instead of connector_hello", @@ -324,7 +331,7 @@ class JsonServer: ) return - connector_hello = message + connector_hello = message.to_pure() context["connector_uuid"] = str(connector_hello.connector_uuid) timeout_event = threading.Event() reject_event = threading.Event() diff --git a/src/dynalab_core/protocols/json/wire.py b/src/dynalab_core/protocols/json/wire.py index 9c1f6d4..0aa4b78 100644 --- a/src/dynalab_core/protocols/json/wire.py +++ b/src/dynalab_core/protocols/json/wire.py @@ -5,19 +5,19 @@ from asyncio import StreamReader, StreamWriter from pydantic import TypeAdapter -from dynalab_core.protocols.packets import ProtocolMessage +from dynalab_core.protocols.packets import JsonProtocolMessage -_MESSAGE_ADAPTER = TypeAdapter(ProtocolMessage) +_MESSAGE_ADAPTER = TypeAdapter(JsonProtocolMessage) -async def write_message(writer: StreamWriter, message: ProtocolMessage) -> None: +async def write_message(writer: StreamWriter, message: JsonProtocolMessage) -> None: writer.write(message.model_dump_json().encode("utf-8") + b"\n") await writer.drain() async def read_message( reader: StreamReader, *, max_frame_bytes: int = 65_536 -) -> ProtocolMessage: +) -> JsonProtocolMessage: line = await reader.readline() if not line: diff --git a/src/dynalab_core/protocols/packets/__init__.py b/src/dynalab_core/protocols/packets/__init__.py index ac041e8..48fd440 100644 --- a/src/dynalab_core/protocols/packets/__init__.py +++ b/src/dynalab_core/protocols/packets/__init__.py @@ -2,27 +2,40 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later -from typing import Annotated - -from pydantic import Field - -from dynalab_core.protocols.packets.data import ValueBatch, ValueDescriptor +from dynalab_core.protocols.packets.data import ( + JsonValueBatch, + JsonValueDescriptor, + ValueBatch, + ValueDescriptor, +) from dynalab_core.protocols.packets.handshake import ( ConnectorHello, DynaLabHello, HandshakeAccepted, HandshakeRejected, + JsonConnectorHello, + JsonDynaLabHello, + JsonHandshakeAccepted, + JsonHandshakeRejected, ) -from dynalab_core.protocols.packets.heartbeat import Heartbeat +from dynalab_core.protocols.packets.heartbeat import Heartbeat, JsonHeartbeat +from dynalab_core.protocols.packets.types import JsonProtocolMessage, ProtocolMessage - -ProtocolMessage = Annotated[ - DynaLabHello - | ConnectorHello - | HandshakeAccepted - | HandshakeRejected - | Heartbeat - | ValueDescriptor - | ValueBatch, - Field(discriminator="type"), +__all__ = [ + "ConnectorHello", + "DynaLabHello", + "HandshakeAccepted", + "HandshakeRejected", + "Heartbeat", + "JsonConnectorHello", + "JsonDynaLabHello", + "JsonHandshakeAccepted", + "JsonHandshakeRejected", + "JsonHeartbeat", + "JsonProtocolMessage", + "JsonValueBatch", + "JsonValueDescriptor", + "ProtocolMessage", + "ValueBatch", + "ValueDescriptor", ] diff --git a/src/dynalab_core/protocols/packets/data.py b/src/dynalab_core/protocols/packets/data.py index 85948d6..7353b1f 100644 --- a/src/dynalab_core/protocols/packets/data.py +++ b/src/dynalab_core/protocols/packets/data.py @@ -2,18 +2,55 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later +from __future__ import annotations +from dataclasses import dataclass from typing import Literal from uuid import UUID -from pydantic import BaseModel +from pydantic import BaseModel, ConfigDict + +@dataclass(slots=True) +class ValueDescriptor: + signal_id: UUID + value: float + timestamp: int + + def __init__(self, signal_id: UUID, value: float, timestamp: int) -> None: + self.signal_id: UUID = signal_id + self.value: float = value + self.timestamp: int = timestamp + + def to_json(self) -> JsonValueDescriptor: + return JsonValueDescriptor.model_validate(self) -class ValueDescriptor(BaseModel): +@dataclass(slots=True) +class ValueBatch: + values: list[ValueDescriptor] + + def __init__(self, values: list[ValueDescriptor]) -> None: + self.values: list[ValueDescriptor] = values + + def to_json(self) -> JsonValueBatch: + return JsonValueBatch.model_validate(self) + + +class JsonValueDescriptor(BaseModel): + model_config = ConfigDict(from_attributes=True) + type: Literal["value_descriptor"] = "value_descriptor" signal_id: UUID value: float timestamp: int + def to_pure(self) -> ValueDescriptor: + return ValueDescriptor(self.signal_id, self.value, self.timestamp) + + +class JsonValueBatch(BaseModel): + model_config = ConfigDict(from_attributes=True) -class ValueBatch(BaseModel): type: Literal["value_batch"] = "value_batch" - values: list[ValueDescriptor] + values: list[JsonValueDescriptor] + + def to_pure(self) -> ValueBatch: + return ValueBatch([value.to_pure() for value in self.values]) diff --git a/src/dynalab_core/protocols/packets/handshake.py b/src/dynalab_core/protocols/packets/handshake.py index 4e812e7..2fb7a4e 100644 --- a/src/dynalab_core/protocols/packets/handshake.py +++ b/src/dynalab_core/protocols/packets/handshake.py @@ -2,15 +2,129 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later +from __future__ import annotations +from dataclasses import dataclass from typing import Literal from uuid import UUID -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field from dynalab_core.protocols.common import VersionDescriptor from dynalab_core.protocols.constants import PROTOCOL_VERSION +@dataclass(slots=True) +class SignalDescriptor: + id: UUID + name: str + type: Literal["number", "binary"] + min_value: float | None + max_value: float | None + unit: str | None + timeout_ms: int + origin: Literal["source", "derived"] + + def __init__( + self, + id: UUID, + name: str, + type: Literal["number", "binary"], + min_value: float | None = None, + max_value: float | None = None, + unit: str | None = None, + timeout_ms: int = 2000, + origin: Literal["source", "derived"] = "source", + ) -> None: + self.id: UUID = id + self.name: str = name + self.type: Literal["number", "binary"] = type + self.min_value: float | None = min_value + self.max_value: float | None = max_value + self.unit: str | None = unit + self.timeout_ms: int = timeout_ms + self.origin: Literal["source", "derived"] = origin + + def to_json(self) -> JsonSignalDescriptor: + return JsonSignalDescriptor.model_validate(self) -class SignalDescriptor(BaseModel): +@dataclass(slots=True) +class DynaLabHello: + instance_id: UUID + core_version: VersionDescriptor + protocol_version: VersionDescriptor + heartbeat_interval_ms: int + heartbeat_timeout_ms: int + + def __init__( + self, + instance_id: UUID, + core_version: VersionDescriptor, + protocol_version: VersionDescriptor = PROTOCOL_VERSION, + heartbeat_interval_ms: int = 1000, + heartbeat_timeout_ms: int = 5000, + ) -> None: + self.instance_id: UUID = instance_id + self.core_version: VersionDescriptor = core_version + self.protocol_version: VersionDescriptor = protocol_version + + self.heartbeat_interval_ms: int = heartbeat_interval_ms + self.heartbeat_timeout_ms: int = heartbeat_timeout_ms + + def to_json(self) -> JsonDynaLabHello: + return JsonDynaLabHello.model_validate(self) + + +@dataclass(slots=True) +class ConnectorHello: + connector_uuid: UUID + protocol_version: VersionDescriptor + connector_name: str + connector_version: str + signals: list[SignalDescriptor] + + def __init__( + self, + connector_uuid: UUID, + connector_name: str, + connector_version: str, + signals: list[SignalDescriptor], + protocol_version: VersionDescriptor = PROTOCOL_VERSION, + ) -> None: + self.connector_uuid: UUID = connector_uuid + self.protocol_version: VersionDescriptor = protocol_version + + self.connector_name: str = connector_name + self.connector_version: str = connector_version + + self.signals: list[SignalDescriptor] = signals + + def to_json(self) -> JsonConnectorHello: + return JsonConnectorHello.model_validate(self) + + +@dataclass(slots=True) +class HandshakeAccepted: + accepted_signals: list[UUID] + + def __init__(self, accepted_signals: list[UUID]) -> None: + self.accepted_signals: list[UUID] = accepted_signals + + def to_json(self) -> JsonHandshakeAccepted: + return JsonHandshakeAccepted.model_validate(self) + + +@dataclass(slots=True) +class HandshakeRejected: + reason: str + + def __init__(self, reason: str) -> None: + self.reason: str = reason + + def to_json(self) -> JsonHandshakeRejected: + return JsonHandshakeRejected.model_validate(self) + + +class JsonSignalDescriptor(BaseModel): + model_config = ConfigDict(from_attributes=True) + id: UUID name: str type: Literal["number", "binary"] @@ -20,18 +134,41 @@ class SignalDescriptor(BaseModel): timeout_ms: int = 2000 origin: Literal["source", "derived"] = "source" + def to_pure(self) -> SignalDescriptor: + return SignalDescriptor( + self.id, + self.name, + self.type, + self.min_value, + self.max_value, + self.unit, + self.timeout_ms, + self.origin, + ) -class DynaLabHello(BaseModel): + +class JsonDynaLabHello(BaseModel): + model_config = ConfigDict(from_attributes=True) type: Literal["dynalab_hello"] = "dynalab_hello" instance_id: UUID core_version: VersionDescriptor protocol_version: VersionDescriptor = PROTOCOL_VERSION - heartbeat_interval_ms: int = Field(default_factory=1000, ge=0) - heartbeat_timeout_ms: int = Field(default_factory=5000, ge=0) + heartbeat_interval_ms: int = Field(default=1000, ge=0) + heartbeat_timeout_ms: int = Field(default=5000, ge=0) + + def to_pure(self) -> DynaLabHello: + return DynaLabHello( + self.instance_id, + self.core_version, + self.protocol_version, + self.heartbeat_interval_ms, + self.heartbeat_timeout_ms, + ) -class ConnectorHello(BaseModel): +class JsonConnectorHello(BaseModel): + model_config = ConfigDict(from_attributes=True) type: Literal["connector_hello"] = "connector_hello" connector_uuid: UUID protocol_version: VersionDescriptor = PROTOCOL_VERSION @@ -39,14 +176,31 @@ class ConnectorHello(BaseModel): connector_name: str connector_version: str - signals: list[SignalDescriptor] + signals: list[JsonSignalDescriptor] + + def to_pure(self) -> ConnectorHello: + return ConnectorHello( + connector_uuid=self.connector_uuid, + connector_name=self.connector_name, + connector_version=self.connector_version, + signals=[signal.to_pure() for signal in self.signals], + protocol_version=self.protocol_version, + ) -class HandshakeAccepted(BaseModel): +class JsonHandshakeAccepted(BaseModel): + model_config = ConfigDict(from_attributes=True) type: Literal["handshake_accepted"] = "handshake_accepted" accepted_signals: list[UUID] + def to_pure(self) -> HandshakeAccepted: + return HandshakeAccepted(self.accepted_signals) -class HandshakeRejected(BaseModel): + +class JsonHandshakeRejected(BaseModel): + model_config = ConfigDict(from_attributes=True) type: Literal["handshake_rejected"] = "handshake_rejected" reason: str + + def to_pure(self) -> HandshakeRejected: + return HandshakeRejected(self.reason) diff --git a/src/dynalab_core/protocols/packets/heartbeat.py b/src/dynalab_core/protocols/packets/heartbeat.py index 35f0d1d..b2ba65b 100644 --- a/src/dynalab_core/protocols/packets/heartbeat.py +++ b/src/dynalab_core/protocols/packets/heartbeat.py @@ -2,13 +2,35 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later - +from __future__ import annotations +from dataclasses import dataclass from typing import Literal -from pydantic import BaseModel +from pydantic import BaseModel, ConfigDict + +@dataclass(slots=True) +class Heartbeat: + sequence: int + send_timestamp: int + return_timestamp: int | None + + def __init__( + self, sequence: int, send_timestamp: int, return_timestamp: int | None = None + ) -> None: + self.sequence: int = sequence + self.send_timestamp: int = send_timestamp + self.return_timestamp: int | None = return_timestamp + + def to_json(self) -> JsonHeartbeat: + return JsonHeartbeat.model_validate(self) -class Heartbeat(BaseModel): +class JsonHeartbeat(BaseModel): + model_config = ConfigDict(from_attributes=True) + type: Literal["heartbeat"] = "heartbeat" sequence: int send_timestamp: int return_timestamp: int | None = None + + def to_pure(self) -> Heartbeat: + return Heartbeat(self.sequence, self.send_timestamp, self.return_timestamp) diff --git a/src/dynalab_core/protocols/packets/types.py b/src/dynalab_core/protocols/packets/types.py new file mode 100644 index 0000000..618adfd --- /dev/null +++ b/src/dynalab_core/protocols/packets/types.py @@ -0,0 +1,55 @@ +# Copyright (C) 2026 Hector van der Aa +# Copyright (C) 2026 Association Exergie +# SPDX-License-Identifier: GPL-3.0-or-later + +"""Aggregate packet types used at package boundaries. + +Leaf packet modules deliberately do not import this module. Keeping the +aggregate unions here prevents the package re-export layer from becoming part +of a packet model's dependency graph. +""" + +from typing import Annotated + +from pydantic import Field + +from dynalab_core.protocols.packets.data import ( + JsonValueBatch, + JsonValueDescriptor, + ValueBatch, + ValueDescriptor, +) +from dynalab_core.protocols.packets.handshake import ( + ConnectorHello, + DynaLabHello, + HandshakeAccepted, + HandshakeRejected, + JsonConnectorHello, + JsonDynaLabHello, + JsonHandshakeAccepted, + JsonHandshakeRejected, +) +from dynalab_core.protocols.packets.heartbeat import Heartbeat, JsonHeartbeat + + +ProtocolMessage = Annotated[ + DynaLabHello + | ConnectorHello + | HandshakeAccepted + | HandshakeRejected + | Heartbeat + | ValueDescriptor + | ValueBatch, + Field(discriminator="type"), +] + +JsonProtocolMessage = Annotated[ + JsonDynaLabHello + | JsonConnectorHello + | JsonHandshakeAccepted + | JsonHandshakeRejected + | JsonHeartbeat + | JsonValueDescriptor + | JsonValueBatch, + Field(discriminator="type"), +] diff --git a/test/manual/peer.py b/test/manual/peer.py index 4d39036..c10781d 100644 --- a/test/manual/peer.py +++ b/test/manual/peer.py @@ -13,6 +13,13 @@ 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 import ( + JsonConnectorHello, + JsonDynaLabHello, + JsonHeartbeat, + JsonValueBatch, + JsonValueDescriptor, +) from dynalab_core.protocols.packets.data import ValueBatch, ValueDescriptor from dynalab_core.protocols.packets.handshake import ( ConnectorHello, @@ -124,13 +131,13 @@ def value_sender_thread( tau * SINE_FREQUENCY_HZ * (perf_counter() - sine_start_time) ) timestamp = monotonic_ns() - message = ValueBatch(values=[]) - value1 = ValueDescriptor( + message = JsonValueBatch(values=[]) + value1 = JsonValueDescriptor( signal_id=signal_id_1, value=2.0 * sine_value, timestamp=timestamp, ) - value2 = ValueDescriptor( + value2 = JsonValueDescriptor( signal_id=signal_id_2, value=sine_value, timestamp=timestamp, @@ -203,7 +210,7 @@ async def perform_handshake( except TimeoutError as exc: raise TimeoutError("Timed out waiting for server hello") from exc - if not isinstance(message, DynaLabHello): + if not isinstance(message, JsonDynaLabHello): raise RuntimeError(f"Expected DynaLabHello, received {type(message).__name__}") log.debug( @@ -211,7 +218,7 @@ async def perform_handshake( message.instance_id, ) - connector_hello = ConnectorHello( + connector_hello = JsonConnectorHello( connector_uuid=uuid4(), protocol_version=PROTOCOL_VERSION, connector_name="Test connector", @@ -252,7 +259,7 @@ async def receive_messages( while True: message = await read_message(reader) - if isinstance(message, Heartbeat): + if isinstance(message, JsonHeartbeat): message.return_timestamp = monotonic_ms() await send_message( diff --git a/test/test_json_server.py b/test/test_json_server.py index 7aff5c3..074ab4e 100644 --- a/test/test_json_server.py +++ b/test/test_json_server.py @@ -10,7 +10,12 @@ from dynalab_core.config import CoreConfig from dynalab_core.errors import JsonServerError from dynalab_core.protocols.constants import PROTOCOL_VERSION from dynalab_core.protocols.common import VersionDescriptor -from dynalab_core.protocols.packets.handshake import ConnectorHello, SignalDescriptor +from dynalab_core.protocols.packets import JsonConnectorHello +from dynalab_core.protocols.packets.handshake import ( + ConnectorHello, + JsonSignalDescriptor, + SignalDescriptor, +) from test.common import find_available_port @@ -141,12 +146,12 @@ def test_rejected_handshake_is_not_logged_as_connected( port = find_available_port(8765) core = Core(CoreConfig(port=port)) connector_uuid = uuid4() - connector_hello = ConnectorHello( + connector_hello = JsonConnectorHello( connector_uuid=connector_uuid, protocol_version=VersionDescriptor(type="alpha", major=999, minor=0, patch=0), connector_name="Incompatible connector", connector_version="0.1.0-test", - signals=[SignalDescriptor(id=uuid4(), name="Dummy signal", type="number")], + signals=[JsonSignalDescriptor(id=uuid4(), name="Dummy signal", type="number")], ) with caplog.at_level(logging.DEBUG, logger="dynalab_core"): @@ -176,12 +181,12 @@ def test_json_server_waits_for_connection_handlers_on_stop( port = find_available_port(8765) core = Core(CoreConfig(port=port)) connector_uuid = uuid4() - connector_hello = ConnectorHello( + connector_hello = JsonConnectorHello( connector_uuid=connector_uuid, protocol_version=PROTOCOL_VERSION, connector_name="Test connector", connector_version="0.1.0-test", - signals=[SignalDescriptor(id=uuid4(), name="Dummy signal", type="number")], + signals=[JsonSignalDescriptor(id=uuid4(), name="Dummy signal", type="number")], ) with caplog.at_level(logging.DEBUG, logger="dynalab_core"):