eric: comparison src/eric7/MicroPython/MicroPythonFileManager.py

-:e9e7eca7efee
+:bf71ee032bb4
 import shutil
 from PyQt6.QtCore import pyqtSlot, pyqtSignal, QObject
 from .MicroPythonFileSystemUtilities import (
-mtime2string, mode2string, decoratedName, listdirStat
+mtime2string,
+mode2string,
+decoratedName,
+listdirStat,
 )
 class MicroPythonFileManager(QObject):
 """
 Class implementing an interface to the device file system commands with
 some additional sugar.
 @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
 rsync is doing
 @signal removeDirectoryDone() emitted after a directory has been deleted
 @signal createDirectoryDone() emitted after a directory was created
 @signal fsinfoDone(fsinfo) emitted after the file system information was
 obtained
 @signal error(exc) emitted with a failure message to indicate a failure
 during the most recent operation
 """
 longListFiles = pyqtSignal(tuple)
 currentDir = pyqtSignal(str)
 currentDirChanged = pyqtSignal(str)
 getFileDone = pyqtSignal(str, str)
 putFileDone = pyqtSignal(str, str)
 rsyncDone = pyqtSignal(str, str)
 rsyncProgressMessage = pyqtSignal(str)
 removeDirectoryDone = pyqtSignal()
 createDirectoryDone = pyqtSignal()
 fsinfoDone = pyqtSignal(tuple)
 error = pyqtSignal(str, str)
 def __init__(self, commandsInterface, parent=None):
 """
 Constructor
 @param commandsInterface reference to the commands interface object
 @type MicroPythonCommandsInterface
 @param parent reference to the parent object
 @type QObject
 """
 super().__init__(parent)
 self.__commandsInterface = commandsInterface
 @pyqtSlot(str)
 def lls(self, dirname, showHidden=False):
 """
 Public slot to get a long listing of the given directory.
 @param dirname name of the directory to list
 @type str
 @param showHidden flag indicating to show hidden files as well
 @type bool
 """
 try:
