#!/usr/bin/env python
import gobject
import time
import dbus
import dbus.service
import dbus.mainloop.glib
class TestException(dbus.DBusException):
_dbus_error_name = 'com.example.TestException'
class TestService(dbus.service.Object):
# ----------------------------------------------------------------------------------------------------
@dbus.service.method("com.example.Frob",
in_signature='s', out_signature='s')
def HelloWorld(self, hello_message):
if str(hello_message) == 'Yo':
raise TestException('Yo is not a proper greeting')
else:
return "You greeted me with '%s'. Thanks!"%(str(hello_message))
@dbus.service.method("com.example.Frob",
in_signature='ss', out_signature='ss')
def DoubleHelloWorld(self, hello1, hello2):
return ("You greeted me with '%s'. Thanks!"%(str(hello1)), "Yo dawg, you uttered '%s'. Thanks!"%(str(hello2)))
@dbus.service.method("com.example.Frob",
in_signature='', out_signature='su')
def PairReturn(self):
return ("foo", 42)
# ----------------------------------------------------------------------------------------------------
@dbus.service.method("com.example.Frob",
in_signature='ybnqiuxtdsog', out_signature='ybnqiuxtdsog')
def TestPrimitiveTypes(self, val_byte, val_boolean, val_int16, val_uint16, val_int32, val_uint32, val_int64, val_uint64, val_double, val_string, val_objpath, val_signature):
return val_byte + 1, not val_boolean, val_int16 + 1, val_uint16 + 1, val_int32 + 1, val_uint32 + 1, val_int64 + 1, val_uint64 + 1, -val_double + 0.123, val_string * 2, val_objpath + "/modified", val_signature * 2
# ----------------------------------------------------------------------------------------------------
@dbus.service.method("com.example.Frob",
in_signature='ayabanaqaiauaxatad', out_signature='ayabanaqaiauaxatad')
def TestArrayOfPrimitiveTypes(self, val_byte, val_boolean, val_int16, val_uint16, val_int32, val_uint32, val_int64, val_uint64, val_double):
return val_byte*2, val_boolean*2, val_int16*2, val_uint16*2, val_int32*2, val_uint32*2, val_int64*2, val_uint64*2, val_double*2
# ----------------------------------------------------------------------------------------------------
@dbus.service.method("com.example.Frob",
in_signature='asaoag', out_signature='asaoag')
def TestArrayOfStringTypes(self, val_string, val_objpath, val_signature):
return val_string * 2, val_objpath * 2, val_signature * 2
# ----------------------------------------------------------------------------------------------------
@dbus.service.method("com.example.Frob",
in_signature = 'a{yy}a{bb}a{nn}a{qq}a{ii}a{uu}a{xx}a{tt}a{dd}a{ss}a{oo}a{gg}',
out_signature = 'a{yy}a{bb}a{nn}a{qq}a{ii}a{uu}a{xx}a{tt}a{dd}a{ss}a{oo}a{gg}')
def TestHashTables(self, hyy, hbb, hnn, hqq, hii, huu, hxx, htt, hdd, hss, hoo, hgg):
ret_hyy = {}
for i in hyy:
ret_hyy[i*2] = (hyy[i]*3) & 255
ret_hbb = {}
for i in hbb:
ret_hbb[i] = True
ret_hnn = {}
for i in hnn:
ret_hnn[i*2] = hnn[i]*3
ret_hqq = {}
for i in hqq:
ret_hqq[i*2] = hqq[i]*3
ret_hii = {}
for i in hii:
ret_hii[i*2] = hii[i]*3
ret_huu = {}
for i in huu:
ret_huu[i*2] = huu[i]*3
ret_hxx = {}
for i in hxx:
ret_hxx[i + 2] = hxx[i] + 1
ret_htt = {}
for i in htt:
ret_htt[i + 2] = htt[i] + 1
ret_hdd = {}
for i in hdd:
ret_hdd[i + 2.5] = hdd[i] + 5.0
ret_hss = {}
for i in hss:
ret_hss[i + "mod"] = hss[i]*2
ret_hoo = {}
for i in hoo:
ret_hoo[i + "/mod"] = hoo[i] + "/mod2"
ret_hgg = {}
for i in hgg:
ret_hgg[i + "assgit"] = hgg[i]*2
return ret_hyy, ret_hbb, ret_hnn, ret_hqq, ret_hii, ret_huu, ret_hxx, ret_htt, ret_hdd, ret_hss, ret_hoo, ret_hgg
# ----------------------------------------------------------------------------------------------------
@dbus.service.method("com.example.Frob",
in_signature='(ii)(s(ii)aya{ss})', out_signature='(ii)(s(ii)aya{ss})')
def TestStructureTypes(self, s1, s2):
(x, y) = s1;
(desc, (x1, y1), ay, hss) = s2;
ret_hss = {}
for i in hss:
ret_hss[i] = hss[i] + " ... in bed!"
return (x + 1, y + 1), (desc + " ... in bed!", (x1 + 2, y1 + 2), ay * 2, ret_hss)
# ----------------------------------------------------------------------------------------------------
@dbus.service.method("com.example.Frob",
in_signature='vb', out_signature='v')
def TestVariant(self, v, modify):
if modify:
if type(v)==dbus.Boolean:
ret = False
elif type(v)==dbus.Dictionary:
ret = {}
for i in v:
ret[i] = v[i] * 2
elif type(v)==dbus.Struct:
ret = ["other struct", dbus.Int16(100)]
else:
ret = v * 2
else:
ret = v
return (type(v))(ret)
# ----------------------------------------------------------------------------------------------------
@dbus.service.method("com.example.Frob",
in_signature='a(ii)aa(ii)aasaa{ss}aayavaav', out_signature='a(ii)aa(ii)aasaa{ss}aayavaav')
def TestComplexArrays(self, aii, aaii, aas, ahashes, aay, av, aav):
return aii * 2, aaii * 2, aas * 2, ahashes * 2, aay * 2, av *2, aav * 2
# ----------------------------------------------------------------------------------------------------
@dbus.service.method("com.example.Frob",
in_signature='a{s(ii)}a{sv}a{sav}a{saav}a{sa(ii)}a{sa{ss}}',
out_signature='a{s(ii)}a{sv}a{sav}a{saav}a{sa(ii)}a{sa{ss}}')
def TestComplexHashTables(self, h_str_to_pair, h_str_to_variant, h_str_to_av, h_str_to_aav,
h_str_to_array_of_pairs, hash_of_hashes):
ret_h_str_to_pair = {}
for i in h_str_to_pair:
ret_h_str_to_pair[i + "_baz"] = h_str_to_pair[i]
ret_h_str_to_variant = {}
for i in h_str_to_variant:
ret_h_str_to_variant[i + "_baz"] = h_str_to_variant[i]
return ret_h_str_to_pair, ret_h_str_to_variant, h_str_to_av, h_str_to_aav, h_str_to_array_of_pairs, hash_of_hashes
# ----------------------------------------------------------------------------------------------------
@dbus.service.method("com.example.Frob",
in_signature='', out_signature='')
def Quit(self):
mainloop.quit()
# ----------------------------------------------------------------------------------------------------
@dbus.service.method("com.example.Frob",
in_signature='sv', out_signature='')
def FrobSetProperty(self, prop_name, prop_value):
self.frob_props[prop_name] = prop_value
message = dbus.lowlevel.SignalMessage("/com/example/TestObject",
"org.freedesktop.DBus.Properties",
"PropertiesChanged")
message.append("com.example.Frob")
message.append({prop_name : prop_value})
message.append([], signature="as")
session_bus.send_message(message)
# ----------------------------------------------------------------------------------------------------
@dbus.service.method("com.example.Frob",
in_signature='', out_signature='')
def FrobInvalidateProperty(self):
self.frob_props["PropertyThatWillBeInvalidated"] = "OMGInvalidated"
message = dbus.lowlevel.SignalMessage("/com/example/TestObject",
"org.freedesktop.DBus.Properties",
"PropertiesChanged")
message.append("com.example.Frob")
message.append({}, signature="a{sv}")
message.append(["PropertyThatWillBeInvalidated"])
session_bus.send_message(message)
# ----------------------------------------------------------------------------------------------------
@dbus.service.signal("com.example.Frob",
signature="sov")
def TestSignal(self, str1, objpath1, variant1):
pass
@dbus.service.method("com.example.Frob",
in_signature='so', out_signature='')
def EmitSignal(self, str1, objpath1):
self.TestSignal (str1 + " .. in bed!", objpath1 + "/in/bed", "a variant")
# ----------------------------------------------------------------------------------------------------
@dbus.service.signal("com.example.Frob",
signature="i")
def TestSignal2(self, int1):
pass
@dbus.service.method("com.example.Frob",
in_signature='', out_signature='')
def EmitSignal2(self):
self.TestSignal2 (42)
# ----------------------------------------------------------------------------------------------------
@dbus.service.method("com.example.Frob", in_signature='i', out_signature='',
async_callbacks=('return_cb', 'raise_cb'))
def Sleep(self, msec, return_cb, raise_cb):
def return_from_async_wait():
return_cb()
return False
gobject.timeout_add(msec, return_from_async_wait)
# ----------------------------------------------------------------------------------------------------
@dbus.service.method("org.freedesktop.DBus.Properties",
in_signature = 'ss',
out_signature = 'v')
def Get(self, interface_name, property_name):
if interface_name == "com.example.Frob":
return self.frob_props[property_name]
else:
raise TestException("No such interface " + interface_name)
@dbus.service.method("org.freedesktop.DBus.Properties",
in_signature = 's',
out_signature = 'a{sv}')
def GetAll(self, interface_name):
if interface_name == "com.example.Frob":
return self.frob_props
else:
raise TestException("No such interface " + interface_name)
if __name__ == '__main__':
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
session_bus = dbus.SessionBus()
name = dbus.service.BusName("com.example.TestService", session_bus)
obj = TestService(session_bus, '/com/example/TestObject')
#print "Our unique name is %s"%(session_bus.get_unique_name())
obj.frob_props = {}
obj.frob_props["y"] = dbus.Byte(1)
obj.frob_props["b"] = dbus.Boolean(True)
obj.frob_props["n"] = dbus.Int16(2)
obj.frob_props["q"] = dbus.UInt16(3)
obj.frob_props["i"] = dbus.Int32(4)
obj.frob_props["u"] = dbus.UInt32(5)
obj.frob_props["x"] = dbus.Int64(6)
obj.frob_props["t"] = dbus.UInt64(7)
obj.frob_props["d"] = dbus.Double(7.5)
obj.frob_props["s"] = dbus.String("a string")
obj.frob_props["o"] = dbus.ObjectPath("/some/path")
obj.frob_props["ay"] = [dbus.Byte(1), dbus.Byte(11)]
obj.frob_props["ab"] = [dbus.Boolean(True), dbus.Boolean(False)]
obj.frob_props["an"] = [dbus.Int16(2), dbus.Int16(12)]
obj.frob_props["aq"] = [dbus.UInt16(3), dbus.UInt16(13)]
obj.frob_props["ai"] = [dbus.Int32(4), dbus.Int32(14)]
obj.frob_props["au"] = [dbus.UInt32(5), dbus.UInt32(15)]
obj.frob_props["ax"] = [dbus.Int64(6), dbus.Int64(16)]
obj.frob_props["at"] = [dbus.UInt64(7), dbus.UInt64(17)]
obj.frob_props["ad"] = [dbus.Double(7.5), dbus.Double(17.5)]
obj.frob_props["as"] = [dbus.String("a string"), dbus.String("another string")]
obj.frob_props["ao"] = [dbus.ObjectPath("/some/path"), dbus.ObjectPath("/another/path")]
obj.frob_props["foo"] = "a frobbed string"
obj.frob_props["PropertyThatWillBeInvalidated"] = "InitialValue"
mainloop = gobject.MainLoop()
mainloop.run()