From c1649176321e5bd7a5bcc72ebdfedbb5817815f5 Mon Sep 17 00:00:00 2001 From: MCKero Date: Fri, 28 Aug 2026 10:06:18 +0100 Subject: [PATCH] Log power events, QEMU stderr and firmware serial; show them in a fixed pane Three sources into one buffer: power events from the supervisor, QEMU's stderr (which run.sh and the tests used to discard), and firmware serial, which the machine model tags SERIAL. default_launcher now captures stderr rather than sending it to DEVNULL, which is what made the last two reachable. The pane is a fixed-height scroll box as asked: 180px with overflow-y:auto, so it never grows with content -- older lines move up out of view and you scroll back to read them. Two details that make that usable rather than annoying: - Autoscroll only sticks when you are already at the bottom. Otherwise a new line arriving would yank the view away from whatever you had scrolled up to read. - MAX_LOG_LINES caps the 
 as well. The box is fixed-height either way, but an
  unbounded DOM node would still grow memory across a long session.

Verified on the live server: power on produced power/qemu/serial lines including
"UV-K5 Firmware, EGZUMER-F4HWN+NR7Y c91cec95", each Reset logs the event and the
banner reappearing, and since= never resent a line. A capacity-500 buffer fed 2000
lines keeps exactly 500 and does not replay evicted entries to a stale cursor.
---
 tools/test_uvk5_supervisor.py | 42 ++++++++++++++++++
 tools/test_webui.py           | 48 +++++++++++++++++++++
 tools/uvk5_supervisor.py      | 28 +++++++++---
 tools/webui.py                | 80 ++++++++++++++++++++++++++++++++---
 4 files changed, 187 insertions(+), 11 deletions(-)

diff --git a/tools/test_uvk5_supervisor.py b/tools/test_uvk5_supervisor.py
index 2967d06..0d1fdb9 100644
--- a/tools/test_uvk5_supervisor.py
+++ b/tools/test_uvk5_supervisor.py
@@ -154,5 +154,47 @@ class TestSupervisor(unittest.TestCase):
         self.assertTrue(self.sup.owns_process())
 
 
+class TestSupervisorLogging(unittest.TestCase):
+    def setUp(self):
+        from uvk5_logs import LogBuffer
+        self.log = LogBuffer(capacity=50)
+        self.procs = []
+
+        def launch():
+            proc = FakeProc()
+            self.procs.append(proc)
+            return proc
+
+        self.sup = Supervisor(launch=launch, connect=FakeClient, log=self.log)
+
+    def texts(self):
+        return [e["text"].lower() for e in self.log.entries()]
+
+    def test_power_on_is_logged(self):
+        self.sup.power_on()
+        self.assertTrue(any("power on" in t for t in self.texts()), self.texts())
+
+    def test_power_off_is_logged(self):
+        self.sup.power_on()
+        self.sup.power_off()
+        self.assertTrue(any("power off" in t for t in self.texts()), self.texts())
+
+    def test_reset_is_logged(self):
+        self.sup.power_on()
+        self.sup.reset()
+        self.assertTrue(any("reset" in t for t in self.texts()), self.texts())
+
+    def test_events_are_tagged_as_power(self):
+        self.sup.power_on()
+        self.assertTrue(any(e["source"] == "power"
+                            for e in self.log.entries()))
+
+    def test_works_without_a_log(self):
+        """The log is optional; a supervisor with none must not crash."""
+        sup = Supervisor(launch=lambda: FakeProc(), connect=FakeClient)
+        sup.power_on()
+        sup.power_off()
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/tools/test_webui.py b/tools/test_webui.py
index c3607a8..16fb29c 100644
--- a/tools/test_webui.py
+++ b/tools/test_webui.py
@@ -506,5 +506,53 @@ class TestOptimisticSend(unittest.TestCase):
         self.assertLess(webui.TAP_MS, 400)
 
 
