eric: comparison eric6/MicroPython/MicroPythonFileSystem.py

-:d1564b590677
+:9a3adf033f90
 from __future__ import unicode_literals
 import ast
 import time
 import stat
+import os
 from PyQt5.QtCore import pyqtSlot, pyqtSignal, QObject, QThread
 from .MicroPythonSerialPort import MicroPythonSerialPort
 <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>
+<li>version: get version info about MicroPython</li>
 </ul>
 """
 def __init__(self, parent=None):
 """
 Constructor
 ]
 out, err = self.__execute(commands)
 if err:
 raise IOError(self.__shortError(err))
-def put(self, hostFileName, deviceFileName):
+def put(self, hostFileName, deviceFileName=None):
 """
 Public method to copy a local file to the connected device.
 @param hostFileName name of the file to be copied
 @type str
 @type str
 @return flag indicating success
 @rtype bool
 @exception IOError raised to indicate an issue with the device
 """
-# TODO: not implemented yet
+if not os.path.isfile(hostFileName):
+raise IOError("No such file: {0}".format(hostFileName))
-def get(self, deviceFileName, hostFileName):
+with open(hostFileName, "rb") as hostFile:
+content = hostFile.read()
+if not deviceFileName:
+deviceFileName = os.path.basename(hostFileName)
+commands = [
+"fd = open('{0}', 'wb')".format(deviceFileName),
+"f = fd.write",
+]
+while content:
+chunk = content[:64]
+commands.append("f(" + repr(chunk) + ")")
+content = content[64:]
+commands.append("fd.close()")
+out, err = self.__execute(commands)
+if err:
+raise IOError(self.__shortError(err))
+return True
+def get(self, deviceFileName, hostFileName=None):
 """
 Public method to copy a file from the connected device.
 @param deviceFileName name of the file to copy
 @type str
 @type str
 @return flag indicating success
 @rtype bool
 @exception IOError raised to indicate an issue with the device
 """
-# TODO: not implemented yet
+if not hostFileName:
+hostFileName = deviceFileName
+commands = [
+"\n".join([
+"try:",
+"    from microbit import uart as u",
+"except ImportError:",
+"    try:",
+"        from machine import UART",
+"        u = UART(0, {0})".format(115200),
+"    except Exception:",
+"        try:",
+"            from sys import stdout as u",
+"        except Exception:",
+"            raise Exception('Could not find UART module in"
+" device.')",
+]),
+"f = open('{0}', 'rb')".format(deviceFileName),
+"r = f.read",
+"result = True",
+"\n".join([
+"while result:",
+"    result = r(32)"
+"    if result:",
+"        u.write(result)",
+]),
+"f.close()",
+]
+out, err = self.__execute(commands)
+if err:
+raise IOError(self.__shortError(err))
+# write the received bytes to the local file
+with open(hostFileName, "wb") as hostFile:
+hostFile.write(out)
+return True
+# TODO: add rsync function
+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
+running MicroPython or there was an issue parsing the output
+"""
+commands = [
+"import os",
+"print(os.uname())",
+]
+try:
+out, err = self.__execute(commands)
+if 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 syncTime(self):
+"""
+Public method to set the time of the connected device to the local
+computer's time.
+@exception IOError raised to indicate an issue with the device
+"""
+now = time.localtime(time.time())
+commands = [
+"\n".join([
+"def set_time(rtc_time):",
+"    rtc = None",
+"    try:",           # Pyboard (it doesn't have machine.RTC())
+"        import pyb",
+"        rtc = pyb.RTC()",
+"        rtc.datetime(rtc_time)",
+"    except:",
+"        try:",
+"            import machine",
+"            rtc = machine.RTC()",
+"            try:",     # ESP8266 uses rtc.datetime()
+"                rtc.datetime(rtc_time)",
+"            except:",  # ESP32 uses rtc.init()
+"                rtc.init(rtc_time)",
+"        except:",
+"            pass",
+]),
+"set_time({0})".format((now.tm_year, now.tm_mon, now.tm_mday,
+now.tm_wday + 1, now.tm_hour, now.tm_min,
+now.tm_sec, 0))
+]
+out, err = self.__execute(commands)
+if err:
+raise IOError(self.__shortError(err))
 class MicroPythonFileManager(QObject):
 """
 Class implementing an interface to the device file system commands with
 @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 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 longListFilesFailed(exc) emitted with a failure message to indicate
 a failed long listing operation
 @signal currentDirFailed(exc) emitted with a failure message to indicate
 that the current directory is not available
+@signal getFileFailed(exc) emitted with a failure message to indicate that
+the file could not be fetched
+@signal putFileFailed(exc) emitted with a failure message to indicate that
+the file could not be copied
+@signal deleteFileFailed(exc) emitted with a failure message to indicate
+that the file could not be deleted on the device
 """
 longListFiles = pyqtSignal(tuple)
 currentDir = pyqtSignal(str)
+getFileDone = pyqtSignal(str, str)
+putFileDone = pyqtSignal(str, str)
+deleteFileDone = pyqtSignal(str)
 longListFilesFailed = pyqtSignal(str)
 currentDirFailed = pyqtSignal(str)
+getFileFailed = pyqtSignal(str)
+putFileFailed = pyqtSignal(str)
+deleteFileFailed = pyqtSignal(str)
 def __init__(self, port, parent=None):
 """
 Constructor
 self.__serial.closeSerialLink()
 @pyqtSlot(str)
 def lls(self, dirname):
 """
-Public method to get a long listing of the given directory.
+Public slot to get a long listing of the given directory.
 @param dirname name of the directory to list
 @type str
 """
 try:
 self.longListFilesFailed.emit(str(exc))
 @pyqtSlot()
 def pwd(self):
 """
-Public method to get the current directory of the device.
+Public slot to get the current directory of the device.
 """
 try:
 pwd = self.__fs.pwd()
 self.currentDir.emit(pwd)
 except Exception as exc:
 self.currentDirFailed.emit(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.__fs.get(deviceFileName, hostFileName)
+self.getFileDone.emit(deviceFileName, hostFileName)
+except Exception as exc:
+self.getFileFailed.emit(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.__fs.put(hostFileName, deviceFileName)
+self.putFileDone(hostFileName, deviceFileName)
+except Exception as exc:
+self.putFileFailed(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.__fs.rm(deviceFileName)
+self.deleteFileDone.emit(deviceFileName)
+except Exception as exc:
+self.deleteFileFailed(str(exc))
 ##################################################################
 ## Utility methods below
 ##################################################################
 @rtype str
 """
 return stat.filemode(mode)
-def decoratedName(name, mode):
+def decoratedName(name, mode, isDir=False):
 """
 Function to decorate the given name according to the given mode.
 @param name file or directory name
 @type str
 @param mode mode value
 @type int
+@param isDir flag indicating that name is a directory
+@type bool
 @return decorated file or directory name
 @rtype str
 """
-if stat.S_ISDIR(mode):
+if stat.S_ISDIR(mode) or isDir:
 # append a '/' for directories
 return name + "/"
 else:
 # no change
 return name

Mercurial Repositories > eric / file comparison

comparison: eric6/MicroPython/MicroPythonFileSystem.py

eric6/MicroPython/MicroPythonFileSystem.py