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