pulsectl.py 文件源码

def _pulse_method_call(pulse_op, func=None, index_arg=True):
        '''Creates following synchronous wrapper for async pa_operation callable:
            wrapper(index, ...) -> pulse_op(index, [*]args_func(...))
            index_arg=False: wrapper(...) -> pulse_op([*]args_func(...))'''
        def _wrapper(self, *args, **kws):
            if index_arg:
                if 'index' in kws: index = kws.pop('index')
                else: index, args = args[0], args[1:]
            pulse_args = func(*args, **kws) if func else list()
            if not is_list(pulse_args): pulse_args = [pulse_args]
            if index_arg: pulse_args = [index] + list(pulse_args)
            with self._pulse_op_cb() as cb:
                try: pulse_op(self._ctx, *(list(pulse_args) + [cb, None]))
                except c.ArgumentError as err: raise TypeError(err.args)
                except c.pa.CallError as err: raise PulseOperationInvalid(err.args[-1])
        func_args = list(inspect.getargspec(func or (lambda: None)))
        func_args[0] = list(func_args[0])
        if index_arg: func_args[0] = ['index'] + func_args[0]
        _wrapper.__name__ = '...'
        _wrapper.__doc__ = 'Signature: func' + inspect.formatargspec(*func_args)
        if func.__doc__: _wrapper.__doc__ += '\n\n' + func.__doc__
        return _wrapper