From 7a947a1a05cb69c9c234ba3ef41b16f60e022cd7 Mon Sep 17 00:00:00 2001
From: Stephen Waits <steve@waits.net>
Date: Sun, 29 Mar 2026 16:56:13 -0600
Subject: [PATCH] refactor: update Python examples for fixed-size LE wire
 format

Remove all varint/zigzag encoding. Every field is now struct.pack/unpack
with fixed-size little-endian format strings. RadioConfig is a single
struct.pack('<IBBBHB', ...) call. Response decoding uses struct.unpack_from
with known offsets. No encoding helpers needed beyond struct.
---
 examples/donglora.py             |  89 +++++++++-----------------
 examples/meshcore/meshcore_rx.py | 104 +++++++++++--------------------
 2 files changed, 63 insertions(+), 130 deletions(-)

diff --git a/examples/donglora.py b/examples/donglora.py
index 3d1259db..04888915 100644
--- a/examples/donglora.py
+++ b/examples/donglora.py
@@ -80,61 +80,27 @@ def read_frame(ser: serial.Serial) -> bytes | None:
         return None
 
 
-# ── Postcard encoding ────────────────────────────────────────────
-
-def varint(n: int) -> bytes:
-    out = []
-    while n >= 0x80:
-        out.append((n & 0x7F) | 0x80)
-        n >>= 7
-    out.append(n & 0x7F)
-    return bytes(out)
-
-
-def zigzag(n: int) -> bytes:
-    return varint((n << 1) ^ (n >> 31) if n >= 0 else ((-n - 1) << 1) | 1)
-
-
-def decode_varint(data: bytes) -> tuple[int, bytes]:
-    n, shift = 0, 0
-    for i, b in enumerate(data):
-        n |= (b & 0x7F) << shift
-        shift += 7
-        if not (b & 0x80):
-            return n, data[i + 1:]
-    return n, b""
-
-
-def decode_zigzag(data: bytes) -> tuple[int, bytes]:
-    n, rest = decode_varint(data)
-    return (n >> 1) ^ -(n & 1), rest
-
-
-# ── Command encoding ─────────────────────────────────────────────
+# ── Fixed-size LE encoding ────────────────────────────────────────
 
 def encode_config(cfg: dict) -> bytes:
-    """Encode a RadioConfig dict to postcard bytes.
-
-    Postcard wire format: u8/i8 are raw bytes, u16/u32 are varint,
-    i16/i32 are zigzag+varint, enums are varint variant index.
-    """
-    out = varint(cfg["freq_hz"])              # u32: varint
-    out += varint(cfg["bw"])                  # enum: varint variant index
-    out += struct.pack("B", cfg["sf"])        # u8: raw byte
-    out += struct.pack("B", cfg["cr"])        # u8: raw byte
-    out += varint(cfg["sync_word"])           # u16: varint
-    out += struct.pack("b", cfg["tx_power_dbm"])  # i8: raw signed byte
-    return out
+    """Encode a RadioConfig to 10 fixed-size LE bytes."""
+    return struct.pack("<IBBBHB",
+        cfg["freq_hz"],
+        cfg["bw"],
+        cfg["sf"],
+        cfg["cr"],
+        cfg["sync_word"],
+        cfg["tx_power_dbm"] & 0xFF,  # i8 as unsigned byte
+    )
 
 
 def encode_command(cmd: str, **kwargs) -> bytes:
