From d9570cd1535cc545a5b95bdf3d8355c909faa9d0 Mon Sep 17 00:00:00 2001 From: Hector van der Aa Date: Sun, 23 Aug 2026 12:27:58 +0100 Subject: [PATCH] Initial DeriveUnit implementation This contains the initial DeriveUnit implementation, including signature validation, this initial implentation only supports ValueDescriptors in the signature alongside the SignalDescriptor and returns a single ValueDescriptor The worker thread still needs to be defined and started but the offline processing has already been validate to work as a first prototype --- src/dynalab_core/derive.py | 107 +++++++++++++++++++++++++ src/dynalab_core/errors.py | 13 +++ src/dynalab_core/protocols/endpoint.py | 5 ++ test/manual/derive_unit.py | 41 ++++++++++ 4 files changed, 166 insertions(+) create mode 100644 src/dynalab_core/derive.py create mode 100644 test/manual/derive_unit.py diff --git a/src/dynalab_core/derive.py b/src/dynalab_core/derive.py new file mode 100644 index 0000000..874a0c5 --- /dev/null +++ b/src/dynalab_core/derive.py @@ -0,0 +1,107 @@ +# Copyright (C) 2026 Hector van der Aa +# Copyright (C) 2026 Association Exergie +# SPDX-License-Identifier: GPL-3.0-or-later + +from collections.abc import Callable +from enum import IntEnum +import inspect +from queue import Empty, Queue +from threading import Thread +import threading +from time import sleep +from typing import Literal +from uuid import UUID + +from dynalab_core.errors import ( + DeriveUnitArgsMismatchError, + DeriveUnitInvalidSignatureError, +) +from dynalab_core.protocols.packets.data import ValueDescriptor +from dynalab_core.protocols.packets.handshake import SignalDescriptor + + +class ParserState(IntEnum): + INIT = 0 + VALUES_DONE = 1 + SIGNAL_DONE = 2 + + +class DeriveUnit: + def __init__( + self, + unit_uuid: UUID, + processing_function: Callable[..., ValueDescriptor], + input_signals: list[SignalDescriptor], + return_signal: SignalDescriptor, + ) -> None: + self._id: UUID = unit_uuid + self._worker_thread: Thread | None = None + self._process_function: Callable = processing_function + self._return_signal: SignalDescriptor = return_signal + self._stop_event: threading.Event = threading.Event() + self._input_queue: Queue[list[ValueDescriptor]] = [] + self._output_queue: Queue[ValueDescriptor] = [] + + self._num_input_args: int = 0 + self._parser_state: int = ParserState.INIT + sig = inspect.signature(self._process_function) + + for name, param in sig.parameters.items(): + if self._parser_state == ParserState.INIT: + if param.annotation is ValueDescriptor: + self._num_input_args += 1 + else: + self._parser_state = ParserState.VALUES_DONE + + if self._parser_state == ParserState.VALUES_DONE: + if param.annotation is SignalDescriptor: + self._parser_state = ParserState.SIGNAL_DONE + continue + else: + raise DeriveUnitInvalidSignatureError + + if self._parser_state == ParserState.SIGNAL_DONE: + raise DeriveUnitInvalidSignatureError + + if sig.return_annotation is not ValueDescriptor: + raise DeriveUnitInvalidSignatureError + + print(f"Signature validated with {self._num_input_args} input values") + # TODO: Define and start worker thread + + def worker_function(self) -> None: + while not self._stop_event.is_set(): + try: + input_args = self._input_queue.get(timeout=0.1) + except Empty: + continue + + if len(input_args) != self._num_input_args: + continue + + self._output_queue.put( + self._process_function(*input_args, self._return_signal) + ) + + def put_data(self, input: list[ValueDescriptor]) -> None: + self._input_queue.put(input) + + def get_output(self) -> ValueDescriptor | None: + try: + return self._output_queue.get_nowait() + except Empty: + return None + + def get_all_output(self) -> list[ValueDescriptor]: + output: list[ValueDescriptor] = [] + + while True: + try: + output.append(self._output_queue.get_nowait()) + except Empty: + return output + + def process_offline(self, input_args: list[ValueDescriptor]) -> ValueDescriptor: + if len(input_args) == self._num_input_args: + return self._process_function(*input_args, self._return_signal) + raise DeriveUnitArgsMismatchError diff --git a/src/dynalab_core/errors.py b/src/dynalab_core/errors.py index efa0c9e..5a26b5c 100644 --- a/src/dynalab_core/errors.py +++ b/src/dynalab_core/errors.py @@ -18,3 +18,16 @@ class DLPakError(Exception): class DLPakNoDataError(DLPakError): """DLPak no data error.""" + + +# DeriveUnit error declarations +class DeriveUnitError(Exception): + """DeriveUnit error.""" + + +class DeriveUnitInvalidSignatureError(DeriveUnitError): + """DeriveUnit invalid processing function signature error.""" + + +class DeriveUnitArgsMismatchError(DeriveUnitError): + """DeriveUnit incorrect arguments provided error.""" diff --git a/src/dynalab_core/protocols/endpoint.py b/src/dynalab_core/protocols/endpoint.py index 64fd843..fe23470 100644 --- a/src/dynalab_core/protocols/endpoint.py +++ b/src/dynalab_core/protocols/endpoint.py @@ -331,6 +331,11 @@ class ConnectorRegistry: 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, diff --git a/test/manual/derive_unit.py b/test/manual/derive_unit.py new file mode 100644 index 0000000..325a9f9 --- /dev/null +++ b/test/manual/derive_unit.py @@ -0,0 +1,41 @@ +from uuid import uuid4 +from dynalab_core.derive import DeriveUnit +from dynalab_core.protocols.packets.data import ValueDescriptor +from dynalab_core.protocols.packets.handshake import SignalDescriptor + + +def process( + a: ValueDescriptor, + b: ValueDescriptor, + r: SignalDescriptor, +) -> ValueDescriptor: + val = ValueDescriptor( + signal_id=r.id, value=a.value + b.value, timestamp=a.timestamp + ) + return val + + +sig_a = SignalDescriptor( + id=uuid4(), + name="Signal A", + type="number", +) + +sig_b = SignalDescriptor( + id=uuid4(), + name="Signal A", + type="number", +) + +sig_r = SignalDescriptor( + id=uuid4(), + name="Signal R", + type="number", +) + +unit = DeriveUnit(uuid4(), process, [sig_a, sig_b], sig_r) + +val_a = ValueDescriptor(signal_id=sig_a.id, value=1, timestamp=0) +val_b = ValueDescriptor(signal_id=sig_a.id, value=2, timestamp=10) + +print(unit.process_offline([val_a, val_b]))