[Qiita][python 3] DBus ことはじめ

README.md

dbus sample

client.py

#!/usr/bin/env python3
import argparse

import dbus
import dbus.service
from dbus.mainloop.glib import DBusGMainLoop
from gi.repository import GObject, GLib


iface = 'jp.co.itage.fooiface'


class FooSession(dbus.service.Object):
    def __init__(self):
        service = 'jp.co.itage'
        obj = '/jp/co/itage/foo'

        bus = dbus.SessionBus()
        bus_name = dbus.service.BusName(service, bus=bus)
        super(FooSession, self).__init__(bus_name, obj)

        self._proxy = bus.get_object(service, obj)

        bus.add_signal_receiver(
            self.finish, dbus_interface=iface, bus_name=service, path=obj)

    def start(self, area):
        print('call start(): {}'.format(area))
        ret = self._proxy.start(area, dbus_interface=iface)
        print('start ret value: {}'.format(ret))

    def stop(self):
        print('call stop()')
        self._proxy.stop(dbus_interface=iface)

    def finish(self, result):
        print('called finish(): {}'.format(result))


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--area',
                        required=True,
                        type=int)
    args = parser.parse_args()

    DBusGMainLoop(set_as_default=True)
    loop = GObject.MainLoop()
    session = FooSession()

    try:
        session.start(args.area)
        loop.run()
    except KeyboardInterrupt:
        session.stop()
        loop.quit()

requirements.txt

dbus-python==1.2.8
pkg-resources==0.0.0
pycairo==1.17.1
PyGObject==3.30.1

server.py

#!/usr/bin/env python3

import dbus
import dbus.service
from dbus.mainloop.glib import DBusGMainLoop
from gi.repository import GObject, GLib


iface = 'jp.co.itage.fooiface'


class FooServer(dbus.service.Object):
    def __init__(self):
        service = 'jp.co.itage'
        obj = '/jp/co/itage/foo'
        bus = dbus.SessionBus()
        bus_name = dbus.service.BusName(service, bus=bus)
        super(FooServer, self).__init__(bus_name, obj)

    @dbus.service.method(iface, in_signature='u', out_signature='u')
    def start(self, area):
        """Register [start method]"""

        print(area)
        self.finish(0)

        if area == 1:
            return 1
        elif area == 2:
            return 2

        return 0

    @dbus.service.method(iface, signature=None)
    def stop(self):
        """Register [stop method]"""

        print('called stop')

    @dbus.service.signal(iface, signature='u')
    def finish(self, result):
        """Register [finish signal]"""

        print('called xfinish: {}'.format(result))


if __name__ == '__main__':
    DBusGMainLoop(set_as_default=True)
    loop = GObject.MainLoop()
    foo = FooServer()

    try:
        loop.run()
    except KeyboardInterrupt:
        loop.quit()