1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234

# SPDX-License-Identifier: MIT
# SPDX-FileCopyrightText: Copyright 2025 Sam Blenny
#
# Related Docs:
# - https://docs.circuitpython.org/projects/tlv320/en/latest/api.html
# - https://learn.adafruit.com/adafruit-tlv320dac3100-i2s-dac/overview
# - https://docs.circuitpython.org/en/latest/shared-bindings/audiobusio/
# - https://docs.circuitpython.org/en/latest/shared-bindings/audiomixer/
# - https://midi.org/specs
# - https://github.com/todbot/circuitpython-synthio-tricks
#
from audiobusio import I2SOut
from board import (
    I2C, I2S_BCLK, I2S_DIN, I2S_MCLK, I2S_WS, PERIPH_RESET
)
from digitalio import DigitalInOut, Direction, Pull
import gc
from micropython import const
import synthio
import sys
from time import sleep
from usb.core import USBError, USBTimeoutError
import usb_host

from adafruit_tlv320 import TLV320DAC3100

from sb_usb_midi import find_usb_device, MIDIInputDevice


# DAC and Synthesis parameters
SAMPLE_RATE = const(11025)
CHAN_COUNT  = const(2)
BUFFER_SIZE = const(1024)
#==============================================================
# CAUTION! When this is set to True, the headphone jack will
# send a line-level output suitable use with a mixer or powered
# speakers, but that will be _way_ too loud for earbuds. For
# finer control of line level volume, adjust LL_DAC_VOLUME.
LINE_LEVEL = const(True)
LL_HEADPHONE_VOLUME = -6.0
#==============================================================

# Change this to True if you want more MIDI output on the serial console
DEBUG = True

# Adjust this to balance the probability of dropped/stuck notes against the
# amount of time you're willing to let event loop spend on blocking IO to wait
# for a USB host read() call to finish. Note that background tasks like synthio
# should work fine while read() is blocked. In my testing, short timeouts like
# 3 or 33 ms would result in some dropped or stuck notes. But, I didn't get
# any of that with longer timeouts like 50, 100, or 300 ms.
#
READ_TIMEOUT_MS = const(50)

# Constants for parsing MIDI messages
CIN_NOTE_OFF = const(0x08)  # Note Off
CIN_NOTE_ON  = const(0x09)  # Note On
CIN_CC       = const(0x0b)  # Control Change
CIN_MPE      = const(0x0a)  # Polyphonic Expression (individual key pressure)
CIN_CP       = const(0x0d)  # Channel Pressure (grouped key pressure)
CIN_PB       = const(0x0e)  # Pitch Bend


def init_dac_audio_synth(i2c):
    # Configure Fruit Jam rev D TLV320 I2S DAC and make a Synthesizer.
    # - i2c: a reference to board.I2C()
    # - returns tuple: (dac: TLV320DAC3100, audio: I2SOut, synth: Synthesizer)

    # 1. Reset DAC (reset is active low)
    rst = DigitalInOut(PERIPH_RESET)
    rst.direction = Direction.OUTPUT
    rst.value = False
    sleep(0.1)
    rst.value = True
    sleep(0.05)
    # 2. Configure sample rate, bit depth, and output port
    dac = TLV320DAC3100(i2c)
    dac.configure_clocks(sample_rate=SAMPLE_RATE, bit_depth=16)
    dac.speaker_output = False
    dac.headphone_output = True
    # 3. Adjust volume for for line-level if needed, otherwise use default
    #    volume set by `dac.headphone_output = True`
    if LINE_LEVEL:
        # This gives a line output level suitable for plugging into a mixer or
        # the AUX input of a powered speaker (THIS IS TOO LOUD FOR HEADPHONES!)
        dac.headphone_volume = LL_HEADPHONE_VOLUME
    if DEBUG:
        print(f"dac.dac_volume = {dac.dac_volume:.1f}")
        print(f"dac.headphone_volume = {dac.headphone_volume:.1f}")
    # 4. Configure I2S for Fruit Jam rev D (rev B swapped WS and MCLK)
    audio = I2SOut(bit_clock=I2S_BCLK, word_select=I2S_WS, data=I2S_DIN)
    # 5. Configure synthio patch to generate audio
    vca = synthio.Envelope(
        attack_time=0.002, decay_time=0.01, sustain_level=0.4,
        release_time=0, attack_level=0.6
    )
    synth = synthio.Synthesizer(
        sample_rate=SAMPLE_RATE, channel_count=CHAN_COUNT, envelope=vca
    )
    audio.play(synth)
    return (dac, audio, synth)


