obexftp

#! /usr/bin/env python
"""ObexFTP Virtual FileSystem for Midnight Commander

Manipulate a cell phone's filesystem calling obexftp binary. This is a complete
user-mode solution, no kernel modules required (unlike SieFS or such). The
script implements all commands of Midnight Commander VFS, except for
undocumented "run"; anyway there are no runnable files in cell phones. The
script is written in Python because I I need to parse XML directory listings
from obexftp, and Python is the best of all languages suited for the task ;).

The script requires Midnight Commander 3.1+
(http://www.midnight-commander.org/), Python 2.3+ (http://www.python.org/),
OpenOBEX 1.0.1+ (http://dev.zuckschwerdt.org/openobex/) and ObexFTP 0.10.4+
(http://triq.net/obexftp).

Edit the script, and correct the full path to the obexftp binary (see
obexftp_prog below). For mc 4.7+ put the script in
$HOME/[.local/share/].mc/extfs.d. For older versions put it in
/usr/[local/][lib|share]/mc/extfs and add a line "obexftp" to the
/usr/[local/][lib|share]/mc/extfs/extfs.ini. Make the script executable.

See detailed installation instructions at
https://phdru.name/Software/mc/INSTALL.html.

Create somewhere a transport file. The transport file may have any name, and is
expected to be a text file with at least one line defining the transport to
your device. Other lines in the file are ignored.

First word in the line is a transport name - Bluetooth, TTY or IrDA. The name
is case-insensitive.

For the Bluetooth transport put there a line "Bluetooth CP:AD:RE:SS channel",
where CP:AD:RE:SS is the hardware address of the device you want to connect to,
and "channel" is the OBEX File Transfer channel; you can discover the address
and the channel for your device by using commands like "hcitool scan" and
"sdptool browse".

For the USB put the interface number: "usb interface".

For the TTY put the device name: "tty /dev/ttyUSB0".

For the IrDA: just put "IrDA" in the file.

Now run this "cd" command in the Midnight Commander (in the "bindings" file the
command is "%cd"): cd transport#obexftp, where "transport" is the name of your
transport file. The script uses obexftp to connect to the device and list files
and directories. Please be warned that opening the VFS for the first time is
VERY slow, because the script needs to scan the entire cell phone's filesystem,
and there are timeouts between connections, which don't make the scanning
faster. Midnight Commander caches the result so you can browse and manipulate
files and directories quickly.

Please note that manipulating the filesystem using your phone's internal
filemanager in parallel with the VFS leads to a disagreement between the VFS
cache and the phone. It is not very dangerous but inconvenient. There is no way
to clear the VFS cache in Midnight Commander and reread the filesystem. You
have to exit the VFS (cd /, for example) and return back using cd
transport#obexftp command. Sometimes even this doesn't help - Midnight
Commander shows the same cached VFS image. Exit Midnight Commander and restart
it.

If something goes wrong set the logging level (see setLevel() below) to INFO or
DEBUG and look in the obexftp-mcextfs.log file. The file is put in the same
directory as the transport file, if it possible; if not the file will be put
into a temporary directory, usually /tmp, or /var/tmp, or whatever directory is
named in $TMP environment variable.

"""

__version__ = "1.4.0"
__author__ = "Oleg Broytman <phd@phdru.name>"
__copyright__ = "Copyright (C) 2004-2013 PhiloSoft Design"
__license__ = "GPL"


# Change this to suite your needs
obexftp_prog = "/usr/bin/obexftp"


import sys, os, shutil
from time import sleep
import xml.dom.minidom
from tempfile import mkstemp, mkdtemp, _candidate_tempdir_list

try:
   import locale
   use_locale = True
except ImportError:
   use_locale = False

if use_locale:
   # Get the default charset.
   try:
      lcAll = locale.getdefaultlocale()
   except locale.Error, err:
      print >>sys.stderr, "WARNING:", err
      lcAll = []

   if len(lcAll) == 2:
      default_encoding = lcAll[1]
   else:
      try:
         default_encoding = locale.getpreferredencoding()
      except locale.Error, err:
         print >>sys.stderr, "WARNING:", err
         default_encoding = sys.getdefaultencoding()
else:
   default_encoding = sys.getdefaultencoding()


