feat: python link simulator and tests

tests/link_sim.py

@@ -0,0 +1,235 @@
+#!/usr/bin/env python3
+"""UART link protocol library + simulator CLI for the looper <-> Box-3 link.
+
+Mirror of shared/looper_link.h (single source of truth for the framing).
+
+Usage (needs pyserial for the CLI, not for the library):
+    python3 tests/link_sim.py --port /dev/cu.usbserial-XXXX monitor
+    python3 tests/link_sim.py --port ... cmd SELECT_TRACK 2
+    python3 tests/link_sim.py --port ... cmd SET_TEMPO 0 90
+    python3 tests/link_sim.py --port ... replay-state
+"""
+from __future__ import annotations
+
+import struct
+
+SOF = 0xA5
+BAUD = 460800
+TRACK_CNT = 4
+
+MSG_STATE = 0x01
+MSG_EVT = 0x02
+MSG_CMD = 0x10
+
+TRK_DATA = 1 << 0
+TRK_ACTIVE = 1 << 1
+TRK_REC = 1 << 2
+TRK_ERASE = 1 << 3
+
+CMD_SELECT_TRACK = 0x01
+CMD_STOP_TRACK = 0x02
+CMD_ERASE_TRACK = 0x03
+CMD_SET_LENGTH = 0x04
+CMD_START_ALL = 0x05
+CMD_STOP_ALL = 0x06
+CMD_GAIN_OUT = 0x07
+CMD_PAN = 0x08
+CMD_GAIN_IN = 0x09
+CMD_CLICK_TOGGLE = 0x0A
+CMD_SET_TEMPO = 0x0B
+CMD_TOGGLE_SRC = 0x0C
+CMD_RESET_ALL = 0x0D
+CMD_SD_SAVE = 0x0E
+CMD_SD_LOAD = 0x0F
+
+CMD_BY_NAME = {
+    name[4:]: value
+    for name, value in list(globals().items())
+    if name.startswith("CMD_")
+}
+
+EVT_NAMES = {1: "SD_PROGRESS", 2: "SD_DONE", 3: "SD_ERROR", 4: "RESET_DONE"}
+
+# layout of link_state_t (little endian, packed)
+STATE_FMT = "<4B4B4BHHHBBB6B"
+STATE_SIZE = struct.calcsize(STATE_FMT)
+assert STATE_SIZE == 27
+
+
+def crc8(data: bytes) -> int:
+    """CRC8 Dallas/Maxim (reflected, poly 0x8C)."""
+    crc = 0
+    for byte in data:
+        for _ in range(8):
+            mix = (crc ^ byte) & 0x01
+            crc >>= 1
+            if mix:
+                crc ^= 0x8C
+            byte >>= 1
+    return crc
+
+
+def encode_frame(msg_type: int, payload: bytes) -> bytes:
+    body = bytes([msg_type, len(payload)]) + payload
+    return bytes([SOF]) + body + bytes([crc8(body)])
+
+
+class Parser:
+    """Incremental frame parser; feed() returns (type, payload) or None."""
+
+    def __init__(self) -> None:
+        self.reset()
+
+    def reset(self) -> None:
+        self._state = 0
+        self._type = 0
+        self._len = 0
+        self._payload = bytearray()
+
+    def feed(self, byte: int):
+        if self._state == 0:
+            if byte == SOF:
+                self._state = 1
+        elif self._state == 1:
+            self._type = byte
+            self._state = 2
+        elif self._state == 2:
+            if byte > 64:
+                self._state = 0
+                return None
+            self._len = byte
+            self._payload = bytearray()
+            self._state = 4 if byte == 0 else 3
+        elif self._state == 3:
+            self._payload.append(byte)
+            if len(self._payload) >= self._len:
+                self._state = 4
+        elif self._state == 4:
+            self._state = 0
+            body = bytes([self._type, self._len]) + bytes(self._payload)
+            if crc8(body) == byte:
+                return (self._type, bytes(self._payload))
+        return None
+
+
+def pack_state(
+    flags=(0, 0, 0, 0),
+    gain_out=(255, 255, 255, 255),
+    pan=(128, 128, 128, 128),
+    pos=0,
+    length=65535,
+    tempo_bpm=120,
+    click_on=0,
+    input_src=0,
+    gain_in=255,
+    vu=(0, 0, 0, 0, 0, 0),
+) -> bytes:
+    return struct.pack(
+        STATE_FMT,
+        *flags,
+        *gain_out,
+        *pan,
+        pos,
+        length,
+        tempo_bpm,
+        click_on,
+        input_src,
+        gain_in,
+        *vu,
+    )
+
+
+def unpack_state(payload: bytes) -> dict:
+    fields = struct.unpack(STATE_FMT, payload)
+    return {
+        "flags": list(fields[0:4]),
+        "gain_out": list(fields[4:8]),
+        "pan": list(fields[8:12]),
+        "pos": fields[12],
+        "len": fields[13],
+        "tempo_bpm": fields[14],
+        "click_on": fields[15],
+        "input_src": fields[16],
+        "gain_in": fields[17],
+        "vu": list(fields[18:24]),
+    }
+
+
+def pack_cmd(op: int, arg0: int = 0, arg1: int = 0) -> bytes:
+    return encode_frame(MSG_CMD, struct.pack("<BBH", op, arg0, arg1))
+
+
+def _fmt_state(state: dict) -> str:
+    def track(i: int) -> str:
+        f = state["flags"][i]
+        mode = (
+            "REC" if f & TRK_REC
+            else "ERA" if f & TRK_ERASE
+            else "PLA" if f & TRK_ACTIVE
+            else "dat" if f & TRK_DATA
+            else "---"
+        )
+        return f"T{i + 1}:{mode} vu={state['vu'][i]:3d}"
+
+    tracks = "  ".join(track(i) for i in range(TRACK_CNT))
+    return (
+        f"{tracks}  pos={state['pos']:5d}/{state['len']:5d}  "
+        f"bpm={state['tempo_bpm']:3d} click={state['click_on']} "
+        f"src={'mic' if state['input_src'] else 'line'} "
+        f"in={state['vu'][4]:3d} out={state['vu'][5]:3d}"
+    )
+
+
+def main() -> int:
+    import argparse
+    import time
+
+    import serial  # pyserial, CLI only
+
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--port", required=True)
+    ap.add_argument("--baud", type=int, default=BAUD)
+    sub = ap.add_subparsers(dest="mode", required=True)
+    sub.add_parser("monitor", help="decode incoming STATE/EVT frames")
+    p_cmd = sub.add_parser("cmd", help="send one CMD frame")
+    p_cmd.add_argument("op", choices=sorted(CMD_BY_NAME))
+    p_cmd.add_argument("arg0", type=int, nargs="?", default=0)
+    p_cmd.add_argument("arg1", type=int, nargs="?", default=0)
+    sub.add_parser("replay-state", help="emit demo STATE frames at 20 Hz")
+    args = ap.parse_args()
+
+    with serial.Serial(args.port, args.baud, timeout=0.1) as ser:
+        if args.mode == "cmd":
+            frame = pack_cmd(CMD_BY_NAME[args.op], args.arg0, args.arg1)
+            ser.write(frame)
+            print(f"sent {frame.hex(' ')}")
+        elif args.mode == "monitor":
+            parser = Parser()
+            while True:
+                for byte in ser.read(256):
+                    result = parser.feed(byte)
+                    if result is None:
+                        continue
+                    msg_type, payload = result
+                    if msg_type == MSG_STATE and len(payload) == STATE_SIZE:
+                        print(_fmt_state(unpack_state(payload)), end="\r")
+                    elif msg_type == MSG_EVT and len(payload) == 2:
+                        name = EVT_NAMES.get(payload[0], hex(payload[0]))
+                        print(f"\nEVT {name} arg={payload[1]}")
+        elif args.mode == "replay-state":
+            pos = 0
+            while True:
+                pos = (pos + 655) % 65536
+                payload = pack_state(
+                    flags=(TRK_DATA | TRK_ACTIVE, TRK_DATA | TRK_REC, 0, 0),
+                    pos=pos,
+                    click_on=1,
+                    vu=(120, 200, 0, 0, 80, 150),
+                )
+                ser.write(encode_frame(MSG_STATE, payload))
+                time.sleep(0.05)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_link_sim.py

