diff --git a/test/exchange-business-cards b/test/exchange-business-cards
new file mode 100755
index 0000000..aace07b
--- /dev/null
+++ b/test/exchange-business-cards
+#!/usr/bin/python
+
+import sys
+import dbus
+
+bus = dbus.SessionBus()
+client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"),
+ "org.bluez.obex.Client")
+
+if (len(sys.argv) < 4):
+ print "Usage: %s <device> <clientfile> <file>" % (sys.argv[0])
+ sys.exit(1)
+
+print "Creating Session"
+path = client.CreateSession(sys.argv[1], { "Target": "OPP" })
+opp = dbus.Interface(bus.get_object("org.bluez.obex.client", path),
+ "org.bluez.obex.ObjectPush")
+
+opp.ExchangeBusinessCards(sys.argv[2], sys.argv[3])
diff --git a/test/ftp-client b/test/ftp-client
new file mode 100755
index 0000000..064a81a
--- /dev/null
+++ b/test/ftp-client
+#!/usr/bin/python
+
+import gobject
+
+import sys
+import dbus
+import dbus.service
+import dbus.mainloop.glib
+import os.path
+from optparse import OptionParser
+
+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()
+
+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)
+
+ bus = dbus.SessionBus()
+ mainloop = gobject.MainLoop()
+
+ client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"),
+ "org.bluez.obex.Client")
+
+ print "Creating Session"
+ path = client.CreateSession(options.device, { "Target": "ftp" })
+
+ ftp_client = FtpClient(path, options.verbose)
+
+ if options.new_dir:
+ ftp_client.change_folder(options.new_dir)
+
+ if options.list_dir:
+ ftp_client.list_folder()
+
+ if options.get_file:
+ ftp_client.get_file(options.get_file)
+
+ if options.put_file:
+ ftp_client.put_file(options.put_file)
+
+ if options.move_file:
+ ftp_client.move_file(options.move_file, options.dest_file)
+
+ if options.copy_file:
+ ftp_client.copy_file(options.copy_file, options.dest_file)
+
+ if options.remove_file:
+ ftp_client.remove_file(options.remove_file)
+
+ mainloop.run()
diff --git a/test/get-obex-capabilities b/test/get-obex-capabilities
new file mode 100755
index 0000000..a06f2eb
--- /dev/null
+++ b/test/get-obex-capabilities
+#!/usr/bin/python
+
+import sys
+import dbus
+
+bus = dbus.SessionBus()
+client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"),
+ "org.bluez.obex.Client")
+
+if (len(sys.argv) < 3):
+ print "Usage: %s <device> <target>" % (sys.argv[0])
+ sys.exit(1)
+
+print "Creating Session"
+session_path = client.CreateSession(sys.argv[1], { "Target": sys.argv[2] })
+session = dbus.Interface(bus.get_object("org.bluez.obex.client", session_path),
+ "org.bluez.obex.Session")
+
+print session.GetCapabilities()
diff --git a/test/list-folders b/test/list-folders
new file mode 100755
index 0000000..ecd456c
--- /dev/null
+++ b/test/list-folders
+#!/usr/bin/python
+
+import sys
+import dbus
+
+
+def list_folder(folder):
+ bus = dbus.SessionBus()
+ client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"),
+ "org.bluez.obex.Client")
+
+ path = client.CreateSession(sys.argv[1], { "Target": "ftp" })
+
+ ftp = dbus.Interface(bus.get_object("org.bluez.obex.client", path),
+ "org.bluez.obex.FileTransfer")
+
+ if folder:
+ for node in folder.split("/"):
+ ftp.ChangeFolder(node)
+
+ for i in ftp.ListFolder():
+ if i["Type"] == "folder":
+ print "%s/" % (i["Name"])
+ else:
+ print "%s" % (i["Name"])
+
+
+if __name__ == '__main__':
+
+ if len(sys.argv) < 2:
+ print "Usage: %s <device> [folder]" % (sys.argv[0])
+ sys.exit(1)
+
+ folder = None
+ if len(sys.argv) == 3:
+ folder = sys.argv[2]
+
+ list_folder(folder)
diff --git a/test/map-client b/test/map-client
new file mode 100755
index 0000000..756ebb8
--- /dev/null
+++ b/test/map-client
+#!/usr/bin/python
+
+import gobject
+
+import sys
+import os
+import dbus
+import dbus.mainloop.glib
+from optparse import OptionParser
+
+from pprint import pformat
+
+def unwrap(x):
+ """Hack to unwrap D-Bus values, so that they're easier to read when
+ printed. Taken from d-feet """
+
+ if isinstance(x, list):
+ return map(unwrap, x)
+
+ if isinstance(x, tuple):
+ return tuple(map(unwrap, x))
+
+ if isinstance(x, dict):
+ return dict([(unwrap(k), unwrap(v)) for k, v in x.iteritems()])
+
+ for t in [unicode, str, long, int, float, bool]:
+ if isinstance(x, t):
+ return t(x)
+
+ return x
+
+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", "--lsdir", action="store_true", dest="ls_dir",
+ help="List folders in current directory")
+ parser.add_option("-v", "--verbose", action="store_true", dest="verbose")
+ parser.add_option("-L", "--lsmsg", action="store", dest="ls_msg",
+ help="List messages in supplied CWD subdir")
+ parser.add_option("-g", "--get", action="store", dest="get_msg",
+ help="Get message contents")
+ parser.add_option("--get-properties", action="store", dest="get_msg_properties",
+ help="Get message properties")
+ parser.add_option("--mark-read", action="store", dest="mark_msg_read",
+ help="Marks the messages as read")
+ parser.add_option("--mark-unread", action="store", dest="mark_msg_unread",
+ help="Marks the messages as unread")
+ parser.add_option("--mark-deleted", action="store", dest="mark_msg_deleted",
+ help="Deletes the message from the folder")
+ parser.add_option("--mark-undeleted", action="store", dest="mark_msg_undeleted",
+ help="Undeletes the message")
+ parser.add_option("-u", "--update-inbox", action="store_true", dest="update_inbox",
+ help="Checks for new mails")
+
+ return parser.parse_args()
+
+def set_folder(session, new_dir):
+ session.SetFolder(new_dir)
+
+class MapClient:
+ def __init__(self, session_path, verbose=False):
+ self.progress = 0
+ self.transfer_path = None
+ self.props = dict()
+ self.verbose = verbose
+ self.path = session_path
+ bus = dbus.SessionBus()
+ obj = bus.get_object("org.bluez.obex.client", session_path)
+ self.session = dbus.Interface(obj, "org.bluez.obex.Session")
+ self.map = dbus.Interface(obj, "org.bluez.obex.MessageAccess")
+ 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")
+
+ def create_transfer_reply(self, reply):
+ (path, properties) = reply
+ self.transfer_path = path
+ self.props[path] = properties
+ if self.verbose:
+ print "Transfer created: %s (file %s)" % (path,
+ properties["Filename"])
+
+ 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"
+ properties = self.props.get(path)
+ if properties == None:
+ return
+ f = open(properties["Filename"], "r")
+ os.remove(properties["Filename"])
+ print f.readlines()
+
+ 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 set_folder(self, new_dir):
+ self.map.SetFolder(new_dir)
+
+ def list_folders(self):
+ for i in self.map.ListFolders(dict()):
+ print "%s/" % (i["Name"])
+
+ def list_messages(self, folder):
+ ret = self.map.ListMessages(folder, dict())
+ print pformat(unwrap(ret))
+
+ def get_message(self, handle):
+ self.map.ListMessages("", dict())
+ path = self.path + "/message" + handle
+ obj = bus.get_object("org.bluez.obex.client", path)
+ msg = dbus.Interface(obj, "org.bluez.obex.Message")
+ msg.Get("", True, reply_handler=self.create_transfer_reply,
+ error_handler=self.error)
+
+ def get_message_properties(self, handle):
+ self.map.ListMessages("", dict())
+ path = self.path + "/message" + handle
+ obj = bus.get_object("org.bluez.obex.client", path)
+ msg = dbus.Interface(obj, "org.bluez.obex.Message")
+ ret = msg.GetProperties()
+ print pformat(unwrap(ret))
+
+ def set_message_property(self, handle, prop, flag):
+ self.map.ListMessages("", dict())
+ path = self.path + "/message" + handle
+ obj = bus.get_object("org.bluez.obex.client", path)
+ msg = dbus.Interface(obj, "org.bluez.obex.Message")
+ msg.SetProperty (prop, flag);
+
+ def update_inbox(self):
+ self.map.UpdateInbox()
+
+
+if __name__ == '__main__':
+
+ dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
+
+ parser = OptionParser()
+
+ (options, args) = parse_options()
+
+ if not options.device:
+ parser.print_help()
+ exit(0)
+
+ bus = dbus.SessionBus()
+ mainloop = gobject.MainLoop()
+
+ client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"),
+ "org.bluez.obex.Client")
+
+ print "Creating Session"
+ path = client.CreateSession(options.device, { "Target": "map" })
+
+ map_client = MapClient(path, options.verbose)
+
+ if options.new_dir:
+ map_client.set_folder(options.new_dir)
+
+ if options.ls_dir:
+ map_client.list_folders()
+
+ if options.ls_msg is not None:
+ map_client.list_messages(options.ls_msg)
+
+ if options.get_msg is not None:
+ map_client.get_message(options.get_msg)
+
+ if options.get_msg_properties is not None:
+ map_client.get_message_properties(options.get_msg_properties)
+
+ if options.mark_msg_read is not None:
+ map_client.set_message_property(options.mark_msg_read, "Read", True)
+
+ if options.mark_msg_unread is not None:
+ map_client.set_message_property(options.mark_msg_unread, "Read", False)
+
+ if options.mark_msg_deleted is not None:
+ map_client.set_message_property(options.mark_msg_deleted, "Deleted", True)
+
+ if options.mark_msg_undeleted is not None:
+ map_client.set_message_property(options.mark_msg_undeleted, "Deleted", False)
+
+ if options.update_inbox:
+ map_client.update_inbox()
+
+ mainloop.run()
diff --git a/test/opp-client b/test/opp-client
new file mode 100755
index 0000000..3d23dfb
--- /dev/null
+++ b/test/opp-client
+#!/usr/bin/python
+
+import sys
+import dbus
+import gobject
+import dbus.mainloop.glib
+import os.path
+from optparse import OptionParser
+
+def parse_options():
+ parser.add_option("-d", "--device", dest="device",
+ help="Device to connect", metavar="DEVICE")
+ parser.add_option("-p", "--pull", dest="pull_to_file",
+ help="Pull vcard and store in FILE", metavar="FILE")
+ parser.add_option("-s", "--send", dest="send_file",
+ help="Send FILE", metavar="FILE")
+ parser.add_option("-v", "--verbose", action="store_true",
+ dest="verbose")
+
+ return parser.parse_args()
+
+class OppClient:
+ def __init__(self, session_path, verbose=False):
+ self.progress = 0
+ self.transfer_path = None
+ 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.opp = dbus.Interface(obj, "org.bluez.obex.ObjectPush")
+ 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 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 pull_business_card(self, filename):
+ self.opp.PullBusinessCard(os.path.abspath(filename),
+ reply_handler=self.create_transfer_reply,
+ error_handler=self.error)
+
+ def send_file(self, filename):
+ self.opp.SendFile(os.path.abspath(filename),
+ reply_handler=self.create_transfer_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)
+
+ bus = dbus.SessionBus()
+ mainloop = gobject.MainLoop()
+
+ client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"),
+ "org.bluez.obex.Client")
+
+ print "Creating Session"
+ path = client.CreateSession(options.device, { "Target": "OPP" })
+
+ opp_client = OppClient(path, options.verbose)
+
+ if options.pull_to_file:
+ opp_client.pull_business_card(options.pull_to_file)
+
+ if options.send_file:
+ opp_client.send_file(options.send_file)
+
+ mainloop.run()
diff --git a/test/pbap-client b/test/pbap-client
new file mode 100755
index 0000000..fbe930c
--- /dev/null
+++ b/test/pbap-client
+#!/usr/bin/python
+
+import gobject
+
+import sys
+import os
+import dbus
+import dbus.service
+import dbus.mainloop.glib
+
+class Transfer:
+ def __init__(self, callback_func):
+ self.callback_func = callback_func
+ self.path = None
+ self.filename = None
+
+class PbapClient:
+ def __init__(self, session_path):
+ self.transfers = 0
+ self.props = dict()
+ self.flush_func = None
+ bus = dbus.SessionBus()
+ obj = bus.get_object("org.bluez.obex.client", session_path)
+ self.session = dbus.Interface(obj, "org.bluez.obex.Session")
+ self.pbap = dbus.Interface(obj,
+ "org.bluez.obex.PhonebookAccess")
+ 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")
+
+ def register(self, reply, transfer):
+ (path, properties) = reply
+ transfer.path = path
+ transfer.filename = properties["Filename"]
+ self.props[path] = transfer
+ print "Transfer created: %s (file %s)" % (path,
+ transfer.filename)
+
+ def error(self, err):
+ print err
+ mainloop.quit()
+
+ def transfer_complete(self, path):
+ req = self.props.get(path)
+ if req == None:
+ return
+ self.transfers -= 1
+ print "Transfer %s finished" % path
+ f = open(req.filename, "r")
+ os.remove(req.filename)
+ lines = f.readlines()
+ del self.props[path]
+ req.callback_func(lines)
+
+ if (len(self.props) == 0) and (self.transfers == 0):
+ if self.flush_func != None:
+ f = self.flush_func
+ self.flush_func = None
+ f()
+
+ def transfer_error(self, code, message, path):
+ req = self.props.get(path)
+ if req == None:
+ return
+ print "Transfer finished with error %s: %s" % (code, message)
+ mainloop.quit()
+
+ def pull(self, vcard, params, func):
+ req = Transfer(func)
+ self.pbap.Pull(vcard, "", params,
+ reply_handler=lambda r: self.register(r, req),
+ error_handler=self.error)
+ self.transfers += 1
+
+ def pull_all(self, params, func):
+ req = Transfer(func)
+ self.pbap.PullAll("", params,
+ reply_handler=lambda r: self.register(r, req),
+ error_handler=self.error)
+ self.transfers += 1
+
+ def flush_transfers(self, func):
+ if (len(self.props) == 0) and (self.transfers == 0):
+ return
+ self.flush_func = func
+
+ def interface(self):
+ return self.pbap
+
+if __name__ == '__main__':
+
+ dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
+
+ bus = dbus.SessionBus()
+ mainloop = gobject.MainLoop()
+
+ client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"),
+ "org.bluez.obex.Client")
+
+ if (len(sys.argv) < 2):
+ print "Usage: %s <device>" % (sys.argv[0])
+ sys.exit(1)
+
+ print "Creating Session"
+ session_path = client.CreateSession(sys.argv[1], { "Target": "PBAP" })
+
+ pbap_client = PbapClient(session_path)
+
+ def process_result(lines, header):
+ if header != None:
+ print header
+ for line in lines:
+ print line,
+ print
+
+ def test_paths(paths):
+ if len(paths) == 0:
+ print
+ print "FINISHED"
+ mainloop.quit()
+ return
+
+ path = paths[0]
+
+ print "\n--- Select Phonebook %s ---\n" % (path)
+ pbap_client.interface().Select("int", path)
+
+ print "\n--- GetSize ---\n"
+ ret = pbap_client.interface().GetSize()
+ print "Size = %d\n" % (ret)
+
+ print "\n--- List vCard ---\n"
+ try:
+ ret = pbap_client.interface().List(dbus.Dictionary())
+ except:
+ ret = []
+
+ params = dbus.Dictionary({ "Format" : "vcard30",
+ "Fields" : [ "VERSION", "FN", "TEL"] })
+ for item in ret:
+ print "%s : %s" % (item[0], item[1])
+ pbap_client.pull(item[0], params,
+ lambda x: process_result(x, None))
+
+ pbap_client.pull_all(params, lambda x: process_result(x,
+ "\n--- PullAll ---\n"))
+
+ pbap_client.flush_transfers(lambda: test_paths(paths[1:]))
+
+ test_paths(["PB", "ICH", "OCH", "MCH", "CCH"])
+
+ mainloop.run()
diff --git a/test/simple-obex-agent b/test/simple-obex-agent
new file mode 100755
index 0000000..1902082
--- /dev/null
+++ b/test/simple-obex-agent
+#!/usr/bin/python
+
+import gobject
+
+import sys
+import dbus
+import dbus.service
+import dbus.mainloop.glib
+
+class Agent(dbus.service.Object):
+ def __init__(self, conn=None, obj_path=None):
+ dbus.service.Object.__init__(self, conn, obj_path)
+ self.pending_auth = False
+
+ @dbus.service.method("org.bluez.obex.Agent", in_signature="osssii",
+ out_signature="s")
+ def Authorize(self, dpath, device, filename, ftype, length, time):
+ global transfers
+
+ self.pending_auth = True
+ print "Authorize (%s, %s, %s) Y/n" % (path, device, filename)
+ auth = raw_input().strip("\n ")
+
+ if auth == "n" or auth == "N":
+ self.pending_auth = False
+ raise dbus.DBusException(
+ "org.bluez.obex.Error.Rejected: "
+ "Not Autorized")
+
+ print "Full filename (including path):"
+ self.pending_auth = False
+
+ transfers.append(Transfer(dpath, filename, 0, length))
+ return raw_input().strip("\n ")
+
+ @dbus.service.method("org.bluez.obex.Agent", in_signature="",
+ out_signature="")
+ def Cancel(self):
+ print "Authorization Canceled"
+ self.pending_auth = False
+
+
+class Transfer(object):
+ def __init__(self, dpath, filename=None, transfered=-1, size=-1):
+ self.dpath = dpath
+ self.filename = filename
+ self.transfered = transfered
+ self.size = size
+
+ def update(self, filename=None, transfered=-1, total=-1):
+ if filename:
+ self.filename = filename
+ self.transfered = transfered
+ self.size = total
+
+ def cancel(self):
+ transfer_iface = dbus.Interface(bus.get_object(
+ "org.bluez.obex", self.dpath),
+ "org.bluez.obex.Transfer")
+ transfer_iface.Cancel()
+
+ def __str__(self):
+ p = float(self.transfered) / float(self.size) * 100
+ return "%s (%s) (%.2f%%)" % (self.filename, self.dpath, p)
+
+ __repr__ = __str__
+
+
+if __name__ == '__main__':
+ global transfers
+
+ def new_transfer(dpath):
+ print "new transfer"
+ bus.add_signal_receiver(progress,
+ dbus_interface="org.bluez.obex.Transfer",
+ signal_name="Progress",
+ path_keyword="dpath")
+
+ def transfer_completed(dpath, success):
+ global transfers
+
+ print "\ntransfer completed => %s" % (success and "Success" or "Fail")
+ transfers = [t for t in transfers if t.dpath != dpath]
+
+ def progress(total, current, dpath):
+ s = []
+ for t in transfers:
+ if t.dpath == dpath:
+ t.update(None, current, total)
+ s.append("%s" % (t))
+ sys.stdout.write(" ".join(s) + "\r")
+ sys.stdout.flush()
+
+
+ transfers = []
+
+ dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
+
+ bus = dbus.SessionBus()
+ manager = dbus.Interface(bus.get_object("org.bluez.obex", "/"),
+ "org.bluez.obex.Manager")
+ bus.add_signal_receiver(new_transfer,
+ dbus_interface="org.bluez.obex.Manager",
+ signal_name="TransferStarted")
+
+ bus.add_signal_receiver(transfer_completed,
+ dbus_interface="org.bluez.obex.Manager",
+ signal_name="TransferCompleted")
+
+ path = "/test/agent"
+ agent = Agent(bus, path)
+
+ mainloop = gobject.MainLoop()
+
+ manager.RegisterAgent(path)
+ print "Agent registered"
+
+ cont = True
+ while cont:
+ try:
+ mainloop.run()
+ except KeyboardInterrupt:
+ if agent.pending_auth:
+ agent.Cancel()
+ elif len(transfers) > 0:
+ for a in transfers:
+ a.cancel()
+ else:
+ cont = False
+
+ # manager.UnregisterAgent(path)
+ # print "Agent unregistered"