example.py

''' Example showing how to use the Multi-Midi library for multi-port UART, PIO and USB MIDI on RP2040/RP2350 based boards
    https://github.com/HLammers/multi-midi

    Copyright (c) 2025 Harm Lammers

    This example is written to use with the Cybo-Drummer hardware (https://github.com/HLammers/cybo-drummer). Adapt to your needs if needed.

    See READYME.md for more instructions on how to use the Multi-Midi library.

    MIT licence:

    Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the
    "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish,
    distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to
    the following conditions:

    The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
    MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
    CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
    SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
'''

import machine
import sys
import asyncio
import time
import gc

from midi_manager import MidiManager
from midi_monitor import MidiMonitor
from log import Log
from midi_test_data import DummyBufferStream

# Allow VSCode to connect to REPL before running main.py
time.sleep(1)

# Settings, in particular useful for debugging
_ENABLE_USB     = True  # If set to False, no USB ports are set up and the USB MIDI driver is not initiated
# IMPORTANT: The MIDI driver turns off the USB serial port (USB CDC) used by the REPL, because a Windows host won’t recognize the MIDI ports
# if CDC and USB are both enabled
_LOG_TO_FILE    = False # If set to True the logger writes to a log file (\log.txt)
_LOG_TX         = False # If set to True the logger will also log sent messages, not only received ones
_RESET_PIN      = const(16) # GPIO number of a pin to which a reset button is connected (wired to ground)
_OUT_INTERVAL   = const(1) # Sleep time between sending MIDI messages to an OUT port
_SLEEP_ON_ERROR = const(10) # Sleep time before shutting down when error occurred (to allow message shown on screen to be read)

# Definitions for USB MIDI
_MANUFACTURER   = 'TestMaker' # Manufacturer string assigned to USB device
_PRODUCT        = 'TestMIDI'  # Product string assigned to USB device

# CALLBACK FUNTIONS CALLED WHEN RECEIVING DATA AND ASYNCIO TASK TO SEND TEST DATA TO A MIDI OUT PORT
# The input side of this example is set up to function as a MIDI monitor

def cb_data(port_id: int, byte_0: int, byte_1: int = 0, byte_2: int = 0) -> None:
    ''' Example callback function which is called each time a MIDI message is received (except for MIDI Real-Time ans SysEx messages)

        args:
            port_id (int): Port number (`MidiManager.in_ports` index number)
            byte_0 (int): First byte of the MIDI message
            byte_1 (int, optional): Second byte of the MIDI message
            byte_2 (int, optional): Third byte of the MIDI message
    '''
    _midi_monitor = MidiMonitor(False)
    _midi_monitor.write(port_id, byte_0, byte_1, byte_2)

def cb_real_time(port_id: int, byte: int) -> None:
    ''' Example callback function which is called each time a MIDI Real-Time message is received

        args:
            port_id (int): Port number (`MidiManager.in_ports` index number)
            byte (int): Single-byte MIDI Real-Time message
    '''
    _midi_monitor = MidiMonitor(False)
    _midi_monitor.write(port_id, byte, 0, 0)

def cb_sysex(port_id: int, buf: bytearray, num_bytes: int) -> None:
    ''' Example callback function which is called each time a MIDI SysEx message is received

        args:
            port_id (int): Port number (`MidiManager.in_ports` index number)
            buf (bytearray): Buffer in which the SysEx block is stored
            num_bytes (int): Length of the SysEx block
    '''
    _midi_monitor = MidiMonitor(False)
    if buf[0] == 0xF0: # SysEx Start
        _midi_monitor.write(port_id, 0xF0, 0, 0)
    if buf[num_bytes - 1] == 0xF7: # End of SysEx
        _midi_monitor.write(port_id, 0xF7, 0, 0)

