asyncdbus

这是python-dbus-next的分支。它底层使用anyio，因此也与Trio和asyncio兼容。

已移除对GLib的支持。如果需要，请在访客模式下使用Trio。

此库可在PyPi上作为asyncdbus找到。

此文档的其余部分来自dbus-next，未经修改。

python-dbus-next

Python下一个伟大的DBus库。

文档

聊天

python-dbus-next是一个Python DBus库，旨在成为一个功能齐全的高级库，主要面向Linux桌面和移动环境的集成。

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

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

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

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

安装

本库可在PyPi上找到，名称为dbus-next。

pip3 install dbus-next

客户端接口

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

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

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

from asyncdbus import MessageBus

import anyio


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 = await bus.get_proxy_object('org.mpris.MediaPlayer2.vlc', '/org/mpris/MediaPlayer2', introspection)
    player = await obj.get_interface('org.mpris.MediaPlayer2.Player')
    properties = await 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 anyio.sleep(99999)

anyio.run(main)

服务接口

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

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

from asyncdbus.service import ServiceInterface, method, dbus_property, signal, Variant
from asyncdbus import MessageBus

import anyio

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')
    await 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 anyio.sleep(99999)

anyio.run(main)

低级接口

低级接口直接与DBus消息一起工作。

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

from asyncdbus.message import Message, MessageType
from asyncdbus import MessageBus

import anyio
import json


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


anyio.run(main)

使用python-dbus-next的项目

• Playerctl测试套件
• i3-dstatus

贡献

欢迎贡献。开发在Github上进行。

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

版权

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

© 2019，Tony Crisci