def main():
    gc.collect()

    # Set up the audio stuff for a basic synthesizer
    i2c = I2C()
    (dac, audio, synth) = init_dac_audio_synth(i2c)

    # Cache function references (MicroPython performance boost trick)
    wr = sys.stdout.write
    panic = synth.release_all
    press = synth.press
    release = synth.release

    # Dictionary to keep track of which notes are active.
    # This is useful for debug printing to watch for stuck notes
    notes = {}

    # Note On helper function with closure for notes dictionary
    def note_off(num):
        release(num)
        if num in notes:
            notes.pop(num)

    # Note Off helper function with closure for notes dictionary
    def note_on(num):
        press(num)
        notes[num] = True

    # Main loop: scan USB host bus for MIDI device, connect, start event loop.
    # This grabs the first MIDI device it finds.
    while True:
        wr("USB Host: scanning bus...\n")
        gc.collect()
        device_cache = {}
        try:
            # This loop will end as soon as it finds a ScanResult object (r)
            r = None
            while r is None:
                sleep(0.4)
                r = find_usb_device(device_cache)
            # Use ScanResult object (r) to check if USB device descriptor info
            # matches the class/subclass/protocol pattern for a MIDI device. If
            # the device doesn't match, MIDIInputDevice will raise an exception
            # and trigger another iteration through the outer while True loop.
            dev = MIDIInputDevice(r, read_timeout=READ_TIMEOUT_MS)
            wr(" found MIDI device vid:pid %04X:%04X\n" % (r.vid, r.pid))
            # Collect garbage to hopefully limit heap fragmentation.
            r = None
            device_cache = {}
            gc.collect()
            # MIDI Event Input Loop: Poll for input until USB error.
            cin = chan = num = val = 0x00
            for data in dev.input_event_generator():
                # Begin handling midi packet which should be None or a 4-byte
                # memoryview.
                if data is None:
                    continue

                # data[0] has CN (Cable Number) and CIN (Code Index Number). By
                # discarding CN with `& 0x0f`, we ignore the virtual MIDI port
                # that the messages arrive from. Ignoring CN would be bad for a
                # fancy DAW or synth setup where you needed to route MIDI
                # among multiple devices. But, that doesn't matter here. We do
                # need CIN to distinguish between note on, note off, Control
                # Change (CC), and so on. For the channel, adding 1 gives us
                # human-friendly channel numbers in the range of 1 to 16.
                #
                # NOTE: As far as I can tell from reading the USB MIDI specs,
                # each MIDI packet will always be 32-bits long and CIN will
                # always be set. That means there's no need to worry about
                # parsing "running status" as would be the case when using UART
                # MIDI with DIN-5 or TRS cables.
                #
                cin = data[0] & 0x0f
                chan = (data[1] & 0xf) + 1
                num = data[2]
                val = data[3]

                # This decodes MIDI events by comparing constants against bytes
                # from a memoryview. Using a class to do this parsing would add
                # many extra heap allocations and dictionary lookups. That
                # stuff is slow, and we want to go _fast_. For details about
                # the MIDI 1.0 standard, see https://midi.org/specs
                #
                if cin == CIN_NOTE_OFF and (21 <= num <= 108):
                    # Note off
                    note_off(num)
                elif cin == CIN_NOTE_ON and (21 <= num <= 108):
                    if val == 0:
                        # Some devices send zero velocity instead of note off
                        note_off(num)
                    else:
                        # Note on
                        note_on(num)
                elif cin == CIN_CC and num == 123 and val == 0:
                    # CC 123 means stop all notes ("panic")
                    panic()
                    wr('PANIC %d %d %d\n' % (chan, num, val))
                if DEBUG:
                    if cin == CIN_NOTE_OFF:
                        # Note On
                        # Debug print this message + list of active notes
                        n = ' '.join(sorted([str(n) for n in notes.keys()]))
                        wr('Off %d %d %3d  notes: %s\n' % (chan, num, val, n))
                    elif cin == CIN_NOTE_ON:
                        # Note Off
                        # Debug print this message + list of active notes
                        n = ' '.join(sorted([str(n) for n in notes.keys()]))
                        wr('On  %d %d %3d  notes: %s\n' % (chan, num, val, n))
                    elif cin == CIN_CC:
                        # CC control change
                        wr('CC  %d %d %d\n' % (chan, num, val))
                    elif cin == CIN_MPE:
                        # MPE polyphonic key pressure (aftertouch)
                        wr('MPE %d %d %d\n' % (chan, num, val))
                    elif cin == CIN_CP:
                        # CP channel key pressure (aftertouch)
                        wr('CP  %d %d %d\n' % (chan, num, val))
                    elif cin == CIN_PB:
                        # PB pitch bend
                        wr('PB  %d %d %d\n' % (chan, num, val))
                    # Ignore the rest: SysEx, System Realtime, or whatever
        except USBError as e:
            # This sometimes happens when devices are unplugged. Not always.
            print("USBError: '%s' (device unplugged?)" % e)
        except ValueError as e:
            # This can happen if an initialization handshake glitches
            print(e)


