eric: comparison eric6/MicroPython/MicroPythonFileSystem.py

-:9a3adf033f90
+:ed510767c096
 from __future__ import unicode_literals
 import ast
 import time
+import os
 import stat
-import os
 from PyQt5.QtCore import pyqtSlot, pyqtSignal, QObject, QThread
 from .MicroPythonSerialPort import MicroPythonSerialPort
+from .MicroPythonFileSystemUtilities import (
+mtime2string, mode2string, decoratedName, listdirStat
+)
 class MicroPythonFileSystem(QObject):
 """
 Class implementing some file system commands for MicroPython.
-Some FTP like commands are provided to perform operations on the file
+Commands are provided to perform operations on the file system of a
-system of a connected MicroPython device. Supported commands are:
+connected MicroPython device. Supported commands are:
 <ul>
 <li>ls: directory listing</li>
 <li>lls: directory listing with meta data</li>
 <li>cd: change directory</li>
 <li>pwd: get the current directory</li>
 <li>put: copy a file to the connected device</li>
 <li>get: get a file from the connected device</li>
 <li>rm: remove a file from the connected device</li>
+<li>rmrf: remove a file/directory recursively (like 'rm -rf' in bash)
 <li>mkdir: create a new directory</li>
 <li>rmdir: remove an empty directory</li>
+</ul>
+There are additional commands related to time and version.
+<ul>
 <li>version: get version info about MicroPython</li>
+<li>syncTime: synchronize the time of the connected device</li>
+<li>showTime: show the current time of the connected device</li>
 </ul>
 """
 def __init__(self, parent=None):
 """
 Constructor
 out, err = self.__execute(commands)
 if err:
 raise IOError(self.__shortError(err))
 return ast.literal_eval(out.decode("utf-8"))
-def lls(self, dirname=""):
+def lls(self, dirname="", fullstat=False):
 """
 Public method to get a long directory listing of the connected device
 including meta data.
 @param dirname name of the directory to be listed
 @type str
-@return list containing the the directory listing with tuple entries
+@param fullstat flag indicating to return the full stat() tuple
-of the name and and a tuple of mode, size and time
+@type bool
-@rtype tuple of str
+@return list containing the directory listing with tuple entries of
+the name and and a tuple of mode, size and time (if fullstat is
+false) or the complete stat() tuple. 'None' is returned in case the
+directory doesn't exist.
+@rtype tuple of (str, tuple)
 @exception IOError raised to indicate an issue with the device
 """
 commands = [
 "import os",
 "\n".join([
 "\n".join([
 "def listdir_stat(dirname):",
 "    try:",
 "        files = os.listdir(dirname)",
 "    except OSError:",
-"        return []",
+"        return None",
 "    if dirname in ('', '/'):",
 "        return list((f, stat(f)) for f in files)",
 "    return list((f, stat(dirname + '/' + f)) for f in files)",
 ]),
 "print(listdir_stat('{0}'))".format(dirname),
 ]
 out, err = self.__execute(commands)
 if err:
 raise IOError(self.__shortError(err))
 fileslist = ast.literal_eval(out.decode("utf-8"))
-return [(f, (s[0], s[6], s[8])) for f, s in fileslist]
+if fileslist is None:
+return None
+else:
+if fullstat:
+return fileslist
+else:
+return [(f, (s[0], s[6], s[8])) for f, s in fileslist]
 def cd(self, dirname):
 """
 Public method to change the current directory on the connected device.
 "os.remove('{0}')".format(filename),
 ]
 out, err = self.__execute(commands)
 if err:
 raise IOError(self.__shortError(err))
+def rmrf(self, name, recursive=False, force=False):
+"""
+Public method to remove a file or directory recursively.
+@param name of the file or directory to remove
+@type str
+@param recursive flag indicating a recursive deletion
+@type bool
+@param force flag indicating to ignore errors
+@type bool
+@return flag indicating success
+@rtype bool
+@exception IOError raised to indicate an issue with the device
+"""
+assert name
+commands = [
+"import os"
+"\n".join([
+"def remove_file(name, recursive=False, force=False):",
+"    try:",
+"        mode = os.stat(name)[0]",
+"        if mode & 0x4000 != 0:",
+"            if recursive:",
+"                for file in os.listdir(name):",
+"                    success = remove_file(name + '/' + file,"
+" recursive, force)",
+"                    if not success and not force:",
+"                        return False",
+"                os.rmdir(name)",
+"            else:",
+"                if not force:",
+"                    return False",
+"        else:",
+"            os.remove(name)",
+"    except:",
+"        if not force:",
+"            return False",
+"    return True",
+]),
+"print(remove_file('{0}', {1}, {2}))".format(name, recursive,
+force),
+]
+out, err = self.__execute(commands)
+if err:
+raise IOError(self.__shortError(err))
+return ast.literal_eval(out.decode("utf-8"))
 def mkdir(self, dirname):
 """
 Public method to create a new directory.
 "f = open('{0}', 'rb')".format(deviceFileName),
 "r = f.read",
 "result = True",
 "\n".join([
 "while result:",
-"    result = r(32)"
+"    result = r(32)",
 "    if result:",
 "        u.write(result)",
 ]),
 "f.close()",
 ]
 # write the received bytes to the local file
 with open(hostFileName, "wb") as hostFile:
 hostFile.write(out)
 return True
-# TODO: add rsync function
 def version(self):
 """
 Public method to get the MicroPython version information of the
 connected device.
 now.tm_sec, 0))
 ]
 out, err = self.__execute(commands)
 if err:
 raise IOError(self.__shortError(err))
