diff --git a/docs/API.md b/docs/API.md index 8eaac0e..af1a9a5 100644 --- a/docs/API.md +++ b/docs/API.md @@ -1,9 +1,9 @@ # DynaLab Core API `Core` is the Python interface for running a DynaLab Core instance, inspecting -connected signals and their current values, recording incoming data, and -creating derived signals. Connectors provide data to a running core using the -[DynaLab protocol](Protocol.md). +signals and their current values, recording incoming data, playing recordings, +and creating derived signals. Connectors provide realtime data to a running +core using the [DynaLab protocol](Protocol.md). > [!WARNING] > DynaLab Core is in alpha. Its public API may change between releases. @@ -63,8 +63,9 @@ Starts the configured server and begins accepting connector connections. This method returns after the server start sequence completes. Call `start()` once per `Core` instance. Calling it in any state other than the -initial state raises `CoreStateMismatchError`. Server startup failures, such as -an address already in use, are propagated by the underlying server. +initial state raises `CoreError` with code `"invalid_state"`. Server startup +failures, such as an address already in use, raise `JsonServerError` while +retaining the underlying exception as their cause. ### `wait(timeout: float | None = None) -> None` @@ -72,15 +73,51 @@ Blocks until `stop()` is called or until `timeout` seconds have elapsed. Omit `timeout` to wait indefinitely. This is useful for keeping a command-line application alive without a busy loop. -### `stop() -> None` +### `stop() -> bool` Stops the server, disconnects connectors, signals the input worker to stop, and -stops all bound derive units. It also releases any call currently blocked in -`wait()`. +stops the playback engine and all bound derive units. It also releases any call +currently blocked in `wait()`. + +Returns `True` when the server, connector registry, derive registry, and input +worker stop within their allowed timeouts, or `False` when shutdown is +incomplete. Call this during application shutdown, including when startup or runtime work raises an exception. A stopped core cannot be started again. +## Modes + +Core operates in one of two modes: + +| Mode | Signal source | Available operation | +| --- | --- | --- | +| `"realtime"` | Connected connectors | Recording | +| `"playback"` | A loaded recording | Playback | + +The initial mode is `"realtime"`. Connected endpoints remain connected in +playback mode, but their incoming values are not routed into Core. + +`CoreMode` is defined in `dynalab_core.types` as the literal union of these two +strings. + +### `set_mode(mode: CoreMode) -> None` + +Changes the active mode and clears all cached live values. `mode` must be +`"realtime"` or `"playback"`; any other value raises `CoreError` with code +`"invalid_mode"`. + +The mode cannot be changed while a recording is active. Attempting to do so +raises `CoreError` with code `"recording_active"`. + +> [!IMPORTANT] +> Changing mode does not automatically pause active playback. Call `pause()` +> before switching from playback to realtime mode. + +### `get_mode() -> CoreMode` + +Returns the active mode as either `"realtime"` or `"playback"`. + ## Signals Signals are registered by connectors during their protocol handshake. A signal @@ -117,17 +154,20 @@ signal = SignalDescriptor( ### `get_signal_descriptor(signal_id: UUID) -> SignalDescriptor | None` Returns the descriptor for `signal_id`, including a derived-signal output, or -`None` when no matching signal is registered. Use a UUID, not its string -representation. +`None` when no matching signal is available. In realtime mode, source +descriptors come from connected connectors. In playback mode, they come from +the loaded recording. Use a UUID, not its string representation. ### `get_all_signal_descriptors() -> list[SignalDescriptor]` -Returns descriptors for every signal currently registered by connected -connectors and bound derive units. The list is empty when no connectors have -completed a handshake and no derive units are bound. +Returns the source descriptors available in the current mode followed by the +descriptors from bound derive units. In realtime mode, source descriptors come +from connected connectors. In playback mode, they come from the loaded +recording; previously recorded derived descriptors are omitted because bound +derive units recalculate those values. Connector membership is dynamic: a disconnected connector's signals are no -longer returned. +longer returned in realtime mode. ## Live Values @@ -160,12 +200,15 @@ Incoming data timestamps use the monotonic-nanosecond timebase. See ## Recording -Recording captures incoming connector values in an in-memory buffer. +Recording captures incoming connector and derived values in an in-memory +buffer. Recording is only available in realtime mode. ### `start_recording() -> None` Clears the current recording buffer, records a new UTC start timestamp, and -begins capturing subsequent values. +begins capturing subsequent values. Calling this method in playback mode raises +`CoreError` with code `"incorrect_mode"`; calling it while already recording +raises `CoreError` with code `"recording_active"`. > [!IMPORTANT] > Starting a recording discards any values captured by a previous recording. @@ -173,16 +216,96 @@ begins capturing subsequent values. ### `stop_recording() -> None` Stops capture, normalizes the recorded data, and creates a processing buffer -containing the completed recording and its signal manifest. +containing the completed recording and its signal manifest. Timestamps in the +completed recording are nanoseconds relative to its first sample. + +Calling this method in playback mode raises `CoreError` with code +`"incorrect_mode"`; calling it when no recording is active raises `CoreError` +with code `"not_recording"`. > [!IMPORTANT] > Stopping a recording replaces the previous processing buffer. Persist or > process it before stopping another recording. -There is currently no public `Core` method for retrieving or writing the -completed processing buffer. The `.dlpak` writer is implemented by the -internal `DLPak` type; applications relying on it must use private state and -therefore should expect that integration to change. +### `get_recording_state() -> bool` + +Returns `True` while a recording is active and `False` otherwise. + +### `write(dir: str | Path, filename: str) -> None` + +Writes the most recently completed recording to a DLPak archive. The `.dlpak` +extension is added to `filename` when it is not already present. + +```python +from pathlib import Path + +core.write(Path("recordings"), "dyno-run-001") +``` + +Complete a recording with `stop_recording()` before calling `write()`. If no +completed recording is available, the method raises `CoreError` with code +`"no_recording"`. Archive write failures raise `DLPakError` with code +`"write_failed"` and retain the underlying exception as their cause. + +## Playback + +Playback feeds a completed recording through the same live-value and derive +pipeline used for realtime data. Switch Core to playback mode and load data +before starting playback: + +```python +from pathlib import Path + +core.set_mode("playback") +core.load_playback_engine(Path("recordings/dyno-run-001.dlpak")) +core.play() +``` + +### `load_playback_engine(input: Path | None = None) -> None` + +Loads playback data. Pass a DLPak archive path to load a saved recording, or +omit `input` to load the recording most recently completed by this `Core` +instance. + +This method is only available in playback mode; otherwise it raises `CoreError` +with code `"invalid_mode"`. Omitting `input` before a recording has been +completed raises `CoreError` with code `"no_playback_data"`. A file that cannot +be read or validated raises `DLPakError` with code `"read_failed"`. Replacing +data while playback is active raises `PlaybackError` with code +`"already_playing"`. + +Loading data does not start playback. + +### `play() -> None` + +Starts or resumes playback from the current position. Playback stops +automatically at the end of the recording and resets its position to the +beginning. + +Calling this method outside playback mode raises `CoreError` with code +`"invalid_mode"`. Starting before data is loaded raises `PlaybackError` with +code `"no_data"`. + +### `pause() -> None` + +Pauses playback. A subsequent `play()` resumes from the current position. +Calling this method outside playback mode raises `CoreError` with code +`"invalid_mode"`. Calling it when playback is already paused has no effect. + +### `seek(timestamp_ns: int) -> None` + +Sets the playback position to `timestamp_ns`, measured in nanoseconds from the +recording's first sample. Samples after that timestamp are emitted when +playback starts or resumes. A negative timestamp raises `PlaybackError` with +code `"invalid_timestamp"`. + +Calling this method outside playback mode raises `CoreError` with code +`"invalid_mode"`. + +### `get_playback_state() -> bool` + +Returns `True` while the playback engine is actively emitting data and `False` +while it is paused, has not started, or has reached the end of the recording. ## Derived Signals @@ -234,9 +357,10 @@ The parameter annotations must use the imported `ValueDescriptor` and `SignalDescriptor` classes shown above. The source string is executed with `exec()`, so only bind code from trusted sources. -The returned UUID identifies the bound unit. Binding source that defines more -or fewer than one function raises `DeriveUnitMultipleFunctionsFoundError`; an -invalid function signature raises `DeriveUnitInvalidSignatureError`. +The returned UUID identifies the bound unit. Invalid source or function shape +raises `DeriveError`. Its code is `"invalid_source"` when the source cannot be +loaded or does not define exactly one function, and `"invalid_signature"` when +the function annotations or parameter order are invalid. ### `unbind_derive_unit(unit_id: UUID) -> None` @@ -248,10 +372,40 @@ to resume derived-value processing. ## Exceptions -Core-specific exceptions are defined in `dynalab_core.errors`. +All expected library exceptions are defined in `dynalab_core.errors` and derive +from `DynaLabError`. Exceptions are grouped by subsystem rather than by every +individual failure reason: -| Exception | Raised when | -| --- | --- | -| `CoreStateMismatchError` | `start()` is called after the core has already started or stopped. | -| `DeriveUnitMultipleFunctionsFoundError` | A derive-unit source string does not define exactly one function. | -| `DeriveUnitInvalidSignatureError` | A derive function's annotations or parameter order are invalid. | +| 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` | +| `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` | +| `JsonServerError` | JSON server startup and timeouts. | `start_timeout`, `startup_failed` | + +Every exception has a human-readable message and a stable, machine-readable +`code`. Catch a subsystem exception when the application can recover from that +area, or catch `DynaLabError` at the application's outer boundary: + +```python +from dynalab_core.errors import CoreError, DynaLabError + +try: + core.start_recording() +except CoreError as error: + if error.code == "incorrect_mode": + core.set_mode("realtime") + core.start_recording() + +try: + run_application() +except DynaLabError as error: + print(f"DynaLab operation failed ({error.code}): {error}") +``` + +Wrapped operating-system, validation, and archive errors are available through +the standard `error.__cause__` attribute. Internal transport conditions such as +a peer disconnect or an invalid incoming frame are handled by the server and +logged instead of being raised from a background task to application code. diff --git a/src/dynalab_core/__init__.py b/src/dynalab_core/__init__.py index 06179c0..dadfbbf 100644 --- a/src/dynalab_core/__init__.py +++ b/src/dynalab_core/__init__.py @@ -19,11 +19,8 @@ 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.errors import ( - CoreModeIncorrectError, - CoreModeSwitchImpossibleError, - CoreStateMismatchError, -) +from dynalab_core.errors import CoreError +from dynalab_core.playback import PlaybackEngine from dynalab_core.protocols.endpoint import ConnectorRegistry from dynalab_core.protocols.common import VersionDescriptor from dynalab_core.protocols.json.server import JsonServer @@ -61,6 +58,8 @@ class Core: ) self._input_worker_thread.start() + self._playback_engine = PlaybackEngine(self._data_input_queue) + self._derive_registry = DeriveRegistry( self._data_input_queue, self._get_live_value_descriptor ) @@ -80,8 +79,10 @@ class Core: "expected_state": "initd", }, ) - raise CoreStateMismatchError( - f'Unable to start DynaLab Core, expected state to be "initd", found {self._state}' + raise CoreError( + f"Cannot start DynaLab Core while it is {self._state!r}; " + "a Core instance can only be started once", + code="invalid_state", ) log.info( @@ -114,6 +115,7 @@ class Core: extra={"event": "core.stopping", "core_state": self._state}, ) json_server_stopped = self._json_server.stop() + self._playback_engine.stop() connector_registry_stopped = self._connector_registry.stop() derive_registry_stopped = self._derive_registry.stop() self._stop_event.set() @@ -157,16 +159,54 @@ class Core: return stopped def set_mode(self, mode: CoreMode) -> None: + if mode not in ("realtime", "playback"): + raise CoreError( + f"Unknown core mode {mode!r}; expected 'realtime' or 'playback'", + code="invalid_mode", + ) if self._recording.is_set(): - raise CoreModeSwitchImpossibleError + raise CoreError( + "Cannot change core mode while a recording is active", + code="recording_active", + ) self._mode.value = mode with self._live_values_lock: self._live_values = {} + def get_mode(self) -> CoreMode: + return self._mode.value + + def load_playback_engine(self, input: Path | None = None) -> None: + if self._mode.value != "playback": + raise CoreError( + "Cannot load playback unless the core is in 'playback' mode", + code="invalid_mode", + ) + if input is None: + if self._processing_buffer is None: + raise CoreError( + "Cannot load playback because no recording is available", + code="no_playback_data", + ) + loaded_data = self._processing_buffer + else: + loaded_data = DLPak().read(input) + self._playback_engine.set_data(loaded_data) + + # def unload_playback_engine + def start_recording(self) -> None: if self._mode.value != "realtime": - raise CoreModeIncorrectError + raise CoreError( + "Cannot start recording unless the core is in 'realtime' mode", + code="incorrect_mode", + ) + if self._recording.is_set(): + raise CoreError( + "Cannot start recording because a recording is already active", + code="recording_active", + ) self._recording_buffer.clear() self._recording_timestamp = datetime.now(timezone.utc) self._recording.set() @@ -180,7 +220,15 @@ class Core: def stop_recording(self) -> None: if self._mode.value != "realtime": - raise CoreModeIncorrectError + raise CoreError( + "Cannot stop recording unless the core is in 'realtime' mode", + code="incorrect_mode", + ) + if not self._recording.is_set(): + raise CoreError( + "Cannot stop recording because no recording is active", + code="not_recording", + ) self._recording.clear() time.sleep(1) self._recording_buffer.normalize() @@ -213,17 +261,29 @@ class Core: }, ) + def get_recording_state(self) -> bool: + return self._recording.is_set() + def get_signal_descriptor(self, signal_id: UUID) -> SignalDescriptor | None: - signal = self._connector_registry.get_signal_descriptor(signal_id) + signal: SignalDescriptor | None = None + if self._mode.value == "realtime": + signal = self._connector_registry.get_signal_descriptor(signal_id) + elif self._mode.value == "playback": + signal = self._playback_engine.get_signal_descriptor(signal_id) + if signal is None: signal = self._derive_registry.get_signal_descriptor(signal_id) return signal def get_all_signal_descriptors(self) -> list[SignalDescriptor]: - connector_signals = self._connector_registry.get_all_signal_descriptors() + root_signals: list[SignalDescriptor] = [] + if self._mode.value == "realtime": + root_signals = self._connector_registry.get_all_signal_descriptors() + elif self._mode.value == "playback": + root_signals = self._playback_engine.get_all_signal_descriptors() derive_signals = self._derive_registry.get_all_signal_descriptors() - return connector_signals + derive_signals + return root_signals + derive_signals def get_live_value(self, signal_id: UUID) -> float | None: with self._live_values_lock: @@ -252,8 +312,40 @@ class Core: self._derive_registry.unregister(unit_id) def write(self, dir: str | Path, filename: str) -> None: - if self._processing_buffer is not None: - self._processing_buffer.write(dir, filename) + if self._processing_buffer is None: + raise CoreError( + "Cannot write recording because no recording is available", + code="no_recording", + ) + self._processing_buffer.write(dir, filename) + + def play(self) -> None: + if self._mode.value != "playback": + raise CoreError( + "Cannot play unless the core is in 'playback' mode", + code="invalid_mode", + ) + self._playback_engine.play() + + def pause(self) -> None: + if self._mode.value != "playback": + raise CoreError( + "Cannot pause unless the core is in 'playback' mode", + code="invalid_mode", + ) + self._playback_engine.pause() + + def seek(self, timestamp_ns: int) -> None: + if self._mode.value != "playback": + raise CoreError( + "Cannot seek unless the core is in 'playback' mode", + code="invalid_mode", + ) + self._playback_engine.seek(timestamp_ns) + + def get_playback_state(self) -> bool: + # TODO: internalize playback state to PlaybackEngine + return self._playback_engine._playing.is_set() def _get_live_value_descriptor(self, signal_id: UUID) -> ValueDescriptor | None: with self._live_values_lock: diff --git a/src/dynalab_core/buffer.py b/src/dynalab_core/buffer.py index c2c55e3..8eb89e8 100644 --- a/src/dynalab_core/buffer.py +++ b/src/dynalab_core/buffer.py @@ -2,6 +2,7 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later +from bisect import bisect_right from collections import defaultdict import csv import io @@ -46,6 +47,62 @@ class ValueBuffer: with self._lock: self._samples.sort(key=lambda sample: sample[0]) + def import_csv(self, input: str) -> None: + reader = csv.reader(io.StringIO(input, newline="")) + + header = next(reader, None) + + if header is None: + raise ValueError("CSV input is empty") + + if not header or header[0] != "timestamp_ns": + raise ValueError("The first CSV column must be 'timestamp_ns'") + + try: + signal_ids = [UUID(column) for column in header[1:]] + except ValueError as error: + raise ValueError( + "CSV contains an invalid signal UUID in its header" + ) from error + + if len(signal_ids) != len(set(signal_ids)): + raise ValueError("CSV contains duplicate signal UUID columns") + + imported_samples: list[tuple[int, UUID, float]] = [] + + for line_number, row in enumerate(reader, start=2): + if not row or all(cell == "" for cell in row): + continue + if len(row) != len(header): + raise ValueError( + f"Line {line_number} has {len(row)} columns; expected {len(header)}" + ) + + try: + timestamp_ns = int(row[0]) + except ValueError as error: + raise ValueError( + f"Line {line_number} has an invalid timestamp: {row[0]!r}" + ) from error + + for signal_id, cell in zip(signal_ids, row[1:]): + if cell == "": + continue + + try: + value = float(cell) + except ValueError as error: + raise ValueError( + f"Line {line_number} has an invalid value: {cell!r}" + ) from error + + imported_samples.append((timestamp_ns, signal_id, value)) + + imported_samples.sort(key=lambda sample: sample[0]) + + with self._lock: + self._samples = imported_samples + def export_csv(self) -> str: with self._lock: samples = list(self._samples) @@ -81,3 +138,12 @@ class ValueBuffer: ) return output.getvalue() + + def get_samples_from(self, timestamp: int) -> list[tuple[int, UUID, float]]: + with self._lock: + index = bisect_right( + self._samples, + timestamp, + key=lambda sample: sample[0], + ) + return self._samples[index:].copy() diff --git a/src/dynalab_core/derive.py b/src/dynalab_core/derive.py index 1eac251..5087fd2 100644 --- a/src/dynalab_core/derive.py +++ b/src/dynalab_core/derive.py @@ -12,12 +12,7 @@ import threading import types from uuid import UUID, uuid4 -from dynalab_core.errors import ( - DeriveRegistryAlreadyRegisteredError, - DeriveUnitArgsMismatchError, - DeriveUnitInvalidSignatureError, - DeriveUnitMultipleFunctionsFoundError, -) +from dynalab_core.errors import DeriveError from dynalab_core.protocols.packets.data import ValueDescriptor from dynalab_core.protocols.packets.handshake import SignalDescriptor @@ -64,13 +59,35 @@ class DeriveUnit: self._parser_state = ParserState.SIGNAL_DONE continue else: - raise DeriveUnitInvalidSignatureError + raise DeriveError( + f"Parameter {name!r} must be annotated as SignalDescriptor", + code="invalid_signature", + ) if self._parser_state == ParserState.SIGNAL_DONE: - raise DeriveUnitInvalidSignatureError + raise DeriveError( + f"Parameter {name!r} appears after the output signal parameter", + code="invalid_signature", + ) if sig.return_annotation is not ValueDescriptor: - raise DeriveUnitInvalidSignatureError + raise DeriveError( + "The derive function must return ValueDescriptor", + code="invalid_signature", + ) + + if self._parser_state != ParserState.SIGNAL_DONE: + raise DeriveError( + "The derive function must end with one SignalDescriptor parameter", + code="invalid_signature", + ) + + if self._num_input_args != len(self._input_signals): + raise DeriveError( + f"Derive function accepts {self._num_input_args} input value(s), but " + f"{len(self._input_signals)} input signal(s) were configured", + code="input_signal_mismatch", + ) self._worker_thread: Thread = Thread( target=self._worker_function, name=f"{self._id}_worker_thread", daemon=True @@ -142,7 +159,11 @@ class DeriveUnit: 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 + raise DeriveError( + f"Derive unit expected {self._num_input_args} input value(s), " + f"received {len(input_args)}", + code="argument_mismatch", + ) def get_input_signals(self) -> list[SignalDescriptor]: return self._input_signals @@ -214,18 +235,30 @@ class DeriveRegistry: "reason": "duplicate_uuid", }, ) - raise DeriveRegistryAlreadyRegisteredError + raise DeriveError( + f"Derive unit {unit_uuid} is already registered", + code="already_registered", + ) unit_id = unit_uuid or uuid4() namespace = {} - exec(function, namespace) + try: + exec(function, namespace) + except Exception as error: + raise DeriveError( + f"Could not load derive function: {error}", + code="invalid_source", + ) from error functions = [ obj for obj in namespace.values() if isinstance(obj, types.FunctionType) ] if len(functions) != 1: - raise DeriveUnitMultipleFunctionsFoundError + raise DeriveError( + f"Derive source must define exactly one function; found {len(functions)}", + code="invalid_source", + ) unit = DeriveUnit( unit_id, functions[0], input_signals, output_signal, self._core_input_queue diff --git a/src/dynalab_core/dlpak.py b/src/dynalab_core/dlpak.py index cb79622..a40efe2 100644 --- a/src/dynalab_core/dlpak.py +++ b/src/dynalab_core/dlpak.py @@ -5,13 +5,13 @@ from datetime import datetime, timezone import logging from pathlib import Path -from typing import Literal +from typing import Literal, Self from zipfile import ZIP_DEFLATED, ZipFile from pydantic import BaseModel, Field from dynalab_core.buffer import ValueBuffer -from dynalab_core.errors import DLPakNoDataError +from dynalab_core.errors import DLPakError from dynalab_core.protocols.packets.handshake import SignalDescriptor log = logging.getLogger(__name__) @@ -38,7 +38,10 @@ class DLPak: self, timestamp: datetime, all_signals: list[SignalDescriptor] ) -> None: if self._data is None: - raise DLPakNoDataError + raise DLPakError( + "Cannot create a DLPak manifest before setting its data", + code="no_data", + ) data_signal_ids = self._data.get_signal_ids() @@ -52,9 +55,62 @@ class DLPak: timestamp=timestamp, signals=manifest_signals, comments="" ) + def read(self, file: str | Path) -> Self: + input_path = Path(file) + + try: + with ZipFile(input_path, mode="r") as archive: + manifest = RecordManifest.model_validate_json( + archive.read("manifest.json") + ) + data = ValueBuffer() + data.import_csv(archive.read("data.csv").decode("utf-8")) + except Exception as error: + log.exception( + "Failed to read DLPak archive from %s", + input_path.name, + extra={ + "event": "dlpak.read_failed", + "input_filename": input_path.name, + "exception_type": type(error).__name__, + }, + ) + raise DLPakError( + f"Could not read DLPak 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 DLPak archive from %s", + input_path.name, + extra={ + "event": "dlpak.read", + "input_filename": input_path.name, + "sample_count": len(data), + "signal_count": len(manifest.signals), + }, + ) + return self + def write(self, dir: str | Path, filename: str) -> None: output_dir = Path(dir) + if self._data is None: + raise DLPakError( + "Cannot write a DLPak archive before setting its data", + code="no_data", + ) + if self._manifest is None: + raise DLPakError( + "Cannot write a DLPak archive before creating its manifest", + code="no_manifest", + ) + if output_dir.is_file(): output_dir = output_dir.parent @@ -81,7 +137,10 @@ class DLPak: "exception_type": type(error).__name__, }, ) - raise + raise DLPakError( + f"Could not write DLPak archive {output_path}: {error}", + code="write_failed", + ) from error log.info( "Wrote DLPak archive to %s", diff --git a/src/dynalab_core/errors.py b/src/dynalab_core/errors.py index 008e6ad..87f3b9d 100644 --- a/src/dynalab_core/errors.py +++ b/src/dynalab_core/errors.py @@ -2,53 +2,53 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later -# Core error declarations -class CoreError(Exception): - """DynaLab Core error.""" +"""Public exceptions raised by DynaLab Core. + +Catch a subsystem exception when recovery depends on where an operation failed, +or catch :class:`DynaLabError` to handle every expected library error. The +``code`` attribute is stable and intended for programmatic decisions; the +exception message is written for people. +""" -class CoreModeIncorrectError(CoreError): - """DynaLab Core mode error.""" +class DynaLabError(Exception): + """Base class for expected errors raised by DynaLab Core.""" + + def __init__(self, message: str, *, code: str) -> None: + self.code = code + super().__init__(message) -class CoreModeSwitchImpossibleError(CoreError): - """DynaLab Core mode error.""" +class CoreError(DynaLabError): + """The core cannot perform the requested lifecycle or mode operation.""" -class CoreStateMismatchError(CoreError): - """DynaLab Core state error.""" +class DLPakError(DynaLabError): + """A DLPak archive cannot be prepared, read, or written.""" -# DLPak error declarations -class DLPakError(Exception): - """DLPak error.""" +class DeriveError(DynaLabError): + """A derive unit cannot be created or used as requested.""" -class DLPakNoDataError(DLPakError): - """DLPak no data error.""" +class PlaybackError(DynaLabError): + """The playback engine cannot perform the requested operation.""" -# DeriveUnit error declarations -class DeriveUnitError(Exception): - """DeriveUnit error.""" +class ConnectorError(DynaLabError): + """A connector endpoint or registry operation failed.""" -class DeriveUnitInvalidSignatureError(DeriveUnitError): - """DeriveUnit invalid processing function signature error.""" +class JsonServerError(DynaLabError): + """The JSON protocol server cannot perform the requested operation.""" -class DeriveUnitArgsMismatchError(DeriveUnitError): - """DeriveUnit incorrect arguments provided error.""" - - -class DeriveUnitMultipleFunctionsFoundError(DeriveUnitError): - """DynaLab Core multiple functions found while trying to create derive unit.""" - - -# DeriveRegistry error declarations -class DeriveRegistryError(Exception): - """Generic ConnectorRegistry Error""" - - -class DeriveRegistryAlreadyRegisteredError(DeriveRegistryError): - """ConnectorRegistry endpoint already registered""" +__all__ = [ + "ConnectorError", + "CoreError", + "DLPakError", + "DeriveError", + "DynaLabError", + "JsonServerError", + "PlaybackError", +] diff --git a/src/dynalab_core/playback.py b/src/dynalab_core/playback.py index 7ace54f..5859290 100644 --- a/src/dynalab_core/playback.py +++ b/src/dynalab_core/playback.py @@ -3,6 +3,155 @@ # SPDX-License-Identifier: GPL-3.0-or-later +import logging +from queue import Queue +import threading +import time +from uuid import UUID + +from dynalab_core.dlpak import DLPak +from dynalab_core.errors import PlaybackError +from dynalab_core.protocols.packets import ProtocolMessage +from dynalab_core.protocols.packets.data import ValueDescriptor +from dynalab_core.protocols.packets.handshake import SignalDescriptor + +log = logging.getLogger(__name__) + + class PlaybackEngine: - def __init__(self) -> None: - pass + def __init__(self, data_queue: Queue[ProtocolMessage]) -> None: + self._data: DLPak | None = None + self._data_queue: Queue[ProtocolMessage] = data_queue + self._playing: threading.Event = threading.Event() + self._internal_timestamp: int = 0 + self._start_timestamp: int = 0 + self._stop_event: threading.Event = threading.Event() + self._stopped_event: threading.Event = threading.Event() + self._ignored_signals: set[UUID] = set() + self._worker_thread: threading.Thread = threading.Thread( + target=self._playback_worker, name="playback_worker_thread", daemon=True + ) + self._worker_thread.start() + + def stop(self) -> None: + self._stop_event.set() + self._stopped_event.wait(5.0) + + def play(self) -> None: + if self._stop_event.is_set(): + raise PlaybackError( + "Cannot start playback after the playback engine has stopped", + code="stopped", + ) + if self._data is None: + raise PlaybackError( + "Cannot start playback before loading data", + code="no_data", + ) + self._playing.set() + + def pause(self) -> None: + self._playing.clear() + + def seek(self, timestamp_ns: int) -> None: + if timestamp_ns < 0: + raise PlaybackError( + "Playback timestamp cannot be negative", + code="invalid_timestamp", + ) + self._internal_timestamp = timestamp_ns + + def get_signal_descriptor(self, uuid: UUID) -> SignalDescriptor | None: + if self._data is not None and self._data._manifest is not None: + for signal in self._data._manifest.signals: + if signal.id == uuid: + return signal + return None + + def get_all_signal_descriptors(self) -> list[SignalDescriptor]: + if self._data is not None and self._data._manifest is not None: + return [ + signal + for signal in self._data._manifest.signals + if signal.origin != "derived" + ] + + return list[SignalDescriptor]() + + def set_data(self, data: DLPak) -> None: + if self._playing.is_set(): + raise PlaybackError( + "Cannot replace playback data while playback is active", + code="already_playing", + ) + if data._data is None or data._manifest is None: + raise PlaybackError( + "Cannot load an incomplete DLPak archive for playback", + code="incomplete_data", + ) + self._data = data + self._ignored_signals = { + signal.id + for signal in self._data._manifest.signals + if signal.origin == "derived" + } + + def _update_internal_timestamp(self) -> None: + self._internal_timestamp = time.monotonic_ns() - self._start_timestamp + + def _playback_worker(self) -> None: + try: + self._run_playback() + except Exception as error: + log.exception( + "Playback worker failed", + extra={ + "event": "playback.worker_failed", + "thread_name": threading.current_thread().name, + "exception_type": type(error).__name__, + }, + ) + finally: + self._playing.clear() + self._stopped_event.set() + + def _run_playback(self) -> None: + while not self._stop_event.is_set(): + if not self._playing.is_set(): + time.sleep(0.05) + continue + + if self._data is None: + self._playing.clear() + continue + samples = self._data._data.get_samples_from(self._internal_timestamp) + + if not samples: + self._playing.clear() + continue + + self._start_timestamp = time.monotonic_ns() - self._internal_timestamp + for timestamp, uuid, value in samples: + if uuid in self._ignored_signals: + continue + while timestamp > self._internal_timestamp: + if not self._playing.is_set() or self._stop_event.is_set(): + break + self._update_internal_timestamp() + remaining = timestamp - self._internal_timestamp + + if remaining > 1_000_000: + time.sleep((remaining - 500_000) / 1_000_000_000) + + if not self._playing.is_set() or self._stop_event.is_set(): + break + self._data_queue.put( + ValueDescriptor( + signal_id=uuid, + value=value, + timestamp=timestamp + self._start_timestamp, + ) + ) + + self._playing.clear() + self._internal_timestamp = 0 diff --git a/src/dynalab_core/protocols/endpoint.py b/src/dynalab_core/protocols/endpoint.py index 64d4079..47b401b 100644 --- a/src/dynalab_core/protocols/endpoint.py +++ b/src/dynalab_core/protocols/endpoint.py @@ -11,10 +11,7 @@ from uuid import UUID from dynalab_core.types import CoreModeState from dynalab_core.constants import HELLO_PACKET, INTERNAL_CONNECTOR_HELLO -from dynalab_core.protocols.errors import ( - ConnectorEndpointQueueFullError, - ConnectorRegistryAlreadyRegisteredError, -) +from dynalab_core.errors import ConnectorError from dynalab_core.protocols.packets import ProtocolMessage from dynalab_core.protocols.packets.data import ValueBatch, ValueDescriptor from dynalab_core.protocols.packets.handshake import ( @@ -135,11 +132,14 @@ class ConnectorEndpoint: def put_ingress_packet(self, packet: ProtocolMessage) -> None: try: self._packet_ingress_queue.put_nowait(packet) - except Full: + except Full as error: self._log_queue_full( "ingress", self._packet_ingress_queue, action="rejected" ) - raise ConnectorEndpointQueueFullError + raise ConnectorError( + f"Connector {self.uuid()} ingress queue is full", + code="queue_full", + ) from error def get_egress_packet(self, timeout: float | None) -> ProtocolMessage: return self._packet_egress_queue.get(block=True, timeout=timeout) @@ -156,9 +156,12 @@ class ConnectorEndpoint: def _put_egress_packet(self, packet: ProtocolMessage) -> None: try: self._packet_egress_queue.put_nowait(packet) - except Full: + except Full as error: self._log_queue_full("egress", self._packet_egress_queue, action="rejected") - raise ConnectorEndpointQueueFullError + raise ConnectorError( + f"Connector {self.uuid()} egress queue is full", + code="queue_full", + ) from error def _log_queue_full( self, @@ -517,7 +520,10 @@ class ConnectorRegistry: "reason": "duplicate_uuid", }, ) - raise ConnectorRegistryAlreadyRegisteredError + raise ConnectorError( + f"Connector {hello.connector_uuid} is already registered", + code="already_registered", + ) endpoint = ConnectorEndpoint( hello, diff --git a/src/dynalab_core/protocols/errors.py b/src/dynalab_core/protocols/errors.py index e443cf7..e5020f6 100644 --- a/src/dynalab_core/protocols/errors.py +++ b/src/dynalab_core/protocols/errors.py @@ -2,21 +2,12 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later -# ConnectorEndpoint error declarations -class ConnectorEndpointError(Exception): - """Generic ConnectorEndpoint Error""" +"""Protocol exception compatibility import. +``ConnectorError`` is defined with the rest of the public exception hierarchy +in :mod:`dynalab_core.errors`. +""" -class ConnectorEndpointQueueFullError(ConnectorEndpointError): - """ConnectorEndpoing queue is full""" +from dynalab_core.errors import ConnectorError - -# ConnectorRegistry error declarations - - -class ConnectorRegistryError(Exception): - """Generic ConnectorRegistry Error""" - - -class ConnectorRegistryAlreadyRegisteredError(ConnectorRegistryError): - """ConnectorRegistry endpoint already registered""" +__all__ = ["ConnectorError"] diff --git a/src/dynalab_core/protocols/json/errors.py b/src/dynalab_core/protocols/json/errors.py index cee5d1b..d16db80 100644 --- a/src/dynalab_core/protocols/json/errors.py +++ b/src/dynalab_core/protocols/json/errors.py @@ -2,25 +2,12 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later -# JsonServer error declarations -class JsonServerError(Exception): - """Generic JsonServer Error""" +"""JSON server exception compatibility import. +``JsonServerError`` is defined with the rest of the public exception hierarchy +in :mod:`dynalab_core.errors`. +""" -class JsonServerTimeoutError(JsonServerError): - """JsonServer timed out""" +from dynalab_core.errors import JsonServerError - -class JsonServerStartupError(JsonServerError): - """JsonServer crashed on startup""" - - def __init__(self, error: Exception) -> None: - self.error = error - super().__init__(f"JSON server failed to start: {error}") - - -class JsonServerValueError(JsonServerError): - """JsonServer incorrect value detected""" - - def __init__(self, message: str) -> None: - super().__init__(str) +__all__ = ["JsonServerError"] diff --git a/src/dynalab_core/protocols/json/server.py b/src/dynalab_core/protocols/json/server.py index 7ad120b..80509d5 100644 --- a/src/dynalab_core/protocols/json/server.py +++ b/src/dynalab_core/protocols/json/server.py @@ -15,11 +15,7 @@ from uuid import uuid4 from dynalab_core.config import CoreConfig from dynalab_core.constants import HELLO_PACKET from dynalab_core.protocols.endpoint import ConnectorEndpoint, ConnectorRegistry -from dynalab_core.protocols.errors import ConnectorRegistryAlreadyRegisteredError -from dynalab_core.protocols.json.errors import ( - JsonServerStartupError, - JsonServerTimeoutError, -) +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 @@ -70,10 +66,16 @@ class JsonServer: "timeout_s": self._timeout, }, ) - raise JsonServerTimeoutError + raise JsonServerError( + f"JSON server did not start within {self._timeout:.1f} seconds", + code="start_timeout", + ) if self._startup_error is not None: - raise JsonServerStartupError(self._startup_error) + raise JsonServerError( + f"JSON server failed to start: {self._startup_error}", + code="startup_failed", + ) from self._startup_error bound_addresses = [ str(sock.getsockname()) for sock in (self._server.sockets or []) @@ -404,17 +406,30 @@ class JsonServer: **context, }, ) - except ConnectorRegistryAlreadyRegisteredError: - reason = "duplicate_connector" - log.debug( - "Connection %s rejected because its connector is already registered", - connection_id, - extra={ - "event": "json_connection.rejected", - "reason": reason, - **context, - }, - ) + except ConnectorError as error: + if error.code == "already_registered": + reason = "duplicate_connector" + log.debug( + "Connection %s rejected because its connector is already registered", + connection_id, + extra={ + "event": "json_connection.rejected", + "reason": reason, + **context, + }, + ) + else: + reason = "connector_error" + log.exception( + "Connection %s failed in the connector subsystem", + connection_id, + extra={ + "event": "json_connection.failed", + "reason": reason, + "error_code": error.code, + **context, + }, + ) except Exception as error: reason = "handler_failed" log.exception( diff --git a/test/manual/core.py b/test/manual/core.py index 9af786e..5605da3 100644 --- a/test/manual/core.py +++ b/test/manual/core.py @@ -43,12 +43,12 @@ unit_id = dl_core.bind_derive_unit( SignalDescriptor(id=uuid4(), name="Dummy sum", type="number", timeout_ms=5000), ) -for i in range(10): - dl_core.wait(1) - values = dl_core.get_all_live_values() - log.debug(f"Core values: {values}") - -dl_core.set_mode("playback") +# for i in range(10): +# dl_core.wait(1) +# values = dl_core.get_all_live_values() +# log.debug(f"Core values: {values}") +# +# dl_core.set_mode("playback") try: diff --git a/test/manual/core_tui.py b/test/manual/core_tui.py new file mode 100644 index 0000000..d52e375 --- /dev/null +++ b/test/manual/core_tui.py @@ -0,0 +1,251 @@ +import logging +import select +import sys +import termios +import time +import tty +from contextlib import contextmanager +from pathlib import Path +from typing import Iterator +from uuid import UUID, uuid4 + +from rich.console import Group +from rich.live import Live +from rich.panel import Panel +from rich.table import Table +from rich.text import Text + +from dynalab_core import Core +from dynalab_core.config import CoreConfig +from dynalab_core.errors import DynaLabError +from dynalab_core.protocols.packets.handshake import SignalDescriptor + + +HOST = "127.0.0.1" +PORT = 8765 +DISPLAY_FREQUENCY_HZ = 5.0 + +signal_id_1: UUID = UUID("3f32cea3-d872-4c16-a0a2-54b57171aeb2") +signal_id_2: UUID = UUID("6578ac37-99d1-410c-b3f7-d049339919b2") + + +@contextmanager +def keyboard_input() -> Iterator[None]: + """Read individual keys and restore the terminal settings on exit.""" + if not sys.stdin.isatty(): + raise RuntimeError("This example must be run in an interactive terminal") + + file_descriptor = sys.stdin.fileno() + previous_settings = termios.tcgetattr(file_descriptor) + tty.setcbreak(file_descriptor) + + try: + yield + finally: + termios.tcsetattr(file_descriptor, termios.TCSADRAIN, previous_settings) + + +class Dashboard: + def __init__(self, core: Core, config: CoreConfig) -> None: + self.core = core + self.config = config + self.running = True + self.playback_loaded = False + self.message = "Waiting for a connector" + self.message_style = "dim" + + def handle_key(self, key: str) -> None: + try: + if key.lower() == "q": + self.running = False + elif key.lower() == "r": + self.toggle_recording() + elif key.lower() == "m": + self.toggle_mode() + elif key.lower() == "l": + self.load_playback() + elif key == " ": + self.toggle_playback() + except DynaLabError as error: + self.set_message(str(error), "bold red") + + def toggle_recording(self) -> None: + if self.core.get_mode() != "realtime": + self.set_message("Recording is only available in realtime mode", "yellow") + return + + if self.core.get_recording_state(): + self.core.stop_recording() + self.set_message("Recording stopped and prepared for playback", "green") + else: + self.core.start_recording() + self.set_message("Recording started", "bold red") + + def toggle_mode(self) -> None: + if self.core.get_recording_state(): + self.set_message("Stop recording before changing mode", "yellow") + return + + if self.core.get_mode() == "realtime": + self.core.set_mode("playback") + self.set_message( + "Playback mode selected; press L to load a recording", "green" + ) + else: + if self.core.get_playback_state(): + self.core.pause() + self.core.set_mode("realtime") + self.set_message("Realtime mode selected", "green") + + def load_playback(self) -> None: + if self.core.get_mode() != "playback": + self.set_message("Switch to playback mode before loading", "yellow") + return + + self.core.load_playback_engine() + self.playback_loaded = True + self.set_message("Internal processing buffer loaded", "green") + + def toggle_playback(self) -> None: + if self.core.get_mode() != "playback": + self.set_message("Switch to playback mode before pressing Space", "yellow") + return + + if self.core.get_playback_state(): + self.core.pause() + self.set_message("Playback paused", "yellow") + else: + self.core.play() + self.set_message("Playback started", "green") + + def set_message(self, message: str, style: str) -> None: + self.message = message + self.message_style = style + + def render(self) -> Group: + mode = self.core.get_mode() + recording = self.core.get_recording_state() + playing = self.core.get_playback_state() + + status = Table.grid(expand=True) + status.add_column(ratio=1) + status.add_column(ratio=1) + status.add_column(ratio=1) + status.add_row( + Text(f"Mode: {mode.title()}", style="bold cyan"), + Text( + "Recording: ON" if recording else "Recording: OFF", + style="bold red" if recording else "dim", + ), + Text( + self.playback_status(playing), + style="bold green" if playing else "dim", + ), + ) + + values = self.core.get_all_live_values() + descriptors = sorted( + self.core.get_all_signal_descriptors(), key=lambda signal: signal.name + ) + value_table = Table(expand=True, show_lines=False) + value_table.add_column("Signal", style="cyan", ratio=2) + value_table.add_column("Origin", style="dim", ratio=1) + value_table.add_column("Value", justify="right", ratio=1) + value_table.add_column("Unit", style="dim", ratio=1) + + if descriptors: + for descriptor in descriptors: + value = values.get(descriptor.id) + value_table.add_row( + descriptor.name, + descriptor.origin, + f"{value: .5f}" if value is not None else "—", + descriptor.unit or "", + ) + else: + value_table.add_row("Waiting for connector…", "", "—", "") + + controls = Text.from_markup( + "[bold]R[/bold] Record [bold]M[/bold] Mode " + "[bold]L[/bold] Load [bold]Space[/bold] Play/Pause " + "[bold]Q[/bold] Quit" + ) + + return Group( + Panel( + status, + title="DynaLab Core", + subtitle=f"{self.config.bind_str()} · values refresh at 5 Hz", + border_style="blue", + ), + 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"), + ) + + def playback_status(self, playing: bool) -> str: + if playing: + return "Playback: PLAYING" + if self.playback_loaded: + return "Playback: LOADED" + return "Playback: NOT LOADED" + + +def main() -> None: + logging.basicConfig(level=logging.CRITICAL) + 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) + + core.bind_derive_unit( + processing_source, + [ + 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 product", + type="number", + timeout_ms=5000, + origin="derived", + ), + ) + + try: + core.start() + with keyboard_input(), Live( + dashboard.render(), + auto_refresh=False, + screen=True, + ) as live: + while dashboard.running: + update_started = time.monotonic() + + while select.select([sys.stdin], [], [], 0)[0]: + dashboard.handle_key(sys.stdin.read(1)) + + live.update(dashboard.render(), refresh=True) + elapsed = time.monotonic() - update_started + time.sleep(max(0.0, 1.0 / DISPLAY_FREQUENCY_HZ - elapsed)) + except KeyboardInterrupt: + pass + finally: + core.stop() + + +if __name__ == "__main__": + main() diff --git a/test/manual/peer.py b/test/manual/peer.py index 812c30e..4d39036 100644 --- a/test/manual/peer.py +++ b/test/manual/peer.py @@ -3,6 +3,7 @@ import logging import threading from concurrent.futures import CancelledError as FutureCancelledError from concurrent.futures import TimeoutError as FutureTimeoutError +from math import sin, tau from statistics import mean from time import monotonic, monotonic_ns, perf_counter from uuid import UUID, uuid4 @@ -27,6 +28,7 @@ PORT = 8765 # Set to None to send as quickly as possible. # For a controlled rate, use something like 5_000.0. TARGET_FREQUENCY_HZ: float | None = 1000 +SINE_FREQUENCY_HZ = 0.5 FREQUENCY_SAMPLE_SIZE = 20000 @@ -88,6 +90,7 @@ def value_sender_thread( """ intervals: list[float] = [] previous_send_time: float | None = None + sine_start_time = perf_counter() if TARGET_FREQUENCY_HZ is not None: period = 1.0 / TARGET_FREQUENCY_HZ @@ -117,16 +120,20 @@ def value_sender_thread( next_send_time += period + sine_value = sin( + tau * SINE_FREQUENCY_HZ * (perf_counter() - sine_start_time) + ) + timestamp = monotonic_ns() message = ValueBatch(values=[]) value1 = ValueDescriptor( signal_id=signal_id_1, - value=2.0, - timestamp=monotonic_ns(), + value=2.0 * sine_value, + timestamp=timestamp, ) value2 = ValueDescriptor( signal_id=signal_id_2, - value=1.0, - timestamp=monotonic_ns(), + value=sine_value, + timestamp=timestamp, ) message.values.append(value1) message.values.append(value2) diff --git a/test/test_core.py b/test/test_core.py index 7a67a73..7589238 100644 --- a/test/test_core.py +++ b/test/test_core.py @@ -5,7 +5,7 @@ from uuid import uuid4 import pytest from dynalab_core import Core from dynalab_core.config import CoreConfig -from dynalab_core.errors import CoreStateMismatchError +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 @@ -20,7 +20,7 @@ def test_core_cannot_start_twice(caplog: pytest.LogCaptureFixture) -> None: core.start() try: - with pytest.raises(CoreStateMismatchError): + with pytest.raises(CoreError, match="can only be started once") as raised: core.start() finally: core.stop() @@ -37,6 +37,7 @@ def test_core_cannot_start_twice(caplog: pytest.LogCaptureFixture) -> None: ) assert rejection.levelno == logging.WARNING assert rejection.core_state == "started" + assert raised.value.code == "invalid_state" def test_core_input_worker_logs_failure( @@ -87,9 +88,14 @@ def test_core_mode_is_shared_with_endpoints() -> None: try: assert endpoint._core_mode is core._mode + assert core.get_mode() == "realtime" core.set_mode("playback") assert endpoint._core_mode.value == "playback" + assert core.get_mode() == "playback" + with pytest.raises(CoreError) as raised: + core.load_playback_engine() + assert raised.value.code == "no_playback_data" finally: core.stop() diff --git a/test/test_dlpak.py b/test/test_dlpak.py index 7577acc..6a1b678 100644 --- a/test/test_dlpak.py +++ b/test/test_dlpak.py @@ -27,3 +27,21 @@ def test_dlpak_logs_written_archive(caplog: pytest.LogCaptureFixture, tmp_path) assert record.output_filename == "recording.dlpak" assert record.sample_count == 1 assert record.signal_count == 1 + + +def test_dlpak_read_restores_manifest_and_data(tmp_path) -> None: + timestamp = datetime.now(timezone.utc) + signal = SignalDescriptor(id=uuid4(), name="Signal", type="number") + buffer = ValueBuffer() + buffer.append(1, signal.id, 2.0) + package = DLPak() + package.set_data(buffer) + package.set_manifest(timestamp, [signal]) + package.write(tmp_path, "recording") + + loaded_package = DLPak() + loaded_package.read(tmp_path / "recording.dlpak") + + assert loaded_package._manifest == package._manifest + assert loaded_package._data is not None + assert loaded_package._data.export_csv() == buffer.export_csv() diff --git a/test/test_errors.py b/test/test_errors.py new file mode 100644 index 0000000..d72b463 --- /dev/null +++ b/test/test_errors.py @@ -0,0 +1,146 @@ +from datetime import datetime, timezone +from queue import Queue +from uuid import uuid4 +from zipfile import BadZipFile + +import pytest + +from dynalab_core.buffer import ValueBuffer +from dynalab_core.derive import DeriveRegistry, DeriveUnit +from dynalab_core.dlpak import DLPak +from dynalab_core.errors import ( + CoreError, + DLPakError, + DeriveError, + DynaLabError, + PlaybackError, +) +from dynalab_core.playback import PlaybackEngine +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, + ) + + +def test_subsystem_errors_share_public_base_and_code() -> None: + error = PlaybackError("Playback data is unavailable", code="no_data") + + assert isinstance(error, DynaLabError) + assert error.code == "no_data" + assert str(error) == "Playback data is unavailable" + + +def test_core_rejects_invalid_mode_and_missing_recording() -> None: + from dynalab_core import Core + + core = Core.__new__(Core) + core._processing_buffer = None + + with pytest.raises(CoreError, match="Unknown core mode") as mode: + core.set_mode("invalid") # type: ignore[arg-type] + with pytest.raises(CoreError, match="no recording") as recording: + core.write(".", "unused") + + assert mode.value.code == "invalid_mode" + assert recording.value.code == "no_recording" + + +def test_dlpak_requires_data_before_manifest() -> None: + package = DLPak() + + with pytest.raises(DLPakError, match="before setting its data") as raised: + package.set_manifest(datetime.now(timezone.utc), []) + + assert raised.value.code == "no_data" + + +def test_dlpak_wraps_archive_read_failures(tmp_path) -> None: + invalid_archive = tmp_path / "invalid.dlpak" + invalid_archive.write_text("not a zip archive") + + with pytest.raises(DLPakError, match="Could not read") as raised: + DLPak().read(invalid_archive) + + assert raised.value.code == "read_failed" + assert isinstance(raised.value.__cause__, BadZipFile) + + +def test_derive_unit_reports_argument_mismatch() -> None: + input_signal = SignalDescriptor(id=uuid4(), name="Input", type="number") + output_signal = SignalDescriptor( + id=uuid4(), name="Output", type="number", origin="derived" + ) + unit = DeriveUnit(uuid4(), _identity, [input_signal], output_signal, Queue()) + + try: + with pytest.raises(DeriveError, match="expected 1 input") as raised: + unit.process_offline([]) + finally: + unit.stop() + + assert raised.value.code == "argument_mismatch" + + +def test_derive_registry_wraps_invalid_source() -> None: + registry = DeriveRegistry(Queue(), lambda signal_id: None) + output_signal = SignalDescriptor( + id=uuid4(), name="Output", type="number", origin="derived" + ) + + try: + with pytest.raises(DeriveError, match="Could not load") as raised: + registry.register("def broken(", [], output_signal) + finally: + registry.stop() + + assert raised.value.code == "invalid_source" + assert isinstance(raised.value.__cause__, SyntaxError) + + +def test_derive_unit_rejects_configured_input_mismatch() -> None: + output_signal = SignalDescriptor( + id=uuid4(), name="Output", type="number", origin="derived" + ) + + with pytest.raises(DeriveError, match="input signal") as raised: + DeriveUnit(uuid4(), _identity, [], output_signal, Queue()) + + assert raised.value.code == "input_signal_mismatch" + + +def test_playback_rejects_incomplete_data_and_negative_seek() -> None: + engine = PlaybackEngine(Queue()) + + try: + with pytest.raises(PlaybackError, match="incomplete") as incomplete: + engine.set_data(DLPak()) + with pytest.raises(PlaybackError, match="cannot be negative") as timestamp: + engine.seek(-1) + finally: + engine.stop() + + assert incomplete.value.code == "incomplete_data" + assert timestamp.value.code == "invalid_timestamp" + + +def test_playback_accepts_complete_dlpak() -> None: + signal = SignalDescriptor(id=uuid4(), name="Input", type="number") + data = ValueBuffer() + data.append(1, signal.id, 2.0) + package = DLPak() + package.set_data(data) + package.set_manifest(datetime.now(timezone.utc), [signal]) + engine = PlaybackEngine(Queue()) + + try: + engine.set_data(package) + finally: + engine.stop() diff --git a/test/test_json_server.py b/test/test_json_server.py index b020f89..7aff5c3 100644 --- a/test/test_json_server.py +++ b/test/test_json_server.py @@ -7,12 +7,9 @@ from uuid import uuid4 import pytest from dynalab_core import Core 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.json.errors import ( - JsonServerStartupError, - JsonServerTimeoutError, -) from dynalab_core.protocols.packets.handshake import ConnectorHello, SignalDescriptor from test.common import find_available_port @@ -39,7 +36,7 @@ def test_json_server_raises_timeout_error( with caplog.at_level(logging.DEBUG, logger="dynalab_core"): try: - with pytest.raises(JsonServerTimeoutError): + with pytest.raises(JsonServerError, match="did not start") as raised: core.start() finally: core.stop() @@ -51,6 +48,7 @@ def test_json_server_raises_timeout_error( ) assert timeout_record.levelno == logging.ERROR assert timeout_record.port == port + assert raised.value.code == "start_timeout" def test_json_server_raises_startup_error( @@ -65,7 +63,7 @@ def test_json_server_raises_startup_error( core1.start() try: - with pytest.raises(JsonServerStartupError): + with pytest.raises(JsonServerError, match="failed to start") as raised: core2.start() finally: core1.stop() @@ -79,6 +77,8 @@ def test_json_server_raises_startup_error( assert failure_record.levelno == logging.ERROR assert failure_record.port == port assert failure_record.exc_info is not None + assert raised.value.code == "startup_failed" + assert isinstance(raised.value.__cause__, OSError) def test_json_server_logs_invalid_handshake( @@ -117,7 +117,7 @@ def test_json_server_reports_unexpected_thread_startup_failure( with caplog.at_level(logging.DEBUG, logger="dynalab_core"): try: - with pytest.raises(JsonServerStartupError): + with pytest.raises(JsonServerError) as raised: core.start() finally: core.stop() @@ -131,6 +131,8 @@ def test_json_server_reports_unexpected_thread_startup_failure( assert failures[0].startup_complete is False assert failures[0].exception_type == "RuntimeError" assert failures[0].exc_info is not None + assert raised.value.code == "startup_failed" + assert isinstance(raised.value.__cause__, RuntimeError) def test_rejected_handshake_is_not_logged_as_connected(