main()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

# SPDX-License-Identifier: MIT
# SPDX-FileCopyrightText: Copyright 2025 Sam Blenny
#
# Driver for USB MIDI devices.
#
# NOTE: USB MIDI is CPU intensive. To help keep latency low, this code uses
# performance boosting tricks with a special focus on limiting the amount of
# heap allocations. Related docs:
# - https://docs.python.org/3/glossary.html#term-generator
# - https://docs.python.org/3/glossary.html#term-iterable
# - https://docs.micropython.org/en/latest/reference/speed_python.html
#
import gc
from micropython import const
from usb import core
from usb.core import USBError, USBTimeoutError

import sb_usb_descriptor


def find_usb_device(device_cache):
    # Find a usb midi device by inspecting usb device descriptors
    # - device_cache: dictionary of previously checked device descriptors
    # - return: ScanResult object for success or None for failure.
    # Exceptions: may raise usb.core.USBError or usb.core.USBTimeoutError
    #
    for device in core.find(find_all=True):
        # Read descriptors to identify devices by type
        try:
            desc = sb_usb_descriptor.Descriptor(device)
            k = str(desc.to_bytes())
            if k in device_cache:
                return None
            # Remember this device to avoid repeatedly checking it later
            device_cache[k] = True
            # Compare descriptor to expected midi device fingerprint
            desc.read_configuration(device)
            # Get tuples of class/subclass/protocol for device and interfaces
            d = desc.dev_class_subclass()
            i0 = desc.int_class_subclass(0)
            i1 = desc.int_class_subclass(1)
            if d == (0, 0) and i0 == (1, 1) and i1 == (1, 3):
                # Reaching here means device's interface 0 is Audio Control and
                # its interface 1 is MIDI Streaming. That's the fingerprint for
                # a class compliant MIDI device.
                return ScanResult(device, desc)
            else:
                print(" IGNORING NON-MIDI USB DEVICE")
                return None
        except ValueError as e:
            # This can happen if we get a 0 length device descriptor. Usually
            # it works fine to ignore the error and try again.
            print(e)
        except USBError as e:
            print(" find_usb_device() USBError: '%s'" % e)
    return None


class ScanResult:
    def __init__(self, device, descriptor):
        self.device = device
        self.descriptor = descriptor
        self.vid = descriptor.idVendor
        self.pid = descriptor.idProduct
        self.dev_info = descriptor.dev_class_subclass()
        self.int0_info = descriptor.int_class_subclass(0)
        self.int1_info = descriptor.int_class_subclass(1)


