python-dbus-next

Python的下一个伟大的DBus库。

文档

聊天

python-dbus-next是一个Python DBus库，旨在成为一个功能齐全的高级库，主要用于将应用程序集成到Linux桌面和移动环境中。

桌面应用程序开发者可以使用此库通过实现常见的DBus标准接口或创建自定义插件接口来将其应用程序集成到桌面环境中。

桌面用户可以使用此库创建自己的脚本和实用程序，以与这些接口交互，从而定制其桌面环境。

python-dbus-next计划通过以下方式改进其他Python DBus库

• 无依赖项，纯Python 3。
• 支持多个IO后端，包括asyncio和GLib主循环。
• 非阻塞IO，适用于GUI开发。
• 针对Python的最新语言特性，提供美观的服务和客户端。
• 无需猜测类型的DBus类型系统完整实现。
• 库所有功能的集成测试。
• 完全记录的公共API。

安装

此库在PyPi上作为dbus-next提供。

pip3 install dbus-next

客户端接口

要在总线上使用服务，库将构造一个代理对象，您可以使用它来调用方法、获取和设置属性，以及监听信号。

了解更多信息，请参阅高级客户端概述。

此示例连接到媒体播放器，并使用MPRIS DBus接口来控制它。

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())

服务接口

要在总线上定义服务，请使用ServiceInterface类，并使用装饰器对类方法进行装饰，以指定DBus方法、属性和信号及其类型签名。

了解更多信息，请参阅高级服务概述。

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())

低级接口

低级接口直接与DBus消息交互。

了解更多信息，请参阅低级接口概述。

from dbus_next.message import Message, MessageType
from dbus_next.aio import MessageBus

import asyncio
import json

loop = asyncio.get_event_loop()


async def main():
    bus = await MessageBus().connect()

    reply = await bus.call(
        Message(destination='org.freedesktop.DBus',
                path='/org/freedesktop/DBus',
                interface='org.freedesktop.DBus',
                member='ListNames'))

    if reply.message_type == MessageType.ERROR:
        raise Exception(reply.body[0])

    print(json.dumps(reply.body[0], indent=2))


loop.run_until_complete(main())

使用python-dbus-next的项目

• Playerctl 测试套件
• i3-dstatus

贡献

欢迎贡献力量。开发在Github上。

在提交之前，运行make以运行代码检查器、代码格式化器和测试套件。

版权

您可以在MIT许可下使用此代码（请参阅LICENSE）。

© 2019，Tony Crisci