-filesList = self.__commandsInterface.lls(
+filesList = self.__commandsInterface.lls(dirname, showHidden=showHidden)
-dirname, showHidden=showHidden)
+result = [
-result = [(decoratedName(name, mode),
+(
-mode2string(mode),
+decoratedName(name, mode),
-str(size),
+mode2string(mode),
-mtime2string(mtime, adjustEpoch=True)) for
+str(size),
-name, (mode, size, mtime) in filesList]
+mtime2string(mtime, adjustEpoch=True),
+)
+for name, (mode, size, mtime) in filesList
+]
 self.longListFiles.emit(tuple(result))
 except Exception as exc:
 self.error.emit("lls", str(exc))
 @pyqtSlot()
 def pwd(self):
 """
 Public slot to get the current directory of the device.
 """
 try:
 pwd = self.__commandsInterface.pwd()
 self.currentDir.emit(pwd)
 except Exception as exc:
 self.error.emit("pwd", 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.__commandsInterface.cd(dirname)
 self.currentDirChanged.emit(dirname)
 except Exception as exc:
 self.error.emit("cd", 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 device
 @type str
 @param hostFileName name of the local file
 @type str
 """
 if hostFileName and os.path.isdir(hostFileName):
 # only a local directory was given
-hostFileName = os.path.join(hostFileName,
+hostFileName = os.path.join(hostFileName, os.path.basename(deviceFileName))
-os.path.basename(deviceFileName))
 try:
 self.__commandsInterface.get(deviceFileName, hostFileName)
 self.getFileDone.emit(deviceFileName, hostFileName)
 except Exception as exc:
 self.error.emit("get", str(exc))
 @pyqtSlot(str)
 @pyqtSlot(str, str)
 def put(self, hostFileName, deviceFileName=""):
 """
 Public slot to put a file onto the device.
 @param hostFileName name of the local file
 @type str
 @param deviceFileName name of the file on the connected device
 @type str
 """
 try:
 self.__commandsInterface.put(hostFileName, deviceFileName)
 self.putFileDone.emit(hostFileName, deviceFileName)
 except Exception as exc:
 self.error.emit("put", str(exc))
 @pyqtSlot(str)
 def delete(self, deviceFileName):
 """
 Public slot to delete a file on the device.
 @param deviceFileName name of the file on the connected device
 @type str
 """
 try:
 self.__commandsInterface.rm(deviceFileName)
 self.deleteFileDone.emit(deviceFileName)
 except Exception as exc:
 self.error.emit("delete", str(exc))
-def __rsync(self, hostDirectory, deviceDirectory, mirror=True,
+def __rsync(
-localDevice=False, indentLevel=0):
+self,
+hostDirectory,
+deviceDirectory,
+mirror=True,
+localDevice=False,
+indentLevel=0,
+):
 """
 Private 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
 @return list of errors
 @rtype list of str
 """
 indent = 4 * "&nbsp;"
 errors = []
 if not os.path.isdir(hostDirectory):
-return [self.tr(
+return [
-"The given name '{0}' is not a directory or does not exist.")
+self.tr(
-.format(hostDirectory)
+"The given name '{0}' is not a directory or does not exist."
+).format(hostDirectory)
 ]
 indentStr = indentLevel * indent
 self.rsyncProgressMessage.emit(
-self.tr("{1}Synchronizing <b>{0}</b>.")
+self.tr("{1}Synchronizing <b>{0}</b>.").format(deviceDirectory, indentStr)
-.format(deviceDirectory, indentStr)
 )
 doneMessage = self.tr("{1}Done synchronizing <b>{0}</b>.").format(
-deviceDirectory, indentStr)
+deviceDirectory, indentStr
+)
 sourceDict = {}
 sourceFiles = listdirStat(hostDirectory)
 for name, nstat in sourceFiles:
 sourceDict[name] = nstat
 destinationDict = {}
 if localDevice:
 if not os.path.isdir(deviceDirectory):
 # simply copy destination to source
 shutil.copytree(hostDirectory, deviceDirectory)
 for name, nstat in destinationFiles:
 destinationDict[name] = nstat
 else:
 try:
 destinationFiles = self.__commandsInterface.lls(
-deviceDirectory, fullstat=True)
+deviceDirectory, fullstat=True
+)
 except Exception as exc:
 return [str(exc)]
 if destinationFiles is None:
 # the destination directory does not exist
 try:
 except Exception as exc:
 return [str(exc)]
 else:
 for name, nstat in destinationFiles:
 destinationDict[name] = nstat
 destinationSet = set(destinationDict.keys())
 sourceSet = set(sourceDict.keys())
-toAdd = sourceSet - destinationSet                  # add to dev
+toAdd = sourceSet - destinationSet  # add to dev
-toDelete = destinationSet - sourceSet               # delete from dev
+toDelete = destinationSet - sourceSet  # delete from dev
-toUpdate = destinationSet.intersection(sourceSet)   # update files
+toUpdate = destinationSet.intersection(sourceSet)  # update files
 indentStr = (indentLevel + 1) * indent
 if localDevice:
 for sourceBasename in toAdd:
 # name exists in source but not in device
 sourceFilename = os.path.join(hostDirectory, sourceBasename)
 destFilename = os.path.join(deviceDirectory, sourceBasename)
 self.rsyncProgressMessage.emit(
-self.tr("{1}Adding <b>{0}</b>...")
+self.tr("{1}Adding <b>{0}</b>...").format(destFilename, indentStr)
-.format(destFilename, indentStr))
+)
 if os.path.isfile(sourceFilename):
 shutil.copy2(sourceFilename, destFilename)
 elif os.path.isdir(sourceFilename):
 # recurse
-errs = self.__rsync(sourceFilename, destFilename,
+errs = self.__rsync(
-mirror=mirror, localDevice=localDevice,
+sourceFilename,
-indentLevel=indentLevel + 1)
+destFilename,
+mirror=mirror,
+localDevice=localDevice,
+indentLevel=indentLevel + 1,
+)
 # just note issues but ignore them otherwise
 errors.extend(errs)
 if mirror:
 for destBasename in toDelete:
 # name exists in device but not local, delete
 destFilename = os.path.join(deviceDirectory, destBasename)
 if os.path.isdir(destFilename):
 shutil.rmtree(destFilename, ignore_errors=True)
 elif os.path.isfile(destFilename):
 os.remove(destFilename)
 for sourceBasename in toUpdate:
 # names exist in both; do an update
 sourceStat = sourceDict[sourceBasename]
 destStat = destinationDict[sourceBasename]
 sourceFilename = os.path.join(hostDirectory, sourceBasename)
 destFilename = os.path.join(deviceDirectory, sourceBasename)
 destMode = destStat[0]
 if os.path.isdir(sourceFilename):
 if os.path.isdir(destFilename):
 # both are directories => recurs
-errs = self.__rsync(sourceFilename, destFilename,
+errs = self.__rsync(
-mirror=mirror,
+sourceFilename,
-localDevice=localDevice,
+destFilename,
-indentLevel=indentLevel + 1)
+mirror=mirror,
+localDevice=localDevice,
+indentLevel=indentLevel + 1,
+)
 # just note issues but ignore them otherwise
 errors.extend(errs)
 else:
 self.rsyncProgressMessage.emit(
-self.tr("Source <b>{0}</b> is a directory and"
+self.tr(
-" destination <b>{1}</b> is a file."
+"Source <b>{0}</b> is a directory and"
-" Ignoring it.")
+" destination <b>{1}</b> is a file."
-.format(sourceFilename, destFilename)
+" Ignoring it."
+).format(sourceFilename, destFilename)
 )
 else:
 if os.path.isdir(destFilename):
 self.rsyncProgressMessage.emit(
-self.tr("Source <b>{0}</b> is a file and"
+self.tr(
-" destination <b>{1}</b> is a directory."
+"Source <b>{0}</b> is a file and"
-" Ignoring it.")
+" destination <b>{1}</b> is a directory."
-.format(sourceFilename, destFilename)
+" Ignoring it."
+).format(sourceFilename, destFilename)
 )
 else:
-if sourceStat[8] > destStat[8]:     # mtime
+if sourceStat[8] > destStat[8]:  # mtime
 self.rsyncProgressMessage.emit(
-self.tr("Updating <b>{0}</b>...")
+self.tr("Updating <b>{0}</b>...").format(destFilename)
-.format(destFilename)
 )
 shutil.copy2(sourceFilename, destFilename)
 else:
 for sourceBasename in toAdd:
 # name exists in source but not in device
 sourceFilename = os.path.join(hostDirectory, sourceBasename)
 destFilename = (
 "/" + sourceBasename
-if deviceDirectory == "/" else
+if deviceDirectory == "/"
-deviceDirectory + "/" + sourceBasename
+else deviceDirectory + "/" + sourceBasename
 )
 self.rsyncProgressMessage.emit(
-self.tr("{1}Adding <b>{0}</b>...")
+self.tr("{1}Adding <b>{0}</b>...").format(destFilename, indentStr)
-.format(destFilename, indentStr))
+)
 if os.path.isfile(sourceFilename):
 try:
-self.__commandsInterface.put(sourceFilename,
+self.__commandsInterface.put(sourceFilename, destFilename)
-destFilename)
 except Exception as exc:
 # just note issues but ignore them otherwise
 errors.append(str(exc))
 elif os.path.isdir(sourceFilename):
 # recurse
-errs = self.__rsync(sourceFilename, destFilename,
+errs = self.__rsync(
-mirror=mirror,
+sourceFilename,
-indentLevel=indentLevel + 1)
+destFilename,
+mirror=mirror,
+indentLevel=indentLevel + 1,
+)
 # just note issues but ignore them otherwise
 errors.extend(errs)
 if mirror:
 for destBasename in toDelete:
 # name exists in device but not local, delete
 destFilename = (
 "/" + sourceBasename
-if deviceDirectory == "/" else
+if deviceDirectory == "/"
-deviceDirectory + "/" + destBasename
+else deviceDirectory + "/" + destBasename
 )
 self.rsyncProgressMessage.emit(
-self.tr("{1}Removing <b>{0}</b>...")
+self.tr("{1}Removing <b>{0}</b>...").format(
-.format(destFilename, indentStr))
+destFilename, indentStr
+)
+)
 try:
-self.__commandsInterface.rmrf(destFilename,
+self.__commandsInterface.rmrf(
-recursive=True,
+destFilename, recursive=True, force=True
-force=True)
+)
 except Exception as exc:
 # just note issues but ignore them otherwise
 errors.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 = (
 "/" + sourceBasename
-if deviceDirectory == "/" else
+if deviceDirectory == "/"
-deviceDirectory + "/" + sourceBasename
+else deviceDirectory + "/" + sourceBasename
 )
 destMode = destStat[0]
 if os.path.isdir(sourceFilename):
 if stat.S_ISDIR(destMode):
 # both are directories => recurs
-errs = self.__rsync(sourceFilename, destFilename,
+errs = self.__rsync(
-mirror=mirror,
+sourceFilename,
-indentLevel=indentLevel + 1)
+destFilename,
+mirror=mirror,
+indentLevel=indentLevel + 1,
+)
 # just note issues but ignore them otherwise
 errors.extend(errs)
 else:
 self.rsyncProgressMessage.emit(
-self.tr("Source <b>{0}</b> is a directory and"
+self.tr(
-" destination <b>{1}</b> is a file."
+"Source <b>{0}</b> is a directory and"
-" Ignoring it.")
+" destination <b>{1}</b> is a file."
-.format(sourceFilename, destFilename)
+" Ignoring it."
+).format(sourceFilename, destFilename)
 )
 else:
 if stat.S_ISDIR(destMode):
 self.rsyncProgressMessage.emit(
-self.tr("Source <b>{0}</b> is a file and"
+self.tr(
-" destination <b>{1}</b> is a directory."
+"Source <b>{0}</b> is a file and"
-" Ignoring it.")
+" destination <b>{1}</b> is a directory."
-.format(sourceFilename, destFilename)
+" Ignoring it."
+).format(sourceFilename, destFilename)
 )
 else:
-if sourceStat[8] > destStat[8]:     # mtime
+if sourceStat[8] > destStat[8]:  # mtime
 self.rsyncProgressMessage.emit(
-self.tr("{1}Updating <b>{0}</b>...")
+self.tr("{1}Updating <b>{0}</b>...").format(
-.format(destFilename, indentStr)
+destFilename, indentStr
+)
 )
 try:
-self.__commandsInterface.put(sourceFilename,
+self.__commandsInterface.put(
-destFilename)
+sourceFilename, destFilename
+)
 except Exception as exc:
 errors.append(str(exc))
 self.rsyncProgressMessage.emit(doneMessage)
 return errors
 @pyqtSlot(str, str)
 @pyqtSlot(str, str, bool)
 @pyqtSlot(str, str, bool, bool)
-def rsync(self, hostDirectory, deviceDirectory, mirror=True,
+def rsync(self, hostDirectory, deviceDirectory, mirror=True, localDevice=False):
-localDevice=False):
 """
 Public slot 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
 @param localDevice flag indicating device access via local file system
 @type bool
 """
-errors = self.__rsync(hostDirectory, deviceDirectory, mirror=mirror,
+errors = self.__rsync(
-localDevice=localDevice)
+hostDirectory, deviceDirectory, mirror=mirror, localDevice=localDevice
+)
 if errors:
 self.error.emit("rsync", "\n".join(errors))
 self.rsyncDone.emit(hostDirectory, deviceDirectory)
 @pyqtSlot(str)
 def mkdir(self, dirname):
 """
 Public slot to create a new directory.
 @param dirname name of the directory to create
 @type str
 """
 try:
 self.__commandsInterface.mkdir(dirname)
 self.createDirectoryDone.emit()
 except Exception as exc:
 self.error.emit("mkdir", str(exc))
 @pyqtSlot(str)
 @pyqtSlot(str, bool)
 def rmdir(self, dirname, recursive=False):
 """
 Public slot to (recursively) remove a directory.
 @param dirname name of the directory to be removed
 @type str
 @param recursive flag indicating a recursive removal
 @type bool
 """
 try:
 if recursive:
-self.__commandsInterface.rmrf(dirname, recursive=True,
+self.__commandsInterface.rmrf(dirname, recursive=True, force=True)
-force=True)
 else:
 self.__commandsInterface.rmdir(dirname)
 self.removeDirectoryDone.emit()
 except Exception as exc:
 self.error.emit("rmdir", str(exc))
 def fileSystemInfo(self):
 """
 Public method to obtain information about the currently mounted file
 systems.
 """

Mercurial Repositories > eric / file comparison

comparison: src/eric7/MicroPython/MicroPythonFileManager.py

src/eric7/MicroPython/MicroPythonFileManager.py