+class TestLogsEndpoint(unittest.TestCase):
+    def setUp(self):
+        self.client, self.http = make_app()
+
+    def test_logs_endpoint_returns_entries(self):
+        resp = self.http.get("/api/logs")
+        self.assertEqual(resp.status_code, 200)
+        body = resp.get_json()
+        self.assertIn("entries", body)
+        self.assertIn("cursor", body)
+
+    def test_logs_accept_a_since_cursor(self):
+        resp = self.http.get("/api/logs?since=0")
+        self.assertEqual(resp.status_code, 200)
+
+    def test_logs_exposes_the_buffer(self):
+        _, sup, http = make_supervised()
+        app_log = http.application.config.get("LOG")
+        self.assertIsNotNone(app_log)
+
+
+class TestLogPane(unittest.TestCase):
+    def setUp(self):
+        _, http = make_app()
+        self.body = http.get("/").get_data(as_text=True)
+
+    def test_page_has_a_log_pane(self):
+        self.assertIn('id="logpane"', self.body)
+        self.assertIn("/api/logs", self.body)
+
+    def test_pane_has_a_fixed_height(self):
+        """The container must not grow with content."""
+        self.assertIn("#logtext", self.body)
+        self.assertIn("height:", self.body)
+
+    def test_pane_scrolls_rather_than_expanding(self):
+        self.assertIn("overflow-y:auto", self.body)
+
+    def test_pane_is_capped_in_line_count(self):
+        """Even a scrollable pane needs a cap, or the DOM grows forever."""
+        self.assertIn("MAX_LOG_LINES", self.body)
+
+    def test_autoscroll_yields_to_manual_scrolling(self):
+        """Scrolling up to read must not be yanked back by the next line."""
+        self.assertIn("scrollHeight", self.body)
+        self.assertIn("clientHeight", self.body)
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/tools/uvk5_supervisor.py b/tools/uvk5_supervisor.py
index 8a16ff9..efd0024 100644
--- a/tools/uvk5_supervisor.py
+++ b/tools/uvk5_supervisor.py
@@ -23,7 +23,7 @@ DEFAULT_QMP = "/tmp/uvk5-qmp.sock"
 
 def default_launcher(qemu: str, flash: str, elf: str,
                      qmp_path: str = DEFAULT_QMP, gdb_port: int = 1234,
-                     capture_stderr: bool = False):
+                     capture_stderr: bool = True):
     """Reproduces the command line in tools/run.sh."""
     def launch():
         # A stale socket makes QEMU fail to bind, which looks like "power on did
@@ -50,13 +50,19 @@ def wait_for_socket(path: str, timeout: float = 15.0) -> bool:
 
 
 class Supervisor:
-    def __init__(self, launch, connect):
+    def __init__(self, launch, connect, log=None):
         self._launch = launch
         self._connect = connect
+        self._log = log
         self._lock = threading.Lock()
         self._proc = None
         self._client = None
 
+    def _note(self, text: str):
+        """Record a power event, if anyone is collecting them."""
+        if self._log is not None:
+            self._log.add("power", text)
+
     def is_running(self) -> bool:
         with self._lock:
             if self._client is None:
@@ -91,7 +97,15 @@ class Supervisor:
                 return False
             self._proc = self._launch()
             self._client = self._connect()
-            return True
+            proc = self._proc
+        self._note("power on")
+        # Forward QEMU's own stderr, which run.sh and the tests used to discard.
+        # Firmware serial arrives here too, tagged SERIAL by the machine model.
+        if self._log is not None and getattr(proc, "stderr", None) is not None:
+            threading.Thread(
+                target=self._log.pump_stream, args=(proc.stderr,),
+                kwargs={"default_source": "qemu"}, daemon=True).start()
+        return True
 
     def power_off(self) -> bool:
         with self._lock:
@@ -108,15 +122,18 @@ class Supervisor:
             client.close()
         except Exception:
             pass
+        code = None
         if proc is not None:
             try:
-                proc.wait(timeout=10)
+                code = proc.wait(timeout=10)
             except Exception:
                 proc.terminate()
                 try:
-                    proc.wait(timeout=5)
+                    code = proc.wait(timeout=5)
                 except Exception:
                     proc.kill()
+        self._note("power off" if code in (None, 0)
+                   else f"power off (qemu exited with {code})")
         return True
 
     def reset(self) -> bool:
@@ -125,6 +142,7 @@ class Supervisor:
         if client is None:
             return self.power_on()
         client.command("system_reset")
+        self._note("reset")
         return True
 
     def pause(self) -> bool:
diff --git a/tools/webui.py b/tools/webui.py
index b6eea5e..52be02c 100644
--- a/tools/webui.py
+++ b/tools/webui.py
@@ -25,6 +25,7 @@ import time
 from flask import Flask, Response, jsonify, request
 
 from uvk5_keys import KEYS, is_valid, normalise
+from uvk5_logs import LogBuffer
 from uvk5_stream import FramePump
 
 KEYPAD_PATH = "/machine/keypad"
@@ -67,6 +68,11 @@ MAX_HOLD_MS = 5000
 LONG_PRESS_AFTER_MS = 400
 LONG_PRESS_MS = 900
 
+# Lines kept in the browser's log pane. The pane is a fixed-height scroll box, so
+# older lines move up out of view; this caps the DOM behind it, which would
+# otherwise grow all session even though only a screenful is visible.
+MAX_LOG_LINES = 500
+
 BOUNDARY = "uvk5frame"
 TARGET_FPS = 15
 
@@ -94,9 +100,13 @@ POWER_ACTIONS = ("on", "off", "reset", "pause", "resume")
 
 
 def create_app(client, frame_addr: int, status_addr: int, scale: int = 4,
-               supervisor=None):
+               supervisor=None, log=None):
     app = Flask(__name__)
 
+    if log is None:
+        log = LogBuffer()
+    app.config["LOG"] = log
+
     # One background grabber for every client. client may be None: the emulator
     # can be powered off, and the page still has to load.
     pump = FramePump(client, frame_addr, status_addr, fps=TARGET_FPS, scale=scale)
