Files
dynalab-core/test/manual/core_tui.py
T
h3cx 6b954287a4 Initial offline process implementation
Implemented the initial offline processing implementation, with
dependency layer building and ordered processing

Processing is still single threaded and unoptimized so it can take some
time to complete, the next step is to move from a single threaded design
to a ThreadPoolExecutor that splits up the buffer into a set number of
chunks for processing
2026-09-27 17:45:57 +02:00

437 lines
15 KiB
Python

from collections import deque
import logging
import select
import sys
import termios
from threading import Lock
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.dlunit import DLUnit
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
LOG_HISTORY_SIZE = 1_000
VISIBLE_LOG_LINES = 20
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 TuiLogHandler(logging.Handler):
"""Keep recent DynaLab Core log messages available to the live display."""
def __init__(self) -> None:
super().__init__()
self._entries: deque[tuple[str, str, str, str]] = deque(maxlen=LOG_HISTORY_SIZE)
self._lock = Lock()
self.setFormatter(
logging.Formatter(
"%(asctime)s %(levelname)-8s %(name)s: %(message)s",
datefmt="%H:%M:%S",
)
)
def emit(self, record: logging.LogRecord) -> None:
try:
message = record.getMessage().replace("\n", " ↳ ")
if record.exc_info is not None and record.exc_info[1] is not None:
message = f"{message} ({record.exc_info[1]})"
level = record.levelname
level_style = {
logging.DEBUG: "dim",
logging.INFO: "white",
logging.WARNING: "yellow",
logging.ERROR: "bold red",
logging.CRITICAL: "bold white on red",
}.get(record.levelno, "white")
with self._lock:
self._entries.append(
(
self.formatter.formatTime(record, datefmt="%H:%M:%S"),
level,
message,
level_style,
)
)
except Exception:
self.handleError(record)
def latest(self, limit: int) -> list[tuple[str, str, str, str]]:
with self._lock:
return list(self._entries)[-limit:]
class Dashboard:
def __init__(
self, core: Core, config: CoreConfig, log_handler: TuiLogHandler
) -> None:
self.core = core
self.config = config
self.log_handler = log_handler
self.running = True
self.mode_selection_open = False
self.offline_loaded = False
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 self.mode_selection_open:
self.select_mode(key)
elif key.lower() == "r":
self.toggle_recording()
elif key.lower() == "m":
self.open_mode_selection()
elif key.lower() == "l":
self.load_current_stage()
elif key.lower() == "p":
self.reprocess_offline()
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; select Offline and press L to load it", "green"
)
else:
self.core.start_recording()
self.set_message("Recording started", "bold red")
def open_mode_selection(self) -> None:
if self.core.get_recording_state():
self.set_message("Stop recording before changing mode", "yellow")
return
self.mode_selection_open = True
self.set_message("Choose a mode: 1 Realtime, 2 Offline, 3 Playback", "cyan")
def select_mode(self, key: str) -> None:
modes = {"1": "realtime", "2": "offline", "3": "playback"}
if key == "\x1b" or key.lower() == "m":
self.mode_selection_open = False
self.set_message("Mode selection cancelled", "dim")
return
mode = modes.get(key)
if mode is None:
self.set_message(
"Choose 1 Realtime, 2 Offline, 3 Playback, or Esc", "yellow"
)
return
if self.core.get_playback_state() and mode != "playback":
self.core.pause()
self.core.set_mode(mode)
self.mode_selection_open = False
if mode == "offline":
self.set_message(
"Offline mode selected; press L to load the recording", "green"
)
elif mode == "playback":
self.set_message(
"Playback mode selected; press L to load the offline buffer", "green"
)
else:
self.set_message("Realtime mode selected", "green")
def load_current_stage(self) -> None:
mode = self.core.get_mode()
if mode == "offline":
self.core.load_processing_buffer()
self.offline_loaded = True
self.playback_loaded = False
self.set_message(
"Recording loaded into the offline buffer; select Playback and press L",
"green",
)
return
if mode == "playback":
self.core.load_playback_engine()
self.playback_loaded = True
self.set_message("Offline buffer loaded into playback", "green")
return
self.set_message("Select Offline or Playback before loading", "yellow")
def reprocess_offline(self) -> None:
if self.core.get_mode() != "offline":
self.set_message("Reprocessing is only available in offline mode", "yellow")
return
self.core.reprocess()
self.set_message("Offline reprocessing requested", "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.buffer_status(mode, playing),
style="bold green" if playing or self.offline_loaded 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] Choose mode "
"[bold]L[/bold] Load stage [bold]P[/bold] Reprocess offline "
"[bold]Space[/bold] Play/Pause "
"[bold]Q[/bold] Quit"
)
log_table = Table.grid(expand=True, padding=(0, 1))
log_table.add_column(style="dim", width=8, no_wrap=True)
log_table.add_column(width=8, no_wrap=True)
log_table.add_column(ratio=1, no_wrap=True, overflow="ellipsis")
log_entries = self.log_handler.latest(VISIBLE_LOG_LINES)
if log_entries:
for timestamp, level, message, style in log_entries:
log_table.add_row(
Text(timestamp),
Text(level, style=style),
Text(message, style=style),
)
else:
log_table.add_row("", "", Text("No DynaLab Core logs yet", style="dim"))
for _ in range(VISIBLE_LOG_LINES - max(1, len(log_entries))):
log_table.add_row("", "", "")
panels = [
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"),
]
if self.mode_selection_open:
mode_picker = Table.grid(expand=True, padding=(0, 2))
mode_picker.add_column(width=3, style="bold cyan")
mode_picker.add_column(ratio=1)
mode_picker.add_row("1", "Realtime — receive connector data and record")
mode_picker.add_row(
"2", "Offline — load a completed recording into a processing buffer"
)
mode_picker.add_row("3", "Playback — load the offline buffer and replay it")
mode_picker.add_row("Esc", "Cancel")
panels.append(
Panel(mode_picker, title="Choose mode", border_style="yellow")
)
panels.append(
Panel(
log_table,
title=f"Logs · latest {VISIBLE_LOG_LINES}",
border_style="magenta",
)
)
return Group(*panels)
def buffer_status(self, mode: str, playing: bool) -> str:
if playing:
return "Playback: PLAYING"
if mode == "playback" and self.playback_loaded:
return "Playback: LOADED"
if mode == "offline" and self.offline_loaded:
return "Offline: LOADED"
if self.offline_loaded:
return "Offline buffer: READY"
return "Offline buffer: NOT LOADED"
def main() -> None:
logging.basicConfig(level=logging.CRITICAL)
log_handler = TuiLogHandler()
dynalab_logger = logging.getLogger("dynalab_core")
previous_log_level = dynalab_logger.level
previous_propagation = dynalab_logger.propagate
dynalab_logger.setLevel(logging.DEBUG)
dynalab_logger.propagate = False
dynalab_logger.addHandler(log_handler)
config = CoreConfig(host=HOST, port=PORT, playback_mode="read")
core = Core(config)
processing_source = (
Path(__file__).with_name("process.py").read_text(encoding="utf-8")
)
processing_source_2 = (
Path(__file__).with_name("process2.py").read_text(encoding="utf-8")
)
dashboard = Dashboard(core, config, log_handler)
derive_unit = DLUnit()
derive_unit.set_manifest(
uuid4(),
[
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",
),
)
derive_unit.set_data(processing_source)
derive_unit_2 = DLUnit()
derive_unit_2.set_manifest(
unit_uuid=uuid4(),
input_signals=[
SignalDescriptor(
id=derive_unit._manifest.return_signal.id,
name="Input signal",
type="number",
timeout_ms=5000,
origin="derived",
)
],
return_signal=SignalDescriptor(
id=uuid4(),
name="Sq derive",
type="number",
timeout_ms=5000,
origin="derived",
),
)
derive_unit_2.set_data(processing_source_2)
core.bind_derive_unit(dlunit=derive_unit)
core.bind_derive_unit(dlunit=derive_unit_2)
try:
core.start()
with (
keyboard_input(),
Live(
dashboard.render(),
auto_refresh=False,
screen=False,
) 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()
dynalab_logger.removeHandler(log_handler)
dynalab_logger.setLevel(previous_log_level)
dynalab_logger.propagate = previous_propagation
if __name__ == "__main__":
main()