MIDI Processing¶
Recipes for MIDI input/output and processing with CoreMIDI.
MIDI Device Discovery¶
List MIDI Devices¶
Discover available MIDI sources and destinations:
from coremusic import capi
# List MIDI sources (input devices)
num_sources = capi.midi_get_number_of_sources()
print(f"MIDI Sources: {num_sources}")
for i in range(num_sources):
source = capi.midi_get_source(i)
name = capi.midi_object_get_string_property(source, "name")
print(f" {i}: {name}")
# List MIDI destinations (output devices)
num_dests = capi.midi_get_number_of_destinations()
print(f"\nMIDI Destinations: {num_dests}")
for i in range(num_dests):
dest = capi.midi_get_destination(i)
name = capi.midi_object_get_string_property(dest, "name")
print(f" {i}: {name}")
Find Device by Name¶
Locate a specific MIDI device:
from coremusic import capi
def find_midi_source(device_name):
"""Find MIDI source by name"""
num_sources = capi.midi_get_number_of_sources()
for i in range(num_sources):
source = capi.midi_get_source(i)
name = capi.midi_object_get_string_property(source, "name")
if device_name.lower() in name.lower():
return source, name
return None, None
# Find a specific device
source, name = find_midi_source("Keyboard")
if source:
print(f"Found: {name}")
else:
print("Device not found")
MIDI Input¶
How Incoming Packets Arrive¶
CoreMIDI delivers incoming data as packets on its own high-priority receive thread. A packet is not the same thing as a MIDI message:
- One packet may contain several complete messages that arrived together.
- A large system-exclusive dump is spread across consecutive packets.
- Real-time bytes such as clock may be interleaved inside another message.
MIDIMessageSplitter turns a stream of packet payloads back into individual
messages, holding the state needed to span packets. Timestamps are host times
(mach absolute ticks); convert them with capi.midi_host_time_to_seconds.
An input port buffers packets by default, so the owning thread controls when Python code runs. Pass a callback instead only when you need the lowest possible latency and can guarantee the callback never blocks.
Receive MIDI Messages¶
Poll the port from your own loop:
import time
from coremusic import capi
from coremusic.midi import MIDIMessageSplitter
client = capi.midi_client_create("MIDI Input")
input_port = capi.midi_input_port_create(client, "Input")
# Connect to the first source, or to one we publish ourselves when nothing
# else is connected
if capi.midi_get_number_of_sources() > 0:
source = capi.midi_get_source(0)
else:
source = capi.midi_source_create(client, "Cookbook Input")
capi.midi_port_connect_source(input_port, source)
splitter = MIDIMessageSplitter()
print("Listening for MIDI for 1s...")
deadline = time.monotonic() + 1.0
try:
while time.monotonic() < deadline:
# Blocks until a packet arrives or the timeout expires.
if not capi.midi_input_wait(input_port, 0.1):
continue
for host_time, payload in capi.midi_input_poll(input_port):
seconds = capi.midi_host_time_to_seconds(host_time)
for data in splitter.push(payload):
status = data[0]
message_type = status & 0xF0
channel = status & 0x0F
if message_type == 0x90 and data[2] > 0: # Note On
print(f"Note On: ch={channel}, note={data[1]}, vel={data[2]}")
elif message_type == 0x80 or message_type == 0x90: # Note Off
print(f"Note Off: ch={channel}, note={data[1]}")
elif message_type == 0xB0: # Control Change
print(f"CC: ch={channel}, ctrl={data[1]}, val={data[2]}")
except KeyboardInterrupt:
print("\nStopped")
# Cleanup
capi.midi_port_disconnect_source(input_port, source)
capi.midi_port_dispose(input_port)
capi.midi_client_dispose(client)
If the loop cannot keep up, the oldest packets are discarded once the buffer is
full. Check capi.midi_input_dropped(input_port) to detect this, and raise
queue_size on midi_input_port_create if it happens.
Receive with a Callback¶
A callback runs on the CoreMIDI receive thread, so it must be short and must not block, allocate heavily, or acquire locks held by slow code. Exceptions raised inside it are swallowed rather than propagated into the framework.
import queue
from coremusic import capi
incoming: queue.SimpleQueue = queue.SimpleQueue()
def midi_callback(data: bytes, host_time: int) -> None:
# Hand off immediately; do the real work on your own thread.
incoming.put((host_time, data))
client = capi.midi_client_create("MIDI Input")
input_port = capi.midi_input_port_create(client, "Input", midi_callback)
# Connect to the first source, or to one we publish ourselves when nothing
# else is connected
if capi.midi_get_number_of_sources() > 0:
source = capi.midi_get_source(0)
else:
source = capi.midi_source_create(client, "Cookbook Input")
capi.midi_port_connect_source(input_port, source)
Filter MIDI Messages¶
Filter specific MIDI message types:
from coremusic import capi
from coremusic.midi import MIDIMessageSplitter
class MIDIFilter:
def __init__(self, filter_notes=False, filter_cc=False):
self.filter_notes = filter_notes
self.filter_cc = filter_cc
self.splitter = MIDIMessageSplitter()
def process(self, payload: bytes) -> None:
for data in self.splitter.push(payload):
message_type = data[0] & 0xF0
# Filter note messages
if message_type in (0x80, 0x90) and self.filter_notes:
continue
# Filter CC messages
if message_type == 0xB0 and self.filter_cc:
continue
print(f"MIDI: {[hex(b) for b in data]}")
# Create filter that blocks notes but allows CC
midi_filter = MIDIFilter(filter_notes=True, filter_cc=False)
client = capi.midi_client_create("Filtered Input")
input_port = capi.midi_input_port_create(client, "Input")
# Connect, then feed polled packets to midi_filter.process()...
MIDI Output¶
Send MIDI Messages¶
Send MIDI messages to an output device:
import time
from coremusic import capi
from coremusic.midi import note_off, note_on
# Create MIDI client and output port
client = capi.midi_client_create("MIDI Output")
output_port = capi.midi_output_port_create(client, "Output")
# Use the first destination, or publish one of our own when nothing else is
# connected, so this runs with or without hardware attached
if capi.midi_get_number_of_destinations() > 0:
dest = capi.midi_get_destination(0)
else:
dest = capi.midi_destination_create(client, "Cookbook Output")
# Send Note On (middle C, velocity 100, channel 0)
capi.midi_send_data(output_port, dest, note_on("C4", 100))
print("Sent Note On")
time.sleep(1.0)
# Send Note Off
capi.midi_send_data(output_port, dest, note_off("C4"))
print("Sent Note Off")
# Cleanup
capi.midi_port_dispose(output_port)
capi.midi_client_dispose(client)
Play MIDI Sequence¶
Send a sequence of MIDI notes:
import time
from coremusic import capi
from coremusic.midi import note_off, note_on
def play_note(port, dest, channel, note, velocity, duration):
"""Play a single note"""
# Note On
capi.midi_send_data(port, dest, note_on(note, velocity, channel=channel))
# Wait
time.sleep(duration)
# Note Off
capi.midi_send_data(port, dest, note_off(note, channel=channel))
# Setup
client = capi.midi_client_create("Sequencer")
output_port = capi.midi_output_port_create(client, "Output")
# Use the first destination, or publish one of our own when nothing else is
# connected, so this runs with or without hardware attached
if capi.midi_get_number_of_destinations() > 0:
dest = capi.midi_get_destination(0)
else:
dest = capi.midi_destination_create(client, "Cookbook Output")
# Play C major scale
scale = [60, 62, 64, 65, 67, 69, 71, 72] # C D E F G A B C
for note in scale:
play_note(output_port, dest, channel=0, note=note, velocity=100, duration=0.5)
time.sleep(0.1) # Gap between notes
# Cleanup
capi.midi_port_dispose(output_port)
capi.midi_client_dispose(client)
Send Control Changes¶
Send MIDI CC messages for automation:
import time
from coremusic import capi
from coremusic.constants import MIDIControlChange
from coremusic.midi import control_change, note_off, note_on
client = capi.midi_client_create("CC Controller")
output_port = capi.midi_output_port_create(client, "Output")
# Use the first destination, or publish one of our own when nothing else is
# connected, so this runs with or without hardware attached
if capi.midi_get_number_of_destinations() > 0:
dest = capi.midi_get_destination(0)
else:
dest = capi.midi_destination_create(client, "Cookbook Output")
# Start a note
capi.midi_send_data(output_port, dest, note_on("C4", 100))
# Fade volume (CC 7) from 127 to 0
for volume in range(127, -1, -5):
cc = control_change(MIDIControlChange.VOLUME, volume)
capi.midi_send_data(output_port, dest, cc)
time.sleep(0.05)
# Stop note
capi.midi_send_data(output_port, dest, note_off("C4"))
# Cleanup
capi.midi_port_dispose(output_port)
capi.midi_client_dispose(client)
MIDI Routing¶
MIDI Thru¶
Route MIDI input directly to output:
import time
from coremusic import capi
# Create client with input and output ports
client = capi.midi_client_create("MIDI Thru")
# Output port
output_port = capi.midi_output_port_create(client, "Output")
# Use the first destination, or publish one of our own when nothing else is
# connected, so this runs with or without hardware attached
if capi.midi_get_number_of_destinations() > 0:
dest = capi.midi_get_destination(0)
else:
dest = capi.midi_destination_create(client, "Cookbook Output")
# Input port
input_port = capi.midi_input_port_create(client, "Input")
# Connect to the first source, or to one we publish ourselves when nothing
# else is connected
if capi.midi_get_number_of_sources() > 0:
source = capi.midi_get_source(0)
else:
source = capi.midi_source_create(client, "Cookbook Input")
capi.midi_port_connect_source(input_port, source)
print("MIDI thru active... (1s)")
deadline = time.monotonic() + 1.0
try:
while time.monotonic() < deadline:
if not capi.midi_input_wait(input_port, 0.1):
continue
# Packets can be forwarded verbatim; no need to split them first.
for _host_time, payload in capi.midi_input_poll(input_port):
capi.midi_send_data(output_port, dest, payload)
except KeyboardInterrupt:
print("\nStopped")
# Cleanup
capi.midi_port_disconnect_source(input_port, source)
capi.midi_port_dispose(input_port)
capi.midi_port_dispose(output_port)
capi.midi_client_dispose(client)
Channel Routing¶
Route MIDI from one channel to another:
import time
from coremusic import capi
from coremusic.midi import MIDIMessageSplitter
class ChannelRouter:
def __init__(self, output_port, dest, input_channel, output_channel):
self.output_port = output_port
self.dest = dest
self.input_channel = input_channel
self.output_channel = output_channel
self.splitter = MIDIMessageSplitter()
def process(self, payload: bytes) -> None:
for message in self.splitter.push(payload):
status = message[0]
# Channel voice messages only; system messages have no channel.
if not 0x80 <= status <= 0xEF:
continue
if (status & 0x0F) == self.input_channel:
data = bytearray(message)
data[0] = (status & 0xF0) | self.output_channel
capi.midi_send_data(self.output_port, self.dest, bytes(data))
# Route channel 0 -> channel 1
client = capi.midi_client_create("Channel Router")
output_port = capi.midi_output_port_create(client, "Output")
# Use the first destination, or publish one of our own when nothing else is
# connected, so this runs with or without hardware attached
if capi.midi_get_number_of_destinations() > 0:
dest = capi.midi_get_destination(0)
else:
dest = capi.midi_destination_create(client, "Cookbook Output")
router = ChannelRouter(output_port, dest, input_channel=0, output_channel=1)
input_port = capi.midi_input_port_create(client, "Input")
# Connect to the first source, or to one we publish ourselves when nothing
# else is connected
if capi.midi_get_number_of_sources() > 0:
source = capi.midi_get_source(0)
else:
source = capi.midi_source_create(client, "Cookbook Input")
capi.midi_port_connect_source(input_port, source)
# Run for a second; a real router would loop until told to stop
deadline = time.monotonic() + 1.0
while time.monotonic() < deadline:
if capi.midi_input_wait(input_port, 0.1):
for _host_time, payload in capi.midi_input_poll(input_port):
router.process(payload)
MIDI Transformation¶
Transpose Notes¶
Transpose all incoming notes:
import time
from coremusic import capi
from coremusic.midi import MIDIMessageSplitter
class Transposer:
def __init__(self, output_port, dest, semitones):
self.output_port = output_port
self.dest = dest
self.semitones = semitones
self.splitter = MIDIMessageSplitter()
def process(self, payload: bytes) -> None:
for message in self.splitter.push(payload):
data = bytearray(message)
message_type = data[0] & 0xF0
# Transpose note on/off messages
if message_type in (0x80, 0x90) and len(data) >= 3:
original_note = data[1]
transposed_note = max(0, min(127, original_note + self.semitones))
data[1] = transposed_note
print(f"Transposed: {original_note} -> {transposed_note}")
# Forward the (possibly modified) message
capi.midi_send_data(self.output_port, self.dest, bytes(data))
# Transpose up one octave
client = capi.midi_client_create("Transposer")
output_port = capi.midi_output_port_create(client, "Output")
# Use the first destination, or publish one of our own when nothing else is
# connected, so this runs with or without hardware attached
if capi.midi_get_number_of_destinations() > 0:
dest = capi.midi_get_destination(0)
else:
dest = capi.midi_destination_create(client, "Cookbook Output")
transposer = Transposer(output_port, dest, semitones=12)
input_port = capi.midi_input_port_create(client, "Input")
# Connect to the first source, or to one we publish ourselves when nothing
# else is connected
if capi.midi_get_number_of_sources() > 0:
source = capi.midi_get_source(0)
else:
source = capi.midi_source_create(client, "Cookbook Input")
capi.midi_port_connect_source(input_port, source)
# Run for a second; a real router would loop until told to stop
deadline = time.monotonic() + 1.0
while time.monotonic() < deadline:
if capi.midi_input_wait(input_port, 0.1):
for _host_time, payload in capi.midi_input_poll(input_port):
transposer.process(payload)
Velocity Scaling¶
Scale note velocities:
from coremusic import capi
from coremusic.constants import MIDIStatus
from coremusic.midi import MIDIMessageSplitter, note_on
class VelocityScaler:
def __init__(self, output_port, dest, scale_factor):
self.output_port = output_port
self.dest = dest
self.scale_factor = scale_factor
self.splitter = MIDIMessageSplitter()
def process(self, payload: bytes) -> None:
for message in self.splitter.push(payload):
data = bytearray(message)
# Note On with a non-zero velocity; a zero velocity is a Note Off
# and must keep its value.
if (
(data[0] & 0xF0) == MIDIStatus.NOTE_ON
and len(data) >= 3
and data[2] > 0
):
original_vel = data[2]
scaled_vel = int(original_vel * self.scale_factor)
data[2] = max(1, min(127, scaled_vel)) # Clamp to 1-127
print(f"Velocity: {original_vel} -> {data[2]}")
capi.midi_send_data(self.output_port, self.dest, bytes(data))
# Scale velocities to 80% (softer)
scaler = VelocityScaler(output_port, dest, scale_factor=0.8)
# Feed it as in the Transpose Notes example; here is one message by hand
scaler.process(note_on("C4", 100))
MIDI Recording¶
Record MIDI Messages¶
Record MIDI to a list with timestamps:
Message times come from the packet host timestamp rather than the wall clock, so they stay accurate even if the polling loop is briefly delayed.
import json
import time
from coremusic import capi
from coremusic.midi import MIDIMessageSplitter
class MIDIRecorder:
def __init__(self, port_id):
self.port_id = port_id
self.splitter = MIDIMessageSplitter()
self.origin = None
self.recorded_messages = []
def record(self, duration):
print("Recording started")
deadline = time.monotonic() + duration
while time.monotonic() < deadline:
if not capi.midi_input_wait(self.port_id, 0.1):
continue
for host_time, payload in capi.midi_input_poll(self.port_id):
# A zero timestamp means "as soon as possible".
if host_time == 0:
host_time = capi.midi_current_host_time()
seconds = capi.midi_host_time_to_seconds(host_time)
if self.origin is None:
self.origin = seconds
for data in self.splitter.push(payload):
self.recorded_messages.append({
'time': seconds - self.origin,
'data': data,
})
dropped = capi.midi_input_dropped(self.port_id)
if dropped:
print(f"Warning: dropped {dropped} packets")
print(f"Recording stopped: {len(self.recorded_messages)} messages")
def save(self, filename):
"""Save recorded messages to file"""
with open(filename, 'w') as f:
messages = [
{'time': msg['time'], 'data': list(msg['data'])}
for msg in self.recorded_messages
]
json.dump(messages, f, indent=2)
print(f"Saved to {filename}")
# Setup recorder
client = capi.midi_client_create("Recorder")
input_port = capi.midi_input_port_create(client, "Input")
# Connect to the first source, or to one we publish ourselves when nothing
# else is connected
if capi.midi_get_number_of_sources() > 0:
source = capi.midi_get_source(0)
else:
source = capi.midi_source_create(client, "Cookbook Input")
capi.midi_port_connect_source(input_port, source)
recorder = MIDIRecorder(input_port)
recorder.record(duration=10)
recorder.save("recorded_midi.json")
# Cleanup
capi.midi_port_disconnect_source(input_port, source)
capi.midi_port_dispose(input_port)
capi.midi_client_dispose(client)
Playback Recorded MIDI¶
Play back recorded MIDI messages:
import json
import time
from coremusic import capi
def playback_midi(filename, output_port, dest):
"""Play back recorded MIDI"""
# Load recorded messages
with open(filename) as f:
messages = json.load(f)
if not messages:
print("No messages to play")
return
print(f"Playing back {len(messages)} messages...")
start_time = time.time()
for msg in messages:
# Wait until scheduled time
target_time = start_time + msg['time']
wait_time = target_time - time.time()
if wait_time > 0:
time.sleep(wait_time)
# Send message
data = bytes(msg['data'])
capi.midi_send_data(output_port, dest, data)
print("Playback complete")
# Setup playback
client = capi.midi_client_create("Playback")
output_port = capi.midi_output_port_create(client, "Output")
# Use the first destination, or publish one of our own when nothing else is
# connected, so this runs with or without hardware attached
if capi.midi_get_number_of_destinations() > 0:
dest = capi.midi_get_destination(0)
else:
dest = capi.midi_destination_create(client, "Cookbook Output")
# Play recording
playback_midi("recorded_midi.json", output_port, dest)
# Cleanup
capi.midi_port_dispose(output_port)
capi.midi_client_dispose(client)
Complete Example: MIDI Monitor¶
Full-featured MIDI monitor with message parsing:
import time
from coremusic import capi
from coremusic.midi import MIDIMessageSplitter
class MIDIMonitor:
def __init__(self):
self.message_count = 0
self.splitter = MIDIMessageSplitter()
def parse_message(self, data):
"""Parse and format MIDI message"""
if len(data) == 0:
return "Empty message"
status = data[0]
message_type = status & 0xF0
channel = (status & 0x0F) + 1
if message_type == 0x80: # Note Off
return f"Note Off | Ch {channel:2d} | Note {data[1]:3d} | Vel {data[2]:3d}"
elif message_type == 0x90: # Note On
if data[2] == 0: # Velocity 0 = Note Off
return f"Note Off | Ch {channel:2d} | Note {data[1]:3d} | Vel {data[2]:3d}"
return f"Note On | Ch {channel:2d} | Note {data[1]:3d} | Vel {data[2]:3d}"
elif message_type == 0xA0: # Poly Aftertouch
return f"Poly AT | Ch {channel:2d} | Note {data[1]:3d} | Pressure {data[2]:3d}"
elif message_type == 0xB0: # Control Change
return f"CC | Ch {channel:2d} | Ctrl {data[1]:3d} | Val {data[2]:3d}"
elif message_type == 0xC0: # Program Change
return f"Program | Ch {channel:2d} | Program {data[1]:3d}"
elif message_type == 0xD0: # Channel Aftertouch
return f"Channel AT| Ch {channel:2d} | Pressure {data[1]:3d}"
elif message_type == 0xE0: # Pitch Bend
value = data[1] + (data[2] << 7)
return f"Pitch Bend| Ch {channel:2d} | Value {value:5d}"
elif status == 0xF8: # Clock
return "MIDI Clock"
elif status == 0xFA: # Start
return "MIDI Start"
elif status == 0xFB: # Continue
return "MIDI Continue"
elif status == 0xFC: # Stop
return "MIDI Stop"
else:
hex_data = ' '.join(f'{b:02X}' for b in data)
return f"Unknown | {hex_data}"
def process(self, payload):
for data in self.splitter.push(payload):
self.message_count += 1
message = self.parse_message(data)
print(f"[{self.message_count:6d}] {message}")
# Setup monitor
monitor = MIDIMonitor()
client = capi.midi_client_create("MIDI Monitor")
input_port = capi.midi_input_port_create(client, "Monitor Input")
# Connect to all MIDI sources
num_sources = capi.midi_get_number_of_sources()
print(f"Monitoring {num_sources} MIDI source(s)\n")
for i in range(num_sources):
source = capi.midi_get_source(i)
capi.midi_port_connect_source(input_port, source)
name = capi.midi_object_get_string_property(source, "name")
print(f"Connected to: {name}")
print("\nMIDI Monitor - running for 1s")
print("-" * 70)
deadline = time.monotonic() + 1.0
try:
while time.monotonic() < deadline:
if not capi.midi_input_wait(input_port, 0.1):
continue
for _host_time, payload in capi.midi_input_poll(input_port):
monitor.process(payload)
except KeyboardInterrupt:
pass
print(f"\n\nStopped - Received {monitor.message_count} messages")
# Cleanup
for i in range(num_sources):
source = capi.midi_get_source(i)
capi.midi_port_disconnect_source(input_port, source)
capi.midi_port_dispose(input_port)
capi.midi_client_dispose(client)
Best Practices¶
Resource Management¶
Always dispose of MIDI resources:
from coremusic import capi
# Create resources
client = capi.midi_client_create("App")
port = capi.midi_output_port_create(client, "Out")
try:
# Use resources
pass
finally:
# Always cleanup
capi.midi_port_dispose(port)
capi.midi_client_dispose(client)
Error Handling¶
Handle MIDI errors gracefully:
from coremusic import capi
# There may be no destinations at all, and midi_get_destination raises rather
# than handing back a bad handle
try:
dest = capi.midi_get_destination(0)
except ValueError:
print("No MIDI destinations available")
dest = capi.midi_destination_create(client, "Fallback Output")
try:
capi.midi_send_data(port, dest, data)
except RuntimeError as e:
print(f"Failed to send MIDI: {e}")
Timing Precision¶
MIDI timestamps are host times: mach absolute ticks on the same monotonic clock CoreAudio uses. Schedule ahead of the current time rather than sending at the moment the note should sound, so the CoreMIDI server absorbs the jitter of your own loop. A timestamp of 0 means "as soon as possible".
from coremusic import capi
# Schedule a note 50 ms from now
when = capi.midi_current_host_time() + capi.midi_seconds_to_host_time(0.05)
capi.midi_send_data(port, dest, note, when)
# Convert an incoming packet timestamp back to seconds
for host_time, _payload in capi.midi_input_poll(input_port):
seconds = capi.midi_host_time_to_seconds(host_time)
Thread Safety¶
CoreMIDI receives on its own thread. Polling with midi_input_poll keeps every
Python object you touch on your own thread, which is the simpler and safer
default.
A callback passed to midi_input_port_create runs on the CoreMIDI receive
thread instead. Keep it to a hand-off, and do the real work elsewhere:
import queue
import time
from coremusic import capi
from coremusic.midi import note_on
class MIDIProcessor:
def __init__(self):
self.incoming: queue.SimpleQueue = queue.SimpleQueue()
def callback(self, data: bytes, host_time: int) -> None:
# Runs on the CoreMIDI thread: hand off and return immediately.
self.incoming.put((host_time, data))
def run(self, deadline: float) -> None:
# Runs on your own thread.
while time.monotonic() < deadline:
try:
host_time, data = self.incoming.get(timeout=0.1)
except queue.Empty:
continue
print(f"{host_time}: {data.hex()}")
processor = MIDIProcessor()
# Whatever the callback is attached to, the hand-off looks like this
processor.callback(note_on("C4", 100), capi.midi_current_host_time())
processor.run(deadline=time.monotonic() + 0.2)
See Also¶
- API Reference - Complete API reference
- AudioUnit Hosting - AudioUnit plugin hosting (instruments)
- Link Integration - Ableton Link tempo sync
- CoreMIDI documentation: https://developer.apple.com/documentation/coremidi