import logging
logger = logging.getLogger('obexftp-mcextfs')
log_err_handler = logging.StreamHandler(sys.stderr)
logger.addHandler(log_err_handler)
logger.setLevel(logging.ERROR)


if len(sys.argv) < 3:
   logger.critical("""\
ObexFTP Virtual FileSystem for Midnight Commander version %s
Author: %s
%s

This is not a program. Put the script in $HOME/[.local/share/].mc/extfs.d or
/usr/[local/][lib|share]/mc/extfs. For more information read the source!""",
   __version__, __author__, __copyright__
)
   sys.exit(1)


tempdirlist = _candidate_tempdir_list()
tempdirlist.insert(0, os.path.abspath(os.path.dirname(sys.argv[2])))

found = False
for tempdir in tempdirlist:
   try:
      logfile_name = os.path.join(tempdir, 'obexftp-mcextfs.log')
      logfile = open(logfile_name, 'w')
   except IOError:
      pass
   else:
      found = True
      logfile.close()
      break

if not found:
   logger.critical("Cannot initialize error log file in directories %s" % str(tempdirlist))
   sys.exit(1)

logger.removeHandler(log_err_handler)
logger.addHandler(logging.FileHandler(logfile_name))

locale.setlocale(locale.LC_ALL, '')


# Parse ObexFTP XML directory listings

class DirectoryEntry(object):
   """Represent a remote file or a directory"""

   def __init__(self, type):
      self.type = type
      self.size = 0
      if type == "file":
         self.perm = "-rw-rw-rw-"
      elif type == "folder":
         self.perm = "drwxrwxrwx"
      else:
         raise ValueError, "unknown type '%s'; expected 'file' or 'folder'" % self.type

   def mtime(self):
      if not hasattr(self, "modified"): # telecom
         return "01-01-70 0:0"
      date, time = self.modified.split('T')
      year, month, day = date[2:4], date[4:6], date[6:8]
      hour, minute = time[:2], time[2:4]
      return "%s-%s-%s %s:%s" % (month, day, year, hour, minute)
   mtime = property(mtime)

   def __repr__(self):
      if self.type == "file":
         return """<%s: type=file, name=%s, size=%s, mtime=%s at 0x%x>""" % (
            self.__class__.__name__, self.name, self.size, self.mtime, id(self)
         )
      if self.type == "folder":
         if hasattr(self, "modified"):
            return """<%s: type=directory, name=%s, mtime=%s at 0x%x>""" % (
               self.__class__.__name__, self.name, self.mtime, id(self)
            )
         else: # telecom
            return """<%s: type=directory, name=%s at 0x%x>""" % (
               self.__class__.__name__, self.name, id(self)
            )
      raise ValueError, "unknown type '%s'; expected 'file' or 'folder'" % self.type

def get_entries(dom, type):
   entries = []
   for obj in dom.getElementsByTagName(type):
      entry = DirectoryEntry(type)
      attrs = obj.attributes
      for i in range(attrs.length):
         attr = attrs.item(i)
         setattr(entry, attr.name, attr.value)
      entries.append(entry)
   return entries


# A unique directory for temporary files
tmpdir_name = None

def setup_tmpdir():
   global tmpdir_name
   tmpdir_name = mkdtemp(".tmp", "mcobex-")
   os.chdir(tmpdir_name)

def cleanup_tmpdir():
   os.chdir(os.pardir)
   shutil.rmtree(tmpdir_name)


def _read(fd):
   out = []
   while True:
      s = os.read(fd, 1024)
      if not s:
         break
      out.append(s)
   return ''.join(out)


def _run(*args):
   """Run the obexftp binary catching errors"""

   out_fd, out_filename = mkstemp(".tmp", "mcobex-", tmpdir_name)
   err_fd, err_filename = mkstemp(".tmp", "mcobex-", tmpdir_name)

   command = "%s %s %s >%s 2>%s" % (obexftp_prog, obexftp_args, ' '.join(args),
      out_filename, err_filename)

   logger.debug("Running command %s", command)
   os.system(command)

   result = _read(out_fd)
   os.remove(out_filename)

   errors = _read(err_fd)
   os.remove(err_filename)

   logger.debug("    result: %s", result)
   logger.debug("    errors: %s", errors)
   return result, errors


