Mercurial Repositories > eric / file revision

# -*- coding: utf-8 -*-

# Copyright (c) 2019 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Module implementing some file system commands for MicroPython.
"""

from __future__ import unicode_literals

import os
import stat

from PyQt5.QtCore import pyqtSlot, pyqtSignal, QObject

from .MicroPythonFileSystemUtilities import (
    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
        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 rsyncProgressMessage(msg) emitted to send a message about what
        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 synchTimeDone() emitted after the time was synchronizde to the
        device
    @signal showTimeDone(dateTime) emitted after the date and time was fetched
        from the connected device
    @signal showVersionDone(versionInfo) emitted after the version information
        was fetched from the connected device
    @signal showImplementationDone(name,version) emitted after the
        implementation information has been 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)
    deleteFileDone = pyqtSignal(str)
    rsyncDone = pyqtSignal(str, str)
    rsyncProgressMessage = pyqtSignal(str)
    removeDirectoryDone = pyqtSignal()
    createDirectoryDone = pyqtSignal()
    fsinfoDone = pyqtSignal(tuple)
    
    synchTimeDone = pyqtSignal()
    showTimeDone = pyqtSignal(str)
    showVersionDone = pyqtSignal(dict)
    showImplementationDone = pyqtSignal(str, str)
    
    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(MicroPythonFileManager, self).__init__(parent)
        
        self.__commandsInterface = commandsInterface
    
    @pyqtSlot(str)
    def lls(self, dirname):
        """
        Public slot to get a long listing of the given directory.
        
        @param dirname name of the directory to list
        @type str
        """
        try:
            filesList = self.__commandsInterface.lls(dirname)
            result = [(decoratedName(name, mode),
                       mode2string(mode),
                       str(size),
                       mtime2string(mtime)) 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,
                                        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):
        """
        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
            the device directory
        @type bool
        @return list of errors
        @rtype list of str
        """
        errors = []
        
        if not os.path.isdir(hostDirectory):
            return [self.tr(
                "The given name '{0}' is not a directory or does not exist.")
                .format(hostDirectory)
            ]
        
        self.rsyncProgressMessage.emit(
            self.tr("Synchronizing <b>{0}</b>.").format(deviceDirectory)
        )
        
        sourceDict = {}
        sourceFiles = listdirStat(hostDirectory)
        for name, nstat in sourceFiles:
            sourceDict[name] = nstat
        
        destinationDict = {}
        try:
            destinationFiles = self.__commandsInterface.lls(deviceDirectory,
                                                            fullstat=True)
        except Exception as exc:
            return [str(exc)]
        if destinationFiles is None:
            # the destination directory does not exist
            try:
                self.__commandsInterface.mkdir(deviceDirectory)
            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
        toDelete = destinationSet - sourceSet               # delete from dev
        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
            self.rsyncProgressMessage.emit(
                self.tr("Adding <b>{0}</b>...").format(destFilename))
            if os.path.isfile(sourceFilename):
                try:
                    self.__commandsInterface.put(sourceFilename, destFilename)
                except Exception as exc:
                    # just note issues but ignore them otherwise
                    errors.append(str(exc))
            if os.path.isdir(sourceFilename):
                # recurse
                errs = self.__rsync(sourceFilename, destFilename,
                                    mirror=mirror)
                # 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 = deviceDirectory + "/" + destBasename
                self.rsyncProgressMessage.emit(
                    self.tr("Removing <b>{0}</b>...").format(destFilename))
                try:
                    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]
            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 => recurs
                    errs = self.__rsync(sourceFilename, destFilename,
                                        mirror=mirror)
                    # 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)
                    )
            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)
                    )
                else:
                    if sourceStat[8] > destStat[8]:     # mtime
                        self.rsyncProgressMessage.emit(
                            self.tr("Updating <b>{0}</b>...")
                            .format(destFilename)
                        )
                        try:
                            self.__commandsInterface.put(sourceFilename,
                                                         destFilename)
                        except Exception as exc:
                            errors.append(str(exc))
        
        self.rsyncProgressMessage.emit(
            self.tr("Done synchronizing <b>{0}</b>.").format(deviceDirectory)
        )
        
        return errors
    
    @pyqtSlot(str, str)
    @pyqtSlot(str, str, bool)
    def rsync(self, hostDirectory, deviceDirectory, mirror=True):
        """
        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
        """
        errors = self.__rsync(hostDirectory, deviceDirectory, mirror=mirror)
        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,
                                              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.
        """
        try:
            fsinfo = self.__commandsInterface.fileSystemInfo()
            self.fsinfoDone.emit(fsinfo)
        except Exception as exc:
            self.error.emit("fileSystemInfo", str(exc))
    
    ##################################################################
    ## some non-filesystem related methods below
    ##################################################################
    
    @pyqtSlot()
    def synchronizeTime(self):
        """
        Public slot to set the time of the connected device to the local
        computer's time.
        """
        try:
            self.__commandsInterface.syncTime()
            self.synchTimeDone.emit()
        except Exception as exc:
            self.error.emit("rmdir", str(exc))
    
    @pyqtSlot()
    def showTime(self):
        """
        Public slot to get the current date and time of the device.
        """
        try:
            dt = self.__commandsInterface.showTime()
            self.showTimeDone.emit(dt)
        except Exception as exc:
            self.error.emit("showTime", str(exc))
    
    @pyqtSlot()
    def showVersion(self):
        """
        Public slot to get the version info for the MicroPython run by the
        connected device.
        """
        try:
            versionInfo = self.__commandsInterface.version()
            self.showVersionDone.emit(versionInfo)
        except Exception as exc:
            self.error.emit("showVersion", str(exc))
    
    @pyqtSlot()
    def showImplementation(self):
        """
        Public slot to obtain some implementation related information.
        """
        try:
            impInfo = self.__commandsInterface.getImplementation()
            if impInfo["name"] == "micropython":
                name = "MicroPython"
            elif impInfo["name"] == "circuitpython":
                name = "CircuitPython"
            elif impInfo["name"] == "unknown":
                name = self.tr("unknown")
            else:
                name = impInfo["name"]
            if impInfo["version"] == "unknown":
                version = self.tr("unknown")
            else:
                version = impInfo["version"]
            self.showImplementationDone.emit(name, version)
        except Exception as exc:
            self.error.emit("showVersion", str(exc))