test/ftp-client

   1 #!/usr/bin/python
   2
   3 import gobject
   4
   5 import sys
   6 import dbus
   7 import dbus.service
   8 import dbus.mainloop.glib
   9 import os.path
  10 from optparse import OptionParser
  11
  12 def parse_options():
  13         parser.add_option("-d", "--device", dest="device",
  14                         help="Device to connect", metavar="DEVICE")
  15         parser.add_option("-c", "--chdir", dest="new_dir",
  16                         help="Change current directory to DIR", metavar="DIR")
  17         parser.add_option("-l", "--list", action="store_true", dest="list_dir",
  18                         help="List the current directory")
  19         parser.add_option("-g", "--get", dest="get_file",
  20                         help="Get FILE", metavar="FILE")
  21         parser.add_option("-p", "--put", dest="put_file",
  22                         help="Put FILE", metavar="FILE")
  23         parser.add_option("-y", "--copy", dest="copy_file",
  24                         help="Copy FILE", metavar="FILE")
  25         parser.add_option("-m", "--move", dest="move_file",
  26                         help="Move FILE", metavar="FILE")
  27         parser.add_option("-n", "--destname", dest="dest_file",
  28                         help="Destination FILE", metavar="FILE")
  29         parser.add_option("-r", "--remove", dest="remove_file",
  30                         help="Remove FILE", metavar="FILE")
  31         parser.add_option("-v", "--verbose", action="store_true",
  32                         dest="verbose")
  33
  34         return parser.parse_args()
  35
  36 class FtpClient:
  37         def __init__(self, session_path, verbose=False):
  38                 self.progress = 0
  39                 self.transfer_path = None
  40                 self.transfer_size = 0
  41                 self.verbose = verbose
  42                 bus = dbus.SessionBus()
  43                 obj = bus.get_object("org.bluez.obex.client", session_path)
  44                 self.session = dbus.Interface(obj, "org.bluez.obex.Session")
  45                 self.ftp = dbus.Interface(obj, "org.bluez.obex.FileTransfer")
  46                 bus.add_signal_receiver(self.transfer_complete,
  47                                 dbus_interface="org.bluez.obex.Transfer",
  48                                 signal_name="Complete",
  49                                 path_keyword="path")
  50                 bus.add_signal_receiver(self.transfer_error,
  51                                 dbus_interface="org.bluez.obex.Transfer",
  52                                 signal_name="Error",
  53                                 path_keyword="path")
  54                 if self.verbose:
  55                         bus.add_signal_receiver(self.transfer_progress,
  56                                 dbus_interface="org.bluez.obex.Transfer",
  57                                 signal_name="PropertyChanged",
  58                                 path_keyword="path")
  59
  60         def create_transfer_reply(self, reply):
  61                 (path, properties) = reply
  62                 self.transfer_path = path
  63                 self.transfer_size = properties["Size"]
  64                 if self.verbose:
  65                         print "Transfer created: %s" % path
  66
  67         def generic_reply(self):
  68                 if self.verbose:
  69                         print "Operation succeeded"
  70
  71         def error(self, err):
  72                 print err
  73                 mainloop.quit()
  74
  75         def transfer_complete(self, path):
  76                 if path != self.transfer_path:
  77                         return
  78                 if self.verbose:
  79                         print "Transfer finished"
  80                 mainloop.quit()
  81
  82         def transfer_error(self, code, message, path):
  83                 if path != self.transfer_path:
  84                         return
  85                 print "Transfer finished with error %s: %s" % (code, message)
  86                 mainloop.quit()
  87
  88         def transfer_progress(self, prop, value, path):
  89                 if path != self.transfer_path:
  90                         return
  91
  92                 if prop != "Progress":
  93                         return
  94
  95                 speed = (value - self.progress) / 1000
  96                 print "Transfer progress %d/%d at %d kBps" % (value,
  97                                                         self.transfer_size,
  98                                                         speed)
  99                 self.progress = value
 100
 101         def change_folder(self, new_dir):
 102                 for node in new_dir.split("/"):
 103                         self.ftp.ChangeFolder(node)
 104
 105         def list_folder(self):
 106                 for i in self.ftp.ListFolder():
 107                         if i["Type"] == "folder":
 108                                 print "%s/" % (i["Name"])
 109                         else:
 110                                 print "%s" % (i["Name"])
 111
 112         def put_file(self, filename):
 113                 self.ftp.PutFile(os.path.abspath(filename),
 114                                 os.path.basename(filename),
 115                                 reply_handler=self.create_transfer_reply,
 116                                 error_handler=self.error)
 117
 118         def get_file(self, filename):
 119                 self.ftp.GetFile(os.path.abspath(filename),
 120                                 os.path.basename(filename),
 121                                 reply_handler=self.create_transfer_reply,
 122                                 error_handler=self.error)
 123
 124         def remove_file(self, filename):
 125                 self.ftp.Delete(filename,
 126                                 reply_handler=self.generic_reply,
 127                                 error_handler=self.error)
 128
 129         def move_file(self, filename, destname):
 130                 self.ftp.MoveFile(filename, destname,
 131                                 reply_handler=self.generic_reply,
 132                                 error_handler=self.error)
 133
 134         def copy_file(self, filename, destname):
 135                 self.ftp.CopyFile(filename, destname,
 136                                 reply_handler=self.generic_reply,
 137                                 error_handler=self.error)
 138
 139 if  __name__ == '__main__':
 140
 141         dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
 142
 143         parser = OptionParser()
 144
 145         (options, args) = parse_options()
 146
 147         if not options.device:
 148                 parser.print_help()
 149                 sys.exit(0)
 150
 151         bus = dbus.SessionBus()
 152         mainloop = gobject.MainLoop()
 153
 154         client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"),
 155                                 "org.bluez.obex.Client")
 156
 157         print "Creating Session"
 158         path = client.CreateSession(options.device, { "Target": "ftp" })
 159
 160         ftp_client = FtpClient(path, options.verbose)
 161
 162         if options.new_dir:
 163                 ftp_client.change_folder(options.new_dir)
 164
 165         if options.list_dir:
 166                 ftp_client.list_folder()
 167
 168         if options.get_file:
 169                 ftp_client.get_file(options.get_file)
 170
 171         if options.put_file:
 172                 ftp_client.put_file(options.put_file)
 173
 174         if options.move_file:
 175                 ftp_client.move_file(options.move_file, options.dest_file)
 176
 177         if options.copy_file:
 178                 ftp_client.copy_file(options.copy_file, options.dest_file)
 179
 180         if options.remove_file:
 181                 ftp_client.remove_file(options.remove_file)
 182
 183         mainloop.run()