class MIDIInputDevice:
    def __init__(self, scan_result, read_timeout=10):
        # Prepare for reading input events from specified device
        # - scan_result: a ScanResult instance
        # - read_timeout: timeout in ms to use for polling read()
        #   CAUTION: setting too low of a timeout may give dropped/stuck notes
        # Exceptions: may raise usb.core.USBError
        #
        device = scan_result.device
        self.device = device
        self.read_timeout = read_timeout
        # Make sure CircuitPython core is not claiming the device
        interface = 1
        if device.is_kernel_driver_active(interface):
            print('Detaching interface %d from kernel' % interface)
            device.detach_kernel_driver(interface)
        # Set configuration
        device.set_configuration()
        # Figure out which endpoints to use
        ins = scan_result.descriptor.input_endpoints(interface)
        outs = scan_result.descriptor.output_endpoints(interface)
        endpoint_in  = None if (len(ins) < 1) else ins[0]
        endpoint_out = None if (len(outs) < 1) else outs[0]
        self.int1_endpoint_in = endpoint_in
        self.int1_endpoint_out = endpoint_out

    def input_event_generator(self):
        # Read USB input events _as efficiently as possible_.
        #
        # This is a generator that makes an iterable for reading input events.
        # The code structure here is weird because it's using MicroPython
        # performance boosting tricks to reduce CPU cycles spent on dictionary
        # lookups, function calls, and heap allocations. The goal is to read
        # input fast enough to avoid audible latency glitches.
        #
        # - returns: iterable that can be used with a for loop
        # - iterable can yield:
        #   1. A memoryview(bytearray(...)) with a 4 byte usb midi packet, or
        #   2. None (read timeout)
        # Exceptions: may raise USBError
        #
        addr = self.int1_endpoint_in.bEndpointAddress
        max_packet = min(64, self.int1_endpoint_in.wMaxPacketSize)
        data = bytearray(max_packet)
        view = memoryview(data)  # using memoryview reduces heap allocations
        read = self.device.read  # caching function avoids dictionary lookups
        ms = self.read_timeout   # read timeout
        while True:
            try:
                # In theory, using a positional argument for the timeout should
                # be faster than using a `timeout=ms` keyword argument
                n = read(addr, data, ms)
                # Bulk read result will be 0 or more 4-byte midi packets, so
                # split that up into 4-byte memoryview slices
                for i in range(0, n, 4):
                    yield view[i:i+4]
                # In case of 0 byte bulk read, we still need to yield something
                if n == 0:
                    yield None
            except USBTimeoutError as e:
                # This is normal. Timeouts happen fairly often.
                yield None
            except USBError as e:
                # This happens when device is unplugged
                raise e

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191

# SPDX-License-Identifier: MIT
# SPDX-FileCopyrightText: Copyright 2025 Sam Blenny
#
# Descriptor parser for USB devices
#
# Related Documentation:
# - https://docs.circuitpython.org/en/latest/shared-bindings/usb/core/index.html
#
from usb import core
from usb.core import USBError, USBTimeoutError


def get_desc(device, desc_type, length=256):
    # Read USB descriptor of type specified by desc_type (index always 0).
    # - device: a usb.core.Device
    # - desc_type: uint8 value for the descriptor type field of wValue
    # - returns: bytearray with results from ctrl_transfer()
    # Exceptions: may raise USBError or USBTimeoutError
    data = bytearray(length)
    bmRequestType = 0x80
    wValue = (desc_type << 8) | 0
    wIndex = 0
    device.ctrl_transfer(bmRequestType, 6, wValue, wIndex, data, 300)
    return data

def split_desc(data):
    # Split a combined descriptor into its individual sub-descriptors
    # - data: a bytearray of descriptor data from ctrl_transfer()
    # - returns: array of bytearrays (first byte of each is length)
    slices = []
    cursor = 0
    limit = len(data)
    data_mv = memoryview(data)  # use memoryview to reduce heap allocations
    for i in range(limit):
        if cursor == limit:
            break
        length = data[cursor]
        if length == 0:
            break
        if cursor + length > limit:
            break
        slices.append(data_mv[cursor:cursor+length])
        cursor += length
    return slices


class ConfigDesc:
    def __init__(self, d):
        # Parse a configuration descriptor
        # - d: bytearray containing a 9 byte configuration descriptor
        if len(d) != 9 or d[0] != 0x09 or d[1] != 0x02:
            raise ValueError("Bad configuration descriptor")
        self.bNumInterfaces      = d[4]
        self.bConfigurationValue = d[5]  # for set_configuration()
        self.bMaxPower           = d[8]  # units are 2 mA


class InterfaceDesc:
    def __init__(self, d):
        # Parse an interface descriptor
        # - d: bytearray containing a 9 byte interface descriptor
        if len(d) != 9 or d[0] != 0x09 or d[1] != 0x04:
            raise ValueError("Bad interface descriptor")
        self.bInterfaceNumber   = d[2]
        self.bNumEndpoints      = d[4]
        self.bInterfaceClass    = d[5]
        self.bInterfaceSubClass = d[6]
        self.bInterfaceProtocol = d[7]
        self.endpoint = []

    def add_endpoint_descriptor(self, data):
        self.endpoint.append(EndpointDesc(data))


