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

-:1ba9d07ba9da
+:286c2a21f36f
 @type QObject
 """
 super().__init__(parent)
 self._deviceType = deviceType
-self._interface = microPythonWidget.deviceInterface()
+self._interface = None
 self.microPython = microPythonWidget
 self._deviceData = {}  # dictionary with essential device data
 self._submitMode = "raw"  # default is 'raw' mode to submit commands
 @type bool
 """
 self._deviceData = {}
 if connected:
+self._interface = self.microPython.deviceInterface()
 with contextlib.suppress(OSError):
 self._deviceData = self.__getDeviceData()
 self._deviceData["local_mip"] = (
 not self._deviceData["mip"] and not self._deviceData["upip"]
 )
 (
 self._deviceData["ethernet"],
 self._deviceData["ethernet_type"],
 ) = self.hasEthernet()
 self._deviceData["ntp"] = self.hasNetworkTime()
+else:
+self._interface = None
 def getDeviceType(self):
 """
 Public method to get the device type.
 return deviceDirectory
 else:
 # user cancelled
 return ""
+def executeCommands(self, commands, *, mode="raw", timeout=0):
+"""
+Public method to send commands to the connected device and return the
+result.
+If no connected interface is available, empty results will be returned.
+@param commands list of commands to be executed
+@type str or list of str
+@keyparam mode submit mode to be used (one of 'raw' or 'paste') (defaults to
+'raw')
+@type str
+@keyparam timeout per command timeout in milliseconds (0 for configured default)
+(defaults to 0)
+@type int (optional)
+@return tuple containing stdout and stderr output of the device
+@rtype tuple of (bytes, bytes)
+"""
+if self._interface is None:
+return b"", b""
+return self._interface.execute(commands, mode=mode, timeout=timeout)
 def sendCommands(self, commandsList):
 """
 Public method to send a list of commands to the device.
 @param commandsList list of commands to be sent to the device
 @type list of str
 """
-if self._submitMode == "raw":
+if self._interface is not None:
-rawOn = [  # sequence of commands to enter raw mode
+self._interface.executeAsync(commandsList, self._submitMode)
-b"\x02",  # Ctrl-B: exit raw repl (just in case)
-b"\r\x03\x03\x03",  # Ctrl-C three times: interrupt any running program
-b"\r\x01",  # Ctrl-A: enter raw REPL
-]
-newLine = [
-b'print("\\n")\r',
-]
-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._interface.executeAsync(commandSequence)
-elif self._submitMode == "paste":
-commands = b"\n".join([c.encode("utf-8)") for c in commandsList])
-commandSequence = ["@PasteOn@", commands]
-self._interface.executeAsyncPaste(commandSequence)
 @pyqtSlot()
 def handleDataFlood(self):
 """
 Public slot handling a data floof from the device.
 print(False)
 del __os_
 """.format(
 repr(pathname)
 )
-out, err = self._interface.execute(command, mode=self._submitMode)
+out, err = self.executeCommands(command, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 return out.strip() == b"True"
 def ls(self, dirname=""):
 print(__os_.listdir('{0}'))
 del __os_
 """.format(
 dirname
 )
