eric: comparison src/eric7/MicroPython/Devices/DeviceBase.py

-:57496966803c
+:6378da868bb0
 """
 Module implementing some utility functions and the MicroPythonDevice base
 class.
 """
+import ast
 import contextlib
 import copy
 import os
+import time
 from PyQt6.QtCore import QObject, pyqtSlot
 from PyQt6.QtWidgets import QInputDialog
 from eric7 import Preferences
 @type QObject
 """
 super().__init__(parent)
 self._deviceType = deviceType
+self._interface = microPythonWidget.deviceInterface()
 self.microPython = microPythonWidget
 self._deviceData = {}  # dictionary with essential device data
 def setConnected(self, connected):
 """
 """
 self._deviceData = {}
 if connected:
 with contextlib.suppress(OSError):
-self._deviceData = self.microPython.commandsInterface().getDeviceData()
+self._deviceData = self.__getDeviceData()
 def getDeviceType(self):
 """
 Public method to get the device type.
 """<p>The device data is not available. Try to connect to the"""
 """ device again. Aborting...</p>"""
 ).format(self.getDeviceType()),
 )
 return False
+def hasCircuitPython(self):
+"""
+Public method to check, if the connected device is flashed with CircuitPython.
+@return flag indicating CircuitPython
+@rtype bool
+"""
+return (
+self.checkDeviceData()
+and self._deviceData["mpy_name"].lower() == "circuitpython"
+)
 def setButtons(self):
 """
 Public method to enable the supported action buttons.
 """
 commands = [c.encode("utf-8)") + b"\r" for c in commandsList]
 commands.append(b"\r")
 commands.append(b"\x04")
 rawOff = [b"\x02", b"\x02"]
 commandSequence = rawOn + newLine + commands + rawOff
-self.microPython.commandsInterface().executeAsync(commandSequence)
+self._interface.executeAsync(commandSequence)
 @pyqtSlot()
 def handleDataFlood(self):
 """
 Public slot handling a data floof from the device.
 @return list of tuples with menu text and URL to be opened for each
 entry
 @rtype list of tuple of (str, str)
 """
 return []
