eric: comparison eric6/MicroPython/MicroPythonCommandsInterface.py

-:62191d1aeeed
+:701cdc76f887
 """
 if not self.__serial:
 return False
 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
 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)
 if self.__serial.hasTimedOut():
-return False
+# it timed out; try it again and than fail
-self.__serial.write(b"\x04")        # send CTRL-D to soft reset
-self.__serial.readUntil(softRebootMessage)
-if self.__serial.hasTimedOut():
-return False
-# 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)
 if self.__serial.hasTimedOut():
 return False
 QCoreApplication.processEvents(QEventLoop.ExcludeUserInputEvents)
 self.__serial.readAll()             # read all data and discard it
 return True
 def __rawOff(self):
 return b"", b"Device not connected or not switched on."
 result = bytearray()
 err = b""
+# switch on raw mode
 self.__blockReadyRead = True
 ok = self.__rawOn()
 if not ok:
 self.__blockReadyRead = False
 return (
 b"",
 b"Could not switch to raw mode. Is the device switched on?"
 )
+# send commands
 QThread.msleep(10)
 for command in commands:
 if command:
 commandBytes = command.encode("utf-8")
 self.__serial.write(commandBytes + b"\x04")
+QCoreApplication.processEvents(
+QEventLoop.ExcludeUserInputEvents)
+ok = self.__serial.readUntil(b"OK")
+if ok != b"OK":
+return (
+b"",
+"Expected 'OK', got '{0}', followed by '{1}'".format(
+ok, self.__serial.readAll()).encode("utf-8")
+)
 # read until prompt
 response = self.__serial.readUntil(b"\x04>")
 if self.__serial.hasTimedOut():
 self.__blockReadyRead = False
 return b"", b"Timeout while processing commands."
-if b"\x04" in response[2:-2]:
+if b"\x04" in response[:-2]:
 # split stdout, stderr
-out, err = response[2:-2].split(b"\x04")
+out, err = response[:-2].split(b"\x04")
 result += out
 else:
 err = b"invalid response received: " + response
 if err:
 self.__blockReadyRead = False
 return b"", err
+# switch off raw mode
 QThread.msleep(10)
 self.__rawOff()
 self.__blockReadyRead = False
 return bytes(result), err
 @return tuple containg the directory listing
 @rtype tuple of str
 @exception IOError raised to indicate an issue with the device
 """
 commands = [
-"import os",
+"import os as __os_",
-"print(os.listdir('{0}'))".format(dirname),
+"print(__os_.listdir('{0}'))".format(dirname),
+"del __os_",
 ]
 out, err = self.execute(commands)
 if err:
 raise IOError(self.__shortError(err))
 return ast.literal_eval(out.decode("utf-8"))
 directory doesn't exist.
 @rtype tuple of (str, tuple)
 @exception IOError raised to indicate an issue with the device
 """
 commands = [
-"import os",
+"import os as __os_",
+"\n".join([
+"def is_visible(filename):",
+"    return (not filename.startswith('.') and"
+" not filename.endswith('~'))",
+]),
 "\n".join([
 "def stat(filename):",
 "    try:",
-"        rstat = os.lstat(filename)",
+"        rstat = __os_.lstat(filename)",
 "    except:",
-"        rstat = os.stat(filename)",
+"        rstat = __os_.stat(filename)",
 "    return tuple(rstat)",
 ]),
 "\n".join([
 "def listdir_stat(dirname):",
 "    try:",
-"        files = os.listdir(dirname)",
+"        files = __os_.listdir(dirname)",
 "    except OSError:",
 "        return None",
 "    if dirname in ('', '/'):",
-"        return list((f, stat(f)) for f in files)",
+"        return list((f, stat(f)) for f in files if"
-"    return list((f, stat(dirname + '/' + f)) for f in files)",
+" is_visible(f))",
+"    return list((f, stat(dirname + '/' + f)) for f in files"
+" if is_visible(f))",
 ]),
 "print(listdir_stat('{0}'))".format(dirname),
+"del __os_, stat, listdir_stat",
 ]
 out, err = self.execute(commands)
 if err:
 raise IOError(self.__shortError(err))
 fileslist = ast.literal_eval(out.decode("utf-8"))
 @exception IOError raised to indicate an issue with the device
 """
 assert dirname
 commands = [
-"import os",
+"import os as __os_",
-"os.chdir('{0}')".format(dirname),
+"__os_.chdir('{0}')".format(dirname),
+"del __os_",
 ]
 out, err = self.execute(commands)
 if err:
 raise IOError(self.__shortError(err))
 @return current directory
 @rtype str
 @exception IOError raised to indicate an issue with the device
 """
 commands = [
-"import os",
+"import os as __os_",
-"print(os.getcwd())",
+"print(__os_.getcwd())",
+"del __os_",
 ]
 out, err = self.execute(commands)
 if err:
 raise IOError(self.__shortError(err))
 return out.decode("utf-8").strip()
+# TODO: test this
 def rm(self, filename):
 """
 Public method to remove a file from the connected device.
 @param filename name of the file to be removed
 @exception IOError raised to indicate an issue with the device
 """
 assert filename
 commands = [
-"import os",
+"import os as __os_",
-"os.remove('{0}')".format(filename),
+"__os_.remove('{0}')".format(filename),
-]
+"del __os_",
-out, err = self.execute(commands)
+]
-if err:
+out, err = self.execute(commands)
-raise IOError(self.__shortError(err))
+if err:
+raise IOError(self.__shortError(err))
+# TODO: test this
 def rmrf(self, name, recursive=False, force=False):
 """
 Public method to remove a file or directory recursively.
 @param name of the file or directory to remove
 @exception IOError raised to indicate an issue with the device
 """
 assert name
 commands = [
-"import os",
+"import os as __os_",
 "\n".join([
 "def remove_file(name, recursive=False, force=False):",
 "    try:",
-"        mode = os.stat(name)[0]",
+"        mode = __os_.stat(name)[0]",
 "        if mode & 0x4000 != 0:",
 "            if recursive:",
-"                for file in os.listdir(name):",
+"                for file in __os_.listdir(name):",
 "                    success = remove_file(name + '/' + file,"
 " recursive, force)",
 "                    if not success and not force:",
 "                        return False",
-"                os.rmdir(name)",
+"                __os_.rmdir(name)",
 "            else:",
 "                if not force:",
 "                    return False",
 "        else:",
-"            os.remove(name)",
+"            __os_.remove(name)",
 "    except:",
 "        if not force:",
 "            return False",
 "    return True",
 ]),
 "print(remove_file('{0}', {1}, {2}))".format(name, recursive,
 force),
+"del __os_, remove_file",
 ]
 out, err = self.execute(commands)
 if err:
 raise IOError(self.__shortError(err))
 return ast.literal_eval(out.decode("utf-8"))
+# TODO: test this
 def mkdir(self, dirname):
 """
 Public method to create a new directory.
 @param dirname name of the directory to create
 @exception IOError raised to indicate an issue with the device
 """
 assert dirname
 commands = [
-"import os",
+"import os as __os_",
-"os.mkdir('{0}')".format(dirname),
+"__os_.mkdir('{0}')".format(dirname),
-]
+"del __os_",
-out, err = self.execute(commands)
+]
-if err:
+out, err = self.execute(commands)
-raise IOError(self.__shortError(err))
+if err:
+raise IOError(self.__shortError(err))
+# TODO: test this
 def rmdir(self, dirname):
 """
 Public method to remove a directory.
 @param dirname name of the directory to be removed
 @exception IOError raised to indicate an issue with the device
 """
 assert dirname
 commands = [
-"import os",
+"import os as __os_",
-"os.rmdir('{0}')".format(dirname),
+"__os_.rmdir('{0}')".format(dirname),
-]
+"del __os_",
-out, err = self.execute(commands)
+]
-if err:
+out, err = self.execute(commands)
-raise IOError(self.__shortError(err))
+if err:
+raise IOError(self.__shortError(err))
+# TODO: test this
 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
 ]
 while content:
 chunk = content[:64]
 commands.append("f(" + repr(chunk) + ")")
 content = content[64:]
-commands.append("fd.close()")
+commands.extend([
+"fd.close()",
+"del f, fd",
+])
 out, err = self.execute(commands)
 if err:
 raise IOError(self.__shortError(err))
 return True
+# TODO: test this
 def get(self, deviceFileName, hostFileName=None):
 """
 Public method to copy a file from the connected device.
 @param deviceFileName name of the file to copy
 if not hostFileName:
 hostFileName = deviceFileName
 commands = [
 "\n".join([
-"try:",
+"def send_data():",
-"    from microbit import uart as u",
-"except ImportError:",
 "    try:",
-"        from machine import UART",
+"        from microbit import uart as u",
-"        u = UART(0, {0})".format(115200),
+"    except ImportError:",
-"    except Exception:",
 "        try:",
-"            from sys import stdout as u",
+"            from machine import UART",
+"            u = UART(0, {0})".format(115200),
 "        except Exception:",
-"            raise Exception('Could not find UART module in"
+"            try:",
-" device.')",
+"                from sys import stdout as u",
-]),
+"            except Exception:",
-"f = open('{0}', 'rb')".format(deviceFileName),
+"                raise Exception('Could not find UART module"
-"r = f.read",
+" in device.')",
-"result = True",
+"    f = open('{0}', 'rb')".format(deviceFileName),
-"\n".join([
+"    r = f.read",
-"while result:",
+"    result = True",
-"    result = r(32)",
+"    while result:",
-"    if result:",
+"        result = r(32)",
-"        u.write(result)",
+"        if result:",
-]),
+"            u.write(result)",
-"f.close()",
+"    f.close()",
+]),
+"send_data()",
 ]
 out, err = self.execute(commands)
 if err:
 raise IOError(self.__shortError(err))
 size, the used size and the free size
 @rtype tuple of tuples of (str, int, int, int)
 @exception IOError raised to indicate an issue with the device
 """
 commands = [
-"import os",
+"import os as __os_",
 "\n".join([
 "def fsinfo():",
 "    infolist = []",
-"    fsnames = os.listdir('/')",
+"    info = __os_.statvfs('/')",
-"    for fs in fsnames:",
+"    if info[0] == 0:",
-"        fs = '/' + fs",
+# assume it is just mount points
-"        infolist.append((fs, os.statvfs(fs)))",
+"        fsnames = __os_.listdir('/')",
+"        for fs in fsnames:",
+"            fs = '/' + fs",
+"            infolist.append((fs, __os_.statvfs(fs)))",
+"    else:",
+"        infolist.append(('/', info))",
 "    return infolist",
 ]),
 "print(fsinfo())",
+"del __os_, fsinfo",
 ]
 out, err = self.execute(commands)
 if err:
 raise IOError(self.__shortError(err))
 infolist = ast.literal_eval(out.decode("utf-8"))
 @return dictionary containing the version information
 @rtype dict
 @exception IOError raised to indicate an issue with the device
 """
 commands = [
-"import os",
+"import os as __os_",
-"print(os.uname())",
+"print(__os_.uname())",
+"del __os_",
 ]
 out, err = self.execute(commands)
 if err:
 raise IOError(self.__shortError(err))
 @return dictionary containing the implementation information
 @rtype dict
 @exception IOError raised to indicate an issue with the device
 """
 commands = [
-"import sys",
+"import sys as __sys_",
 "res = {}",                             # __IGNORE_WARNING_M613__
 "\n".join([
 "try:",
-"    res['name'] = sys.implementation.name",
+"    res['name'] = __sys_.implementation.name",
 "except AttributeError:",
 "    res['name'] = 'unknown'",
 ]),
 "\n".join([
 "try:",
 "    res['version'] = '.'.join((str(i) for i in"
-" sys.implementation.version))",
+" __sys_.implementation.version))",
 "except AttributeError:",
 "    res['version'] = 'unknown'",
 ]),
 "print(res)",
+"del res, __sys_",
 ]
 out, err = self.execute(commands)
 if err:
 raise IOError(self.__shortError(err))
 return ast.literal_eval(out.decode("utf-8"))
 commands = [
 "\n".join([
 "def set_time(rtc_time):",
 "    rtc = None",
 "    try:",           # Pyboard (it doesn't have machine.RTC())
-"        import pyb",
+"        import pyb as __pyb_",
-"        rtc = pyb.RTC()",
+"        rtc = __pyb_.RTC()",
 "        clock_time = rtc_time[:6] + (rtc_time[6] + 1, 0)",
 "        rtc.datetime(clock_time)",
-"    except:",
+"        del __pyb_",
+"    except Exception:",
 "        try:",
-"            import machine",
+"            import machine as __machine_",
-"            rtc = machine.RTC()",
+"            rtc = __machine_.RTC()",
 "            try:",     # ESP8266 may use rtc.datetime()
 "                clock_time = rtc_time[:6] +"
 " (rtc_time[6] + 1, 0)",
 "                rtc.datetime(clock_time)",
-"            except:",  # ESP32 uses rtc.init()
+"            except Exception:",  # ESP32 uses rtc.init()
 "                rtc.init(rtc_time[:6])",
+"            del __machine_",
 "        except:",
 "            try:",
-"                import rtc, time",
+"                import rtc as __rtc_",
-"                clock=rtc.RTC()",
+"                import time as __time_",
-"                clock.datetime = time.struct_time(rtc_time +"
+"                clock=__rtc_.RTC()",
-" (-1, -1))",
+"                clock.datetime = __time_.struct_time("
+"rtc_time + (-1, -1))",
+"                del __rtc_, __time_",
 "            except:",
 "                pass",
 ]),
 "set_time({0})".format((now.tm_year, now.tm_mon, now.tm_mday,
 now.tm_hour, now.tm_min, now.tm_sec,
-now.tm_wday))
+now.tm_wday)),
+"del set_time",
 ]
 out, err = self.execute(commands)
 if err:
 raise IOError(self.__shortError(err))
 @return time of the device
 @rtype str
 @exception IOError raised to indicate an issue with the device
 """
 commands = [
-"import time",
+"import time as __time_",
 "\n".join([
 "try:",
-"    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()"
+"    print(__time_.strftime('%Y-%m-%d %H:%M:%S',"
-"))",
+" __time_.localtime()))",
 # __IGNORE_WARNING_M601__
 "except AttributeError:",
-"    tm = time.localtime()",
+"    tm = __time_.localtime()",
 "    print('{0:04d}-{1:02d}-{2:02d} {3:02d}:{4:02d}:{5:02d}'"
 ".format(tm.tm_year, tm.tm_mon, tm.tm_mday, tm.tm_hour,"
 " tm.tm_min, tm.tm_sec))",
-]),
+"    del tm",
+]),
+"del __time_"
 ]
 out, err = self.execute(commands)
 if err:
 raise IOError(self.__shortError(err))
 return out.decode("utf-8").strip()

Mercurial Repositories > eric / file comparison

comparison: eric6/MicroPython/MicroPythonCommandsInterface.py

eric6/MicroPython/MicroPythonCommandsInterface.py