在PulseAudio的DBus接口上获得工作信号？
在PulseAudio的DBus接口上获得工作信号？
提问于 2015-10-21 12:17:48
我正在尝试获取一个D-Bus信号处理程序，以便在PulseAudio中接收器的状态发生变化(例如变为非活动状态)时被调用。不幸的是，它没有被调用，坦率地说，我不确定为什么。
import dbus
import dbus.mainloop.glib
from gi.repository import GObject
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
bus = dbus.SessionBus()
def signal_handler(*args, **kwargs):
print('sig: ', args, kwargs)
def connect():
import os
if 'PULSE_DBUS_SERVER' in os.environ:
address = os.environ['PULSE_DBUS_SERVER']
else:
bus = dbus.SessionBus()
server_lookup = bus.get_object("org.PulseAudio1", "/org/pulseaudio/server_lookup1")
address = server_lookup.Get("org.PulseAudio.ServerLookup1", "Address", dbus_interface="org.freedesktop.DBus.Properties")
return dbus.connection.Connection(address)
conn = connect()
core = conn.get_object(object_path='/org/pulseaudio/core1')
core.connect_to_signal('StateUpdated', signal_handler)
core.ListenForSignal('org.PulseAudio.Core1.Device.StateUpdated', dbus.Array(signature='o'), dbus_interface='org.PulseAudio.Core1')
loop = GObject.MainLoop()
loop.run()回答 2
Stack Overflow用户
发布于 2015-11-18 17:47:48
试试这个，对我很管用。
import dbus
import os
from dbus.mainloop.glib import DBusGMainLoop
import gobject
def pulse_bus_address():
if 'PULSE_DBUS_SERVER' in os.environ:
address = os.environ['PULSE_DBUS_SERVER']
else:
bus = dbus.SessionBus()
server_lookup = bus.get_object("org.PulseAudio1", "/org/pulseaudio/server_lookup1")
address = server_lookup.Get("org.PulseAudio.ServerLookup1", "Address", dbus_interface="org.freedesktop.DBus.Properties")
print(address)
return address
def sig_handler(state):
print("State changed to %s" % state)
if state == 0:
print("Pulseaudio running.")
elif state == 1:
print("Pulseaudio idle.")
elif state == 2:
print("Pulseaudio suspended")
# setup the glib mainloop
DBusGMainLoop(set_as_default=True)
loop = gobject.MainLoop()
pulse_bus = dbus.connection.Connection(pulse_bus_address())
pulse_core = pulse_bus.get_object(object_path='/org/pulseaudio/core1')
pulse_core.ListenForSignal('org.PulseAudio.Core1.Device.StateUpdated', dbus.Array(signature='o'), dbus_interface='org.PulseAudio.Core1')
pulse_bus.add_signal_receiver(sig_handler, 'StateUpdated')
loop.run()需要pulseaudio的default.pa具有以下内容：
.ifexists module-dbus-protocol.so
load-module module-dbus-protocol
.endif编辑:对于那些想知道@conf-f-use关于应用程序名称的问题的人。事实证明，他们自己回答了这个问题，并在这里发布了答案：https://askubuntu.com/questions/906160/is-there-a-way-to-detect-whether-a-skype-call-is-in-progress-dbus-pulseaudio
窃取@con-f-use的一段代码并应用到我上面的代码中，我们得到了一个监视器，它可以跟踪状态，并能够告诉您正在播放的应用程序名称、艺术家、标题和名称。
Cheers @con-f-use :)
import dbus
import os
from dbus.mainloop.glib import DBusGMainLoop
import gobject
def pulse_bus_address():
if 'PULSE_DBUS_SERVER' in os.environ:
address = os.environ['PULSE_DBUS_SERVER']
else:
bus = dbus.SessionBus()
server_lookup = bus.get_object("org.PulseAudio1", "/org/pulseaudio/server_lookup1")
address = server_lookup.Get("org.PulseAudio.ServerLookup1", "Address", dbus_interface="org.freedesktop.DBus.Properties")
print(address)
return address
# convert byte array to string
def dbus2str(db):
if type(db)==dbus.Struct:
return str(tuple(dbus2str(i) for i in db))
if type(db)==dbus.Array:
return "".join([dbus2str(i) for i in db])
if type(db)==dbus.Dictionary:
return dict((dbus2str(k), dbus2str(v)) for k, v in db.items())
if type(db)==dbus.String:
return db+''
if type(db)==dbus.UInt32:
return str(db+0)
if type(db)==dbus.Byte:
return chr(db)
if type(db)==dbus.Boolean:
return db==True
if type(db)==dict:
return dict((dbus2str(k), dbus2str(v)) for k, v in db.items())
return "(%s:%s)" % (type(db), db)
def sig_handler(state):
print("State changed to %s" % state)
if state == 0:
print("Pulseaudio running.")
elif state == 1:
print("Pulseaudio idle.")
elif state == 2:
print("Pulseaudio suspended")
dbus_pstreams = (
dbus.Interface(
pulse_bus.get_object(object_path=path),
dbus_interface='org.freedesktop.DBus.Properties'
) for path in pulse_core.Get(
'org.PulseAudio.Core1',
'PlaybackStreams',
dbus_interface='org.freedesktop.DBus.Properties' )
)
pstreams = {}
for pstream in dbus_pstreams:
try:
pstreams[pstream.Get('org.PulseAudio.Core1.Stream', 'Index')] = pstream
except dbus.exceptions.DBusException:
pass
if pstreams:
for stream in pstreams.keys():
plist = pstreams[stream].Get('org.PulseAudio.Core1.Stream', 'PropertyList')
appname = dbus2str(plist.get('application.name', None))
artist = dbus2str(plist.get('media.artist', None))
title = dbus2str(plist.get('media.title', None))
name = dbus2str(plist.get('media.name', None))
print appname,artist,title,name
# setup the glib mainloop
DBusGMainLoop(set_as_default=True)
loop = gobject.MainLoop()
pulse_bus = dbus.connection.Connection(pulse_bus_address())
pulse_core = pulse_bus.get_object(object_path='/org/pulseaudio/core1')
#pulse_clients = pulse_bus.get_object(object_path='/org/pulseaudio/core1/Clients')
#print dir(pulse_clients)
pulse_core.ListenForSignal('org.PulseAudio.Core1.Device.StateUpdated', dbus.Array(signature='o'), dbus_interface='org.PulseAudio.Core1')
pulse_bus.add_signal_receiver(sig_handler, 'StateUpdated')
loop.run()Stack Overflow用户
发布于 2015-11-18 16:23:02
在我的环境中，我运行了dbus和pulseaudio，但是发现的地址不存在：
>>> import dbus
>>> import dbus.mainloop.glib
>>> from gi.repository import GObject
>>> dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
<dbus.mainloop.NativeMainLoop object at 0x7f3c98ffd4e0>
>>> bus = dbus.SessionBus()
>>> server_lookup = bus.get_object("org.PulseAudio1", "/org/pulseaudio/server_lookup1")
>>> address = server_lookup.Get("org.PulseAudio.ServerLookup1", "Address", dbus_interface="org.freedesktop.DBus.Properties")
>>> address
dbus.String('unix:path=/run/user/1000/pulse/dbus-socket', variant_level=1)
$ dbus-monitor --address 'unix:path=/run/user/1000/pulse/dbus-socket'
Failed to open connection to unix:path=/run/user/1000/pulse/dbus-socket: Failed to connect to socket /run/user/1000/pulse/dbus-socket: No such file or directory
$ ls /run/user/1000/pulse/
cli native pid我不知道我的配置是否是默认的，但看起来dbus集成就是不在那里！
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/33250864
复制相关文章
相似问题