def recursive_list(directory='/'):
   """List the directory recursively"""
   listing, errors = _run("-l '%s'" % directory)

   if not listing:
      logger.error("Error reading XML listing: %s", errors)
      return

   dom = xml.dom.minidom.parseString(listing)
   directories = get_entries(dom, "folder")
   files = get_entries(dom, "file")

   for entry in directories + files:
      fullpath = "%s/%s" % (directory, entry.name)
      fullpath = fullpath.encode(default_encoding)
      if fullpath.startswith('//'): fullpath = fullpath[1:]
      print entry.perm, "1 user group", entry.size, entry.mtime, fullpath

   for entry in directories:
      fullpath = "%s/%s" % (directory, entry.name)
      if fullpath.startswith('//'): fullpath = fullpath[1:]
      sleep(1)
      recursive_list(fullpath)

def mcobex_list():
   """List the entire VFS"""
   setup_tmpdir()
   try:
      recursive_list()
   finally:
      cleanup_tmpdir()


def mcobex_copyout():
   """Get a file from the VFS"""
   obex_filename = sys.argv[3]
   real_filename = sys.argv[4]

   setup_tmpdir()
   try:
      _run("-g '%s'" % obex_filename)
      try:
         os.rename(os.path.basename(obex_filename), real_filename)
      except OSError:
         logger.exception("Error CopyOut %s to %s", obex_filename, real_filename)
   finally:
      cleanup_tmpdir()


def mcobex_copyin():
   """Put a file to the VFS"""
   obex_filename = sys.argv[3]
   real_filename = sys.argv[4]
   dirname, filename = os.path.split(obex_filename)

   setup_tmpdir()
   try:
      try:
         os.rename(real_filename, filename)
         _run("-c '%s' -p '%s'" % (dirname, filename))
         os.rename(filename, real_filename) # by some reason MC wants the file back
      except OSError:
         logger.exception("Error CopyIn %s to %s", real_filename, obex_filename)
   finally:
      cleanup_tmpdir()


def mcobex_rm():
   """Remove a file from the VFS"""
   obex_filename = sys.argv[3]
   try:
      _run("-k '%s'" % obex_filename)
   finally:
      cleanup_tmpdir()


def mcobex_mkdir():
   """Create a directory in the VFS"""
   obex_dirname = sys.argv[3]
   try:
      _run("-C '%s'" % obex_dirname)
   finally:
      cleanup_tmpdir()


mcobex_rmdir = mcobex_rm


def transport_error(error_str):
   logger.critical("Error parsing the transport file: %s" % error_str)
   sys.exit(1)

def setup_transport():
   """Setup transport parameters for the obexftp program"""
   try:
      transport_file = open(sys.argv[2], 'r')
      line = transport_file.readline()
      transport_file.close()
   except IOError:
      transport_error("cannot read '%s'" % sys.argv[2])

   parts = line.strip().split()
   transport = parts[0].lower()

   if transport == "bluetooth":
      if len(parts) < 3:
         transport_error("not enough arguments for 'bluetooth' transport")
      elif len(parts) > 3:
         transport_error("too many arguments for 'bluetooth' transport")
      return ' '.join(["-b", parts[1], "-B", parts[2]])
   elif transport == "usb":
      if len(parts) < 2:
         transport_error("not enough arguments for 'usb' transport")
      elif len(parts) > 2:
         transport_error("too many arguments for 'usb' transport")
      return ' '.join(["-u", parts[1]])
   elif transport == "tty":
      if len(parts) < 2:
         transport_error("not enough arguments for 'tty' transport")
      elif len(parts) > 2:
         transport_error("too many arguments for 'tty' transport")
      return ' '.join(["-t", parts[1]])
   elif transport == "irda":
      if len(parts) > 1:
         transport_error("too many arguments for 'irda' transport")
      return "-i"
   else:
      logger.critical("Unknown transport '%s'; expected 'bluetooth', 'tty' or 'irda'", transport)
      sys.exit(1)


command = sys.argv[1]
procname = "mcobex_" + command

g = globals()
if not g.has_key(procname):
   logger.critical("Unknown command %s", command)
   sys.exit(1)


try:
   obexftp_args = setup_transport()
except SystemExit:
   raise
except:
   logger.exception("Error parsing the transport file")
   sys.exit(1)

try:
   g[procname]()
except SystemExit:
   raise
except:
   logger.exception("Error during run")