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

Revised error codes and functional PlaybackEngine

Browse Source 
All internal and public errors have been revised (GPT 5.6 Sol) to return
more detail without horrendously long error classes, this has reduced
the number of error types but each now carries a reason and a code for
differentiation if needed

PlaybackEngine has been implemented and is functional, core mode changes
what signal descriptors are returned to avoid duplicates, the playback
engine currently discards the derived signals that would be recorded in
the DLPak and derived signals are reprocessed live

Future will require the behavior listed above to be selectable so one of
two options are possible:
A - Derived signals are reprocessed live by the relevant derive units
(current behavior
B - Derived signals are played back directly from the PlaybackEngine,
this requires turning off the routing to the live derive units to avoid
duplicate values

This will also require the option to perform offline processing with
derive units to be able to create derived signals after the fact, this
will likely incur a new 'offline' mode in the core to gate things
correctly, this will likely be useful for heavier processing that cannot
be done live, filters that require a lookahead, or more precise derived
signals using interpolated signals for higher acurracy which also brings
the ability to offline process on a fixed timebase instead of following
either input signal to the derived signal
This commit is contained in:
h3cx
2026-09-13 12:26:12 +02:00
parent bbc5aac891
commit 9407bfeb79
18 changed files with 1160 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/manual/core.py
+6 -6
View File
@@ -43,12 +43,12 @@ unit_id = dl_core.bind_derive_unit(
SignalDescriptor(id=uuid4(), name="Dummy sum", type="number", timeout_ms=5000),
)
for i in range(10):
dl_core.wait(1)
values = dl_core.get_all_live_values()
log.debug(f"Core values: {values}")
dl_core.set_mode("playback")
# for i in range(10):
# dl_core.wait(1)
# values = dl_core.get_all_live_values()
# log.debug(f"Core values: {values}")
#
# dl_core.set_mode("playback")
try:
test/manual/core_tui.py
+251
View File
@@ -0,0 +1,251 @@
import logging
import select
import sys
import termios
import time
import tty
from contextlib import contextmanager
from pathlib import Path
from typing import Iterator
from uuid import UUID, uuid4
from rich.console import Group
from rich.live import Live
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
from dynalab_core import Core
from dynalab_core.config import CoreConfig
from dynalab_core.errors import DynaLabError
from dynalab_core.protocols.packets.handshake import SignalDescriptor
HOST = "127.0.0.1"
PORT = 8765
DISPLAY_FREQUENCY_HZ = 5.0
signal_id_1: UUID = UUID("3f32cea3-d872-4c16-a0a2-54b57171aeb2")
signal_id_2: UUID = UUID("6578ac37-99d1-410c-b3f7-d049339919b2")
@contextmanager
def keyboard_input() -> Iterator[None]:
"""Read individual keys and restore the terminal settings on exit."""
if not sys.stdin.isatty():
raise RuntimeError("This example must be run in an interactive terminal")
file_descriptor = sys.stdin.fileno()
previous_settings = termios.tcgetattr(file_descriptor)
tty.setcbreak(file_descriptor)
try:
yield
finally:
termios.tcsetattr(file_descriptor, termios.TCSADRAIN, previous_settings)
class Dashboard:
def __init__(self, core: Core, config: CoreConfig) -> None:
self.core = core
self.config = config
self.running = True
self.playback_loaded = False
self.message = "Waiting for a connector"
self.message_style = "dim"
def handle_key(self, key: str) -> None:
try:
if key.lower() == "q":
self.running = False
elif key.lower() == "r":
self.toggle_recording()
elif key.lower() == "m":
self.toggle_mode()
elif key.lower() == "l":
self.load_playback()
elif key == " ":
self.toggle_playback()
except DynaLabError as error:
self.set_message(str(error), "bold red")
def toggle_recording(self) -> None:
if self.core.get_mode() != "realtime":
self.set_message("Recording is only available in realtime mode", "yellow")
return
if self.core.get_recording_state():
self.core.stop_recording()
self.set_message("Recording stopped and prepared for playback", "green")
else:
self.core.start_recording()
self.set_message("Recording started", "bold red")
def toggle_mode(self) -> None:
if self.core.get_recording_state():
self.set_message("Stop recording before changing mode", "yellow")
return
if self.core.get_mode() == "realtime":
self.core.set_mode("playback")
self.set_message(
"Playback mode selected; press L to load a recording", "green"
)
else:
if self.core.get_playback_state():
self.core.pause()
self.core.set_mode("realtime")
self.set_message("Realtime mode selected", "green")
def load_playback(self) -> None:
if self.core.get_mode() != "playback":
self.set_message("Switch to playback mode before loading", "yellow")
return
self.core.load_playback_engine()
self.playback_loaded = True
self.set_message("Internal processing buffer loaded", "green")
def toggle_playback(self) -> None:
if self.core.get_mode() != "playback":
self.set_message("Switch to playback mode before pressing Space", "yellow")
return
if self.core.get_playback_state():
self.core.pause()
self.set_message("Playback paused", "yellow")
else:
self.core.play()
self.set_message("Playback started", "green")
def set_message(self, message: str, style: str) -> None:
self.message = message
self.message_style = style
def render(self) -> Group:
mode = self.core.get_mode()
recording = self.core.get_recording_state()
playing = self.core.get_playback_state()
status = Table.grid(expand=True)
status.add_column(ratio=1)
status.add_column(ratio=1)
status.add_column(ratio=1)
status.add_row(
Text(f"Mode: {mode.title()}", style="bold cyan"),
Text(
"Recording: ON" if recording else "Recording: OFF",
style="bold red" if recording else "dim",
),
Text(
self.playback_status(playing),
style="bold green" if playing else "dim",
),
)
values = self.core.get_all_live_values()
descriptors = sorted(
self.core.get_all_signal_descriptors(), key=lambda signal: signal.name
)
value_table = Table(expand=True, show_lines=False)
value_table.add_column("Signal", style="cyan", ratio=2)
value_table.add_column("Origin", style="dim", ratio=1)
value_table.add_column("Value", justify="right", ratio=1)
value_table.add_column("Unit", style="dim", ratio=1)
if descriptors:
for descriptor in descriptors:
value = values.get(descriptor.id)
value_table.add_row(
descriptor.name,
descriptor.origin,
f"{value: .5f}" if value is not None else "—",
descriptor.unit or "",
)
else:
value_table.add_row("Waiting for connector…", "", "—", "")
controls = Text.from_markup(
"[bold]R[/bold] Record [bold]M[/bold] Mode "
"[bold]L[/bold] Load [bold]Space[/bold] Play/Pause "
"[bold]Q[/bold] Quit"
)
return Group(
Panel(
status,
title="DynaLab Core",
subtitle=f"{self.config.bind_str()} · values refresh at 5 Hz",
border_style="blue",
),
Panel(value_table, title="Live values", border_style="cyan"),
Panel(Text(self.message, style=self.message_style), title="Status"),
Panel(controls, title="Hotkeys", border_style="blue"),
)
def playback_status(self, playing: bool) -> str:
if playing:
return "Playback: PLAYING"
if self.playback_loaded:
return "Playback: LOADED"
return "Playback: NOT LOADED"
def main() -> None:
logging.basicConfig(level=logging.CRITICAL)
config = CoreConfig(host=HOST, port=PORT)
core = Core(config)
processing_source = Path(__file__).with_name("process.py").read_text(
encoding="utf-8"
)
dashboard = Dashboard(core, config)
core.bind_derive_unit(
processing_source,
[
SignalDescriptor(
id=signal_id_1,
name="Dummy signal 1",
type="number",
timeout_ms=5000,
),
SignalDescriptor(
id=signal_id_2,
name="Dummy signal 2",
type="number",
timeout_ms=5000,
),
],
SignalDescriptor(
id=uuid4(),
name="Dummy product",
type="number",
timeout_ms=5000,
origin="derived",
),
)
try:
core.start()
with keyboard_input(), Live(
dashboard.render(),
auto_refresh=False,
screen=True,
) as live:
while dashboard.running:
update_started = time.monotonic()
while select.select([sys.stdin], [], [], 0)[0]:
dashboard.handle_key(sys.stdin.read(1))
live.update(dashboard.render(), refresh=True)
elapsed = time.monotonic() - update_started
time.sleep(max(0.0, 1.0 / DISPLAY_FREQUENCY_HZ - elapsed))
except KeyboardInterrupt:
pass
finally:
core.stop()
if __name__ == "__main__":
main()
test/manual/peer.py
+11 -4
View File
@@ -3,6 +3,7 @@ import logging
import threading
from concurrent.futures import CancelledError as FutureCancelledError
from concurrent.futures import TimeoutError as FutureTimeoutError
from math import sin, tau
from statistics import mean
from time import monotonic, monotonic_ns, perf_counter
from uuid import UUID, uuid4
@@ -27,6 +28,7 @@ PORT = 8765
# Set to None to send as quickly as possible.
# For a controlled rate, use something like 5_000.0.
TARGET_FREQUENCY_HZ: float | None = 1000
SINE_FREQUENCY_HZ = 0.5
FREQUENCY_SAMPLE_SIZE = 20000
@@ -88,6 +90,7 @@ def value_sender_thread(
"""
intervals: list[float] = []
previous_send_time: float | None = None
sine_start_time = perf_counter()
if TARGET_FREQUENCY_HZ is not None:
period = 1.0 / TARGET_FREQUENCY_HZ
@@ -117,16 +120,20 @@ def value_sender_thread(
next_send_time += period
sine_value = sin(
tau * SINE_FREQUENCY_HZ * (perf_counter() - sine_start_time)
)
timestamp = monotonic_ns()
message = ValueBatch(values=[])
value1 = ValueDescriptor(
signal_id=signal_id_1,
value=2.0,
timestamp=monotonic_ns(),
value=2.0 * sine_value,
timestamp=timestamp,
)
value2 = ValueDescriptor(
signal_id=signal_id_2,
value=1.0,
timestamp=monotonic_ns(),
value=sine_value,
timestamp=timestamp,
)
message.values.append(value1)
message.values.append(value2)
test/test_core.py
+8 -2
View File
@@ -5,7 +5,7 @@ from uuid import uuid4
import pytest
from dynalab_core import Core
from dynalab_core.config import CoreConfig
from dynalab_core.errors import CoreStateMismatchError
from dynalab_core.errors import CoreError
from dynalab_core.protocols.endpoint import ConnectorEndpoint
from dynalab_core.protocols.packets.data import ValueDescriptor
from dynalab_core.protocols.packets.handshake import ConnectorHello
@@ -20,7 +20,7 @@ def test_core_cannot_start_twice(caplog: pytest.LogCaptureFixture) -> None:
core.start()
try:
with pytest.raises(CoreStateMismatchError):
with pytest.raises(CoreError, match="can only be started once") as raised:
core.start()
finally:
core.stop()
@@ -37,6 +37,7 @@ def test_core_cannot_start_twice(caplog: pytest.LogCaptureFixture) -> None:
)
assert rejection.levelno == logging.WARNING
assert rejection.core_state == "started"
assert raised.value.code == "invalid_state"
def test_core_input_worker_logs_failure(
@@ -87,9 +88,14 @@ def test_core_mode_is_shared_with_endpoints() -> None:
try:
assert endpoint._core_mode is core._mode
assert core.get_mode() == "realtime"
core.set_mode("playback")
assert endpoint._core_mode.value == "playback"
assert core.get_mode() == "playback"
with pytest.raises(CoreError) as raised:
core.load_playback_engine()
assert raised.value.code == "no_playback_data"
finally:
core.stop()
test/test_dlpak.py
+18
View File
@@ -27,3 +27,21 @@ def test_dlpak_logs_written_archive(caplog: pytest.LogCaptureFixture, tmp_path)
assert record.output_filename == "recording.dlpak"
assert record.sample_count == 1
assert record.signal_count == 1
def test_dlpak_read_restores_manifest_and_data(tmp_path) -> None:
timestamp = datetime.now(timezone.utc)
signal = SignalDescriptor(id=uuid4(), name="Signal", type="number")
buffer = ValueBuffer()
buffer.append(1, signal.id, 2.0)
package = DLPak()
package.set_data(buffer)
package.set_manifest(timestamp, [signal])
package.write(tmp_path, "recording")
loaded_package = DLPak()
loaded_package.read(tmp_path / "recording.dlpak")
assert loaded_package._manifest == package._manifest
assert loaded_package._data is not None
assert loaded_package._data.export_csv() == buffer.export_csv()
test/test_errors.py
+146
View File
@@ -0,0 +1,146 @@
from datetime import datetime, timezone
from queue import Queue
from uuid import uuid4
from zipfile import BadZipFile
import pytest
from dynalab_core.buffer import ValueBuffer
from dynalab_core.derive import DeriveRegistry, DeriveUnit
from dynalab_core.dlpak import DLPak
from dynalab_core.errors import (
CoreError,
DLPakError,
DeriveError,
DynaLabError,
PlaybackError,
)
from dynalab_core.playback import PlaybackEngine
from dynalab_core.protocols.packets.data import ValueDescriptor
from dynalab_core.protocols.packets.handshake import SignalDescriptor
def _identity(
value: ValueDescriptor, output: SignalDescriptor
) -> ValueDescriptor:
return ValueDescriptor(
signal_id=output.id,
value=value.value,
timestamp=value.timestamp,
)
def test_subsystem_errors_share_public_base_and_code() -> None:
error = PlaybackError("Playback data is unavailable", code="no_data")
assert isinstance(error, DynaLabError)
assert error.code == "no_data"
assert str(error) == "Playback data is unavailable"
def test_core_rejects_invalid_mode_and_missing_recording() -> None:
from dynalab_core import Core
core = Core.__new__(Core)
core._processing_buffer = None
with pytest.raises(CoreError, match="Unknown core mode") as mode:
core.set_mode("invalid") # type: ignore[arg-type]
with pytest.raises(CoreError, match="no recording") as recording:
core.write(".", "unused")
assert mode.value.code == "invalid_mode"
assert recording.value.code == "no_recording"
def test_dlpak_requires_data_before_manifest() -> None:
package = DLPak()
with pytest.raises(DLPakError, match="before setting its data") as raised:
package.set_manifest(datetime.now(timezone.utc), [])
assert raised.value.code == "no_data"
def test_dlpak_wraps_archive_read_failures(tmp_path) -> None:
invalid_archive = tmp_path / "invalid.dlpak"
invalid_archive.write_text("not a zip archive")
with pytest.raises(DLPakError, match="Could not read") as raised:
DLPak().read(invalid_archive)
assert raised.value.code == "read_failed"
assert isinstance(raised.value.__cause__, BadZipFile)
def test_derive_unit_reports_argument_mismatch() -> None:
input_signal = SignalDescriptor(id=uuid4(), name="Input", type="number")
output_signal = SignalDescriptor(
id=uuid4(), name="Output", type="number", origin="derived"
)
unit = DeriveUnit(uuid4(), _identity, [input_signal], output_signal, Queue())
try:
with pytest.raises(DeriveError, match="expected 1 input") as raised:
unit.process_offline([])
finally:
unit.stop()
assert raised.value.code == "argument_mismatch"
def test_derive_registry_wraps_invalid_source() -> None:
registry = DeriveRegistry(Queue(), lambda signal_id: None)
output_signal = SignalDescriptor(
id=uuid4(), name="Output", type="number", origin="derived"
)
try:
with pytest.raises(DeriveError, match="Could not load") as raised:
registry.register("def broken(", [], output_signal)
finally:
registry.stop()
assert raised.value.code == "invalid_source"
assert isinstance(raised.value.__cause__, SyntaxError)
def test_derive_unit_rejects_configured_input_mismatch() -> None:
output_signal = SignalDescriptor(
id=uuid4(), name="Output", type="number", origin="derived"
)
with pytest.raises(DeriveError, match="input signal") as raised:
DeriveUnit(uuid4(), _identity, [], output_signal, Queue())
assert raised.value.code == "input_signal_mismatch"
def test_playback_rejects_incomplete_data_and_negative_seek() -> None:
engine = PlaybackEngine(Queue())
try:
with pytest.raises(PlaybackError, match="incomplete") as incomplete:
engine.set_data(DLPak())
with pytest.raises(PlaybackError, match="cannot be negative") as timestamp:
engine.seek(-1)
finally:
engine.stop()
assert incomplete.value.code == "incomplete_data"
assert timestamp.value.code == "invalid_timestamp"
def test_playback_accepts_complete_dlpak() -> None:
signal = SignalDescriptor(id=uuid4(), name="Input", type="number")
data = ValueBuffer()
data.append(1, signal.id, 2.0)
package = DLPak()
package.set_data(data)
package.set_manifest(datetime.now(timezone.utc), [signal])
engine = PlaybackEngine(Queue())
try:
engine.set_data(package)
finally:
engine.stop()
test/test_json_server.py
+9 -7
View File
@@ -7,12 +7,9 @@ from uuid import uuid4
import pytest
from dynalab_core import Core
from dynalab_core.config import CoreConfig
from dynalab_core.errors import JsonServerError
from dynalab_core.protocols.constants import PROTOCOL_VERSION
from dynalab_core.protocols.common import VersionDescriptor
from dynalab_core.protocols.json.errors import (
JsonServerStartupError,
JsonServerTimeoutError,
)
from dynalab_core.protocols.packets.handshake import ConnectorHello, SignalDescriptor
from test.common import find_available_port
@@ -39,7 +36,7 @@ def test_json_server_raises_timeout_error(
with caplog.at_level(logging.DEBUG, logger="dynalab_core"):
try:
with pytest.raises(JsonServerTimeoutError):
with pytest.raises(JsonServerError, match="did not start") as raised:
core.start()
finally:
core.stop()
@@ -51,6 +48,7 @@ def test_json_server_raises_timeout_error(
)
assert timeout_record.levelno == logging.ERROR
assert timeout_record.port == port
assert raised.value.code == "start_timeout"
def test_json_server_raises_startup_error(
@@ -65,7 +63,7 @@ def test_json_server_raises_startup_error(
core1.start()
try:
with pytest.raises(JsonServerStartupError):
with pytest.raises(JsonServerError, match="failed to start") as raised:
core2.start()
finally:
core1.stop()
@@ -79,6 +77,8 @@ def test_json_server_raises_startup_error(
assert failure_record.levelno == logging.ERROR
assert failure_record.port == port
assert failure_record.exc_info is not None
assert raised.value.code == "startup_failed"
assert isinstance(raised.value.__cause__, OSError)
def test_json_server_logs_invalid_handshake(
@@ -117,7 +117,7 @@ def test_json_server_reports_unexpected_thread_startup_failure(
with caplog.at_level(logging.DEBUG, logger="dynalab_core"):
try:
with pytest.raises(JsonServerStartupError):
with pytest.raises(JsonServerError) as raised:
core.start()
finally:
core.stop()
@@ -131,6 +131,8 @@ def test_json_server_reports_unexpected_thread_startup_failure(
assert failures[0].startup_complete is False
assert failures[0].exception_type == "RuntimeError"
assert failures[0].exc_info is not None
assert raised.value.code == "startup_failed"
assert isinstance(raised.value.__cause__, RuntimeError)
def test_rejected_handshake_is_not_logged_as_connected(