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
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
|
from ._core import *
import time
import threading
class MidiListener(threading.Thread):
"""
Self-contained midi listener thread.
This object allows to setup a Midi server that is independent
of the audio server (mainly to be able to receive Midi data even
when the audio server is stopped). Although it runs in a separated
thread, the same device can't be used by this object and the audio
server at the same time. It is adviced to call the deactivateMidi()
method on the audio server to avoid conflicts.
:Parent: threading.Thread
:Args:
function: Python function (can't be a list)
Function that will be called when a new midi event is available.
This function is called with the incoming midi data as
arguments. The signature of the function must be:
def myfunc(status, data1, data2)
mididev: int or list of ints, optional
Sets the midi input device (see `pm_list_devices()` for the
available devices). The default, -1, means the system default
device. A number greater than the highest portmidi device index
will open all available input devices. Specific devices can be
set with a list of integers.
reportdevice: boolean, optional
If True, the device ID will be reported as a fourth argument to
the callback. The signature will then be:
def myfunc(status, data1, data2, id)
Available at initialization only. Defaults to False.
.. note::
This object is available only if pyo is built with portmidi support
(see withPortmidi function).
>>> s = Server()
>>> s.deactivateMidi()
>>> s.boot()
>>> def midicall(status, data1, data2):
... print(status, data1, data2)
>>> listen = MidiListener(midicall, 5)
>>> listen.start()
"""
def __init__(self, function, mididev=-1, reportdevice=False):
threading.Thread.__init__(self)
self.daemon = True
self._function = WeakMethod(function)
self._mididev = mididev
if type(mididev) is not list:
mididev = [mididev]
self._reportdevice = reportdevice
self._listener = MidiListener_base(self._function, mididev, self._reportdevice)
def run(self):
"""
Starts the process. The thread runs as daemon, so no need to stop it.
"""
self._listener.play()
while True:
try:
time.sleep(0.001)
except:
pass
def stop(self):
"""
Stops the listener and properly close the midi ports.
"""
self._listener.stop()
def getDeviceInfos(self):
"""
Returns infos about connected midi devices.
This method returns a list of dictionaries, one per device.
Dictionary format is:
{"id": device_id (int), "name": device_name (str), "interface": interface (str)}
"""
infos = self._listener.getDeviceInfos()
if infos:
lst = []
for info in infos:
dct = {}
items = info.split(", ")
for item in items:
isplit = item.split(": ")
dct[isplit[0]] = isplit[1]
dct["id"] = int(dct["id"])
lst.append(dct)
return lst
return []
class MidiDispatcher(threading.Thread):
"""
Self-contained midi dispatcher thread.
This object allows to setup a Midi server that is independent
of the audio server (mainly to be able to send Midi data even
when the audio server is stopped). Although it runs in a separated
thread, the same device can't be used by this object and the audio
server at the same time. It is adviced to call the deactivateMidi()
method on the audio server to avoid conflicts.
Use the `send` method to send midi event to connected devices.
Use the `sendx` method to send sysex event to connected devices.
:Parent: threading.Thread
:Args:
mididev: int or list of ints, optional
Sets the midi output device (see `pm_list_devices()` for the
available devices). The default, -1, means the system default
device. A number greater than the highest portmidi device index
will open all available input devices. Specific devices can be
set with a list of integers.
.. note::
This object is available only if pyo is built with portmidi support
(see withPortmidi function).
>>> s = Server()
>>> s.deactivateMidi()
>>> s.boot()
>>> dispatch = MidiDispatcher(5)
>>> dispatch.start()
>>> dispatch.send(144, 60, 127)
"""
def __init__(self, mididev=-1):
threading.Thread.__init__(self)
self.daemon = True
self._mididev = mididev
if type(mididev) is not list:
mididev = [mididev]
self._dispatcher = MidiDispatcher_base(mididev)
def run(self):
"""
Starts the process. The thread runs as daemon, so no need to stop it.
"""
self._dispatcher.play()
while True:
try:
time.sleep(0.001)
except:
pass
def send(self, status, data1, data2=0, timestamp=0, device=-1):
"""
Send a MIDI message to the selected midi output device.
Arguments can be list of values to generate multiple events
in one call.
:Args:
status: int
Status byte.
data1: int
First data byte.
data2: int, optional
Second data byte. Defaults to 0.
timestamp: int, optional
The delay time, in milliseconds, before the note
is sent on the portmidi stream. A value of 0 means
to play the note now. Defaults to 0.
device: int, optional
The index of the device to which the message will
be sent. The default (-1) means all devices. See
`getDeviceInfos()` to retrieve device indexes.
"""
status, data1, data2, timestamp, device, lmax = convertArgsToLists(status, data1, data2, timestamp, device)
[
self._dispatcher.send(wrap(status, i), wrap(data1, i), wrap(data2, i), wrap(timestamp, i), wrap(device, i))
for i in range(lmax)
]
def sendx(self, msg, timestamp=0, device=-1):
"""
Send a MIDI system exclusive message to the selected midi output device.
Arguments can be list of values to generate multiple events
in one call.
:Args:
msg: str
A valid system exclusive message as a string. The first byte
must be 0xf0 and the last one must be 0xf7.
timestamp: int, optional
The delay time, in milliseconds, before the note
is sent on the portmidi stream. A value of 0 means
to play the note now. Defaults to 0.
device: int, optional
The index of the device to which the message will
be sent. The default (-1) means all devices. See
`getDeviceInfos()` to retrieve device indexes.
"""
msg, timestamp, device, lmax = convertArgsToLists(msg, timestamp, device)
[self._dispatcher.sendx(wrap(msg, i), wrap(timestamp, i), wrap(device, i)) for i in range(lmax)]
def getDeviceInfos(self):
"""
Returns infos about connected midi devices.
This method returns a list of dictionaries, one per device.
Dictionary format is:
{"id": device_id (int), "name": device_name (str), "interface": interface (str)}
"""
infos = self._dispatcher.getDeviceInfos()
if infos:
lst = []
for info in infos:
dct = {}
items = info.split(", ")
for item in items:
isplit = item.split(": ")
dct[isplit[0]] = isplit[1]
dct["id"] = int(dct["id"])
lst.append(dct)
return lst
return []
OscListenerLock = threading.Lock()
class OscListener(threading.Thread):
"""
Self-contained OSC listener thread.
This object allows to setup an OSC server that is independent
of the audio server (mainly to be able to receive OSC data even
when the audio server is stopped).
:Parent: threadind.Thread
:Args:
function: Python function (can't be a list)
Function that will be called when a new OSC event is available.
This function is called with the incoming address and values as
arguments. The signature of the function must be::
def myfunc(address, *args)
port: int, optional
The OSC port on which the values are received. Defaults to 9000.
>>> s = Server().boot()
>>> def call(address, *args):
... print(address, args)
>>> listen = OscListener(call, 9901)
>>> listen.start()
"""
def __init__(self, function, port=9000):
threading.Thread.__init__(self)
self.daemon = True
self._function = WeakMethod(function)
self._port = port
self._listener = OscListener_base(self._oscrecv, self._port)
def _oscrecv(self, address, *args):
with OscListenerLock:
self._function(address, *args)
def run(self):
"""
Starts the process. The thread runs as daemon, so no need to stop it.
"""
while True:
self._listener.get()
try:
time.sleep(0.001)
except:
pass
|
