eric: comparison eric6/MicroPython/MicroPythonFileSystem.py

-:a09a30251d4e
+:3368ce0e7879
 Module implementing some file system commands for MicroPython.
 """
 from __future__ import unicode_literals
-from PyQt5.QtCore import QObject
+import ast
+import time
+import stat
+from PyQt5.QtCore import QObject, QThread
 class MicroPythonFileSystem(QObject):
 """
 Class implementing some file system commands for MicroPython.
 <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>mkdir: create a new directory</li>
+<li>rmdir: remove an empty directory</li>
 </ul>
 """
 def __init__(self, parent=None):
 """
 Constructor
 @param parent reference to the parent object
 @type QObject
 """
 super(MicroPythonFileSystem, self).__init__(parent)
-def ls(self):
+self.__serial = None
+def setSerial(self, serial):
+"""
+Public method to set the serial port to be used.
+Note: The serial port should be initialized and open already.
+@param serial open serial port
+@type MicroPythonSerialPort
+"""
+self.__serial = serial
+def __rawOn(self):
+"""
+Private method to switch the connected device to 'raw' mode.
+Note: switching to raw mode is done with synchroneous writes.
+"""
+if not self.__serial:
+return
+rawReplMessage = b"raw REPL; CTRL-B to exit\r\n"
+softRebootMessage = b"soft reboot\r\n"
+self.__serial.write(b"\x02")        # end raw mode if required
+self.__serial.waitForBytesWritten()
+for _i in range(3):
+# CTRL-C three times to break out of loops
+self.__serial.write(b"\r\x03")
+self.__serial.waitForBytesWritten()
+QThread.msleep(10)
+self.__serial.readAll()             # read all data and discard it
+self.__serial.write(b"\r\x01")      # send CTRL-A to enter raw mode
+self.__serial.readUntil(rawReplMessage)
+self.__serial.write(b"\x04")        # send CTRL-D to soft reset
+self.__serial.readUntil(softRebootMessage)
+# some MicroPython devices seem to need to be convinced in some
+# special way
+data = self.__serial.readUntil(rawReplMessage)
+if not data.endswith(rawReplMessage):
+self.__serial.write(b"\r\x01")  # send CTRL-A again
+self.__serial.readUntil(rawReplMessage)
+self.__serial.readAll()             # read all data and discard it
+def __rawOff(self):
+"""
+Private method to switch 'raw' mode off.
+"""
+if self.__serial:
+self.__serial.write(b"\x02")    # send CTRL-B to cancel raw mode
+def __execute(self, commands):
+"""
+Private method to send commands to the connected device and return the
+result.
+If no serial connection is available, empty results will be returned.
+@param commands list of commands to be executed
+@type str
+@return tuple containing stdout and stderr output of the device
+@rtype tuple of (bytes, bytes)
+"""
+if not self.__serial:
+return b"", b""
+result = bytearray()
+err = b""
+self.__rawOn()
+QThread.msleep(10)
+for command in commands:
+if command:
+commandBytes = command.encode("utf-8")
+self.__serial.write(commandBytes + b"\x04")
+# read until prompt
+response = self.__serial.readUntil(b"\x04>")
+# split stdout, stderr
+out, err = response[2:-2].split(b"\x04")
+result += out
+if err:
+return b"", err
+QThread.msleep(10)
+self.__rawOff()
+return bytes(result), err
+def __shortError(self, error):
+"""
+Private method to create a shortened error message.
+@param error verbose error message
+@type bytes
+@return shortened error message
+@rtype str
+"""
+if error:
+decodedError = error.decode("utf-8")
+try:
+return decodedError.split["\r\n"][-2]
+except Exception:
+return decodedError
+return self.tr("Detected an error without indications.")
+##################################################################
+## Methods below implement the file system commands
+##################################################################
+def ls(self, dirname=""):
 """
 Public method to get a directory listing of the connected device.
+@param dirname name of the directory to be listed
+@type str
 @return tuple containg the directory listing
 @rtype tuple of str
-"""
+@exception IOError raised to indicate an issue with the device
-# TODO: not implemented yet
+"""
+commands = [
-def lls(self):
+"import os",
+"print(os.listdir('{0}'))".format(dirname),
+]
+out, err = self.__execute(commands)
+if err:
+raise IOError(self.__shortError(err))
+return ast.literal_eval(out.decode("utf-8"))
+def lls(self, dirname=""):
 """
 Public method to get a long directory listing of the connected device
 including meta data.
-@return tuple containg the the directory listing with tuple entries
+@param dirname name of the directory to be listed
-containing the name, size, time and mode
+@type str
+@return list containing the the directory listing with tuple entries
+of the name and and a tuple of mode, size and time
 @rtype tuple of str
-"""
+@exception IOError raised to indicate an issue with the device
-# TODO: not implemented yet
+"""
+commands = [
-def cd(self, path):
+"import os",
+"\n".join([
+"def stat(filename):",
+"    try:",
+"        rstat = os.lstat(filename)",
+"    except:",
+"        rstat = os.stat(filename)",
+"    return tuple(rstat)",
+]),
+"\n".join([
+"def listdir_stat(dirname):",
+"    try:",
+"        files = os.listdir(dirname)",
+"    except OSError:",
+"        return []",
+"    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]
+def cd(self, dirname):
 """
 Public method to change the current directory on the connected device.
