test/opp-client

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