Take the 2-minute tour ×
Ask Ubuntu is a question and answer site for Ubuntu users and developers. It's 100% free, no registration required.

Accessing PulseAudio with Python2 is a bit tricky, but it works with the help of ctypes. The problem is that the same code fails to work with Python3. Apparently something was changed in ctypes, but I can't figure out what exactly.

Here's the sample code, which will list all the source devices. It works in Python2, but segfaults horribly in Python3.

# PulseAudio Source list:

import time
from ctypes import *
PA = CDLL('libpulse.so.0')
PA_CONTEXT_UNCONNECTED = 0
PA_CONTEXT_CONNECTING = 1
PA_CONTEXT_AUTHORIZING = 2
PA_CONTEXT_SETTING_NAME = 3
PA_CONTEXT_READY = 4
PA_CONTEXT_FAILED = 5
PA_CONTEXT_TERMINATED = 6
PA_OPERATION_RUNNING = 0
PA_OPERATION_DONE = 1
PA_OPERATION_CANCELLED = 2
STRING = c_char_p
size_t = c_ulong
uint32_t = c_uint32
uint8_t = c_uint8
class pa_mainloop_api(Structure):
    pass
class pa_threaded_mainloop(Structure):
    pass
pa_threaded_mainloop._fields_ = []
pa_threaded_mainloop_new = PA.pa_threaded_mainloop_new
pa_threaded_mainloop_new.restype = POINTER(pa_threaded_mainloop)
pa_threaded_mainloop_new.argtypes = []
pa_threaded_mainloop_free = PA.pa_threaded_mainloop_free
pa_threaded_mainloop_free.restype = None
pa_threaded_mainloop_free.argtypes = [POINTER(pa_threaded_mainloop)]
pa_threaded_mainloop_start = PA.pa_threaded_mainloop_start
pa_threaded_mainloop_start.restype = c_int
pa_threaded_mainloop_start.argtypes = [POINTER(pa_threaded_mainloop)]
pa_threaded_mainloop_stop = PA.pa_threaded_mainloop_stop
pa_threaded_mainloop_stop.restype = None
pa_threaded_mainloop_stop.argtypes = [POINTER(pa_threaded_mainloop)]
pa_threaded_mainloop_get_api = PA.pa_threaded_mainloop_get_api
pa_threaded_mainloop_get_api.restype = POINTER(pa_mainloop_api)
pa_threaded_mainloop_get_api.argtypes = [POINTER(pa_threaded_mainloop)]
class pa_context(Structure):
    pass
pa_context._fields_ = []
class pa_spawn_api(Structure):
    pass
class pa_stream(Structure):
    pass
pa_stream._fields_ = []
class pa_operation(Structure):
        pass
class pa_source_info(Structure):
    pass
pa_source_info._fields_ = [
    ('name', STRING),
    ('index', uint32_t),
    ('description', STRING),
]
pa_context_flags = c_int
pa_context_flags_t = pa_context_flags
pa_context_state = c_int
pa_context_state_t = pa_context_state
pa_context_notify_cb_t = CFUNCTYPE(None, POINTER(pa_context), c_void_p)
pa_context_success_cb_t = CFUNCTYPE(None, POINTER(pa_context), c_int, c_void_p)
pa_stream_success_cb_t = CFUNCTYPE(None, POINTER(pa_stream), c_int, c_void_p)
pa_stream_request_cb_t = CFUNCTYPE(None, POINTER(pa_stream), size_t, c_void_p)
pa_stream_notify_cb_t = CFUNCTYPE(None, POINTER(pa_stream), c_void_p)
pa_source_info_cb_t = CFUNCTYPE(None, POINTER(pa_context), POINTER(pa_source_info), c_int, c_void_p)

pa_context_new = PA.pa_context_new
pa_context_new.restype = POINTER(pa_context)
pa_context_new.argtypes = [POINTER(pa_mainloop_api), STRING]
pa_context_connect = PA.pa_context_connect
pa_context_connect.restype = c_int
pa_context_connect.argtypes = [POINTER(pa_context), STRING, pa_context_flags_t, POINTER(pa_spawn_api)]
pa_context_set_state_callback = PA.pa_context_set_state_callback
pa_context_set_state_callback.restype = None
pa_context_set_state_callback.argtypes = [POINTER(pa_context), pa_context_notify_cb_t, c_void_p]
pa_context_get_state = PA.pa_context_get_state
pa_context_get_state.restype = pa_context_state_t
pa_context_get_state.argtypes = [POINTER(pa_context)]
pa_stream_set_state_callback = PA.pa_stream_set_state_callback
pa_stream_set_state_callback.restype = None
pa_stream_set_state_callback.argtypes = [POINTER(pa_stream), pa_stream_notify_cb_t, c_void_p]