-out, err = self._interface.execute(command, mode=self._submitMode)
+out, err = self.executeCommands(command, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 return ast.literal_eval(out.decode("utf-8"))
 def lls(self, dirname="", fullstat=False, showHidden=False):
 print(listdir_stat('{0}', {1}))
 del __os_, stat, listdir_stat, is_visible
 """.format(
 dirname, showHidden
 )
-out, err = self._interface.execute(command, mode=self._submitMode)
+out, err = self.executeCommands(command, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 fileslist = ast.literal_eval(out.decode("utf-8"))
 if fileslist is None:
 return None
 __os_.chdir('{0}')
 del __os_
 """.format(
 dirname
 )
-out, err = self._interface.execute(command, mode=self._submitMode)
+out, err = self.executeCommands(command, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 def pwd(self):
 """
 command = """
 import os as __os_
 print(__os_.getcwd())
 del __os_
 """
-out, err = self._interface.execute(command, mode=self._submitMode)
+out, err = self.executeCommands(command, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 return out.decode("utf-8").strip()
 def rm(self, filename):
 raise err
 del __os_
 """.format(
 filename
 )
-out, err = self._interface.execute(command, mode=self._submitMode)
+out, err = self.executeCommands(command, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 def rmrf(self, name, recursive=False, force=False):
 """
 print(remove_file('{0}', {1}, {2}))
 del __os_, remove_file
 """.format(
 name, recursive, force
 )
-out, err = self._interface.execute(
+out, err = self.executeCommands(
 command, mode=self._submitMode, timeout=20000
 )
 if err:
 raise OSError(self._shortError(err))
 return ast.literal_eval(out.decode("utf-8"))
 __os_.mkdir('{0}')
 del __os_
 """.format(
 dirname
 )
-out, err = self._interface.execute(command, mode=self._submitMode)
+out, err = self.executeCommands(command, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 def rmdir(self, dirname):
 """
 raise
 del __os_
 """.format(
 dirname
 )
-out, err = self._interface.execute(command, mode=self._submitMode)
+out, err = self.executeCommands(command, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 def put(self, hostFileName, deviceFileName=None):
 """
 "del f, fd",
 ]
 )
 command = "\n".join(commands)
-out, err = self._interface.execute(command, mode=self._submitMode)
+out, err = self.executeCommands(command, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 return True
 def get(self, deviceFileName, hostFileName=None):
 send_data()
 del send_data
 """.format(
 deviceFileName
 )
-out, err = self._interface.execute(command, mode=self._submitMode)
+out, err = self.executeCommands(command, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 # write the received bytes to the local file
 # convert eol to "\n"
 return infodict
 print(fsinfo())
 del __os_, fsinfo
 """
-out, err = self._interface.execute(command, mode=self._submitMode)
+out, err = self.executeCommands(command, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 infodict = ast.literal_eval(out.decode("utf-8"))
 if infodict is None:
 return None
 return res
 print(get_device_data())
 del get_device_data
 """
-out, err = self._interface.execute(command, mode=self._submitMode)
+out, err = self.executeCommands(command, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 return ast.literal_eval(out.decode("utf-8"))
 def getBoardInformation(self):
 return res
 print(get_board_info())
 del get_board_info
 """
-out, err = self._interface.execute(command, mode=self._submitMode)
+out, err = self.executeCommands(command, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 return ast.literal_eval(out.decode("utf-8"))
 def getModules(self):
 @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, mode=self._submitMode)
+out, err = self.executeCommands(commands, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 modules = []
 for line in out.decode("utf-8").splitlines()[:-1]:
 )
 get_time()
 del get_time
 """
-out, err = self._interface.execute(command, mode=self._submitMode)
+out, err = self.executeCommands(command, mode=self._submitMode)
 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()
 now.tm_sec,
 now.tm_yday,
 now.tm_isdst,
 ),
 )
-out, err = self._interface.execute(command, mode=self._submitMode)
+out, err = self.executeCommands(command, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 ##################################################################
 ## Methods below implement package management related methods
 upip_install()
 del upip_install
 """.format(
 repr(packages)
 )
-return self._interface.execute(command, mode=self._submitMode, timeout=60000)
+return self.executeCommands(command, mode=self._submitMode, timeout=60000)
 def mipInstall(self, package, index=None, target=None, version=None, mpy=True):
 """
 Public method to install packages using 'mip'.
 mip_install()
 del mip_install
 """.format(
 parameterStr
 )
-return self._interface.execute(command, mode=self._submitMode, timeout=60000)
+return self.executeCommands(command, mode=self._submitMode, timeout=60000)
 def getLibPaths(self):
 """
 Public method to get the list of library paths contained in 'sys.path'.
 print([p for p in sys.path if p.endswith('/lib')])
 lib_paths()
 del lib_paths
 """
-out, err = self._interface.execute(command, mode=self._submitMode)
+out, err = self.executeCommands(command, mode=self._submitMode)
 if err:
 raise OSError(self._shortError(err))
 return ast.literal_eval(out.decode("utf-8"))

Mercurial Repositories > eric / file comparison

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

src/eric7/MicroPython/Devices/DeviceBase.py