Exergie/dynalab-core
1
1 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Final tweaks for beta

Browse Source
This commit is contained in:
h3cx committed 2026-10-14 22:13:14 +02:00
1 parent 0ae0d90f72
commit 93b7ff40ba
12 files changed
+1161 -394
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/dynalab_core/__init__.py
+338 -164
View File
@@ -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={
src/dynalab_core/buffer.py
+7
View File
@@ -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
src/dynalab_core/config.py
+2
View File
@@ -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}"
src/dynalab_core/derive.py
+296 -103
View File
@@ -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",
src/dynalab_core/dlpak.py
+33 -5
View File
@@ -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",
src/dynalab_core/dlunit.py
+24 -6
View File
@@ -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",
src/dynalab_core/operations.py
+142
View File
@@ -0,0 +1,142 @@
# Copyright (C) 2026 Hector van der Aa <hector@h3cx.dev>
# Copyright (C) 2026 Association Exergie <association.exergie@gmail.com>
# 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)
src/dynalab_core/playback.py
+179 -104
View File
@@ -2,14 +2,17 @@
# Copyright (C) 2026 Association Exergie <association.exergie@gmail.com>
# 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
src/dynalab_core/protocols/endpoint.py
+79 -5
View File
@@ -2,6 +2,8 @@
# Copyright (C) 2026 Association Exergie <association.exergie@gmail.com>
# 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)",
src/dynalab_core/protocols/json/server.py
+22 -2
View File
@@ -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()
src/dynalab_core/types.py
+28
View File
@@ -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
src/dynalab_core/values.py
+11 -5
View File
@@ -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