+def showTime(self):
+"""
+Public method to get the current time of the device.
+@return time of the device
+@rtype str
+@exception IOError raised to indicate an issue with the device
+"""
+commands = [
+"import time",
+"print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()))",
+# __IGNORE_WARNING_M601__
+]
+out, err = self.__execute(commands)
+if err:
+raise IOError(self.__shortError(err))
+return out.decode("utf-8").strip()
 class MicroPythonFileManager(QObject):
 """
 Class implementing an interface to the device file system commands with
 @signal longListFiles(result) emitted with a tuple of tuples containing the
 name, mode, size and time for each directory entry
 @signal currentDir(dirname) emitted to report the current directory of the
 device
+@signal currentDirChanged(dirname) emitted to report back a change of the
+current directory
 @signal getFileDone(deviceFile, localFile) emitted after the file was
 fetched from the connected device and written to the local file system
 @signal putFileDone(localFile, deviceFile) emitted after the file was
 copied to the connected device
 @signal deleteFileDone(deviceFile) emitted after the file has been deleted
 on the connected device
+@signal rsyncDone(localName, deviceName) emitted after the rsync operation
+has been completed
+@signal rsyncMessages(list) emitted with a list of messages
 @signal longListFilesFailed(exc) emitted with a failure message to indicate
 a failed long listing operation
 @signal currentDirFailed(exc) emitted with a failure message to indicate
 that the current directory is not available
+@signal currentDirChangeFailed(exc) emitted with a failure message to
+indicate that the current directory could not be changed
 @signal getFileFailed(exc) emitted with a failure message to indicate that
 the file could not be fetched
 @signal putFileFailed(exc) emitted with a failure message to indicate that
 the file could not be copied
 @signal deleteFileFailed(exc) emitted with a failure message to indicate
 that the file could not be deleted on the device
+@signal rsyncFailed(exc) emitted with a failure message to indicate that
+the rsync operation could not be completed
 """
 longListFiles = pyqtSignal(tuple)
 currentDir = pyqtSignal(str)
+currentDirChanged = pyqtSignal(str)
 getFileDone = pyqtSignal(str, str)
 putFileDone = pyqtSignal(str, str)
 deleteFileDone = pyqtSignal(str)
+rsyncDone = pyqtSignal(str, str)
+rsyncMessages = pyqtSignal(list)
 longListFilesFailed = pyqtSignal(str)
 currentDirFailed = pyqtSignal(str)
+currentDirChangeFailed = pyqtSignal(str)
 getFileFailed = pyqtSignal(str)
 putFileFailed = pyqtSignal(str)
 deleteFileFailed = pyqtSignal(str)
+rsyncFailed = pyqtSignal(str)
 def __init__(self, port, parent=None):
 """
 Constructor
 self.currentDir.emit(pwd)
 except Exception as exc:
 self.currentDirFailed.emit(str(exc))
 @pyqtSlot(str)
+def cd(self, dirname):
+"""
+Public slot to change the current directory of the device.
+@param dirname name of the desired current directory
+@type str
+"""
+try:
+self.__fs.cd(dirname)
+self.currentDirChanged.emit(dirname)
+except Exception as exc:
+self.currentDirChangeFailed.emit(str(exc))
+@pyqtSlot(str)
 @pyqtSlot(str, str)
 def get(self, deviceFileName, hostFileName=""):
 """
 Public slot to get a file from the connected device.
 @param deviceFileName name of the file on the connected device
 @type str
 """
 try:
 self.__fs.put(hostFileName, deviceFileName)
-self.putFileDone(hostFileName, deviceFileName)
+self.putFileDone.emit(hostFileName, deviceFileName)
 except Exception as exc:
-self.putFileFailed(str(exc))
+self.putFileFailed.emit(str(exc))
 @pyqtSlot(str)
 def delete(self, deviceFileName):
 """
 Public slot to delete a file on the device.
 """
 try:
 self.__fs.rm(deviceFileName)
 self.deleteFileDone.emit(deviceFileName)
 except Exception as exc:
-self.deleteFileFailed(str(exc))
+self.deleteFileFailed.emit(str(exc))
-##################################################################
+def __rsync(self, hostDirectory, deviceDirectory, mirror=True):
-## Utility methods below
+"""
-##################################################################
+Private method to synchronize a local directory to the device.
+@param hostDirectory name of the local directory
-def mtime2string(mtime):
+@type str
-"""
+@param deviceDirectory name of the directory on the device
-Function to convert a time value to a string representation.
+@type str
+@param mirror flag indicating to mirror the local directory to
-@param mtime time value
+the device directory
-@type int
+@type bool
-@return string representation of the given time
+@return tuple containing a list of messages and list of errors
-@rtype str
+@rtype tuple of (list of str, list of str)
 """
