janikandClaude Opus 4.8
d2dfc73f9e
The mppt-testbench Python tooling had drifted from the flashed firmware
(bundled fw e7a23a3 vs live 1b85532) and could no longer communicate:
- stm32_link.py: CRC8 -> CRC-16/CCITT-FALSE; telemetry 68B -> 78B
(btemp, cmp_outer/inner, iout_slow, vfly_ofs_applied); add PTYPE_INT16
and commands 0x12-0x18; replace PARAMS with the current 37-param map
(single dt_normal, no dt brackets; test_corr/phase_ofs, phase PI,
precharge PI, duty dither; vfly_active 0-3).
- tuner.py: retire per-bracket deadtime; sweep the single dt_normal.
- cli.py: update tune-deadtime, help/examples, btemp readout;
default ports COM11 (load) / COM4 (stm32).
- debug console TUI: sync protocol.py/app.py/status_bar/telemetry_panel
from live (new command keys, link RX/TX/loss stats, single dead-time,
new telemetry fields, param-write auto-retry); add duty_fft.py.
- README: rewrite parameter table, deadtime section, ports, keybindings.
Verified: protocol round-trip self-tests + live `bench stm32-read`
reading all 37 params over COM4.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
296 lines
12 KiB
Python
296 lines
12 KiB
Python
"""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}")
|