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

@@ -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)

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)
 

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)

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."""

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

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)

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
+            ),
         ],
     )
 

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