Implement serial receive, unlocking the UV-K5 programming protocol

Nothing could be sent to the firmware. Three separate pieces were missing.

USART1 was a register stub with no chardev, so there was no source of incoming
bytes. It now takes a chardev property, defaulting to serial0, so -serial works.

The DMA model never serviced USART. App/driver/uart.c receives over a circular
peripheral-to-memory channel and never reads DR; it finds new data with

    write_ptr = sizeof(UART_DMA_Buffer) - LL_DMA_GetDataLength(DMA1, CHANNEL_2)

so leaving CNDTR at its programmed value made the buffer look permanently empty
however many bytes arrived. DMA now drains USART1's queue byte by byte, decrementing
CNDTR and reloading it in circular mode. Channels also remember the length they were
given, since CNDTR counts down and the offset into the buffer has to be derived from
the difference.

Servicing happens on a CNDTR read rather than from a timer: that read is precisely
how the driver looks for data, so no polling is needed and no byte can be delivered
before the guest asks.

And transmit was invisible to the far end. DR writes went to stderr only, so a host
tool would send a command, the firmware would answer, and the answer went nowhere the
tool could see -- indistinguishable from being ignored. This cost a debugging round:
the first test run reported "no reply at all" with 0 bytes of boot output, which
looked like receive failing when the boot banner was in fact being written to stderr
as always. DR now also writes the raw byte to the chardev when one is connected.

SR reports RXNE when bytes are queued and DR consumes one, so a polling firmware
would work too, even though this one uses DMA.

tools/test_serial_rx.py speaks the real wire protocol -- AB CD framing, the fixed XOR
obfuscation, CRC-16/XMODEM -- and checks two exchanges end to end:

    0x0514 hello        -> 0x0515 ack
    0x051B EEPROM read  -> 0x051C with the requested 8 bytes at 0x0E70

Both pass. CPS/CHIRP-style tools can now talk to the emulator. keypad_test.py,
test_flash_persist.py, test_freq_entry.py and the 143 unit tests still pass.

qemu/py32f071.c

@@ -28,9 +28,12 @@
 #include "hw/arm/armv7m.h"
 #include "hw/boards.h"
 #include "hw/qdev-properties.h"
+/* Serial receive: USART1 takes a chardev so a host tool can drive the firmware. */
+#include "hw/qdev-properties-system.h"
+#include "chardev/char-fe.h"
+#include "sysemu/sysemu.h"
 #include "hw/sysbus.h"
 #include "exec/address-spaces.h"
-#include "sysemu/sysemu.h"
 #include "qom/object.h"
 
 /* ---------------------------------------------------------------- memory map */
@@ -820,8 +823,19 @@ OBJECT_DECLARE_SIMPLE_TYPE(PY32DmaState, PY32_DMA)
 
 typedef struct {
     uint32_t ccr, cndtr, cpar, cmar;
+    /*
+     * The length the guest programmed, kept separately because cndtr counts down.
+     * Needed to derive how far into the buffer a transfer has got, and to reload
+     * the count in circular mode.
+     */
+    uint32_t total;
 } PY32DmaChannel;
 
