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.
This commit is contained in:
2 files changed
+63
-130
No files matched your search
@@ -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 ──────────────────────────────────────
|
||||
|
||||
Reference in new issue
Block a user