test/opp-client

   1 #!/usr/bin/python
   2
   3 import sys
   4 import dbus
   5 import gobject
   6 import dbus.mainloop.glib
   7 import os.path
   8 from optparse import OptionParser
   9
  10 def parse_options():
  11         parser.add_option("-d", "--device", dest="device",
  12                         help="Device to connect", metavar="DEVICE")
  13         parser.add_option("-p", "--pull", dest="pull_to_file",
  14                         help="Pull vcard and store in FILE", metavar="FILE")
  15         parser.add_option("-s", "--send", dest="send_file",
  16                         help="Send FILE", metavar="FILE")
  17         parser.add_option("-v", "--verbose", action="store_true",
  18                         dest="verbose")
  19
  20         return parser.parse_args()
  21
  22 class OppClient:
  23         def __init__(self, session_path, verbose=False):
  24                 self.progress = 0
  25                 self.transfer_path = None
  26                 self.verbose = verbose
  27                 bus = dbus.SessionBus()
  28                 obj = bus.get_object("org.bluez.obex.client", session_path)
  29                 self.session = dbus.Interface(obj, "org.bluez.obex.Session")
  30                 self.opp = dbus.Interface(obj, "org.bluez.obex.ObjectPush")
  31                 bus.add_signal_receiver(self.transfer_complete,
  32                                 dbus_interface="org.bluez.obex.Transfer",
  33                                 signal_name="Complete",
  34                                 path_keyword="path")
  35                 bus.add_signal_receiver(self.transfer_error,
  36                                 dbus_interface="org.bluez.obex.Transfer",
  37                                 signal_name="Error",
  38                                 path_keyword="path")
  39                 if self.verbose:
  40                         bus.add_signal_receiver(self.transfer_progress,
  41                                 dbus_interface="org.bluez.obex.Transfer",
  42                                 signal_name="PropertyChanged",
  43                                 path_keyword="path")
  44
  45         def create_transfer_reply(self, reply):
  46                 (path, properties) = reply
  47                 self.transfer_path = path
  48                 self.transfer_size = properties["Size"]
  49                 if self.verbose:
  50                         print "Transfer created: %s" % path
  51
  52         def error(self, err):
  53                 print err
  54                 mainloop.quit()
  55
  56         def transfer_complete(self, path):
  57                 if path != self.transfer_path:
  58                         return
  59                 if self.verbose:
  60                         print "Transfer finished"
  61                 mainloop.quit()
  62
  63         def transfer_error(self, code, message, path):
  64                 if path != self.transfer_path:
  65                         return
  66                 print "Transfer finished with error %s: %s" % (code, message)
  67                 mainloop.quit()
  68
  69         def transfer_progress(self, prop, value, path):
  70                 if path != self.transfer_path:
  71                         return
  72
  73                 if prop != "Progress":
  74                         return
  75
  76                 speed = (value - self.progress) / 1000
  77                 print "Transfer progress %d/%d at %d kBps" % (value,
  78                                                         self.transfer_size,
  79                                                         speed)
  80                 self.progress = value
  81
  82         def pull_business_card(self, filename):
  83                 self.opp.PullBusinessCard(os.path.abspath(filename),
  84                                 reply_handler=self.create_transfer_reply,
  85                                 error_handler=self.error)
  86
  87         def send_file(self, filename):
  88                 self.opp.SendFile(os.path.abspath(filename),
  89                                 reply_handler=self.create_transfer_reply,
  90                                 error_handler=self.error)
  91
  92 if  __name__ == '__main__':
  93
  94         dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
  95
  96         parser = OptionParser()
  97
  98         (options, args) = parse_options()
  99
 100         if not options.device:
 101                 parser.print_help()
 102                 sys.exit(0)
 103
 104         bus = dbus.SessionBus()
 105         mainloop = gobject.MainLoop()
 106
 107         client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"),
 108                                 "org.bluez.obex.Client")
 109
 110         print "Creating Session"
 111         path = client.CreateSession(options.device, { "Target": "OPP" })
 112
 113         opp_client = OppClient(path, options.verbose)
 114
 115         if options.pull_to_file:
 116                 opp_client.pull_business_card(options.pull_to_file)
 117
 118         if options.send_file:
 119                 opp_client.send_file(options.send_file)
 120
 121         mainloop.run()