The First Wire to Another Machine

# labs/hardware_protocol.py
import json

MAGIC = b"GLAD"

def encode_message(payload: dict) -> bytes:
    body = json.dumps(payload).encode("utf-8")
    return MAGIC + body

def decode_message(data: bytes) -> dict:
    if data[:4] != MAGIC:
        raise ValueError("Invalid magic bytes")
    return json.loads(data[4:].decode("utf-8"))

if __name__ == "__main__":
    msg = {"command": "arm.move", "params": {"angle": 90}}
    print("Round-trip OK:", decode_message(encode_message(msg)) == msg)

$ uv run python -m labs.hardware_protocol
Round-trip OK: True

Four bytes that spell nothing to anyone else are the cheapest question a receiver can ask: are these mine? Junk from a wrong port, a leftover boot message, a stream read half a byte out of step, all fail here for a few nanoseconds of comparison instead of arriving at json.loads dressed as data. The stage still cheats, though: data[4:] assumes the buffer holds exactly one body.

import struct

def encode_message(payload: dict) -> bytes:
    body = json.dumps(payload).encode("utf-8")
    return MAGIC + struct.pack(">I", len(body)) + body

def decode_message(data: bytes) -> dict:
    if data[:4] != MAGIC:
        raise ValueError("Invalid magic bytes")
    length = struct.unpack(">I", data[4:8])[0]
    return json.loads(data[8:8 + length].decode("utf-8"))

if __name__ == "__main__":
    frame = encode_message({"command": "arm.move", "params": {"angle": 90}})
    print("Body length declared:", struct.unpack(">I", frame[4:8])[0], "bytes")

$ uv run python -m labs.hardware_protocol
Round-trip OK: True
Body length declared: 48 bytes

struct.pack(">I", n) writes a 32-bit unsigned integer as four bytes, and the > pins big-endian order so the two ends never argue about which byte comes first. That argument is real: your laptop stores integers low byte first, plenty of other silicon does not, and a length field read backwards asks for two billion bytes. Declare the order in the format string and neither machine gets a vote.

import hashlib

def encode_message(payload: dict) -> bytes:
    body = json.dumps(payload).encode("utf-8")
    checksum = hashlib.md5(body).digest()[:4]
    return MAGIC + struct.pack(">I", len(body)) + checksum + body

def decode_message(data: bytes) -> dict:
    if data[:4] != MAGIC:
        raise ValueError("Invalid magic bytes")
    length, checksum = struct.unpack(">I4s", data[4:12])
    body = data[12:12 + length]
    if len(body) < length:
        raise ValueError(f"Short frame: {len(body)} of {length} body bytes")
    if hashlib.md5(body).digest()[:4] != checksum:
        raise ValueError("Checksum mismatch")
    return json.loads(body.decode("utf-8"))

if __name__ == "__main__":
    frame = encode_message({"command": "arm.move", "params": {"angle": 90}})
    print("Frame size:", len(frame), "bytes")
    print("Header:", frame[:12].hex())
    tampered = bytearray(frame)
    tampered[15] ^= 0xFF          # flip every bit of one body byte
    try:
        decode_message(bytes(tampered))
    except ValueError as err:
        print("Tamper detected:", err)

$ uv run python -m labs.hardware_protocol
Round-trip OK: True
Body length declared: 48 bytes
Frame size: 60 bytes
Header: 474c414400000030b0740f21
Tamper detected: Checksum mismatch

Read that header by eye and the layout stops being abstract: 474c4144 is GLAD in ASCII, 00000030 is 48 in hexadecimal, and b0740f21 is the first four bytes of the body's MD5 digest. Twelve plus forty-eight is the sixty the program printed. Byte 15 lands in the body, since the header ends at 12, so flipping it changes the recomputed digest and the guard fires. Four bytes of digest miss a corruption roughly once in four billion, which is a different league from the zero protection the line offers on its own.

from typing import Callable

def read_frame(read: Callable[[int], bytes]) -> dict:
    def read_exactly(n: int) -> bytes:
        buf = b""
        while len(buf) < n:
            chunk = read(n - len(buf))
            if not chunk:
                raise ConnectionError("stream ended mid-frame")
            buf += chunk
        return buf

    window = read_exactly(4)
    while window != MAGIC:                      # resync one byte at a time
        window = window[1:] + read_exactly(1)
    length, checksum = struct.unpack(">I4s", read_exactly(8))
    body = read_exactly(length)
    if hashlib.md5(body).digest()[:4] != checksum:
        raise ValueError("Checksum mismatch")
    return json.loads(body.decode("utf-8"))

if __name__ == "__main__":
    JUNK = b"ets Jun  8 2016\r\n"
    stream = JUNK + encode_message({"command": "arm.move", "params": {"angle": 90}})
    pos = 0

    def dribble(n: int) -> bytes:               # hand back 3 bytes at most
        global pos
        chunk = stream[pos:pos + min(n, 3)]
        pos += len(chunk)
        return chunk

    print(f"Skipped {len(JUNK)} junk bytes, read:", read_frame(dribble))

