From c67316dd90e3c845f0c8c6fca2fd8cce669df6f1 Mon Sep 17 00:00:00 2001 From: Hector van der Aa Date: Wed, 9 Sep 2026 16:09:02 +0200 Subject: [PATCH] Added derive units to core Derive units are now available in the core and has their relevant input values routed and outputs are looped back into the core router Fixed the dlpak manifest which had no signal descriptors --- src/dynalab_core/__init__.py | 72 ++++++++++++++++++++++++++++++++++-- src/dynalab_core/buffer.py | 2 + src/dynalab_core/derive.py | 21 ++++++++--- src/dynalab_core/errors.py | 4 ++ src/dynalab_core/values.py | 30 +++++++++++++-- test/manual/core.py | 30 +++++++++++++++ test/manual/peer.py | 29 ++++++++++----- test/manual/process.py | 11 ++++++ 8 files changed, 176 insertions(+), 23 deletions(-) create mode 100644 test/manual/process.py diff --git a/src/dynalab_core/__init__.py b/src/dynalab_core/__init__.py index 18528e1..35a82d4 100644 --- a/src/dynalab_core/__init__.py +++ b/src/dynalab_core/__init__.py @@ -3,21 +3,25 @@ # SPDX-License-Identifier: GPL-3.0-or-later +from collections.abc import Callable from datetime import datetime, timezone +import inspect import logging from queue import Empty, Queue import threading from threading import Lock, Thread import time +import types from typing import Literal -from uuid import UUID +from uuid import UUID, uuid4 from dynalab_core.buffer import ValueBuffer from dynalab_core.config import CoreConfig from dynalab_core.constants import CORE_VERSION +from dynalab_core.derive import DeriveUnit from dynalab_core.dlpak import DLPak -from dynalab_core.errors import CoreStateMismatchError +from dynalab_core.errors import CoreMultipleFunctionsFoundError, CoreStateMismatchError from dynalab_core.protocols.endpoint import ConnectorRegistry from dynalab_core.protocols.common import VersionDescriptor from dynalab_core.protocols.json.server import JsonServer @@ -46,6 +50,8 @@ class Core: self._live_values: dict[UUID, Value] = {} self._live_values_lock = Lock() + self._derive_units: dict[UUID, DeriveUnit] = {} + self._data_input_queue: Queue[ProtocolMessage] = Queue(524288) self._input_worker_thread = Thread( target=self._input_worker, name="input_worker_thread", daemon=True @@ -104,6 +110,8 @@ class Core: self._json_server.stop() self._connector_registry.stop() self._stop_event.set() + for unit in self._derive_units.values(): + unit.stop() self._state = "stopped" log.info( "Core stopped", @@ -121,9 +129,12 @@ class Core: self._recording_buffer.normalize() self._processing_buffer = DLPak() self._processing_buffer.set_data(self._recording_buffer) + signal_descriptors = self._connector_registry.get_all_signal_descriptors() + for unit in self._derive_units.values(): + signal_descriptors.append(unit._return_signal) + self._processing_buffer.set_manifest( - self._recording_timestamp, - self._connector_registry.get_all_signal_descriptors(), + self._recording_timestamp, signal_descriptors ) # TODO: remove debug behavior default saving to output self._processing_buffer.write("./", "output") @@ -148,6 +159,35 @@ class Core: return {signal_id: value.get() for signal_id, value in live_values} + def bind_derive_unit( + self, + function: str, + input_signals: list[SignalDescriptor], + output_signal: SignalDescriptor, + ) -> UUID: + unit_id = uuid4() + + namespace = {} + exec(function, namespace) + + functions = [ + obj for obj in namespace.values() if isinstance(obj, types.FunctionType) + ] + + if len(functions) != 1: + raise CoreMultipleFunctionsFoundError + + unit = DeriveUnit( + unit_id, functions[0], input_signals, output_signal, self._data_input_queue + ) + + self._derive_units[unit_id] = unit + + return unit_id + + def unbind_derive_unit(self, unit_id: UUID) -> None: + self._derive_units[unit_id].stop() + def _input_worker(self) -> None: while not self._stop_event.is_set(): try: @@ -177,3 +217,27 @@ class Core: self._recording_buffer.append( message.timestamp, message.signal_id, message.value ) + + for unit in self._derive_units.values(): + input_signals = unit.get_input_signals() + if message.signal_id in [signal.id for signal in input_signals]: + args: list[ValueDescriptor] = [] + for signal in input_signals: + if signal.id == message.signal_id: + args.append(message) + continue + + with self._live_values_lock: + live_value = self._live_values.get(signal.id) + + value_descriptor = ( + live_value.to_descriptor(signal.id) + if live_value is not None + else None + ) + if value_descriptor is None: + break + + args.append(value_descriptor) + else: + unit.put_data(args) diff --git a/src/dynalab_core/buffer.py b/src/dynalab_core/buffer.py index 2267f18..61a4463 100644 --- a/src/dynalab_core/buffer.py +++ b/src/dynalab_core/buffer.py @@ -32,6 +32,8 @@ class ValueBuffer: return len(self._samples) def normalize(self) -> None: + if len(self) == 0: + return with self._lock: minimum = min(sample[0] for sample in self._samples) diff --git a/src/dynalab_core/derive.py b/src/dynalab_core/derive.py index 3391618..d97dfad 100644 --- a/src/dynalab_core/derive.py +++ b/src/dynalab_core/derive.py @@ -8,8 +8,6 @@ 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 ( @@ -33,13 +31,18 @@ class DeriveUnit: processing_function: Callable[..., ValueDescriptor], input_signals: list[SignalDescriptor], return_signal: SignalDescriptor, + return_queue: Queue[ValueDescriptor], ) -> None: self._id: UUID = unit_uuid self._process_function: Callable = processing_function self._return_signal: SignalDescriptor = return_signal + self._input_signals: list[SignalDescriptor] = input_signals self._stop_event: threading.Event = threading.Event() + self._stopped_event: threading.Event = threading.Event() self._input_queue: Queue[list[ValueDescriptor]] = Queue() + # TODO: Deprecate output queue self._output_queue: Queue[ValueDescriptor] = Queue() + self._return_queue: Queue[ValueDescriptor] = return_queue self._num_input_args: int = 0 self._parser_state: int = ParserState.INIT @@ -80,9 +83,10 @@ class DeriveUnit: if len(input_args) != self._num_input_args: continue - self._output_queue.put( - self._process_function(*input_args, self._return_signal) - ) + value = self._process_function(*input_args, self._return_signal) + self._return_queue.put(value) + self._output_queue.put(value) + self._stopped_event.set() def put_data(self, input: list[ValueDescriptor]) -> None: self._input_queue.put(input) @@ -106,3 +110,10 @@ class DeriveUnit: if len(input_args) == self._num_input_args: return self._process_function(*input_args, self._return_signal) raise DeriveUnitArgsMismatchError + + def get_input_signals(self) -> list[SignalDescriptor]: + return self._input_signals + + def stop(self) -> None: + self._stop_event.set() + self._stopped_event.wait(5.0) diff --git a/src/dynalab_core/errors.py b/src/dynalab_core/errors.py index 5a26b5c..e4e098e 100644 --- a/src/dynalab_core/errors.py +++ b/src/dynalab_core/errors.py @@ -11,6 +11,10 @@ class CoreStateMismatchError(CoreError): """DynaLab Core state error.""" +class CoreMultipleFunctionsFoundError(CoreError): + """DynaLab Core multiple functions found while trying to create derive unit.""" + + # DLPak error declarations class DLPakError(Exception): """DLPak error.""" diff --git a/src/dynalab_core/values.py b/src/dynalab_core/values.py index d3f33b2..037fbb2 100644 --- a/src/dynalab_core/values.py +++ b/src/dynalab_core/values.py @@ -5,6 +5,9 @@ from threading import Lock from time import monotonic, monotonic_ns +from uuid import UUID + +from dynalab_core.protocols.packets.data import ValueDescriptor class Value: @@ -20,11 +23,30 @@ class Value: self._last_updated = timestamp def get(self) -> float | None: + current_value = self._get_current_value() + return current_value[0] if current_value is not None else None + + def to_descriptor(self, signal_id: UUID) -> ValueDescriptor | None: + current_value = self._get_current_value() + if current_value is None: + return None + + value, timestamp = current_value + return ValueDescriptor( + signal_id=signal_id, + value=value, + timestamp=timestamp, + ) + + def _get_current_value(self) -> tuple[float, int] | None: + with self._lock: + value = self._value + last_updated = self._last_updated + now = monotonic_ns() if ( - now > self._last_updated + self._timeout_ms * 1_000_000 - or self._last_updated == 0 + now > last_updated + self._timeout_ms * 1_000_000 + or last_updated == 0 ): return None - with self._lock: - return self._value + return value, last_updated diff --git a/test/manual/core.py b/test/manual/core.py index 47e758f..e1f6fd4 100644 --- a/test/manual/core.py +++ b/test/manual/core.py @@ -1,9 +1,13 @@ import logging +from pathlib import Path +from time import sleep +from uuid import UUID, uuid4 from rich.logging import RichHandler from dynalab_core import Core from dynalab_core.config import CoreConfig +from dynalab_core.protocols.packets.handshake import SignalDescriptor logging.basicConfig( level=logging.DEBUG, @@ -14,13 +18,39 @@ logging.basicConfig( ) log = logging.getLogger(__name__) +processing_str = Path( + "/home/hector/projects/Exergie/dynalab-core/test/manual/process.py" +).read_text(encoding="utf-8") +signal_id_1: UUID = UUID("3f32cea3-d872-4c16-a0a2-54b57171aeb2") +signal_id_2: UUID = UUID("6578ac37-99d1-410c-b3f7-d049339919b2") config = CoreConfig(port=8765) log.info("Configured manual core on %s", config.bind_str()) dl_core = Core(config) dl_core.start() + + dl_core.start_recording() +for i in range(2): + dl_core.wait(1) + values = dl_core.get_all_live_values() + log.debug(f"Core values: {values}") + +unit_id = dl_core.bind_derive_unit( + processing_str, + [ + SignalDescriptor( + id=signal_id_1, name="Dummy signal 1", type="number", timeout_ms=5000 + ), + SignalDescriptor( + id=signal_id_2, name="Dummy signal 2", type="number", timeout_ms=5000 + ), + ], + SignalDescriptor(id=uuid4(), name="Dummy sum", type="number", timeout_ms=5000), +) + + try: while True: dl_core.wait(1) diff --git a/test/manual/peer.py b/test/manual/peer.py index 9c90bd0..812c30e 100644 --- a/test/manual/peer.py +++ b/test/manual/peer.py @@ -52,7 +52,8 @@ logging.basicConfig( log = logging.getLogger(__name__) -signal_id: UUID = uuid4() +signal_id_1: UUID = UUID("3f32cea3-d872-4c16-a0a2-54b57171aeb2") +signal_id_2: UUID = UUID("6578ac37-99d1-410c-b3f7-d049339919b2") def monotonic_ms() -> int: @@ -117,13 +118,18 @@ def value_sender_thread( next_send_time += period message = ValueBatch(values=[]) - for i in range(10): - value = ValueDescriptor( - signal_id=signal_id, - value=2.0, - timestamp=monotonic_ns(), - ) - message.values.append(value) + value1 = ValueDescriptor( + signal_id=signal_id_1, + value=2.0, + timestamp=monotonic_ns(), + ) + value2 = ValueDescriptor( + signal_id=signal_id_2, + value=1.0, + timestamp=monotonic_ns(), + ) + message.values.append(value1) + message.values.append(value2) future = asyncio.run_coroutine_threadsafe( send_message( @@ -205,8 +211,11 @@ async def perform_handshake( connector_version=CONNECTOR_VERSION.get_version(), signals=[ SignalDescriptor( - id=signal_id, name="Dummy signal", type="number", timeout_ms=5000 - ) + id=signal_id_1, name="Dummy signal 1", type="number", timeout_ms=5000 + ), + SignalDescriptor( + id=signal_id_2, name="Dummy signal 2", type="number", timeout_ms=5000 + ), ], ) diff --git a/test/manual/process.py b/test/manual/process.py new file mode 100644 index 0000000..f29ebe7 --- /dev/null +++ b/test/manual/process.py @@ -0,0 +1,11 @@ +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