async def test_port(port, port_id: int) -> None:
    ''' `asyncio` task to send test data to a MIDI out port

    Args:
        port_id (int): Port number (`MidiManager.out_ports` index number)
        port (OutPortUART | OutPortPIO | OutPortUSB): MIDI OUT port object to send
    '''
    # Cache as many things as possible to improve performance of the `while True` loop
    _midi_ports = MidiManager() # this works because Midiports is a singleton, so this returns the already initiated instance
    _usb_is_active = _midi_ports.usb_is_active
    _test_stream_next_msg = DummyBufferStream(True).next_message
    channel = 0
    _midi_monitor = MidiMonitor(False)
    _monitor_write = _midi_monitor.write
    _write_sysex = port.write_sysex
    _write_real_time = port.write_real_time
    _write_data = port.write_data
    _sleep = asyncio.sleep
    while True:
        # Wait for USB to be active (if initiated)
        # try:
        #     await _usb_is_active() # pyright: ignore[reportOptionalCall]
        # except TypeError: # raised if the USB MIDI device is not initiated (midi_ports.usb_is_active is None)
        #     pass
        await _usb_is_active() # pyright: ignore[reportOptionalCall]
        # Get next 3-byte MIDI message or variable length SysEx block from test stream
        data = _test_stream_next_msg()
        # Send to the right port write function, depending on the type of message / SysEx block
        status = data[0]
        if status == 0xF0: # SysEx block
            _write_sysex(data, len(data)) # write SysEx block to MIDI OUT port
            if _LOG_TX:
                _monitor_write(port_id, 0xF0, 0, 0) # monitor SysEx Start
                _monitor_write(port_id, 0xF7, 0, 0) # monitor End of SysEx
        elif status >= 0xF8: # System Real-Time message
            _write_real_time(status) # write Midi Real-Time message to MIDI OUT port
            if _LOG_TX:
                _monitor_write(port_id, status, 0, 0) # send to MIDI monitor
        else: # all other types of MIDI messages
            if status <= 0xEF: # Channel Voice message or Channel Common message
                data[0] |= channel # set channel
            _write_data(*data) # write MIDI message to MIDI OUT port
            if _LOG_TX:
                _monitor_write(port_id, *data) # send to MIDI monitor
        channel = (channel + 1) & 0xF # loop over all 16 MIDI channels
        await _sleep(_OUT_INTERVAL)

# Global variable to store test_port tasks
_test_port_tasks = []

# Add a reset button (useful during development and debugging)
reset_button = machine.Pin(_RESET_PIN, machine.Pin.IN, machine.Pin.PULL_UP)
reset_button.irq(lambda _: machine.reset(), machine.Pin.IRQ_RISING, hard = True)

# First time the Log logger singleton is initiated, so this is the moment to set options; without this it would use the default options
_log = Log(_LOG_TO_FILE)

# Define exception handler for asyncio event loop
def _handle_exception(loop, context) -> None:
    _log.sync_write('ERROR: ' + str(context['exception']))
    # Delay to allow message shown on screen to be read
    time.sleep(_SLEEP_ON_ERROR)
    # Cancel all tasks and deinitialize everything
    _log_task.cancel()
    _log.deinit()
    for task in _test_port_tasks:
        task.cancel()
    _midi_manager.deinit()
    loop.stop()
    # Shut down
    sys.exit()

