h3cx
563a35ee2c
CoreConfig now have a boolean thread debug option, when this is enabled, it prints out the CPU usage of each thread to help identify which threads are the most CPU hungry for later optimisation
341 lines
11 KiB
Python
341 lines
11 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.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"
|
|
)
|
|
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("", "", "")
|
|
|
|
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"),
|
|
Panel(
|
|
log_table,
|
|
title=f"Logs · latest {VISIBLE_LOG_LINES}",
|
|
border_style="magenta",
|
|
),
|
|
)
|
|
|
|
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)
|
|
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, thread_debug=False)
|
|
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()
|