diff --git a/.gitignore b/.gitignore index db3d538..e8079ad 100644 --- a/.gitignore +++ b/.gitignore @@ -46,3 +46,6 @@ _build .idea .vscode *~ + +.claude +CLAUDE.md diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index ff19dde..bf62649 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -4,18 +4,18 @@ repos: - repo: https://github.com/pre-commit/pre-commit-hooks - rev: v4.5.0 + rev: v6.0.0 hooks: - id: check-yaml - id: end-of-file-fixer - id: trailing-whitespace - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.3.4 + rev: v0.15.12 hooks: - id: ruff-format - id: ruff args: ["--fix"] - repo: https://github.com/fsfe/reuse-tool - rev: v3.0.1 + rev: v6.2.0 hooks: - id: reuse diff --git a/README.rst b/README.rst index edcaae7..3cd2ab1 100644 --- a/README.rst +++ b/README.rst @@ -17,6 +17,17 @@ Introduction MIDI library for CircuitPython +TMIDI is a MIDI library for CircuitPython / MicroPython. It attempts to +be a provide a simpler API and perform faster, especially for MIDI parsing, +compared to ``adafruit_midi`` (`adafruit_midi github `_) +TMIDI is derived from ``winterbloom_smolmidi`` +(`smolmidi github `_) +by Thea Flowers for Winterbloom. + +Like ``adafruit_midi``, TMIDI works on CircuitPython's ``usb_midi`` and ``busio.UART``, +or any stream-like object that supports ``.readinto()`` and ``.write()`` + + Dependencies ============= @@ -112,8 +123,7 @@ For information on building library documentation, please check out Testing ======= -Install ``pytest`` with ``pip3 install pytest --upgrade`` and run ``pytest -v`` -To build docs: +Install ``pytest`` with ``pip3 install pytest circuitpython-mocks --upgrade`` and run ``pytest -v`` Contributing ============ diff --git a/docs/index.rst b/docs/index.rst index 536aaa5..4939ec5 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -36,10 +36,7 @@ Table of Contents Download Library Bundle CircuitPython Reference Documentation CircuitPython Support Forum - Discord Chat - Adafruit Learning System - Adafruit Blog - Adafruit Store + Indices and tables ================== diff --git a/examples/tmidi_adafruit_midi_comparison.py b/examples/tmidi_adafruit_midi_comparison.py new file mode 100644 index 0000000..b53e8e5 --- /dev/null +++ b/examples/tmidi_adafruit_midi_comparison.py @@ -0,0 +1,117 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 Tod Kurt +# +# SPDX-License-Identifier: MIT + +""" +Synthetic benchmark comparing tmidi vs adafruit_midi receive() performance. +Run on a CircuitPython board (tested on RP2350 Pico 2). + +Copy both tmidi.py and adafruit_midi/ to the board's lib/ folder, then run this +script. Each library is tested against the same pre-loaded byte buffer so results +are directly comparable. Results print as µs/msg and net heap change. + +Example output (RP2350 Pico 2): + tmidi : 6000 msgs, 160 µs/msg, heap: -32 bytes + adafruit_midi : 6000 msgs, 385 µs/msg, heap: -32 bytes + tmidi is 2.4x faster than adafruit_midi +""" + +import gc +import time + +# One repetition of every message type to test. Add/remove rows freely. +# fmt: off +_PATTERN = bytes([ + 0x90, 60, 100, # NOTE_ON + 0xF8, # CLOCK + 0x80, 60, 0, # NOTE_OFF + 0xB0, 74, 64, # CC + 0xF8, # CLOCK + 0xD0, 64, # CHANNEL_PRESSURE + 0xC0, 42, # PROGRAM_CHANGE + 0xFA, # START + 0xE0, 0x00, 0x40, # PITCH_BEND center + 0xFC, # STOP +]) +# fmt: on +_MSGS_PER_PATTERN = 10 # update this when adding/removing rows above +REPS = 1000 +BENCH_BYTES = _PATTERN * REPS + + +class BufPort: + """Fake MIDI port backed by a pre-loaded byte buffer. + Implements both read() (adafruit_midi) and readinto() (tmidi). + """ + + def __init__(self, data): + self._data = memoryview(data) + self._pos = 0 + + def read(self, numbytes=1): + if self._pos >= len(self._data): + return b"" + end = min(self._pos + numbytes, len(self._data)) + chunk = bytes(self._data[self._pos : end]) + self._pos = end + return chunk + + def readinto(self, buf, numbytes=1): + if self._pos >= len(self._data): + return 0 + buf[0] = self._data[self._pos] + self._pos += 1 + return 1 + + +def run_benchmark(midi, label): + gc.collect() + free_before = gc.mem_free() + t0 = time.monotonic_ns() + count = 0 + while True: + if midi.receive() is None: + break + count += 1 + elapsed_us = (time.monotonic_ns() - t0) // 1000 + gc.collect() + heap_delta = gc.mem_free() - free_before + us_per_msg = elapsed_us // count if count else 0 + print(f" {label:<15}: {count} msgs, {us_per_msg} µs/msg, heap: {heap_delta} bytes") + return us_per_msg + + +print("tmidi vs adafruit_midi synthetic benchmark") +print("=" * 50) + +results = {} + +try: + import tmidi + # import tmidi_old as tmidi + + midi = tmidi.MIDI(midi_in=BufPort(BENCH_BYTES)) + results["tmidi"] = run_benchmark(midi, "tmidi") +except ImportError: + print(" tmidi: not installed") + +try: + import adafruit_midi + from adafruit_midi.control_change import ControlChange # noqa: F401 + from adafruit_midi.note_on import NoteOn # noqa: F401 + from adafruit_midi.note_off import NoteOff # noqa: F401 + from adafruit_midi.pitch_bend import PitchBend # noqa: F401 + from adafruit_midi.channel_pressure import ChannelPressure # noqa: F401 + from adafruit_midi.program_change import ProgramChange # noqa: F401 + from adafruit_midi.timing_clock import TimingClock # noqa: F401 + from adafruit_midi.start import Start # noqa: F401 + from adafruit_midi.stop import Stop # noqa: F401 + + midi = adafruit_midi.MIDI(midi_in=BufPort(BENCH_BYTES), in_channel=0) + results["adafruit_midi"] = run_benchmark(midi, "adafruit_midi") +except ImportError: + print(" adafruit_midi: not installed") + +if "tmidi" in results and "adafruit_midi" in results: + ratio = results["adafruit_midi"] / results["tmidi"] + print(f"tmidi is {ratio:.1f}x faster than adafruit_midi\n") diff --git a/examples/tmidi_simple_arpeggiator.py b/examples/tmidi_simple_arpeggiator.py index 5669105..3857e93 100644 --- a/examples/tmidi_simple_arpeggiator.py +++ b/examples/tmidi_simple_arpeggiator.py @@ -9,10 +9,11 @@ import time import usb_midi import tmidi +from tmidi import NOTE_ON, NOTE_OFF midi = tmidi.MIDI(midi_in=usb_midi.ports[0], midi_out=usb_midi.ports[1]) # if serial midi -# uart = busio.UART(rx=board.RX, tx=board.TX, timeout=0.000) +# uart = busio.UART(rx=board.RX, tx=board.TX, timeout=0.001) # midi = tmidi.MIDI(midi_in=uart, midi_out=uart) tempo = 120 # bpm @@ -27,32 +28,30 @@ while True: # handle midi input if msg := midi.receive(): - if msg.type == tmidi.NOTE_ON and msg.velocity > 0: + if msg.type == NOTE_ON and msg.velocity > 0: print("note on", msg) pressed_notes.append(msg.note) - elif msg.type == tmidi.NOTE_OFF or ( - msg.type == tmidi.NOTE_ON and msg.velocity == 0 - ): + elif msg.type == NOTE_OFF or (msg.type == NOTE_ON and msg.velocity == 0): if msg.note in pressed_notes: midi.send(msg) # send the note off pressed_notes.remove(msg.note) note_i = 0 - # do midi output + # do midi output, if we have pressed notes if len(pressed_notes) == 0: continue now = time.monotonic() if now - last_note_time >= note_time: last_note_time = now - note_on = tmidi.Message(tmidi.NOTE_ON, pressed_notes[note_i], 127) + note_on = tmidi.Message(NOTE_ON, pressed_notes[note_i], 127) print("arp note_on: ", note_on) midi.send(note_on) gate_time = note_time * gate_percent if gate_time > 0 and now - last_note_time >= gate_time: gate_time = 0 - note_off = tmidi.Message(tmidi.NOTE_OFF, pressed_notes[note_i], 127) + note_off = tmidi.Message(NOTE_OFF, pressed_notes[note_i], 127) print("arp note_off:", note_off) midi.send(note_off) note_i = (note_i + 1) % len(pressed_notes) diff --git a/examples/tmidi_simple_receiver.py b/examples/tmidi_simple_receiver.py index 7f5156a..6c7c2f7 100644 --- a/examples/tmidi_simple_receiver.py +++ b/examples/tmidi_simple_receiver.py @@ -12,20 +12,10 @@ while True: if msg := midi_usb.receive(): if msg.type == tmidi.NOTE_ON and msg.velocity > 0: - print( - "note on: note:", - msg.note, - "vel:", - msg.velocity, - "channel:", - msg.channel, - ) - elif msg.type == tmidi.NOTE_OFF or msg.velocity == 0: - print( - "note off: note:", - msg.note, - "vel:", - msg.velocity, - "channel:", - msg.channel, - ) + print(f"note on: note: {msg.note} vel: {msg.velocity} chan: {msg.channel}") + elif msg.type == tmidi.NOTE_OFF or ( + msg.type == tmidi.NOTE_ON and msg.velocity == 0 + ): + print(f"note off: note: {msg.note} vel: {msg.velocity} chan: {msg.channel}") + else: + print("msg:", msg) diff --git a/examples/tmidi_sysex.py b/examples/tmidi_sysex.py new file mode 100644 index 0000000..618c775 --- /dev/null +++ b/examples/tmidi_sysex.py @@ -0,0 +1,17 @@ +# SPDX-FileCopyrightText: Copyright (c) 2024 Tod Kurt +# +# SPDX-License-Identifier: MIT + +""" +# in CPython this looks like: + +import rtmidi, time +midiin = rtmidi.MidiIn(); midiout = rtmidi.MidiOut() +midiin.open_port(0); midiout.open_port(0) +midiin.ignore_types(sysex=False) +midiout.send_message(bytes([0xf0, 0x7e, 0x7f, 0x06, 0x01, 0xf7])) # device inquiry 0x7f means 'any device' +time.sleep(0.01) +resp = midiin.get_message() +print(resp) + +""" diff --git a/examples/tmidi_test_device.py b/examples/tmidi_test_device.py new file mode 100644 index 0000000..76059a5 --- /dev/null +++ b/examples/tmidi_test_device.py @@ -0,0 +1,94 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 Tod Kurt +# +# SPDX-License-Identifier: MIT + +# MIDI echo server for hardware-in-the-loop testing with tmidi_test_host.py. +# +# Receives every MIDI message and echoes it back unchanged, except SysEx +# which is consumed-and-discarded (parser reads until 0xF7 to stay in sync). +# +# Three modes — set MODE below: +# +# "usb_midi" (default): +# Just plug in and run tmidi_test_host.py --usb-midi on the host. +# +# "usb_cdc" (simulates busio.UART byte stream over USB): +# 1. Put this in boot.py on the device: +# import usb_cdc +# usb_cdc.enable(console=True, data=True) +# 2. Set MODE = "usb_cdc" below. +# 3. Run tmidi_test_host.py --serial /dev/cu.usbmodemXXXX on the host. +# (The REPL is on the first CDC port; use the *second* one for MIDI.) +# +# "uart" (hardware UART / TRS MIDI): +# 1. Set MODE = "uart" below. +# 2. Optionally override UART_TX and UART_RX to use different pins. +# 3. Connect a MIDI interface circuit to those pins. +# 4. Run tmidi_test_host.py --serial /dev/cu.usbmodemXXXX on the host +# (or use a hardware MIDI loopback at 31250 baud). + +import board +import busio +import digitalio +import usb_midi + +import tmidi + +MODE = "usb_midi" # "usb_midi" | "usb_cdc" | "uart" + +UART_TX = board.TX # override for non-default pins +UART_RX = board.RX + +# ---- setup ----------------------------------------------------------------- + +try: + led = digitalio.DigitalInOut(board.LED) + led.direction = digitalio.Direction.OUTPUT +except AttributeError: + led = None # board has no LED pin + +if MODE == "usb_midi": + midi = tmidi.MIDI(midi_in=usb_midi.ports[0], midi_out=usb_midi.ports[1]) + print("tmidi_test_device: USB MIDI mode") + +elif MODE == "usb_cdc": + import usb_cdc # noqa: E402 + + midi = tmidi.MIDI(midi_in=usb_cdc.data, midi_out=usb_cdc.data) + print("tmidi_test_device: USB CDC serial mode") + +elif MODE == "uart": + uart = busio.UART(tx=UART_TX, rx=UART_RX, baudrate=31250, timeout=0) + midi = tmidi.MIDI(midi_in=uart, midi_out=uart) + print(f"tmidi_test_device: UART mode (TX={UART_TX} RX={UART_RX})") + +else: + raise ValueError(f"Unknown MODE: {MODE!r} — must be 'usb_midi', 'usb_cdc', or 'uart'") + +print("Echoing MIDI (SysEx consumed but not echoed).") + +# ---- main loop ------------------------------------------------------------- + +msg_count = 0 +last_errors = 0 + +while True: + msg = midi.receive() + if msg is None: + continue + + if led: + led.value = True + + msg_count += 1 + print(msg_count, msg) + + if msg.type != tmidi.SYSEX: + midi.send(msg) + + if led: + led.value = False + + if midi.error_count != last_errors: + last_errors = midi.error_count + print(" parse errors:", last_errors) diff --git a/examples/tmidi_test_host.py b/examples/tmidi_test_host.py new file mode 100644 index 0000000..92fe4d4 --- /dev/null +++ b/examples/tmidi_test_host.py @@ -0,0 +1,304 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2024 Tod Kurt +# +# SPDX-License-Identifier: MIT + +# MIDI hardware-in-the-loop test host. +# +# Sends a battery of MIDI messages to a CircuitPython RP2040 running +# tmidi_test_device.py and verifies the echoed replies match. +# +# Usage: +# python tmidi_test_host.py --usb-midi [--port "CircuitPython Audio"] +# python tmidi_test_host.py --serial /dev/cu.usbmodemXXXX +# python tmidi_test_host.py --serial /dev/ttyACM1 +# +# Dependencies: +# USB MIDI mode: pip install mido python-rtmidi +# Serial mode: pip install pyserial + +import argparse +import sys +import time + +ECHO_TIMEOUT = 2.0 # seconds to wait for each echo + +# --------------------------------------------------------------------------- +# Test cases +# +# Each entry: (description, [send_bytes, ...], expect_bytes) +# send_bytes — list of raw MIDI byte sequences to send in order +# expect_bytes — raw bytes expected as the echo, or None for no echo +# +# Pitch bend encoding (standard MIDI, LSB first): +# center (0) [0xE0, 0x00, 0x40] (14-bit value 0x2000 = 8192) +# full up (+8191) [0xE0, 0x7F, 0x7F] (14-bit value 0x3FFF) +# full down (-8192) [0xE0, 0x00, 0x00] (14-bit value 0x0000) +# --------------------------------------------------------------------------- + +TESTS = [ + ( + "NOTE_ON ch=0 note=60 vel=100", + [bytes([0x90, 60, 100])], + bytes([0x90, 60, 100]), + ), + ( + "NOTE_OFF ch=0 note=60 vel=0", + [bytes([0x80, 60, 0])], + bytes([0x80, 60, 0]), + ), + ( + "CONTROL_CHANGE ch=3 cc=74 val=64", + [bytes([0xB3, 74, 64])], + bytes([0xB3, 74, 64]), + ), + ( + "PROGRAM_CHANGE ch=5 prog=42", + [bytes([0xC5, 42])], + bytes([0xC5, 42]), + ), + ( + "PITCH_BEND ch=0 center (0)", + [bytes([0xE0, 0x00, 0x40])], + bytes([0xE0, 0x00, 0x40]), + ), + ( + "PITCH_BEND ch=0 full up (+8191)", + [bytes([0xE0, 0x7F, 0x7F])], + bytes([0xE0, 0x7F, 0x7F]), + ), + ( + "PITCH_BEND ch=0 full down (-8192)", + [bytes([0xE0, 0x00, 0x00])], + bytes([0xE0, 0x00, 0x00]), + ), + ( + "CLOCK (single-byte real-time)", + [bytes([0xF8])], + bytes([0xF8]), + ), + ( + "SYSEX desync: NOTE_ON must survive a SysEx burst (ch=2, note=85, vel=99)", + [ + bytes([0xF0, 0x7E, 0x7F, 0x06, 0x01, 0xF7]), # SysEx — consumed, not echoed + bytes([0x92, 85, 99]), # NOTE_ON — must be echoed cleanly + ], + bytes([0x92, 85, 99]), + ), +] + + +# --------------------------------------------------------------------------- +# USB MIDI transport — uses mido + python-rtmidi +# --------------------------------------------------------------------------- + + +class MidiTransportUSB: + def __init__(self, port_name=None): + try: + import mido + except ImportError: + sys.exit("ERROR: mido not installed. Run: pip install mido python-rtmidi") + self._mido = mido + + out_names = mido.get_output_names() + in_names = mido.get_input_names() + + if not out_names: + sys.exit("ERROR: No MIDI output ports found. Is the device plugged in?") + + if port_name is None: + keywords = ("circuit", "pico", "tmidi") + for kw in keywords: + hits = [n for n in out_names if kw in n.lower()] + if hits: + port_name = hits[0] + break + if port_name is None: + print("Available MIDI output ports:") + for n in out_names: + print(" ", n) + sys.exit("Use --port to specify one.") + + # Match the input port by name (may have a numeric suffix on some OSes) + in_name = next( + (n for n in in_names if port_name in n or n in port_name), + in_names[0] if in_names else port_name, + ) + + print("Output ->", port_name) + print("Input <-", in_name) + self._out = mido.open_output(port_name) + self._inp = mido.open_input(in_name) + + def send(self, raw_bytes): + for msg in self._mido.parse_all(raw_bytes): + self._out.send(msg) + + def receive(self, timeout=ECHO_TIMEOUT): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + msg = self._inp.receive(block=False) + if msg is not None: + return bytes(msg.bytes()) + time.sleep(0.001) + return None + + def close(self): + self._out.close() + self._inp.close() + + +# --------------------------------------------------------------------------- +# Serial transport — uses pyserial (USB CDC or physical UART) +# --------------------------------------------------------------------------- + + +class MidiTransportSerial: + # Data byte counts for parsing the raw MIDI byte stream on the host. + _DATA_LEN = { + 0x80: 2, + 0x90: 2, + 0xA0: 2, + 0xB0: 2, + 0xC0: 1, + 0xD0: 1, + 0xE0: 2, + 0xF2: 2, + 0xF3: 1, + 0xF5: 1, + } + + def __init__(self, port_path): + try: + import serial + except ImportError: + sys.exit("ERROR: pyserial not installed. Run: pip install pyserial") + # Baud rate is nominal for USB CDC (USB ignores it); use 31250 for + # a physical UART connected to a hardware MIDI port. + self._ser = serial.Serial(port_path, 115200, timeout=0.05) + print("Serial port:", port_path) + time.sleep(0.5) # let the device settle after the port opens + self._ser.reset_input_buffer() + + def send(self, raw_bytes): + self._ser.write(raw_bytes) + + def receive(self, timeout=ECHO_TIMEOUT): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + b = self._ser.read(1) + if not b: + continue + status = b[0] + if not (status & 0x80): + continue # skip stray data bytes + msg_type = status & 0xF0 if status < 0xF0 else status + data_len = self._DATA_LEN.get(msg_type, 0) + data = self._ser.read(data_len) if data_len else b"" + return bytes([status]) + data + return None + + def close(self): + self._ser.close() + + +# --------------------------------------------------------------------------- +# Test runner +# --------------------------------------------------------------------------- + + +def run_tests(transport): + passed = 0 + failed = 0 + + for i, (name, send_seqs, expect) in enumerate(TESTS, 1): + print("[%d/%d] %s ... " % (i, len(TESTS), name), end="", flush=True) + t0 = time.monotonic() + + for raw in send_seqs: + transport.send(raw) + time.sleep(0.01) # small gap so burst messages don't merge + + if expect is None: + time.sleep(0.05) + print("OK (no echo)") + passed += 1 + continue + + reply = transport.receive() + elapsed_ms = (time.monotonic() - t0) * 1000 + + if reply is None: + print("FAIL (timeout after %.1fs)" % ECHO_TIMEOUT) + failed += 1 + elif bytes(reply) == bytes(expect): + print("PASS (%.0f ms)" % elapsed_ms) + passed += 1 + else: + exp_hex = " ".join("%02X" % b for b in expect) + got_hex = " ".join("%02X" % b for b in reply) + print("FAIL (expected %s, got %s)" % (exp_hex, got_hex)) + failed += 1 + + time.sleep(0.02) # inter-test gap + + print() + result = "PASSED" if failed == 0 else "FAILED" + print("%s: %d/%d tests" % (result, passed, len(TESTS))) + return failed == 0 + + +# --------------------------------------------------------------------------- +# Entry point +# --------------------------------------------------------------------------- + + +def main(): + parser = argparse.ArgumentParser( + description="tmidi hardware-in-the-loop test host", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +examples: + %(prog)s --usb-midi + %(prog)s --usb-midi --port "CircuitPython Audio" + %(prog)s --serial /dev/cu.usbmodemXXXX (macOS) + %(prog)s --serial /dev/ttyACM1 (Linux) +""", + ) + mode = parser.add_mutually_exclusive_group(required=True) + mode.add_argument( + "--usb-midi", + action="store_true", + help="test via USB MIDI (requires mido + python-rtmidi)", + ) + mode.add_argument( + "--serial", metavar="PORT", help="test via serial port (requires pyserial)" + ) + parser.add_argument( + "--port", metavar="NAME", help="MIDI output port name (--usb-midi only)" + ) + args = parser.parse_args() + + print("tmidi hardware-in-the-loop test") + print("=" * 40) + + if args.usb_midi: + transport = MidiTransportUSB(port_name=args.port) + else: + transport = MidiTransportSerial(args.serial) + + print() + try: + ok = run_tests(transport) + except KeyboardInterrupt: + print("\nAborted.") + ok = False + finally: + transport.close() + + sys.exit(0 if ok else 1) + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml index 17dbc0f..7494eb3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -46,3 +46,6 @@ py-modules = ["tmidi"] [tool.setuptools.dynamic] dependencies = {file = ["requirements.txt"]} optional-dependencies = {optional = {file = ["optional_requirements.txt"]}} + +[tool.ruff] +line-length = 90 diff --git a/tests/test_tmidi_messages.py b/tests/test_tmidi_messages.py index 61f7bfd..63d404f 100644 --- a/tests/test_tmidi_messages.py +++ b/tests/test_tmidi_messages.py @@ -1,4 +1,4 @@ -# SPDX-FileCopyrightText: Copyright (c) 2024 Tod Kurt +# SPDX-FileCopyrightText: Copyright (c) 2026 Tod Kurt # SPDX-License-Identifier: MIT @@ -44,3 +44,34 @@ def test_message_program_change_send(): msg = tmidi.Message(tmidi.PROGRAM_CHANGE, 33) port.expected = [0xC0, 33] midi_out.send(msg) + + +def test_send_channel_zero(): + # channel=0 must not be treated as falsy and ignored + port = PortStub(iter([])) + midi_out = tmidi.MIDI(midi_out=port) + + msg = tmidi.Message(tmidi.NOTE_ON, 60, 100, channel=5) + port.expected = [0x90, 60, 100] # channel overridden to 0 + midi_out.send(msg, channel=0) + + +def test_send_clock(): + port = PortStub(iter([])) + midi_out = tmidi.MIDI(midi_out=port) + + msg = tmidi.Message(tmidi.CLOCK) + port.expected = [0xF8] + midi_out.send(msg) + + +def test_send_batch(): + port = PortStub(iter([])) + midi_out = tmidi.MIDI(midi_out=port) + + msgs = [ + tmidi.Message(tmidi.NOTE_ON, 60, 100), + tmidi.Message(tmidi.NOTE_OFF, 60, 0), + ] + port.expected = [0x90, 60, 100, 0x80, 60, 0] + midi_out.send(msgs) diff --git a/tests/test_tmidi_one.py b/tests/test_tmidi_one.py index 886067e..5e994bc 100644 --- a/tests/test_tmidi_one.py +++ b/tests/test_tmidi_one.py @@ -62,3 +62,93 @@ def test_midi_in_invalid_data(): assert msg is None assert midi_in.error_count == 1 + + +def test_note_on_receive(): + port = PortStub(iter([0x90, 60, 100])) + midi_in = tmidi.MIDI(midi_in=port) + + msg = midi_in.receive() + + assert msg is not None + assert msg.type == tmidi.NOTE_ON + assert msg.channel == 0 + assert msg.note == 60 + assert msg.velocity == 100 + assert midi_in.error_count == 0 + + +def test_note_on_receive_roundtrip(): + port = PortStub(iter([0x93, 60, 100])) + midi_in = tmidi.MIDI(midi_in=port) + + msg = midi_in.receive() + + assert msg is not None + assert bytes(msg.__bytes__()) == bytes([0x93, 60, 100]) + + +def test_pitch_bend_receive(): + # Center: wire [0xE0, 0x00, 0x40] → pitch_bend = 0 + port = PortStub(iter([0xE0, 0x00, 0x40])) + midi_in = tmidi.MIDI(midi_in=port) + msg = midi_in.receive() + assert msg is not None + assert msg.type == tmidi.PITCH_BEND + assert msg.pitch_bend == 0 + assert list(msg.__bytes__()) == [0xE0, 0x00, 0x40] + + +def test_pitch_bend_full_down(): + # Full down: wire [0xE0, 0x00, 0x00] → pitch_bend = -8192 + # data1=0x00 must not trigger the __init__ user-API special case + port = PortStub(iter([0xE0, 0x00, 0x00])) + midi_in = tmidi.MIDI(midi_in=port) + msg = midi_in.receive() + assert msg is not None + assert msg.type == tmidi.PITCH_BEND + assert msg.pitch_bend == -8192 + assert list(msg.__bytes__()) == [0xE0, 0x00, 0x00] + + +def test_pitch_bend_full_up(): + # Full up: wire [0xE0, 0x7F, 0x7F] → pitch_bend = 8191 + port = PortStub(iter([0xE0, 0x7F, 0x7F])) + midi_in = tmidi.MIDI(midi_in=port) + msg = midi_in.receive() + assert msg is not None + assert msg.type == tmidi.PITCH_BEND + assert msg.pitch_bend == 8191 + assert list(msg.__bytes__()) == [0xE0, 0x7F, 0x7F] + + +def test_data_byte_corruption(): + # Status byte appearing in data position should trigger error + port = PortStub(iter([0x90, 0x90, 60])) + midi_in = tmidi.MIDI(midi_in=port) + + msg = midi_in.receive() + + assert msg is None + assert midi_in.error_count == 1 + + +def test_sysex_receive_no_desync(): + # SysEx followed by Note On: parser must consume the SysEx payload + # so the Note On is parsed cleanly with no error count. + sysex = [0xF0, 0x7E, 0x7F, 0x06, 0x01, 0xF7] + note_on = [0x91, 60, 127] + port = PortStub(iter(sysex + note_on)) + midi_in = tmidi.MIDI(midi_in=port) + + msg1 = midi_in.receive() + assert msg1 is not None + assert msg1.type == tmidi.SYSEX + + msg2 = midi_in.receive() + assert msg2 is not None + assert msg2.type == tmidi.NOTE_ON + assert msg2.channel == 1 + assert msg2.note == 60 + assert msg2.velocity == 127 + assert midi_in.error_count == 0 diff --git a/tests/test_tmidi_sysex.py b/tests/test_tmidi_sysex.py new file mode 100644 index 0000000..f5739fd --- /dev/null +++ b/tests/test_tmidi_sysex.py @@ -0,0 +1,132 @@ +# SPDX-FileCopyrightText: Copyright (c) 2024 Tod Kurt +# SPDX-License-Identifier: MIT + +import tmidi + + +class PortStub: + def __init__(self, data): + self.data = data + + def readinto(self, buf, numbytes=1): + bytes_read = 0 + for n in range(numbytes): + try: + value = next(self.data) + buf[n] = value + bytes_read += 1 + except StopIteration: + break + return bytes_read + + +def test_sysex_empty(): + # Bare F0 F7 with no payload + port = PortStub(iter([0xF0, 0xF7])) + midi_in = tmidi.MIDI(midi_in=port) + msg = midi_in.receive() + assert msg is not None + assert msg.type == tmidi.SYSEX + assert midi_in.error_count == 0 + + +def test_sysex_with_payload(): + # Standard device inquiry SysEx + port = PortStub(iter([0xF0, 0x7E, 0x7F, 0x06, 0x01, 0xF7])) + midi_in = tmidi.MIDI(midi_in=port) + msg = midi_in.receive() + assert msg is not None + assert msg.type == tmidi.SYSEX + assert midi_in.error_count == 0 + + +def test_sysex_long_payload(): + # 64-byte payload — parser must consume all of it + payload = list(range(64)) # 0x00–0x3F, all valid data bytes + port = PortStub(iter([0xF0] + payload + [0xF7])) + midi_in = tmidi.MIDI(midi_in=port) + msg = midi_in.receive() + assert msg is not None + assert msg.type == tmidi.SYSEX + assert midi_in.error_count == 0 + + +def test_sysex_consecutive(): + # Two SysEx messages back to back, then a CC — all three must parse cleanly + sysex1 = [0xF0, 0x41, 0x10, 0x42, 0xF7] + sysex2 = [0xF0, 0x7E, 0x7F, 0x06, 0x01, 0xF7] + cc = [0xB2, 74, 64] + port = PortStub(iter(sysex1 + sysex2 + cc)) + midi_in = tmidi.MIDI(midi_in=port) + + msg1 = midi_in.receive() + assert msg1 is not None and msg1.type == tmidi.SYSEX + + msg2 = midi_in.receive() + assert msg2 is not None and msg2.type == tmidi.SYSEX + + msg3 = midi_in.receive() + assert msg3 is not None + assert msg3.type == tmidi.CC + assert msg3.channel == 2 + assert msg3.data0 == 74 + assert msg3.data1 == 64 + assert midi_in.error_count == 0 + + +def test_sysex_buffer_captures_payload(): + payload = [0x7E, 0x7F, 0x06, 0x01] + port = PortStub(iter([0xF0] + payload + [0xF7])) + sysex_buf = bytearray(16) + midi_in = tmidi.MIDI(midi_in=port, sysex_buffer=sysex_buf) + msg = midi_in.receive() + assert msg is not None + assert msg.type == tmidi.SYSEX + assert msg.data0 == len(payload) + assert list(sysex_buf[: msg.data0]) == payload + assert midi_in.error_count == 0 + + +def test_sysex_buffer_truncates_when_full(): + # Payload longer than buffer — stream must stay in sync, no hang. + payload = list(range(32)) + port = PortStub(iter([0xF0] + payload + [0xF7, 0x90, 60, 100])) + sysex_buf = bytearray(8) + midi_in = tmidi.MIDI(midi_in=port, sysex_buffer=sysex_buf) + msg1 = midi_in.receive() + assert msg1 is not None and msg1.type == tmidi.SYSEX + assert msg1.data0 == 8 # buffer filled to capacity + assert list(sysex_buf) == payload[:8] + msg2 = midi_in.receive() + assert msg2 is not None and msg2.type == tmidi.NOTE_ON + assert midi_in.error_count == 0 + + +def test_sysex_no_buffer_data0_is_zero(): + port = PortStub(iter([0xF0, 0x7E, 0x7F, 0xF7])) + midi_in = tmidi.MIDI(midi_in=port) + msg = midi_in.receive() + assert msg is not None and msg.type == tmidi.SYSEX + assert msg.data0 == 0 + + +def test_sysex_does_not_set_running_status(): + # SysEx is a system message and must not update running status. + # A data byte after SysEx with no new status should be an error, + # not silently reuse a pre-SysEx running status. + note_on = [0x90, 60, 100] + sysex = [0xF0, 0x01, 0x02, 0xF7] + bare_data = [0x40] # data byte with no status — invalid on its own + port = PortStub(iter(note_on + sysex + bare_data)) + midi_in = tmidi.MIDI(midi_in=port, enable_running_status=True) + + msg1 = midi_in.receive() + assert msg1 is not None and msg1.type == tmidi.NOTE_ON # sets running status + + msg2 = midi_in.receive() + assert msg2 is not None and msg2.type == tmidi.SYSEX # must NOT update running status + + msg3 = midi_in.receive() + # bare_data after SysEx: running status from before SysEx should NOT apply + assert msg3 is None + assert midi_in.error_count == 1 diff --git a/tmidi.py b/tmidi.py index 75f920d..496ac58 100644 --- a/tmidi.py +++ b/tmidi.py @@ -1,5 +1,5 @@ # SPDX-FileCopyrightText: Copyright (c) 2019 Alethea Flowers for Winterbloom -# SPDX-FileCopyrightText: Copyright (c) 2024 Tod Kurt +# SPDX-FileCopyrightText: Copyright (c) 2026 Tod Kurt # # SPDX-License-Identifier: MIT """ @@ -103,22 +103,22 @@ SYSTEM_RESET = const(0xFF) """System Reset""" -_LEN_0_MESSAGES = set( - [ - TUNE_REQUEST, - SYSEX, - SYSEX_END, - CLOCK, - TICK, - START, - CONTINUE, - STOP, - ACTIVE_SENSING, - SYSTEM_RESET, - ] -) -_LEN_1_MESSAGES = set([PROGRAM_CHANGE, CHANNEL_PRESSURE, SONG_SELECT, BUS_SELECT]) -_LEN_2_MESSAGES = set([NOTE_OFF, NOTE_ON, AFTERTOUCH, CC, PITCH_BEND, SONG_POSITION]) +_MSG_DATA_LEN = { + NOTE_OFF: 2, + NOTE_ON: 2, + AFTERTOUCH: 2, + CC: 2, + PITCH_BEND: 2, + SONG_POSITION: 2, + PROGRAM_CHANGE: 1, + CHANNEL_PRESSURE: 1, + SONG_SELECT: 1, + BUS_SELECT: 1, +} + +# Pre-allocated scratch buffer for __bytes__() — avoids per-call list allocation. +# Safe because CircuitPython is single-threaded. +_msg_buf = bytearray(3) _MSG_TYPE_NAMES = { NOTE_OFF: "NoteOff", @@ -148,26 +148,6 @@ def _is_channel_message(status_byte): return status_byte >= NOTE_OFF and status_byte < SYSEX -def _read_byte(port): - while not (buf := port.read(1)): - pass - return buf[0] - - -# def _read_n_bytes(port, buf, dest, num_bytes): -# while num_bytes: -# if port.readinto(buf): -# dest.append(buf[0]) -# num_bytes -= 1 - - -# def _read_byte_works(port): -# while True: -# buf = port.read(1) -# if buf: -# return buf[0] - - class Message: """ MIDI Message. @@ -208,18 +188,23 @@ def __init__(self, mtype=SYSTEM_RESET, data0=0, data1=0, channel=0): self.channel = channel self.data0 = data0 self.data1 = data1 - if mtype == PITCH_BEND and data1 == 0: + if mtype == PITCH_BEND and data1 == 0 and not (0 <= data0 <= 0x7F): self.pitch_bend = data0 def __bytes__(self): status_byte = self.type if _is_channel_message(status_byte): status_byte |= self.channel - if self.type in _LEN_2_MESSAGES: - return bytes([status_byte, self.data0, self.data1]) - elif self.type in _LEN_1_MESSAGES: - return bytes([status_byte, self.data0]) - return bytes([status_byte]) + _msg_buf[0] = status_byte + n = _MSG_DATA_LEN.get(self.type, 0) + if n == 2: + _msg_buf[1] = self.data0 + _msg_buf[2] = self.data1 + return bytes(_msg_buf) + if n == 1: + _msg_buf[1] = self.data0 + return bytes(_msg_buf[:2]) + return bytes(_msg_buf[:1]) def __repr__(self): return self.__str__() @@ -230,9 +215,14 @@ def __str__(self): ch_str = "ch:%d" % self.channel if _is_channel_message(mtype) else "-" if mtype == PITCH_BEND: return "%s %s %d)" % (type_str, ch_str, self.pitch_bend) - if mtype in _LEN_2_MESSAGES: + if mtype == SYSEX: + if self.data0: + return "%s %d bytes)" % (type_str, self.data0) + return "%s)" % type_str + n = _MSG_DATA_LEN.get(mtype, 0) + if n == 2: return "%s %s %d %d)" % (type_str, ch_str, self.data0, self.data1) - if mtype in _LEN_1_MESSAGES: + if n == 1: return "%s %s %d)" % (type_str, ch_str, self.data0) return "%s)" % type_str @@ -285,6 +275,27 @@ class MIDI: :param midi_out: an object which implements ``write(buffer, length)``, set to ``usb_midi.ports[1]`` for USB MIDI, default None. :param bool enable_running_status: Allow running status messages to work, default False. + :param bytearray sysex_buffer: Optional pre-allocated buffer for capturing SysEx payloads. + When provided, received SysEx payload bytes are written into this buffer and + ``msg.data0`` on the returned message holds the number of bytes written. + If the payload exceeds the buffer length, the overflow is consumed and discarded + so the stream stays in sync. When ``None`` (default), SysEx payloads are discarded + and ``msg.data0`` is 0. + + Example of receiving SysEx with payload capture: + + .. code-block:: python + + import usb_midi + import tmidi + sysex_buf = bytearray(128) + midi = tmidi.MIDI(midi_in=usb_midi.ports[0], sysex_buffer=sysex_buf) + + while True: + if msg := midi.receive(): + if msg.type == tmidi.SYSEX: + payload = memoryview(sysex_buf)[: msg.data0] + print("SysEx:", list(payload)) Example of sending MIDI over USB: @@ -327,12 +338,15 @@ class MIDI: print("uart midi:", msg) """ - def __init__(self, midi_in=None, midi_out=None, enable_running_status=False): + def __init__( + self, midi_in=None, midi_out=None, enable_running_status=False, sysex_buffer=None + ): self._in_port = midi_in self._out_port = midi_out self._running_status_enabled = enable_running_status self._running_status = None self._error_count = 0 + self._sysex_buf = sysex_buffer # This input buffer holds what has been read from midi_in self._read_buf = bytearray(1) @@ -350,55 +364,72 @@ def receive(self): :returns Message object: Returns object or None for nothing. """ + in_port = self._in_port + read_buf = self._read_buf - # Read the status byte for the next message. - # note: this will block if the port is set to have a timeout - status_byte_buf = self._in_port.read(1) - - # No message ready. - if not status_byte_buf: + # Non-blocking: return None if no byte is waiting. + if not in_port.readinto(read_buf): return None - # Is this actually a status byte? - status_byte = status_byte_buf[0] - is_status = status_byte & 0x80 + status_byte = read_buf[0] - # If not, see if we have a running status byte. - if not is_status: + # If not a status byte, try running status, otherwise discard. + if not (status_byte & 0x80): if self._running_status_enabled and self._running_status: status_byte = self._running_status - # If not a status byte and no running status, this is invalid data. else: self._error_count += 1 return None - message = Message(status_byte) + msg_type = status_byte + msg_channel = 0 - # Is this a channel message, if so, let's figure out the right - # message type and set the message's channel property. if _is_channel_message(status_byte): - # Only set the running status byte for channel messages. self._running_status = status_byte - # Mask off the channel nibble. - message.type = status_byte & 0xF0 - message.channel = status_byte & 0x0F - - # Read the appropriate number of bytes for each message type. - if message.type in _LEN_2_MESSAGES: - message.data0 = _read_byte(self._in_port) - message.data1 = _read_byte(self._in_port) - elif message.type in _LEN_1_MESSAGES: - message.data0 = _read_byte(self._in_port) - - # Check the data bytes for corruption. status bytes in data - # means we're out of sync, so discard. - # TODO: Figure out a better way to detect and deal with this upstream. - for b in (message.data0 or 0, message.data1 or 0): - if b & 0x80: + msg_type = status_byte & 0xF0 + msg_channel = status_byte & 0x0F + + # Consume SysEx payload byte-by-byte until the terminator so the + # stream stays in sync. SysEx cancels running status per the MIDI spec. + # If sysex_buffer was provided, payload bytes are written into it; + # the returned message's data0 holds the number of bytes written. + if msg_type == SYSEX: + self._running_status = None + sysex_buf = self._sysex_buf + sysex_len = 0 + while True: + while not in_port.readinto(read_buf): + pass + b = read_buf[0] + if b == SYSEX_END: + break + if sysex_buf is not None and sysex_len < len(sysex_buf): + sysex_buf[sysex_len] = b + sysex_len += 1 + return Message(SYSEX, sysex_len) + + data_len = _MSG_DATA_LEN.get(msg_type, 0) + data0 = 0 + data1 = 0 + + if data_len >= 1: + while not in_port.readinto(read_buf): + pass + data0 = read_buf[0] + # A status byte appearing where data is expected means we're out of sync. + if data0 & 0x80: + self._error_count += 1 + return None + + if data_len >= 2: + while not in_port.readinto(read_buf): + pass + data1 = read_buf[0] + if data1 & 0x80: self._error_count += 1 return None - return message + return Message(msg_type, data0, data1, msg_channel) def send(self, msg, channel=None): """Send a MIDI message. @@ -409,14 +440,14 @@ def send(self, msg, channel=None): """ if isinstance(msg, Message): - if channel: + if channel is not None: msg.channel = channel # bytes(object) does not work in uPy data = msg.__bytes__() else: data = bytearray() for each_msg in msg: - if channel: + if channel is not None: each_msg.channel = channel data.extend(each_msg.__bytes__())