Wprowadzenie do DBus i DBus-Python

28 October 2008 Comments

D-Bus to wysokopoziomowa usługa umożliwiająca porozumiewanie się ze sobą aplikacji. Możliwe jest porozumiewanie się aplikacje między sobą, czy też przekazywane powiadomień systemowych do aplikacji (np. podłączenie nowego urządzenia, włożenie płyty CD). Porozumiewanie następuje za pomocą komunikatów i jest asymetryczne. Za pomocą DBusa możemy korzystać z demonów takich jak networkmanager, hal, czy też z aplikacji KDE4, GNOME i innych.

Aplikacje świadczące usługi rejestrują się pod stałym identyfikatorem. Zazwyczaj identyfikatorem będzie odwrócona notacja domenowa, choć możemy też trafić na identyfikator w postaci :NUMER.NUMER. By wylistować znane D-Busowi usługi wystarczy wykonać polecenie:

dbus-send --system --dest=org.freedesktop.DBus --type=method_call --print-reply / org.freedesktop.DBus.ListNames

Efekt może być np. taki:

array [
      string "org.freedesktop.DBus"
      string ":1.102"
      string ":1.7"
      string "com.redhat.dhcp"
      string "org.freedesktop.NetworkManagerInfo"
      string ":1.8"
      string ":1.4"
      string ":1.0"
      string ":1.5"
      string "org.freedesktop.NetworkManager"
      string ":1.101"
      string "org.freedesktop.Hal"
   ]

By dowiedzieć się więcej o danej usłudze wystarczy wykonać polecenie:

dbus-send --session --dest=org.freedesktop.Hal --type=method_call --print-reply /org/freedesktop/Hal org.freedesktop.DBus.Introspectable.Introspect

Gdzie /org/freedesktop/Hal i org.freedesktop.Hal to nazwa usługi, o której chcemy uzyskać informacje (nie wszystkie usługi "obsługują" introspekcję). Usługi możemy przeglądać za pomocą dbus-inspector:

dbus-python

Pakiet ten znajdziemy praktycznie w każdej większej dystrybucji Linuksa. Podstawą korzystania z DBusa jest połączenie się z określoną magistralą:

import dbus

session_bus = dbus.SessionBus()

Magistrala systemowa domyślnie uruchamiana jest wraz ze startem systemu i zawiera usługi takie jak NetworkManager, Hal, Udev:

import dbus

system_bus = dbus.SystemBus()

Wysyłanie wiadomości

Aplikacje D-Bus mogą udostępniać obiekty dla innych aplikacji. By wykorzystać taki obiekt musisz znać:

Nazwę magistrali / identyfikator: Identyfikuje aplikację, którą chcemy użyć. Zazwyczaj operować będziemy na identyfikatora zapisanych w formie odwróconej notacji domenowej, np " org.freedesktop.NetworkManager".
Ścieżka do obiektu: Aplikacje mogą udostępniać wiele obiektów i należy określić, który obiekt chcemy wykorzystać. Ścieżka do danego obiektu przedstawiana jest podobnie jak ścieżka do pliku.

By operować na dostępnym obiekcie używamy obiektu proxy Pythona. Gdy wywołamy metodę na obiekcie proxy dbus-python wyśle wiadomość do wskazanego obiektu DBusa zwracając wszystkie dane zwrócone przez obiekt.

By uzyskać obiekt Proxy należy wykonać metodę get_object na obiekcie magistrali podając identyfikator i ścieżkę obiektu. Przykładowo dla NetworkManager:

import dbus
bus = dbus.SystemBus()
proxy = bus.get_object('org.freedesktop.NetworkManager',
                       '/org/freedesktop/NetworkManager/Devices/eth0')
#proxy to dbus.proxies.ProxyObject
print proxy

Interfejsy i metody

D-Bus używa interfejsów by udostępnić metody w przestrzeni nazw. Interfejs to grupa powiązanych metod i sygnałów. Dla przykładu każdy obiekt NetworkManager reprezentujący interfejs sieciowy implementuje interfejs org.freedesktop.NetworkManager.Devices, który posiada metody takie jak getProperties. By wywołać taką metodę - wywołaj metodę o tej samej nazwie na obiekcie Proxy przekazując jako argument nazwę interfejsu:

import dbus
import pprint

bus = dbus.SystemBus()
eth0 = bus.get_object('org.freedesktop.NetworkManager',
                      '/org/freedesktop/NetworkManager/Devices/wlan0')
props = eth0.getProperties(dbus_interface='org.freedesktop.NetworkManager.Devices')
pprint.pprint(props)

