test/test-ussd

   1 #!/usr/bin/python
   2
   3 import sys
   4 import gobject
   5 import os
   6
   7 import dbus
   8 import dbus.mainloop.glib
   9
  10 state = None
  11
  12 def ussd_notification_received(content):
  13         print("Network sent a Notification: " + content)
  14
  15 def ussd_request_received(content):
  16         print("Network sent a Request: " + content)
  17         ss.Cancel()
  18
  19 def ussd_property_changed(name, value):
  20         global state
  21         if name != "State":
  22                 return
  23         print("USSD session state is " + value)
  24         state = str(value)
  25
  26 def stdin_handler(fd, condition):
  27         s = os.read(fd.fileno(), 160).rstrip()
  28         if not s:
  29                 ss.Cancel()
  30         elif state == "user-response":
  31                 print ss.Respond(s, timeout = 100)
  32         elif state == "idle":
  33                 print ss.Initiate(s, timeout = 100)
  34         else:
  35                 print "Invalid state", state
  36         return True
  37
  38 if __name__ == "__main__":
  39         if (len(sys.argv) < 2):
  40                 print "Usage: %s [modem] <ussd-string>" % (sys.argv[0])
  41                 sys.exit(1)
  42
  43         dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
  44
  45         bus = dbus.SystemBus()
  46
  47         manager = dbus.Interface(bus.get_object('org.ofono', '/'),
  48                                                         'org.ofono.Manager')
  49
  50         modems = manager.GetModems()
  51         modem = modems[0][0]
  52
  53         if (len(sys.argv) == 2):
  54                 ussd = sys.argv[1]
  55         else:
  56                 modem = sys.argv[1]
  57                 ussd = sys.argv[2]
  58
  59         ss = dbus.Interface(bus.get_object('org.ofono', modem),
  60                                         'org.ofono.SupplementaryServices')
  61
  62         props = ss.GetProperties()
  63         for p in props:
  64                 ussd_property_changed(p, props[p])
  65
  66         ss.connect_to_signal("NotificationReceived", ussd_notification_received)
  67         ss.connect_to_signal("RequestReceived", ussd_request_received)
  68         ss.connect_to_signal("PropertyChanged", ussd_property_changed)
  69
  70         print ss.Initiate(ussd, timeout=100)
  71
  72         gobject.io_add_watch(sys.stdin, gobject.IO_IN, stdin_handler)
  73
  74         mainloop = gobject.MainLoop()
  75         mainloop.run()