$ uv run python -m labs.hardware_protocol
Skipped 17 junk bytes, read: {'command': 'arm.move', 'params': {'angle': 90}}

read_frame takes a read function, not a socket or a serial port, so the same code drives a fake stream in a test and a real cable in a minute. Two ideas earn their lines. read_exactly loops until it has every byte it asked for, because one read is never one message. And the resync loop slides a four-byte window forward until the magic appears, which turns garbage on the line from a fatal error into a delay of a few bytes.

# esp32/main.py — copied to the board with: mpremote cp esp32/main.py :main.py
import hashlib
import json
import struct
import sys

import micropython
from machine import Pin

MAGIC = b"GLAD"
LED = Pin(2, Pin.OUT)

def read_exactly(n):
    buf = b""
    while len(buf) < n:
        chunk = sys.stdin.buffer.read(n - len(buf))
        if chunk:
            buf += chunk
    return buf

def read_frame():
    window = read_exactly(4)
    while window != MAGIC:
        window = window[1:] + read_exactly(1)
    length, checksum = struct.unpack(">I4s", read_exactly(8))
    body = read_exactly(length)
    if hashlib.md5(body).digest()[:4] != checksum:
        raise ValueError("Checksum mismatch")
    return json.loads(body.decode("utf-8"))

def send_frame(payload):
    body = json.dumps(payload).encode("utf-8")
    checksum = hashlib.md5(body).digest()[:4]
    sys.stdout.buffer.write(MAGIC + struct.pack(">I", len(body)) + checksum + body)

def handle(msg):
    command = msg.get("command")
    if command == "ping":
        return {"status": "ok", "command": "ping"}
    if command == "led.set":
        LED.value(1 if msg.get("params", {}).get("state") == "on" else 0)
        return {"status": "ok", "command": command, "led": LED.value()}
    return {"status": "error", "reason": "unknown command: %s" % command}

def main():
    micropython.kbd_intr(-1)      # 0x03 in a frame is data, not Ctrl-C
    while True:
        try:
            msg = read_frame()
        except ValueError as err:
            send_frame({"status": "error", "reason": str(err)})
            continue
        send_frame(handle(msg))

main()

The board's half is the same protocol written twice, and that duplication is deliberate: MicroPython cannot import your laptop's package, so the frame layout lives in two files that must agree byte for byte. Note what the far end does with a bad checksum. It answers, saying what went wrong, and goes back to reading. A microcontroller that stops talking is a microcontroller you drive to the bench and power-cycle, so every path through main ends back at read_frame.

# labs/hardware_link.py
import time

import serial

from labs.hardware_protocol import encode_message, read_frame

PORT = "/dev/ttyUSB0"
BAUD = 115200

def open_link(port: str = PORT, baud: int = BAUD) -> serial.Serial:
    link = serial.Serial(port, baud, timeout=2)
    time.sleep(2.0)              # opening the port resets the board
    link.reset_input_buffer()    # drop whatever the bootloader printed
    return link

def command(link: serial.Serial, payload: dict) -> dict:
    link.write(encode_message(payload))
    return read_frame(link.read)

def main() -> None:
    with open_link() as link:
        for payload in [
            {"command": "ping"},
            {"command": "led.set", "params": {"state": "on"}},
            {"command": "led.set", "params": {"state": "off"}},
            {"command": "dance"},
        ]:
            started = time.perf_counter()
            reply = command(link, payload)
            elapsed_ms = (time.perf_counter() - started) * 1000
            print(f"{payload['command']:<8} -> {reply}  ({elapsed_ms:.0f} ms)")

if __name__ == "__main__":
    main()

$ uv run python -m labs.hardware_link   # measured on the bench — yours will vary
ping     -> {'status': 'ok', 'command': 'ping'}  (11 ms)
led.set  -> {'status': 'ok', 'command': 'led.set', 'led': 1}  (14 ms)
led.set  -> {'status': 'ok', 'command': 'led.set', 'led': 0}  (13 ms)
dance    -> {'status': 'error', 'reason': 'unknown command: dance'}  (13 ms)

The blue light on the board comes on and goes off. That is the whole point of the chapter. Set PORT to whatever your machine calls the board: /dev/ttyUSB0 for the common USB-serial chips on Linux, /dev/ttyACM0 for boards with native USB, /dev/tty.usbserial-0001 on macOS, COM3 or thereabouts on Windows. Run uv run python -m serial.tools.list_ports -v and the machine will tell you. The round trips landed in the low tens of milliseconds here; roughly half of that is the bytes themselves at 115200 baud and the rest is MicroPython parsing JSON, so your figures will sit wherever your board and cable put them.