class EndpointDesc:
    def __init__(self, d):
        # Parse an endpoint descriptor
        # - d: bytearray containing a 7-9 byte endpoint descriptor
        if len(d) < 7 or d[0] < 0x07 or d[1] != 0x05:
            raise ValueError("Bad endpoint descriptor")
        self.bEndpointAddress = d[2]
        # bmAttributes low 2 bits: 0:control, 1:iso., 2:bulk, 3:interrupt
        self.bmAttributes     = d[3]
        self.wMaxPacketSize   = (d[5] << 8) | d[4]
        self.bInterval        = d[6]

    def attribute_str(self):
        a = self.bmAttributes & 0x3
        if a == 0:
            return 'control'
        elif a == 1:
            return 'iso'
        elif a == 2:
            return 'bulk'
        elif a == 3:
            return 'interrupt'
        return ''


class Descriptor:
    def __init__(self, device):
        # Read and parse USB device descriptor
        # - device: usb.core.Device
        #
        device_desc = get_desc(device, 0x01, length=18)
        length = device_desc[0]
        if length != 18:
            raise ValueError('Bad Device Descriptor Length: %d' % length)
        d = device_desc
        self.device_desc_bytes = d
        self.bcdUSB          = (d[ 3] << 8) | d[ 2]
        self.bDeviceClass    = d[4]
        self.bDeviceSubClass = d[5]
        self.bDeviceProtocol = d[6]
        self.bMaxPacketSize0 = d[7]
        self.idVendor        = (d[ 9] << 8) | d[ 8]
        self.idProduct       = (d[11] << 8) | d[10]
        self.bNumConfigurations = d[17]
        # Make an empty placeholder configuration
        self.config_desc_list = []
        self.configs = []
        self.interfaces = []

    def vid_pid(self):
        return (self.idVendor, self.idProduct)

    def dev_class_subclass(self):
        # Get device descriptor's class and subclass
        return (self.bDeviceClass, self.bDeviceSubClass)

    def int_class_subclass(self, interface):
        # Get requested interface descriptor's class and subclass
        for i in self.interfaces:
            if i.bInterfaceNumber == interface:
                return (i.bInterfaceClass, i.bInterfaceSubClass)
        return (None, None)

    def output_endpoints(self, interface):
        # Get list of output endpoints for requested interface
        arr = []
        input_mask = 0x80
        for i in self.interfaces:
            if i.bInterfaceNumber == interface:
                for e in i.endpoint:
                    if not (e.bEndpointAddress & input_mask):
                        arr.append(e)
        return arr

    def input_endpoints(self, interface):
        # Get list of input endpoints for interface 0
        arr = []
        input_mask = 0x80
        for i in self.interfaces:
            if i.bInterfaceNumber == interface:
                for e in i.endpoint:
                    if (e.bEndpointAddress & input_mask):
                        arr.append(e)
        return arr

    def read_configuration(self, device):
        # Read and parse USB configuration descriptor
        # - device: usb.core.Device
        config_desc_list = split_desc(get_desc(device, 0x02, length=256))
        if len(config_desc_list) == 0:
            raise ValueError("Empty Configuration Descriptor")
        self.config_desc_list = config_desc_list
        self.configs    = []
        self.interfaces = []
        interface_num = -1
        for d in config_desc_list:
            if len(d) < 2:
                continue
            bLength = d[0]
            bDescriptorType = d[1]
            tag = (bLength << 8) | bDescriptorType
            if tag == 0x0902:
                # Configuration
                self.configs.append(ConfigDesc(d))
            elif tag == 0x0904:
                # Interface
                self.interfaces.append(InterfaceDesc(d))
                interface_num += 1
            elif 7 <= bLength <= 9 and bDescriptorType == 0x05:
                # Endpoint
                if interface_num >= 0:
                    self.interfaces[interface_num].add_endpoint_descriptor(d)
                else:
                    raise ValueError("Found endpoint before interface")

    def to_bytes(self):
        return self.device_desc_bytes