@@ -136,6 +146,11 @@ def create_app(client, frame_addr: int, status_addr: int, scale: int = 4,
             return jsonify(powered=False, status="unreachable", error=str(exc))
         return jsonify(powered=True, **info)
 
+    @app.get("/api/logs")
+    def api_logs():
+        since = request.args.get("since", type=int, default=0)
+        return jsonify(entries=log.entries(since=since), cursor=log.cursor())
+
     @app.post("/api/power/")
     def api_power(action):
         action = (action or "").strip().lower()
@@ -312,6 +327,19 @@ def render_index(scale: int) -> str:
   #powerstate.on {{ color:#3fb950; }}
   /* Powered off is a dark panel, not a frozen last frame. */
   .screen-off {{ background:#0b0d10 !important; }}
+  #logpane {{ align-self:stretch; }}
+  #logpane summary {{ font-size:12px; color:#6e7681; cursor:pointer;
+                     user-select:none; }}
+  /*
+   * Fixed height, so the box never grows with content: older lines move up out
+   * of view and you scroll back to read them. min-height matches height so a
+   * nearly empty pane does not jump around as the first lines arrive.
+   */
+  #logtext {{ height:180px; min-height:180px; overflow-y:auto; margin:6px 0 0;
+             padding:8px; background:#0d1117; border:1px solid #2d333b;
+             border-radius:6px; white-space:pre-wrap; word-break:break-all;
+             font:12px/1.45 ui-monospace,SFMono-Regular,Menlo,monospace;
+             color:#8b949e; }}
 
 
 

@@ -328,9 +356,13 @@ def render_index(scale: int) -> str:
     
{grid}

   

   
connecting...

-  
How long you hold a key is measured here and sent as a number,
-  so a slow link cannot turn a tap into a long press. Over 400 ms the firmware
-  treats it as held, which is a different event. Arrows move, Enter is MENU,
+  

+    Logs (firmware serial, qemu, power)
+    

+  

+  
Keys are sent the moment you press, so a slow link does not
+  add the click duration to the delay. Keep holding past 400 ms for a long press,
+  which the firmware treats as a separate event. Arrows move, Enter is MENU,
   Esc is EXIT, digits map straight through. No PTT button -- the keypad model
   has no PTT line.

 
@@ -468,6 +500,39 @@ async function poll() {{
 }}
 poll();
 setInterval(poll, 3000);
+
+// Cap the DOM as well as the server-side buffer. The pane scrolls, but an
+// unbounded 
 would still grow memory over a long session.
+const MAX_LOG_LINES = {MAX_LOG_LINES};
+let logCursor = 0;
+
+async function pollLogs() {{
+  const pre = document.getElementById('logtext');
+  try {{
+    const r = await fetch('/api/logs?since=' + logCursor);
+    const j = await r.json();
+    logCursor = j.cursor;
+    if (!j.entries.length) return;
+
+    // Only stick to the bottom if the user is already there. Otherwise a new
+    // line would yank the view away from whatever they scrolled up to read.
+    const atBottom =
+      pre.scrollHeight - pre.scrollTop - pre.clientHeight < 24;
+
+    for (const e of j.entries) {{
+      pre.textContent += e.time + ' [' + e.source + '] ' + e.text + '\\n';
+    }}
+    const lines = pre.textContent.split('\\n');
+    if (lines.length > MAX_LOG_LINES) {{
+      pre.textContent = lines.slice(-MAX_LOG_LINES).join('\\n');
+    }}
+    if (atBottom) pre.scrollTop = pre.scrollHeight;
+  }} catch (err) {{
+    /* leave the pane as it is; the next poll will catch up */
+  }}
+}}
+pollLogs();
+setInterval(pollLogs, 2000);
 
 """
 
@@ -504,10 +569,13 @@ def main() -> int:
             raise RuntimeError(f"QMP socket never appeared at {args.qmp}")
         return QmpClient(args.qmp)
 
+    # One buffer shared by the supervisor and the HTTP layer, so power events,
+    # QEMU stderr and firmware serial all land in the same place.
+    log = LogBuffer()
     supervisor = Supervisor(
         launch=default_launcher(args.qemu, args.flash, args.elf, args.qmp,
                                 gdb_port=args.gdb_port),
-        connect=connect)
+        connect=connect, log=log)
 
     if args.attach:
         # Someone else owns the process; adopt it so the screen works, but Off
@@ -517,7 +585,7 @@ def main() -> int:
     # page behaves like walking up to a machine rather than finding it booted.
 
     app = create_app(supervisor.client(), args.frame_addr, args.status_addr,
-                     args.scale, supervisor=supervisor)
+                     args.scale, supervisor=supervisor, log=log)
     print(f"serving on http://{args.host}:{args.port}/")
     print("attached to a running emulator" if args.attach
           else "emulator is OFF; press On in the browser to boot it")