-return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(mtime))
+messages = []
+errors = []
-def mode2string(mode):
+if not os.isdir(hostDirectory):
-"""
+return ([], [self.tr(
-Function to convert a mode value to a string representation.
+"The given name '{0}' is not a directory or does not exist.")
+.format(hostDirectory)
-@param mode mode value
+])
-@type int
-@return string representation of the given mode value
+sourceDict = {}
-@rtype str
+sourceFiles = listdirStat(hostDirectory)
-"""
+for name, nstat in sourceFiles:
-return stat.filemode(mode)
+sourceDict[name] = nstat
+destinationDict = {}
-def decoratedName(name, mode, isDir=False):
+try:
-"""
+destinationFiles = self.__fs.lls(deviceDirectory, fullstat=True)
-Function to decorate the given name according to the given mode.
+except Exception as exc:
+return ([], [str(exc)])
-@param name file or directory name
+if destinationFiles is None:
-@type str
+# the destination directory does not exist
-@param mode mode value
+try:
-@type int
+self.__fs.mkdir(deviceDirectory)
-@param isDir flag indicating that name is a directory
+except Exception as exc:
-@type bool
+return ([], [str(exc)])
-@return decorated file or directory name
+else:
-@rtype str
+for name, nstat in destinationFiles:
-"""
+destinationDict[name] = nstat
-if stat.S_ISDIR(mode) or isDir:
-# append a '/' for directories
+destinationSet = set(destinationDict.keys())
-return name + "/"
+sourceSet = set(sourceDict.keys())
-else:
+toAdd = sourceSet - destinationSet                  # add to dev
-# no change
+toDelete = destinationSet - sourceSet               # delete from dev
-return name
+toUpdate = destinationSet.intersection(sourceSet)   # update files
+for sourceBasename in toAdd:
+# name exists in source but not in device
+sourceFilename = os.path.join(hostDirectory, sourceBasename)
+destFilename = deviceDirectory + "/" + sourceBasename
+if os.path.isfile(sourceFilename):
+try:
+self.__fs.put(sourceFilename, destFilename)
+except Exception as exc:
+messages.append(str(exc))
+if os.path.isdir(sourceFilename):
+# recurse
+msg, err = self.__rsync(sourceFilename, destFilename,
+mirror=mirror)
+messages.extend(msg)
+errors.extend(err)
+if mirror:
+for destBasename in toDelete:
+# name exists in device but not local, delete
+destFilename = deviceDirectory + "/" + destBasename
+try:
+self.__fs.rmrf(destFilename, recursive=True, force=True)
+except Exception as exc:
+# ignore errors here
+messages.append(str(exc))
+for sourceBasename in toUpdate:
+# names exist in both; do an update
+sourceStat = sourceDict[sourceBasename]
+destStat = destinationDict[sourceBasename]
+sourceFilename = os.path.join(hostDirectory, sourceBasename)
+destFilename = deviceDirectory + "/" + sourceBasename
+destMode = destStat[0]
+if os.path.isdir(sourceFilename):
+if stat.S_ISDIR(destMode):
+# both are directories => recurse
+msg, err = self.__rsync(sourceFilename, destFilename,
+mirror=mirror)
+messages.extend(msg)
+errors.extend(err)
+else:
+messages.append(self.tr(
+"Source '{0}' is a directory and destination '{1}'"
+" is a file. Ignoring it."
+).format(sourceFilename, destFilename))
+else:
+if stat.S_ISDIR(destMode):
+messages.append(self.tr(
+"Source '{0}' is a file and destination '{1}' is"
+" a directory. Ignoring it."
+).format(sourceFilename, destFilename))
+else:
+if sourceStat[8] > destStat[8]:     # mtime
+messages.append(self.tr(
+"'{0}' is newer than '{1}' - copying"
+).format(sourceFilename, destFilename))
+try:
+self.__fs.put(sourceFilename, destFilename)
+except Exception as exc:
+messages.append(str(exc))
+return messages, errors
+def rsync(self, hostDirectory, deviceDirectory, mirror=True):
+"""
+Public method to synchronize a local directory to the device.
+@param hostDirectory name of the local directory
+@type str
+@param deviceDirectory name of the directory on the device
+@type str
+@param mirror flag indicating to mirror the local directory to
+the device directory
+@type bool
+"""
+messages, errors = self.__rsync(hostDirectory, deviceDirectory,
+mirror=mirror)
+if errors:
+self.rsyncFailed.emit("\n".join(errors))
+if messages:
+self.rsyncMessages.emit(messages)
+self.rsyncDone.emit(hostDirectory, deviceDirectory)

Mercurial Repositories > eric / file comparison

comparison: eric6/MicroPython/MicroPythonFileSystem.py

eric6/MicroPython/MicroPythonFileSystem.py