-    """Encode a command name + kwargs to postcard bytes."""
-    commands = {
+    """Encode a command to fixed-size LE bytes."""
+    tags = {
         "Ping": 0, "GetConfig": 1, "SetConfig": 2, "StartRx": 3,
         "StopRx": 4, "Transmit": 5, "DisplayOn": 6, "DisplayOff": 7,
     }
-    idx = commands[cmd]
-    out = varint(idx)
+    out = bytes([tags[cmd]])
     if cmd == "SetConfig":
         out += encode_config(kwargs["config"])
     elif cmd == "Transmit":
@@ -144,36 +110,37 @@ def encode_command(cmd: str, **kwargs) -> bytes:
         else:
             out += b"\x01" + encode_config(config)
         payload = kwargs["payload"]
-        out += varint(len(payload)) + payload
+        out += struct.pack("<H", len(payload)) + payload
     return out
 
 
 # ── Response decoding ────────────────────────────────────────────
 
 def decode_response(data: bytes) -> dict:
-    """Decode a postcard-encoded response into a dict."""
+    """Decode a fixed-size LE response."""
     if not data:
         return {"type": "Empty"}
-    variant = data[0]
+    tag = data[0]
     rest = data[1:]
-    if variant == 0:
+    if tag == 0:
         return {"type": "Pong"}
-    elif variant == 1:
+    elif tag == 1:
         return {"type": "Config", "raw": rest.hex()}
-    elif variant == 2:
-        rssi, rest = decode_zigzag(rest)
-        snr, rest = decode_zigzag(rest)
-        plen, rest = decode_varint(rest)
-        return {"type": "RxPacket", "rssi": rssi, "snr": snr, "payload": rest[:plen]}
-    elif variant == 3:
+    elif tag == 2:
+        rssi = struct.unpack_from("<h", rest, 0)[0]
+        snr = struct.unpack_from("<h", rest, 2)[0]
+        plen = struct.unpack_from("<H", rest, 4)[0]
+        payload = rest[6:6 + plen]
+        return {"type": "RxPacket", "rssi": rssi, "snr": snr, "payload": payload}
+    elif tag == 3:
         return {"type": "TxDone"}
-    elif variant == 4:
+    elif tag == 4:
         return {"type": "Ok"}
-    elif variant == 5:
+    elif tag == 5:
         code = rest[0] if rest else -1
         return {"type": "Error", "code": code}
     else:
-        return {"type": f"Unknown({variant})"}
+        return {"type": f"Unknown({tag})"}
 
 
 # ── High-level helpers ───────────────────────────────────────────
diff --git a/examples/meshcore/meshcore_rx.py b/examples/meshcore/meshcore_rx.py
index 67842f04..4ded82e0 100755
--- a/examples/meshcore/meshcore_rx.py
+++ b/examples/meshcore/meshcore_rx.py
@@ -409,101 +409,67 @@ def read_frame(ser: serial.Serial) -> bytes | None:
         return None
 
 
-# ── Postcard serialization ─────────────────────────────────────────
-
-
-def varint(n: int) -> bytes:
-    out = []
-    while n >= 0x80:
-        out.append((n & 0x7F) | 0x80)
-        n >>= 7
-    out.append(n & 0x7F)
-    return bytes(out)
-
-
-def zigzag(n: int) -> bytes:
-    return varint((n << 1) ^ (n >> 31) if n >= 0 else ((-n - 1) << 1) | 1)
+# ── Fixed-size LE serialization ────────────────────────────────────
 
 
 def encode_radio_config(cfg: dict) -> bytes:
-    out = varint(cfg["freq_hz"])     # u32: varint
-    out += varint(cfg["bw"])         # enum: varint variant index
-    out += struct.pack("B", cfg["sf"])  # u8: raw byte
-    out += struct.pack("B", cfg["cr"])  # u8: raw byte
-    out += varint(cfg["sync_word"])  # u16: varint
-    out += struct.pack("b", cfg["tx_power_dbm"])  # i8: raw byte (signed)
-    return out
+    """Encode RadioConfig to 10 fixed-size LE bytes."""
+    return struct.pack("<IBBBHB",
+        cfg["freq_hz"],
+        cfg["bw"],
+        cfg["sf"],
+        cfg["cr"],
+        cfg["sync_word"],
+        cfg["tx_power_dbm"] & 0xFF,
+    )
 
 
 def encode_command(cmd: dict) -> bytes:
+    """Encode a command dict to fixed-size LE bytes."""
     kind = cmd["type"]
-    if kind == "Ping":
-        return varint(0)
-    elif kind == "GetConfig":
-        return varint(1)
-    elif kind == "SetConfig":
-        return varint(2) + encode_radio_config(cmd["config"])
-    elif kind == "StartRx":
-        return varint(3)
-    elif kind == "StopRx":
-        return varint(4)
+    tags = {
+        "Ping": 0, "GetConfig": 1, "SetConfig": 2, "StartRx": 3,
+        "StopRx": 4, "Transmit": 5, "DisplayOn": 6, "DisplayOff": 7,
+    }
+    out = bytes([tags[kind]])
+    if kind == "SetConfig":
+        out += encode_radio_config(cmd["config"])
     elif kind == "Transmit":
-        out = varint(5)
         config = cmd.get("config")
         if config is None:
-            out += b"\x00"  # Option::None
+            out += b"\x00"
         else:
-            out += b"\x01" + encode_radio_config(config)  # Option::Some
+            out += b"\x01" + encode_radio_config(config)
         payload = cmd["payload"]
-        out += varint(len(payload)) + payload
-        return out
-    elif kind == "DisplayOn":
-        return varint(6)
-    elif kind == "DisplayOff":
-        return varint(7)
-    else:
-        raise ValueError(f"Unknown command type: {kind}")
-
-
-def decode_varint(data: bytes) -> tuple[int, bytes]:
-    n, shift = 0, 0
-    for i, b in enumerate(data):
-        n |= (b & 0x7F) << shift
-        shift += 7
-        if not (b & 0x80):
-            return n, data[i + 1 :]
-    return n, b""
-
-
-def decode_zigzag_varint(data: bytes) -> tuple[int, bytes]:
-    n, rest = decode_varint(data)
-    return (n >> 1) ^ -(n & 1), rest
+        out += struct.pack("<H", len(payload)) + payload
+    return out
 
 
 def decode_response(data: bytes) -> dict:
+    """Decode a fixed-size LE response."""
     if not data:
         return {"type": "Empty"}
-    variant = data[0]
+    tag = data[0]
     rest = data[1:]
-    if variant == 0:
+    if tag == 0:
         return {"type": "Pong"}
-    elif variant == 1:
+    elif tag == 1:
         return {"type": "Config", "raw": rest.hex()}
-    elif variant == 2:
-        rssi, rest = decode_zigzag_varint(rest)
-        snr, rest = decode_zigzag_varint(rest)
-        plen, rest = decode_varint(rest)
-        payload = rest[:plen]
+    elif tag == 2:
+        rssi = struct.unpack_from("<h", rest, 0)[0]
+        snr = struct.unpack_from("<h", rest, 2)[0]
+        plen = struct.unpack_from("<H", rest, 4)[0]
+        payload = rest[6:6 + plen]
         return {"type": "RxPacket", "rssi": rssi, "snr": snr, "payload": payload}
-    elif variant == 3:
+    elif tag == 3:
         return {"type": "TxDone"}
-    elif variant == 4:
+    elif tag == 4:
         return {"type": "Ok"}
-    elif variant == 5:
+    elif tag == 5:
         code = rest[0] if rest else -1
         return {"type": "Error", "code": code}
     else:
-        return {"type": f"Unknown({variant})", "raw": rest.hex()}
+        return {"type": f"Unknown({tag})", "raw": rest.hex()}
 
 
 # ── MeshCore packet decoder ──────────────────────────────────────