You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

1218 lines
46 KiB

from io import BytesIO
import struct
import sys
import traceback
try:
from _cbox2 import * #local file _cbox2.py
import metadata #local file metadata.py
except ModuleNotFoundError:
from ._cbox2 import *
from . import metadata #local file metadata.py
type_wrapper_debug = False
is_python3 = not sys.version.startswith("2")
###############################################################################
# Ugly internals. Please skip this section for your own sanity.
###############################################################################
class GetUUID:
"""An object that calls a C layer command, receives a /uuid callback from it
and stores the passed UUID in its uuid attribute.
Example use: GetUUID('/command', arg1, arg2...).uuid
"""
def __init__(self, cmd, *cmd_args):
def callback(cmd, fb, args):
if cmd == "/uuid" and len(args) == 1:
self.uuid = args[0]
else:
raise ValueException("Unexpected callback: %s" % cmd)
self.callback = callback
self.uuid = None
do_cmd(cmd, self, list(cmd_args))
def __call__(self, *args):
self.callback(*args)
class GetThings:
"""A generic callback object that receives various forms of information from
C layer and converts then into object's Python attributes.
This is an obsolete interface, to be replaced by GetUUID or metaclass
based type-safe autoconverter. However, there are still some cases that
aren't (yet) handled by either.
"""
@staticmethod
def by_uuid(uuid, cmd, anames, args):
return GetThings(Document.uuid_cmd(uuid, cmd), anames, args)
def __init__(self, cmd, anames, args):
for i in anames:
if i.startswith("*"):
setattr(self, i[1:], [])
elif i.startswith("%"):
setattr(self, i[1:], {})
else:
setattr(self, i, None)
anames = set(anames)
self.seq = []
def update_callback(cmd, fb, args):
self.seq.append((cmd, fb, args))
cmd = cmd[1:]
if cmd in anames:
if len(args) == 1:
setattr(self, cmd, args[0])
else:
setattr(self, cmd, args)
elif "*" + cmd in anames:
if len(args) == 1:
getattr(self, cmd).append(args[0])
else:
getattr(self, cmd).append(args)
elif "%" + cmd in anames:
if len(args) == 2:
getattr(self, cmd)[args[0]] = args[1]
else:
getattr(self, cmd)[args[0]] = args[1:]
elif "?" + cmd in anames:
setattr(self, cmd, bool(args[0]))
elif len(args) == 1:
setattr(self, cmd, args[0])
do_cmd(cmd, update_callback, args)
def __str__(self):
return str(self.seq)
class PropertyDecorator(object):
"""Abstract property decorator."""
def __init__(self, base):
self.base = base
def get_base(self):
return self.base
def map_cmd(self, cmd):
return cmd
class AltPropName(PropertyDecorator):
"""Command-name-changing property decorator. Binds a property to the
specified /path, different from the default one, which based on property name,
with -s and -es suffix removed for lists and dicts."""
def __init__(self, alt_name, base):
PropertyDecorator.__init__(self, base)
self.alt_name = alt_name
def map_cmd(self, cmd):
return self.alt_name
def execute(self, property, proptype, klass):
pass
class SettableProperty(PropertyDecorator):
"""Decorator that creates a setter method for the property."""
def execute(self, property, proptype, klass):
if type(proptype) is dict:
setattr(klass, 'set_' + property, lambda self, key, value: self.cmd('/' + property, None, key, value))
elif type(proptype) is bool:
setattr(klass, 'set_' + property, lambda self, value: self.cmd('/' + property, None, 1 if value else 0))
elif issubclass(proptype, DocObj):
setattr(klass, 'set_' + property, lambda self, value: self.cmd('/' + property, None, value.uuid))
else:
setattr(klass, 'set_' + property, lambda self, value: self.cmd('/' + property, None, proptype(value)))
def new_get_things(obj, cmd, settermap, args):
"""Call C command with arguments 'args', populating a return object obj
using settermap to interpret callback commands and initialise the return
object."""
def update_callback(cmd2, fb, args2):
try:
if cmd2 in settermap:
settermap[cmd2](obj, args2)
elif cmd2 != '/uuid': # Ignore UUID as it's usually safe to do so
print ("Unexpected command: %s" % cmd2)
except Exception as error:
traceback.print_exc()
raise
# Set initial values for the properties (None or empty dict/list)
for setterobj in settermap.values():
setattr(obj, setterobj.property, setterobj.init_value())
# Call command and apply callback commands via setters to the object
do_cmd(cmd, update_callback, args)
return obj
def _error_arg_mismatch(required, passed):
raise ValueError("Types required: %s, values passed: %s" % (repr(required), repr(passed)))
def _handle_object_wrapping(t):
if issubclass(t, DocObj):
return lambda uuid: Document.map_uuid_and_check(uuid, t)
return t
def _make_args_to_type_lambda(t):
t = _handle_object_wrapping(t)
return lambda args: t(*args)
def _make_args_to_tuple_of_types_lambda(ts):
ts = list(map(_handle_object_wrapping, ts))
return lambda args: tuple([ts[i](args[i]) for i in range(max(len(ts), len(args)))]) if len(ts) == len(args) else _error_arg_mismatch(ts, args)
def _make_args_decoder(t):
if type(t) is tuple:
return _make_args_to_tuple_of_types_lambda(t)
else:
return _make_args_to_type_lambda(t)
def get_thing(cmd, fieldcmd, datatype, *args):
pull = False
if type(datatype) is list:
assert (len(datatype) == 1)
decoder = _make_args_decoder(datatype[0])
value = []
def adder(data):
value.append(decoder(data))
elif type(datatype) is dict:
assert (len(datatype) == 1)
key_type, value_type = list(datatype.items())[0]
key_decoder = _make_args_decoder(key_type)
value_decoder = _make_args_decoder(value_type)
value = {}
def adder(data):
value[key_decoder([data[0]])] = value_decoder(data[1:])
else:
decoder = _make_args_decoder(datatype)
def adder(data):
value[0] = decoder(data)
value = [None]
pull = True
def callback(cmd2, fb, args2):
if cmd2 == fieldcmd:
adder(args2)
else:
print ("Unexpected command %s" % cmd2)
do_cmd(cmd, callback, list(args))
if pull:
return value[0]
else:
return value
class SetterWithConversion:
"""A setter object class that sets a specific property to a typed value or a tuple of typed value."""
def __init__(self, property, extractor):
self.property = property
self.extractor = extractor
def init_value(self):
return None
def __call__(self, obj, args):
# print ("Setting attr %s on object %s" % (self.property, obj))
setattr(obj, self.property, self.extractor(args))
class ListAdderWithConversion:
"""A setter object class that adds a tuple filled with type-converted arguments of the
callback to a list. E.g. ListAdderWithConversion('foo', (int, int))(obj, [1,2])
adds a tuple: (int(1), int(2)) to the list obj.foo"""
def __init__(self, property, extractor):
self.property = property
self.extractor = extractor
def init_value(self):
return []
def __call__(self, obj, args):
getattr(obj, self.property).append(self.extractor(args))
class DictAdderWithConversion:
"""A setter object class that adds a tuple filled with type-converted
arguments of the callback to a dictionary under a key passed as first argument
i.e. DictAdderWithConversion('foo', str, (int, int))(obj, ['bar',1,2]) adds
a tuple: (int(1), int(2)) under key 'bar' to obj.foo"""
def __init__(self, property, keytype, valueextractor):
self.property = property
self.keytype = keytype
self.valueextractor = valueextractor
def init_value(self):
return {}
def __call__(self, obj, args):
getattr(obj, self.property)[self.keytype(args[0])] = self.valueextractor(args[1:])
def _type_properties(base_type):
return {prop: getattr(base_type, prop) for prop in dir(base_type) if not prop.startswith("__")}
def _create_setter(prop, t):
if type(t) in [type, tuple] or issubclass(type(t), DocObj):
if type_wrapper_debug:
print ("%s is type %s" % (prop, repr(t)))
return SetterWithConversion(prop, _make_args_decoder(t))
elif type(t) is dict:
assert(len(t) == 1)
tkey, tvalue = list(t.items())[0]
if type_wrapper_debug:
print ("%s is type: %s -> %s" % (prop, repr(tkey), repr(tvalue)))
return DictAdderWithConversion(prop, tkey, _make_args_decoder(tvalue))
elif type(t) is list:
assert(len(t) == 1)
if type_wrapper_debug:
print ("%s is array of %s" % (prop, repr(t[0])))
return ListAdderWithConversion(prop, _make_args_decoder(t[0]))
else:
raise ValueError("Don't know what to do with property '%s' of type %s" % (prop, repr(t)))
def _create_unmarshaller(name, base_type, object_wrapper = False, property_grabber = _type_properties):
all_decorators = {}
prop_types = {}
settermap = {}
if type_wrapper_debug:
print ("Wrapping type: %s" % name)
print ("-----")
for prop, proptype in property_grabber(base_type).items():
decorators = []
propcmd = '/' + prop
if type(proptype) in [list, dict]:
if propcmd.endswith('s'):
if propcmd.endswith('es'):
propcmd = propcmd[:-2]
else:
propcmd = propcmd[:-1]
while isinstance(proptype, PropertyDecorator):
decorators.append(proptype)
propcmd = proptype.map_cmd(propcmd)
proptype = proptype.get_base()
settermap[propcmd] = _create_setter(prop, proptype)
all_decorators[prop] = decorators
prop_types[prop] = proptype
base_type.__str__ = lambda self: (str(name) + ":" + " ".join(["%s=%s" % (v.property, str(getattr(self, v.property))) for v in settermap.values()]))
if type_wrapper_debug:
print ("")
def exec_cmds(o):
for propname, decorators in all_decorators.items():
for decorator in decorators:
decorator.execute(propname, prop_types[propname], o)
if object_wrapper:
return exec_cmds, lambda cmd: (lambda self, *args: new_get_things(base_type(), self.path + cmd, settermap, list(args)))
else:
return lambda cmd, *args: new_get_things(base_type(), cmd, settermap, list(args))
class NonDocObj(object):
"""Root class for all wrapper classes that wrap objects that don't have
their own identity/UUID.
This covers various singletons and inner objects (e.g. engine in instruments)."""
class Status:
pass
def __init__(self, path):
self.path = path
def __new__(classObj, *args, **kwargs):
if is_python3:
result = object.__new__(classObj)
result.__init__(*args, **kwargs)
else:
result = object.__new__(classObj, *args, **kwargs)
name = classObj.__name__
if getattr(classObj, 'wrapped_class', None) != name:
classfinaliser, cmdwrapper = _create_unmarshaller(name, classObj.Status, object_wrapper = True)
classfinaliser(classObj)
classObj.status = cmdwrapper('/status')
classObj.wrapped_class = name
return result
def cmd(self, cmd, fb = None, *args):
do_cmd(self.path + cmd, fb, list(args))
def cmd_makeobj(self, cmd, *args):
return Document.map_uuid(GetUUID(self.path + cmd, *args).uuid)
def get_things(self, cmd, fields, *args):
return GetThings(self.path + cmd, fields, list(args))
def get_thing(self, cmd, fieldcmd, type, *args):
return get_thing(self.path + cmd, fieldcmd, type, *args)
def make_path(self, path):
return self.path + path
def __str__(self):
return "%s<%s>" % (self.__class__.__name__, self.path)
class DocObj(NonDocObj):
"""Root class for all wrapper classes that wrap first-class document objects."""
class Status:
pass
def __init__(self, uuid):
NonDocObj.__init__(self, Document.uuid_cmd(uuid, ''))
self.uuid = uuid
def delete(self):
self.cmd("/delete")
def __str__(self):
return "%s<%s>" % (self.__class__.__name__, self.uuid)
class VarPath:
def __init__(self, path, args = []):
self.path = path
self.args = args
def plus(self, subpath, *args):
return VarPath(self.path if subpath is None else self.path + "/" + subpath, self.args + list(args))
def set(self, *values):
do_cmd(self.path, None, self.args + list(values))
###############################################################################
# And those are the proper user-accessible objects.
###############################################################################
class Config:
class KeysUnmarshaller:
keys = [str]
keys_unmarshaller = _create_unmarshaller('Config.keys()', KeysUnmarshaller)
"""INI file manipulation class."""
@staticmethod
def sections(prefix = ""):
"""Return a list of configuration sections."""
return [CfgSection(name) for name in get_thing('/config/sections', '/section', [str], prefix)]
@staticmethod
def keys(section, prefix = ""):
"""Return a list of configuration keys in a section, with optional prefix filtering."""
return Config.keys_unmarshaller('/config/keys', str(section), str(prefix)).keys
@staticmethod
def get(section, key):
"""Return a string value of a given key."""
return get_thing('/config/get', '/value', str, str(section), str(key))
@staticmethod
def set(section, key, value):
"""Set a string value for a given key."""
do_cmd('/config/set', None, [str(section), str(key), str(value)])
@staticmethod
def delete(section, key):
"""Delete a given key."""
do_cmd('/config/delete', None, [str(section), str(key)])
@staticmethod
def save(filename = None):
"""Save config, either into current INI file or some other file."""
if filename is None:
do_cmd('/config/save', None, [])
else:
do_cmd('/config/save', None, [str(filename)])
@staticmethod
def add_section(section, content):
"""Populate a config section based on a string with key=value lists.
This is a toy/debug function, it doesn't handle any edge cases."""
for line in content.splitlines():
line = line.strip()
if line == '' or line.startswith('#'):
continue
try:
key, value = line.split("=", 2)
except ValueError as err:
raise ValueError("Cannot parse config line '%s'" % line)
Config.set(section, key.strip(), value.strip())
class Transport:
@staticmethod
def seek_ppqn(ppqn):
do_cmd('/master/seek_ppqn', None, [int(ppqn)])
@staticmethod
def seek_samples(samples):
do_cmd('/master/seek_samples', None, [int(samples)])
@staticmethod
def set_tempo(tempo):
do_cmd('/master/set_tempo', None, [float(tempo)])
@staticmethod
def set_timesig(nom, denom):
do_cmd('/master/set_timesig', None, [int(nom), int(denom)])
@staticmethod
def set_ppqn_factor(factor):
do_cmd('/master/set_ppqn_factor', None, [int(factor)])
@staticmethod
def play():
do_cmd('/master/play', None, [])
@staticmethod
def stop():
do_cmd('/master/stop', None, [])
@staticmethod
def panic():
do_cmd('/master/panic', None, [])
@staticmethod
def status():
return GetThings("/master/status", ['pos', 'pos_ppqn', 'tempo', 'timesig', 'sample_rate', 'playing', 'ppqn_factor'], [])
@staticmethod
def tell():
return GetThings("/master/tell", ['pos', 'pos_ppqn', 'playing'], [])
@staticmethod
def ppqn_to_samples(pos_ppqn):
return get_thing("/master/ppqn_to_samples", '/value', int, pos_ppqn)
@staticmethod
def samples_to_ppqn(pos_samples):
return get_thing("/master/samples_to_ppqn", '/value', int, pos_samples)
# Currently responsible for both JACK and USB I/O - not all functionality is
# supported by both.
class JackIO:
AUDIO_TYPE = "32 bit float mono audio"
MIDI_TYPE = "8 bit raw midi"
PORT_IS_SINK = 0x1
PORT_IS_SOURCE = 0x2
PORT_IS_PHYSICAL = 0x4
PORT_CAN_MONITOR = 0x8
PORT_IS_TERMINAL = 0x10
metadata.get_thing = get_thing #avoid circular dependency and redundant code
Metadata = metadata.Metadata #use with cbox.JackIO.Metadata.get_all_properties()
@staticmethod
def status():
# Some of these only make sense for JACK
return GetThings("/io/status", ['client_type', 'client_name',
'audio_inputs', 'audio_outputs', 'buffer_size', '*midi_output',
'*midi_input', 'sample_rate', 'output_resolution',
'*usb_midi_input', '*usb_midi_output', '?external_tempo'], [])
@staticmethod
def jack_transport_position():
# Some of these only make sense for JACK
return GetThings("/io/jack_transport_position", ['state', 'unique_lo',
'unique_hi', 'usecs_lo', 'usecs_hi', 'frame_rate', 'frame', 'bar',
'beat', 'tick', 'bar_start_tick', 'bbt_frame_offset', 'beats_per_bar',
'beat_type', 'ticks_per_beat', 'beats_per_minute', 'is_master'], [])
@staticmethod
def jack_transport_locate(pos):
do_cmd("/io/jack_transport_locate", None, [pos])
@staticmethod
def transport_mode(master = True, conditional = False):
if master:
do_cmd("/io/transport_mode", None, [1 if conditional else 2])
else:
do_cmd("/io/transport_mode", None, [0])
@staticmethod
def create_midi_input(name, autoconnect_spec = None):
uuid = GetUUID("/io/create_midi_input", name).uuid
if autoconnect_spec is not None and autoconnect_spec != '':
JackIO.autoconnect(uuid, autoconnect_spec)
return uuid
@staticmethod
def create_midi_output(name, autoconnect_spec = None):
uuid = GetUUID("/io/create_midi_output", name).uuid
if autoconnect_spec is not None and autoconnect_spec != '':
JackIO.autoconnect(uuid, autoconnect_spec)
return uuid
@staticmethod
def autoconnect(uuid, autoconnect_spec = None):
if autoconnect_spec is not None:
do_cmd("/io/autoconnect", None, [uuid, autoconnect_spec])
else:
do_cmd("/io/autoconnect", None, [uuid, ''])
autoconnect_midi_input = autoconnect
autoconnect_midi_output = autoconnect
autoconnect_audio_output = autoconnect
@staticmethod
def rename_midi_output(uuid, new_name):
do_cmd("/io/rename_midi_port", None, [uuid, new_name])
rename_midi_input = rename_midi_output
@staticmethod
def disconnect_midi_port(uuid):
do_cmd("/io/disconnect_midi_port", None, [uuid])
@staticmethod
def disconnect_midi_output(uuid):
do_cmd("/io/disconnect_midi_output", None, [uuid])
@staticmethod
def disconnect_midi_input(uuid):
do_cmd("/io/disconnect_midi_input", None, [uuid])
@staticmethod
def delete_midi_input(uuid):
do_cmd("/io/delete_midi_input", None, [uuid])
@staticmethod
def delete_midi_output(uuid):
do_cmd("/io/delete_midi_output", None, [uuid])
@staticmethod
def route_midi_input(input_uuid, scene_uuid):
do_cmd("/io/route_midi_input", None, [input_uuid, scene_uuid])
@staticmethod
def set_appsink_for_midi_input(input_uuid, enabled):
do_cmd("/io/set_appsink_for_midi_input", None, [input_uuid, 1 if enabled else 0])
@staticmethod
def get_new_events(input_uuid):
seq = []
do_cmd("/io/get_new_events", (lambda cmd, fb, args: seq.append((cmd, fb, args))), [input_uuid])
return seq
@staticmethod
def create_audio_output(name, autoconnect_spec = None):
uuid = GetUUID("/io/create_audio_output", name).uuid
if autoconnect_spec is not None and autoconnect_spec != '':
JackIO.autoconnect(uuid, autoconnect_spec)
return uuid
@staticmethod
def create_audio_output_router(uuid_left, uuid_right):
return get_thing("/io/create_audio_output_router", "/uuid", DocRecorder, uuid_left, uuid_right)
@staticmethod
def delete_audio_output(uuid):
do_cmd("/io/delete_audio_output", None, [uuid])
@staticmethod
def rename_audio_output(uuid, new_name):
do_cmd("/io/rename_audio_port", None, [uuid, new_name])
@staticmethod
def disconnect_audio_output(uuid):
do_cmd("/io/disconnect_audio_output", None, [uuid])
@staticmethod
def port_connect(pfrom, pto):
do_cmd("/io/port_connect", None, [pfrom, pto])
@staticmethod
def port_disconnect(pfrom, pto):
do_cmd("/io/port_disconnect", None, [pfrom, pto])
@staticmethod
def get_ports(name_mask = ".*", type_mask = ".*", flag_mask = 0):
return get_thing("/io/get_ports", '/port', [str], name_mask, type_mask, int(flag_mask))
@staticmethod
def get_connected_ports(port):
return get_thing("/io/get_connected_ports", '/port', [str], port)
@staticmethod
def external_tempo(enable):
"""Enable reacting to JACK transport tempo"""
do_cmd('/io/external_tempo', None, [1 if enable else 0])
def call_on_idle(callback = None):
do_cmd("/on_idle", callback, [])
def get_new_events():
seq = []
do_cmd("/on_idle", (lambda cmd, fb, args: seq.append((cmd, fb, args))), [])
return seq
def send_midi_event(*data, **kwargs):
output = kwargs.get('output', None)
do_cmd('/send_event_to', None, [output if output is not None else ''] + list(data))
def send_sysex(data, output = None):
do_cmd('/send_sysex_to', None, [output if output is not None else '', bytearray(data)])
def flush_rt():
do_cmd('/rt/flush', None, [])
class CfgSection:
def __init__(self, name):
self.name = name
def __getitem__(self, key):
return Config.get(self.name, key)
def __setitem__(self, key, value):
Config.set(self.name, key, value)
def __delitem__(self, key):
Config.delete(self.name, key)
def keys(self, prefix = ""):
return Config.keys(self.name, prefix)
class Pattern:
@staticmethod
def get_pattern():
pat_data = get_thing("/get_pattern", '/pattern', (bytes, int))
if pat_data is not None:
pat_blob, length = pat_data
pat_data = []
ofs = 0
while ofs < len(pat_blob):
data = list(struct.unpack_from("iBBbb", pat_blob, ofs))
data[1:2] = []
pat_data.append(tuple(data))
ofs += 8
return pat_data, length
return None
@staticmethod
def serialize_event(time, *data):
if len(data) >= 1 and len(data) <= 3:
return struct.pack("iBBbb"[0:2 + len(data)], int(time), len(data), *[int(v) for v in data])
raise ValueError("Invalid length of an event (%d)" % len(data))
class Document:
"""Document singleton."""
classmap = {}
objmap = {}
@staticmethod
def dump():
"""Print all objects in the documents to stdout. Only used for debugging."""
do_cmd("/doc/dump", None, [])
@staticmethod
def uuid_cmd(uuid, cmd):
"""Internal: execute a given request on an object with specific UUID."""
return "/doc/uuid/%s%s" % (uuid, cmd)
@staticmethod
def get_uuid(path):
"""Internal: retrieve an UUID of an object that has specified path."""
return GetUUID('%s/get_uuid' % path).uuid
@staticmethod
def map_path(path, *args):
"""Internal: return an object corresponding to a path"""
return Document.map_uuid(Document.get_uuid(path))
@staticmethod
def cmd_makeobj(cmd, *args):
"""Internal: create an object from the UUID result of a command"""
return Document.map_uuid(GetUUID(cmd, *args).uuid)
@staticmethod
def get_obj_class(uuid):
"""Internal: retrieve an internal class type of an object that has specified path."""
return get_thing(Document.uuid_cmd(uuid, "/get_class_name"), '/class_name', str)
@staticmethod
def get_song():
"""Retrieve the current song object of a given document. Each document can
only have one current song."""
return Document.map_path("/song")
@staticmethod
def get_scene():
"""Retrieve the first scene object of a default engine. This function
is considered obsolete-ish, because of multiple scene support."""
return Document.map_path("/scene")
@staticmethod
def get_engine():
"""Retrieve the current RT engine object of a given document. Each document can
only have one current RT engine."""
return Document.map_path("/rt/engine")
@staticmethod
def get_rt():
"""Retrieve the RT singleton. RT is an object used to communicate between
realtime and user thread, and is currently also used to access the audio
engine."""
return Document.map_path("/rt")
@staticmethod
def new_engine(srate, bufsize):
"""Create a new off-line engine object. This new engine object cannot be used for
audio playback - that's only allowed for default engine."""
return Document.cmd_makeobj('/new_engine', int(srate), int(bufsize))
@staticmethod
def map_uuid(uuid):
"""Create or retrieve a Python-side accessor proxy for a C-side object."""
if uuid is None:
return None
if uuid in Document.objmap:
return Document.objmap[uuid]
try:
oclass = Document.get_obj_class(uuid)
except Exception as e:
print ("Note: Cannot get class for " + uuid)
Document.dump()
raise
o = Document.classmap[oclass](uuid)
Document.objmap[uuid] = o
if hasattr(o, 'init_object'):
o.init_object()
return o
@staticmethod
def map_uuid_and_check(uuid, t):
o = Document.map_uuid(uuid)
if not isinstance(o, t):
raise TypeError("UUID %s is of type %s, expected %s" % (uuid, o.__class__, t))
return o
class DocPattern(DocObj):
class Status:
event_count = int
loop_end = int
name = str
def __init__(self, uuid):
DocObj.__init__(self, uuid)
def set_name(self, name):
self.cmd("/name", None, name)
Document.classmap['cbox_midi_pattern'] = DocPattern
class ClipItem:
def __init__(self, pos, offset, length, pattern, clip):
self.pos = pos
self.offset = offset
self.length = length
self.pattern = Document.map_uuid(pattern)
self.clip = Document.map_uuid(clip)
def __str__(self):
return "pos=%d offset=%d length=%d pattern=%s clip=%s" % (self.pos, self.offset, self.length, self.pattern.uuid, self.clip.uuid)
def __eq__(self, other):
return str(self) == str(other)
class DocTrackClip(DocObj):
class Status:
pos = SettableProperty(int)
offset = SettableProperty(int)
length = SettableProperty(int)
pattern = SettableProperty(DocPattern)
def __init__(self, uuid):
DocObj.__init__(self, uuid)
Document.classmap['cbox_track_item'] = DocTrackClip
class DocTrack(DocObj):
class Status:
clips = [ClipItem]
name = SettableProperty(str)
external_output = SettableProperty(str)
mute = SettableProperty(int)
def add_clip(self, pos, offset, length, pattern):
return self.cmd_makeobj("/add_clip", int(pos), int(offset), int(length), pattern.uuid)
def clear_clips(self):
return self.cmd_makeobj("/clear_clips")
Document.classmap['cbox_track'] = DocTrack
class TrackItem:
def __init__(self, name, count, track):
self.name = name
self.count = count
self.track = Document.map_uuid(track)
class PatternItem:
def __init__(self, name, length, pattern):
self.name = name
self.length = length
self.pattern = Document.map_uuid(pattern)
class MtiItem:
def __init__(self, pos, tempo, timesig_num, timesig_denom):
self.pos = pos
self.tempo = tempo
# Original misspelling
self.timesig_num = timesig_num
self.timesig_denom = timesig_denom
def __getattr__(self, name):
if name == 'timesig_nom':
return self.timesig_num
raise AttributeError(name)
def __setattr__(self, name, value):
if name == 'timesig_nom':
self.timesig_num = value
else:
self.__dict__[name] = value
def __eq__(self, o):
return self.pos == o.pos and self.tempo == o.tempo and self.timesig_num == o.timesig_num and self.timesig_denom == o.timesig_denom
def __repr__(self):
return ("pos: {}, bpm: {}, timesig: {}/{}".format(self.pos, self.tempo, self.timesig_num, self.timesig_denom))
class DocSongStatus:
tracks = None
patterns = None
class DocSong(DocObj):
class Status:
tracks = [TrackItem]
patterns = [PatternItem]
mtis = [MtiItem]
loop_start = int
loop_end = int
def clear(self):
return self.cmd("/clear", None)
def set_loop(self, ls, le):
return self.cmd("/set_loop", None, int(ls), int(le))
def set_mti(self, pos, tempo = None, timesig_num = None, timesig_denom = None, timesig_nom = None):
if timesig_nom is not None:
timesig_num = timesig_nom
self.cmd("/set_mti", None, int(pos), float(tempo) if tempo is not None else -1.0, int(timesig_num) if timesig_num is not None else -1, int(timesig_denom) if timesig_denom else -1)
def delete_mti(self, pos):
"""Deleting works only if we set everything to exactly 0. Not None, not -1"""
self.set_mti(pos, tempo = 0, timesig_num = 0, timesig_denom = 0, timesig_nom = 0)
def add_track(self):
return self.cmd_makeobj("/add_track")
def load_drum_pattern(self, name):
return self.cmd_makeobj("/load_pattern", name, 1)
def load_drum_track(self, name):
return self.cmd_makeobj("/load_track", name, 1)
def pattern_from_blob(self, blob, length):
return self.cmd_makeobj("/load_blob", bytearray(blob), int(length))
def loop_single_pattern(self, loader):
self.clear()
track = self.add_track()
pat = loader()
length = pat.status().loop_end
track.add_clip(0, 0, length, pat)
self.set_loop(0, length)
self.update_playback()
def update_playback(self):
# XXXKF Maybe make it a song-level API instead of global
do_cmd("/update_playback", None, [])
Document.classmap['cbox_song'] = DocSong
class UnknownModule(NonDocObj):
class Status:
pass
class DocRecorder(DocObj):
class Status:
filename = str
gain = SettableProperty(float)
Document.classmap['cbox_recorder'] = DocRecorder
class RecSource(NonDocObj):
class Status:
handler = [DocRecorder]
def attach(self, recorder):
self.cmd('/attach', None, recorder.uuid)
def detach(self, recorder):
self.cmd('/detach', None, recorder.uuid)
class EffectSlot(NonDocObj):
class Status:
insert_preset = SettableProperty(str)
insert_engine = SettableProperty(str)
bypass = SettableProperty(bool)
def init_object(self):
# XXXKF add wrapper classes for effect engines
self.engine = UnknownModule(self.path + "/engine")
class InstrumentOutput(EffectSlot):
class Status(EffectSlot.Status):
gain_linear = float
gain = float
output = SettableProperty(int)
def init_object(self):
EffectSlot.init_object(self)
self.rec_dry = RecSource(self.make_path('/rec_dry'))
self.rec_wet = RecSource(self.make_path('/rec_wet'))
class DocInstrument(DocObj):
class Status:
name = str
outputs = int
aux_offset = int
engine = str
def init_object(self):
s = self.status()
engine = s.engine
if engine in engine_classes:
self.engine = engine_classes[engine]("/doc/uuid/" + self.uuid + "/engine")
else:
raise ValueError("Unknown engine %s" % engine)
self.output_slots = []
for i in range(s.outputs):
io = InstrumentOutput(self.make_path('/output/%d' % (i + 1)))
io.init_object()
self.output_slots.append(io)
def move_to(self, target_scene, pos = 0):
return self.cmd_makeobj("/move_to", target_scene.uuid, pos + 1)
def get_output_slot(self, slot):
return self.output_slots[slot]
Document.classmap['cbox_instrument'] = DocInstrument
class DocLayer(DocObj):
class Status:
name = str
instrument_name = str
instrument = AltPropName('/instrument_uuid', DocInstrument)
enable = SettableProperty(bool)
low_note = SettableProperty(int)
high_note = SettableProperty(int)
fixed_note = SettableProperty(int)
in_channel = SettableProperty(int)
out_channel = SettableProperty(int)
disable_aftertouch = SettableProperty(bool)
invert_sustain = SettableProperty(bool)
consume = SettableProperty(bool)
ignore_scene_transpose = SettableProperty(bool)
ignore_program_changes = SettableProperty(bool)
transpose = SettableProperty(int)
external_output = SettableProperty(str)
def get_instrument(self):
return self.status().instrument
Document.classmap['cbox_layer'] = DocLayer
class SamplerEngine(NonDocObj):
class Status(object):
"""Maximum number of voices playing at the same time."""
polyphony = int
"""Current number of voices playing."""
active_voices = int
"""Current number of delayed-startup voices waiting to be played."""
active_prevoices = int
"""Current number of disk streams."""
active_pipes = int
"""GM volume (14-bit) per MIDI channel."""
volume = {int:int}
"""GM pan (14-bit) per MIDI channel."""
pan = {int:int}
"""Output offset per MIDI channel."""
output = {int:int}
"""Current number of voices playing per MIDI channel."""
channel_voices = AltPropName('/channel_voices', {int:int})
"""Current number of voices waiting to be played per MIDI channel."""
channel_prevoices = AltPropName('/channel_prevoices', {int:int})
"""MIDI channel -> (program number, program name)"""
patches = {int:(int, str)}
def load_patch_from_cfg(self, patch_no, cfg_section, display_name):
"""Load a sampler program from an 'spgm:' config section."""
return self.cmd_makeobj("/load_patch", int(patch_no), cfg_section, display_name)
def load_patch_from_string(self, patch_no, sample_dir, sfz_data, display_name):
"""Load a sampler program from a string, using given filesystem path for sample directory."""
return self.cmd_makeobj("/load_patch_from_string", int(patch_no), sample_dir, sfz_data, display_name)
def load_patch_from_file(self, patch_no, sfz_name, display_name):
"""Load a sampler program from a filesystem file."""
return self.cmd_makeobj("/load_patch_from_file", int(patch_no), sfz_name, display_name)
def load_patch_from_tar(self, patch_no, tar_name, sfz_name, display_name):
"""Load a sampler program from a tar file."""
return self.cmd_makeobj("/load_patch_from_file", int(patch_no), "sbtar:%s;%s" % (tar_name, sfz_name), display_name)
def set_patch(self, channel, patch_no):
"""Select patch identified by patch_no in a specified MIDI channel."""
self.cmd("/set_patch", None, int(channel), int(patch_no))
def set_output(self, channel, output):
"""Set output offset value in a specified MIDI channel."""
self.cmd("/set_output", None, int(channel), int(output))
def get_unused_program(self):
"""Returns first program number that has no program associated with it."""
return self.get_thing("/get_unused_program", '/program_no', int)
def set_polyphony(self, polyphony):
"""Set a maximum number of voices that can be played at a given time."""
self.cmd("/polyphony", None, int(polyphony))
def get_patches(self):
"""Return a map of program identifiers to program objects."""
return self.get_thing("/patches", '/patch', {int : (str, SamplerProgram, int)})
def get_keyswitch_state(self, channel, group):
"""Return a map of program identifiers to program objects."""
return self.get_thing("/keyswitch_state", '/last_key', int, channel, group)
class FluidsynthEngine(NonDocObj):
class Status:
polyphony = int
soundfont = str
patch = {int: (int, str)}
def load_soundfont(self, filename):
return self.cmd_makeobj("/load_soundfont", filename)
def set_patch(self, channel, patch_no):
self.cmd("/set_patch", None, int(channel), int(patch_no))
def set_polyphony(self, polyphony):
self.cmd("/polyphony", None, int(polyphony))
def get_patches(self):
return self.get_thing("/patches", '/patch', {int: str})
class StreamPlayerEngine(NonDocObj):
class Status:
filename = str
pos = int
length = int
playing = int
def play(self):
self.cmd('/play')
def stop(self):
self.cmd('/stop')
def seek(self, place):
self.cmd('/seek', None, int(place))
def load(self, filename, loop_start = -1):
self.cmd('/load', None, filename, int(loop_start))
def unload(self):
self.cmd('/unload')
class TonewheelOrganEngine(NonDocObj):
class Status:
upper_drawbar = SettableProperty({int: int})
lower_drawbar = SettableProperty({int: int})
pedal_drawbar = SettableProperty({int: int})
upper_vibrato = SettableProperty(bool)
lower_vibrato = SettableProperty(bool)
vibrato_mode = SettableProperty(int)
vibrato_chorus = SettableProperty(int)
percussion_enable = SettableProperty(bool)
percussion_3rd = SettableProperty(bool)
class JackInputEngine(NonDocObj):
class Status:
inputs = (int, int)
engine_classes = {
'sampler' : SamplerEngine,
'fluidsynth' : FluidsynthEngine,
'stream_player' : StreamPlayerEngine,
'tonewheel_organ' : TonewheelOrganEngine,
'jack_input' : JackInputEngine,
}
class DocAuxBus(DocObj):
class Status:
name = str
def init_object(self):
self.slot = EffectSlot("/doc/uuid/" + self.uuid + "/slot")
self.slot.init_object()
Document.classmap['cbox_aux_bus'] = DocAuxBus
class DocScene(DocObj):
class Status:
name = str
title = str
transpose = int
layers = [DocLayer]
instruments = {str: (str, DocInstrument)}
auxes = {str: DocAuxBus}
enable_default_song_input = SettableProperty(bool)
enable_default_external_input = SettableProperty(bool)
def clear(self):
self.cmd("/clear", None)
def load(self, name):
self.cmd("/load", None, name)
def load_aux(self, aux):
return self.cmd_makeobj("/load_aux", aux)
def delete_aux(self, aux):
return self.cmd("/delete_aux", None, aux)
def delete_layer(self, pos):
self.cmd("/delete_layer", None, int(1 + pos))
def move_layer(self, old_pos, new_pos):
self.cmd("/move_layer", None, int(old_pos + 1), int(new_pos + 1))
#Layer positions are 0 for "append" and other positions are 1...n which need to be unique
def add_layer(self, aux, pos = None):
if pos is None:
return self.cmd_makeobj("/add_layer", 0, aux)
else:
# Note: The positions in high-level API are zero-based.
return self.cmd_makeobj("/add_layer", int(1 + pos), aux)
def add_instrument_layer(self, name, pos = None):
if pos is None:
return self.cmd_makeobj("/add_instrument_layer", 0, name)
else:
return self.cmd_makeobj("/add_instrument_layer", int(1 + pos), name)
def add_new_instrument_layer(self, name, engine, pos = None):
if pos is None:
return self.cmd_makeobj("/add_new_instrument_layer", 0, name, engine)
else:
return self.cmd_makeobj("/add_new_instrument_layer", int(1 + pos), name, engine)
def add_new_midi_layer(self, ext_output_uuid, pos = None):
if pos is None:
return self.cmd_makeobj("/add_midi_layer", 0, ext_output_uuid)
else:
return self.cmd_makeobj("/add_midi_layer", int(1 + pos), ext_output_uuid)
def send_midi_event(self, *data):
self.cmd('/send_event', None, *data)
def play_pattern(self, pattern, tempo, id = 0):
self.cmd('/play_pattern', None, pattern.uuid, float(tempo), int(id))
Document.classmap['cbox_scene'] = DocScene
class DocRt(DocObj):
class Status:
audio_channels = (int, int)
state = (int, str)
Document.classmap['cbox_rt'] = DocRt
class DocModule(DocObj):
class Status:
pass
Document.classmap['cbox_module'] = DocModule
class DocEngine(DocObj):
class Status:
scenes = AltPropName('/scene', [DocScene])
def init_object(self):
self.master_effect = EffectSlot(self.path + "/master_effect")
self.master_effect.init_object()
def new_scene(self):
return self.cmd_makeobj('/new_scene')
def new_recorder(self, filename):
return self.cmd_makeobj("/new_recorder", filename)
def render_stereo(self, samples):
return self.get_thing("/render_stereo", '/data', bytes, samples)
Document.classmap['cbox_engine'] = DocEngine
class SamplerProgram(DocObj):
class Status:
name = str
sample_dir = str
source_file = str
program_no = int
in_use = int
def get_regions(self):
return self.get_thing("/regions", '/region', [SamplerLayer])
def get_global(self):
return self.cmd_makeobj("/global")
def get_hierarchy(self):
"""see SamplerLayer.get_hierarchy"""
return {self.get_global() : self.get_global().get_hierarchy()}
def get_control_inits(self):
return self.get_thing("/control_inits", '/control_init', [(int, int)])
def get_control_labels(self):
return self.get_thing("/control_labels", '/control_label', {int : str})
def get_key_labels(self):
return self.get_thing("/key_labels", '/key_label', {int : str})
def get_output_labels(self):
return self.get_thing("/output_labels", '/output_label', {int : str})
def get_keyswitch_groups(self):
return self.get_thing("/keyswitch_groups", '/key_range', [(int, int)])
def new_group(self):
# Obsolete
return self.cmd_makeobj("/new_group")
def add_control_init(self, controller, value):
return self.cmd("/add_control_init", None, controller, value)
def add_control_label(self, controller, label):
return self.cmd("/add_control_label", None, controller, label)
# which = -1 -> remove all controllers with that number from the list
def delete_control_init(self, controller, which = 0):
return self.cmd("/delete_control_init", None, controller, which)
def load_file(self, filename, max_size = -1):
"""Return an in-memory file corresponding to a given file inside sfbank.
This can be used for things like scripts, images, descriptions etc."""
data = self.get_thing("/load_file", '/data', bytes, filename, max_size)
if data is None:
return data
return BytesIO(data)
def clone_to(self, dest_module, prog_index):
return self.cmd_makeobj('/clone_to', dest_module.uuid, int(prog_index))
Document.classmap['sampler_program'] = SamplerProgram
class SamplerLayer(DocObj):
class Status:
parent_program = SamplerProgram
parent = DocObj
level = str
def get_children(self):
"""Return all children SamplerLayer.
The hierarchy is always global-master-group-region
Will be empty if this is
an sfz <region>, which has no further children.
"""
return self.get_thing("/get_children", '/child', [SamplerLayer])
def get_hierarchy(self):
"""Returns either a level of hierarchy, e.g. <global> or <group>
or None, if this is a childless layer, such as a <region>.
The hierarchy is always global-master-group-region.
Regions alre always on the fourth level. But not all levels might have regions.
Hint: Print with pprint during development."""
children = self.get_children()
if children:
result = {}
for childLayer in children:
result[childLayer] = childLayer.get_hierarchy()
else:
result = None
return result
def as_dict(self):
"""Returns a dictionary of parameters set at this level of the
layer hierarchy."""
return self.get_thing("/as_list", '/value', {str: str})
def as_dict_full(self):
"""Returns a dictionary of parameters set either at this level of the
layer hierarchy or at one of the ancestors."""
return self.get_thing("/as_list_full", '/value', {str: str})
def as_string(self):
"""A space separated string of all sampler values at this level
in the hierarchy, for example ampeg_decay.
This only includes non-default values, e.g. from the sfz file"""
return self.get_thing("/as_string", '/value', str)
def as_string_full(self):
"""A space separated string of all sampler values at this level
in the hierarchy, for example ampeg_decay.
This includes all default values.
To access the values as dict with number data types use
get_params_full().
'_oncc1' will be converted to '_cc1'
"""
return self.get_thing("/as_string_full", '/value', str)
def set_param(self, key, value):
self.cmd("/set_param", None, key, str(value))
def unset_param(self, key):
self.cmd("/unset_param", None, key)
def new_child(self):
return self.cmd_makeobj("/new_child")
Document.classmap['sampler_layer'] = SamplerLayer