import datetime
import json
import sys
import time
from pathlib import Path
import serial
PORT = sys.argv[1] if len(sys.argv) > 1 else "COM7"
OUTPUT = (
Path(sys.argv[2])
if len(sys.argv) > 2
else Path(__file__).parent.parent / "tests" / "fixtures" / "meshtastic_config.json"
)
LABEL = sys.argv[3] if len(sys.argv) > 3 else "config"
WANT_CONFIG_FIELD = 3
if OUTPUT.exists():
raise SystemExit(f"refusing to overwrite existing capture: {OUTPUT}")
def frame(pb):
return bytes([0x94, 0xC3]) + len(pb).to_bytes(2, "big") + pb
def varint(v):
out = bytearray()
while True:
b = v & 0x7F
v >>= 7
out.append(b | 0x80 if v else b)
if not v:
return bytes(out)
def deframe(buf):
frames, i = [], 0
while i < len(buf) - 3:
if buf[i] == 0x94 and buf[i + 1] == 0xC3:
ln = (buf[i + 2] << 8) | buf[i + 3]
if i + 4 + ln <= len(buf):
frames.append(bytes(buf[i + 4 : i + 4 + ln]))
i += 4 + ln
continue
i += 1
return frames
s = serial.Serial(PORT, 115200, timeout=0.2)
time.sleep(0.4)
s.reset_input_buffer()
s.write(frame(varint((WANT_CONFIG_FIELD << 3) | 0) + varint(0x11223344)))
s.flush()
buf = bytearray()
last = time.time()
while time.time() - last < 1.5: c = s.read(8192)
if c:
buf.extend(c)
last = time.time()
s.close()
frames = deframe(buf)
print(f"captured {len(buf)} bytes, {len(frames)} FromRadio frames")
OUTPUT.write_text(
json.dumps(
{
"_comment": (
"Meshtastic-compatible device config stream, captured black-box "
f"{datetime.date.today().isoformat()}. "
"want_config request = field 3 (discovered by empirically probing which field "
"number triggers a response; no schema was read). Stream API framing (0x94 0xc3 "
"+ BE u16 len). frames = the raw FromRadio payloads streamed back, hex. Field "
"NUMBERS/wire TYPES are observable facts; meanings are NOT asserted here."
),
"label": LABEL,
"port": PORT,
"want_config_field": WANT_CONFIG_FIELD,
"frame_count": len(frames),
"frames": [f.hex() for f in frames],
},
indent=1,
),
encoding="utf-8",
)
print(f"wrote {OUTPUT}")