dbus-next

To expose an interface on the D-Bus, you must subclass ServiceInterface and define your members using specific decorators. Once defined, you use MessageBus.export() to make the interface available at a specific object path on a connected bus.

Core Components

• ServiceInterface: The base class for all exported services. You must call super().__init__('interface.name') in your constructor.
• @method(): Defines a D-Bus method. Parameters and return values must be annotated with DBus type signature strings (e.g., 'i' for integer, 's' for string). If using aio.MessageBus, methods can be coroutines.
• @dbus_property(): Defines a D-Bus property getter. Use the standard Python @<property_name>.setter syntax to define a setter. Property getters/setters can be coroutines in aio.MessageBus.
• @signal(): Defines a D-Bus signal. The value returned by the decorated method is broadcast to listeners. If the signal has multiple arguments, return them as a list.
• DBusError: You can raise DBusError within methods or properties to send detailed error information back to the D-Bus client.

Requesting a Well-Known Name

If you want clients to find your service using a predictable name rather than a connection-specific one, call MessageBus.request_name('your.well-known.name') after connecting.

from dbus_next.aio import MessageBus
from dbus_next.service import ServiceInterface, method, dbus_property, signal

class ExampleInterface(ServiceInterface):
    def __init__(self):
        super().__init__('com.example.SampleInterface0')
        self._bar = 105

    @method()
    def Frobate(self, foo: 'i', bar: 's') -> 'a{us}':
        return {1: 'one', 2: 'two'}

    @dbus_property()
    def Bar(self) -> 'y':
        return self._bar

    @Bar.setter
    def Bar(self, val: 'y'):
        self._bar = val

    @signal()
    def Changed(self) -> 'b':
        return True

async def main():
    bus = await MessageBus().connect()
    interface = ExampleInterface()
    bus.export('/com/example/sample0', interface)
    await bus.request_name('com.example.name')
    await bus.wait_for_disconnect()

To consume a service on the DBus, use the MessageBus to connect and create a proxy object. The proxy object allows you to call methods, access properties, and listen to signals on a specific interface.

Common workflow:

1. Connect to the bus using await MessageBus().connect().
2. Introspect the object to get its structure using await bus.introspect(service_name, object_path).
3. Get a proxy object via bus.get_proxy_object(service_name, object_path, introspection).
4. Access specific interfaces using obj.get_interface(interface_name).
5. Call methods with await interface.method_name() or listen to signals using interface.on_signal_name(callback).

from dbus_next.aio import MessageBus
import asyncio

async def main():
    bus = await MessageBus().connect()
    # Introspect the service
    introspection = await bus.introspect('org.mpris.MediaPlayer2.vlc', '/org/mpris/MediaPlayer2')

    # Get proxy object and interfaces
    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 a method
    await player.call_play()

    # Listen to a signal
    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)

asyncio.run(main())

To export a service on the DBus, inherit from ServiceInterface. Use decorators to define the DBus interface members:

• @method(): Defines a DBus method. You can use Python type hints to specify DBus type signatures (e.g., what: 's' for a string).
• @dbus_property(): Defines a DBus property. Use a corresponding .setter decorator to implement property updates.
• @signal(): Defines a DBus signal.

To make the service available:

1. Instantiate your class with a name.
2. Export it to a path using bus.export(path, interface_instance).
3. Request the service name on the bus using await bus.request_name(name).

from dbus_next.service import ServiceInterface, method, dbus_property, signal, Variant
from dbus_next.aio import 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

    @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)
    await bus.request_name('test.name')
    await asyncio.get_event_loop().create_future()

asyncio.run(main())