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         if t is str:
  52                 return "%s" % d
  53
  54         return str(d)
  55
  56 def property_changed(name, value, path, interface):
  57         iface = interface[interface.rfind(".") + 1:]
  58         print "{%s} [%s] %s = %s" % (iface, path, name, pretty(value))
  59
  60 def added(name, value, member, path, interface):
  61         iface = interface[interface.rfind(".") + 1:]
  62         print "{%s} [%s] %s %s" % (iface, member, name, pretty(value))
  63
  64 def removed(name, member, path, interface):
  65         iface = interface[interface.rfind(".") + 1:]
  66         print "{%s} [%s] %s" % (iface, name, member)
  67
  68 def event(member, path, interface):
  69         iface = interface[interface.rfind(".") + 1:]
  70         print "{%s} [%s] %s" % (iface, path, member)
  71
  72 def message(msg, args, member, path, interface):
  73         iface = interface[interface.rfind(".") + 1:]
  74         print "{%s} [%s] %s %s (%s)" % (iface, path, member,
  75                                         msg, pretty(args))
  76
  77 def ussd(msg, member, path, interface):
  78         iface = interface[interface.rfind(".") + 1:]
  79         print "{%s} [%s] %s %s" % (iface, path, member, msg)
  80
  81 def value(value, member, path, interface):
  82         iface = interface[interface.rfind(".") + 1:]
  83         print "{%s} [%s] %s %s" % (iface, path, member, str(value))
  84
  85 if __name__ == '__main__':
  86         dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
  87
  88         bus = dbus.SystemBus()
  89
  90         bus.add_signal_receiver(property_changed,
  91                                 bus_name="org.ofono",
  92                                 signal_name = "PropertyChanged",
  93                                 path_keyword="path",
  94                                 interface_keyword="interface")
  95
  96         for member in ["IncomingBarringInEffect",
  97                         "OutgoingBarringInEffect",
  98                         "NearMaximumWarning"]:
  99                 bus.add_signal_receiver(event,
 100                                         bus_name="org.ofono",
 101                                         signal_name = member,
 102                                         member_keyword="member",
 103                                         path_keyword="path",
 104                                         interface_keyword="interface")
 105
 106         for member in ["ModemAdded",
 107                         "ContextAdded",
 108                         "CallAdded",
 109                         "MessageAdded"]:
 110                 bus.add_signal_receiver(added,
 111                                         bus_name="org.ofono",
 112                                         signal_name = member,
 113                                         member_keyword="member",
 114                                         path_keyword="path",
 115                                         interface_keyword="interface")
 116
 117         for member in ["ModemRemoved",
 118                         "ContextRemoved",
 119                         "CallRemoved",
 120                         "MessageRemoved"]:
 121                 bus.add_signal_receiver(removed,
 122                                         bus_name="org.ofono",
 123                                         signal_name = member,
 124                                         member_keyword="member",
 125                                         path_keyword="path",
 126                                         interface_keyword="interface")
 127
 128         for member in ["DisconnectReason", "Forwarded", "BarringActive"]:
 129                 bus.add_signal_receiver(value,
 130                                         bus_name="org.ofono",
 131                                         signal_name = member,
 132                                         member_keyword="member",
 133                                         path_keyword="path",
 134                                         interface_keyword="interface")
 135
 136         for member in ["IncomingBroadcast", "EmergencyBroadcast",
 137                         "IncomingMessage", "ImmediateMessage"]:
 138                 bus.add_signal_receiver(message,
 139                                         bus_name="org.ofono",
 140                                         signal_name = member,
 141                                         member_keyword="member",
 142                                         path_keyword="path",
 143                                         interface_keyword="interface")
 144
 145         for member in ["NotificationReceived", "RequestReceived"]:
 146                 bus.add_signal_receiver(ussd,
 147                                         bus_name="org.ofono",
 148                                         signal_name = member,
 149                                         member_keyword="member",
 150                                         path_keyword="path",
 151                                         interface_keyword="interface")
 152
 153         mainloop = gobject.MainLoop()
 154         mainloop.run()