christianandClaude Opus 5.5
b793490d90
- NA's CAMPUS page (99): tap = acknowledge, hold = fix, at the campus shown. NA's settings page turns the cell into that campus's key and writes the command into that campus's Companion (sp_cmd), exactly as a press on its own deck does. Logged on both sides. - coremidi.JXA_CONNECT: connect_peer in JavaScript for Automation, used when a Mac has no python (no Command Line Tools): a MIDI hold now reconnects Woonsocket's Tracks, FOH and Lights. - Woonsocket's MIDI links are on wired addresses now (Tracks to Lighting 10.2.2.2, Tracks to FOH 10.2.1.12; the other ends list Tracks at 10.2.1.219), and the wired-network rule is back on there. - Auto-setup keeps one check per MIDI link (the end that reports the link's state): two pulled links read 4 DOWN before. Tried live: a hold sent from NA's page 99 reconnected the FOH link at Woonsocket. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
446 lines
18 KiB
Python
446 lines
18 KiB
Python
"""Minimal CoreMIDI access through ctypes (no PyObjC, stdlib only).
|
|
|
|
Two things the status panel needs that Companion cannot see:
|
|
|
|
* sessions() - the live Network MIDI (RTP-MIDI) sessions on this Mac and the
|
|
peers currently joined to each, read from the network driver's
|
|
"apple.midirtp.session" entity property.
|
|
* Listener - timestamps the last message that arrived on a named MIDI
|
|
source, so a tile can say "last MIDI 3s ago".
|
|
* connect_peer() - re-add a peer to a named session, the way Audio MIDI Setup's
|
|
"Connect" does (macOS's MIDINetworkSession API only reaches the
|
|
default session, not named ones like "NA Tracks").
|
|
"""
|
|
import ctypes
|
|
import ctypes.util
|
|
import plistlib
|
|
import threading
|
|
import time
|
|
|
|
_cm = ctypes.CDLL(ctypes.util.find_library("CoreMIDI"))
|
|
_cf = ctypes.CDLL(ctypes.util.find_library("CoreFoundation"))
|
|
|
|
_U32 = ctypes.c_uint32
|
|
_UTF8 = 0x08000100
|
|
|
|
_cm.MIDIGetNumberOfDevices.restype = ctypes.c_ulong
|
|
_cm.MIDIGetDevice.restype = _U32
|
|
_cm.MIDIGetDevice.argtypes = [ctypes.c_ulong]
|
|
_cm.MIDIDeviceGetNumberOfEntities.restype = ctypes.c_ulong
|
|
_cm.MIDIDeviceGetNumberOfEntities.argtypes = [_U32]
|
|
_cm.MIDIDeviceGetEntity.restype = _U32
|
|
_cm.MIDIDeviceGetEntity.argtypes = [_U32, ctypes.c_ulong]
|
|
_cm.MIDIGetNumberOfSources.restype = ctypes.c_ulong
|
|
_cm.MIDIGetSource.restype = _U32
|
|
_cm.MIDIGetSource.argtypes = [ctypes.c_ulong]
|
|
_cm.MIDIObjectGetStringProperty.argtypes = [_U32, ctypes.c_void_p, ctypes.POINTER(ctypes.c_void_p)]
|
|
_cm.MIDIObjectGetDictionaryProperty.argtypes = [_U32, ctypes.c_void_p, ctypes.POINTER(ctypes.c_void_p)]
|
|
_cm.MIDIObjectGetIntegerProperty.argtypes = [_U32, ctypes.c_void_p, ctypes.POINTER(ctypes.c_int32)]
|
|
_cm.MIDIObjectSetDictionaryProperty.argtypes = [_U32, ctypes.c_void_p, ctypes.c_void_p]
|
|
|
|
_cf.CFStringCreateWithCString.restype = ctypes.c_void_p
|
|
_cf.CFStringCreateWithCString.argtypes = [ctypes.c_void_p, ctypes.c_char_p, _U32]
|
|
_cf.CFStringGetCString.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_long, _U32]
|
|
_cf.CFRelease.argtypes = [ctypes.c_void_p]
|
|
_cf.CFPropertyListCreateData.restype = ctypes.c_void_p
|
|
_cf.CFPropertyListCreateData.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_long, ctypes.c_ulong, ctypes.c_void_p]
|
|
_cf.CFDataGetLength.restype = ctypes.c_long
|
|
_cf.CFDataGetLength.argtypes = [ctypes.c_void_p]
|
|
_cf.CFDataGetBytePtr.restype = ctypes.c_void_p
|
|
_cf.CFDataGetBytePtr.argtypes = [ctypes.c_void_p]
|
|
_cf.CFDataCreate.restype = ctypes.c_void_p
|
|
_cf.CFDataCreate.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_long]
|
|
_cf.CFPropertyListCreateWithData.restype = ctypes.c_void_p
|
|
_cf.CFPropertyListCreateWithData.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_ulong,
|
|
ctypes.c_void_p, ctypes.c_void_p]
|
|
|
|
# Apple's network MIDI driver: "AppleMIDIRTPDriver" on macOS 26.2 (ProPresenter Mac),
|
|
# "AppleMIDINetworkDriver" on 26.5 (NA Ingest, NA Tracks).
|
|
NETWORK_DRIVERS = ("com.apple.AppleMIDINetworkDriver", "com.apple.AppleMIDIRTPDriver")
|
|
|
|
# One lock around every CoreMIDI call: concurrent first calls from two threads
|
|
# (session reads vs. MIDIClientCreate) intermittently fail with -2.
|
|
_lock = threading.RLock()
|
|
|
|
|
|
def _cfstr(s):
|
|
return _cf.CFStringCreateWithCString(None, s.encode("utf-8"), _UTF8)
|
|
|
|
|
|
def _string_prop(obj, key):
|
|
out = ctypes.c_void_p()
|
|
k = _cfstr(key)
|
|
err = _cm.MIDIObjectGetStringProperty(obj, k, ctypes.byref(out))
|
|
_cf.CFRelease(k)
|
|
if err or not out.value:
|
|
return None
|
|
buf = ctypes.create_string_buffer(1024)
|
|
_cf.CFStringGetCString(out, buf, 1024, _UTF8)
|
|
_cf.CFRelease(out)
|
|
return buf.value.decode("utf-8", "replace")
|
|
|
|
|
|
def _dict_prop(obj, key):
|
|
out = ctypes.c_void_p()
|
|
k = _cfstr(key)
|
|
err = _cm.MIDIObjectGetDictionaryProperty(obj, k, ctypes.byref(out))
|
|
_cf.CFRelease(k)
|
|
if err or not out.value:
|
|
return None
|
|
data = _cf.CFPropertyListCreateData(None, out, 100, 0, None) # 100 = XML format
|
|
raw = ctypes.string_at(_cf.CFDataGetBytePtr(data), _cf.CFDataGetLength(data))
|
|
_cf.CFRelease(data)
|
|
_cf.CFRelease(out)
|
|
return plistlib.loads(raw)
|
|
|
|
|
|
def _set_dict_prop(obj, key, value):
|
|
raw = plistlib.dumps(value)
|
|
data = _cf.CFDataCreate(None, raw, len(raw))
|
|
plist = _cf.CFPropertyListCreateWithData(None, data, 0, None, None)
|
|
_cf.CFRelease(data)
|
|
if not plist:
|
|
raise OSError("could not build the property list")
|
|
k = _cfstr(key)
|
|
err = _cm.MIDIObjectSetDictionaryProperty(obj, k, plist)
|
|
_cf.CFRelease(k)
|
|
_cf.CFRelease(plist)
|
|
return err
|
|
|
|
|
|
def _session_entities():
|
|
"""[(entity, its "apple.midirtp.session" dict)] for every network MIDI session on this Mac."""
|
|
out = []
|
|
for i in range(_cm.MIDIGetNumberOfDevices()):
|
|
dev = _cm.MIDIGetDevice(i)
|
|
if _string_prop(dev, "driver") not in NETWORK_DRIVERS:
|
|
continue
|
|
for j in range(_cm.MIDIDeviceGetNumberOfEntities(dev)):
|
|
ent = _cm.MIDIDeviceGetEntity(dev, j)
|
|
sess = _dict_prop(ent, "apple.midirtp.session")
|
|
if sess is not None:
|
|
out.append((ent, sess))
|
|
return out
|
|
|
|
|
|
def with_peer(sess, peer_name, address):
|
|
"""The session settings with this peer added, or None if it is already in the session.
|
|
Peers already there are passed back exactly as read, so they are left alone."""
|
|
peers = list(sess.get("peers", []))
|
|
if any(p.get("name") == peer_name or p.get("address") == address for p in peers):
|
|
return None
|
|
new = dict(sess)
|
|
new["peers"] = peers + [{"address": address, "name": peer_name}]
|
|
return new
|
|
|
|
|
|
def without_peer(sess, peer_name):
|
|
"""The session settings without this peer, or None if it is not in the session."""
|
|
peers = list(sess.get("peers", []))
|
|
keep = [p for p in peers if p.get("name") != peer_name]
|
|
if len(keep) == len(peers):
|
|
return None
|
|
new = dict(sess)
|
|
new["peers"] = keep
|
|
return new
|
|
|
|
|
|
def _edit_session(session_name, edit):
|
|
with _lock:
|
|
for ent, sess in _session_entities():
|
|
if (sess.get("dns-sd-name") or _string_prop(ent, "name")) != session_name:
|
|
continue
|
|
new = edit(sess)
|
|
if new is None:
|
|
return "unchanged"
|
|
err = _set_dict_prop(ent, "apple.midirtp.session", new)
|
|
if err:
|
|
raise OSError("MIDIObjectSetDictionaryProperty failed: %d" % err)
|
|
return "asked"
|
|
raise LookupError("no network MIDI session named %r on this Mac" % session_name)
|
|
|
|
|
|
LIVE_STATES = ("running", "connected") # "connected" for ~6 s while a peer joins, then "running"
|
|
|
|
|
|
def stuck_peer(sess, peer_name):
|
|
"""True if the peer is listed but not live (e.g. stuck "initializing" after a network drop)."""
|
|
return any(p.get("name") == peer_name and p.get("state") not in LIVE_STATES for p in sess.get("peers", []))
|
|
|
|
|
|
def listed_elsewhere(sess, peer_name, address):
|
|
"""True if the peer is in the session at another address (a Bonjour name, a Wi-Fi address)."""
|
|
return any(p.get("name") == peer_name and p.get("address") != address for p in sess.get("peers", []))
|
|
|
|
|
|
def connect_peer(session_name, peer_name, address, settle=time.sleep, repoint=False):
|
|
"""Ask macOS to connect `peer_name` at `address` ("10.1.10.206:5004") to the named session,
|
|
like Audio MIDI Setup's Connect. Returns "already" (it is in the session and live), "asked",
|
|
"kicked" (it was listed but stuck: dropped, then connected again) or, with `repoint`, "moved"
|
|
(it was listed at another address - a Bonjour name, Wi-Fi - and now joins at this one)."""
|
|
with _lock:
|
|
found = [(e, s) for e, s in _session_entities()
|
|
if (s.get("dns-sd-name") or _string_prop(e, "name")) == session_name]
|
|
moving = bool(found and repoint and listed_elsewhere(found[0][1], peer_name, address))
|
|
if found and (moving or stuck_peer(found[0][1], peer_name)):
|
|
ent, sess = found[0]
|
|
dropped = without_peer(sess, peer_name)
|
|
with _lock:
|
|
err = _set_dict_prop(ent, "apple.midirtp.session", dropped)
|
|
if err:
|
|
raise OSError("MIDIObjectSetDictionaryProperty failed: %d" % err)
|
|
settle(1.0)
|
|
# Built from what was just written, not a fresh read that may still list the old entry.
|
|
with _lock:
|
|
err = _set_dict_prop(ent, "apple.midirtp.session", with_peer(dropped, peer_name, address))
|
|
if err:
|
|
raise OSError("MIDIObjectSetDictionaryProperty failed: %d" % err)
|
|
return "moved" if moving else "kicked"
|
|
result = _edit_session(session_name, lambda sess: with_peer(sess, peer_name, address))
|
|
return "already" if result == "unchanged" else result
|
|
|
|
|
|
def disconnect_peer(session_name, peer_name):
|
|
"""Drop a peer from the named session, like Audio MIDI Setup's Disconnect (to test the alarms).
|
|
Returns "not connected" or "asked"."""
|
|
result = _edit_session(session_name, lambda sess: without_peer(sess, peer_name))
|
|
return "not connected" if result == "unchanged" else result
|
|
|
|
|
|
# The same read as sessions(), for a Mac without python (no Command Line Tools): macOS's own
|
|
# JavaScript for Automation calls CoreMIDI directly, nothing to install. Prints the sessions as JSON.
|
|
# Older macOS keeps no "state" for a peer; a peer without one is listed as configured, no verdict.
|
|
JXA_SESSIONS = r"""
|
|
ObjC.import('CoreMIDI'); ObjC.import('Foundation');
|
|
function str(o, k) { var r = Ref(); if ($.MIDIObjectGetStringProperty(o, k, r) !== 0) return null; return ObjC.unwrap(ObjC.castRefToObject(r[0])); }
|
|
function dict(o, k) { var r = Ref(); if ($.MIDIObjectGetDictionaryProperty(o, k, r) !== 0) return null; return ObjC.deepUnwrap(ObjC.castRefToObject(r[0])); }
|
|
var out = {}, n = $.MIDIGetNumberOfDevices();
|
|
for (var i = 0; i < n; i++) {
|
|
var dev = $.MIDIGetDevice(i), drv = str(dev, $('driver')) || '';
|
|
if (drv.indexOf('Network') < 0 && drv.toLowerCase().indexOf('rtp') < 0) continue;
|
|
for (var j = 0; j < $.MIDIDeviceGetNumberOfEntities(dev); j++) {
|
|
var ent = $.MIDIDeviceGetEntity(dev, j), s = dict(ent, $('apple.midirtp.session'));
|
|
if (!s) continue;
|
|
out[s['dns-sd-name'] || str(ent, $('name')) || '?'] = {port: s.port, peers: (s.peers || []).map(function (p) {
|
|
var q = {name: p.name, address: p.address}; if (p.state !== undefined) q.state = p.state; return q; })};
|
|
}
|
|
}
|
|
'SPMIDI' + JSON.stringify(out);
|
|
"""
|
|
|
|
|
|
|
|
# Connect, for a Mac without python: what connect_peer does, in JavaScript for Automation. Filled in
|
|
# with jxa_connect(); prints SPMIDI"already" / "asked" / "kicked" / "moved", like connect_peer.
|
|
JXA_CONNECT = r"""
|
|
ObjC.import('CoreMIDI'); ObjC.import('Foundation');
|
|
var SESSION = __SESSION__, PEER = __PEER__, ADDRESS = __ADDRESS__, REPOINT = __REPOINT__;
|
|
function str(o, k) { var r = Ref(); if ($.MIDIObjectGetStringProperty(o, k, r) !== 0) return null; return ObjC.unwrap(ObjC.castRefToObject(r[0])); }
|
|
function dict(o, k) { var r = Ref(); if ($.MIDIObjectGetDictionaryProperty(o, k, r) !== 0) return null; return ObjC.deepUnwrap(ObjC.castRefToObject(r[0])); }
|
|
function put(ent, s) { var e = $.MIDIObjectSetDictionaryProperty(ent, $('apple.midirtp.session'), $(s)); if (e !== 0) throw new Error('MIDIObjectSetDictionaryProperty failed: ' + e); }
|
|
var found = null, n = $.MIDIGetNumberOfDevices();
|
|
for (var i = 0; i < n && !found; i++) {
|
|
var dev = $.MIDIGetDevice(i), drv = str(dev, $('driver')) || '';
|
|
if (drv.indexOf('Network') < 0 && drv.toLowerCase().indexOf('rtp') < 0) continue;
|
|
for (var j = 0; j < $.MIDIDeviceGetNumberOfEntities(dev); j++) {
|
|
var ent = $.MIDIDeviceGetEntity(dev, j), s = dict(ent, $('apple.midirtp.session'));
|
|
if (s && (s['dns-sd-name'] || str(ent, $('name'))) === SESSION) { found = [ent, s]; break; }
|
|
}
|
|
}
|
|
if (!found) throw new Error('no network MIDI session named ' + SESSION + ' on this Mac');
|
|
var ent = found[0], s = found[1], peers = s.peers || [], result;
|
|
var mine = peers.filter(function (p) { return p.name === PEER; });
|
|
var stuck = mine.some(function (p) { return p.state !== undefined && p.state !== 'running' && p.state !== 'connected'; });
|
|
var elsewhere = mine.some(function (p) { return p.address !== ADDRESS; });
|
|
if (mine.length && (stuck || (REPOINT && elsewhere))) {
|
|
s.peers = peers.filter(function (p) { return p.name !== PEER; });
|
|
put(ent, s); delay(1);
|
|
s.peers = s.peers.concat([{address: ADDRESS, name: PEER}]); put(ent, s);
|
|
result = (REPOINT && elsewhere) ? 'moved' : 'kicked';
|
|
} else if (mine.length || peers.some(function (p) { return p.address === ADDRESS; })) {
|
|
result = 'already';
|
|
} else {
|
|
s.peers = peers.concat([{address: ADDRESS, name: PEER}]); put(ent, s); result = 'asked';
|
|
}
|
|
'SPMIDI' + JSON.stringify(result);
|
|
"""
|
|
|
|
|
|
def jxa_connect(session_name, peer_name, address, repoint=False):
|
|
"""The JXA_CONNECT script for one peer (values go in as JSON, never as code)."""
|
|
import json as _json
|
|
return (JXA_CONNECT.replace("__SESSION__", _json.dumps(session_name)).replace("__PEER__", _json.dumps(peer_name))
|
|
.replace("__ADDRESS__", _json.dumps(address)).replace("__REPOINT__", "true" if repoint else "false"))
|
|
|
|
def sessions():
|
|
"""{session name: {"port": int, "peers": [{"name", "address", "state"}]}}"""
|
|
found = {}
|
|
with _lock:
|
|
for ent, sess in _session_entities():
|
|
name = sess.get("dns-sd-name") or _string_prop(ent, "name") or "?"
|
|
found[name] = {
|
|
"port": sess.get("port"),
|
|
"peers": [
|
|
{"name": p.get("name", ""), "address": p.get("address", ""), "state": p.get("state", "")}
|
|
for p in sess.get("peers", [])
|
|
],
|
|
}
|
|
return found
|
|
|
|
|
|
def source_names():
|
|
with _lock:
|
|
return [_string_prop(_cm.MIDIGetSource(i), "displayName") or "" for i in range(_cm.MIDIGetNumberOfSources())]
|
|
|
|
|
|
# --- Listener -----------------------------------------------------------
|
|
|
|
class _MIDIPacket(ctypes.Structure):
|
|
_pack_ = 4
|
|
_fields_ = [("timeStamp", ctypes.c_uint64), ("length", ctypes.c_uint16), ("data", ctypes.c_ubyte * 256)]
|
|
|
|
|
|
class _MIDIPacketList(ctypes.Structure):
|
|
_pack_ = 4
|
|
_fields_ = [("numPackets", ctypes.c_uint32), ("packet", _MIDIPacket * 1)]
|
|
|
|
|
|
_READPROC = ctypes.CFUNCTYPE(None, ctypes.POINTER(_MIDIPacketList), ctypes.c_void_p, ctypes.c_void_p)
|
|
|
|
_cm.MIDIClientCreate.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p, ctypes.POINTER(_U32)]
|
|
_cm.MIDIInputPortCreate.argtypes = [_U32, ctypes.c_void_p, _READPROC, ctypes.c_void_p, ctypes.POINTER(_U32)]
|
|
_cm.MIDIPortConnectSource.argtypes = [_U32, _U32, ctypes.c_void_p]
|
|
_cm.MIDIPortDisconnectSource.argtypes = [_U32, _U32]
|
|
|
|
_client = None
|
|
|
|
|
|
def _get_client():
|
|
global _client
|
|
with _lock:
|
|
if _client is None:
|
|
ref = _U32()
|
|
name = _cfstr("status-panel")
|
|
err = _cm.MIDIClientCreate(name, None, None, ctypes.byref(ref))
|
|
_cf.CFRelease(name)
|
|
if err:
|
|
raise OSError("MIDIClientCreate failed: %d" % err)
|
|
_client = ref.value
|
|
return _client
|
|
|
|
|
|
def describe(data):
|
|
"""Short human label for a MIDI message: 'NOTE 36', 'CC 7=64', 'PC 5', 'MTC'."""
|
|
if not data:
|
|
return ""
|
|
st = data[0] & 0xF0
|
|
if st == 0x90 and len(data) >= 3 and data[2]:
|
|
return "NOTE %d" % data[1]
|
|
if st in (0x80, 0x90) and len(data) >= 2:
|
|
return "OFF %d" % data[1]
|
|
if st == 0xB0 and len(data) >= 3:
|
|
return "CC %d=%d" % (data[1], data[2])
|
|
if st == 0xC0 and len(data) >= 2:
|
|
return "PC %d" % data[1]
|
|
if data[0] == 0xF1:
|
|
return "MTC"
|
|
if data[0] == 0xF0:
|
|
return "SYSEX"
|
|
return "0x%02X" % data[0]
|
|
|
|
|
|
class Listener:
|
|
"""Connects an input port to the named source (re-checked on poll())."""
|
|
|
|
def __init__(self, source_name):
|
|
self.source_name = source_name
|
|
self.last_time = None
|
|
self.last_msg = ""
|
|
self.count = 0
|
|
self._endpoint = None
|
|
self._port = None
|
|
self._proc = _READPROC(self._on_packets) # keep a reference or it is collected
|
|
|
|
def _on_packets(self, pktlist, _ref, _src):
|
|
try:
|
|
pl = pktlist.contents
|
|
if pl.numPackets:
|
|
pkt = pl.packet[0]
|
|
n = min(pkt.length, 256)
|
|
data = bytes(pkt.data[:n])
|
|
# Skip active sensing / clock so they do not mask real traffic.
|
|
if data and data[0] not in (0xF8, 0xFE):
|
|
self.last_msg = describe(data)
|
|
self.last_time = time.time()
|
|
self.count += 1
|
|
except Exception: # never raise into CoreMIDI's thread
|
|
pass
|
|
|
|
def _find(self):
|
|
with _lock:
|
|
for i in range(_cm.MIDIGetNumberOfSources()):
|
|
ep = _cm.MIDIGetSource(i)
|
|
if _string_prop(ep, "displayName") == self.source_name or _string_prop(ep, "name") == self.source_name:
|
|
return ep
|
|
return None
|
|
|
|
@property
|
|
def connected(self):
|
|
return self._endpoint is not None
|
|
|
|
def poll(self):
|
|
"""(Re)connect if the source appeared, disappeared or changed."""
|
|
with _lock:
|
|
self._poll()
|
|
|
|
def _poll(self):
|
|
ep = self._find()
|
|
if ep == self._endpoint:
|
|
return
|
|
if self._port is None:
|
|
ref = _U32()
|
|
name = _cfstr("status-panel in")
|
|
err = _cm.MIDIInputPortCreate(_get_client(), name, self._proc, None, ctypes.byref(ref))
|
|
_cf.CFRelease(name)
|
|
if err:
|
|
raise OSError("MIDIInputPortCreate failed: %d" % err)
|
|
self._port = ref.value
|
|
if self._endpoint is not None:
|
|
_cm.MIDIPortDisconnectSource(self._port, self._endpoint)
|
|
self._endpoint = None
|
|
if ep is not None and _cm.MIDIPortConnectSource(self._port, ep, None) == 0:
|
|
self._endpoint = ep
|
|
|
|
|
|
# --- test helper: a virtual source we can send into ----------------------
|
|
|
|
_cm.MIDISourceCreate.argtypes = [_U32, ctypes.c_void_p, ctypes.POINTER(_U32)]
|
|
_cm.MIDIReceived.argtypes = [_U32, ctypes.POINTER(_MIDIPacketList)]
|
|
_cm.MIDIEndpointDispose.argtypes = [_U32]
|
|
|
|
|
|
class VirtualSource:
|
|
"""Only used by the tests: a source whose messages a Listener can receive."""
|
|
|
|
def __init__(self, name):
|
|
ref = _U32()
|
|
cname = _cfstr(name)
|
|
with _lock:
|
|
err = _cm.MIDISourceCreate(_get_client(), cname, ctypes.byref(ref))
|
|
_cf.CFRelease(cname)
|
|
if err:
|
|
raise OSError("MIDISourceCreate failed: %d" % err)
|
|
self.ref = ref.value
|
|
|
|
def send(self, data):
|
|
pl = _MIDIPacketList()
|
|
pl.numPackets = 1
|
|
pl.packet[0].timeStamp = 0
|
|
pl.packet[0].length = len(data)
|
|
for i, b in enumerate(data):
|
|
pl.packet[0].data[i] = b
|
|
_cm.MIDIReceived(self.ref, ctypes.byref(pl))
|
|
|
|
def close(self):
|
|
_cm.MIDIEndpointDispose(self.ref)
|