diff --git a/docs/API.md b/docs/API.md index af1a9a5..0d5191c 100644 --- a/docs/API.md +++ b/docs/API.md @@ -310,14 +310,23 @@ while it is paused, has not started, or has reached the end of the recording. ## Derived Signals A derive unit calculates a new value whenever one of its input signals receives -an update and all inputs have non-stale live values. Define the calculation as a -Python source string and bind it with `bind_derive_unit()`. +an update and all inputs have non-stale live values. Package the calculation, +its input descriptors, and its output descriptor in a `DLUnit`, then bind it +with `bind_derive_unit()`. -### `bind_derive_unit(function, input_signals, output_signal) -> UUID` +### `bind_derive_unit(dlunit: DLUnit | None = None, file: Path | None = None) -> UUID` + +Provide exactly one of `dlunit` or `file`: + +- `dlunit` binds an in-memory `DLUnit`. +- `file` loads and binds a `.dlunit` archive. + +To create and bind an in-memory unit: ```python from uuid import uuid4 +from dynalab_core.dlunit import DLUnit from dynalab_core.protocols.packets.handshake import SignalDescriptor processing_function = """ @@ -332,25 +341,42 @@ def add(left: ValueDescriptor, right: ValueDescriptor, output: SignalDescriptor) ) """ -output_signal = SignalDescriptor( +return_signal = SignalDescriptor( id=uuid4(), name="Total", type="number", origin="derived", ) +derive_unit = DLUnit() +derive_unit.set_manifest( + uuid4(), + [left_signal, right_signal], + return_signal, +) +derive_unit.set_data(processing_function) + +unit_id = core.bind_derive_unit(dlunit=derive_unit, file=None) +``` + +To bind a previously saved archive instead, pass `None` for `dlunit` and its +path for `file`: + +```python +from pathlib import Path + unit_id = core.bind_derive_unit( - processing_function, - input_signals=[left_signal, right_signal], - output_signal=output_signal, + dlunit=None, + file=Path("units/total.dlunit"), ) ``` -`function` must define exactly one Python function. Its parameters must be: +The unit's Python source must define exactly one function. Its parameters must +be: 1. Zero or more `ValueDescriptor` parameters, in the same order as - `input_signals`. -2. One final `SignalDescriptor` parameter for `output_signal`. + the manifest's `input_signals`. +2. One final `SignalDescriptor` parameter for the manifest's `return_signal`. 3. A return annotation of `ValueDescriptor`. The parameter annotations must use the imported `ValueDescriptor` and @@ -378,8 +404,9 @@ individual failure reason: | Exception | Area | Codes | | --- | --- | --- | -| `CoreError` | Core lifecycle, mode, and recording operations. | `invalid_state`, `invalid_mode`, `incorrect_mode`, `recording_active`, `not_recording`, `no_recording`, `no_playback_data` | -| `DLPakError` | Archive preparation, reading, and writing. | `no_data`, `no_manifest`, `read_failed`, `write_failed` | +| `CoreError` | Core lifecycle, mode, recording, and derive-unit binding operations. | `invalid_state`, `invalid_mode`, `incorrect_mode`, `recording_active`, `not_recording`, `no_recording`, `no_playback_data`, `bind_derive_failed` | +| `DLPakError` | Recording archive preparation, reading, and writing. | `no_data`, `no_manifest`, `read_failed`, `write_failed` | +| `DLUnitError` | Derive-unit archive preparation, reading, and writing. | `no_data`, `no_manifest`, `read_failed`, `write_failed` | | `DeriveError` | Derive source, signatures, arguments, and registration. | `invalid_source`, `invalid_signature`, `input_signal_mismatch`, `argument_mismatch`, `already_registered` | | `PlaybackError` | Playback state and data. | `no_data`, `incomplete_data`, `already_playing`, `invalid_timestamp`, `stopped` | | `ConnectorError` | Connector queues and registration. | `queue_full`, `already_registered` | diff --git a/src/dynalab_core/__init__.py b/src/dynalab_core/__init__.py index dadfbbf..3e38de7 100644 --- a/src/dynalab_core/__init__.py +++ b/src/dynalab_core/__init__.py @@ -19,6 +19,7 @@ from dynalab_core.config import CoreConfig from dynalab_core.constants import CORE_VERSION from dynalab_core.derive import DeriveRegistry from dynalab_core.dlpak import DLPak +from dynalab_core.dlunit import DLUnit from dynalab_core.errors import CoreError from dynalab_core.playback import PlaybackEngine from dynalab_core.protocols.endpoint import ConnectorRegistry @@ -300,12 +301,22 @@ 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, + self, dlunit: DLUnit | None = None, file: Path | None = None ) -> UUID: - unit = self._derive_registry.register(function, input_signals, output_signal) + if dlunit is not None and file is not None: + raise CoreError( + "Bind a derive unit from either a DLUnit or a file, not both", + code="bind_derive_failed", + ) + if dlunit is None and file is not None: + dlunit = DLUnit().read(file) + elif dlunit is None and file is None: + raise CoreError( + "Cannot bind derive unit with no dlunit OR file specified", + code="bind_derive_failed", + ) + + unit = self._derive_registry.register(dlunit) return unit.uuid() def unbind_derive_unit(self, unit_id: UUID) -> None: diff --git a/src/dynalab_core/derive.py b/src/dynalab_core/derive.py index 5087fd2..829d5f0 100644 --- a/src/dynalab_core/derive.py +++ b/src/dynalab_core/derive.py @@ -12,6 +12,7 @@ import threading import types from uuid import UUID, uuid4 +from dynalab_core.dlunit import DLUnit from dynalab_core.errors import DeriveError from dynalab_core.protocols.packets.data import ValueDescriptor from dynalab_core.protocols.packets.handshake import SignalDescriptor @@ -214,17 +215,11 @@ class DeriveRegistry: ) self._routing_worker_thread.start() - def register( - self, - function: str, - input_signals: list[SignalDescriptor], - output_signal: SignalDescriptor, - unit_uuid: UUID | None = None, - ) -> DeriveUnit: + def register(self, dlunit: DLUnit) -> DeriveUnit: - if unit_uuid is not None: + if dlunit._manifest.unit_uuid is not None: with self._lock: - current = self.get(unit_uuid) + current = self.get(dlunit._manifest.unit_uuid) if current is not None: log.warning( "Derive Unit %s is already registered", @@ -236,14 +231,14 @@ class DeriveRegistry: }, ) raise DeriveError( - f"Derive unit {unit_uuid} is already registered", + f"Derive unit {dlunit._manifest.unit_uuid} is already registered", code="already_registered", ) - unit_id = unit_uuid or uuid4() + unit_id = dlunit._manifest.unit_uuid or uuid4() namespace = {} try: - exec(function, namespace) + exec(dlunit._data, namespace) except Exception as error: raise DeriveError( f"Could not load derive function: {error}", @@ -261,7 +256,11 @@ class DeriveRegistry: ) unit = DeriveUnit( - unit_id, functions[0], input_signals, output_signal, self._core_input_queue + unit_id, + functions[0], + dlunit._manifest.input_signals, + dlunit._manifest.return_signal, + self._core_input_queue, ) with self._lock: @@ -273,8 +272,10 @@ class DeriveRegistry: extra={ "event": "derive.registered", "derive_uuid": str(unit_id), - "input_signal_ids": [str(signal.id) for signal in input_signals], - "output_signal_id": str(output_signal.id), + "input_signal_ids": [ + str(signal.id) for signal in dlunit._manifest.input_signals + ], + "output_signal_id": str(dlunit._manifest.return_signal.id), }, ) diff --git a/src/dynalab_core/dlunit.py b/src/dynalab_core/dlunit.py new file mode 100644 index 0000000..22ab478 --- /dev/null +++ b/src/dynalab_core/dlunit.py @@ -0,0 +1,142 @@ +# Copyright (C) 2026 Hector van der Aa +# Copyright (C) 2026 Association Exergie +# SPDX-License-Identifier: GPL-3.0-or-later + +import logging +from pathlib import Path +from typing import Literal, Self +from uuid import UUID +from zipfile import ZIP_DEFLATED, ZipFile + +from pydantic import BaseModel, Field + +from dynalab_core.errors import DLPakError, DLUnitError +from dynalab_core.protocols.packets.handshake import SignalDescriptor + + +log = logging.getLogger(__name__) + + +class DeriveManifest(BaseModel): + unit_uuid: UUID + input_signals: list[SignalDescriptor] = Field(default_factory=list) + return_signal: SignalDescriptor + comments: str = "" + + +class DLUnit: + def __init__(self) -> None: + self._dir: Path | None = None + self._filename: str | None = None + self._savetype: Literal["zip", "folder"] = "zip" + self._data: str | None = None + self._manifest: DeriveManifest | None = None + + def set_data(self, data: str) -> None: + self._data = data + + def set_manifest( + self, + unit_uuid: UUID, + input_signals: list[SignalDescriptor], + return_signal: SignalDescriptor, + ) -> None: + + self._manifest = DeriveManifest( + unit_uuid=unit_uuid, + input_signals=input_signals, + return_signal=return_signal, + ) + + def read(self, file: str | Path) -> Self: + input_path = Path(file) + + try: + with ZipFile(input_path, mode="r") as archive: + manifest = DeriveManifest.model_validate_json( + archive.read("manifest.json") + ) + data: str = archive.read("unit.py").decode("utf-8") + except Exception as error: + log.exception( + "Failed to read DLUnit archive from %s", + input_path.name, + extra={ + "event": "dlunit.read_failed", + "input_filename": input_path.name, + "exception_type": type(error).__name__, + }, + ) + raise DLUnitError( + f"Could not read DLUnit archive {input_path}: {error}", + code="read_failed", + ) from error + + self._dir = input_path.parent + self._filename = input_path.name + self._data = data + self._manifest = manifest + + log.info( + "Read DLUnit archive from %s", + input_path.name, + extra={ + "event": "dlunit.read", + "input_filename": input_path.name, + }, + ) + return self + + def write(self, dir: str | Path, filename: str) -> None: + output_dir = Path(dir) + + if self._data is None: + raise DLUnitError( + "Cannot write a DLUnit archive before setting its data", + code="no_data", + ) + if self._manifest is None: + raise DLUnitError( + "Cannot write a DLUnit archive before creating its manifest", + code="no_manifest", + ) + + if output_dir.is_file(): + output_dir = output_dir.parent + + file_path = Path(filename) + + if file_path.suffix.lower() != ".dlunit": + file_path = file_path.with_suffix(".dlunit") + + output_path = output_dir / file_path + + try: + with ZipFile(output_path, mode="w", compression=ZIP_DEFLATED) as archive: + archive.writestr( + "manifest.json", self._manifest.model_dump_json(indent=2) + ) + archive.writestr("unit.py", self._data.export_csv()) + except Exception as error: + log.exception( + "Failed to write DLUnit archive to %s", + output_path.name, + extra={ + "event": "dlunit.write_failed", + "output_filename": output_path.name, + "exception_type": type(error).__name__, + }, + ) + raise DLUnitError( + f"Could not write DLUnit archive {output_path}: {error}", + code="write_failed", + ) from error + + log.info( + "Wrote DLUnit archive to %s", + output_path.name, + extra={ + "event": "dlunit.written", + "output_filename": output_path.name, + }, + ) diff --git a/src/dynalab_core/errors.py b/src/dynalab_core/errors.py index 87f3b9d..32d7d83 100644 --- a/src/dynalab_core/errors.py +++ b/src/dynalab_core/errors.py @@ -27,6 +27,10 @@ class DLPakError(DynaLabError): """A DLPak archive cannot be prepared, read, or written.""" +class DLUnitError(DynaLabError): + """A DLPak archive cannot be prepared, read, or written.""" + + class DeriveError(DynaLabError): """A derive unit cannot be created or used as requested.""" diff --git a/test/manual/core.py b/test/manual/core.py index 5605da3..8f9de07 100644 --- a/test/manual/core.py +++ b/test/manual/core.py @@ -7,6 +7,7 @@ from rich.logging import RichHandler from dynalab_core import Core from dynalab_core.config import CoreConfig +from dynalab_core.dlunit import DLUnit from dynalab_core.protocols.packets.handshake import SignalDescriptor logging.basicConfig( @@ -29,9 +30,9 @@ log.info("Configured manual core on %s", config.bind_str()) dl_core = Core(config) dl_core.start() - -unit_id = dl_core.bind_derive_unit( - processing_str, +dl_unit = DLUnit() +dl_unit.set_manifest( + uuid4(), [ SignalDescriptor( id=signal_id_1, name="Dummy signal 1", type="number", timeout_ms=5000 @@ -42,6 +43,9 @@ unit_id = dl_core.bind_derive_unit( ], SignalDescriptor(id=uuid4(), name="Dummy sum", type="number", timeout_ms=5000), ) +dl_unit.set_data(processing_str) + +unit_id = dl_core.bind_derive_unit(dl_unit) # for i in range(10): # dl_core.wait(1) diff --git a/test/manual/core_tui.py b/test/manual/core_tui.py index d52e375..709ee9e 100644 --- a/test/manual/core_tui.py +++ b/test/manual/core_tui.py @@ -1,7 +1,9 @@ +from collections import deque import logging import select import sys import termios +from threading import Lock import time import tty from contextlib import contextmanager @@ -17,6 +19,7 @@ from rich.text import Text from dynalab_core import Core from dynalab_core.config import CoreConfig +from dynalab_core.dlunit import DLUnit from dynalab_core.errors import DynaLabError from dynalab_core.protocols.packets.handshake import SignalDescriptor @@ -24,6 +27,8 @@ from dynalab_core.protocols.packets.handshake import SignalDescriptor HOST = "127.0.0.1" PORT = 8765 DISPLAY_FREQUENCY_HZ = 5.0 +LOG_HISTORY_SIZE = 1_000 +VISIBLE_LOG_LINES = 12 signal_id_1: UUID = UUID("3f32cea3-d872-4c16-a0a2-54b57171aeb2") signal_id_2: UUID = UUID("6578ac37-99d1-410c-b3f7-d049339919b2") @@ -45,10 +50,59 @@ def keyboard_input() -> Iterator[None]: termios.tcsetattr(file_descriptor, termios.TCSADRAIN, previous_settings) +class TuiLogHandler(logging.Handler): + """Keep recent DynaLab Core log messages available to the live display.""" + + def __init__(self) -> None: + super().__init__() + self._entries: deque[tuple[str, str, str, str]] = deque( + maxlen=LOG_HISTORY_SIZE + ) + self._lock = Lock() + self.setFormatter( + logging.Formatter( + "%(asctime)s %(levelname)-8s %(name)s: %(message)s", + datefmt="%H:%M:%S", + ) + ) + + def emit(self, record: logging.LogRecord) -> None: + try: + message = record.getMessage().replace("\n", " ↳ ") + if record.exc_info is not None and record.exc_info[1] is not None: + message = f"{message} ({record.exc_info[1]})" + level = record.levelname + level_style = { + logging.DEBUG: "dim", + logging.INFO: "white", + logging.WARNING: "yellow", + logging.ERROR: "bold red", + logging.CRITICAL: "bold white on red", + }.get(record.levelno, "white") + with self._lock: + self._entries.append( + ( + self.formatter.formatTime(record, datefmt="%H:%M:%S"), + level, + message, + level_style, + ) + ) + except Exception: + self.handleError(record) + + def latest(self, limit: int) -> list[tuple[str, str, str, str]]: + with self._lock: + return list(self._entries)[-limit:] + + class Dashboard: - def __init__(self, core: Core, config: CoreConfig) -> None: + def __init__( + self, core: Core, config: CoreConfig, log_handler: TuiLogHandler + ) -> None: self.core = core self.config = config + self.log_handler = log_handler self.running = True self.playback_loaded = False self.message = "Waiting for a connector" @@ -170,6 +224,22 @@ class Dashboard: "[bold]L[/bold] Load [bold]Space[/bold] Play/Pause " "[bold]Q[/bold] Quit" ) + log_table = Table.grid(expand=True, padding=(0, 1)) + log_table.add_column(style="dim", width=8, no_wrap=True) + log_table.add_column(width=8, no_wrap=True) + log_table.add_column(ratio=1, no_wrap=True, overflow="ellipsis") + log_entries = self.log_handler.latest(VISIBLE_LOG_LINES) + if log_entries: + for timestamp, level, message, style in log_entries: + log_table.add_row( + Text(timestamp), + Text(level, style=style), + Text(message, style=style), + ) + else: + log_table.add_row("", "", Text("No DynaLab Core logs yet", style="dim")) + for _ in range(VISIBLE_LOG_LINES - max(1, len(log_entries))): + log_table.add_row("", "", "") return Group( Panel( @@ -181,6 +251,11 @@ class Dashboard: Panel(value_table, title="Live values", border_style="cyan"), Panel(Text(self.message, style=self.message_style), title="Status"), Panel(controls, title="Hotkeys", border_style="blue"), + Panel( + log_table, + title=f"Logs · latest {VISIBLE_LOG_LINES}", + border_style="magenta", + ), ) def playback_status(self, playing: bool) -> str: @@ -193,15 +268,23 @@ class Dashboard: def main() -> None: logging.basicConfig(level=logging.CRITICAL) + log_handler = TuiLogHandler() + dynalab_logger = logging.getLogger("dynalab_core") + previous_log_level = dynalab_logger.level + previous_propagation = dynalab_logger.propagate + dynalab_logger.setLevel(logging.DEBUG) + dynalab_logger.propagate = False + dynalab_logger.addHandler(log_handler) config = CoreConfig(host=HOST, port=PORT) core = Core(config) processing_source = Path(__file__).with_name("process.py").read_text( encoding="utf-8" ) - dashboard = Dashboard(core, config) + dashboard = Dashboard(core, config, log_handler) - core.bind_derive_unit( - processing_source, + derive_unit = DLUnit() + derive_unit.set_manifest( + uuid4(), [ SignalDescriptor( id=signal_id_1, @@ -224,13 +307,15 @@ def main() -> None: origin="derived", ), ) + derive_unit.set_data(processing_source) + core.bind_derive_unit(dlunit=derive_unit, file=None) try: core.start() with keyboard_input(), Live( dashboard.render(), auto_refresh=False, - screen=True, + screen=False, ) as live: while dashboard.running: update_started = time.monotonic() @@ -245,6 +330,9 @@ def main() -> None: pass finally: core.stop() + dynalab_logger.removeHandler(log_handler) + dynalab_logger.setLevel(previous_log_level) + dynalab_logger.propagate = previous_propagation if __name__ == "__main__": diff --git a/test/test_core.py b/test/test_core.py index 7589238..5b9306e 100644 --- a/test/test_core.py +++ b/test/test_core.py @@ -1,14 +1,16 @@ import logging +from pathlib import Path import threading from uuid import uuid4 import pytest from dynalab_core import Core from dynalab_core.config import CoreConfig +from dynalab_core.dlunit import DLUnit from dynalab_core.errors import CoreError from dynalab_core.protocols.endpoint import ConnectorEndpoint from dynalab_core.protocols.packets.data import ValueDescriptor -from dynalab_core.protocols.packets.handshake import ConnectorHello +from dynalab_core.protocols.packets.handshake import ConnectorHello, SignalDescriptor from test.common import find_available_port @@ -99,3 +101,34 @@ def test_core_mode_is_shared_with_endpoints() -> None: assert raised.value.code == "no_playback_data" finally: core.stop() + + +def test_core_binds_in_memory_dlunit() -> None: + core = Core(CoreConfig(port=find_available_port(8765))) + input_signal = SignalDescriptor(id=uuid4(), name="Input", type="number") + return_signal = SignalDescriptor( + id=uuid4(), name="Output", type="number", origin="derived" + ) + dlunit = DLUnit() + dlunit.set_manifest(uuid4(), [input_signal], return_signal) + dlunit.set_data( + """ +from dynalab_core.protocols.packets.data import ValueDescriptor +from dynalab_core.protocols.packets.handshake import SignalDescriptor + +def identity(value: ValueDescriptor, output: SignalDescriptor) -> ValueDescriptor: + return ValueDescriptor(signal_id=output.id, value=value.value, timestamp=value.timestamp) +""" + ) + + try: + unit_id = core.bind_derive_unit(dlunit) + assert core.get_signal_descriptor(return_signal.id) == return_signal + + with pytest.raises(CoreError, match="either a DLUnit or a file") as raised: + core.bind_derive_unit(dlunit, file=Path("unit.dlunit")) + + assert raised.value.code == "bind_derive_failed" + core.unbind_derive_unit(unit_id) + finally: + core.stop() diff --git a/test/test_derive.py b/test/test_derive.py index 05b5259..3a65780 100644 --- a/test/test_derive.py +++ b/test/test_derive.py @@ -5,6 +5,7 @@ from uuid import UUID, uuid4 import pytest from dynalab_core.derive import DeriveRegistry +from dynalab_core.dlunit import DLUnit from dynalab_core.protocols.packets.data import ValueDescriptor from dynalab_core.protocols.packets.handshake import SignalDescriptor @@ -30,6 +31,17 @@ def fail(value: ValueDescriptor, output: SignalDescriptor) -> ValueDescriptor: """ +def make_dlunit( + source: str, + input_signals: list[SignalDescriptor], + return_signal: SignalDescriptor, +) -> DLUnit: + dlunit = DLUnit() + dlunit.set_manifest(uuid4(), input_signals, return_signal) + dlunit.set_data(source) + return dlunit + + def test_derive_registry_routes_with_live_values() -> None: output_queue: Queue[ValueDescriptor] = Queue() left_signal = SignalDescriptor(id=uuid4(), name="Left", type="number") @@ -40,7 +52,11 @@ def test_derive_registry_routes_with_live_values() -> None: try: unit = registry.register( - PROCESSING_FUNCTION, [left_signal, right_signal], output_signal + make_dlunit( + PROCESSING_FUNCTION, + [left_signal, right_signal], + output_signal, + ) ) right_value = ValueDescriptor(signal_id=right_signal.id, value=2.0, timestamp=1) live_values[right_signal.id] = right_value @@ -87,7 +103,11 @@ def test_derive_worker_logs_failure_and_stops( with caplog.at_level(logging.DEBUG, logger="dynalab_core"): try: unit = registry.register( - FAILING_PROCESSING_FUNCTION, [input_signal], output_signal + make_dlunit( + FAILING_PROCESSING_FUNCTION, + [input_signal], + output_signal, + ) ) unit.put_data( [ValueDescriptor(signal_id=input_signal.id, value=1.0, timestamp=1)] diff --git a/test/test_errors.py b/test/test_errors.py index d72b463..4ce88fc 100644 --- a/test/test_errors.py +++ b/test/test_errors.py @@ -8,6 +8,7 @@ import pytest from dynalab_core.buffer import ValueBuffer from dynalab_core.derive import DeriveRegistry, DeriveUnit from dynalab_core.dlpak import DLPak +from dynalab_core.dlunit import DLUnit from dynalab_core.errors import ( CoreError, DLPakError, @@ -97,7 +98,10 @@ def test_derive_registry_wraps_invalid_source() -> None: try: with pytest.raises(DeriveError, match="Could not load") as raised: - registry.register("def broken(", [], output_signal) + dlunit = DLUnit() + dlunit.set_manifest(uuid4(), [], output_signal) + dlunit.set_data("def broken(") + registry.register(dlunit) finally: registry.stop()