(dbus.ObjectPath('/org/freedesktop/NetworkManager/Devices/wlan0'),
 dbus.String(u'wlan0'),
 dbus.UInt32(2L),
 dbus.String(u'/org/freedesktop/Hal/devices/net_00_14_a4_25_03_44'),
 dbus.Boolean(True),
 dbus.UInt32(7L),
 dbus.String(u'192.168.1.4'),
 dbus.String(u'255.255.255.0'),
 dbus.String(u'192.168.1.255'),
 dbus.String(u'00:14:A4:25:03:33'),
 dbus.String(u'192.168.1.1'),
 dbus.String(u'192.168.1.1'),
 dbus.String(u'0.0.0.0'),
 dbus.Int32(2),
 dbus.Int32(33),
 dbus.Boolean(True),
 dbus.Int32(2),
 dbus.UInt32(5L),
 dbus.UInt32(61647L),
 dbus.String(u'/org/freedesktop/NetworkManager/Devices/wlan0/Networks/TEST'),
 dbus.Array([dbus.String(u'/org/freedesktop/NetworkManager/Devices/wlan0/Networks/TEST'), dbus.String(u'/org/freedesktop/NetworkManager/Devices/wlan0/Networks/D_20_O_20_M_20_'), dbus.String(u'/org/freedesktop/NetworkManager/Devices/wlan0/Networks/faxon')], signature=dbus.Signature('s')))

Interfejsy i metody

W odróżnieniu od Pythona D-Bus używa statycznego typowania - każda metoda posiada sygnaturę reprezentującą jej argumenty i nie przyjmie argumentów o innych typach. Dbus-python próbuje określić typy argumentów za pomocą introspekcji DBusa. Jeżeli nie uda mu się może wyrzucić wyjątek TypeError w przypadku niewłaściwych typów.

Typ Pythona	Typ DBus
dbus.Boolean	Boolean (signature 'b')
dbus.Byte	byte (signature 'y')
dbus.Int16	16-bit signed integer ('n')
dbus.Int32	32-bit signed integer ('i')
dbus.Int64	64-bit signed integer ('x')
dbus.UInt16	16-bit unsigned integer ('q')
dbus.UInt32	32-bit unsigned integer ('u')
dbus.UInt64	64-bit unsigned integer ('t')
dbus.Double	double-precision float ('d')
dbus.ObjectPath	object path ('o')
dbus.Signature	signature ('g')
dbus.String	string ('s')
dbus.UTF8String	string ('s')
bool	Boolean ('b')
int	32-bit signed integer ('i')
long	64-bit signed integer ('x')
float	double-precision float ('d')
str	string ('s')
unicode	string ('s')

D-Bus wspiera też cztery typy kontenerowe: tablice, struct, słownik i variant. Tablice reprezentowane są przez listy pythonowe lub dbus.Array. Struct reprezentowane są przez tuple.

Asynchroniczne wywoływanie metod

Asynchroniczne (nie-blokujące) wywoływanie metod umożliwia zlecenie wielu wywołań jednocześnie, jak i normalne działanie aplikacji w czasie oczekiwania na odpowiedź. By móc wysyłać asynchroniczne wywołania potrzebna jest pętla zdarzeń (event loop). dbus-python obsługuje pętlę zdarzeń Glib oraz pętlę Qt >= 4.2:

import dbus
from dbus.mainloop.glib import DBusGMainLoop

dbus_loop = DBusGMainLoop()

bus = dbus.SessionBus(mainloop=dbus_loop)

W przypadky PyQt wystarczy użyć dbus.mainloop.qt.DBusQtMainLoop zamiast dbus.mainloop.glib.DBusGMainLoop. By wykonać asynchroniczne wywołanie należy podać w metodzie Proxy dwa argumenty - reply_handler i error_handler. Metoda Proxy od razu zwróci None, lecz po pewnym czasie zostanie wywołany reply_handler z argumentami takimi jak wartości zwrócone przez metodę DBusa, lub error_handler z argumentem DBusException.

# -*- coding: utf-8 -*-
import pprint

import gobject
import dbus
from dbus.mainloop.glib import DBusGMainLoop

def handle_response(*args):
	print 'RESPONSE'
	pprint.pprint(args)

def handle_exception(*args):
	print 'EXCEPTION'
	pprint.pprint(args)

def make_calls():
	bus = dbus.SystemBus()
	eth0 = bus.get_object('org.freedesktop.NetworkManager',
			'/org/freedesktop/NetworkManager/Devices/wlan0')
	eth0.getProperties(dbus_interface='org.freedesktop.NetworkManager.Devices', reply_handler=handle_response, error_handler=handle_exception)

DBusGMainLoop(set_as_default=True)
# Wywołanie metod z opóźnieniem
gobject.timeout_add(1000, make_calls)

failed = False
hello_replied = False
raise_replied = False

loop = gobject.MainLoop()
loop.run()
if failed:
   raise SystemExit("Example async client failed!")

Odbieranie sygnałów - wiadomości

