--- a/src/eric7/MicroPython/MicroPythonFileManager.py Wed Jul 13 11:16:20 2022 +0200
+++ b/src/eric7/MicroPython/MicroPythonFileManager.py Wed Jul 13 14:55:47 2022 +0200
@@ -14,7 +14,10 @@
from PyQt6.QtCore import pyqtSlot, pyqtSignal, QObject
from .MicroPythonFileSystemUtilities import (
- mtime2string, mode2string, decoratedName, listdirStat
+ mtime2string,
+ mode2string,
+ decoratedName,
+ listdirStat,
)
@@ -22,7 +25,7 @@
"""
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
@@ -43,10 +46,11 @@
@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)
@@ -58,44 +62,47 @@
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(
- dirname, showHidden=showHidden)
- result = [(decoratedName(name, mode),
- mode2string(mode),
- str(size),
- mtime2string(mtime, adjustEpoch=True)) for
- name, (mode, size, mtime) in filesList]
+ filesList = self.__commandsInterface.lls(dirname, showHidden=showHidden)
+ result = [
+ (
+ decoratedName(name, mode),
+ mode2string(mode),
+ str(size),
+ 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):
"""
@@ -106,12 +113,12 @@
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
"""
@@ -120,13 +127,13 @@
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
@@ -134,20 +141,19 @@
"""
if hostFileName and os.path.isdir(hostFileName):
# only a local directory was given
- hostFileName = os.path.join(hostFileName,
- os.path.basename(deviceFileName))
+ hostFileName = os.path.join(hostFileName, 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
@@ -158,12 +164,12 @@
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
"""
@@ -172,12 +178,18 @@
self.deleteFileDone.emit(deviceFileName)
except Exception as exc:
self.error.emit("delete", str(exc))
-
- def __rsync(self, hostDirectory, deviceDirectory, mirror=True,
- localDevice=False, indentLevel=0):
+
+ def __rsync(
+ 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
@@ -194,27 +206,28 @@
"""
indent = 4 * " "
errors = []
-
+
if not os.path.isdir(hostDirectory):
- return [self.tr(
- "The given name '{0}' is not a directory or does not exist.")
- .format(hostDirectory)
+ return [
+ self.tr(
+ "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>.")
- .format(deviceDirectory, indentStr)
+ self.tr("{1}Synchronizing <b>{0}</b>.").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):
@@ -229,7 +242,8 @@
else:
try:
destinationFiles = self.__commandsInterface.lls(
- deviceDirectory, fullstat=True)
+ deviceDirectory, fullstat=True
+ )
except Exception as exc:
return [str(exc)]
if destinationFiles is None:
@@ -241,32 +255,36 @@
else:
for name, nstat in destinationFiles:
destinationDict[name] = nstat
-
+
destinationSet = set(destinationDict.keys())
sourceSet = set(sourceDict.keys())
- toAdd = sourceSet - destinationSet # add to dev
- toDelete = destinationSet - sourceSet # delete from dev
- toUpdate = destinationSet.intersection(sourceSet) # update files
+ toAdd = sourceSet - destinationSet # add to dev
+ toDelete = destinationSet - sourceSet # delete from dev
+ 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>...")
- .format(destFilename, indentStr))
+ self.tr("{1}Adding <b>{0}</b>...").format(destFilename, indentStr)
+ )
if os.path.isfile(sourceFilename):
shutil.copy2(sourceFilename, destFilename)
elif os.path.isdir(sourceFilename):
# recurse
- errs = self.__rsync(sourceFilename, destFilename,
- mirror=mirror, localDevice=localDevice,
- indentLevel=indentLevel + 1)
+ errs = self.__rsync(
+ sourceFilename,
+ 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
@@ -275,7 +293,7 @@
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]
@@ -286,32 +304,36 @@
if os.path.isdir(sourceFilename):
if os.path.isdir(destFilename):
# both are directories => recurs
- errs = self.__rsync(sourceFilename, destFilename,
- mirror=mirror,
- localDevice=localDevice,
- indentLevel=indentLevel + 1)
+ errs = self.__rsync(
+ sourceFilename,
+ destFilename,
+ 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"
- " destination <b>{1}</b> is a file."
- " Ignoring it.")
- .format(sourceFilename, destFilename)
+ self.tr(
+ "Source <b>{0}</b> is a directory and"
+ " destination <b>{1}</b> is a file."
+ " Ignoring it."
+ ).format(sourceFilename, destFilename)
)
else:
if os.path.isdir(destFilename):
self.rsyncProgressMessage.emit(
- self.tr("Source <b>{0}</b> is a file and"
- " destination <b>{1}</b> is a directory."
- " Ignoring it.")
- .format(sourceFilename, destFilename)
+ self.tr(
+ "Source <b>{0}</b> is a file and"
+ " destination <b>{1}</b> is a directory."
+ " 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>...")
- .format(destFilename)
+ self.tr("Updating <b>{0}</b>...").format(destFilename)
)
shutil.copy2(sourceFilename, destFilename)
else:
@@ -320,46 +342,50 @@
sourceFilename = os.path.join(hostDirectory, sourceBasename)
destFilename = (
"/" + sourceBasename
- if deviceDirectory == "/" else
- deviceDirectory + "/" + sourceBasename
+ if deviceDirectory == "/"
+ else deviceDirectory + "/" + sourceBasename
)
self.rsyncProgressMessage.emit(
- self.tr("{1}Adding <b>{0}</b>...")
- .format(destFilename, indentStr))
+ self.tr("{1}Adding <b>{0}</b>...").format(destFilename, indentStr)
+ )
if os.path.isfile(sourceFilename):
try:
- self.__commandsInterface.put(sourceFilename,
- destFilename)
+ self.__commandsInterface.put(sourceFilename, 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,
- mirror=mirror,
- indentLevel=indentLevel + 1)
+ errs = self.__rsync(
+ sourceFilename,
+ 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
- deviceDirectory + "/" + destBasename
+ if deviceDirectory == "/"
+ else deviceDirectory + "/" + destBasename
)
self.rsyncProgressMessage.emit(
- self.tr("{1}Removing <b>{0}</b>...")
- .format(destFilename, indentStr))
+ self.tr("{1}Removing <b>{0}</b>...").format(
+ destFilename, indentStr
+ )
+ )
try:
- self.__commandsInterface.rmrf(destFilename,
- recursive=True,
- force=True)
+ self.__commandsInterface.rmrf(
+ destFilename, recursive=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]
@@ -367,57 +393,63 @@
sourceFilename = os.path.join(hostDirectory, sourceBasename)
destFilename = (
"/" + sourceBasename
- if deviceDirectory == "/" else
- deviceDirectory + "/" + sourceBasename
+ if deviceDirectory == "/"
+ 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,
- mirror=mirror,
- indentLevel=indentLevel + 1)
+ errs = self.__rsync(
+ sourceFilename,
+ 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"
- " destination <b>{1}</b> is a file."
- " Ignoring it.")
- .format(sourceFilename, destFilename)
+ self.tr(
+ "Source <b>{0}</b> is a directory and"
+ " destination <b>{1}</b> is a file."
+ " Ignoring it."
+ ).format(sourceFilename, destFilename)
)
else:
if stat.S_ISDIR(destMode):
self.rsyncProgressMessage.emit(
- self.tr("Source <b>{0}</b> is a file and"
- " destination <b>{1}</b> is a directory."
- " Ignoring it.")
- .format(sourceFilename, destFilename)
+ self.tr(
+ "Source <b>{0}</b> is a file and"
+ " destination <b>{1}</b> is a directory."
+ " 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>...")
- .format(destFilename, indentStr)
+ self.tr("{1}Updating <b>{0}</b>...").format(
+ destFilename, indentStr
+ )
)
try:
- self.__commandsInterface.put(sourceFilename,
- destFilename)
+ self.__commandsInterface.put(
+ 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,
- localDevice=False):
+ def rsync(self, hostDirectory, deviceDirectory, mirror=True, 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
@@ -428,18 +460,19 @@
@param localDevice flag indicating device access via local file system
@type bool
"""
- errors = self.__rsync(hostDirectory, deviceDirectory, mirror=mirror,
- localDevice=localDevice)
+ errors = self.__rsync(
+ 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
"""
@@ -448,13 +481,13 @@
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
@@ -462,14 +495,13 @@
"""
try:
if recursive:
- self.__commandsInterface.rmrf(dirname, recursive=True,
- force=True)
+ self.__commandsInterface.rmrf(dirname, recursive=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