# Define asyncio event loop - this is the core part of the Example application
async def main():
    global _log_task, _midi_manager
    # Start the logger’s background processes and assign an exception handler to the asyncio event loop
    _log_task = asyncio.create_task(_log.run())
    asyncio.get_event_loop().set_exception_handler(_handle_exception)
    
    # THE FOLLOWING STEPS DO THE ACTUAL INITIATION OF THE MULTI-PORT MIDI MANAGER
    
    # Initiate the MIDI manager, which is the primary interface of the Multi Midi library
    _midi_manager = MidiManager()
    # Set USB device’s manufacturer name, product name and serial number
    # TIP: use machine.unique_id() to get a byte string with a unique identifier of a board/SoC to use as serial_str
    if _ENABLE_USB:
        _midi_manager.set_usb_strings(_MANUFACTURER, _PRODUCT, machine.unique_id())
    # Assign callback functions to be called when receiving MIDI data
    _midi_manager.assign_callbacks(cb_real_time, cb_sysex, cb_data)
    # Initiatie 6 hardware MIDI IN ports, 6 hardware MIDI OUT ports and 7 USB MIDI IN/OUT ports, compatible with the Cybo-Drummer hardware
    # IMPORTANT: USB MIDI port names need to be at least 2 characters long (otherwise a Windows host would fail to recognize the device) and
    # are not shown on a Windows host
    _midi_manager.add_uart_in(0)                      # in_ports[0]:   hardware IN port 1     (UART0, default pins)
    _midi_manager.add_uart_in(1)                      # in_ports[1]:   hardware IN port 2     (UART1, default pins)
    _midi_manager.add_pio_in(0, 6)                    # in_ports[2]:   hardware IN port 3     (PIO 0, GPIO 6)
    _midi_manager.add_pio_in(1, 7)                    # in_ports[3]:   hardware IN port 4     (PIO 1, GPIO 7)
    _midi_manager.add_pio_in(2, 8)                    # in_ports[4]:   hardware IN port 5     (PIO 2, GPIO 8)
    _midi_manager.add_pio_in(3, 9)                    # in_ports[5]:   hardware IN port 6     (PIO 3, GPIO 9)
    if _ENABLE_USB:
        _midi_manager.add_usb_in(0, 'Port 1')         # in_ports[6]:   USB IN port ‘Port 1’   (virtual cable 0, with External Jack)
        _midi_manager.add_usb_in(1, 'Port 2')         # in_ports[7]:   USB IN port ‘Port 2’   (virtual cable 1, with External Jack)
        _midi_manager.add_usb_in(2, 'Port 3')         # in_ports[8]:   USB IN port ‘Port 3’   (virtual cable 2, with External Jack)
        _midi_manager.add_usb_in(3, 'Port 4')         # in_ports[9]:   USB IN port ‘Port 4’   (virtual cable 3, with External Jack)
        _midi_manager.add_usb_in(4, 'Port 5')         # in_ports[10]:  USB IN port ‘Port 5’   (virtual cable 4, with External Jack)
        _midi_manager.add_usb_in(5, 'Port 6')         # in_ports[11]:  USB IN port ‘Port 6’   (virtual cable 5, with External Jack)
        _midi_manager.add_usb_in(6, 'Control', False) # in_ports[12]:  USB IN port ‘Control’  (virtual cable 6, without External Jack)
    _midi_manager.add_uart_out(0)                     # out_ports[0]:  hardware OUT port 1    (UART0, default pins)
    _midi_manager.add_uart_out(1)                     # out_ports[1]:  hardware OUT port 2    (UART1, default pins)
    _midi_manager.add_pio_out(4, 10)                  # out_ports[2]:  hardware OUT port 3    (PIO 4, GPIO 10)
    _midi_manager.add_pio_out(5, 11)                  # out_ports[3]:  hardware OUT port 4    (PIO 5, GPIO 11)
    _midi_manager.add_pio_out(6, 12)                  # out_ports[4]:  hardware OUT port 5    (PIO 6, GPIO 12)
    _midi_manager.add_pio_out(7, 13)                  # out_ports[5]:  hardware OUT port 6    (PIO 7, GPIO 13)
    if _ENABLE_USB:
        _midi_manager.add_usb_out(0)                  # out_ports[6]:  USB OUT port ‘Port 1’  (virtual cable 0, with External Jack)
        _midi_manager.add_usb_out(1)                  # out_ports[7]:  USB OUT port ‘Port 2’  (virtual cable 1, with External Jack)
        _midi_manager.add_usb_out(2)                  # out_ports[8]:  USB OUT port ‘Port 3’  (virtual cable 2, with External Jack)
        _midi_manager.add_usb_out(3)                  # out_ports[9]:  USB OUT port ‘Port 4’  (virtual cable 3, with External Jack)
        _midi_manager.add_usb_out(4)                  # out_ports[10]: USB OUT port ‘Port 5’  (virtual cable 4, with External Jack)
        _midi_manager.add_usb_out(5)                  # out_ports[11]: USB OUT port ‘Port 6’  (virtual cable 5, with External Jack)
        _midi_manager.add_usb_out(6)                  # out_ports[12]: USB OUT port ‘Control’ (virtual cable 6, without External Jack)
    # Start the midi manager
    await _midi_manager.run()

    # Create tasks to test each MIDI OUT port by sending out test data
    test_port_tasks = []
    for i, port in enumerate(_midi_manager.out_ports):
        test_port_tasks.append(asyncio.create_task(test_port(port, i))) # run port test which sends out MIDI events eternally
    # Confirm that everything is up and running
    _log.write('running')
    await asyncio.Event().wait() # keep alive (wait eternally)

# Run the asyncio event loop
try:
    asyncio.run(main())
finally:
    # Cancel all tasks and deinitialize everything
    _log_task.cancel()
    _log.deinit()
    for task in _test_port_tasks:
        task.cancel()
    _midi_manager.deinit()
    asyncio.new_event_loop() # clear retained state