The [[http://www.freedesktop.org/wiki/Software/dbus|D-Bus library]] is a messaging library used by various desktop environments (GNOME, KDE, etc) for interprocess communication. There are multiple Python bindings for DBus: * GDbus and QtDbus are wrappers over the C/C++ APIs of GLib and Qt * [[https://github.com/LEW21/pydbus|pydbus]] is a modern, pythonic API with the goal of being as invisible a layer between the exported API and its user as possible * [[https://github.com/rhinstaller/dasbus|dasbus]] is a Python3-only alternative to pydbus with additional features and better flexibility * [[https://dbus.freedesktop.org/doc/dbus-python/|dbus-python]] is a legacy API, built with a deprecated dbus-glib library, and involving a lot of type-guessing (despite "explicit is better than implicit" and "resist the temptation to guess"). * [[https://pypi.python.org/pypi/txdbus|txdbus]] is a native Python implementation of the D-Bus protocol for the Twisted networking framework. * [[https://pypi.org/project/dbus-next/|dbus-next]] is a native Python library for D-Bus with support for multiple IO backends suitable for GUI develeopment and cross platform projects. * [[https://github.com/python-sdbus/python-sdbus|python-sdbus]] modern D-Bus library supporting both asyncio and blocking calls, unified client-server classes and type hints. See also: [[https://www.freedesktop.org/wiki/Software/DBusBindings/#python|DBusBindings]] on Freedesktop wiki. The `dbus-viewer` and `qdbusviewer` programs let you browse through the services and interfaces available on your system. = pydbus = For more information see [[https://github.com/LEW21/pydbus/blob/master/README.rst|pydbus's Readme]]. == Introspection == {{{ from pydbus import SessionBus bus = SessionBus() # Create an object that will proxy for a particular remote object. remote_object = bus.get( "org.freedesktop.DBus", # Bus name "/org/freedesktop/DBus" # Object path ) # Introspection returns an XML document containing information # about the methods supported by an interface. print("Introspection data:\n") print(remote_object.Introspect()) }}} Output: {{{ Introspection data: ... }}} == Calling an interface method == After executing the introspection example: {{{ print(remote_object.ListNames()) }}} Output: {{{ ['org.freedesktop.DBus', 'org.freedesktop.Notifications', 'org.freedesktop.PowerManagement', ':1.8', ':1.9', 'org.kde.kaccess', 'org.kde.kded', 'org.kde.StatusNotifierItem-655-1', 'org.freedesktop.systemd1', 'org.ktorrent.ktorrent', 'org.kde.StatusNotifierItem-656-1', 'org.kde.konversation', 'org.pulseaudio.Server', 'org.kde.KScreen', 'org.kde.krunner', 'org.kde.konsole', ':1.40', 'org.a11y.Bus', ':1.41', ':1.42', ':1.20', ':1.43', 'org.kde.klauncher5', ':1.21', ':1.23', 'org.kde.dolphin-3012', 'org.freedesktop.PowerManagement.Inhibit', 'org.kde.Solid.PowerManagement', ':1.24', ':1.25', ':1.49', 'org.kde.kmix', 'org.kde.screensaver', 'org.kde.KWin', 'org.bluez.obex', ':1.29', 'ca.desrt.dconf', 'org.kde.kgpg', 'org.freedesktop.ScreenSaver', 'org.kde.plasmashell', 'org.kde.plasmanetworkmanagement', 'org.kde.StatusNotifierItem-666-1', 'org.kde.kglobalaccel', 'org.freedesktop.FileManager1', 'org.kde.kwalletd5', 'org.PulseAudio1', 'org.kde.polkit-kde-authentication-agent-1', ':1.93', 'org.kde.kded5', 'org.kde.ActivityManager', 'org.kde.keyboard', 'org.kde.kate-3030', ':1.31', 'org.kde.kuiserver', ':1.32', ':1.55', ':1.33', ':1.11', 'org.kde.kwin.Screenshot', ':1.56', ':1.34', 'org.kde.StatusNotifierWatcher', 'org.kde.JobViewServer', ':1.35', ':1.0', ':1.13', ':1.58', 'org.kde.StatusNotifierHost-616', ':1.14', ':1.59', ':1.15', ':1.38', ':1.16', 'org.kde.ksmserver', ':1.39', ':1.17', ':1.5', 'org.kde.Solid.PowerManagement.PolicyAgent', ':1.18', 'org.kde.klauncher', ':1.6', ':1.19'] }}} The following example makes your system hibernate: {{{ # Get the power management object power = bus.get('org.gnome.PowerManager', '/org/gnome/PowerManager') # Hibernate the system if power.CanHibernate(): power.Hibernate() }}} = dasbus = For more information see [[https://dasbus.readthedocs.io/|dasbus's documentation]]. == Introspection of a remote object == Introspection returns an XML string containing information about interfaces, methods, properties and signals of the remote object. {{{ from dasbus.connection import SessionMessageBus bus = SessionMessageBus() # Create an object that will be a proxy for a particular remote object. remote_object = bus.get_proxy( "org.freedesktop.DBus", # The bus name "/org/freedesktop/DBus" # The object path ) # Call the Introspect method of the remote object. print(remote_object.Introspect()) }}} == Accessing a remote property == The following example prints the current hostname. {{{ from dasbus.connection import SystemMessageBus bus = SystemMessageBus() proxy = bus.get_proxy( "org.freedesktop.hostname1", "/org/freedesktop/hostname1" ) print(proxy.Hostname) }}} == Calling a remote method == The following example sends a notification to the notification server. {{{ from dasbus.connection import SessionMessageBus bus = SessionMessageBus() proxy = bus.get_proxy( "org.freedesktop.Notifications", "/org/freedesktop/Notifications" ) id = proxy.Notify( "", 0, "face-smile", "My notification", "Hello World!", [], {}, 0 ) print("The notification {} was sent.".format(id)) }}} = dbus-next = For more information see [[https://python-dbus-next.readthedocs.io/|dbus-next's documentation]]. == The client interface == To use a service on the bus, the library constructs a proxy object you can use to call methods, get and set properties, and listen to signals. This example connects to a media player and controls it with the MPRIS DBus interface using python's asyncio backend. {{{#!highlight python from dbus_next.aio import MessageBus import asyncio loop = asyncio.get_event_loop() async def main(): bus = await MessageBus().connect() # the introspection xml would normally be included in your project, but # this is convenient for development introspection = await bus.introspect('org.mpris.MediaPlayer2.vlc', '/org/mpris/MediaPlayer2') obj = bus.get_proxy_object('org.mpris.MediaPlayer2.vlc', '/org/mpris/MediaPlayer2', introspection) player = obj.get_interface('org.mpris.MediaPlayer2.Player') properties = obj.get_interface('org.freedesktop.DBus.Properties') # call methods on the interface (this causes the media player to play) await player.call_play() volume = await player.get_volume() print(f'current volume: {volume}, setting to 0.5') await player.set_volume(0.5) # listen to signals def on_properties_changed(interface_name, changed_properties, invalidated_properties): for changed, variant in changed_properties.items(): print(f'property changed: {changed} - {variant.value}') properties.on_properties_changed(on_properties_changed) await loop.create_future() loop.run_until_complete(main()) }}} == The service interface == To define a service on the bus, use the ServiceInterface class and decorate class methods to specify DBus methods, properties, and signals with their type signatures. {{{#!highlight python from dbus_next.service import ServiceInterface, method, dbus_property, signal, Variant from dbus_next.aio MessageBus import asyncio class ExampleInterface(ServiceInterface): def __init__(self, name): super().__init__(name) self._string_prop = 'kevin' @method() def Echo(self, what: 's') -> 's': return what @method() def GetVariantDict() -> 'a{sv}': return { 'foo': Variant('s', 'bar'), 'bat': Variant('x', -55), 'a_list': Variant('as', ['hello', 'world']) } @dbus_property() def string_prop(self) -> 's': return self._string_prop @string_prop.setter def string_prop_setter(self, val: 's'): self._string_prop = val @signal() def signal_simple(self) -> 's': return 'hello' async def main(): bus = await MessageBus().connect() interface = ExampleInterface('test.interface') bus.export('/test/path', interface) # now that we are ready to handle requests, we can request name from D-Bus await bus.request_name('test.name') # wait indefinitely await asyncio.get_event_loop().create_future() asyncio.get_event_loop().run_until_complete(main()) }}} = python-sdbus = See [[https://python-sdbus.readthedocs.io/en/latest/index.html|python-sdbus documentation for API reference and quickstart guides]]. Python-sdbus has a unified client and server classes where one class can be used both as a client proxy and as a server. == Interface class definition == {{{#!highlight python from sdbus import (DbusInterfaceCommonAsync, dbus_method_async, dbus_property_async, dbus_signal_async) # This is file only contains interface definition for easy import # in server and client files class ExampleInterface( DbusInterfaceCommonAsync, interface_name='org.example.interface' ): @dbus_method_async( input_signature='s', result_signature='s', ) async def upper(self, string: str) -> str: return string.upper() @dbus_property_async( property_signature='s', ) def hello_world(self) -> str: return 'Hello, World!' @dbus_signal_async( signal_signature='i' ) def clock(self) -> int: raise NotImplementedError }}} == Server == {{{#!highlight python from asyncio import new_event_loop, sleep from random import randint from time import time from example_interface import ExampleInterface from sdbus import request_default_bus_name_async loop = new_event_loop() export_object = ExampleInterface() async def clock() -> None: """ This coroutine will sleep a random time and emit a signal with current clock """ while True: await sleep(randint(2, 7)) # Sleep a random time current_time = int(time()) # The interface we defined uses integers export_object.clock.emit(current_time) async def startup() -> None: """Perform async startup actions""" # Acquire a known name on the bus # Clients will use that name to address this server await request_default_bus_name_async('org.example.test') # Export the object to D-Bus export_object.export_to_dbus('/') loop.run_until_complete(startup()) task_clock = loop.create_task(clock()) loop.run_forever() }}} == Client == {{{#!highlight python from asyncio import new_event_loop from example_interface import ExampleInterface # Create a new proxied object example_object = ExampleInterface.new_proxy('org.example.test', '/') async def print_clock() -> None: # Use async for loop to print clock signals we receive async for x in example_object.clock: print('Got clock: ', x) async def call_upper() -> None: s = 'test string' s_after = await example_object.upper(s) print('Initial string: ', s) print('After call: ', s_after) async def get_hello_world() -> None: print('Remote property: ', await example_object.hello_world) loop = new_event_loop() # Always bind your tasks to a variable task_upper = loop.create_task(call_upper()) task_clock = loop.create_task(print_clock()) task_hello_world = loop.create_task(get_hello_world()) loop.run_forever() }}}