import os.path
from optparse import OptionParser
-class Agent(dbus.service.Object):
- def __init__(self, conn=None, obj_path=None, verbose=False):
- dbus.service.Object.__init__(self, conn, obj_path)
- self.verbose = verbose
-
- @dbus.service.method("org.openobex.Agent",
- in_signature="o", out_signature="s")
- def Request(self, path):
- return ""
-
- @dbus.service.method("org.openobex.Agent",
- in_signature="ot", out_signature="")
- def Progress(self, path, transferred):
- if self.verbose:
- print "Transfer progress (%d bytes)" % (transferred)
- return
-
- @dbus.service.method("org.openobex.Agent",
- in_signature="o", out_signature="")
- def Complete(self, path):
- if self.verbose:
- print "Transfer finished"
- mainloop.quit()
-
- @dbus.service.method("org.openobex.Agent",
- in_signature="os", out_signature="")
- def Error(self, path, error):
- print "Transfer finished with an error: %s" % (error)
- mainloop.quit()
-
- @dbus.service.method("org.openobex.Agent",
- in_signature="", out_signature="")
- def Release(self):
- mainloop.quit()
-
-
def parse_options():
- parser.add_option("-d", "--device", dest="device",
- help="Device to connect", metavar="DEVICE")
- parser.add_option("-c", "--chdir", dest="new_dir",
- help="Change current directory to DIR", metavar="DIR")
- parser.add_option("-l", "--list", action="store_true", dest="list_dir",
- help="List the current directory")
- parser.add_option("-g", "--get", dest="get_file",
- help="Get FILE", metavar="FILE")
- parser.add_option("-p", "--put", dest="put_file",
- help="Put FILE", metavar="FILE")
- parser.add_option("-y", "--copy", dest="copy_file",
- help="Copy FILE", metavar="FILE")
- parser.add_option("-m", "--move", dest="move_file",
- help="Move FILE", metavar="FILE")
- parser.add_option("-n", "--destname", dest="dest_file",
- help="Destination FILE", metavar="FILE")
- parser.add_option("-r", "--remove", dest="remove_file",
- help="Remove FILE", metavar="FILE")
- parser.add_option("-v", "--verbose", action="store_true", dest="verbose")
-
- return parser.parse_args()
-
-def error(err):
- print err
-
-def void_reply():
- pass
-
-def change_folder(session, new_dir):
- for node in new_dir.split("/"):
- session.ChangeFolder(node)
-
-def list_folder(session):
- for i in session.ListFolder():
- if i["Type"] == "folder":
- print "%s/" % (i["Name"])
- else:
- print "%s" % (i["Name"])
-
-def put_file(session, filename):
- session.PutFile(os.path.abspath(filename),
- os.path.basename(filename),
- reply_handler=void_reply,
- error_handler=error)
-
-def get_file(session, filename):
- session.GetFile(os.path.abspath(filename),
- os.path.basename(filename),
- reply_handler=void_reply,
- error_handler=error)
-
-def remove_file(session, filename):
- session.Delete(filename,
- reply_handler=void_reply,
- error_handler=error)
-
-def move_file(session, filename, destname):
- session.MoveFile(filename,
- destname,
- reply_handler=void_reply,
- error_handler=error)
-
-def copy_file(session, filename, destname):
- session.CopyFile(filename,
- destname,
- reply_handler=void_reply,
- error_handler=error)
+ parser.add_option("-d", "--device", dest="device",
+ help="Device to connect", metavar="DEVICE")
+ parser.add_option("-c", "--chdir", dest="new_dir",
+ help="Change current directory to DIR", metavar="DIR")
+ parser.add_option("-l", "--list", action="store_true", dest="list_dir",
+ help="List the current directory")
+ parser.add_option("-g", "--get", dest="get_file",
+ help="Get FILE", metavar="FILE")
+ parser.add_option("-p", "--put", dest="put_file",
+ help="Put FILE", metavar="FILE")
+ parser.add_option("-y", "--copy", dest="copy_file",
+ help="Copy FILE", metavar="FILE")
+ parser.add_option("-m", "--move", dest="move_file",
+ help="Move FILE", metavar="FILE")
+ parser.add_option("-n", "--destname", dest="dest_file",
+ help="Destination FILE", metavar="FILE")
+ parser.add_option("-r", "--remove", dest="remove_file",
+ help="Remove FILE", metavar="FILE")
+ parser.add_option("-v", "--verbose", action="store_true",
+ dest="verbose")
+
+ return parser.parse_args()
+
+class FtpClient:
+ def __init__(self, session_path, verbose=False):
+ self.progress = 0
+ self.transfer_path = None
+ self.transfer_size = 0
+ self.verbose = verbose
+ bus = dbus.SessionBus()
+ obj = bus.get_object("org.bluez.obex.client", session_path)
+ self.session = dbus.Interface(obj, "org.bluez.obex.Session")
+ self.ftp = dbus.Interface(obj, "org.bluez.obex.FileTransfer")
+ bus.add_signal_receiver(self.transfer_complete,
+ dbus_interface="org.bluez.obex.Transfer",
+ signal_name="Complete",
+ path_keyword="path")
+ bus.add_signal_receiver(self.transfer_error,
+ dbus_interface="org.bluez.obex.Transfer",
+ signal_name="Error",
+ path_keyword="path")
+ if self.verbose:
+ bus.add_signal_receiver(self.transfer_progress,
+ dbus_interface="org.bluez.obex.Transfer",
+ signal_name="PropertyChanged",
+ path_keyword="path")
+
+ def create_transfer_reply(self, reply):
+ (path, properties) = reply
+ self.transfer_path = path
+ self.transfer_size = properties["Size"]
+ if self.verbose:
+ print "Transfer created: %s" % path
+
+ def generic_reply(self):
+ if self.verbose:
+ print "Operation succeeded"
+
+ def error(self, err):
+ print err
+ mainloop.quit()
+
+ def transfer_complete(self, path):
+ if path != self.transfer_path:
+ return
+ if self.verbose:
+ print "Transfer finished"
+ mainloop.quit()
+
+ def transfer_error(self, code, message, path):
+ if path != self.transfer_path:
+ return
+ print "Transfer finished with error %s: %s" % (code, message)
+ mainloop.quit()
+
+ def transfer_progress(self, prop, value, path):
+ if path != self.transfer_path:
+ return
+
+ if prop != "Progress":
+ return
+
+ speed = (value - self.progress) / 1000
+ print "Transfer progress %d/%d at %d kBps" % (value,
+ self.transfer_size,
+ speed)
+ self.progress = value
+
+ def change_folder(self, new_dir):
+ for node in new_dir.split("/"):
+ self.ftp.ChangeFolder(node)
+
+ def list_folder(self):
+ for i in self.ftp.ListFolder():
+ if i["Type"] == "folder":
+ print "%s/" % (i["Name"])
+ else:
+ print "%s" % (i["Name"])
+
+ def put_file(self, filename):
+ self.ftp.PutFile(os.path.abspath(filename),
+ os.path.basename(filename),
+ reply_handler=self.create_transfer_reply,
+ error_handler=self.error)
+
+ def get_file(self, filename):
+ self.ftp.GetFile(os.path.abspath(filename),
+ os.path.basename(filename),
+ reply_handler=self.create_transfer_reply,
+ error_handler=self.error)
+
+ def remove_file(self, filename):
+ self.ftp.Delete(filename,
+ reply_handler=self.generic_reply,
+ error_handler=self.error)
+
+ def move_file(self, filename, destname):
+ self.ftp.MoveFile(filename, destname,
+ reply_handler=self.generic_reply,
+ error_handler=self.error)
+
+ def copy_file(self, filename, destname):
+ self.ftp.CopyFile(filename, destname,
+ reply_handler=self.generic_reply,
+ error_handler=self.error)
if __name__ == '__main__':
- dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
-
- parser = OptionParser()
-
- (options, args) = parse_options()
-
- if not options.device:
- parser.print_help()
- sys.exit(0)
+ dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
- bus = dbus.SessionBus()
- mainloop = gobject.MainLoop()
+ parser = OptionParser()
- path = "/test/agent"
- agent = Agent(bus, path, options.verbose)
+ (options, args) = parse_options()
- client = dbus.Interface(bus.get_object("org.openobex.client", "/"),
- "org.openobex.Client")
+ if not options.device:
+ parser.print_help()
+ sys.exit(0)
- session_path = client.CreateSession({ "Destination": options.device,
- "Target": "ftp"})
+ bus = dbus.SessionBus()
+ mainloop = gobject.MainLoop()
- session = dbus.Interface(bus.get_object("org.openobex.client", session_path),
- "org.openobex.Session")
+ client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"),
+ "org.bluez.obex.Client")
- session.AssignAgent(path)
+ print "Creating Session"
+ path = client.CreateSession(options.device, { "Target": "ftp" })
- ftp = dbus.Interface(bus.get_object("org.openobex.client", session_path),
- "org.openobex.FileTransfer")
+ ftp_client = FtpClient(path, options.verbose)
- if options.new_dir:
- change_folder(ftp, options.new_dir)
+ if options.new_dir:
+ ftp_client.change_folder(options.new_dir)
- if options.list_dir:
- list_folder(ftp)
+ if options.list_dir:
+ ftp_client.list_folder()
- if options.get_file:
- get_file(ftp, options.get_file)
+ if options.get_file:
+ ftp_client.get_file(options.get_file)
- if options.put_file:
- put_file(ftp, options.put_file)
+ if options.put_file:
+ ftp_client.put_file(options.put_file)
- if options.move_file:
- move_file(ftp, options.move_file, options.dest_file)
+ if options.move_file:
+ ftp_client.move_file(options.move_file, options.dest_file)
- if options.copy_file:
- copy_file(ftp, options.copy_file, options.dest_file)
+ if options.copy_file:
+ ftp_client.copy_file(options.copy_file, options.dest_file)
- if options.remove_file:
- remove_file(ftp, options.remove_file)
+ if options.remove_file:
+ ftp_client.remove_file(options.remove_file)
- mainloop.run()
+ mainloop.run()
projects / profile / ivi / obexd.git / blobdiff