Add a background frame pump so QMP load is constant

One thread grabs the LCD at a fixed rate into a shared buffer; HTTP clients serve
from that buffer. Before, every /stream iteration issued its own QMP reads, so
load scaled with client count and reconnects, and a slow reader could stall the
grab loop.

Encodes only when the framebuffer bytes change, and exposes generation so a client
can tell a new frame from the same one without comparing bytes.

rebind() is here for the power work in a later task: the emulator comes and goes
under the server, and rebind(None) blanks the screen instead of leaving a stale
frame that looks live. The run loop re-checks the grabber identity under the lock
after a read, so a rebind landing mid-read cannot be undone by the frame it was
already fetching.

A client that raises is swallowed on purpose -- a dead emulator must not kill the
pump, since power may come back.

Verified against a real emulator: 5000 latest() calls in 1 ms with no QMP traffic,
generation flat at 1 over a second of static screen, guest still running, and
rebind(None) going dark. 9 unit tests.

tools/test_uvk5_stream.py

@@ -0,0 +1,134 @@
+#!/usr/bin/env python3
+"""Unit tests for the background frame pump. No emulator needed."""
+import tempfile
+import threading
+import time
+import unittest
+
+from uvk5_stream import FramePump
+
+
+class CountingClient:
+    """Counts memsave calls, so QMP load can be asserted."""
+
+    def __init__(self):
+        self.reads = 0
+        self.lock = threading.Lock()
+
+    def command(self, name, **args):
+        if name == "pmemsave":
+            raise AssertionError("pmemsave reads physical addresses; use memsave")
+        if name != "memsave":
+            raise AssertionError(f"unexpected {name}")
+        with self.lock:
+            self.reads += 1
+        with open(args["filename"], "wb") as fh:
+            fh.write(bytes(args["size"]))
+        return {}
+
+
+def wait_for_frame(pump, timeout=3.0):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if pump.latest() is not None:
+            return True
+        time.sleep(0.01)
+    return False
+
+
+class TestFramePump(unittest.TestCase):
+    def _pump(self, client=None, fps=50):
+        pump = FramePump(client if client is not None else CountingClient(),
+                         0x1000, 0x2000, fps=fps, spool_dir=tempfile.mkdtemp())
+        self.addCleanup(pump.stop)
+        return pump
+
+    def test_latest_returns_a_png_once_started(self):
+        pump = self._pump()
+        pump.start()
+        self.assertTrue(wait_for_frame(pump), "no frame produced")
+        self.assertTrue(pump.latest().startswith(b"\x89PNG\r\n\x1a\n"))
+
+    def test_latest_is_none_before_start(self):
+        pump = self._pump()
+        self.assertIsNone(pump.latest())
+
+    def test_read_rate_is_independent_of_reader_count(self):
+        """QMP load must not scale with clients: that is the whole point."""
+        client = CountingClient()
+        pump = self._pump(client, fps=20)
+        pump.start()
+        self.assertTrue(wait_for_frame(pump))
+
+        time.sleep(0.5)
+        with client.lock:
+            first = client.reads
+        # Hammer latest() the way many readers would. It must cause no reads.
+        for _ in range(2000):
+            pump.latest()
+        with client.lock:
+            after_hammer = client.reads
+        self.assertLess(after_hammer - first, 30,
+                        "serving frames must not trigger fresh QMP reads")
+
+    def test_stop_halts_reading(self):
+        client = CountingClient()
+        pump = self._pump(client)
+        pump.start()
+        self.assertTrue(wait_for_frame(pump))
+        pump.stop()
+        with client.lock:
+            settled = client.reads
+        time.sleep(0.3)
+        with client.lock:
+            self.assertEqual(client.reads, settled)
+
+    def test_generation_does_not_advance_on_an_unchanged_frame(self):
+        """Encoding only on change is what keeps idle CPU near zero."""
+        pump = self._pump()
+        pump.start()
+        self.assertTrue(wait_for_frame(pump))
+        time.sleep(0.2)
+        gen = pump.generation()
+        time.sleep(0.3)
+        # The stub always returns the same zero frame, so nothing changed.
+        self.assertEqual(pump.generation(), gen)
+
+    def test_start_is_idempotent(self):
+        pump = self._pump()
+        pump.start()
+        pump.start()
+        self.assertTrue(wait_for_frame(pump))
+
+    def test_survives_a_client_that_raises(self):
+        """A dead emulator must not kill the pump; power may come back."""
+        class Broken:
+            def command(self, name, **args):
+                raise RuntimeError("emulator gone")
+
+        pump = self._pump(Broken())
+        pump.start()
+        time.sleep(0.2)
+        self.assertIsNone(pump.latest())      # nothing to show, but still alive
+        pump.stop()
+
+    def test_rebind_to_none_blanks_the_screen(self):
+        """Power off must go dark, not keep showing a stale frame."""
+        pump = self._pump()
+        pump.start()
+        self.assertTrue(wait_for_frame(pump))
+        pump.rebind(None)
+        self.assertIsNone(pump.latest())
+
+    def test_rebind_to_a_client_resumes(self):
+        pump = self._pump()
+        pump.start()
+        self.assertTrue(wait_for_frame(pump))
+        pump.rebind(None)
+        self.assertIsNone(pump.latest())
+        pump.rebind(CountingClient())
+        self.assertTrue(wait_for_frame(pump), "did not resume after rebind")
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/uvk5_stream.py

