<s> """<STR_LIT>""" <EOL> __version__ = "<STR_LIT>" <EOL> import errno as _errno <EOL> import platform as _platform <EOL> from cffi import FFI as _FFI <EOL> _ffi = _FFI ( ) <EOL> _ffi . cdef ( """<STR_LIT>""" ) <EOL> _ffi . cdef ( """<STR_LIT>""" , packed = True ) <EOL> if _platform . system ( ) == '<STR_LIT>' : <EOL> if _platform . architecture ( ) [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> _lib = _ffi . dlopen ( "<STR_LIT>" ) <EOL> else : <EOL> _lib = _ffi . dlopen ( "<STR_LIT>" ) <EOL> else : <EOL> _lib = _ffi . dlopen ( "<STR_LIT>" ) <EOL> _AUDIO = b"<STR_LIT>" <EOL> _MIDI = b"<STR_LIT>" <EOL> STOPPED = _lib . JackTransportStopped <EOL> """<STR_LIT>""" <EOL> ROLLING = _lib . JackTransportRolling <EOL> """<STR_LIT>""" <EOL> STARTING = _lib . JackTransportStarting <EOL> """<STR_LIT>""" <EOL> NETSTARTING = _lib . JackTransportNetStarting <EOL> """<STR_LIT>""" <EOL> _SUCCESS = <NUM_LIT:0> <EOL> _FAILURE = <NUM_LIT:1> <EOL> class Client ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name , use_exact_name = False , no_start_server = False , <EOL> servername = None , session_id = None ) : <EOL> """<STR_LIT>""" <EOL> status = _ffi . new ( "<STR_LIT>" ) <EOL> options = _lib . JackNullOption <EOL> optargs = [ ] <EOL> if use_exact_name : <EOL> options |= _lib . JackUseExactName <EOL> if no_start_server : <EOL> options |= _lib . JackNoStartServer <EOL> if servername : <EOL> options |= _lib . JackServerName <EOL> optargs . append ( _ffi . new ( "<STR_LIT>" , servername . encode ( ) ) ) <EOL> if session_id : <EOL> options |= _lib . JackSessionID <EOL> optargs . append ( _ffi . new ( "<STR_LIT>" , session_id . encode ( ) ) ) <EOL> self . _ptr = _lib . jack_client_open ( name . encode ( ) , options , status , <EOL> * optargs ) <EOL> self . _status = Status ( status [ <NUM_LIT:0> ] ) <EOL> if not self . _ptr : <EOL> raise JackError ( str ( self . status ) ) <EOL> self . _inports = Ports ( self , _AUDIO , _lib . JackPortIsInput ) <EOL> self . _outports = Ports ( self , _AUDIO , _lib . JackPortIsOutput ) <EOL> self . _midi_inports = Ports ( self , _MIDI , _lib . JackPortIsInput ) <EOL> self . _midi_outports = Ports ( self , _MIDI , _lib . JackPortIsOutput ) <EOL> self . _keepalive = [ ] <EOL> self . _position = _ffi . new ( "<STR_LIT>" ) <EOL> _ptr = _ffi . NULL <EOL> def __enter__ ( self ) : <EOL> self . activate ( ) <EOL> return self <EOL> def __exit__ ( self , * args ) : <EOL> self . deactivate ( ) <EOL> self . close ( ) <EOL> def __del__ ( self ) : <EOL> """<STR_LIT>""" <EOL> self . close ( ) <EOL> @ property <EOL> def name ( self ) : <EOL> """<STR_LIT>""" <EOL> return _ffi . string ( _lib . jack_get_client_name ( self . _ptr ) ) . decode ( ) <EOL> @ property <EOL> def samplerate ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . jack_get_sample_rate ( self . _ptr ) <EOL> @ property <EOL> def blocksize ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . jack_get_buffer_size ( self . _ptr ) <EOL> @ blocksize . setter <EOL> def blocksize ( self , blocksize ) : <EOL> _check ( _lib . jack_set_buffer_size ( self . _ptr , blocksize ) , <EOL> "<STR_LIT>" ) <EOL> @ property <EOL> def status ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _status <EOL> @ property <EOL> def realtime ( self ) : <EOL> """<STR_LIT>""" <EOL> return bool ( _lib . jack_is_realtime ( self . _ptr ) ) <EOL> @ property <EOL> def frames_since_cycle_start ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . jack_frames_since_cycle_start ( self . _ptr ) <EOL> @ property <EOL> def frame_time ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . jack_frame_time ( self . _ptr ) <EOL> @ property <EOL> def last_frame_time ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . jack_last_frame_time ( self . _ptr ) <EOL> @ property <EOL> def inports ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _inports <EOL> @ property <EOL> def outports ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _outports <EOL> @ property <EOL> def midi_inports ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _midi_inports <EOL> @ property <EOL> def midi_outports ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _midi_outports <EOL> def owns ( self , port ) : <EOL> """<STR_LIT>""" <EOL> port = self . _get_port_ptr ( port ) <EOL> return bool ( _lib . jack_port_is_mine ( self . _ptr , port ) ) <EOL> def activate ( self ) : <EOL> """<STR_LIT>""" <EOL> _check ( _lib . jack_activate ( self . _ptr ) , "<STR_LIT>" ) <EOL> def deactivate ( self , ignore_errors = True ) : <EOL> """<STR_LIT>""" <EOL> err = _lib . jack_deactivate ( self . _ptr ) <EOL> if not ignore_errors : <EOL> _check ( err , "<STR_LIT>" ) <EOL> def cpu_load ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . jack_cpu_load ( self . _ptr ) <EOL> def close ( self , ignore_errors = True ) : <EOL> """<STR_LIT>""" <EOL> if self . _ptr : <EOL> err = _lib . jack_client_close ( self . _ptr ) <EOL> self . _ptr = _ffi . NULL <EOL> if not ignore_errors : <EOL> _check ( err , "<STR_LIT>" ) <EOL> def connect ( self , source , destination ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( source , Port ) : <EOL> source = source . name <EOL> if isinstance ( destination , Port ) : <EOL> destination = destination . name <EOL> err = _lib . jack_connect ( self . _ptr , source . encode ( ) , <EOL> destination . encode ( ) ) <EOL> if err == _errno . EEXIST : <EOL> raise JackError ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( source , destination ) ) <EOL> _check ( err , <EOL> "<STR_LIT>" . format ( source , destination ) ) <EOL> def disconnect ( self , source , destination ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( source , Port ) : <EOL> source = source . name <EOL> if isinstance ( destination , Port ) : <EOL> destination = destination . name <EOL> _check ( _lib . jack_disconnect ( <EOL> self . _ptr , source . encode ( ) , destination . encode ( ) ) , <EOL> "<STR_LIT>" . format ( source , destination ) ) <EOL> def transport_start ( self ) : <EOL> """<STR_LIT>""" <EOL> _lib . jack_transport_start ( self . _ptr ) <EOL> def transport_stop ( self ) : <EOL> """<STR_LIT>""" <EOL> _lib . jack_transport_stop ( self . _ptr ) <EOL> @ property <EOL> def transport_state ( self ) : <EOL> """<STR_LIT>""" <EOL> return TransportState ( _lib . jack_transport_query ( self . _ptr , _ffi . NULL ) ) <EOL> def transport_locate ( self , frame ) : <EOL> """<STR_LIT>""" <EOL> _check ( _lib . jack_transport_locate ( self . _ptr , frame ) , <EOL> "<STR_LIT>" ) <EOL> def transport_query ( self ) : <EOL> """<STR_LIT>""" <EOL> state , pos = self . transport_query_struct ( ) <EOL> return TransportState ( state ) , position2dict ( pos ) <EOL> def transport_query_struct ( self ) : <EOL> """<STR_LIT>""" <EOL> state = _lib . jack_transport_query ( self . _ptr , self . _position ) <EOL> return state , self . _position <EOL> def transport_reposition_struct ( self , position ) : <EOL> """<STR_LIT>""" <EOL> _check ( _lib . jack_transport_reposition ( self . _ptr , position ) , <EOL> "<STR_LIT>" ) <EOL> def set_freewheel ( self , onoff ) : <EOL> """<STR_LIT>""" <EOL> _check ( _lib . jack_set_freewheel ( self . _ptr , onoff ) , <EOL> "<STR_LIT>" ) <EOL> def set_shutdown_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> @ self . _callback ( "<STR_LIT>" ) <EOL> def callback_wrapper ( code , reason , _ ) : <EOL> callback ( Status ( code ) , _ffi . string ( reason ) . decode ( ) ) <EOL> _lib . jack_on_info_shutdown ( self . _ptr , callback_wrapper , _ffi . NULL ) <EOL> def set_process_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> @ self . _callback ( "<STR_LIT>" , error = _FAILURE ) <EOL> def callback_wrapper ( frames , _ ) : <EOL> try : <EOL> callback ( frames ) <EOL> except CallbackExit : <EOL> return _FAILURE <EOL> return _SUCCESS <EOL> _check ( _lib . jack_set_process_callback ( <EOL> self . _ptr , callback_wrapper , _ffi . NULL ) , <EOL> "<STR_LIT>" ) <EOL> def set_freewheel_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> @ self . _callback ( "<STR_LIT>" ) <EOL> def callback_wrapper ( starting , _ ) : <EOL> callback ( bool ( starting ) ) <EOL> _check ( _lib . jack_set_freewheel_callback ( <EOL> self . _ptr , callback_wrapper , _ffi . NULL ) , <EOL> "<STR_LIT>" ) <EOL> def set_blocksize_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> @ self . _callback ( "<STR_LIT>" , error = _FAILURE ) <EOL> def callback_wrapper ( blocksize , _ ) : <EOL> try : <EOL> callback ( blocksize ) <EOL> except CallbackExit : <EOL> return _FAILURE <EOL> return _SUCCESS <EOL> _check ( _lib . jack_set_buffer_size_callback ( <EOL> self . _ptr , callback_wrapper , _ffi . NULL ) , <EOL> "<STR_LIT>" ) <EOL> def set_samplerate_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> @ self . _callback ( "<STR_LIT>" , error = _FAILURE ) <EOL> def callback_wrapper ( samplerate , _ ) : <EOL> try : <EOL> callback ( samplerate ) <EOL> except CallbackExit : <EOL> return _FAILURE <EOL> return _SUCCESS <EOL> _check ( _lib . jack_set_sample_rate_callback ( <EOL> self . _ptr , callback_wrapper , _ffi . NULL ) , <EOL> "<STR_LIT>" ) <EOL> def set_client_registration_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> @ self . _callback ( "<STR_LIT>" ) <EOL> def callback_wrapper ( name , register , _ ) : <EOL> callback ( _ffi . string ( name ) . decode ( ) , bool ( register ) ) <EOL> _check ( _lib . jack_set_client_registration_callback ( <EOL> self . _ptr , callback_wrapper , _ffi . NULL ) , <EOL> "<STR_LIT>" ) <EOL> def set_port_registration_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> @ self . _callback ( "<STR_LIT>" ) <EOL> def callback_wrapper ( port , register , _ ) : <EOL> port = self . _wrap_port_ptr ( _lib . jack_port_by_id ( self . _ptr , port ) ) <EOL> callback ( port , bool ( register ) ) <EOL> _check ( _lib . jack_set_port_registration_callback ( <EOL> self . _ptr , callback_wrapper , _ffi . NULL ) , <EOL> "<STR_LIT>" ) <EOL> def set_port_connect_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> @ self . _callback ( "<STR_LIT>" ) <EOL> def callback_wrapper ( a , b , connect , _ ) : <EOL> a = self . _wrap_port_ptr ( _lib . jack_port_by_id ( self . _ptr , a ) ) <EOL> b = self . _wrap_port_ptr ( _lib . jack_port_by_id ( self . _ptr , b ) ) <EOL> callback ( a , b , bool ( connect ) ) <EOL> _check ( _lib . jack_set_port_connect_callback ( <EOL> self . _ptr , callback_wrapper , _ffi . NULL ) , <EOL> "<STR_LIT>" ) <EOL> def set_port_rename_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> @ self . _callback ( "<STR_LIT>" , error = _FAILURE ) <EOL> def callback_wrapper ( port , old_name , new_name , _ ) : <EOL> port = self . _wrap_port_ptr ( _lib . jack_port_by_id ( self . _ptr , port ) ) <EOL> try : <EOL> callback ( port , _ffi . string ( old_name ) . decode ( ) , <EOL> _ffi . string ( new_name ) . decode ( ) ) <EOL> except CallbackExit : <EOL> return _FAILURE <EOL> return _SUCCESS <EOL> _check ( _lib . jack_set_port_rename_callback ( <EOL> self . _ptr , callback_wrapper , _ffi . NULL ) , <EOL> "<STR_LIT>" ) <EOL> def set_graph_order_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> @ self . _callback ( "<STR_LIT>" , error = _FAILURE ) <EOL> def callback_wrapper ( _ ) : <EOL> try : <EOL> callback ( ) <EOL> except CallbackExit : <EOL> return _FAILURE <EOL> return _SUCCESS <EOL> _check ( _lib . jack_set_graph_order_callback ( <EOL> self . _ptr , callback_wrapper , _ffi . NULL ) , <EOL> "<STR_LIT>" ) <EOL> def set_xrun_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> @ self . _callback ( "<STR_LIT>" , error = _FAILURE ) <EOL> def callback_wrapper ( _ ) : <EOL> try : <EOL> callback ( _lib . jack_get_xrun_delayed_usecs ( self . _ptr ) ) <EOL> except CallbackExit : <EOL> return _FAILURE <EOL> return _SUCCESS <EOL> _check ( _lib . jack_set_xrun_callback ( <EOL> self . _ptr , callback_wrapper , _ffi . NULL ) , <EOL> "<STR_LIT>" ) <EOL> def set_timebase_callback ( self , callback = None , conditional = False ) : <EOL> """<STR_LIT>""" <EOL> if callback is None : <EOL> return lambda cb : self . set_timebase_callback ( cb , conditional ) <EOL> @ self . _callback ( "<STR_LIT>" ) <EOL> def callback_wrapper ( state , blocksize , pos , new_pos , _ ) : <EOL> callback ( state , blocksize , pos , bool ( new_pos ) ) <EOL> err = _lib . jack_set_timebase_callback ( self . _ptr , conditional , <EOL> callback_wrapper , _ffi . NULL ) <EOL> if conditional and err in ( _errno . EBUSY , - <NUM_LIT:1> ) : <EOL> return False <EOL> _check ( err , "<STR_LIT>" ) <EOL> return True <EOL> def get_uuid_for_client_name ( self , name ) : <EOL> """<STR_LIT>""" <EOL> uuid = _ffi . gc ( _lib . jack_get_uuid_for_client_name ( <EOL> self . _ptr , name . encode ( ) ) , _lib . jack_free ) <EOL> if not uuid : <EOL> raise JackError ( "<STR_LIT>" . format ( name ) ) <EOL> return _ffi . string ( uuid ) . decode ( ) <EOL> def get_client_name_by_uuid ( self , uuid ) : <EOL> """<STR_LIT>""" <EOL> name = _ffi . gc ( _lib . jack_get_client_name_by_uuid ( <EOL> self . _ptr , uuid . encode ( ) ) , _lib . jack_free ) <EOL> if not name : <EOL> raise JackError ( "<STR_LIT>" . format ( uuid ) ) <EOL> return _ffi . string ( name ) . decode ( ) <EOL> def get_port_by_name ( self , name ) : <EOL> """<STR_LIT>""" <EOL> port_ptr = _lib . jack_port_by_name ( self . _ptr , name . encode ( ) ) <EOL> if not port_ptr : <EOL> raise JackError ( "<STR_LIT>" . format ( name ) ) <EOL> return self . _wrap_port_ptr ( port_ptr ) <EOL> def get_all_connections ( self , port ) : <EOL> """<STR_LIT>""" <EOL> port = self . _get_port_ptr ( port ) <EOL> names = _ffi . gc ( _lib . jack_port_get_all_connections ( self . _ptr , port ) , <EOL> _lib . jack_free ) <EOL> return self . _port_list_from_pointers ( names ) <EOL> def get_ports ( self , name_pattern = "<STR_LIT>" , is_audio = False , is_midi = False , <EOL> is_input = False , is_output = False , is_physical = False , <EOL> can_monitor = False , is_terminal = False ) : <EOL> """<STR_LIT>""" <EOL> if is_audio and not is_midi : <EOL> type_pattern = _AUDIO <EOL> elif is_midi and not is_audio : <EOL> type_pattern = _MIDI <EOL> else : <EOL> type_pattern = b"<STR_LIT>" <EOL> flags = <NUM_LIT> <EOL> if is_input : <EOL> flags |= _lib . JackPortIsInput <EOL> if is_output : <EOL> flags |= _lib . JackPortIsOutput <EOL> if is_physical : <EOL> flags |= _lib . JackPortIsPhysical <EOL> if can_monitor : <EOL> flags |= _lib . JackPortCanMonitor <EOL> if is_terminal : <EOL> flags |= _lib . JackPortIsTerminal <EOL> names = _ffi . gc ( _lib . jack_get_ports ( <EOL> self . _ptr , name_pattern . encode ( ) , type_pattern , flags ) , <EOL> _lib . jack_free ) <EOL> return self . _port_list_from_pointers ( names ) <EOL> def _callback ( self , cdecl , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> def callback_decorator ( python_callable ) : <EOL> function_ptr = _ffi . callback ( cdecl , python_callable , ** kwargs ) <EOL> self . _keepalive . append ( function_ptr ) <EOL> return function_ptr <EOL> return callback_decorator <EOL> def _register_port ( self , name , porttype , is_terminal , is_physical , flags ) : <EOL> """<STR_LIT>""" <EOL> if is_terminal : <EOL> flags |= _lib . JackPortIsTerminal <EOL> if is_physical : <EOL> flags |= _lib . JackPortIsPhysical <EOL> port_ptr = _lib . jack_port_register ( self . _ptr , name . encode ( ) , porttype , <EOL> flags , <NUM_LIT:0> ) <EOL> if not port_ptr : <EOL> raise JackError ( <EOL> "<STR_LIT>" . format ( name ) ) <EOL> return self . _wrap_port_ptr ( port_ptr ) <EOL> def _port_list_from_pointers ( self , names ) : <EOL> """<STR_LIT>""" <EOL> ports = [ ] <EOL> if names : <EOL> idx = <NUM_LIT:0> <EOL> while True : <EOL> name = names [ idx ] <EOL> if not name : <EOL> break <EOL> ports . append ( self . get_port_by_name ( _ffi . string ( name ) . decode ( ) ) ) <EOL> idx += <NUM_LIT:1> <EOL> return ports <EOL> def _get_port_ptr ( self , port ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( port , Port ) : <EOL> port = port . _ptr <EOL> elif isinstance ( port , str ) : <EOL> port = self . get_port_by_name ( port ) . _ptr <EOL> return port <EOL> def _wrap_port_ptr ( self , ptr ) : <EOL> """<STR_LIT>""" <EOL> porttype = _ffi . string ( _lib . jack_port_type ( ptr ) ) <EOL> if porttype == _AUDIO : <EOL> port = OwnPort ( ptr , self ) if self . owns ( ptr ) else Port ( ptr ) <EOL> elif porttype == _MIDI : <EOL> port = OwnMidiPort ( ptr , self ) if self . owns ( ptr ) else MidiPort ( ptr ) <EOL> else : <EOL> assert False <EOL> return port <EOL> class Port ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , port_ptr ) : <EOL> self . _ptr = port_ptr <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" . format ( self ) <EOL> def __eq__ ( self , other ) : <EOL> """<STR_LIT>""" <EOL> return self . _ptr == other . _ptr <EOL> def __ne__ ( self , other ) : <EOL> """<STR_LIT>""" <EOL> return not self . __eq__ ( other ) <EOL> @ property <EOL> def name ( self ) : <EOL> """<STR_LIT>""" <EOL> return _ffi . string ( _lib . jack_port_name ( self . _ptr ) ) . decode ( ) <EOL> @ property <EOL> def shortname ( self ) : <EOL> """<STR_LIT>""" <EOL> return _ffi . string ( _lib . jack_port_short_name ( self . _ptr ) ) . decode ( ) <EOL> @ shortname . setter <EOL> def shortname ( self , shortname ) : <EOL> _check ( _lib . jack_port_set_name ( self . _ptr , shortname . encode ( ) ) , <EOL> "<STR_LIT>" ) <EOL> @ property <EOL> def uuid ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . jack_port_uuid ( self . _ptr ) <EOL> is_audio = property ( lambda self : True , doc = "<STR_LIT>" ) <EOL> is_midi = property ( lambda self : False , doc = "<STR_LIT>" ) <EOL> @ property <EOL> def is_input ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackPortIsInput ) <EOL> @ property <EOL> def is_output ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackPortIsOutput ) <EOL> @ property <EOL> def is_physical ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackPortIsPhysical ) <EOL> @ property <EOL> def can_monitor ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackPortCanMonitor ) <EOL> @ property <EOL> def is_terminal ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackPortIsTerminal ) <EOL> def request_monitor ( self , onoff ) : <EOL> """<STR_LIT>""" <EOL> _check ( _lib . jack_port_request_monitor ( self . _ptr , onoff ) , <EOL> "<STR_LIT>" ) <EOL> def _hasflag ( self , flag ) : <EOL> """<STR_LIT>""" <EOL> return bool ( _lib . jack_port_flags ( self . _ptr ) & flag ) <EOL> class MidiPort ( Port ) : <EOL> """<STR_LIT>""" <EOL> is_audio = property ( lambda self : False , doc = "<STR_LIT>" ) <EOL> is_midi = property ( lambda self : True , doc = "<STR_LIT>" ) <EOL> class OwnPort ( Port ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , port_ptr , client ) : <EOL> Port . __init__ ( self , port_ptr ) <EOL> self . _client = client <EOL> @ property <EOL> def number_of_connections ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . jack_port_connected ( self . _ptr ) <EOL> @ property <EOL> def connections ( self ) : <EOL> """<STR_LIT>""" <EOL> names = _ffi . gc ( _lib . jack_port_get_connections ( self . _ptr ) , <EOL> _lib . jack_free ) <EOL> return self . _client . _port_list_from_pointers ( names ) <EOL> def is_connected_to ( self , port ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( port , Port ) : <EOL> port = port . name <EOL> return bool ( _lib . jack_port_connected_to ( self . _ptr , port . encode ( ) ) ) <EOL> def connect ( self , port ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( port , Port ) : <EOL> port = self . _client . get_port_by_name ( port ) <EOL> if self . is_output : <EOL> source = self <EOL> if not port . is_input : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> destination = port <EOL> elif self . is_input : <EOL> destination = self <EOL> if not port . is_output : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> source = port <EOL> else : <EOL> assert False <EOL> self . _client . connect ( source . name , destination . name ) <EOL> def disconnect ( self , other = None ) : <EOL> """<STR_LIT>""" <EOL> if other is None : <EOL> _check ( _lib . jack_port_disconnect ( self . _client . _ptr , self . _ptr ) , <EOL> "<STR_LIT>" . format ( self . name ) ) <EOL> else : <EOL> if self . is_output : <EOL> args = self , other <EOL> elif self . is_input : <EOL> args = other , self <EOL> self . _client . disconnect ( * args ) <EOL> def unregister ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . is_audio : <EOL> listname = "<STR_LIT>" <EOL> elif self . is_midi : <EOL> listname = "<STR_LIT>" <EOL> if self . is_input : <EOL> listname += "<STR_LIT>" <EOL> elif self . is_output : <EOL> listname += "<STR_LIT>" <EOL> ports = getattr ( self . _client , listname ) <EOL> ports . _portlist . remove ( self ) <EOL> _check ( _lib . jack_port_unregister ( self . _client . _ptr , self . _ptr ) , <EOL> "<STR_LIT>" . format ( self . name ) ) <EOL> def get_buffer ( self ) : <EOL> """<STR_LIT>""" <EOL> blocksize = self . _client . blocksize <EOL> return _ffi . buffer ( _lib . jack_port_get_buffer ( self . _ptr , blocksize ) , <EOL> blocksize * _ffi . sizeof ( "<STR_LIT:float>" ) ) <EOL> def get_array ( self ) : <EOL> """<STR_LIT>""" <EOL> import numpy as np <EOL> return np . frombuffer ( self . get_buffer ( ) , dtype = np . float32 ) <EOL> class OwnMidiPort ( MidiPort , OwnPort ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> OwnPort . __init__ ( self , * args , ** kwargs ) <EOL> self . _event = _ffi . new ( "<STR_LIT>" ) <EOL> def get_buffer ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def get_array ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> @ property <EOL> def max_event_size ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . jack_midi_max_event_size ( <EOL> _lib . jack_port_get_buffer ( self . _ptr , self . _client . blocksize ) ) <EOL> @ property <EOL> def lost_midi_events ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . jack_midi_get_lost_event_count ( <EOL> _lib . jack_port_get_buffer ( self . _ptr , self . _client . blocksize ) ) <EOL> def incoming_midi_events ( self ) : <EOL> """<STR_LIT>""" <EOL> event = self . _event <EOL> buf = _lib . jack_port_get_buffer ( self . _ptr , self . _client . blocksize ) <EOL> for i in range ( _lib . jack_midi_get_event_count ( buf ) ) : <EOL> err = _lib . jack_midi_event_get ( event , buf , i ) <EOL> assert not err , err <EOL> yield event . time , _ffi . buffer ( event . buffer , event . size ) <EOL> def clear_buffer ( self ) : <EOL> """<STR_LIT>""" <EOL> _lib . jack_midi_clear_buffer ( <EOL> _lib . jack_port_get_buffer ( self . _ptr , self . _client . blocksize ) ) <EOL> def write_midi_event ( self , time , event ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> event = _ffi . from_buffer ( event ) <EOL> except AttributeError : <EOL> pass <EOL> except TypeError : <EOL> pass <EOL> _check ( _lib . jack_midi_event_write ( <EOL> _lib . jack_port_get_buffer ( self . _ptr , self . _client . blocksize ) , <EOL> time , event , len ( event ) ) , "<STR_LIT>" ) <EOL> def reserve_midi_event ( self , time , size ) : <EOL> """<STR_LIT>""" <EOL> buf = _lib . jack_midi_event_reserve ( <EOL> _lib . jack_port_get_buffer ( self . _ptr , self . _client . blocksize ) , <EOL> time , size ) <EOL> return _ffi . buffer ( buf , size if buf else <NUM_LIT:0> ) <EOL> class Ports ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , client , porttype , flag ) : <EOL> self . _client = client <EOL> self . _type = porttype <EOL> self . _flag = flag <EOL> self . _portlist = [ ] <EOL> def __len__ ( self ) : <EOL> return self . _portlist . __len__ ( ) <EOL> def __getitem__ ( self , name ) : <EOL> return self . _portlist . __getitem__ ( name ) <EOL> def __iter__ ( self ) : <EOL> return self . _portlist . __iter__ ( ) <EOL> def __repr__ ( self ) : <EOL> return self . _portlist . __repr__ ( ) <EOL> def register ( self , shortname , is_terminal = False , is_physical = False ) : <EOL> """<STR_LIT>""" <EOL> port = self . _client . _register_port ( <EOL> shortname , self . _type , is_terminal , is_physical , self . _flag ) <EOL> self . _portlist . append ( port ) <EOL> return port <EOL> def clear ( self ) : <EOL> """<STR_LIT>""" <EOL> while self . _portlist : <EOL> self . _portlist [ <NUM_LIT:0> ] . unregister ( ) <EOL> class RingBuffer ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , size ) : <EOL> """<STR_LIT>""" <EOL> ptr = _lib . jack_ringbuffer_create ( size ) <EOL> if not ptr : <EOL> raise JackError ( "<STR_LIT>" ) <EOL> self . _ptr = _ffi . gc ( ptr , _lib . jack_ringbuffer_free ) <EOL> @ property <EOL> def write_space ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . jack_ringbuffer_write_space ( self . _ptr ) <EOL> def write ( self , data ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> data = _ffi . from_buffer ( data ) <EOL> except AttributeError : <EOL> pass <EOL> except TypeError : <EOL> pass <EOL> return _lib . jack_ringbuffer_write ( self . _ptr , data , len ( data ) ) <EOL> @ property <EOL> def write_buffers ( self ) : <EOL> """<STR_LIT>""" <EOL> vectors = _ffi . new ( "<STR_LIT>" ) <EOL> _lib . jack_ringbuffer_get_write_vector ( self . _ptr , vectors ) <EOL> return ( <EOL> _ffi . buffer ( vectors [ <NUM_LIT:0> ] . buf , vectors [ <NUM_LIT:0> ] . len ) , <EOL> _ffi . buffer ( vectors [ <NUM_LIT:1> ] . buf , vectors [ <NUM_LIT:1> ] . len ) <EOL> ) <EOL> def write_advance ( self , size ) : <EOL> """<STR_LIT>""" <EOL> _lib . jack_ringbuffer_write_advance ( self . _ptr , size ) <EOL> @ property <EOL> def read_space ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . jack_ringbuffer_read_space ( self . _ptr ) <EOL> def read ( self , size ) : <EOL> """<STR_LIT>""" <EOL> data = _ffi . new ( "<STR_LIT>" , size ) <EOL> size = _lib . jack_ringbuffer_read ( self . _ptr , data , size ) <EOL> return _ffi . buffer ( data , size ) <EOL> def peek ( self , size ) : <EOL> """<STR_LIT>""" <EOL> data = _ffi . new ( "<STR_LIT>" , size ) <EOL> size = _lib . jack_ringbuffer_peek ( self . _ptr , data , size ) <EOL> return _ffi . buffer ( data , size ) <EOL> @ property <EOL> def read_buffers ( self ) : <EOL> """<STR_LIT>""" <EOL> vectors = _ffi . new ( "<STR_LIT>" ) <EOL> _lib . jack_ringbuffer_get_read_vector ( self . _ptr , vectors ) <EOL> return ( <EOL> _ffi . buffer ( vectors [ <NUM_LIT:0> ] . buf , vectors [ <NUM_LIT:0> ] . len ) , <EOL> _ffi . buffer ( vectors [ <NUM_LIT:1> ] . buf , vectors [ <NUM_LIT:1> ] . len ) <EOL> ) <EOL> def read_advance ( self , size ) : <EOL> """<STR_LIT>""" <EOL> _lib . jack_ringbuffer_read_advance ( self . _ptr , size ) <EOL> def mlock ( self ) : <EOL> """<STR_LIT>""" <EOL> _check ( _lib . jack_ringbuffer_mlock ( self . _ptr ) , <EOL> "<STR_LIT>" ) <EOL> def reset ( self , size = None ) : <EOL> """<STR_LIT>""" <EOL> if size is None : <EOL> _lib . jack_ringbuffer_reset ( self . _ptr ) <EOL> else : <EOL> _lib . jack_ringbuffer_reset_size ( self . _ptr , size ) <EOL> @ property <EOL> def size ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _ptr . size <EOL> class Status ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = '<STR_LIT>' <EOL> def __init__ ( self , code ) : <EOL> self . _code = code <EOL> def __repr__ ( self ) : <EOL> flags = "<STR_LIT:U+002CU+0020>" . join ( name for name in dir ( self ) <EOL> if not name . startswith ( '<STR_LIT:_>' ) and getattr ( self , name ) ) <EOL> if not flags : <EOL> flags = "<STR_LIT>" <EOL> return "<STR_LIT>" . format ( self . _code , flags ) <EOL> @ property <EOL> def failure ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackFailure ) <EOL> @ property <EOL> def invalid_option ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackInvalidOption ) <EOL> @ property <EOL> def name_not_unique ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackNameNotUnique ) <EOL> @ property <EOL> def server_started ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackServerStarted ) <EOL> @ property <EOL> def server_failed ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackServerFailed ) <EOL> @ property <EOL> def server_error ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackServerError ) <EOL> @ property <EOL> def no_such_client ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackNoSuchClient ) <EOL> @ property <EOL> def load_failure ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackLoadFailure ) <EOL> @ property <EOL> def init_failure ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackInitFailure ) <EOL> @ property <EOL> def shm_failure ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackShmFailure ) <EOL> @ property <EOL> def version_error ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackVersionError ) <EOL> @ property <EOL> def backend_error ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackBackendError ) <EOL> @ property <EOL> def client_zombie ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hasflag ( _lib . JackClientZombie ) <EOL> def _hasflag ( self , flag ) : <EOL> """<STR_LIT>""" <EOL> return bool ( self . _code & flag ) <EOL> class TransportState ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = '<STR_LIT>' <EOL> def __init__ ( self , code ) : <EOL> self . _code = code <EOL> def __eq__ ( self , other ) : <EOL> return self . _code == other <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" + { <EOL> _lib . JackTransportStopped : '<STR_LIT>' , <EOL> _lib . JackTransportRolling : '<STR_LIT>' , <EOL> _lib . JackTransportStarting : '<STR_LIT>' , <EOL> _lib . JackTransportNetStarting : '<STR_LIT>' , <EOL> } [ self . _code ] <EOL> class JackError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class CallbackExit ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def position2dict ( pos ) : <EOL> """<STR_LIT>""" <EOL> assert pos . unique_1 == pos . unique_2 <EOL> keys = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> if pos . valid & _lib . JackPositionBBT : <EOL> keys += [ '<STR_LIT:bar>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> if pos . valid & _lib . JackPositionTimecode : <EOL> keys += [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> if pos . valid & _lib . JackBBTFrameOffset : <EOL> keys += [ '<STR_LIT>' ] <EOL> if pos . valid & _lib . JackAudioVideoRatio : <EOL> keys += [ '<STR_LIT>' ] <EOL> if pos . valid & _lib . JackVideoFrameOffset : <EOL> keys += [ '<STR_LIT>' ] <EOL> return dict ( ( k , getattr ( pos , k ) ) for k in keys ) <EOL> def version ( ) : <EOL> """<STR_LIT>""" <EOL> v = _ffi . new ( "<STR_LIT>" ) <EOL> _lib . jack_get_version ( v + <NUM_LIT:0> , v + <NUM_LIT:1> , v + <NUM_LIT:2> , v + <NUM_LIT:3> ) <EOL> return tuple ( v ) <EOL> def version_string ( ) : <EOL> """<STR_LIT>""" <EOL> return _ffi . string ( _lib . jack_get_version_string ( ) ) . decode ( ) <EOL> def client_name_size ( ) : <EOL> """<STR_LIT>""" <EOL> return _lib . jack_client_name_size ( ) <EOL> def port_name_size ( ) : <EOL> """<STR_LIT>""" <EOL> return _lib . jack_port_name_size ( ) <EOL> def set_error_function ( callback = None ) : <EOL> """<STR_LIT>""" <EOL> _set_error_or_info_function ( callback , _lib . jack_set_error_function ) <EOL> def set_info_function ( callback = None ) : <EOL> """<STR_LIT>""" <EOL> _set_error_or_info_function ( callback , _lib . jack_set_info_function ) <EOL> def client_pid ( name ) : <EOL> """<STR_LIT>""" <EOL> return _lib . jack_get_client_pid ( name . encode ( ) ) <EOL> def _set_error_or_info_function ( callback , setter ) : <EOL> """<STR_LIT>""" <EOL> if callback is None : <EOL> callback_wrapper = _ffi . NULL <EOL> else : <EOL> @ _ffi . callback ( "<STR_LIT>" ) <EOL> def callback_wrapper ( msg ) : <EOL> callback ( _ffi . string ( msg ) . decode ( ) ) <EOL> _keepalive [ setter ] = callback_wrapper <EOL> setter ( callback_wrapper ) <EOL> _keepalive = { } <EOL> def _check ( error_code , msg ) : <EOL> """<STR_LIT>""" <EOL> if error_code : <EOL> raise JackError ( "<STR_LIT>" . format ( msg , error_code ) ) </s>