@@ -0,0 +1,72 @@
+"""Unit tests for tests/link_sim.py (UART link protocol, python side).
+
+Parity with tests/test_link_protocol.c: same CRC, same layout, same frozen
+vector for CMD SELECT_TRACK track 2.
+"""
+import pytest
+
+from link_sim import (
+    CMD_SELECT_TRACK,
+    MSG_CMD,
+    MSG_STATE,
+    Parser,
+    crc8,
+    encode_frame,
+    pack_cmd,
+    pack_state,
+    unpack_state,
+)
+
+# printed by tests/test_link_protocol.c (frozen, do not change)
+CMD_VECTOR = bytes.fromhex("A5 10 04 01 02 00 00 84".replace(" ", ""))
+
+
+def test_cmd_vector_parity_with_c():
+    assert pack_cmd(CMD_SELECT_TRACK, 2) == CMD_VECTOR
+
+
+def test_state_roundtrip():
+    payload = pack_state(pos=1234, tempo_bpm=120, vu=[0, 0, 0, 0, 0, 200])
+    assert len(payload) == 27
+    frame = encode_frame(MSG_STATE, payload)
+    parser = Parser()
+    frames = [f for b in frame if (f := parser.feed(b)) is not None]
+    assert len(frames) == 1
+    msg_type, rx = frames[0]
+    assert msg_type == MSG_STATE
+    state = unpack_state(rx)
+    assert state["pos"] == 1234
+    assert state["tempo_bpm"] == 120
+    assert state["vu"][5] == 200
+
+
+def test_bad_crc_rejected_then_resync():
+    payload = pack_state()
+    frame = bytearray(encode_frame(MSG_STATE, payload))
+    frame[-1] ^= 0xFF
+    parser = Parser()
+    assert all(parser.feed(b) is None for b in frame)
+    # noise, then a clean frame is still accepted
+    for b in b"\x00\xa5\x42":
+        parser.feed(b)
+    parser.reset()
+    good = encode_frame(MSG_STATE, payload)
+    frames = [f for b in good if (f := parser.feed(b)) is not None]
+    assert len(frames) == 1
+
+
+def test_crc8_known_value():
+    # CRC over type+len+payload of the frozen CMD vector
+    assert crc8(CMD_VECTOR[1:-1]) == CMD_VECTOR[-1]
+
+
+def test_encode_frame_layout():
+    frame = encode_frame(MSG_CMD, b"\x01\x02\x00\x00")
+    assert frame[0] == 0xA5
+    assert frame[1] == MSG_CMD
+    assert frame[2] == 4
+    assert len(frame) == 8
+
+
+if __name__ == "__main__":
+    raise SystemExit(pytest.main([__file__, "-v"]))