-@param path directory to change to
+@param dirname directory to change to
 @type str
-"""
+@exception IOError raised to indicate an issue with the device
-# TODO: not implemented yet
+"""
+assert dirname
+commands = [
+"import os",
+"os.chdir('{0}')".format(dirname),
+]
+out, err = self.__execute(commands)
+if err:
+raise IOError(self.__shortError(err))
 def pwd(self):
 """
 Public method to get the current directory of the connected device.
 @return current directory
 @rtype str
-"""
+@exception IOError raised to indicate an issue with the device
-# TODO: not implemented yet
+"""
+commands = [
+"import os",
+"print(os.getcwd())",
+]
+out, err = self.__execute(commands)
+if err:
+raise IOError(self.__shortError(err))
+return out.decode("utf-8").strip()
+def rm(self, filename):
+"""
+Public method to remove a file from the connected device.
+@param filename name of the file to be removed
+@type str
+@exception IOError raised to indicate an issue with the device
+"""
+assert filename
+commands = [
+"import os",
+"os.remove('{0}')".format(filename),
+]
+out, err = self.__execute(commands)
+if err:
+raise IOError(self.__shortError(err))
+def mkdir(self, dirname):
+"""
+Public method to create a new directory.
+@param dirname name of the directory to create
+@type str
+@exception IOError raised to indicate an issue with the device
+"""
+assert dirname
+commands = [
+"import os",
+"os.mkdir('{0}')".format(dirname),
+]
+out, err = self.__execute(commands)
+if err:
+raise IOError(self.__shortError(err))
+def rmdir(self, dirname):
+"""
+Public method to remove a directory.
+@param dirname name of the directory to be removed
+@type str
+@exception IOError raised to indicate an issue with the device
+"""
+assert dirname
+commands = [
+"import os",
+"os.rmdir('{0}')".format(dirname),
+]
+out, err = self.__execute(commands)
+if err:
+raise IOError(self.__shortError(err))
 def put(self, hostFileName, deviceFileName):
 """
 Public method to copy a local file to the connected device.
 @type str
 @param deviceFileName name of the file to copy to
 @type str
 @return flag indicating success
 @rtype bool
+@exception IOError raised to indicate an issue with the device
 """
 # TODO: not implemented yet
 def get(self, deviceFileName, hostFileName):
 """
 @type str
 @param hostFileName name of the file to copy to
 @type str
 @return flag indicating success
 @rtype bool
+@exception IOError raised to indicate an issue with the device
 """
 # TODO: not implemented yet
+##################################################################
+## Utility methods below
+##################################################################
+def mtime2string(self, mtime):
+"""
+Public method to convert a time value to a string representation.
+@param mtime time value
+@type int
+@return string representation of the given time
+@rtype str
+"""
+return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(mtime))
+def mode2string(self, mode):
+"""
+Public method to convert a mode value to a string representation.
+@param mode mode value
+@type int
+@return string representation of the given mode value
+@rtype str
+"""
+return stat.filemode(mode)
+# TODO: remove this
+##
+##if __name__ == "__main__":
+##    from PyQt5.QtCore import QCoreApplication, QTimer
+##    from MicroPythonSerialPort import MicroPythonSerialPort
+##
+##    app = QCoreApplication([])
+##
+##    serial = MicroPythonSerialPort()
+##    serial.openSerialLink("/dev/ttyUSB0")
+##    fs = MicroPythonFileSystem()
+##    fs.setSerial(serial)
+##
+##    def tf():
+##        fs.cd("/flash")
+##        print(fs.pwd())
+##        fs.cd("odroid_go")
+##        print(fs.pwd())
+##        ll = fs.lls()
+##        print(ll)
+##        for f, (m, s, t) in ll:
+##            print(fs.mode2string(m), s, fs.mtime2string(t), f)
+##        fs.cd("..")
+##        print(fs.pwd())
+##        ll = fs.lls("odroid_go")
+##        print(ll)
+##        for f, (m, s, t) in ll:
+##            print(fs.mode2string(m), s, fs.mtime2string(t), f)
+##
+##    QTimer.singleShot(0, tf)
+##    app.exec_()

Mercurial Repositories > eric / file comparison

comparison: eric6/MicroPython/MicroPythonFileSystem.py

eric6/MicroPython/MicroPythonFileSystem.py