diff --git a/src/dynalab_core/__init__.py b/src/dynalab_core/__init__.py index 92c2eda..23a9259 100644 --- a/src/dynalab_core/__init__.py +++ b/src/dynalab_core/__init__.py @@ -3,13 +3,15 @@ # SPDX-License-Identifier: GPL-3.0-or-later -from copy import copy +from collections.abc import Callable +from copy import copy, deepcopy from datetime import datetime, timezone +from dataclasses import dataclass, field import logging from pathlib import Path -from queue import Empty, Queue +from queue import Empty, Full, Queue import threading -from threading import Lock, Thread +from threading import Lock, RLock, Thread import time from typing import Literal from uuid import UUID @@ -24,26 +26,46 @@ 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.playback import PlaybackEngine, PlaybackValueDescriptor from dynalab_core.protocols.endpoint import ConnectorRegistry from dynalab_core.protocols.common import VersionDescriptor from dynalab_core.protocols.json.server import JsonServer -from dynalab_core.protocols.packets import ProtocolMessage from dynalab_core.protocols.packets.data import ValueDescriptor from dynalab_core.protocols.packets.handshake import SignalDescriptor -from dynalab_core.types import BufferSource, CoreMode, CoreModeState -from dynalab_core.values import Value +from dynalab_core.types import ( + BufferSource, + CoreMode, + CoreModeState, + WorkerStatus, + ConnectorStatus, + DeriveUnitStatus, +) +from dynalab_core.values import Value, SignalValueDescriptor +from dynalab_core.operations import OperationManager, OperationStatus log = logging.getLogger(__name__) log.addHandler(logging.NullHandler()) +@dataclass +class _RecordingBoundary: + active: bool + finished: threading.Event = field(default_factory=threading.Event) + data: ValueBuffer | None = None + signals: list[SignalDescriptor] = field(default_factory=list) + error: Exception | None = None + + class Core: def __init__(self, config: CoreConfig) -> None: self._state: Literal[ "uninitd", "initd", "started", "stopping", "stopped", "stop_incomplete" - ] = "unintid" + ] = "uninitd" + self._control_lock = RLock() + self._operations = OperationManager(self._control_lock) + self._input_error: str | None = None + self._recording_signals: dict[UUID, SignalDescriptor] = {} self._mode: CoreModeState = CoreModeState() self._recording = threading.Event() self._recording_buffer = ValueBuffer() @@ -57,26 +79,32 @@ class Core: self._live_values: dict[UUID, Value] = {} self._live_values_lock = Lock() - self._data_input_queue: Queue[ProtocolMessage] = Queue(524288) + self._data_input_queue: Queue = Queue(524288) self._input_worker_thread = Thread( target=self._input_worker, name="input_worker_thread", daemon=True ) - self._input_worker_thread.start() - self._playback_engine = PlaybackEngine( self._core_config, self._data_input_queue ) + self._connector_registry = ConnectorRegistry( + self._data_input_queue, + self._mode, + lambda: self._derive_registry.get_output_ids(), + ) self._derive_registry = DeriveRegistry( self._data_input_queue, self._mode, self._get_live_value_descriptor, self.get_all_signal_descriptors, + self._playback_engine.is_current, + self._core_config.derive_queue_capacity, + self._get_source_revision, ) - self._connector_registry = ConnectorRegistry(self._data_input_queue, self._mode) self._json_server = JsonServer(self._core_config, self._connector_registry) self._state = "initd" + self._input_worker_thread.start() self._debug_thread: Thread | None = None @@ -185,8 +213,11 @@ class Core: "Core stopping", extra={"event": "core.stopping", "core_state": self._state}, ) + with self._control_lock: + self._state = "stopping" + operations_stopped = self._operations.wait_active(5.0) json_server_stopped = self._json_server.stop() - self._playback_engine.stop() + playback_stopped = self._playback_engine.stop() connector_registry_stopped = self._connector_registry.stop() derive_registry_stopped = self._derive_registry.stop() self._stop_event.set() @@ -202,7 +233,9 @@ class Core: ) stopped = all( ( + operations_stopped, json_server_stopped, + playback_stopped, connector_registry_stopped, derive_registry_stopped, input_worker_stopped, @@ -221,7 +254,9 @@ class Core: extra={ "event": "core.stop_incomplete", "core_state": self._state, + "operations_stopped": operations_stopped, "json_server_stopped": json_server_stopped, + "playback_stopped": playback_stopped, "connector_registry_stopped": connector_registry_stopped, "derive_registry_stopped": derive_registry_stopped, "input_worker_stopped": input_worker_stopped, @@ -229,21 +264,100 @@ class Core: ) return stopped - def set_mode(self, mode: CoreMode) -> None: - if self._recording.is_set(): - raise CoreError( - "Cannot change core mode while a recording is active", - code="recording_active", - ) + def _require_mutable(self) -> None: + if self._state not in ("initd", "started"): + raise CoreError("Core is stopping or stopped", code="invalid_state") + self._operations.require_idle() - self._mode.value = mode - with self._live_values_lock: - self._live_values = {} + def set_mode(self, mode: CoreMode) -> None: + with self._control_lock: + self._require_mutable() + if mode not in ("realtime", "offline", "playback"): + raise CoreError("Unknown core mode", code="invalid_mode") + if self._recording.is_set(): + raise CoreError( + "Cannot change mode while recording", code="recording_active" + ) + with self._live_values_lock: + if self._mode.value == "playback" and mode != "playback": + self._playback_engine.pause() + self._mode.value = mode + self._live_values.clear() + + def _get_source_revision(self) -> object: + return ( + self._mode.value, + id(self._processing_buffer), + id(self._playback_engine._data), + self._connector_registry.get_revision(), + ) + + def get_connectors(self) -> list[ConnectorStatus]: + return self._connector_registry.get_statuses() + + def get_derive_units(self) -> list[DeriveUnitStatus]: + return self._derive_registry.get_statuses() + + def validate_derive_units(self) -> None: + self._derive_registry.validate() + + def get_worker_statuses(self) -> list[WorkerStatus]: + return [ + WorkerStatus( + self._input_worker_thread.name, + "failed" + if self._input_error + else "stopped" + if self._input_worker_stopped_event.is_set() + else "running", + self._input_error, + self._data_input_queue.qsize(), + self._data_input_queue.maxsize, + ), + self._json_server.get_status(), + self._playback_engine.get_status(), + self._derive_registry.get_worker_status(), + *(unit.worker for unit in self.get_derive_units()), + ] + + def get_operation(self, operation_id: UUID | None = None) -> OperationStatus | None: + return self._operations.get(operation_id) + + def wait_operation( + self, operation_id: UUID, timeout: float | None = None + ) -> OperationStatus: + return self._operations.wait(operation_id, timeout) + + def _run_operation( + self, name: str, action: Callable[[], None], background: bool + ) -> UUID | None: + with self._control_lock: + self._require_mutable() + if self._recording.is_set(): + raise CoreError( + "Stop recording before starting a data operation", + code="recording_active", + ) + operation_id = self._operations.start(name, action) + if background: + return operation_id + self._operations.wait(operation_id) + return None + + def get_state(self) -> str: + return self._state def get_mode(self) -> CoreMode: return copy(self._mode.value) - def load_playback_engine(self, input: Path | None = None) -> None: + def load_playback_engine( + self, input: Path | None = None, *, background: bool = False + ) -> UUID | None: + return self._run_operation( + "load_playback", lambda: self._load_playback_engine(input), background + ) + + 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", @@ -255,12 +369,23 @@ class Core: "Cannot load playback because no processing buffer is available", code="no_playback_data", ) - loaded_data = copy(self._processing_buffer) + loaded_data = self._processing_buffer else: - loaded_data = DLPak().read(input) - self._playback_engine.set_data(loaded_data) + loaded_data = DLPak().read(input, progress=self._operations.update) + with self._live_values_lock: + if self._state not in ("initd", "started"): + raise CoreError("Core stopped during loading", code="invalid_state") + self._playback_engine.set_data(loaded_data) + self._live_values.clear() - def load_processing_buffer(self, input: Path | None = None) -> None: + def load_processing_buffer( + self, input: Path | None = None, *, background: bool = False + ) -> UUID | None: + return self._run_operation( + "load_processing", lambda: self._load_processing_buffer(input), background + ) + + def _load_processing_buffer(self, input: Path | None = None) -> None: if self._mode.value != "offline": raise CoreError( "Cannot load processing unless the core is in 'offline' mode", @@ -273,28 +398,17 @@ class Core: "Cannot load processing because no recording buffer is available", code="no_playback_data", ) - loaded_data = copy(self._offload_buffer) + loaded_data = deepcopy(self._offload_buffer) else: - loaded_data = DLPak().read(input) + loaded_data = DLPak().read(input, progress=self._operations.update) + if self._state not in ("initd", "started"): + raise CoreError("Core stopped during loading", code="invalid_state") self._processing_buffer = loaded_data - def _reprocess_worker( - self, - result: dict, - ) -> None: - try: - self._processing_buffer.remove_derived() + def reprocess(self, *, background: bool = False) -> UUID | None: + return self._run_operation("reprocess", self._reprocess, background) - processed = self._derive_registry.process_offline(self._processing_buffer) - - self._processing_buffer.merge(processed) - - result["processed"] = processed - - except Exception as err: - result["error"] = err - - def reprocess(self) -> None: + def _reprocess(self) -> None: if self._mode.value != "offline": raise CoreError( "Cannot reprocess unless the core is in 'offline' mode", @@ -320,27 +434,24 @@ class Core: "sample_count": len(self._processing_buffer._data), }, ) - result: dict[str, DLPak | Exception] = {} - - reprocess_thread = Thread( - target=self._reprocess_worker, - name="core_reprocess_thread", - args=(result,), - ) - reprocess_thread.start() - reprocess_thread.join() - - if error := result.get("error"): - log.exception( - "Offline reprocessing failed", - extra={ - "event": "core.reprocess_failed", - "exception_type": type(error).__name__, - }, + snapshot = deepcopy(self._processing_buffer) + snapshot.remove_derived() + try: + processed = self._derive_registry.process_offline( + snapshot, self._operations.update ) - raise error - - processed = result["processed"] + snapshot.merge(processed) + except Exception: + log.exception( + "Offline reprocessing failed", extra={"event": "core.reprocess_failed"} + ) + raise + with self._control_lock: + if self._state not in ("initd", "started"): + raise CoreError( + "Core stopped during reprocessing", code="invalid_state" + ) + self._processing_buffer = snapshot log.info( "Offline reprocessing completed with %d derived sample(s)", @@ -352,71 +463,75 @@ class Core: }, ) + def _recording_boundary(self, active: bool) -> _RecordingBoundary: + boundary = _RecordingBoundary(active) + try: + self._data_input_queue.put(boundary, timeout=5.0) + except Full as error: + raise CoreError("Input queue is full", code="input_overload") from error + while not boundary.finished.wait(0.1): + if not self._input_worker_thread.is_alive(): + raise CoreError("Input worker has stopped", code="worker_failed") + if boundary.error is not None: + raise boundary.error + return boundary + def start_recording(self) -> None: - if self._mode.value != "realtime": - 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() - log.info( - "Recording started", - extra={ - "event": "recording.started", - "recording_timestamp": self._recording_timestamp.isoformat(), - }, - ) + with self._control_lock: + self._require_mutable() + if self._mode.value != "realtime": + raise CoreError( + "Recording requires realtime mode", code="incorrect_mode" + ) + if self._recording.is_set(): + raise CoreError( + "A recording is already active", code="recording_active" + ) + self._recording_boundary(True) + log.info("Recording started", extra={"event": "recording.started"}) def stop_recording(self) -> None: - if self._mode.value != "realtime": - 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() - self._recording_buffer.sort() - self._offload_buffer = DLPak() - self._offload_buffer.set_data(copy(self._recording_buffer)) - connector_signal_descriptors = ( - self._connector_registry.get_all_signal_descriptors() - ) - derive_signal_descriptors = self._derive_registry.get_all_signal_descriptors() - - self._offload_buffer.set_manifest( - self._recording_timestamp, - connector_signal_descriptors + derive_signal_descriptors, - ) + with self._control_lock: + self._require_mutable() + if self._mode.value != "realtime": + raise CoreError( + "Recording requires realtime mode", code="incorrect_mode" + ) + if not self._recording.is_set(): + raise CoreError("No recording is active", code="not_recording") + boundary = self._recording_boundary(False) + completed = DLPak() + boundary.data.normalize() + boundary.data.sort() + completed.set_data(boundary.data) + completed.set_manifest(self._recording_timestamp, boundary.signals) + self._offload_buffer = completed log.info( - "Recording stopped with %d sample(s) from %d signal(s)", - len(self._recording_buffer), - len(self._recording_buffer.get_signal_ids()), - extra={ - "event": "recording.stopped", - "sample_count": len(self._recording_buffer), - "signal_count": len(self._recording_buffer.get_signal_ids()), - "duration_ms": round( - ( - datetime.now(timezone.utc) - self._recording_timestamp - ).total_seconds() - * 1000 - ), - }, + "Recording stopped with %d samples", + len(boundary.data), + extra={"event": "recording.stopped", "sample_count": len(boundary.data)}, ) + def _apply_recording_boundary(self, boundary: _RecordingBoundary) -> None: + try: + with self._live_values_lock: + if boundary.active: + self._recording_buffer = ValueBuffer() + self._recording_timestamp = datetime.now(timezone.utc) + self._recording_signals = { + signal.id: copy(signal) + for signal in self.get_all_signal_descriptors() + } + self._recording.set() + else: + self._recording.clear() + boundary.data = self._recording_buffer + boundary.signals = list(self._recording_signals.values()) + except Exception as error: + boundary.error = error + finally: + boundary.finished.set() + def get_recording_state(self) -> bool: return self._recording.is_set() @@ -475,10 +590,21 @@ class Core: with self._live_values_lock: live_values = list(self._live_values.items()) - return {signal_id: value.get() for signal_id, value in live_values} + return { + signal_id: current + for signal_id, value in live_values + if (current := value.get()) is not None + } def bind_derive_unit( self, dlunit: DLUnit | None = None, file: Path | None = None + ) -> UUID: + with self._control_lock: + self._require_mutable() + return self._bind_derive_unit(dlunit, file) + + def _bind_derive_unit( + self, dlunit: DLUnit | None = None, file: Path | None = None ) -> UUID: if dlunit is not None and file is not None: raise CoreError( @@ -497,7 +623,10 @@ class Core: return copy(unit.uuid()) def unbind_derive_unit(self, unit_id: UUID) -> None: - self._derive_registry.unregister(unit_id) + with self._control_lock: + self._require_mutable() + if not self._derive_registry.unregister(unit_id): + raise CoreError("Derive worker could not stop", code="stop_incomplete") def _get_archive(self, source: BufferSource) -> DLPak: if source == "recording": @@ -549,42 +678,69 @@ class Core: return self._get_value_buffer(source).get_time_range() def write( - self, dir: str | Path, filename: str, *, source: BufferSource = "processing" - ) -> None: + self, + dir: str | Path, + filename: str, + *, + source: BufferSource = "processing", + background: bool = False, + ) -> UUID | None: + return self._run_operation( + "write", lambda: self._write(dir, filename, source), background + ) + + def _write(self, dir: str | Path, filename: str, source: BufferSource) -> None: """Save processing data by default, or explicitly save another buffer.""" if source == "recording" and self._recording.is_set(): raise CoreError( "Stop recording before saving its archive", code="recording_active" ) - self._get_archive(source).write(dir, filename) + self._get_archive(source).write(dir, filename, progress=self._operations.update) 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() + with self._control_lock: + self._require_mutable() + if self._mode.value != "playback": + raise CoreError( + "Cannot play unless the core is in 'playback' mode", + code="invalid_mode", + ) + with self._live_values_lock: + if ( + not self._playback_engine.is_playing() + and self._playback_engine.get_position() == 0 + ): + self._live_values.clear() + 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() + with self._control_lock: + self._require_mutable() + if self._mode.value != "playback": + raise CoreError( + "Cannot pause unless the core is in 'playback' mode", + code="invalid_mode", + ) + with self._live_values_lock: + 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) + with self._control_lock: + self._require_mutable() + if self._mode.value != "playback": + raise CoreError( + "Cannot seek unless the core is in 'playback' mode", + code="invalid_mode", + ) + with self._live_values_lock: + self._playback_engine.seek(timestamp_ns) + self._live_values.clear() def get_playback_state(self) -> bool: - # TODO: internalize playback state to PlaybackEngine - return self._playback_engine._playing.is_set() + return self._playback_engine.is_playing() + + def get_playback_position(self) -> int: + return self._playback_engine.get_position() def _get_live_value_descriptor(self, signal_id: UUID) -> ValueDescriptor | None: with self._live_values_lock: @@ -610,15 +766,28 @@ class Core: self._stop_event.wait(0.01) continue + if isinstance(message, _RecordingBoundary): + self._apply_recording_boundary(message) + continue if isinstance(message, ValueDescriptor): with self._live_values_lock: + if isinstance(message, PlaybackValueDescriptor): + if ( + self._mode.value != "playback" + or not self._playback_engine.is_current(message) + ): + continue + elif self._mode.value != "realtime": + continue + signal_descriptor = ( + message.signal + if isinstance(message, SignalValueDescriptor) + else self.get_signal_descriptor(message.signal_id) + ) + if signal_descriptor is None: + continue live_value = self._live_values.get(message.signal_id) - if live_value is None: - signal_descriptor = self.get_signal_descriptor( - message.signal_id - ) - live_value = Value( signal_descriptor.timeout_ms if signal_descriptor @@ -626,18 +795,23 @@ class Core: ) self._live_values[message.signal_id] = live_value - live_value.update(message.value, message.timestamp) - if self._recording.is_set(): - self._recording_buffer.append( - message.timestamp, message.signal_id, message.value - ) + live_value.update(message.value, message.timestamp) + if self._recording.is_set(): + if message.signal_id not in self._recording_signals: + self._recording_signals[message.signal_id] = copy( + signal_descriptor + ) + self._recording_buffer.append( + message.timestamp, message.signal_id, message.value + ) - if ( - self._mode.value == "realtime" - or self._core_config.playback_mode == "reprocess" - ): - self._derive_registry.put_data(message) + if ( + self._mode.value == "realtime" + or self._core_config.playback_mode == "reprocess" + ): + self._derive_registry.put_data(message) except Exception as error: + self._input_error = str(error) log.exception( "Core input worker failed", extra={ diff --git a/src/dynalab_core/buffer.py b/src/dynalab_core/buffer.py index dc6e34e..3d4d6d1 100644 --- a/src/dynalab_core/buffer.py +++ b/src/dynalab_core/buffer.py @@ -25,6 +25,13 @@ class ValueBuffer: buffer.merge(self) return buffer + def __deepcopy__(self, memo: dict) -> ValueBuffer: + # Samples contain only immutable scalars and UUIDs. Rebuild every mutable + # container and create a new lock rather than attempting to copy a lock. + buffer = self.__copy__() + memo[id(self)] = buffer + return buffer + def _append_no_lock(self, timestamp: int, uuid: UUID, value: float) -> None: if self._samples and timestamp < self._samples[-1][0]: self._sorted = False diff --git a/src/dynalab_core/config.py b/src/dynalab_core/config.py index 9fb4ed0..cfe5aac 100644 --- a/src/dynalab_core/config.py +++ b/src/dynalab_core/config.py @@ -14,5 +14,7 @@ class CoreConfig(BaseModel): thread_debug: bool = False playback_mode: PlaybackMode = "reprocess" + derive_queue_capacity: int = Field(default=16384, ge=1) + def bind_str(self) -> str: return f"{self.host}:{self.port}" diff --git a/src/dynalab_core/derive.py b/src/dynalab_core/derive.py index 2326df5..648da30 100644 --- a/src/dynalab_core/derive.py +++ b/src/dynalab_core/derive.py @@ -7,7 +7,7 @@ from datetime import datetime, timezone from enum import IntEnum import inspect import logging -from queue import Empty, Queue +from queue import Empty, Full, Queue from threading import RLock, Thread import threading import types @@ -17,9 +17,12 @@ from dynalab_core.buffer import ValueBuffer from dynalab_core.dlpak import DLPak from dynalab_core.dlunit import DLUnit, DeriveManifest from dynalab_core.errors import DeriveError +from dynalab_core.playback import PlaybackValueDescriptor from dynalab_core.protocols.packets.data import ValueDescriptor from dynalab_core.protocols.packets.handshake import SignalDescriptor -from dynalab_core.types import CoreModeState +from dynalab_core.types import CoreModeState, DeriveUnitStatus, WorkerStatus +from dynalab_core.values import SignalValueDescriptor +from dynalab_core.operations import ProgressCallback log = logging.getLogger(__name__) @@ -38,6 +41,9 @@ class DeriveUnit: input_signals: list[SignalDescriptor], return_signal: SignalDescriptor, return_queue: Queue[ValueDescriptor], + is_current_playback_value: Callable[[PlaybackValueDescriptor], bool] + | None = None, + queue_capacity: int = 16384, ) -> None: self._id: UUID = unit_uuid self._process_function: Callable = processing_function @@ -45,8 +51,12 @@ class DeriveUnit: 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() + self._input_queue: Queue[list[ValueDescriptor]] = Queue(queue_capacity) + self._error: str | None = None + self._dropped_samples = 0 + self._status_lock = threading.Lock() self._return_queue: Queue[ValueDescriptor] = return_queue + self._is_current_playback_value = is_current_playback_value self._num_input_args: int = 0 self._parser_state: int = ParserState.INIT @@ -143,9 +153,38 @@ class DeriveUnit: ) continue - value = self._process_function(*input_args, self._return_signal) - self._return_queue.put(value) + if self._has_stale_playback_input(input_args): + continue + try: + value = self._process_function(*input_args, self._return_signal) + self._validate_result(value) + except Exception: + if self._has_stale_playback_input(input_args): + continue + raise + if input_args and isinstance(input_args[0], PlaybackValueDescriptor): + value = PlaybackValueDescriptor( + signal_id=value.signal_id, + value=value.value, + timestamp=value.timestamp, + generation=input_args[0].generation, + ) + else: + value = SignalValueDescriptor( + value.signal_id, + value.value, + value.timestamp, + self._return_signal, + ) + while not self._stop_event.is_set(): + try: + self._return_queue.put(value, timeout=0.1) + break + except Full: + continue except Exception as error: + with self._status_lock: + self._error = str(error) log.exception( "Derive worker %s failed", derive_uuid, @@ -168,9 +207,59 @@ class DeriveUnit: }, ) - def put_data(self, input: list[ValueDescriptor]) -> None: - if not self._stopped_event.is_set(): - self._input_queue.put(input) + def put_data(self, input: list[ValueDescriptor]) -> bool: + if self._stop_event.is_set() or self._stopped_event.is_set(): + return False + try: + self._input_queue.put_nowait(input) + return True + except Full: + with self._status_lock: + self._dropped_samples += 1 + if self._dropped_samples == 1: + log.warning( + "Derive %s queue is full; dropping new requests", self._id + ) + return False + + def get_status(self) -> DeriveUnitStatus: + with self._status_lock: + worker = WorkerStatus( + name=self._worker_thread.name, + state="failed" + if self._error is not None + else "running" + if self.is_running() + else "stopped", + error=self._error, + queue_size=self._input_queue.qsize(), + queue_capacity=self._input_queue.maxsize, + dropped_samples=self._dropped_samples, + ) + return DeriveUnitStatus( + self._id, + tuple(s.id for s in self._input_signals), + self._return_signal.id, + worker, + ) + + def _validate_result(self, value: ValueDescriptor) -> None: + if ( + not isinstance(value, ValueDescriptor) + or value.signal_id != self._return_signal.id + ): + raise DeriveError( + "Derive function must return a value for its configured output signal", + code="invalid_result", + ) + + def _has_stale_playback_input(self, args: list[ValueDescriptor]) -> bool: + return bool( + args + and isinstance(args[0], PlaybackValueDescriptor) + and self._is_current_playback_value is not None + and not self._is_current_playback_value(args[0]) + ) def get_input_signals(self) -> list[SignalDescriptor]: return self._input_signals @@ -220,6 +309,7 @@ class DeriveUnit: input_args.append(ValueDescriptor(signal.id, input_value, timestamp)) result = self._process_function(*input_args, self._return_signal) + self._validate_result(result) output_buffer.append(result.timestamp, self._return_signal.id, result.value) return output_buffer @@ -231,13 +321,25 @@ class DeriveRegistry: core_mode: CoreModeState, get_live_value: Callable[[UUID], ValueDescriptor | None], get_all_core_signal_descriptors: Callable[[], list[SignalDescriptor]], + is_current_playback_value: Callable[[PlaybackValueDescriptor], bool] + | None = None, + queue_capacity: int = 16384, + get_source_revision: Callable[[], object] | None = None, ) -> None: self._lock = RLock() self._units: dict[UUID, DeriveUnit] = {} self._core_input_queue = core_input_queue self._get_live_value = get_live_value self._get_all_core_signal_descriptors = get_all_core_signal_descriptors - self._routing_queue: Queue[ValueDescriptor] = Queue() + self._is_current_playback_value = is_current_playback_value + self._routing_queue: Queue[ValueDescriptor] = Queue(queue_capacity) + self._queue_capacity = queue_capacity + self._get_source_revision = get_source_revision + self._units_revision = 0 + self._validated_revision: object = None + self._validation_error: str | None = None + self._error: str | None = None + self._dropped_samples = 0 self._routing_stop_event = threading.Event() self._routing_stopped_event = threading.Event() self._routing_worker_thread = Thread( @@ -246,10 +348,13 @@ class DeriveRegistry: daemon=True, ) self._routing_worker_thread.start() - self._dependency_layers: list[list[UUID]] = [] self._core_mode = core_mode def register(self, dlunit: DLUnit) -> DeriveUnit: + if dlunit._manifest is None or dlunit._data is None: + raise DeriveError( + "Derive unit needs a manifest and source", code="incomplete_unit" + ) if dlunit._manifest.unit_uuid is not None: with self._lock: @@ -294,16 +399,31 @@ class DeriveRegistry: ] pure_return_signal = dlunit._manifest.return_signal.to_pure() - unit = DeriveUnit( - unit_id, - functions[0], - pure_input_signals, - pure_return_signal, - self._core_input_queue, - ) - + sources = { + signal.id + for signal in self._get_all_core_signal_descriptors() + if signal.origin == "source" + } with self._lock: + if unit_id in self._units: + raise DeriveError( + "Derive unit is already registered", code="already_registered" + ) + manifests = [unit.get_manifest() for unit in self._units.values()] + manifests.append(dlunit._manifest) + self._build_layers(manifests, sources, allow_missing=True) + unit = DeriveUnit( + unit_id, + functions[0], + pure_input_signals, + pure_return_signal, + self._core_input_queue, + self._is_current_playback_value, + self._queue_capacity, + ) self._units[unit_id] = unit + self._units_revision += 1 + self._validated_revision = None log.info( "Registered derive unit %s", @@ -323,30 +443,15 @@ class DeriveRegistry: def unregister(self, unit_uuid: UUID) -> bool: with self._lock: current = self._units.get(unit_uuid) - - if current is not None: - if not current.stop(): - log.error( - "Unable to unregister derive unit %s because it did not stop", - unit_uuid, - extra={ - "event": "derive.unregister_incomplete", - "derive_uuid": str(unit_uuid), - }, - ) - return False + if current is None: + return True + if not current.stop(): + return False + with self._lock: + if self._units.get(unit_uuid) is current: del self._units[unit_uuid] - else: - log.debug( - "Unit %s was not registered", - unit_uuid, - extra={ - "event": "derive.unregister_noop", - "derive_uuid": str(unit_uuid), - }, - ) - return True - + self._units_revision += 1 + self._validated_revision = None log.info( "Unregistered derive unit %s", unit_uuid, @@ -385,9 +490,44 @@ class DeriveRegistry: return signals - def put_data(self, message: ValueDescriptor) -> None: - if not self._routing_stop_event.is_set(): + def get_output_ids(self) -> set[UUID]: + with self._lock: + return {unit.return_signal().id for unit in self._units.values()} + + def get_statuses(self) -> list[DeriveUnitStatus]: + with self._lock: + return [unit.get_status() for unit in self._units.values()] + + def get_worker_status(self) -> WorkerStatus: + self._refresh_validation() + with self._lock: + return WorkerStatus( + self._routing_worker_thread.name, + "failed" + if self._error + else "stopped" + if self._routing_stopped_event.is_set() + else "blocked" + if self._validation_error + else "running", + self._error or self._validation_error, + self._routing_queue.qsize(), + self._routing_queue.maxsize, + self._dropped_samples, + ) + + def put_data(self, message: ValueDescriptor) -> bool: + if self._routing_stop_event.is_set() or self._routing_stopped_event.is_set(): + return False + try: self._routing_queue.put_nowait(message) + return True + except Full: + with self._lock: + self._dropped_samples += 1 + if self._dropped_samples == 1: + log.warning("Derive routing queue is full; dropping new requests") + return False def stop(self) -> bool: log.debug( @@ -410,9 +550,11 @@ class DeriveRegistry: with self._lock: units = list(self._units.values()) - unit_results = [unit.stop() for unit in units] - units_stopped = all(unit_results) - self._units.clear() + unit_results = [unit.stop() for unit in units] + units_stopped = all(unit_results) + with self._lock: + if units_stopped: + self._units.clear() stopped = routing_stopped and units_stopped if stopped: log.debug( @@ -430,67 +572,83 @@ class DeriveRegistry: ) return stopped - def _build_dependency_layers(self) -> None: - all_signals = self._get_all_core_signal_descriptors() - - source_signals: list[SignalDescriptor] = [ - signal for signal in all_signals if signal.origin == "source" - ] - - units = [unit.get_manifest() for unit in self._units.values()] - - dependent_units = [ - unit - for unit in units - if any(signal.origin == "derived" for signal in unit.input_signals) - ] - - L0_units = [ - unit - for unit in units - if not any(signal.origin == "derived" for signal in unit.input_signals) - ] - - unit_layers: list[list[DeriveManifest]] = [] - - unit_layers.append(L0_units) - - while len(dependent_units) > 0: - available_signal_ids = {signal.id for signal in source_signals} - available_signal_ids.update( - unit.return_signal.id for layer in unit_layers for unit in layer + @staticmethod + def _build_layers( + units: list[DeriveManifest], + source_ids: set[UUID], + *, + allow_missing: bool = False, + ) -> list[list[UUID]]: + if any(not unit.input_signals for unit in units): + raise DeriveError( + "Derive units need at least one input signal", code="missing_inputs" ) - Lx_units = [ + output_ids = [unit.return_signal.id for unit in units] + if len(output_ids) != len(set(output_ids)) or source_ids.intersection( + output_ids + ): + raise DeriveError( + "Derived output IDs must be unique and distinct from sources", + code="duplicate_output", + ) + if any(unit.return_signal.origin != "derived" for unit in units): + raise DeriveError( + "Derived outputs must have derived origin", code="invalid_origin" + ) + inputs = {signal.id for unit in units for signal in unit.input_signals} + missing = inputs - source_ids - set(output_ids) + if missing and not allow_missing: + raise DeriveError( + "Missing input signals: " + ", ".join(sorted(map(str, missing))), + code="missing_inputs", + ) + available = source_ids | (missing if allow_missing else set()) + pending = list(units) + layers: list[list[UUID]] = [] + while pending: + ready = [ unit - for unit in dependent_units - if all(sig.id in available_signal_ids for sig in unit.input_signals) + for unit in pending + if all(signal.id in available for signal in unit.input_signals) ] - - if not Lx_units: - unresolved = ", ".join(str(unit.unit_uuid) for unit in dependent_units) + if not ready: raise DeriveError( - f"Unresolved derive dependencies for units: {unresolved}", - code="unresolved_dependencies", + "Derive dependencies contain a cycle", code="dependency_cycle" ) + layers.append([unit.unit_uuid for unit in ready]) + available.update(unit.return_signal.id for unit in ready) + pending = [unit for unit in pending if unit not in ready] + return layers - ready_ids = {unit.unit_uuid for unit in Lx_units} - dependent_units = [ - unit for unit in dependent_units if unit.unit_uuid not in ready_ids - ] - unit_layers.append(Lx_units) - - self._dependency_layers = [ - [manifest.unit_uuid for manifest in layer] for layer in unit_layers - ] - log.debug( - "Built %d offline derive dependency layer(s)", - len(self._dependency_layers), - extra={ - "event": "derive.offline_layers_built", - "layer_count": len(self._dependency_layers), - "unit_count": len(units), - }, + def _build_dependency_layers(self, buffer: DLPak | None = None) -> list[list[UUID]]: + signals = ( + buffer.get_all_signal_descriptors() + if buffer is not None + else self._get_all_core_signal_descriptors() ) + sources = {signal.id for signal in signals if signal.origin == "source"} + if buffer is not None: + sources.intersection_update(buffer._data.get_signal_ids()) + with self._lock: + units = [unit.get_manifest() for unit in self._units.values()] + return self._build_layers(units, sources) + + def validate(self) -> None: + self._build_dependency_layers() + + def _refresh_validation(self) -> None: + revision = ( + self._get_source_revision() if self._get_source_revision else object(), + self._units_revision, + ) + if self._validated_revision == revision: + return + try: + self.validate() + self._validation_error = None + except DeriveError as error: + self._validation_error = str(error) + self._validated_revision = revision def _routing_worker(self) -> None: log.debug( @@ -509,6 +667,7 @@ class DeriveRegistry: self._route_data(message) except Exception as error: + self._error = str(error) log.exception( "Derive routing worker failed", extra={ @@ -528,6 +687,15 @@ class DeriveRegistry: ) def _route_data(self, message: ValueDescriptor) -> None: + if ( + isinstance(message, PlaybackValueDescriptor) + and self._is_current_playback_value is not None + and not self._is_current_playback_value(message) + ): + return + self._refresh_validation() + if self._validation_error is not None: + return with self._lock: units = list(self._units.values()) @@ -548,19 +716,44 @@ class DeriveRegistry: args.append(value_descriptor) else: + if isinstance(message, PlaybackValueDescriptor): + args = [ + PlaybackValueDescriptor( + signal_id=value.signal_id, + value=value.value, + timestamp=value.timestamp, + generation=message.generation, + ) + for value in args + ] unit.put_data(args) - def process_offline(self, buffer: DLPak) -> DLPak: - self._build_dependency_layers() + def process_offline( + self, buffer: DLPak, progress: ProgressCallback | None = None + ) -> DLPak: + layers = self._build_dependency_layers(buffer) + total_units = sum(len(layer) for layer in layers) + completed_units = 0 + if progress: + progress(0, max(total_units, 1), "Preparing derived signals") working_buffer = ValueBuffer() for timestamp, signal_id, value in buffer._data._samples: working_buffer.append(timestamp, signal_id, value) processed_buffer = ValueBuffer() - for layer_index, layer in enumerate(self._dependency_layers): + for layer_index, layer in enumerate(layers): layer_units = [self._units[uuid] for uuid in layer] for unit in layer_units: + if progress: + progress( + completed_units, + total_units, + f"Processing derive unit {unit.uuid()}", + ) returned_buffer = unit.process_offline(working_buffer) + completed_units += 1 + if progress: + progress(completed_units, total_units, "Derived unit completed") if not returned_buffer._samples: log.warning( "Offline derive unit %s returned no values", diff --git a/src/dynalab_core/dlpak.py b/src/dynalab_core/dlpak.py index 15ac4be..a080af4 100644 --- a/src/dynalab_core/dlpak.py +++ b/src/dynalab_core/dlpak.py @@ -3,7 +3,7 @@ # SPDX-License-Identifier: GPL-3.0-or-later from __future__ import annotations -from copy import copy +from copy import copy, deepcopy from datetime import datetime, timezone import logging import os @@ -16,6 +16,7 @@ from zipfile import ZIP_DEFLATED, ZipFile from pydantic import BaseModel, Field from dynalab_core.buffer import ValueBuffer +from dynalab_core.operations import ProgressCallback from dynalab_core.errors import DLPakError from dynalab_core.protocols.packets.handshake import ( JsonSignalDescriptor, @@ -40,8 +41,11 @@ class DLPak: self._data: ValueBuffer | None = None self._manifest: RecordManifest | None = None + def __copy__(self) -> DLPak: + return deepcopy(self) + def set_data(self, data: ValueBuffer) -> None: - self._data = data + self._data = deepcopy(data) def set_manifest( self, timestamp: datetime, all_signals: list[SignalDescriptor] @@ -70,7 +74,7 @@ class DLPak: self._data.sort() for signal in dlpak._manifest.signals: - self._manifest.signals.append(signal) + self._manifest.signals.append(signal.model_copy(deep=True)) log.debug( "Merged %d sample(s) and %d signal(s) into DLPak", len(dlpak._data), @@ -142,16 +146,24 @@ class DLPak: code="invalid_manifest", ) - def read(self, file: str | Path) -> Self: + def read( + self, file: str | Path, *, progress: ProgressCallback | None = None + ) -> Self: input_path = Path(file) try: + if progress: + progress(0, 3, "Opening archive") with ZipFile(input_path, mode="r") as archive: manifest = RecordManifest.model_validate_json( archive.read("manifest.json") ) + if progress: + progress(1, 3, "Reading samples") data = ValueBuffer() data.import_csv(archive.read("data.csv").decode("utf-8")) + if progress: + progress(2, 3, "Validating archive") self._validate_contents(data, manifest) except Exception as error: log.exception( @@ -183,9 +195,17 @@ class DLPak: "signal_count": len(manifest.signals), }, ) + if progress: + progress(3, 3, "Archive loaded") return self - def write(self, dir: str | Path, filename: str) -> None: + def write( + self, + dir: str | Path, + filename: str, + *, + progress: ProgressCallback | None = None, + ) -> None: output_dir = Path(dir) if self._data is None: @@ -211,6 +231,8 @@ class DLPak: temporary_path: Path | None = None try: + if progress: + progress(0, 3, "Validating archive") self._validate_contents(self._data, self._manifest) with NamedTemporaryFile( dir=output_path.parent, @@ -225,10 +247,16 @@ class DLPak: archive.writestr( "manifest.json", self._manifest.model_dump_json(indent=2) ) + if progress: + progress(1, 3, "Writing samples") archive.writestr("data.csv", self._data.export_csv()) temporary_file.flush() os.fsync(temporary_file.fileno()) + if progress: + progress(2, 3, "Committing archive") temporary_path.replace(output_path) + if progress: + progress(3, 3, "Archive saved") except Exception as error: log.exception( "Failed to write DLPak archive to %s", diff --git a/src/dynalab_core/dlunit.py b/src/dynalab_core/dlunit.py index 4c5b27a..18194d2 100644 --- a/src/dynalab_core/dlunit.py +++ b/src/dynalab_core/dlunit.py @@ -3,14 +3,16 @@ # SPDX-License-Identifier: GPL-3.0-or-later import logging +import os from pathlib import Path +from tempfile import NamedTemporaryFile 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.errors import DLUnitError from dynalab_core.protocols.packets.handshake import ( JsonSignalDescriptor, SignalDescriptor, @@ -117,12 +119,25 @@ class DLUnit: output_path = output_dir / file_path + temporary_path: Path | None = None 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()) + with NamedTemporaryFile( + dir=output_path.parent, + prefix=f".{output_path.name}.", + suffix=".tmp", + delete=False, + ) as temporary_file: + temporary_path = Path(temporary_file.name) + with ZipFile( + temporary_file, mode="w", compression=ZIP_DEFLATED + ) as archive: + archive.writestr( + "manifest.json", self._manifest.model_dump_json(indent=2) + ) + archive.writestr("unit.py", self._data) + temporary_file.flush() + os.fsync(temporary_file.fileno()) + temporary_path.replace(output_path) except Exception as error: log.exception( "Failed to write DLUnit archive to %s", @@ -137,6 +152,9 @@ class DLUnit: f"Could not write DLUnit archive {output_path}: {error}", code="write_failed", ) from error + finally: + if temporary_path is not None: + temporary_path.unlink(missing_ok=True) log.info( "Wrote DLUnit archive to %s", diff --git a/src/dynalab_core/operations.py b/src/dynalab_core/operations.py new file mode 100644 index 0000000..d7f5404 --- /dev/null +++ b/src/dynalab_core/operations.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 + +from collections.abc import Callable +from dataclasses import dataclass, replace +from threading import Event, RLock, Thread +from typing import Literal +from uuid import UUID, uuid4 + +from dynalab_core.errors import CoreError + +ProgressCallback = Callable[[int, int, str], None] + + +@dataclass(frozen=True) +class OperationStatus: + id: UUID + name: str + state: Literal["running", "completed", "failed"] = "running" + completed: int = 0 + total: int = 1 + message: str = "" + error_code: str | None = None + error_message: str | None = None + + @property + def progress(self) -> float: + return self.completed / self.total if self.total else 0.0 + + +@dataclass +class _Operation: + status: OperationStatus + finished: Event + error: Exception | None = None + + +class OperationManager: + """Serialize file/processing operations and retain the latest 100 results.""" + + def __init__(self, lock: RLock) -> None: + self._lock = lock + self._operations: dict[UUID, _Operation] = {} + self._active: UUID | None = None + + def require_idle(self) -> None: + with self._lock: + if self._active is not None: + raise CoreError( + "A data operation is already running", code="operation_active" + ) + + def start(self, name: str, action: Callable[[], None]) -> UUID: + with self._lock: + self.require_idle() + operation_id = uuid4() + operation = _Operation(OperationStatus(operation_id, name), Event()) + self._operations[operation_id] = operation + self._active = operation_id + while len(self._operations) > 100: + del self._operations[next(iter(self._operations))] + thread = Thread( + target=self._execute, + args=(operation, action), + name=f"core_{name}_thread", + daemon=True, + ) + try: + thread.start() + except Exception as error: + self._finish(operation, error) + raise + return operation_id + + def _execute(self, operation: _Operation, action: Callable[[], None]) -> None: + error = None + try: + action() + except BaseException as caught: + if isinstance(caught, Exception): + error = caught + else: + error = CoreError( + f"Operation aborted by {type(caught).__name__}", + code="operation_aborted", + ) + error.__cause__ = caught + finally: + self._finish(operation, error) + + def _finish(self, operation: _Operation, error: Exception | None) -> None: + with self._lock: + operation.error = error + operation.status = replace( + operation.status, + state="failed" if error is not None else "completed", + completed=operation.status.completed + if error is not None + else operation.status.total, + error_code=getattr(error, "code", "operation_failed") + if error is not None + else None, + error_message=str(error) if error is not None else None, + ) + self._active = None + operation.finished.set() + + def update(self, completed: int, total: int, message: str) -> None: + with self._lock: + if self._active is not None: + operation = self._operations[self._active] + operation.status = replace( + operation.status, completed=completed, total=total, message=message + ) + + def get(self, operation_id: UUID | None = None) -> OperationStatus | None: + with self._lock: + if operation_id is None: + operation_id = next(reversed(self._operations), None) + if operation_id is None: + return None + if operation_id not in self._operations: + raise CoreError("Unknown operation ID", code="unknown_operation") + return self._operations[operation_id].status + + def wait(self, operation_id: UUID, timeout: float | None = None) -> OperationStatus: + with self._lock: + if operation_id not in self._operations: + raise CoreError("Unknown operation ID", code="unknown_operation") + operation = self._operations[operation_id] + operation.finished.wait(timeout) + with self._lock: + status, error = operation.status, operation.error + if error is not None: + raise error + return status + + def wait_active(self, timeout: float) -> bool: + with self._lock: + operation = self._operations.get(self._active) + return operation is None or operation.finished.wait(timeout) diff --git a/src/dynalab_core/playback.py b/src/dynalab_core/playback.py index 5f9712d..0a3b7d0 100644 --- a/src/dynalab_core/playback.py +++ b/src/dynalab_core/playback.py @@ -2,14 +2,17 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later - +from bisect import bisect_left +from copy import deepcopy +from dataclasses import dataclass import logging -from queue import Queue +from queue import Full, Queue import threading import time from uuid import UUID from dynalab_core.config import CoreConfig +from dynalab_core.types import WorkerStatus from dynalab_core.dlpak import DLPak from dynalab_core.errors import PlaybackError from dynalab_core.protocols.packets import ProtocolMessage @@ -19,103 +22,182 @@ from dynalab_core.protocols.packets.handshake import SignalDescriptor log = logging.getLogger(__name__) +@dataclass(slots=True) +class PlaybackValueDescriptor(ValueDescriptor): + """Internal sample carrying its playback generation through derive workers.""" + + generation: int + + class PlaybackEngine: def __init__( self, core_config: CoreConfig, data_queue: Queue[ProtocolMessage] ) -> None: self._core_config = core_config 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( + self._data_queue = data_queue + self._samples: list[tuple[int, UUID, float]] = [] + self._cursor = 0 + self._duration = 0 + self._generation = 0 + self._error: str | None = None + self._playing = threading.Event() + self._internal_timestamp = 0 + self._start_timestamp = 0 + self._condition = threading.Condition() + self._stop_event = threading.Event() + self._stopped_event = threading.Event() + self._worker_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 stop(self) -> bool: + with self._condition: + self._internal_timestamp = self._get_position_no_lock() + self._playing.clear() + self._generation += 1 + self._stop_event.set() + self._condition.notify_all() + stopped = self._stopped_event.wait(5.0) + if not stopped: + log.error("Playback worker did not stop within 5 seconds") + return stopped 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() + with self._condition: + if self._stop_event.is_set() or self._stopped_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" + ) + if not self._playing.is_set(): + self._start_timestamp = time.monotonic_ns() - self._internal_timestamp + self._generation += 1 + self._playing.set() + self._condition.notify_all() def pause(self) -> None: - self._playing.clear() + with self._condition: + self._internal_timestamp = self._get_position_no_lock() + self._playing.clear() + self._generation += 1 + self._condition.notify_all() def seek(self, timestamp_ns: int) -> None: if timestamp_ns < 0: raise PlaybackError( - "Playback timestamp cannot be negative", - code="invalid_timestamp", + "Playback timestamp cannot be negative", code="invalid_timestamp" ) - self._internal_timestamp = timestamp_ns + with self._condition: + if self._data is None: + raise PlaybackError("Cannot seek before loading data", code="no_data") + self._internal_timestamp = min(timestamp_ns, self._duration) + self._start_timestamp = time.monotonic_ns() - self._internal_timestamp + self._cursor = bisect_left( + self._samples, self._internal_timestamp, key=lambda sample: sample[0] + ) + self._generation += 1 + self._condition.notify_all() + + def is_playing(self) -> bool: + return self._playing.is_set() + + def is_current(self, message: PlaybackValueDescriptor) -> bool: + with self._condition: + return ( + not self._stop_event.is_set() and message.generation == self._generation + ) + + def get_status(self) -> WorkerStatus: + return WorkerStatus( + self._worker_thread.name, + "failed" + if self._error + else "stopped" + if self._stopped_event.is_set() + else "running", + self._error, + self._data_queue.qsize(), + self._data_queue.maxsize, + ) + + def _get_position_no_lock(self) -> int: + if self._playing.is_set(): + return min(self._duration, time.monotonic_ns() - self._start_timestamp) + return self._internal_timestamp + + def get_position(self) -> int: + with self._condition: + return self._get_position_no_lock() 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: - if ( - signal.origin == "derived" - and self._core_config.playback_mode == "reprocess" - ): - return None - return signal.to_pure() - return None + return next( + ( + signal + for signal in self.get_all_signal_descriptors() + if signal.id == uuid + ), + None, + ) def get_all_signal_descriptors(self) -> list[SignalDescriptor]: - if self._data is not None and self._data._manifest is not None: - if self._core_config.playback_mode == "reprocess": - return [ - signal - for signal in self._data._manifest.signals - if signal.origin != "derived" - ] - elif self._core_config.playback_mode == "read": - return [signal.to_pure() for signal in self._data._manifest.signals] - - return list[SignalDescriptor]() + with self._condition: + if self._data is None: + return [] + return [ + signal.to_pure() + for signal in self._data._manifest.signals + if self._core_config.playback_mode != "reprocess" + or signal.origin != "derived" + ] 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 - if self._core_config.playback_mode == "reprocess": - 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 + with self._condition: + 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", + ) + snapshot = deepcopy(data) + snapshot._data.sort() + samples = snapshot._data.get_samples() + first_timestamp = samples[0][0] if samples else 0 + self._duration = samples[-1][0] - first_timestamp if samples else 0 + samples = [ + (timestamp - first_timestamp, uuid, value) + for timestamp, uuid, value in samples + ] + if self._core_config.playback_mode == "reprocess": + ignored_signals = { + signal.id + for signal in snapshot._manifest.signals + if signal.origin == "derived" + } + samples = [ + sample for sample in samples if sample[1] not in ignored_signals + ] + self._data = snapshot + self._samples = samples + self._cursor = 0 + self._internal_timestamp = 0 + self._generation += 1 + self._condition.notify_all() def _playback_worker(self) -> None: try: self._run_playback() except Exception as error: + self._error = str(error) log.exception( "Playback worker failed", extra={ @@ -130,41 +212,34 @@ class PlaybackEngine: 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: + with self._condition: + if not self._playing.is_set(): + self._condition.wait() + continue + if self._cursor == len(self._samples): + self._playing.clear() + self._internal_timestamp = 0 + self._cursor = 0 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, + timestamp, uuid, value = self._samples[self._cursor] + self._internal_timestamp = self._get_position_no_lock() + remaining = timestamp - self._internal_timestamp + if remaining > 0: + # A control command wakes this wait, even across long gaps. + self._condition.wait(remaining / 1_000_000_000) + continue + try: + self._data_queue.put_nowait( + PlaybackValueDescriptor( + signal_id=uuid, + value=value, + timestamp=timestamp + self._start_timestamp, + generation=self._generation, + ) ) - ) - - self._playing.clear() - self._internal_timestamp = 0 + except Full: + # Keep controls responsive while the consumer catches up. + self._condition.wait(0.01) + continue + self._cursor += 1 diff --git a/src/dynalab_core/protocols/endpoint.py b/src/dynalab_core/protocols/endpoint.py index 6f3baa5..07360b1 100644 --- a/src/dynalab_core/protocols/endpoint.py +++ b/src/dynalab_core/protocols/endpoint.py @@ -2,6 +2,8 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later +from collections.abc import Callable +from copy import deepcopy import logging from queue import Empty, Full, Queue import threading @@ -9,7 +11,8 @@ from threading import Lock, RLock, Thread from time import monotonic, sleep from uuid import UUID -from dynalab_core.types import CoreModeState +from dynalab_core.types import CoreModeState, ConnectorStatus +from dynalab_core.values import SignalValueDescriptor from dynalab_core.constants import HELLO_PACKET from dynalab_core.errors import ConnectorError from dynalab_core.protocols.packets import ProtocolMessage @@ -44,7 +47,8 @@ class ConnectorEndpoint: self._core_mode = core_mode - self._connector_hello = hello + self._connector_hello = deepcopy(hello) + self._signals = {signal.id: signal for signal in self._connector_hello.signals} self._timed_out_event = timeout_event self._reject_event = reject_event self._stop_event = threading.Event() @@ -327,7 +331,14 @@ class ConnectorEndpoint: signal.id for signal in self._connector_hello.signals ): try: - self._core_input_queue.put_nowait(message) + self._core_input_queue.put_nowait( + SignalValueDescriptor( + message.signal_id, + message.value, + message.timestamp, + self._signals[message.signal_id], + ) + ) except Full: self._log_queue_full( "core_input", @@ -347,7 +358,14 @@ class ConnectorEndpoint: for signal in self._connector_hello.signals ): try: - self._core_input_queue.put_nowait(value) + self._core_input_queue.put_nowait( + SignalValueDescriptor( + value.signal_id, + value.value, + value.timestamp, + self._signals[value.signal_id], + ) + ) except Full: self._log_queue_full( "core_input", @@ -484,7 +502,14 @@ class ConnectorEndpoint: class ConnectorRegistry: - def __init__(self, core_input_queue: Queue, core_mode: CoreModeState) -> None: + def __init__( + self, + core_input_queue: Queue, + core_mode: CoreModeState, + get_derived_ids: Callable[[], set[UUID]] | None = None, + ) -> None: + self._revision = 0 + self._get_derived_ids = get_derived_ids self._endpoints: dict[UUID, ConnectorEndpoint] = {} self._lock = RLock() self._core_input_queue = core_input_queue @@ -514,6 +539,25 @@ class ConnectorRegistry: code="already_registered", ) + signal_ids = [signal.id for signal in hello.signals] + used_ids = { + signal.id + for endpoint in self._endpoints.values() + for signal in endpoint.get_all_signal_descriptors() + } + if self._get_derived_ids is not None: + used_ids.update(self._get_derived_ids()) + if len(signal_ids) != len(set(signal_ids)) or used_ids.intersection( + signal_ids + ): + raise ConnectorError( + "Signal IDs must be unique across connectors and derives", + code="duplicate_signal", + ) + if any(signal.origin != "source" for signal in hello.signals): + raise ConnectorError( + "Connector signals must have source origin", code="invalid_signal" + ) endpoint = ConnectorEndpoint( hello, timeout_event, @@ -523,6 +567,7 @@ class ConnectorRegistry: ) self._endpoints[hello.connector_uuid] = endpoint + self._revision += 1 endpoint_count = len(self._endpoints) try: @@ -532,6 +577,7 @@ class ConnectorRegistry: with self._lock: if self._endpoints.get(hello.connector_uuid) is endpoint: del self._endpoints[hello.connector_uuid] + self._revision += 1 log.exception( "Failed to start connector endpoint %s", hello.connector_uuid, @@ -576,6 +622,7 @@ class ConnectorRegistry: ) return False del self._endpoints[connector_uuid] + self._revision += 1 endpoint_count = len(self._endpoints) else: log.debug( @@ -599,6 +646,32 @@ class ConnectorRegistry: ) return True + def get_revision(self) -> int: + with self._lock: + return self._revision + + def get_statuses(self) -> list[ConnectorStatus]: + with self._lock: + return [ + ConnectorStatus( + id=endpoint.uuid(), + name=endpoint._connector_hello.connector_name, + version=endpoint._connector_hello.connector_version, + signal_ids=tuple( + signal.id for signal in endpoint.get_all_signal_descriptors() + ), + state="failed" + if ( + endpoint._reject_event.is_set() + or endpoint._timed_out_event.is_set() + ) + else "stopping" + if endpoint._stop_event.is_set() + else "connected", + ) + for endpoint in self._endpoints.values() + ] + def get(self, connector_uuid: UUID) -> ConnectorEndpoint | None: with self._lock: return self._endpoints.get(connector_uuid) @@ -630,6 +703,7 @@ class ConnectorRegistry: with self._lock: endpoints = tuple(self._endpoints.values()) self._endpoints.clear() + self._revision += 1 log.debug( "Stopping connector registry with %d endpoint(s)", diff --git a/src/dynalab_core/protocols/json/server.py b/src/dynalab_core/protocols/json/server.py index 95925fd..6f7bd0c 100644 --- a/src/dynalab_core/protocols/json/server.py +++ b/src/dynalab_core/protocols/json/server.py @@ -17,13 +17,13 @@ from dynalab_core.constants import HELLO_PACKET from dynalab_core.protocols.endpoint import ConnectorEndpoint, ConnectorRegistry from dynalab_core.errors import ConnectorError, JsonServerError from dynalab_core.protocols.json.wire import read_message, write_message -from dynalab_core.protocols.packets.data import JsonValueBatch, JsonValueDescriptor from dynalab_core.protocols.packets.handshake import ( - ConnectorHello, HandshakeAccepted, JsonConnectorHello, ) +from dynalab_core.types import WorkerStatus + log = logging.getLogger(__name__) @@ -35,6 +35,7 @@ class JsonServer: self._server: Server | None = None self._started_event: threading.Event = threading.Event() self._startup_error: Exception | None = None + self._error: str | None = None self._debug_timeout_test: bool = False self._config = config self._timeout: float = 10 @@ -44,6 +45,10 @@ class JsonServer: self._handler_tasks: set[asyncio.Task[None]] = set() def start(self) -> None: + if self._stop_event.is_set() or self._thread is not None: + raise JsonServerError( + "Server cannot be started in its current state", code="invalid_state" + ) log.info( "Starting JSON server on %s", self._config.bind_str(), @@ -97,6 +102,8 @@ class JsonServer: def stop(self) -> bool: log.info("Stopping JSON server", extra={"event": "json_server.stopping"}) self._stop_event.set() + if self._thread is None: + self._stopped_event.set() if not self._stopped_event.wait(10): log.error( "JSON server did not stop within 10 seconds", @@ -107,6 +114,17 @@ class JsonServer: log.info("JSON server stopped", extra={"event": "json_server.stopped"}) return True + def get_status(self) -> WorkerStatus: + return WorkerStatus( + "json_server_thread", + "failed" + if self._error + else "running" + if self._thread and self._thread.is_alive() + else "stopped", + self._error, + ) + def _json_server_thread_main(self) -> None: log.debug( "JSON server thread started", @@ -118,6 +136,7 @@ class JsonServer: try: asyncio.run(self._run_json_server()) except Exception as error: + self._error = str(error) startup_complete = self._server is not None log.exception( "JSON server thread failed", @@ -162,6 +181,7 @@ class JsonServer: }, ) self._startup_error = error + self._error = str(error) if not self._debug_timeout_test: self._started_event.set() diff --git a/src/dynalab_core/types.py b/src/dynalab_core/types.py index 88a627d..cd4e809 100644 --- a/src/dynalab_core/types.py +++ b/src/dynalab_core/types.py @@ -5,6 +5,7 @@ from dataclasses import dataclass from typing import Literal +from uuid import UUID CoreMode = Literal["realtime", "playback", "offline"] @@ -14,3 +15,30 @@ BufferSource = Literal["recording", "processing", "playback"] @dataclass class CoreModeState: value: CoreMode = "realtime" + + +@dataclass(frozen=True) +class WorkerStatus: + name: str + state: Literal["running", "stopped", "failed", "blocked"] + error: str | None = None + queue_size: int = 0 + queue_capacity: int = 0 + dropped_samples: int = 0 + + +@dataclass(frozen=True) +class ConnectorStatus: + id: "UUID" + name: str + version: str + signal_ids: tuple["UUID", ...] + state: Literal["connected", "stopping", "failed"] + + +@dataclass(frozen=True) +class DeriveUnitStatus: + id: "UUID" + input_signal_ids: tuple["UUID", ...] + output_signal_id: "UUID" + worker: WorkerStatus diff --git a/src/dynalab_core/values.py b/src/dynalab_core/values.py index 037fbb2..c8b48ad 100644 --- a/src/dynalab_core/values.py +++ b/src/dynalab_core/values.py @@ -3,11 +3,20 @@ # SPDX-License-Identifier: GPL-3.0-or-later +from dataclasses import dataclass from threading import Lock -from time import monotonic, monotonic_ns +from time import monotonic_ns from uuid import UUID from dynalab_core.protocols.packets.data import ValueDescriptor +from dynalab_core.protocols.packets.handshake import SignalDescriptor + + +@dataclass(slots=True) +class SignalValueDescriptor(ValueDescriptor): + """Ingress/derive value retaining metadata across disconnects and unbinding.""" + + signal: SignalDescriptor class Value: @@ -44,9 +53,6 @@ class Value: last_updated = self._last_updated now = monotonic_ns() - if ( - now > last_updated + self._timeout_ms * 1_000_000 - or last_updated == 0 - ): + if now > last_updated + self._timeout_ms * 1_000_000 or last_updated == 0: return None return value, last_updated