@@ -0,0 +1,95 @@
+#!/usr/bin/env python3
+"""Background frame pump for the web UI.
+
+One thread reads the LCD at a fixed rate into a shared buffer, and every HTTP
+client serves from that buffer. Previously each /stream iteration did its own QMP
+reads, so load scaled with the number of clients and a slow reader could stall the
+grab loop.
+
+Encoding happens only when the framebuffer bytes actually change -- the LCD is
+static most of the time, so idle CPU stays near zero. `generation` lets a client
+tell "no new frame" from "same frame again" without comparing bytes itself.
+
+`rebind` exists for power cycling: the emulator can come and go under the server,
+and rebinding to None blanks the screen rather than leaving a stale frame that
+looks live.
+"""
+import threading
+import time
+
+from uvk5_lcd import FrameGrabber, encode_png, unpack
+
+
+class FramePump:
+    def __init__(self, client, frame_addr: int, status_addr: int,
+                 fps: int = 15, scale: int = 4, spool_dir: str = "/dev/shm"):
+        self._frame_addr = frame_addr
+        self._status_addr = status_addr
+        self._spool_dir = spool_dir
+        self._interval = 1.0 / fps
+        self._scale = scale
+        self._lock = threading.Lock()
+        self._grabber = (FrameGrabber(client, frame_addr, status_addr, spool_dir)
+                         if client is not None else None)
+        self._png = None
+        self._raw = None
+        self._generation = 0
+        self._stop = threading.Event()
+        self._thread = None
+
+    def start(self):
+        if self._thread is not None:
+            return
+        self._thread = threading.Thread(target=self._run, daemon=True)
+        self._thread.start()
+
+    def stop(self):
+        self._stop.set()
+        thread, self._thread = self._thread, None
+        if thread is not None:
+            thread.join(timeout=2)
+
+    def rebind(self, client):
+        """Point at a new QMP client, or None when the emulator is off."""
+        with self._lock:
+            self._grabber = (
+                FrameGrabber(client, self._frame_addr, self._status_addr,
+                             self._spool_dir)
+                if client is not None else None)
+            if client is None:
+                self._png = None        # dark screen, not a stale frame
+                self._raw = None
+            self._generation += 1
+
+    def _run(self):
+        while not self._stop.is_set():
+            started = time.monotonic()
+            with self._lock:
+                grabber = self._grabber
+            if grabber is not None:
+                try:
+                    status, frame = grabber.raw()
+                    current = (status, frame)
+                    with self._lock:
+                        # Re-check: a rebind may have landed mid-read, and its
+                        # blanking must not be undone by this stale frame.
+                        if self._grabber is grabber and current != self._raw:
+                            self._raw = current
+                            self._png = encode_png(unpack(status, frame),
+                                                   self._scale)
+                            self._generation += 1
+                except Exception:
+                    # A dead emulator must not kill the pump: power may come
+                    # back, and latest() keeps serving the last good frame.
+                    pass
+            slack = self._interval - (time.monotonic() - started)
+            if slack > 0:
+                self._stop.wait(slack)
+
+    def latest(self):
+        with self._lock:
+            return self._png
+
+    def generation(self) -> int:
+        with self._lock:
+            return self._generation