Add a function get_new_events() which is like call_on_idle except it returns a list.
[calfbox.git] / py / cbox.py
blob3a4b059c49ae0fc40d357c524d5dae53e67da7ad
1 from _cbox import *
2 import struct
4 class GetUUID:
5 def __init__(self):
6 def callback(cmd, fb, args):
7 if cmd == "/uuid" and len(args) == 1:
8 self.uuid = args[0]
9 else:
10 raise ValueException("Unexpected callback: %s" % cmd)
11 self.callback = callback
12 def __call__(self, *args):
13 self.callback(*args)
15 class GetThings:
16 @staticmethod
17 def by_uuid(uuid, cmd, anames, args):
18 return GetThings(Document.uuid_cmd(uuid, cmd), anames, args)
19 def __init__(self, cmd, anames, args):
20 for i in anames:
21 if i.startswith("*"):
22 setattr(self, i[1:], [])
23 elif i.startswith("%"):
24 setattr(self, i[1:], {})
25 else:
26 setattr(self, i, None)
27 anames = set(anames)
28 self.seq = []
29 def update_callback(cmd, fb, args):
30 self.seq.append((cmd, fb, args))
31 cmd = cmd[1:]
32 if cmd in anames:
33 if len(args) == 1:
34 setattr(self, cmd, args[0])
35 else:
36 setattr(self, cmd, args)
37 elif "*" + cmd in anames:
38 if len(args) == 1:
39 getattr(self, cmd).append(args[0])
40 else:
41 getattr(self, cmd).append(args)
42 elif "%" + cmd in anames:
43 if len(args) == 2:
44 getattr(self, cmd)[args[0]] = args[1]
45 else:
46 getattr(self, cmd)[args[0]] = args[1:]
47 elif len(args) == 1:
48 setattr(self, cmd, args[0])
49 do_cmd(cmd, update_callback, args)
50 def __str__(self):
51 return str(self.seq)
53 class VarPath:
54 def __init__(self, path, args = []):
55 self.path = path
56 self.args = args
57 def plus(self, subpath, *args):
58 return VarPath(self.path if subpath is None else self.path + "/" + subpath, self.args + list(args))
59 def set(self, *values):
60 do_cmd(self.path, None, self.args + list(values))
62 class Config:
63 @staticmethod
64 def sections(prefix = ""):
65 return [CfgSection(name) for name in GetThings('/config/sections', ['*section'], [str(prefix)]).section]
67 @staticmethod
68 def keys(section, prefix = ""):
69 return GetThings('/config/keys', ['*key'], [str(section), str(prefix)]).key
71 @staticmethod
72 def get(section, key):
73 return GetThings('/config/get', ['value'], [str(section), str(key)]).value
75 @staticmethod
76 def set(section, key, value):
77 do_cmd('/config/set', None, [str(section), str(key), str(value)])
79 @staticmethod
80 def delete(section, key):
81 do_cmd('/config/delete', None, [str(section), str(key)])
83 @staticmethod
84 def save(filename = None):
85 if filename is None:
86 do_cmd('/config/save', None, [])
87 else:
88 do_cmd('/config/save', None, [str(filename)])
90 @staticmethod
91 def add_section(section, content):
92 for line in content.splitlines():
93 line = line.strip()
94 if line == '' or line.startswith('#'):
95 continue
96 try:
97 key, value = line.split("=", 2)
98 except ValueError as err:
99 raise ValueError("Cannot parse config line '%s'" % line)
100 Config.set(section, key.strip(), value.strip())
102 class Transport:
103 @staticmethod
104 def seek_ppqn(ppqn):
105 do_cmd('/master/seek_ppqn', None, [int(ppqn)])
106 @staticmethod
107 def seek_samples(samples):
108 do_cmd('/master/seek_samples', None, [int(samples)])
109 @staticmethod
110 def set_tempo(tempo):
111 do_cmd('/master/set_tempo', None, [float(tempo)])
112 @staticmethod
113 def set_timesig(nom, denom):
114 do_cmd('/master/set_timesig', None, [int(nom), int(denom)])
115 @staticmethod
116 def play():
117 do_cmd('/master/play', None, [])
118 @staticmethod
119 def stop():
120 do_cmd('/master/stop', None, [])
121 @staticmethod
122 def panic():
123 do_cmd('/master/panic', None, [])
124 @staticmethod
125 def status():
126 return GetThings("/master/status", ['pos', 'pos_ppqn', 'tempo', 'timesig', 'sample_rate', 'playing'], [])
127 @staticmethod
128 def tell():
129 return GetThings("/master/tell", ['pos', 'pos_ppqn', 'playing'], [])
131 # Currently responsible for both JACK and USB I/O - not all functionality is
132 # supported by both.
133 class JackIO:
134 AUDIO_TYPE = "32 bit float mono audio"
135 MIDI_TYPE = "8 bit raw midi"
136 PORT_IS_SINK = 0x1
137 PORT_IS_SOURCE = 0x2
138 PORT_IS_PHYSICAL = 0x4
139 PORT_CAN_MONITOR = 0x8
140 PORT_IS_TERMINAL = 0x10
141 @staticmethod
142 def status():
143 # Some of these only make sense for
144 return GetThings("/io/status", ['client_type', 'client_name', 'audio_inputs', 'audio_outputs', 'buffer_size', '*midi_output', '*midi_input', 'sample_rate', 'output_resolution'], [])
145 @staticmethod
146 def create_midi_output(name, autoconnect_spec = None):
147 fb = GetUUID()
148 do_cmd("/io/create_midi_output", fb, [name])
149 uuid = fb.uuid
150 if autoconnect_spec is not None and autoconnect_spec != '':
151 JackIO.autoconnect(uuid, autoconnect_spec)
152 return uuid
153 @staticmethod
154 def autoconnect_midi_output(uuid, autoconnect_spec = None):
155 if autoconnect_spec is not None:
156 do_cmd("/io/autoconnect", None, [uuid, autoconnect_spec])
157 else:
158 do_cmd("/io/autoconnect", None, [uuid, ''])
159 @staticmethod
160 def rename_midi_output(uuid, new_name):
161 do_cmd("/io/rename_midi_output", None, [uuid, new_name])
162 @staticmethod
163 def disconnect_midi_output(uuid):
164 do_cmd("/io/disconnect_midi_output", None, [uuid])
165 @staticmethod
166 def delete_midi_output(uuid):
167 do_cmd("/io/delete_midi_output", None, [uuid])
168 @staticmethod
169 def port_connect(pfrom, pto):
170 do_cmd("/io/port_connect", None, [pfrom, pto])
171 @staticmethod
172 def port_disconnect(pfrom, pto):
173 do_cmd("/io/port_disconnect", None, [pfrom, pto])
174 @staticmethod
175 def get_ports(name_mask = ".*", type_mask = ".*", flag_mask = 0):
176 return GetThings("/io/get_ports", ['*port'], [name_mask, type_mask, int(flag_mask)]).port
178 def call_on_idle(callback = None):
179 do_cmd("/on_idle", callback, [])
181 def get_new_events():
182 return GetThings("/on_idle", [], []).seq
184 class CfgSection:
185 def __init__(self, name):
186 self.name = name
188 def __getitem__(self, key):
189 return Config.get(self.name, key)
191 def __setitem__(self, key, value):
192 Config.set(self.name, key, value)
194 def __delitem__(self, key):
195 Config.delete(self.name, key)
197 def keys(self, prefix = ""):
198 return Config.keys(self.name, prefix)
201 class Pattern:
202 @staticmethod
203 def get_pattern():
204 pat_data = GetThings("/get_pattern", ['pattern'], []).pattern
205 if pat_data is not None:
206 pat_blob, length = pat_data
207 pat_data = []
208 ofs = 0
209 while ofs < len(pat_blob):
210 data = list(struct.unpack_from("iBBbb", pat_blob, ofs))
211 data[1:2] = []
212 pat_data.append(tuple(data))
213 ofs += 8
214 return pat_data, length
215 return None
217 @staticmethod
218 def serialize_event(time, *data):
219 if len(data) >= 1 and len(data) <= 3:
220 return struct.pack("iBBbb"[0:2 + len(data)], int(time), len(data), *[int(v) for v in data])
221 raise ValueError("Invalid length of an event (%d)" % len(data))
223 class Document:
224 classmap = {}
225 objmap = {}
226 @staticmethod
227 def dump():
228 do_cmd("/doc/dump", None, [])
229 @staticmethod
230 def uuid_cmd(uuid, cmd):
231 return "/doc/uuid/%s%s" % (uuid, cmd)
232 @staticmethod
233 def get_uuid(path):
234 return GetThings("%s/get_uuid" % path, ["uuid"], []).uuid
235 @staticmethod
236 def get_obj_class(uuid):
237 return GetThings(Document.uuid_cmd(uuid, "/get_class_name"), ["class_name"], []).class_name
238 @staticmethod
239 def get_song():
240 return Document.map_uuid(Document.get_uuid("/song"))
241 @staticmethod
242 def get_scene():
243 return Document.map_uuid(Document.get_uuid("/scene"))
244 @staticmethod
245 def get_rt():
246 return Document.map_uuid(Document.get_uuid("/rt"))
247 @staticmethod
248 def new_scene(srate, bufsize):
249 fb = GetUUID()
250 do_cmd("/new_scene", fb, [int(srate), int(bufsize)])
251 return Document.map_uuid(fb.uuid)
252 @staticmethod
253 def map_uuid(uuid):
254 if uuid in Document.objmap:
255 return Document.objmap[uuid]
256 try:
257 oclass = Document.get_obj_class(uuid)
258 except Exception as e:
259 print ("Note: Cannot get class for " + uuid)
260 Document.dump()
261 raise
262 o = Document.classmap[oclass](uuid)
263 Document.objmap[uuid] = o
264 if hasattr(o, 'init_object'):
265 o.init_object()
266 return o
268 class SetterMaker():
269 def __init__(self, obj, path):
270 self.obj = obj
271 self.path = path
272 def set(self, value):
273 self.obj.cmd(self.path, None, value)
274 def set2(self, key, value):
275 self.obj.cmd(self.path, None, key, value)
277 class NonDocObj(object):
278 def __init__(self, path, status_field_list):
279 self.path = path
280 self.status_fields = []
281 for sf in status_field_list:
282 if sf.startswith("="):
283 sf = sf[1:]
284 if sf.startswith("%"):
285 sf2 = sf[1:]
286 self.__dict__['set_' + sf2] = SetterMaker(self, "/" + sf2).set2
287 else:
288 self.__dict__['set_' + sf] = SetterMaker(self, "/" + sf).set
289 self.status_fields.append(sf)
291 def cmd(self, cmd, fb = None, *args):
292 do_cmd(self.path + cmd, fb, list(args))
294 def cmd_makeobj(self, cmd, *args):
295 fb = GetUUID()
296 do_cmd(self.path + cmd, fb, list(args))
297 return Document.map_uuid(fb.uuid)
299 def get_things(self, cmd, fields, *args):
300 return GetThings(self.path + cmd, fields, list(args))
302 def make_path(self, path):
303 return self.path + path
305 def status(self):
306 return self.transform_status(self.get_things("/status", self.status_fields))
308 def transform_status(self, status):
309 return status
311 class DocObj(NonDocObj):
312 def __init__(self, uuid, status_field_list):
313 NonDocObj.__init__(self, Document.uuid_cmd(uuid, ''), status_field_list)
314 self.uuid = uuid
316 def delete(self):
317 self.cmd("/delete")
319 class DocPattern(DocObj):
320 def __init__(self, uuid):
321 DocObj.__init__(self, uuid, ["event_count", "loop_end", "name"])
322 def set_name(self, name):
323 self.cmd("/name", None, name)
324 Document.classmap['cbox_midi_pattern'] = DocPattern
326 class ClipItem:
327 def __init__(self, pos, offset, length, pattern, clip):
328 self.pos = pos
329 self.offset = offset
330 self.length = length
331 self.pattern = Document.map_uuid(pattern)
332 self.clip = Document.map_uuid(clip)
333 def __str__(self):
334 return "pos=%d offset=%d length=%d pattern=%s clip=%s" % (self.pos, self.offset, self.length, self.pattern.uuid, self.clip.uuid)
335 def __eq__(self, other):
336 return str(self) == str(other)
338 class DocTrackClip(DocObj):
339 def __init__(self, uuid):
340 DocObj.__init__(self, uuid, ["pos", "offset", "length", "pattern", "uuid"])
341 def transform_status(self, status):
342 return ClipItem(status.pos, status.offset, status.length, status.pattern, status.uuid)
344 Document.classmap['cbox_track_item'] = DocTrackClip
346 class DocTrackStatus:
347 name = None
348 clips = None
349 external_output = None
351 class DocTrack(DocObj):
352 def __init__(self, uuid):
353 DocObj.__init__(self, uuid, ["*clip", "=name", "=external_output"])
354 def add_clip(self, pos, offset, length, pattern):
355 return self.cmd_makeobj("/add_clip", int(pos), int(offset), int(length), pattern.uuid)
356 def transform_status(self, status):
357 res = DocTrackStatus()
358 res.name = status.name
359 res.clips = [ClipItem(*c) for c in status.clip]
360 res.external_output = status.external_output
361 return res
362 Document.classmap['cbox_track'] = DocTrack
364 class TrackItem:
365 def __init__(self, name, count, track):
366 self.name = name
367 self.count = count
368 self.track = Document.map_uuid(track)
370 class PatternItem:
371 def __init__(self, name, length, pattern):
372 self.name = name
373 self.length = length
374 self.pattern = Document.map_uuid(pattern)
376 class DocSongStatus:
377 tracks = None
378 patterns = None
380 class DocSong(DocObj):
381 def __init__(self, uuid):
382 DocObj.__init__(self, uuid, ["*track", "*pattern", "*mti", 'loop_start', 'loop_end'])
383 def clear(self):
384 return self.cmd("/clear", None)
385 def set_loop(self, ls, le):
386 return self.cmd("/set_loop", None, int(ls), int(le))
387 def set_mti(self, pos, tempo = None, timesig_nom = None, timesig_denom = None):
388 self.cmd("/set_mti", None, int(pos), float(tempo) if tempo is not None else -1.0, int(timesig_nom) if timesig_nom is not None else -1, int(timesig_denom) if timesig_denom else -1)
389 def add_track(self):
390 return self.cmd_makeobj("/add_track")
391 def load_drum_pattern(self, name):
392 return self.cmd_makeobj("/load_pattern", name, 1)
393 def load_drum_track(self, name):
394 return self.cmd_makeobj("/load_track", name, 1)
395 def pattern_from_blob(self, blob, length):
396 return self.cmd_makeobj("/load_blob", bytearray(blob), int(length))
397 def loop_single_pattern(self, loader):
398 self.clear()
399 track = self.add_track()
400 pat = loader()
401 length = pat.status().loop_end
402 track.add_clip(0, 0, length, pat)
403 self.set_loop(0, length)
404 self.update_playback()
406 def transform_status(self, status):
407 res = DocSongStatus()
408 res.tracks = [TrackItem(*t) for t in status.track]
409 res.patterns = [PatternItem(*t) for t in status.pattern]
410 res.mtis = [tuple(t) for t in status.mti]
411 return res
412 def update_playback(self):
413 # XXXKF Maybe make it a song-level API instead of global
414 do_cmd("/update_playback", None, [])
415 Document.classmap['cbox_song'] = DocSong
417 class DocLayer(DocObj):
418 def __init__(self, uuid):
419 DocObj.__init__(self, uuid, ["name", "instrument_name", "instrument_uuid", "=enable", "=low_note", "=high_note", "=fixed_note", "=in_channel", "=out_channel", "=aftertouch", "=invert_sustain", "=consume", "=ignore_scene_transpose", "=transpose"])
420 def get_instrument(self):
421 return Document.map_uuid(self.status().instrument_uuid)
422 Document.classmap['cbox_layer'] = DocLayer
424 class SamplerEngine(NonDocObj):
425 def __init__(self, path):
426 NonDocObj.__init__(self, path, ['polyphony', 'active_voices', '%volume', '%patch', '%pan'])
427 def load_patch_from_cfg(self, patch_no, cfg_section, display_name):
428 return self.cmd_makeobj("/load_patch", int(patch_no), cfg_section, display_name)
429 def load_patch_from_string(self, patch_no, sample_dir, sfz_data, display_name):
430 return self.cmd_makeobj("/load_patch_from_string", int(patch_no), sample_dir, sfz_data, display_name)
431 def load_patch_from_file(self, patch_no, sfz_name, display_name):
432 return self.cmd_makeobj("/load_patch_from_file", int(patch_no), sfz_name, display_name)
433 def set_patch(self, channel, patch_no):
434 self.cmd("/set_patch", None, int(channel), int(patch_no))
435 def get_unused_program(self):
436 return self.get_things("/get_unused_program", ['program_no']).program_no
437 def set_polyphony(self, polyphony):
438 self.cmd("/polyphony", None, int(polyphony))
439 def get_patches(self):
440 return self.get_things("/patches", ['%patch']).patch
441 def transform_status(self, status):
442 status.patches = status.patch
443 return status
445 class FluidsynthEngine(NonDocObj):
446 def __init__(self, path):
447 NonDocObj.__init__(self, path, ['polyphony', 'soundfont', '%patch'])
448 def load_soundfont(self, filename):
449 return self.cmd_makeobj("/load_soundfont", filename)
450 def set_patch(self, channel, patch_no):
451 self.cmd("/set_patch", None, int(channel), int(patch_no))
452 def set_polyphony(self, polyphony):
453 self.cmd("/polyphony", None, int(polyphony))
454 def get_patches(self):
455 return self.get_things("/patches", ['%patch']).patch
456 def transform_status(self, status):
457 status.patches = status.patch
458 return status
460 class StreamPlayerEngine(NonDocObj):
461 def __init__(self, path):
462 NonDocObj.__init__(self, path, ['filename', 'pos', 'length', 'playing'])
463 def play(self):
464 self.cmd('/play')
465 def stop(self):
466 self.cmd('/stop')
467 def seek(self, place):
468 self.cmd('/seek', None, int(place))
469 def load(self, filename, loop_start = -1):
470 self.cmd('/load', None, filename, int(loop_start))
471 def unload(self, filename, loop_start = -1):
472 self.cmd('/unload')
474 class TonewheelOrganEngine(NonDocObj):
475 def __init__(self, path):
476 NonDocObj.__init__(self, path, ['=%upper_drawbar', '=%lower_drawbar', '=%pedal_drawbar',
477 '=upper_vibrato', '=lower_vibrato', '=vibrato_mode', '=vibrato_chorus',
478 '=percussion_enable', '=percussion_3rd'])
480 engine_classes = {
481 'sampler' : SamplerEngine,
482 'fluidsynth' : FluidsynthEngine,
483 'stream_player' : StreamPlayerEngine,
484 'tonewheel_organ' : TonewheelOrganEngine,
487 class DocInstrument(DocObj):
488 def __init__(self, uuid):
489 DocObj.__init__(self, uuid, ["name", "outputs", "aux_offset", "engine"])
490 def init_object(self):
491 engine = self.status().engine
492 if engine in engine_classes:
493 self.engine = engine_classes[engine]("/doc/uuid/" + self.uuid + "/engine")
494 Document.classmap['cbox_instrument'] = DocInstrument
496 class DocScene(DocObj):
497 def __init__(self, uuid):
498 DocObj.__init__(self, uuid, ["name", "title", "transpose", "*layer", "*instrument", '*aux'])
499 def clear(self):
500 self.cmd("/clear", None)
501 def load(self, name):
502 self.cmd("/load", None, name)
503 def load_aux(self, aux):
504 return self.cmd_makeobj("/load_aux", aux)
505 def delete_aux(self, aux):
506 return self.cmd("/delete_aux", None, aux)
507 def delete_layer(self, pos):
508 self.cmd("/delete_layer", None, int(1 + pos))
509 def move_layer(self, old_pos, new_pos):
510 self.cmd("/move_layer", None, int(old_pos + 1), int(new_pos + 1))
512 def add_layer(self, aux, pos = None):
513 if pos is None:
514 return self.cmd_makeobj("/add_layer", 0, aux)
515 else:
516 # Note: The positions in high-level API are zero-based.
517 return self.cmd_makeobj("/add_layer", int(1 + pos), aux)
518 def add_instrument_layer(self, name, pos = None):
519 if pos is None:
520 return self.cmd_makeobj("/add_instrument_layer", 0, name)
521 else:
522 return self.cmd_makeobj("/add_instrument_layer", int(1 + pos), name)
523 def add_new_instrument_layer(self, name, engine, pos = None):
524 if pos is None:
525 return self.cmd_makeobj("/add_new_instrument_layer", 0, name, engine)
526 else:
527 return self.cmd_makeobj("/add_new_instrument_layer", int(1 + pos), name, engine)
528 def transform_status(self, status):
529 status.layers = [Document.map_uuid(i) for i in status.layer]
530 delattr(status, 'layer')
531 status.auxes = dict([(name, Document.map_uuid(uuid)) for name, uuid in status.aux])
532 delattr(status, 'aux')
533 status.instruments = dict([(name, (engine, Document.map_uuid(uuid))) for name, engine, uuid in status.instrument])
534 delattr(status, 'instrument')
535 return status
536 Document.classmap['cbox_scene'] = DocScene
538 class DocRt(DocObj):
539 def __init__(self, uuid):
540 DocObj.__init__(self, uuid, ["name", "instrument_name", "instrument_uuid", "=enable", "=low_note", "=high_note", "=fixed_note", "=in_channel", "=out_channel", "=aftertouch", "=invert_sustain", "=consume", "=ignore_scene_transpose"])
541 Document.classmap['cbox_rt'] = DocRt
543 class DocAuxBus(DocObj):
544 def __init__(self, uuid):
545 DocObj.__init__(self, uuid, ["name"])
546 #def transform_status(self, status):
547 # status.slot = Document.map_uuid(status.slot_uuid)
548 # return status
549 def get_slot_engine(self):
550 return self.cmd_makeobj("/slot/engine/get_uuid")
551 def get_slot_status(self):
552 return self.get_things("/slot/status", ["insert_preset", "insert_engine"])
553 Document.classmap['cbox_aux_bus'] = DocAuxBus
555 class DocModule(DocObj):
556 def __init__(self, uuid):
557 DocObj.__init__(self, uuid, [])
558 Document.classmap['cbox_module'] = DocModule
560 class SamplerProgram(DocObj):
561 def __init__(self, uuid):
562 DocObj.__init__(self, uuid, [])
563 def get_regions(self):
564 return map(Document.map_uuid, self.get_things("/regions", ['*region']).region)
565 def get_groups(self):
566 g = self.get_things("/groups", ['*group', 'default_group'])
567 return [Document.map_uuid(g.default_group)] + list(map(Document.map_uuid, g.group))
568 def new_group(self):
569 return self.cmd_makeobj("/new_group")
570 Document.classmap['sampler_program'] = SamplerProgram
572 class SamplerLayer(DocObj):
573 def __init__(self, uuid):
574 DocObj.__init__(self, uuid, ['parent_program', 'parent_group'])
575 def get_children(self):
576 return map(Document.map_uuid, self.get_things("/get_children", ['*region']).region)
577 def as_string(self):
578 return self.get_things("/as_string", ['value']).value
579 def as_string_full(self):
580 return self.get_things("/as_string_full", ['value']).value
581 def set_param(self, key, value):
582 self.cmd("/set_param", None, key, str(value))
583 def new_region(self):
584 return self.cmd_makeobj("/new_region")
585 Document.classmap['sampler_layer'] = SamplerLayer