"""ctypes interface to the pulse library based on asyncio.""" import sys import asyncio import logging import re import pprint import ctypes as ct from abc import ABC from functools import partialmethod from . import __version__ from .libpulse_ctypes import PulseCTypes from .mainloop import MainLoop, pulse_ctypes, callback_func_ptr from .pulse_functions import pulse_functions from .pulse_enums import pulse_enums struct_ctypes = pulse_ctypes.struct_ctypes logger = logging.getLogger('libpuls') def _add_pulse_to_namespace(): def add_obj(name, obj): assert getattr(module, name, None) is None, f'{name} is duplicated' setattr(module, name, obj) # Add the pulse constants and functions to the module namespace. module = sys.modules[__name__] for name in pulse_functions['signatures']: func = pulse_ctypes.get_prototype(name) add_obj(name, func) for enum, constants in pulse_enums.items(): for name, value in constants.items(): add_obj(name, value) _add_pulse_to_namespace() del _add_pulse_to_namespace # /usr/include/pulse/def.h: # #define PA_INVALID_INDEX ((uint32_t) -1) PA_INVALID_INDEX = ct.c_uint32(-1).value # 'pa_volume_t' values defined in volume.h. PA_VOLUME_NORM = 0x10000 # Normal volume (100%, 0 dB). PA_VOLUME_MUTED = 0 # Muted (minimal valid) volume (0%, -inf dB). PA_VOLUME_MAX = 0xffffffff # UINT32_MAX/2 Maximum volume we can store. PA_VOLUME_INVALID = 0xffffffff # Special 'invalid' volume. def PA_VOLUME_IS_VALID(v): return v <= PA_VOLUME_MAX PA_CHANNELS_MAX = 32 # Defined in sample.h C_UINT_ARRAY_32 = ct.c_uint * PA_CHANNELS_MAX C_INT_ARRAY_32 = ct.c_int * PA_CHANNELS_MAX # Map values to their name. CTX_STATES = dict((eval(state), state) for state in ('PA_CONTEXT_UNCONNECTED', 'PA_CONTEXT_CONNECTING', 'PA_CONTEXT_AUTHORIZING', 'PA_CONTEXT_SETTING_NAME', 'PA_CONTEXT_READY', 'PA_CONTEXT_FAILED', 'PA_CONTEXT_TERMINATED')) OPERATION_STATES = dict((eval(state), state) for state in ('PA_OPERATION_CANCELLED', 'PA_OPERATION_DONE', 'PA_OPERATION_RUNNING')) def event_codes_to_names(): def build_events_dict(mask): for fac in globals(): if fac.startswith(prefix): val = eval(fac) if (val & mask) and val != mask: yield val, fac[prefix_len:].lower() prefix = 'PA_SUBSCRIPTION_EVENT_' prefix_len = len(prefix) facilities = {0 : 'sink'} facilities.update(build_events_dict(PA_SUBSCRIPTION_EVENT_FACILITY_MASK)) event_types = {0: 'new'} event_types.update(build_events_dict(PA_SUBSCRIPTION_EVENT_TYPE_MASK)) return facilities, event_types # Dictionaries mapping libpulse events values to their names. EVENT_FACILITIES, EVENT_TYPES = event_codes_to_names() def run_in_task(coro): """Decorator to wrap a coroutine in a task of AsyncioTasks instance.""" async def wrapper(*args, **kwargs): def get_coro_arg(): length = len(args) coro_arg = '' if length >=2: coro_arg += f'{args[1].__qualname__}(' if length >= 3: coro_arg += f'{args[2]})' else: coro_arg += ')' return coro_arg if 0: # When enabled while running the test suite, will print all the # pulseaudio coroutines that are being tested. print(f'{coro.__qualname__}({get_coro_arg()})', file=sys.stderr) lib_pulse = LibPulse._get_instance() if lib_pulse is None: raise LibPulseClosedError try: return await lib_pulse.libpulse_tasks.create_task( coro(*args, **kwargs)) except asyncio.CancelledError: logger.warning(f'{coro.__qualname__}({get_coro_arg()})' ' has been cancelled') raise return wrapper class LibPulseError(Exception): pass class LibPulseClosedError(LibPulseError): pass class LibPulseStateError(LibPulseError): pass class LibPulseOperationError(LibPulseError): pass class LibPulseClosedIteratorError(LibPulseError): pass class LibPulseInstanceExistsError(LibPulseError): pass class LibPulseArgumentError(LibPulseError): pass class EventIterator: """Pulse events asynchronous iterator.""" QUEUE_CLOSED = object() def __init__(self): self.event_queue = asyncio.Queue() self.closed = False # Public methods. def close(self): self.closed = True # Private methods. def abort(self): while True: try: self.event_queue.get_nowait() except asyncio.QueueEmpty: break self.put_nowait(self.QUEUE_CLOSED) def put_nowait(self, obj): if not self.closed: self.event_queue.put_nowait(obj) def __aiter__(self): return self async def __anext__(self): if self.closed: logger.info('Events Asynchronous Iterator is closed') raise StopAsyncIteration try: event = await self.event_queue.get() except asyncio.CancelledError: self.close() raise StopAsyncIteration if event is not self.QUEUE_CLOSED: return event self.close() raise LibPulseClosedIteratorError('Got QUEUE_CLOSED') class AsyncioTasks: def __init__(self): self._tasks = set() def create_task(self, coro): task = asyncio.create_task(coro) self._tasks.add(task) task.add_done_callback(lambda t: self._tasks.remove(t)) return task def __iter__(self): for t in self._tasks: yield t class PulseEvent: """A libpulse event. Use the event_facilities() and event_types() static methods to get all the values currently defined by the libpulse library for 'facility' and 'type'. They correspond to some of the variables defined in the pulse_enums module under the pa_subscription_event_type Enum. attributes: facility: str - name of the facility, for example 'sink'. index: int - index of the facility. type: str - type of event, normaly 'new', 'change' or 'remove'. """ def __init__(self, event_type, index): fac = event_type & PA_SUBSCRIPTION_EVENT_FACILITY_MASK assert fac in EVENT_FACILITIES self.facility = EVENT_FACILITIES[fac] type = event_type & PA_SUBSCRIPTION_EVENT_TYPE_MASK assert type in EVENT_TYPES self.type = EVENT_TYPES[type] self.index = index @staticmethod def event_facilities(): return list(EVENT_FACILITIES.values()) @staticmethod def event_types(): return list(EVENT_TYPES.values()) class PropList(dict): """Dictionary of the elements of a proplist whose value is a string.""" def __init__(self, c_pa_proplist): super().__init__() null_ptr = ct.POINTER(ct.c_void_p)() null_ptr_ptr = ct.pointer(null_ptr) while True: key = pa_proplist_iterate(c_pa_proplist, null_ptr_ptr) if not key: break elif isinstance(key, bytes): val = pa_proplist_gets(c_pa_proplist, key) if val: self[key.decode()] = val.decode() class PulseStructure: """The representation of a ctypes Structure. When returned by a callback as a pointer to a structure, one must make a deep copy of the elements of the structure as they are only temporarily available. """ ignored_pointer_names = set() array_sizes = { 'pa_card_port_info': 'n_ports', 'pa_source_port_info': 'n_ports', 'pa_sink_port_info': 'n_ports', 'pa_card_profile_info': 'n_profiles', 'pa_card_profile_info2': 'n_profiles', 'pa_format_info': 'n_formats', } def __init__(self, c_struct, c_structure_type): for name, c_type in c_structure_type._fields_: fq_name = f'{c_structure_type.__name__}.{name}: {c_type.__name__}' if fq_name in self.ignored_pointer_names: continue try: c_struct_val = getattr(c_struct, name) except AttributeError: assert False, (f'{fq_name} not found while instantiating' f' a PulseStructure') if c_type in PulseCTypes.numeric_types.values(): setattr(self, name, c_struct_val) # A NULL pointer. elif not c_struct_val: setattr(self, name, None) elif c_type is ct.c_char_p: setattr(self, name, c_struct_val.decode()) # A proplist pointer. elif (isinstance(c_struct_val, ct._Pointer) and c_struct_val._type_.__name__ == 'pa_proplist'): setattr(self, name, PropList(c_struct_val)) # An array. elif isinstance(c_struct_val, ct.Array): # All libpulse arrays have a numeric type. setattr(self, name, c_struct_val[:]) # An array of pointers. elif (isinstance(c_struct_val, ct._Pointer) and isinstance(c_struct_val.contents, ct._Pointer)): ctype = c_struct_val.contents._type_ if ctype.__name__ in struct_ctypes: val = [] ptr = c_struct_val size_attr = self.array_sizes[ctype.__name__] array_size = getattr(self, size_attr) for i in range(array_size): if not ptr[i]: break val.append(PulseStructure(ptr[i].contents, ctype)) setattr(self, name, val) else: self.ignore_member(fq_name) # A pointer. elif isinstance(c_struct_val, ct._Pointer): if c_struct_val._type_.__name__ in struct_ctypes: setattr(self, name, PulseStructure(c_struct_val.contents, c_struct_val._type_)) else: self.ignore_member(fq_name) # A structure. else: if c_type.__name__ in struct_ctypes: setattr(self, name, PulseStructure(c_struct_val, c_type)) else: self.ignore_member(fq_name) def ignore_member(self, name): self.ignored_pointer_names.add(name) logger.debug(f"Ignoring '{name}' structure member") def __repr__(self): return pprint.pformat(self.__dict__, sort_dicts=False) class CtypesPulseStructure(ABC): """Container for an instance of a subclass of ctypes.Structure. Scalar attributes of the instance may be updated after instanciation as well as the scalar elements of an array, but not the array itself or the other aggregate attributes. """ @staticmethod def get_fields_names(struct_name): return [field[0] for field in struct_ctypes[struct_name]._fields_] def __getattr__(self, name): if name in self.fields_names: return getattr(self.ct_struct, name) else: raise AttributeError( f"'{self.__class__.__name__}' object has no attribute '{name}'") def __setattr__(self, name, val): if name in self.fields_names: setattr(self.c_obj, name, val) else: object.__setattr__(self, name, val) def byref(self): return ct.byref(self.ct_struct) def to_pulse_structure(self): return PulseStructure(self.ct_struct, struct_ctypes[self.struct_name]) class Pa_buffer_attr(CtypesPulseStructure): struct_name = 'pa_buffer_attr' fields_names = CtypesPulseStructure.get_fields_names(struct_name) def __init__(self, maxlength, tlength, prebuf, minreq, fragsize): self.ct_struct = struct_ctypes[self.struct_name]( maxlength, tlength, prebuf, minreq, fragsize) class Pa_cvolume(CtypesPulseStructure): struct_name = 'pa_cvolume' fields_names = CtypesPulseStructure.get_fields_names(struct_name) def __init__(self, channels, values): length = len(values) assert length <= PA_CHANNELS_MAX values += [0 for i in range(PA_CHANNELS_MAX - length)] self.ct_struct = struct_ctypes[self.struct_name]( channels, C_UINT_ARRAY_32(*values)) class Pa_channel_map(CtypesPulseStructure): struct_name = 'pa_channel_map' fields_names = CtypesPulseStructure.get_fields_names(struct_name) def __init__(self, channels, map): length = len(map) assert length <= PA_CHANNELS_MAX map += [0 for i in range(PA_CHANNELS_MAX - length)] self.ct_struct = struct_ctypes[self.struct_name]( channels, C_INT_ARRAY_32(*map)) class Pa_format_info(CtypesPulseStructure): struct_name = 'pa_format_info' fields_names = CtypesPulseStructure.get_fields_names(struct_name) def __init__(self, encoding, plist): # 'plist' is an instance of ctypes._Pointer that has been built using # some of the pa_proplist_ functions. self.ct_struct = struct_ctypes[self.struct_name](encoding, plist) class Pa_sample_spec(CtypesPulseStructure): struct_name = 'pa_sample_spec' fields_names = CtypesPulseStructure.get_fields_names(struct_name) def __init__(self, format, rate, channels): self.ct_struct = struct_ctypes[self.struct_name]( format, rate, channels) class LibPulse: """Interface to libpulse library as an asynchronous context manager.""" ASYNCIO_LOOPS = dict() # {asyncio loop: LibPulse instance} # Function signature: (pa_operation *, # [pa_context *, [args...], cb_t, void *]) # Callback signature: (void, [pa_context *, [objs...], void *]) context_methods = ( 'pa_context_add_autoload', 'pa_context_drain', 'pa_context_get_server_info', 'pa_context_load_module', 'pa_context_play_sample_with_proplist', # pa_context_send_message_to_object() is the only method that is not # in the 'LibPulse.context_list_methods' list and that returns a list # of objects. See the 'pa_context_string_cb_t' callback signature in # the pulse_functions module. 'pa_context_send_message_to_object', # The context state is monitored by the LibPulse instance. # 'pa_context_set_state_callback', # Reception of events is monitored by the LibPulse instance. # 'pa_context_set_subscribe_callback', 'pa_context_stat', 'pa_ext_device_restore_test', ) # Function signature: (pa_operation *, # [pa_context *, [args...], cb_t, void *]) # Callback signature: (void, [pa_context *, int, void *]) context_success_methods = ( 'pa_context_exit_daemon', 'pa_context_kill_client', 'pa_context_kill_sink_input', 'pa_context_kill_source_output', 'pa_context_move_sink_input_by_index', 'pa_context_move_sink_input_by_name', 'pa_context_move_source_output_by_index', 'pa_context_move_source_output_by_name', 'pa_context_play_sample', 'pa_context_proplist_remove', 'pa_context_proplist_update', 'pa_context_remove_autoload_by_index', 'pa_context_remove_autoload_by_name', 'pa_context_remove_sample', 'pa_context_set_card_profile_by_index', 'pa_context_set_card_profile_by_name', 'pa_context_set_default_sink', 'pa_context_set_default_source', 'pa_context_set_name', 'pa_context_set_port_latency_offset', 'pa_context_set_sink_input_mute', 'pa_context_set_sink_input_volume', 'pa_context_set_sink_mute_by_index', 'pa_context_set_sink_mute_by_name', 'pa_context_set_sink_port_by_index', 'pa_context_set_sink_port_by_name', 'pa_context_set_sink_volume_by_index', 'pa_context_set_sink_volume_by_name', 'pa_context_set_source_mute_by_index', 'pa_context_set_source_mute_by_name', 'pa_context_set_source_output_mute', 'pa_context_set_source_output_volume', 'pa_context_set_source_port_by_index', 'pa_context_set_source_port_by_name', 'pa_context_set_source_volume_by_index', 'pa_context_set_source_volume_by_name', 'pa_context_subscribe', 'pa_context_suspend_sink_by_index', 'pa_context_suspend_sink_by_name', 'pa_context_suspend_source_by_index', 'pa_context_suspend_source_by_name', 'pa_context_unload_module', 'pa_ext_device_restore_save_formats', 'pa_ext_device_restore_subscribe', ) # Function signature: (pa_operation *, [pa_context *, cb_t, void *]) # Callback signature: (void, [pa_context *, struct *, int, void *]) context_list_methods = ( 'pa_context_get_autoload_info_by_index', 'pa_context_get_autoload_info_by_name', 'pa_context_get_autoload_info_list', 'pa_context_get_card_info_by_index', 'pa_context_get_card_info_by_name', 'pa_context_get_card_info_list', 'pa_context_get_client_info', 'pa_context_get_client_info_list', 'pa_context_get_module_info', 'pa_context_get_module_info_list', 'pa_context_get_sample_info_by_index', 'pa_context_get_sample_info_by_name', 'pa_context_get_sample_info_list', 'pa_context_get_sink_info_by_index', 'pa_context_get_sink_info_by_name', 'pa_context_get_sink_info_list', 'pa_context_get_sink_input_info', 'pa_context_get_sink_input_info_list', 'pa_context_get_source_info_by_index', 'pa_context_get_source_info_by_name', 'pa_context_get_source_info_list', 'pa_context_get_source_output_info', 'pa_context_get_source_output_info_list', 'pa_ext_device_restore_read_formats', 'pa_ext_device_restore_read_formats_all', ) # Function signature: (pa_operation *, # [pa_stream *, [args...], cb_t, void *]) # Callback signature: (void, [pa_stream *, int, void *]) stream_success_methods = ( 'pa_stream_cork', 'pa_stream_drain', 'pa_stream_flush', 'pa_stream_prebuf', 'pa_stream_proplist_remove', 'pa_stream_proplist_update', 'pa_stream_set_buffer_attr', 'pa_stream_set_name', 'pa_stream_trigger', 'pa_stream_update_sample_rate', 'pa_stream_update_timing_info', ) def __init__(self, name, server=None, flags=PA_CONTEXT_NOAUTOSPAWN): """ Constructor arguments: - 'name' Name of the application. - 'server' Server name, if 'server' is None, connect to the default server. - 'flags' 'flags' and 'server' are arguments of pa_context_connect() used to connect to the server. """ logger.info(f'Python libpulse version {__version__}') assert isinstance(name, str) if server is not None: assert isinstance(server, str) server = server.encode() self.server = server self.flags = flags self.loop = asyncio.get_running_loop() if self.loop in self.ASYNCIO_LOOPS: raise LibPulseInstanceExistsError self.c_context = pa_context_new(MainLoop.C_MAINLOOP_API, name.encode()) # From the ctypes documentation: "NULL pointers have a False # boolean value". if not self.c_context: raise RuntimeError('Cannot get context from libpulse library') self.closed = False self.state = ('PA_CONTEXT_UNCONNECTED', 'PA_OK') self.main_task = asyncio.current_task(self.loop) self.libpulse_tasks = AsyncioTasks() self.state_notification = self.loop.create_future() self.event_iterator = None self.ASYNCIO_LOOPS[self.loop] = self LibPulse.add_async_methods() # Keep a reference to prevent garbage collection. self.c_context_state_callback = callback_func_ptr( 'pa_context_notify_cb_t', LibPulse.context_state_callback) self.c_context_subscribe_callback = callback_func_ptr( 'pa_context_subscribe_cb_t', LibPulse.context_subscribe_callback) # Initialisation. @staticmethod def add_async_methods(): # Register the partial methods. method_types = { 'context_methods': LibPulse._pa_context_get, 'context_success_methods': LibPulse._pa_context_op_success, 'context_list_methods': LibPulse._pa_context_get_list, 'stream_success_methods': LibPulse._pa_stream_op_success, } this_module = sys.modules[__name__] for method_type, libpulse_method in method_types.items(): func_names = getattr(LibPulse, method_type) for func_name in func_names: setattr(LibPulse, func_name, partialmethod(libpulse_method, func_name)) if hasattr(this_module, func_name): delattr(this_module, func_name) @staticmethod def _get_instance(): """Get the LibPulse instance running on the asyncio loop. The instance may not be yet connected or be in the closing state. Prefer using get_current_instance(). """ loop = asyncio.get_running_loop() try: return LibPulse.ASYNCIO_LOOPS[loop] except KeyError: return None @staticmethod def context_state_callback(c_context, c_userdata): """Call back that monitors the connection state.""" lib_pulse = LibPulse._get_instance() if lib_pulse is None: return st = pa_context_get_state(c_context) st = CTX_STATES[st] if st in ('PA_CONTEXT_READY', 'PA_CONTEXT_FAILED', 'PA_CONTEXT_TERMINATED'): error = pa_strerror(pa_context_errno(c_context)) state = (st, error.decode()) logger.info(f'LibPulse connection: {state}') state_notification = lib_pulse.state_notification lib_pulse.state = state if not state_notification.done(): state_notification.set_result(state) elif not lib_pulse.closed and st != 'PA_CONTEXT_READY': # A task is used here instead of calling directly abort() so # that pa_context_connect() has the time to handle a # previous PA_CONTEXT_READY state. asyncio.create_task(lib_pulse.abort(state)) else: logger.debug(f'LibPulse connection: {st}') @run_in_task async def _pa_context_connect(self): """Connect the context to the default server.""" pa_context_set_state_callback(self.c_context, self.c_context_state_callback, None) rc = pa_context_connect(self.c_context, self.server, self.flags, None) logger.debug(f'pa_context_connect return code: {rc}') await self.state_notification if self.state[0] != 'PA_CONTEXT_READY': raise LibPulseStateError(self.state) @staticmethod def context_subscribe_callback(c_context, event_type, index, c_userdata): """Call back to handle pulseaudio events.""" lib_pulse = LibPulse._get_instance() if lib_pulse is None: return if lib_pulse.event_iterator is not None: lib_pulse.event_iterator.put_nowait(PulseEvent(event_type, index)) # Libpulse async methods workers. @staticmethod def get_callback_data(func_name): # Get name and signature of the callback argument of 'func_name'. func_sig = pulse_functions['signatures'][func_name] args = func_sig[1] for arg in args: if arg in pulse_functions['callbacks']: callback_name = arg callback_sig = pulse_functions['callbacks'][arg] assert len(args) >= 3 and arg == args[-2] return callback_name, callback_sig def call_ctypes_func(self, func_name, operation_type, cb_func_ptr, *func_args): # Call the 'func_name' ctypes function. args = [] for arg in func_args: arg = arg.encode() if isinstance(arg, str) else arg args.append(arg) func_proto = pulse_ctypes.get_prototype(func_name) try: c_operation = func_proto(operation_type, *args, cb_func_ptr, None) except ct.ArgumentError as e: first_arg = ('c_context' if operation_type == self.c_context else 'pa_stream') raise LibPulseArgumentError( f"\nException reported by ctypes:\n" f" {e!r}" f"\nFunction arguments:\n" f" {func_name}{(first_arg, *args, cb_func_ptr, None)}\n" ) return c_operation @staticmethod async def handle_operation(c_operation, future): # From the ctypes documentation: "NULL pointers have a False # boolean value". if not c_operation: future.cancel() error = "NULL 'pa_operation' pointer" lib_pulse = LibPulse._get_instance() if lib_pulse is not None: errmsg = pa_strerror(pa_context_errno(lib_pulse.c_context)) error += f': {errmsg.decode()}' raise LibPulseOperationError(error) try: await future except asyncio.CancelledError: pa_operation_cancel(c_operation) raise finally: pa_operation_unref(c_operation) async def _pa_get(self, func_name, operation_type, *func_args): """Call an asynchronous pulse function that does not return a list. 'func_args' is the sequence of the arguments of the function preceding the callback in the function signature. The last argument (i.e. 'userdata') is set to None by call_ctypes_func(). """ def callback_func(c_operation_type, *c_results): results = [] try: for arg, c_result in zip(callback_sig[1][1:-1], c_results[:-1]): arg_list = arg.split() if arg_list[-1] == '*' and arg_list[0] in struct_ctypes: struct_name = arg_list[0] if not c_result: results.append(None) else: results.append(PulseStructure(c_result.contents, struct_ctypes[struct_name])) else: if arg == 'char *': c_result = c_result.decode() results.append(c_result) except Exception as e: results = e finally: if not notification.done(): notification.set_result(results) callback_data = self.get_callback_data(func_name) assert callback_data, f'{func_name} signature without a callback' callback_name, callback_sig = callback_data notification = self.loop.create_future() # Await on the future. cb_func_ptr = callback_func_ptr(callback_name, callback_func) c_operation = self.call_ctypes_func(func_name, operation_type, cb_func_ptr, *func_args) await LibPulse.handle_operation(c_operation, notification) results = notification.result() if isinstance(results, Exception): raise results for result in results: if result is None: raise LibPulseOperationError( 'NULL pointer result returned by the callback') if len(results) == 1: return results[0] return results async def _pa_get_list(self, func_name, operation_type, *func_args): """Call an asynchronous pulse function that returns a list. 'func_args' is the sequence of the arguments of the function preceding the callback in the function signature. The last argument (i.e. 'userdata') is set to None by call_ctypes_func(). """ def info_callback(c_operation_type, c_info, eol, c_userdata): # From the ctypes documentation: "NULL pointers have a False # boolean value". if not c_info: if not notification.done(): notification.set_result(eol) else: try: arg = callback_sig[1][1] arg_list = arg.split() assert arg_list[-1] == '*' assert arg_list[0] in struct_ctypes struct_name = arg_list[0] infos.append(PulseStructure(c_info.contents, struct_ctypes[struct_name])) except Exception as e: if not notification.done(): notification.set_result(e) callback_data = self.get_callback_data(func_name) assert callback_data, f'{func_name} signature without a callback' callback_name, callback_sig = callback_data infos = [] notification = self.loop.create_future() # Await on the future. cb_func_ptr = callback_func_ptr(callback_name, info_callback) c_operation = self.call_ctypes_func(func_name, operation_type, cb_func_ptr, *func_args) await LibPulse.handle_operation(c_operation, notification) eol = notification.result() if isinstance(eol, Exception): raise eol if eol < 0: error = "'eol' set to a negative value by the callback" errmsg = pa_strerror(pa_context_errno(self.c_context)) error += f': {errmsg.decode()}' raise LibPulseOperationError(error) if func_name.endswith(('_by_name', '_by_index', '_info', '_formats')): assert len(infos) == 1 return infos[0] return infos @run_in_task async def _run_in_task(self, method, func_name, operation_type, *func_args): try: return await method(func_name, operation_type, *func_args) except (Exception, asyncio.CancelledError) as e: return e async def _pa_context_get(self, func_name, *func_args): result = await self._run_in_task(self._pa_get, func_name, self.c_context, *func_args) if isinstance(result, (Exception, asyncio.CancelledError)): raise result return result async def _pa_success(self, func_name, operation_type, *func_args): success = await self._run_in_task(self._pa_get, func_name, operation_type, *func_args) if isinstance(success, (Exception, asyncio.CancelledError)): raise success return success async def _pa_context_op_success(self, func_name, *func_args): result = await self._pa_success(func_name, self.c_context, *func_args) if result != PA_OPERATION_DONE: error = pa_strerror(pa_context_errno(self.c_context)) raise LibPulseOperationError(f'Failure: {error.decode()}: ' f'{OPERATION_STATES[result]!r}') return result async def _pa_stream_op_success(self, func_name, pa_stream, *func_args): return await self._pa_success(func_name, pa_stream, *func_args) async def _pa_context_get_list(self, func_name, *func_args): result = await self._run_in_task(self._pa_get_list, func_name, self.c_context, *func_args) if isinstance(result, (Exception, asyncio.CancelledError)): raise result else: return result # Context manager. async def abort(self, state): # Cancelling the main task does close the LibPulse context manager. logger.error(f'The LibPulse instance has been aborted: {state}') self.main_task.cancel() async def close(self): if self.closed: return self.closed = True try: for task in self.libpulse_tasks: task.cancel() if self.event_iterator is not None: self.event_iterator.abort() pa_context_set_state_callback(self.c_context, None, None) pa_context_set_subscribe_callback(self.c_context, None, None) if self.state[0] == 'PA_CONTEXT_READY': try: await self.pa_context_drain() except (Exception, asyncio.CancelledError): # Quoting pulse documentation: # If there is nothing to drain, the function returns NULL. pass pa_context_disconnect(self.c_context) logger.info('Disconnected from libpulse context') finally: pa_context_unref(self.c_context) for loop, lib_pulse in list(self.ASYNCIO_LOOPS.items()): if lib_pulse is self: del self.ASYNCIO_LOOPS[loop] break else: logger.error('Cannot remove LibPulse instance upon closing') MainLoop.close() logger.debug('LibPulse instance closed') async def __aenter__(self): try: # Set up the two callbacks that live until this instance is # closed. self.main_task = asyncio.current_task(self.loop) await self._pa_context_connect() pa_context_set_subscribe_callback(self.c_context, self.c_context_subscribe_callback, None) return self except asyncio.CancelledError: await self.close() if self.state[0] != 'PA_CONTEXT_READY': raise LibPulseStateError(self.state) except Exception: await self.close() raise async def __aexit__(self, exc_type, exc_value, traceback): await self.close() if exc_type is asyncio.CancelledError: if self.state[0] != 'PA_CONTEXT_READY': raise LibPulseStateError(self.state) # Public methods. @staticmethod async def get_current_instance(): """Get the LibPulse running instance. Raises LibPulseStateError if the instance is not in the PA_CONTEXT_READY state. """ lib_pulse = LibPulse._get_instance() if lib_pulse is not None: await lib_pulse.state_notification if lib_pulse.state[0] != 'PA_CONTEXT_READY': raise LibPulseStateError(lib_pulse.state) return lib_pulse def get_events_iterator(self): """Return an Asynchronous Iterator of libpulse events. The iterator is used to run an async for loop over the PulseEvent instances. The async for loop can be terminated by invoking the close() method of the iterator from within the loop or from another task. """ if self.closed: raise LibPulseOperationError('The LibPulse instance is closed') if self.event_iterator is not None and not self.event_iterator.closed: raise LibPulseError('Not allowed: the current Asynchronous' ' Iterator must be closed first') self.event_iterator = EventIterator() return self.event_iterator async def log_server_info(self): if self.state[0] != 'PA_CONTEXT_READY': raise LibPulseStateError(self.state) server_info = await self.pa_context_get_server_info() server_name = server_info.server_name if re.match(r'.*\d+\.\d', server_name): # Pipewire includes the server version in the server name. logger.info(f'Server: {server_name}') else: logger.info(f'Server: {server_name} {server_info.server_version}') version = pa_context_get_protocol_version(self.c_context) server_ver = pa_context_get_server_protocol_version(self.c_context) logger.debug(f'libpulse library/server versions: ' f'{version}/{server_ver}') # 'server' is the name of the socket libpulse is connected to. server = pa_context_get_server(self.c_context) logger.debug(f'{server_name} connected to {server.decode()}')