+##################################################################
+## Methods below implement the file system commands
+##################################################################
+def _shortError(self, error):
+"""
+Protected 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.")
+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 OSError raised to indicate an issue with the device
+"""
+command = """
+import os as __os_
+print(__os_.listdir('{0}'))
+del __os_
+""".format(
+dirname
+)
+out, err = self._interface.execute(command)
+if err:
+raise OSError(self._shortError(err))
+return ast.literal_eval(out.decode("utf-8"))
+def lls(self, dirname="", fullstat=False, showHidden=False):
+"""
+Public method to get a long directory listing of the connected device
+including meta data.
+@param dirname name of the directory to be listed
+@type str
+@param fullstat flag indicating to return the full stat() tuple
+@type bool
+@param showHidden flag indicating to show hidden files as well
+@type bool
+@return list containing the directory listing with tuple entries of
+the name and and a tuple of mode, size and time (if fullstat is
+false) or the complete stat() tuple. 'None' is returned in case the
+directory doesn't exist.
+@rtype tuple of (str, tuple)
+@exception OSError raised to indicate an issue with the device
+"""
+command = """
+import os as __os_
+def is_visible(filename, showHidden):
+return showHidden or (filename[0] != '.' and filename[-1] != '~')
+def stat(filename):
+try:
+rstat = __os_.lstat(filename)
+except:
+rstat = __os_.stat(filename)
+return tuple(rstat)
+def listdir_stat(dirname, showHidden):
+try:
+files = __os_.listdir(dirname)
+except OSError:
+return []
+if dirname in ('', '/'):
+return list((f, stat(f)) for f in files if is_visible(f, showHidden))
+return list(
+(f, stat(dirname + '/' + f)) for f in files if is_visible(f, showHidden)
+)
+print(listdir_stat('{0}', {1}))
+del __os_, stat, listdir_stat, is_visible
+""".format(
+dirname, showHidden
+)
+out, err = self._interface.execute(command)
+if err:
+raise OSError(self._shortError(err))
+fileslist = ast.literal_eval(out.decode("utf-8"))
+if fileslist is None:
+return None
+else:
+if fullstat:
+return fileslist
+else:
+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 dirname directory to change to
+@type str
+@exception OSError raised to indicate an issue with the device
+"""
+if dirname:
+command = """
+import os as __os_
+__os_.chdir('{0}')
+del __os_
+""".format(
+dirname
+)
+out, err = self._interface.execute(command)
+if err:
+raise OSError(self._shortError(err))
+def pwd(self):
+"""
+Public method to get the current directory of the connected device.
+@return current directory
+@rtype str
+@exception OSError raised to indicate an issue with the device
+"""
+command = """
+import os as __os_
+print(__os_.getcwd())
+del __os_
+"""
+out, err = self._interface.execute(command)
+if err:
+raise OSError(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 OSError raised to indicate an issue with the device
+"""
+if filename:
+command = """
+import os as __os_
+__os_.remove('{0}')
+del __os_
+""".format(
+filename
+)
+out, err = self._interface.execute(command)
+if err:
+raise OSError(self._shortError(err))
+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
+@type str
+@param recursive flag indicating a recursive deletion
+@type bool
+@param force flag indicating to ignore errors
+@type bool
+@return flag indicating success
+@rtype bool
+@exception OSError raised to indicate an issue with the device
+"""
+if name:
+command = """
+import os as __os_
+def remove_file(name, recursive=False, force=False):
+try:
+mode = __os_.stat(name)[0]
+if mode & 0x4000 != 0:
+if recursive:
+for file in __os_.listdir(name):
+success = remove_file(name + '/' + file, recursive, force)
+if not success and not force:
+return False
+__os_.rmdir(name)
+else:
+if not force:
+return False
+else:
+__os_.remove(name)
+except:
+if not force:
+return False
+return True
+print(remove_file('{0}', {1}, {2}))
+del __os_, remove_file
+""".format(
+name, recursive, force
+)
+out, err = self._interface.execute(command)
+if err:
+raise OSError(self._shortError(err))
+return ast.literal_eval(out.decode("utf-8"))
+return False
+def mkdir(self, dirname):
+"""
+Public method to create a new directory.
+@param dirname name of the directory to create
+@type str
+@exception OSError raised to indicate an issue with the device
+"""
+if dirname:
+command = """
+import os as __os_
+__os_.mkdir('{0}')
+del __os_
+""".format(
+dirname
+)
+out, err = self._interface.execute(command)
+if err:
+raise OSError(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 OSError raised to indicate an issue with the device
+"""
+if dirname:
+command = """
+import os as __os_
+__os_.rmdir('{0}')
+del __os_
+""".format(
+dirname
+)
+out, err = self._interface.execute(command)
+if err:
+raise OSError(self._shortError(err))
+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
+@param deviceFileName name of the file to copy to
+@type str
+@return flag indicating success
+@rtype bool
+@exception OSError raised to indicate an issue with the device
+"""
+if not os.path.isfile(hostFileName):
+raise OSError("No such file: {0}".format(hostFileName))
+if not deviceFileName:
+deviceFileName = os.path.basename(hostFileName)
+with open(hostFileName, "rb") as hostFile:
+content = hostFile.read()
+return self.putData(deviceFileName, content)
+def putData(self, deviceFileName, content):
+"""
+Public method to write the given data to the connected device.
+@param deviceFileName name of the file to write to
+@type str
+@param content data to write
+@type bytes
+@return flag indicating success
+@rtype bool
+@exception OSError raised to indicate an issue with the device
+"""
+if not deviceFileName:
+raise OSError("Missing device file name")
+# convert eol '\r'
+content = content.replace(b"\r\n", b"\r")
+content = content.replace(b"\n", b"\r")
+commands = [
+"fd = open('{0}', 'wb')".format(deviceFileName),
+"f = fd.write",
+]
+while content:
+chunk = content[:64]
+commands.append("f(" + repr(chunk) + ")")
+content = content[64:]
+commands.extend(
+[
+"fd.close()",
+"del f, fd",
+]
+)
+command = "\n".join(commands)
+out, err = self._interface.execute(command)
+if err:
+raise OSError(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
+@param hostFileName name of the file to copy to
+@type str
+@return flag indicating success
+@rtype bool
+@exception OSError raised to indicate an issue with the device
+"""
+if not deviceFileName:
+raise OSError("Missing device file name")
+if not hostFileName:
+hostFileName = deviceFileName
+out = self.getData(deviceFileName)
+with open(hostFileName, "wb") as hostFile:
+hostFile.write(out)
+return True
+def getData(self, deviceFileName):
+"""
+Public method to read data from the connected device.
+@param deviceFileName name of the file to read from
+@type str
+@return data read from the device
+@rtype bytes
+@exception OSError raised to indicate an issue with the device
+"""
+if not deviceFileName:
+raise OSError("Missing device file name")
+command = """
+def send_data():
+try:
+from microbit import uart as u
+except ImportError:
+try:
+from sys import stdout as u
+except ImportError:
+try:
+from machine import UART
+u = UART(0, 115200)
+except Exception:
+raise Exception('Could not find UART module in device.')
+f = open('{0}', 'rb')
+r = f.read
+result = True
+while result:
+result = r(32)
+if result:
+u.write(result)
+f.close()
+send_data()
+del send_data
+""".format(
+deviceFileName
+)
+out, err = self._interface.execute(command)
+if err:
+raise OSError(self._shortError(err))
+# write the received bytes to the local file
+# convert eol to "\n"
+out = out.replace(b"\r\n", b"\n")
+out = out.replace(b"\r", b"\n")
+return out
+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)
+@exception OSError raised to indicate an issue with the device
+"""
+command = """
+import os as __os_
+def fsinfo():
+infolist = []
+info = __os_.statvfs('/')
+if info[0] == 0:
+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._interface.execute(command)
+if err:
+raise OSError(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)
+##################################################################
+## board information related methods below
+##################################################################
+def __getDeviceData(self):
+"""
+Private method to get some essential data for the connected board.
+@return dictionary containing the determined data
+@rtype dict
+@exception OSError raised to indicate an issue with the device
+"""
+command = """
+res = {}
+import os as __os_
+uname = __os_.uname()
+res['sysname'] = uname.sysname
+res['nodename'] = uname.nodename
+res['release'] = uname.release
+res['version'] = uname.version
+res['machine'] = uname.machine
+import sys as __sys_
+res['py_platform'] = __sys_.platform
+res['py_version'] = __sys_.version
+try:
+res['mpy_name'] = __sys_.implementation.name
+except AttributeError:
+res['mpy_name'] = 'unknown'
+try:
+res['mpy_version'] = '.'.join((str(i) for i in __sys_.implementation.version))
+except AttributeError:
+res['mpy_version'] = 'unknown'
+try:
+import pimoroni as __pimoroni_
+res['mpy_variant'] = 'Pimoroni'
+del __pimoroni_
+except ImportError:
+res['mpy_variant'] = ''
+print(res)
+del res, uname, __os_, __sys_
+"""
+out, err = self._interface.execute(command)
+if err:
+raise OSError(self._shortError(err))
+return ast.literal_eval(out.decode("utf-8"))
+def getBoardInformation(self):
+"""
+Public method to get some information data of the connected board.
+@return dictionary containing the determined data
+@rtype dict
+@exception OSError raised to indicate an issue with the device
+"""
+command = """
+res = {}
+import gc as __gc_
+__gc_.enable()
+__gc_.collect()
+mem_alloc = __gc_.mem_alloc()
+mem_free = __gc_.mem_free()
+mem_total = mem_alloc + mem_free
+res['mem_total_kb'] = mem_total / 1024.0
+res['mem_used_kb'] = mem_alloc / 1024.0
+res['mem_used_pc'] = mem_alloc / mem_total * 100.0
+res['mem_free_kb'] = mem_free / 1024.0
+res['mem_free_pc'] = mem_free / mem_total * 100.0
+del __gc_, mem_alloc, mem_free, mem_total
+import os as __os_
+uname = __os_.uname()
+res['sysname'] = uname.sysname
+res['nodename'] = uname.nodename
+res['release'] = uname.release
+res['version'] = uname.version
+res['machine'] = uname.machine
+import sys as __sys_
+res['py_platform'] = __sys_.platform
+res['py_version'] = __sys_.version
+try:
+res['mpy_name'] = __sys_.implementation.name
+except AttributeError:
+res['mpy_name'] = 'unknown'
+try:
+res['mpy_version'] = '.'.join((str(i) for i in __sys_.implementation.version))
+except AttributeError:
+res['mpy_version'] = 'unknown'
+try:
+import pimoroni as __pimoroni_
+res['mpy_variant'] = 'Pimoroni'
+del __pimoroni_
+except ImportError:
+res['mpy_variant'] = ''
+try:
+stat_ = __os_.statvfs('/flash')
+res['flash_info_available'] = True
+res['flash_total_kb'] = stat_[2] * stat_[0] / 1024.0
+res['flash_free_kb'] = stat_[3] * stat_[0] / 1024.0
+res['flash_used_kb'] = res['flash_total_kb'] - res['flash_free_kb']
+res['flash_free_pc'] = res['flash_free_kb'] / res['flash_total_kb'] * 100.0
+res['flash_used_pc'] = res['flash_used_kb'] / res['flash_total_kb'] * 100.0
+del stat_
+except AttributeError:
+res['flash_info_available'] = False
+try:
+import machine as __mc_
+if isinstance(__mc_.freq(), tuple):
+res['mc_frequency_mhz'] = __mc_.freq()[0] / 1000000.0
+else:
+res['mc_frequency_mhz'] = __mc_.freq() / 1000000.0
+res['mc_id'] = ':'.join(['{0:X}'.format(x) for x in __mc_.unique_id()])
+del __mc_
+except ImportError:
+try:
+import microcontroller as __mc_
+res['mc_frequency_mhz'] = __mc_.cpu.frequency / 1000000.0
+res['mc_temp_c'] = __mc_.cpu.temperature
+res['mc_id'] = ':'.join(['{0:X}'.format(x) for x in __mc_.cpu.uid])
+del __mc_
+except ImportError:
+res['mc_frequency'] = None
+res['mc_temp'] = None
+try:
+import ulab as __ulab_
+res['ulab'] = __ulab_.__version__
+del __ulab_
+except ImportError:
+res['ulab'] = None
+print(res)
+del res, __os_, __sys_
+"""
+out, err = self._interface.execute(command)
+if err:
+raise OSError(self._shortError(err))
+return ast.literal_eval(out.decode("utf-8"))
+def getModules(self):
+"""
+Public method to show a list of modules built into the firmware.
+@return list of builtin modules
+@rtype list of str
+@exception OSError raised to indicate an issue with the device
+"""
+commands = ["help('modules')"]
+out, err = self._interface.execute(commands)
+if err:
+raise OSError(self._shortError(err))
+modules = []
+for line in out.decode("utf-8").splitlines()[:-1]:
+modules.extend(line.split())
+return modules
+##################################################################
+## time related methods below
+##################################################################
+def getTime(self):
+"""
+Public method to get the current time of the device.
+@return time of the device
+@rtype str
+@exception OSError raised to indicate an issue with the device
+"""
+command = """
+try:
+import rtc as __rtc_
+print(
+'{0:04d}-{1:02d}-{2:02d} {3:02d}:{4:02d}:{5:02d}'
+.format(*__rtc_.RTC().datetime[:6])
+)
+del __rtc_
+except:
+import time as __time_
+try:
+print(__time_.strftime('%Y-%m-%d %H:%M:%S', __time_.localtime()))
+except AttributeError:
+tm = __time_.localtime()
+print(
+'{0:04d}-{1:02d}-{2:02d} {3:02d}:{4:02d}:{5:02d}'
+.format(tm[0], tm[1], tm[2], tm[3], tm[4], tm[5])
+)
+del tm
+del __time_
+"""
+out, err = self._interface.execute(command)
+if err:
+if b"NotImplementedError" in err:
+return "&lt;unsupported&gt; &lt;unsupported&gt;"
+raise OSError(self._shortError(err))
+return out.decode("utf-8").strip()
+def _getSetTimeCode(self):
+"""
+Protected method to get the device code to set the time.
+Note: This method must be implemented in the various device specific
+subclasses.
+@return code to be executed on the connected device to set the time
+@rtype str
+"""
+# rtc_time[0] - year    4 digit
+# rtc_time[1] - month   1..12
+# rtc_time[2] - day     1..31
+# rtc_time[3] - weekday 1..7 1=Monday
+# rtc_time[4] - hour    0..23
+# rtc_time[5] - minute  0..59
+# rtc_time[6] - second  0..59
+# rtc_time[7] - yearday 1..366
+# rtc_time[8] - isdst   0, 1, or -1
+if self.hasCircuitPython():
+# CircuitPython is handled here in order to not duplicate the code in all
+# specific boards able to be flashed with CircuitPython or MicroPython
+return """
+def set_time(rtc_time):
+import rtc
+import time
+clock = rtc.RTC()
+clock_time = rtc_time[:3] + rtc_time[4:7] + (rtc_time[3], rtc_time[7], rtc_time[8])
+clock.datetime = time.struct_time(clock_time)
+"""
+else:
+return ""
+def syncTime(self, deviceType, hasCPy=False):
+"""
+Public method to set the time of the connected device to the local
+computer's time.
+@param deviceType type of board to sync time to
+@type str
+@param hasCPy flag indicating that the device has CircuitPython loadede
+(defaults to False)
+@type bool
+@exception OSError raised to indicate an issue with the device
+"""
+setTimeCode = self._getSetTimeCode()
+if setTimeCode:
+now = time.localtime(time.time())
+command = """{0}
+set_time({1})
+del set_time
+""".format(
+setTimeCode,
+(
+now.tm_year,
+now.tm_mon,
+now.tm_mday,
+now.tm_wday + 1,
+now.tm_hour,
+now.tm_min,
+now.tm_sec,
+now.tm_yday,
+now.tm_isdst,
+),
+)
+out, err = self._interface.execute(command)
+if err:
+raise OSError(self._shortError(err))
+#
+# eflag: noqa = M613

Mercurial Repositories > eric / file comparison

comparison: src/eric7/MicroPython/Devices/DeviceBase.py

src/eric7/MicroPython/Devices/DeviceBase.py