mppt-testbench/code64/debug_console/app.py

"""Textual TUI application for LVSolarBuck debug console."""
import time
from collections import deque
from textual.app import App, ComposeResult
from textual.containers import Horizontal, Vertical, VerticalScroll
from textual.widgets import Header, Footer, RichLog

from .protocol import (
    CMD_TELEMETRY, CMD_PARAM_WRITE, CMD_PARAM_WRITE_ACK, CMD_PARAM_VALUE, CMD_PONG, CMD_ERROR_MSG,
    decode_telemetry, decode_param_value, build_ping, build_shutdown, build_reset, build_test_50, build_param_read_all,
    build_relay_on, build_relay_off, build_hold_converter, build_toggle_precharge,
    PARAM_BY_ID,
)
from .serial_worker import SerialWorker
from .data_logger import DataLogger
from .widgets.telemetry_panel import TelemetryPanel
from .widgets.param_group import ParamGroup
from .widgets.status_bar import StatusBar


class DebugConsoleApp(App):
    """LVSolarBuck Debug Console"""

    TITLE = "LVSolarBuck Debug Console"

    CSS = """
    Screen {
        layout: vertical;
    }
    #main {
        layout: horizontal;
        height: 1fr;
    }
    #left-panel {
        width: 35;
        min-width: 30;
        border-right: solid $primary;
    }
    #right-panel {
        width: 1fr;
        min-width: 40;
    }
    #error-log {
        height: 8;
        min-height: 4;
        border-top: solid $error;
        background: $surface;
    }
    """

    BINDINGS = [
        ("q", "quit", "Quit"),
        ("p", "ping", "Ping"),
        ("f", "toggle_filter", "Filter On/Off"),
        ("s", "shutdown", "SHUTOFF"),
        ("x", "reset", "RESET"),
        ("t", "test50", "TEST 50%"),
        ("c", "relay_on", "Relay ON"),
        ("d", "relay_off", "Relay OFF"),
        ("h", "hold_converter", "HOLD CONV"),
        ("g", "toggle_precharge", "PRECHARGE"),
    ]

    def __init__(self, port: str, baudrate: int = 460800):
        super().__init__()
        self.serial_port = port
        self.serial_baudrate = baudrate
        self.worker: SerialWorker | None = None
        self.logger: DataLogger | None = None
        # Per-tick accumulators (bumped in _on_frame); _sent_count counts the
        # frames the MCU generated (incl. lost) from the telemetry seq advance.
        self._telem_count = 0
        self._sent_count = 0
        # Sliding-window link-rate estimator: a deque of (time, recv, sent)
        # buckets covering the last _rate_window_s, so the Hz/loss readout is a
        # moving average refreshed every status tick — smooth (no ±1-frame
        # 8/12 Hz quantization) yet still responsive.
        self._rate_window_s = 3.0
        self._rate_buckets: deque = deque()
        self._rate_window_start = time.monotonic()
        self._last_seq = -1
        # Un-ACKed param writes awaiting retry: the MCU's RX side is EMI-lossy
        # while the converter is switching (TX/telemetry unaffected), so writes
        # are resent until the matching CMD_PARAM_WRITE_ACK arrives.
        # param_id -> [frame_bytes, tries, deadline]
        self._pending_writes: dict[int, list] = {}

    def compose(self) -> ComposeResult:
        yield Header()
        with Horizontal(id="main"):
            with Vertical(id="left-panel"):
                yield TelemetryPanel()
            with VerticalScroll(id="right-panel"):
                yield ParamGroup("Global", self._send_data)
                yield ParamGroup("Compensator", self._send_data)
                yield ParamGroup("Vfly", self._send_data)
                yield ParamGroup("CC", self._send_data)
                yield ParamGroup("MPPT", self._send_data)
                yield ParamGroup("Deadtime", self._send_data)
                yield ParamGroup("Manual", self._send_data)
        yield RichLog(id="error-log", markup=True, wrap=True)
        yield StatusBar()
        yield Footer()

    def on_mount(self) -> None:
        self.sub_title = f"{self.serial_port} {self.serial_baudrate}"
        log = self.query_one("#error-log", RichLog)
        log.write("[dim]Error log ready[/dim]")
        self.logger = DataLogger(log_dir="logs")
        self.logger.log_session_info(self.serial_port, self.serial_baudrate)
        log.write(f"[dim]Logging to {self.logger.db_path}[/dim]")
        self.worker = SerialWorker(self.serial_port, self.serial_baudrate)
        self.worker.set_frame_callback(self._on_frame)
        self.worker.set_status_callback(self._on_status)
        self.worker.start()
        self.set_interval(0.5, self._update_status)
        self.set_interval(2.0, self._request_params)
        self.set_interval(0.3, self._flush_pending_writes)

    def on_unmount(self) -> None:
        if self.worker:
            self.worker.stop()
        if self.logger:
            self.logger.close()

    def _send_data(self, data: bytes) -> None:
        if self.worker:
            # Track param writes for ACK-based auto-retry. Frame layout:
            # [SYNC, cmd, len, payload...]; ParamWritePayload starts with
            # param_id. A newer write to the same param replaces the pending
            # entry, so retries never resurrect a stale value.
            if len(data) > 3 and data[1] == CMD_PARAM_WRITE:
                self._pending_writes[data[3]] = [data, 0, time.monotonic() + 0.3]
            self.worker.send(data)

    def _flush_pending_writes(self) -> None:
        if not self.worker or not self._pending_writes:
            return
        now = time.monotonic()
        for pid in list(self._pending_writes):
            entry = self._pending_writes.get(pid)
            if entry is None or now < entry[2]:
                continue
            if entry[1] >= 30:
                self._pending_writes.pop(pid, None)
                p = PARAM_BY_ID.get(pid)
                name = p.name if p else f"0x{pid:02X}"
                self.notify(f"Param write '{name}' got no ACK after 30 tries",
                            severity="error", timeout=10)
                continue
            entry[1] += 1
            entry[2] = now + 0.3
            self.worker.send(entry[0])

    def _request_params(self) -> None:
        if self.worker and self.worker.connected:
            self._send_data(build_param_read_all())

    def _on_frame(self, cmd: int, payload: bytes) -> None:
        if cmd == CMD_TELEMETRY:
            t = decode_telemetry(payload)
            if t is not None:
                self._telem_count += 1
                if self._last_seq >= 0:
                    # seq increments per frame the MCU SENT, so the advance
                    # counts generated frames whether or not they arrived.
                    self._sent_count += (t.seq - self._last_seq) & 0xFF
                    expected = (self._last_seq + 1) & 0xFF
                    if t.seq != expected:
                        diff = (t.seq - expected) & 0xFF
                        if self.worker:
                            self.worker.drop_count += diff
                else:
                    self._sent_count += 1
                self._last_seq = t.seq
                if self.logger:
                    self.logger.log_telemetry(t)
                self.call_from_thread(self._update_ui, t)
        elif cmd == CMD_PARAM_VALUE:
            result = decode_param_value(payload)
            if result:
                param_id, value = result
                if self.logger:
                    self.logger.log_param(param_id, value)
                self.call_from_thread(self._update_param, param_id, value)
        elif cmd == CMD_PARAM_WRITE_ACK:
            result = decode_param_value(payload)
            if result:
                param_id, value = result
                self._pending_writes.pop(param_id, None)
                if self.logger:
                    self.logger.log_param(param_id, value)
                self.call_from_thread(self._update_param, param_id, value)
            self.call_from_thread(self.notify, "Parameter ACK", severity="information")
        elif cmd == CMD_ERROR_MSG:
            msg = payload.decode("ascii", errors="replace")
            self.call_from_thread(self._log_error, msg)
        elif cmd == CMD_PONG:
            self.call_from_thread(self.notify, "PONG received")

    def _on_status(self, connected: bool) -> None:
        self.call_from_thread(self._update_connected, connected)

    def _log_error(self, msg: str) -> None:
        log = self.query_one("#error-log", RichLog)
        ts = time.strftime("%H:%M:%S")
        is_guard = msg.startswith("GUARD") or msg.startswith("DIAG") or msg.startswith("RAW") or msg.startswith("CTRL") or msg.startswith("HW ") or msg.startswith("HRTIM") or msg.startswith("FLAGS") or msg == "Startup"
        if is_guard:
            log.write(f"[yellow][{ts}] {msg}[/yellow]")
        else:
            log.write(f"[bold red][{ts}] {msg}[/bold red]")
            self.notify(f"STM32: {msg}", severity="error", timeout=10)

    def _update_ui(self, t) -> None:
        self.query_one(TelemetryPanel).update_telemetry(t)
        status = self.query_one(StatusBar)
        if self.worker:
            status.pkt_count = self.worker.pkt_count
            status.drop_count = self.worker.drop_count
        status.last_seq = t.seq
        status.refresh_status()

    def _update_param(self, param_id: int, value: float) -> None:
        for group in self.query(ParamGroup):
            group.update_param(param_id, value)
        self.query_one(TelemetryPanel).update_dt_param(param_id, value)

    def _update_connected(self, connected: bool) -> None:
        status = self.query_one(StatusBar)
        status.connected = connected
        status.refresh_status()
        if connected:
            self._request_params()

    def _update_status(self) -> None:
        now = time.monotonic()
        # Push this tick's counts, then slide the window forward, dropping
        # buckets older than _rate_window_s. The retained buckets exactly cover
        # (_rate_window_start, now], so summing their counts and dividing by
        # that span is an unbiased moving-average rate — no short-window
        # quantization, and it tracks real changes within the window.
        self._rate_buckets.append((now, self._telem_count, self._sent_count))
        self._telem_count = 0
        self._sent_count = 0
        while len(self._rate_buckets) > 1 and now - self._rate_buckets[0][0] > self._rate_window_s:
            self._rate_window_start = self._rate_buckets.popleft()[0]
        span = now - self._rate_window_start
        recv = sum(b[1] for b in self._rate_buckets)
        sent = sum(b[2] for b in self._rate_buckets)
        fps = recv / span if span > 0 else 0.0
        sent_fps = sent / span if span > 0 else 0.0
        loss_pct = 100.0 * (1.0 - recv / sent) if sent > 0 else 0.0
        status = self.query_one(StatusBar)
        status.fps = fps
        status.sent_fps = sent_fps
        status.loss_pct = loss_pct
        if self.worker:
            status.connected = self.worker.connected
        status.refresh_status()
        # The StatusBar is hidden behind the Footer on most terminals, so the
        # link stats are mirrored into the always-visible telemetry panel.
        panel = self.query_one(TelemetryPanel)
        panel.link_rx = fps
        panel.link_tx = sent_fps
        panel.link_loss = loss_pct

    def action_ping(self) -> None:
        self._send_data(build_ping())

    def action_shutdown(self) -> None:
        self._send_data(build_shutdown())

    def action_reset(self) -> None:
        self._send_data(build_reset())

    def action_test50(self) -> None:
        self._send_data(build_test_50())

    def action_relay_on(self) -> None:
        self._send_data(build_relay_on())

    def action_relay_off(self) -> None:
        self._send_data(build_relay_off())

    def action_hold_converter(self) -> None:
        self._send_data(build_hold_converter())

    def action_toggle_precharge(self) -> None:
        self._send_data(build_toggle_precharge())

    def action_toggle_filter(self) -> None:
        panel = self.query_one(TelemetryPanel)
        panel.filter_enabled = not panel.filter_enabled
        state = "ON" if panel.filter_enabled else "OFF"
        self.notify(f"Filter {state}")