dynalab-core/test/manual/core_tui.py

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 == " ":
                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 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]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")
    )
    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)
    core.bind_derive_unit(dlunit=derive_unit, file=None)

    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()