eric: comparison eric6/MicroPython/MicroPythonFileSystem.py

-:2ca7fb61a82f
+:e29b0ee86b29
 from .MicroPythonSerialPort import MicroPythonSerialPort
 from .MicroPythonFileSystemUtilities import (
 mtime2string, mode2string, decoratedName, listdirStat
 )
+import Preferences
 class MicroPythonFileSystem(QObject):
 """
 Class implementing some file system commands for MicroPython.
 out = out.replace(b"\r", b"\n")
 with open(hostFileName, "wb") as hostFile:
 hostFile.write(out)
 return True
+def fileSystemInfo(self):
+"""
+Public method to obtain information about the currently mounted file
+systems.
+@return tuple of tuples containing the file system name, the total
+size, the used size and the free size
+@rtype tuple of tuples of (str, int, int, int)
+"""
+commands = [
+"import os",
+"\n".join([
+"def fsinfo():",
+"    infolist = []",
+"    fsnames = os.listdir('/')",
+"    for fs in fsnames:",
+"        fs = '/' + fs",
+"        infolist.append((fs, os.statvfs(fs)))",
+"    return infolist",
+]),
+"print(fsinfo())",
+]
+out, err = self.__execute(commands)
+if err:
+raise IOError(self.__shortError(err))
+infolist = ast.literal_eval(out.decode("utf-8"))
+if infolist is None:
+return None
+else:
+filesystemInfos = []
+for fs, info in infolist:
+totalSize = info[2] * info[1]
+freeSize = info[4] * info[1]
+usedSize = totalSize - freeSize
+filesystemInfos.append((fs, totalSize, usedSize, freeSize))
+return tuple(filesystemInfos)
+##################################################################
+## non-filesystem related methods below
+##################################################################
 def version(self):
 """
 Public method to get the MicroPython version information of the
 connected device.
 @return dictionary containing the version information
 @rtype dict
-@exception ValueError raised to indicate that the device might not be
+@exception IOError raised to indicate an issue with the device
-running MicroPython or there was an issue parsing the output
 """
 commands = [
 "import os",
 "print(os.uname())",
 ]
-try:
+out, err = self.__execute(commands)
-out, err = self.__execute(commands)
+if err:
-if err:
+raise IOError(self.__shortError(err))
-raise ValueError(self.__shortError(err))
-except ValueError:
-# just re-raise it
-raise
-except Exception:
-# Raise a value error to indicate being unable to find something
-# on the device that will return parseable information about the
-# version. It doesn't matter what the error is, it just needs to
-# report a failure with the expected ValueError exception.
-raise ValueError("Unable to determine version information.")
 rawOutput = out.decode("utf-8").strip()
 rawOutput = rawOutput[1:-1]
 items = rawOutput.split(",")
 result = {}
 for item in items:
 key, value = item.strip().split("=")
 result[key.strip()] = value.strip()[1:-1]
 return result
+def getImplementation(self):
+"""
+Public method to get some implementation information of the connected
+device.
+@return dictionary containing the implementation information
+@rtype dict
+@exception IOError raised to indicate an issue with the device
+"""
+commands = [
+"import sys",
+"res = {}",
+"\n".join([
+"try:",
+"    res['name'] = sys.implementation.name",
+"except AttributeError:",
+"    res['name'] = 'unknown'",
+]),
+"\n".join([
+"try:",
+"    res['version'] = '.'.join((str(i) for i in"
+" sys.implementation.version))",
+"except AttributeError:",
+"    res['version'] = 'unknown'",
+]),
+"print(res)",
+]
+out, err = self.__execute(commands)
+if err:
+raise IOError(self.__shortError(err))
+return ast.literal_eval(out.decode("utf-8"))
 def syncTime(self):
 """
 Public method to set the time of the connected device to the local
 computer's time.
 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)
 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, port, parent=None):
 """
 @type QObject
 """
 super(MicroPythonFileManager, self).__init__(parent)
 self.__serialPort = port
-self.__serial = MicroPythonSerialPort(parent=self)
+self.__serial = MicroPythonSerialPort(
+timeout=Preferences.getMicroPython("SerialTimeout"),
+parent=self)
 self.__fs = MicroPythonFileSystem(parent=self)
 @pyqtSlot()
-def connect(self):
+def connectToDevice(self):
 """
 Public slot to start the manager.
 """
 self.__serial.openSerialLink(self.__serialPort)
 self.__fs.setSerial(self.__serial)
 @pyqtSlot()
-def disconnect(self):
+def disconnectFromDevice(self):
 """
 Public slot to stop the thread.
 """
 self.__serial.closeSerialLink()
+@pyqtSlot()
+def handlePreferencesChanged(self):
+"""
+Public slot to handle a change of the preferences.
+"""
+self.__serial.setTimeout(Preferences.getMicroPython("SerialTimeout"))
 @pyqtSlot(str)
 def lls(self, dirname):
 """
 Public slot to get a long listing of the given directory.
 self.__fs.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.__fs.fileSystemInfo()
+self.fsinfoDone.emit(fsinfo)
+except Exception as exc:
+self.error.emit("fileSystemInfo", str(exc))
 ##################################################################
 ## some non-filesystem related methods below
 ##################################################################
 @pyqtSlot()
 try:
 versionInfo = self.__fs.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.__fs.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))

Mercurial Repositories > eric / file comparison

comparison: eric6/MicroPython/MicroPythonFileSystem.py

eric6/MicroPython/MicroPythonFileSystem.py