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")
-
- return parser.parse_args()
+ 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")
+
+ return parser.parse_args()
def set_folder(session, new_dir):
- for node in new_dir.split("/"):
- session.SetFolder(node)
+ 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.GetFolderListing(dict()):
+ print "%s/" % (i["Name"])
+
+ def list_messages(self, folder):
+ ret = self.map.GetMessageListing(folder, dict())
+ print pformat(unwrap(ret))
+
+ def get_message(self, handle):
+ self.map.GetMessageListing("", 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("",reply_handler=self.create_transfer_reply,
+ error_handler=self.error)
if __name__ == '__main__':
- dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
+ dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
- parser = OptionParser()
+ parser = OptionParser()
- (options, args) = parse_options()
+ (options, args) = parse_options()
- if not options.device:
- parser.print_help()
- exit(0)
+ if not options.device:
+ parser.print_help()
+ exit(0)
- bus = dbus.SessionBus()
- mainloop = gobject.MainLoop()
+ bus = dbus.SessionBus()
+ mainloop = gobject.MainLoop()
- client = dbus.Interface(bus.get_object("org.openobex.client", "/"),
- "org.openobex.Client")
+ client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"),
+ "org.bluez.obex.Client")
- session_path = client.CreateSession({ "Destination": options.device,
- "Target": "map"})
+ print "Creating Session"
+ path = client.CreateSession(options.device, { "Target": "map" })
- session = dbus.Interface(bus.get_object("org.openobex.client", session_path),
- "org.openobex.Session")
+ map_client = MapClient(path, options.verbose)
- map = dbus.Interface(bus.get_object("org.openobex.client", session_path),
- "org.openobex.MessageAccess")
+ if options.new_dir:
+ map_client.set_folder(options.new_dir)
- if options.new_dir:
- set_folder(map, options.new_dir)
+ if options.ls_dir:
+ map_client.list_folders()
- if options.ls_dir:
- print map.GetFolderListing(dict())
+ if options.ls_msg is not None:
+ map_client.list_messages(options.ls_msg)
- if options.ls_msg is not None:
- print map.GetMessageListing(options.ls_msg, dict())
+ if options.get_msg is not None:
+ map_client.get_message(options.get_msg)
- mainloop.run()
+ mainloop.run()
projects / profile / ivi / obexd.git / blobdiff