pa_context_get_source_info_list = PA.pa_context_get_source_info_list
pa_context_get_source_info_list.restype = POINTER(pa_operation)
pa_context_get_source_info_list.argtypes = [POINTER(pa_context), pa_source_info_cb_t, c_void_p]

def pa_state_cb(context, userdata):
    global pa_state
    state = pa_context_get_state(context)
    if state in [PA_CONTEXT_UNCONNECTED, PA_CONTEXT_CONNECTING, PA_CONTEXT_AUTHORIZING,
        PA_CONTEXT_SETTING_NAME]:
        pa_state = 0
    elif state == PA_CONTEXT_FAILED:
        pa_state = 2
    elif state == PA_CONTEXT_READY:
        pa_state = 1
    return  0

def pa_sourcelist_cb(context, source_info, eol, userdata):
    if eol==0:
        print("IDX:", source_info.contents.index, " ", source_info.contents.name, "--", source_info.contents.description)
    return  0

pa_ml = pa_threaded_mainloop_new()
pa_mlapi = pa_threaded_mainloop_get_api(pa_ml)

pa_ctx = pa_context_new(pa_mlapi, b"kazam-pulse")

error = -1
pa_state = 0
if pa_context_connect(pa_ctx, None, 0, None)!=0:
    print("cannot connect..")
    sys.exit(1)

pa_context_set_state_callback(pa_ctx, pa_context_notify_cb_t(pa_state_cb), None)
ret = pa_threaded_mainloop_start(pa_ml)

ctc = pa_context_get_state(pa_ctx)

time.sleep(1)
pa_context_get_source_info_list(pa_ctx, pa_source_info_cb_t(pa_sourcelist_cb), None);
time.sleep(1)

And here's the backtrace:

#0  0x00007ffff7e11fc1 in ?? ()
#1  0x00007ffff5e82d27 in pa_context_set_state (st=PA_CONTEXT_AUTHORIZING, c=0xae7d40) at pulse/context.c:291
#2  pa_context_set_state (c=0xae7d40, st=PA_CONTEXT_AUTHORIZING) at pulse/context.c:279
#3  0x00007ffff5e85691 in setup_context (io=0x7fffe8000910, c=0xae7d40) at pulse/context.c:561
#4  on_connection (client=<optimized out>, io=0x7fffe8000910, userdata=0xae7d40) at pulse/context.c:890
#5  0x00007ffff5a440e6 in do_call (c=0x9976c0) at pulsecore/socket-client.c:161
#6  0x00007ffff5e96ba7 in dispatch_defer (m=0x9ce180) at pulse/mainloop.c:704
#7  pa_mainloop_dispatch (m=0x9ce180) at pulse/mainloop.c:920
#8  0x00007ffff5e96de5 in pa_mainloop_iterate (m=0x9ce180, block=<optimized out>, retval=0x0) at pulse/mainloop.c:960
#9  0x00007ffff5e96e90 in pa_mainloop_run (m=0x9ce180, retval=0x0) at pulse/mainloop.c:975
#10 0x00007ffff5ea530f in thread (userdata=0x9a64a0) at pulse/thread-mainloop.c:88
#11 0x00007ffff5a4ed18 in internal_thread_func (userdata=0xa66840) at pulsecore/thread-posix.c:83
#12 0x00007ffff7bc4e9a in start_thread () from /lib/x86_64-linux-gnu/libpthread.so.0
#13 0x00007ffff65844bd in clone () from /lib/x86_64-linux-gnu/libc.so.6
#14 0x0000000000000000 in ?? ()
share|improve this question
    
Works for me here with Python 2.7 and 3.2. Since you've got a backtrace, might be interesting to go one step up the stack ('up' in gdb) and see what's actually going wrong ('list' to see where the code is at the time). –  user59264 May 1 '12 at 18:02
    
The code works for you? This probably means that something is wrong with my local installation. Can you tell me which version of Ubuntu are you running? –  BigWhale May 1 '12 at 18:09
    
I tried this on a fresh ubuntu 12.04 install with the python3.2-minimal package and I get a segfault in python3.2mu... –  Smotko May 1 '12 at 18:54
add comment

1 Answer

up vote 1 down vote accepted

Just a follow up. This problem was resolved. Using Python 3.2.3 in Precise Pangolin the posted code works as expected.

share|improve this answer
add comment

Your Answer

 
discard

By posting your answer, you agree to the privacy policy and terms of service.

Not the answer you're looking for? Browse other questions tagged or ask your own question.