By odbierać sygnały magistrala musi być podłączona do pętli zdarzeń. Sygnały będą odbierane tylko gdy pętla działa. By odpowiadać na sygnały należy użyć metody add_signal_receiver na obiekcie DBus. To określa funkcję do wywołania gdy odebrany zostanie pasujący sygnał. Argumenty to: funkcja odpowiadająca na sygnał, nazwa sygnału, interfejs DBus, nazwa magistrali, ścieżka. W dbus-inspector możemy wyszukać sygnały istniejących usług. Przykładowo w Hal możemy znaleźć sygnały dotyczące napędu CDROM. W moim laptopie wygląda to tak:

Przykładowy kod:

# -*- coding: utf-8 -*-
import pprint

import gobject
import dbus
from dbus.mainloop.glib import DBusGMainLoop

def handler_cond(*args):
	print "Odebrano Sygnał Condition"
	pprint.pprint(args)
	print

def handler_prop(*args):
	print "Odebrano Sygnał PropertyModified"
	pprint.pprint(args)
	print

def make_calls():
	bus = dbus.SystemBus()
	proxy = bus.get_object('org.freedesktop.Hal',
			'/org/freedesktop/Hal/devices/storage_model_DVDRW_SOSW_833S')
	proxy.connect_to_signal("Condition", handler_cond)
	proxy.connect_to_signal("PropertyModified", handler_prop)

DBusGMainLoop(set_as_default=True)
# Wywołanie metod z opóźnieniem
gobject.timeout_add(100, make_calls)

loop = gobject.MainLoop()
loop.run()

Jeżeli w czasie działania skryptu włożymy pustą płytę CD to odbierzemy sygnały:

Odebrano Sygnał PropertyModified
(dbus.Int32(1),
 dbus.Array([dbus.Struct((dbus.String(u'storage.removable.media_available'), dbus.Boolean(False), dbus.Boolean(False)), signature=None)], signature=dbus.Signature('(sbb)')))

Odebrano Sygnał PropertyModified
(dbus.Int32(2),
 dbus.Array([dbus.Struct((dbus.String(u'storage.cdrom.write_speeds'), dbus.Boolean(False), dbus.Boolean(False)), signature=None), dbus.Struct((dbus.String(u'storage.cdrom.read_speed'), dbus.Boolean(False), dbus.Boolean(False)), signature=None)], signature=dbus.Signature('(sbb)')))

Wysunięcie płyty podobnie:

Odebrano Sygnał PropertyModified
(dbus.Int32(1),
 dbus.Array([dbus.Struct((dbus.String(u'storage.removable.media_available'), dbus.Boolean(False), dbus.Boolean(False)), signature=None)], signature=dbus.Signature('(sbb)')))

Odebrano Sygnał PropertyModified
(dbus.Int32(2),
 dbus.Array([dbus.Struct((dbus.String(u'storage.cdrom.write_speeds'), dbus.Boolean(False), dbus.Boolean(False)), signature=None), dbus.Struct((dbus.String(u'storage.cdrom.read_speed'), dbus.Boolean(False), dbus.Boolean(False)), signature=None)], signature=dbus.Signature('(sbb)')))

Normalna aplikacja najpierw odpytałaby się o listę urządzeń, dla których zaczęłaby nasłuchiwać sygnałów.

Udostępnianie obiektów

By udostępniać obiekty magistrala musi być podłączona do pętli zdarzeń, która musi działać. By udostępnić obiekt (klasę) wystarczy dziedziczyć dbus.service.Object, a w konstruktorze dodać argument object_path na ścieżkę usługi.

class Example(dbus.service.Object):
    def __init__(self, object_path):
        dbus.service.Object.__init__(self, dbus.SessionBus(), path)

Kolejny etap to udostępnianie metod za pomocą dekoratora dbus.service.method, np:

class Example(dbus.service.Object):
    def __init__(self, object_path):
        dbus.service.Object.__init__(self, dbus.SessionBus(), path)

    @dbus.service.method(dbus_interface='com.example.Sample',
                         in_signature='v', out_signature='s')
    def StringifyVariant(self, variant):
        return str(variant)

By udostępnić metodę jako sygnał należy użyć dekoratora dbus.service.signal:

class Example(dbus.service.Object):
    def __init__(self, object_path):
        dbus.service.Object.__init__(self, dbus.SessionBus(), path)

    @dbus.service.signal(dbus_interface='com.example.Sample',
                         signature='us')
    def NumberOfBottlesChanged(self, number, contents):
        print "%d bottles of %s on the wall" % (number, contents)

e = Example('/bottle-counter')
e.NumberOfBottlesChanged(100, 'beer')
# -> emits com.example.Sample.NumberOfBottlesChanged(100, 'beer')
#    and prints "100 bottles of beer on the wall"

Przykładowe skrypty znajdziemy także w katalogu examples paczki dbus-python.
dbus-python i interfejsy do innych języków
Katalog z paczkami dbus-python

RkBlog

Programowanie Sieciowe, 28 October 2008