+/* Defined further down; DMA drains its receive queue. */
+typedef struct PY32StubState PY32StubState;
+static bool py32_stub_rx_empty(PY32StubState *s);
+static bool py32_stub_rx_pop(PY32StubState *s, uint8_t *out);
+
 struct PY32DmaState {
     SysBusDevice parent_obj;
     MemoryRegion iomem;
@@ -836,6 +850,12 @@ struct PY32DmaState {
     /* Set by the SoC: lets the DMA clock bytes through an SPI controller. */
     PY32SpiState *spi[2];
 
+    /*
+     * Set by the SoC. USART1's receive queue is drained from here because the
+     * firmware's UART driver never reads DR -- it watches the DMA count instead.
+     */
+    PY32StubState *usart1;
+
     /*
      * The address space DMA transfers move bytes through.
      *
@@ -975,6 +995,73 @@ static void py32_dma_kick(void *dma, PY32SpiState *spi)
     py32_dma_run_for_spi((PY32DmaState *)dma, spi);
 }
 
+/*
+ * Move queued USART bytes into the guest buffer, one at a time, decrementing the
+ * channel's remaining count.
+ *
+ * The count is the whole point. App/driver/uart.c configures a circular
+ * peripheral-to-memory channel and never reads DR; it locates new data with
+ *
+ *     write_ptr = sizeof(UART_DMA_Buffer) - LL_DMA_GetDataLength(...)
+ *
+ * so a model that leaves CNDTR at its initial value reports an empty buffer
+ * forever, no matter how many bytes arrived. Serial receive was dead for exactly
+ * that reason, and with it the whole UV-K5 programming protocol.
+ *
+ * Circular mode reloads the count and wraps the address on completion rather than
+ * stopping, which is what makes the firmware's pointer arithmetic work across the
+ * end of the buffer.
+ */
+static void py32_dma_service_usart_rx(PY32DmaState *s)
+{
+    AddressSpace *as = s->as;
+
+    if (!as || !s->usart1) {
+        return;
+    }
+
+    for (int ch = 0; ch < PY32_DMA_CHANNELS; ch++) {
+        PY32DmaChannel *c = &s->ch[ch];
+
+        if (!(c->ccr & DMA_CCR_EN) || (c->ccr & DMA_CCR_DIR)) {
+            continue;                 /* disabled, or memory-to-peripheral */
+        }
+        if ((c->cpar & ~0x3ffu) != PY32_USART1_BASE) {
+            continue;
+        }
+        if (c->total == 0) {
+            continue;                 /* never configured with a length */
+        }
+
+        while (!py32_stub_rx_empty(s->usart1)) {
+            uint8_t byte;
+            if (!py32_stub_rx_pop(s->usart1, &byte)) {
+                break;
+            }
+
+            const uint32_t done = c->total - c->cndtr;
+            const uint32_t dest = c->cmar + ((c->ccr & DMA_CCR_MINC) ? done : 0);
+            address_space_write(as, dest, MEMTXATTRS_UNSPECIFIED, &byte, 1);
+
+            if (c->cndtr > 0) {
+                c->cndtr--;
+            }
+
+            if (c->cndtr == 0) {
+                if (c->ccr & DMA_CCR_CIRC) {
+                    c->cndtr = c->total;      /* wrap, keep running */
+                } else {
+                    c->ccr &= ~DMA_CCR_EN;
+                    break;
+                }
+            }
+        }
+
+        s->isr |= DMA_FLAG_GIF(ch);
+        py32_dma_update_irq(s);
+    }
+}
+
 static uint64_t py32_dma_read(void *opaque, hwaddr addr, unsigned size)
 {
     PY32DmaState *s = opaque;
@@ -991,7 +1078,16 @@ static uint64_t py32_dma_read(void *opaque, hwaddr addr, unsigned size)
         if (ch < PY32_DMA_CHANNELS) {
             switch (reg) {
             case DMA_CCR:   return s->ch[ch].ccr;
-            case DMA_CNDTR: return s->ch[ch].cndtr;
+            case DMA_CNDTR:
+                /*
+                 * Deliver any pending serial bytes before answering. This read is
+                 * precisely how App/driver/uart.c discovers new data -- it computes
+                 * a write pointer from the remaining count -- so servicing here
+                 * needs no timer and cannot deliver bytes the guest has not asked
+                 * about yet.
+                 */
+                py32_dma_service_usart_rx(s);
+                return s->ch[ch].cndtr;
             case DMA_CPAR:  return s->ch[ch].cpar;
             case DMA_CMAR:  return s->ch[ch].cmar;
             default: break;
@@ -1021,7 +1117,10 @@ static void py32_dma_write(void *opaque, hwaddr addr, uint64_t value, unsigned s
     }
 
     switch (reg) {
-    case DMA_CNDTR: s->ch[ch].cndtr = value; break;
+    case DMA_CNDTR:
+        s->ch[ch].cndtr = value;
+        s->ch[ch].total = value;      /* remember it; cndtr counts down */
+        break;
     case DMA_CPAR:  s->ch[ch].cpar = value;  break;
     case DMA_CMAR:  s->ch[ch].cmar = value;  break;
     case DMA_CCR: {
@@ -1479,12 +1578,62 @@ struct PY32StubState {
     char        *stub_name;
     uint32_t     size;
     uint32_t     regs[0x100];
+
+    /*
+     * Receive path, USART1 only.
+     *
+     * A chardev supplies bytes; DR hands them to the guest. The DMA model drains
+     * this queue on behalf of the circular receive channel, because
+     * App/driver/uart.c never reads DR directly -- it derives a write pointer from
+     * the channel's remaining count.
+     */
+    CharBackend  chr;
+    uint8_t      rx_fifo[256];
+    unsigned     rx_head, rx_tail;
 };
 
-/* USART_SR transmit flags, from the vendor header: TXE is bit 7, TC bit 6. */
+/* USART_SR flags, from the vendor header. */
+#define PY32_USART_SR_RXNE (1u << 5)
 #define PY32_USART_SR_TC   (1u << 6)
 #define PY32_USART_SR_TXE  (1u << 7)
 
+static bool py32_stub_rx_empty(PY32StubState *s)
+{
+    return s->rx_head == s->rx_tail;
+}
+
+/* Pull one received byte, or return false when nothing is queued. */
+static bool py32_stub_rx_pop(PY32StubState *s, uint8_t *out)
+{
+    if (py32_stub_rx_empty(s)) {
+        return false;
+    }
+    *out = s->rx_fifo[s->rx_tail];
+    s->rx_tail = (s->rx_tail + 1) % sizeof(s->rx_fifo);
+    return true;
+}
+
+static int py32_stub_can_receive(void *opaque)
+{
+    PY32StubState *s = opaque;
+    const unsigned used = (s->rx_head - s->rx_tail) % sizeof(s->rx_fifo);
+    return sizeof(s->rx_fifo) - 1 - used;
+}
+
+static void py32_stub_receive(void *opaque, const uint8_t *buf, int size)
+{
+    PY32StubState *s = opaque;
+
+    for (int i = 0; i < size; i++) {
+        const unsigned next = (s->rx_head + 1) % sizeof(s->rx_fifo);
+        if (next == s->rx_tail) {
+            break;                    /* full; drop rather than overwrite */
+        }
+        s->rx_fifo[s->rx_head] = buf[i];
+        s->rx_head = next;
+    }
+}
+
 static uint64_t py32_stub_read(void *opaque, hwaddr addr, unsigned size)
 {
     PY32StubState *s = opaque;
@@ -1504,6 +1653,19 @@ static uint64_t py32_stub_read(void *opaque, hwaddr addr, unsigned size)
      */
     if (addr == 0x00 && s->stub_name && !strcmp(s->stub_name, "usart1")) {
         value |= PY32_USART_SR_TXE | PY32_USART_SR_TC;
+        /* RXNE so a firmware that polls instead of using DMA also works. */
+        if (!py32_stub_rx_empty(s)) {
+            value |= PY32_USART_SR_RXNE;
+        }
+    }
+
+    /* Reading DR consumes a received byte, as on hardware. */
+    if (addr == 0x04 && s->stub_name && !strcmp(s->stub_name, "usart1")) {
+        uint8_t byte;
+        if (py32_stub_rx_pop(s, &byte)) {
+            return byte;
+        }
+        return 0;
     }
 
     qemu_log_mask(LOG_UNIMP, "py32-%s: read 0x%03" HWADDR_PRIx " -> 0x%08x\n",
@@ -1559,7 +1721,20 @@ static void py32_stub_write(void *opaque, hwaddr addr, uint64_t value, unsigned
         s->regs[idx] = value;
     }
     if (addr == 0x04 && s->stub_name && !strcmp(s->stub_name, "usart1")) {
-        py32_stub_serial_byte((char)(value & 0xff));
+        const uint8_t byte = value & 0xff;
+
+        /* Human-readable copy on stderr, which is what the web UI log reads. */
+        py32_stub_serial_byte((char)byte);
+
+        /*
+         * And the raw byte to the chardev, if one is attached. Without this the
+         * transmit side is invisible to anything on the other end of the port: a
+         * host tool sends a command, the firmware answers, and the answer only ever
+         * reaches stderr -- which looks exactly like the firmware ignoring it.
+         */
+        if (qemu_chr_fe_backend_connected(&s->chr)) {
+            qemu_chr_fe_write_all(&s->chr, &byte, 1);
+        }
     }
     qemu_log_mask(LOG_UNIMP, "py32-%s: write 0x%03" HWADDR_PRIx " = 0x%08" PRIx64 "\n",
                   s->stub_name ?: "stub", addr, value);
@@ -1581,11 +1756,22 @@ static void py32_stub_realize(DeviceState *dev, Error **errp)
                           s->stub_name ?: TYPE_PY32_STUB,
                           s->size ? s->size : 0x400);
     sysbus_init_mmio(SYS_BUS_DEVICE(dev), &s->iomem);
+
+    /*
+     * Only USART1 takes a chardev: it is the firmware's console and the port the
+     * UV-K5 programming protocol speaks over. Harmless when unset -- without a
+     * backend the receive queue simply stays empty, which is the old behaviour.
+     */
+    if (s->stub_name && !strcmp(s->stub_name, "usart1")) {
+        qemu_chr_fe_set_handlers(&s->chr, py32_stub_can_receive,
+                                 py32_stub_receive, NULL, NULL, s, NULL, true);
+    }
 }
 
 static Property py32_stub_properties[] = {
     DEFINE_PROP_STRING("stub-name", PY32StubState, stub_name),
     DEFINE_PROP_UINT32("size", PY32StubState, size, 0x400),
+    DEFINE_PROP_CHR("chardev", PY32StubState, chr),
     DEFINE_PROP_END_OF_LIST(),
 };
 
@@ -1797,11 +1983,30 @@ static void py32f071_soc_realize(DeviceState *dev_soc, Error **errp)
     for (int i = 0; i < PY32_NUM_STUB; i++) {
         qdev_prop_set_string(DEVICE(&s->stub[i]), "stub-name", py32_stubs[i].name);
         qdev_prop_set_uint32(DEVICE(&s->stub[i]), "size", py32_stubs[i].size);
+
+        /*
+         * Give USART1 a chardev so something can talk *to* the firmware. This is
+         * the port the UV-K5 programming protocol runs over (App/app/uart.c:
+         * 0x0514 handshake, 0x051B/0x051D EEPROM read and write, 0x05DD reset).
+         * Defaults to "serial0", so -serial on the command line just works.
+         */
+        if (!strcmp(py32_stubs[i].name, "usart1")) {
+            Chardev *chr = serial_hd(0);
+            if (chr) {
+                qdev_prop_set_chr(DEVICE(&s->stub[i]), "chardev", chr);
+            }
+        }
+
         if (!sysbus_realize(SYS_BUS_DEVICE(&s->stub[i]), errp)) {
             return;
         }
         memory_region_add_subregion(&s->container, py32_stubs[i].base,
                                     sysbus_mmio_get_region(SYS_BUS_DEVICE(&s->stub[i]), 0));
+
+        /* DMA drains USART1's receive queue; see py32_dma_service_usart_rx. */
+        if (!strcmp(py32_stubs[i].name, "usart1")) {
+            s->dma.usart1 = &s->stub[i];
+        }
     }
 
     /*

tools/test_serial_rx.py

@@ -0,0 +1,220 @@
+#!/usr/bin/env python3
+"""The firmware must receive serial bytes and answer a programming command.
+
+Serial receive used to be impossible. USART1 was a register stub with no chardev, so
+nothing could be sent in; and App/driver/uart.c locates incoming data with
+
+    write_ptr = sizeof(UART_DMA_Buffer) - LL_DMA_GetDataLength(DMA1, CHANNEL_2)
+
+over a circular DMA channel, while the DMA model only ever serviced SPI and never
+decremented CNDTR for USART. That expression was therefore always 0 and the buffer
+always looked empty, which made the whole UV-K5 programming protocol in App/app/uart.c
+unreachable: 0x0514 handshake, 0x051B/0x051D EEPROM read and write, 0x05DD reset.
+
+This sends a real 0x0514 handshake and checks for a reply.
+
+Wire format, from App/app/uart.c:
+    AB CD <len:16 LE> <payload> <crc:16 LE> DC BA
+The payload is obfuscated with a fixed XOR key, and the CRC is CRC-16/XMODEM over the
+payload. Replies use the same framing.
+"""
+import gzip
+import os
+import shutil
+import socket
+import struct
+import subprocess
+import sys
+import tempfile
+import time
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+ROOT = os.path.dirname(HERE)
+QEMU = os.path.expanduser("~/qemu-build/qemu-7.2+dfsg/build/qemu-system-arm")
+ELF = os.path.expanduser("~/uvk5-port/uvk5-sat/build/CW/nr7y.cw.elf")
+PRISTINE = os.path.join(ROOT, "assets", "pristine", "flash-pristine.img.gz")
+
+BOOT_SECONDS = 20
+
+# App/app/uart.c: Obfuscation[] applied to the payload of every frame.
+XOR_KEY = bytes([
+    0x16, 0x6C, 0x14, 0xE6, 0x2E, 0x91, 0x0D, 0x40,
+    0x21, 0x35, 0xD5, 0x40, 0x13, 0x03, 0xE9, 0x80,
+])
+
+
+def crc16_xmodem(data: bytes) -> int:
+    crc = 0
+    for byte in data:
+        crc ^= byte << 8
+        for _ in range(8):
+            crc = ((crc << 1) ^ 0x1021) & 0xFFFF if crc & 0x8000 else (crc << 1) & 0xFFFF
+    return crc
+
+
+def obfuscate(payload: bytes) -> bytes:
+    return bytes(b ^ XOR_KEY[i % len(XOR_KEY)] for i, b in enumerate(payload))
+
+
+def build_frame(payload: bytes) -> bytes:
+    body = payload + struct.pack("<H", crc16_xmodem(payload))
+    return b"\xab\xcd" + struct.pack("<H", len(payload)) + obfuscate(body) + b"\xdc\xba"
+
+
+def parse_frames(buf: bytes):
+    """Yield deobfuscated payloads found in buf."""
+    out = []
+    i = 0
+    while True:
+        start = buf.find(b"\xab\xcd", i)
+        if start < 0 or start + 4 > len(buf):
+            break
+        length = struct.unpack_from("<H", buf, start + 2)[0]
+        end = start + 4 + length + 2
+        if end + 2 > len(buf) or length > 512:
+            i = start + 2
+            continue
+        body = obfuscate(buf[start + 4:end])
+        out.append(body[:length])
+        i = end + 2
+    return out
+
+
+def main():
+    for path, what in ((QEMU, "QEMU"), (ELF, "firmware"), (PRISTINE, "pristine image")):
+        if not os.path.exists(path):
+            sys.exit(f"missing {what}: {path}")
+
+    workdir = tempfile.mkdtemp(prefix="uvk5-serial-")
+    image = os.path.join(workdir, "flash.img")
+    sock_path = os.path.join(workdir, "serial.sock")
+    with gzip.open(PRISTINE, "rb") as src, open(image, "wb") as dst:
+        shutil.copyfileobj(src, dst)
+
+    # A listening socket for QEMU's serial chardev to connect back to.
+    srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    srv.bind(sock_path)
+    srv.listen(1)
+    srv.settimeout(40)
+
+    proc = subprocess.Popen(
+        [QEMU, "-M", f"uv-k5-v3,flash-image={image}", "-nographic", "-monitor", "none",
+         "-serial", f"unix:{sock_path}", "-kernel", ELF],
+        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+
+    failures = []
+    try:
+        conn, _ = srv.accept()
+        conn.settimeout(15)
+        print("serial connected")
+        time.sleep(BOOT_SECONDS)
+
+        # Drain the boot banner the firmware transmits, so it is not mistaken for a
+        # reply. Transmit already worked; this test is about the other direction.
+        conn.setblocking(False)
+        banner = b""
+        deadline = time.time() + 2
+        while time.time() < deadline:
+            try:
+                chunk = conn.recv(4096)
+                if not chunk:
+                    break
+                banner += chunk
+            except BlockingIOError:
+                time.sleep(0.1)
+        print(f"boot output: {len(banner)} bytes"
+              f"{' (' + banner[:40].decode(errors='replace').strip() + ')' if banner else ''}")
+        conn.setblocking(True)
+
+        # 0x0514: hello. Payload is the command id plus a 4-byte timestamp.
+        payload = struct.pack("<HH", 0x0514, 4) + struct.pack("<I", 0x12345678)
+        conn.sendall(build_frame(payload))
+        print("sent 0x0514 hello")
+
+        conn.settimeout(12)
+        reply = b""
+        deadline = time.time() + 12
+        while time.time() < deadline:
+            try:
+                chunk = conn.recv(4096)
+            except socket.timeout:
+                break
+            if not chunk:
+                break
+            reply += chunk
+            if b"\xdc\xba" in reply:
+                break
+
+        if not reply:
+            failures.append("no reply at all -- the firmware never saw the command")
+        else:
+            print(f"got {len(reply)} bytes back: {reply[:24].hex()}")
+            frames = parse_frames(reply)
+            if not frames:
+                failures.append(f"reply was not a valid frame: {reply[:32].hex()}")
+            else:
+                cmd = struct.unpack_from("<H", frames[0], 0)[0]
+                print(f"reply command id: 0x{cmd:04X}")
+                # 0x0515 is the hello ack.
+                if cmd != 0x0515:
+                    failures.append(f"expected 0x0515 ack, got 0x{cmd:04X}")
+                else:
+                    print("PASS  firmware answered the handshake")
+
+                    # 0x051B: read EEPROM. Ask for 8 bytes at 0x0E70 (VFO indices),
+                    # which proves the receive path carries a real request and that
+                    # flash contents come back over the wire.
+                    req = struct.pack("<HHHBBI", 0x051B, 8, 0x0E70, 8, 0,
+                                      0x12345678)
+                    conn.sendall(build_frame(req))
+                    print("sent 0x051B read of 8 bytes at 0x0E70")
+
+                    data = b""
+                    deadline = time.time() + 12
+                    while time.time() < deadline:
+                        try:
+                            chunk = conn.recv(4096)
+                        except socket.timeout:
+                            break
+                        if not chunk:
+                            break
+                        data += chunk
+                        if b"\xdc\xba" in data:
+                            break
+
+                    read_frames = parse_frames(data)
+                    if not read_frames:
+                        failures.append(
+                            f"no valid frame for the EEPROM read: {data[:32].hex()}")
+                    else:
+                        rcmd = struct.unpack_from("<H", read_frames[0], 0)[0]
+                        print(f"reply command id: 0x{rcmd:04X}")
+                        if rcmd != 0x051C:
+                            failures.append(
+                                f"expected 0x051C read reply, got 0x{rcmd:04X}")
+                        else:
+                            body = read_frames[0]
+                            print(f"payload: {body.hex()}")
+                            print("PASS  firmware served an EEPROM read")
+
+    except socket.timeout:
+        failures.append("QEMU never connected its serial port")
+    finally:
+        try:
+            proc.terminate()
+            proc.wait(timeout=15)
+        except Exception:
+            proc.kill()
+        srv.close()
+        shutil.rmtree(workdir, ignore_errors=True)
+
+    if failures:
+        print()
+        for f in failures:
+            print(f"FAIL {f}")
+        sys.exit(1)
+    print("\nserial receive works: the firmware accepts programming commands")
+
+
+if __name__ == "__main__":
+    main()