test/monitor-ofono

   1 #!/usr/bin/python
   2
   3 import gobject
   4
   5 import dbus
   6 import dbus.mainloop.glib
   7
   8 _dbus2py = {
   9         dbus.String : unicode,
  10         dbus.UInt32 : int,
  11         dbus.Int32 : int,
  12         dbus.Int16 : int,
  13         dbus.UInt16 : int,
  14         dbus.UInt64 : int,
  15         dbus.Int64 : int,
  16         dbus.Byte : int,
  17         dbus.Boolean : bool,
  18         dbus.ByteArray : str,
  19         dbus.ObjectPath : str
  20     }
  21
  22 def dbus2py(d):
  23         t = type(d)
  24         if t in _dbus2py:
  25                 return _dbus2py[t](d)
  26         if t is dbus.Dictionary:
  27                 return dict([(dbus2py(k), dbus2py(v)) for k, v in d.items()])
  28         if t is dbus.Array and d.signature == "y":
  29                 return "".join([chr(b) for b in d])
  30         if t is dbus.Array or t is list:
  31                 return [dbus2py(v) for v in d]
  32         if t is dbus.Struct or t is tuple:
  33                 return tuple([dbus2py(v) for v in d])
  34         return d
  35
  36 def pretty(d):
  37         d = dbus2py(d)
  38         t = type(d)
  39
  40         if t in (dict, tuple, list) and len(d) > 0:
  41                 if t is dict:
  42                         d = ", ".join(["%s = %s" % (k, pretty(v))
  43                                         for k, v in d.items()])
  44                         return "{ %s }" % d
  45
  46                 d = " ".join([pretty(e) for e in d])
  47
  48                 if t is tuple:
  49                         return "( %s )" % d
  50
  51         return str(d)
  52
  53 def property_changed(name, value, path, interface):
  54         iface = interface[interface.rfind(".") + 1:]
  55         print "{%s} [%s] %s = %s" % (iface, path, name, pretty(value))
  56
  57 def added(name, value, member, path, interface):
  58         iface = interface[interface.rfind(".") + 1:]
  59         print "{%s} [%s] %s %s" % (iface, member, name, pretty(value))
  60
  61 def removed(name, member, path, interface):
  62         iface = interface[interface.rfind(".") + 1:]
  63         print "{%s} [%s] %s" % (iface, name, member)
  64
  65 def event(member, path, interface):
  66         iface = interface[interface.rfind(".") + 1:]
  67         print "{%s} [%s] %s" % (iface, path, member)
  68
  69 def message(msg, args, member, path, interface):
  70         iface = interface[interface.rfind(".") + 1:]
  71         print "{%s} [%s] %s %s (%s)" % (iface, path, member,
  72                                         msg, pretty(args))
  73
  74 def ussd(msg, member, path, interface):
  75         iface = interface[interface.rfind(".") + 1:]
  76         print "{%s} [%s] %s %s" % (iface, path, member, msg)
  77
  78 def value(value, member, path, interface):
  79         iface = interface[interface.rfind(".") + 1:]
  80         print "{%s} [%s] %s %s" % (iface, path, member, str(value))
  81
  82 if __name__ == '__main__':
  83         dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
  84
  85         bus = dbus.SystemBus()
  86
  87         bus.add_signal_receiver(property_changed,
  88                                 bus_name="org.ofono",
  89                                 signal_name = "PropertyChanged",
  90                                 path_keyword="path",
  91                                 interface_keyword="interface")
  92
  93         for member in ["IncomingBarringInEffect",
  94                         "OutgoingBarringInEffect",
  95                         "NearMaximumWarning"]:
  96                 bus.add_signal_receiver(event,
  97                                         bus_name="org.ofono",
  98                                         signal_name = member,
  99                                         member_keyword="member",
 100                                         path_keyword="path",
 101                                         interface_keyword="interface")
 102
 103         for member in ["ModemAdded",
 104                         "ContextAdded",
 105                         "CallAdded",
 106                         "MessageAdded"]:
 107                 bus.add_signal_receiver(added,
 108                                         bus_name="org.ofono",
 109                                         signal_name = member,
 110                                         member_keyword="member",
 111                                         path_keyword="path",
 112                                         interface_keyword="interface")
 113
 114         for member in ["ModemRemoved",
 115                         "ContextRemoved",
 116                         "CallRemoved",
 117                         "MessageRemoved"]:
 118                 bus.add_signal_receiver(removed,
 119                                         bus_name="org.ofono",
 120                                         signal_name = member,
 121                                         member_keyword="member",
 122                                         path_keyword="path",
 123                                         interface_keyword="interface")
 124
 125         for member in ["DisconnectReason", "Forwarded", "BarringActive"]:
 126                 bus.add_signal_receiver(value,
 127                                         bus_name="org.ofono",
 128                                         signal_name = member,
 129                                         member_keyword="member",
 130                                         path_keyword="path",
 131                                         interface_keyword="interface")
 132
 133         for member in ["IncomingBroadcast", "EmergencyBroadcast",
 134                         "IncomingMessage", "ImmediateMessage"]:
 135                 bus.add_signal_receiver(message,
 136                                         bus_name="org.ofono",
 137                                         signal_name = member,
 138                                         member_keyword="member",
 139                                         path_keyword="path",
 140                                         interface_keyword="interface")
 141
 142         for member in ["NotificationReceived", "RequestReceived"]:
 143                 bus.add_signal_receiver(ussd,
 144                                         bus_name="org.ofono",
 145                                         signal_name = member,
 146                                         member_keyword="member",
 147                                         path_keyword="path",
 148                                         interface_keyword="interface")
 149
 150         mainloop = gobject.MainLoop()
 151         mainloop.run()