Mercurial Repositories > eric / file comparison

comparison: src/eric7/MicroPython/MicroPythonDeviceInterface.py

src/eric7/MicroPython/MicroPythonDeviceInterface.py

branch
mpy_network
changeset 9990
54c614d91eff
parent 9989
286c2a21f36f
child 10008
c5bcafe3485c
equal deleted inserted replaced
9989:286c2a21f36f 9990:54c614d91eff
1 # -*- coding: utf-8 -*- 1 # -*- coding: utf-8 -*-
2 2
3 # Copyright (c) 2019 - 2023 Detlev Offenbach <detlev@die-offenbachs.de> 3 # Copyright (c) 2023 Detlev Offenbach <detlev@die-offenbachs.de>
4 # 4 #
5 5
6 """ 6 """
7 Module implementing some file system commands for MicroPython. 7 Module implementing an interface base class to talk to a connected MicroPython device.
8 """ 8 """
9 9
10 from PyQt6.QtCore import ( 10 from PyQt6.QtCore import QObject, pyqtSignal, pyqtSlot
11 QCoreApplication,
12 QEventLoop,
13 QObject,
14 QThread,
15 QTimer,
16 pyqtSignal,
17 pyqtSlot,
18 )
19
20 from eric7 import Preferences
21
22 from .MicroPythonSerialPort import MicroPythonSerialPort
23 11
24 12
25 class MicroPythonDeviceInterface(QObject): 13 class MicroPythonDeviceInterface(QObject):
26 """ 14 """
27 Class implementing an interface to talk to a connected MicroPython device. 15 Class implementing an interface to talk to a connected MicroPython device.
28 16
29 @signal executeAsyncFinished() emitted to indicate the end of an 17 @signal executeAsyncFinished() emitted to indicate the end of an
30 asynchronously executed list of commands (e.g. a script) 18 asynchronously executed list of commands (e.g. a script)
31 @signal dataReceived(data) emitted to send data received via the serial 19 @signal dataReceived(data) emitted to send data received via the connection
32 connection for further processing 20 for further processing
33 """ 21 """
34 22
35 executeAsyncFinished = pyqtSignal() 23 executeAsyncFinished = pyqtSignal()
36 dataReceived = pyqtSignal(bytes) 24 dataReceived = pyqtSignal(bytes)
37
38 PasteModePrompt = b"=== "
39 TracebackMarker = b"Traceback (most recent call last):"
40 25
41 def __init__(self, parent=None): 26 def __init__(self, parent=None):
42 """ 27 """
43 Constructor 28 Constructor
44 29
45 @param parent reference to the parent object 30 @param parent reference to the parent object
46 @type QObject 31 @type QObject
47 """ 32 """
48 super().__init__(parent) 33 super().__init__(parent)
49 34
50 self.__repl = parent 35 @pyqtSlot()
36 def connectToDevice(self, connection):
37 """
38 Public slot to connect to the device.
51 39
52 self.__blockReadyRead = False 40 @param connection name of the connection to be used
53
54 self.__serial = MicroPythonSerialPort(
55 timeout=Preferences.getMicroPython("SerialTimeout"), parent=self
56 )
57 self.__serial.readyRead.connect(self.__readSerial)
58
59 @pyqtSlot()
60 def __readSerial(self):
61 """
62 Private slot to read all available serial data and emit it with the
63 "dataReceived" signal for further processing.
64 """
65 if not self.__blockReadyRead:
66 data = bytes(self.__serial.readAll())
67 self.dataReceived.emit(data)
68
69 @pyqtSlot()
70 def connectToDevice(self, port):
71 """
72 Public slot to start the manager.
73
74 @param port name of the port to be used
75 @type str 41 @type str
76 @return flag indicating success 42 @return flag indicating success
77 @rtype bool 43 @rtype bool
44 @exception NotImplementedError raised to indicate that this method needs to
45 be implemented in a derived class
78 """ 46 """
79 return self.__serial.openSerialLink(port) 47 raise NotImplementedError(
48 "This method needs to be implemented in a derived class."
49 )
50
51 return False
80 52
81 @pyqtSlot() 53 @pyqtSlot()
82 def disconnectFromDevice(self): 54 def disconnectFromDevice(self):
83 """ 55 """
84 Public slot to stop the thread. 56 Public slot to disconnect from the device.
57
58 @exception NotImplementedError raised to indicate that this method needs to
59 be implemented in a derived class
85 """ 60 """
86 self.__serial.closeSerialLink() 61 raise NotImplementedError(
62 "This method needs to be implemented in a derived class."
63 )
87 64
88 def isConnected(self): 65 def isConnected(self):
89 """ 66 """
90 Public method to get the connection status. 67 Public method to get the connection status.
91 68
92 @return flag indicating the connection status 69 @return flag indicating the connection status
93 @rtype bool 70 @rtype bool
71 @exception NotImplementedError raised to indicate that this method needs to
72 be implemented in a derived class
94 """ 73 """
95 return self.__serial.isConnected() 74 raise NotImplementedError(
75 "This method needs to be implemented in a derived class."
76 )
77
78 return False
96 79
97 @pyqtSlot() 80 @pyqtSlot()
98 def handlePreferencesChanged(self): 81 def handlePreferencesChanged(self):
99 """ 82 """
100 Public slot to handle a change of the preferences. 83 Public slot to handle a change of the preferences.
101 """ 84 """
102 self.__serial.setTimeout(Preferences.getMicroPython("SerialTimeout")) 85 pass
103 86
104 def write(self, data): 87 def write(self, data):
105 """ 88 """
106 Public method to write data to the connected device. 89 Public method to write data to the connected device.
107 90
108 @param data data to be written 91 @param data data to be written
109 @type bytes or bytearray 92 @type bytes or bytearray
93 @exception NotImplementedError raised to indicate that this method needs to
94 be implemented in a derived class
110 """ 95 """
111 self.__serial.isConnected() and self.__serial.write(data) 96 raise NotImplementedError(
112 97 "This method needs to be implemented in a derived class."
113 def __pasteOn(self):
114 """
115 Private method to switch the connected device to 'paste' mode.
116
117 Note: switching to paste mode is done with synchronous writes.
118
119 @return flag indicating success
120 @rtype bool
121 """
122 if not self.__serial:
123 return False
124
125 pasteMessage = b"paste mode; Ctrl-C to cancel, Ctrl-D to finish\r\n=== "
126
127 self.__serial.clear() # clear any buffered output before entering paste mode
128 self.__serial.write(b"\x02") # end raw mode if required
129 written = self.__serial.waitForBytesWritten(500)
130 # time out after 500ms if device is not responding
131 if not written:
132 return False
133 for _i in range(3):
134 # CTRL-C three times to break out of loops
135 self.__serial.write(b"\r\x03")
136 written = self.__serial.waitForBytesWritten(500)
137 # time out after 500ms if device is not responding
138 if not written:
139 return False
140 QThread.msleep(10)
141 self.__serial.readAll() # read all data and discard it
142 self.__serial.write(b"\r\x05") # send CTRL-E to enter paste mode
143 self.__serial.readUntil(pasteMessage)
144
145 if self.__serial.hasTimedOut():
146 # it timed out; try it again and than fail
147 self.__serial.write(b"\r\x05") # send CTRL-E again
148 self.__serial.readUntil(pasteMessage)
149 if self.__serial.hasTimedOut():
150 return False
151
152 QCoreApplication.processEvents(
153 QEventLoop.ProcessEventsFlag.ExcludeUserInputEvents
154 ) 98 )
155 self.__serial.readAll() # read all data and discard it
156 return True
157
158 def __pasteOff(self):
159 """
160 Private method to switch 'paste' mode off.
161 """
162 if self.__serial:
163 self.__serial.write(b"\x04") # send CTRL-D to cancel paste mode
164
165 def __rawOn(self):
166 """
167 Private method to switch the connected device to 'raw' mode.
168
169 Note: switching to raw mode is done with synchronous writes.
170
171 @return flag indicating success
172 @rtype bool
173 """
174 if not self.__serial:
175 return False
176
177 rawReplMessage = b"raw REPL; CTRL-B to exit\r\n>"
178
179 self.__serial.write(b"\x02") # end raw mode if required
180 written = self.__serial.waitForBytesWritten(500)
181 # time out after 500ms if device is not responding
182 if not written:
183 return False
184 for _i in range(3):
185 # CTRL-C three times to break out of loops
186 self.__serial.write(b"\r\x03")
187 written = self.__serial.waitForBytesWritten(500)
188 # time out after 500ms if device is not responding
189 if not written:
190 return False
191 QThread.msleep(10)
192 self.__serial.readAll() # read all data and discard it
193 self.__serial.write(b"\r\x01") # send CTRL-A to enter raw mode
194 self.__serial.readUntil(rawReplMessage)
195 if self.__serial.hasTimedOut():
196 # it timed out; try it again and than fail
197 self.__serial.write(b"\r\x01") # send CTRL-A again
198 self.__serial.readUntil(rawReplMessage)
199 if self.__serial.hasTimedOut():
200 return False
201
202 QCoreApplication.processEvents(
203 QEventLoop.ProcessEventsFlag.ExcludeUserInputEvents
204 )
205 self.__serial.readAll() # read all data and discard it
206 return True
207
208 def __rawOff(self):
209 """
210 Private method to switch 'raw' mode off.
211 """
212 if self.__serial:
213 self.__serial.write(b"\x02") # send CTRL-B to cancel raw mode
214 self.__serial.readUntil(b">>> ") # read until Python prompt
215 self.__serial.readAll() # read all data and discard it
216 99
217 def probeDevice(self): 100 def probeDevice(self):
218 """ 101 """
219 Public method to check the device is responding. 102 Public method to check the device is responding.
220 103
221 If the device has not been flashed with a MicroPython formware, the 104 If the device has not been flashed with a MicroPython firmware, the
222 probe will fail. 105 probe will fail.
223 106
224 @return flag indicating a communicating MicroPython device 107 @return flag indicating a communicating MicroPython device
225 @rtype bool 108 @rtype bool
109 @exception NotImplementedError raised to indicate that this method needs to
110 be implemented in a derived class
226 """ 111 """
227 if not self.__serial: 112 raise NotImplementedError(
228 return False 113 "This method needs to be implemented in a derived class."
114 )
229 115
230 if not self.__serial.isConnected(): 116 return False
231 return False
232
233 # switch on raw mode
234 self.__blockReadyRead = True
235 ok = self.__pasteOn()
236 if not ok:
237 self.__blockReadyRead = False
238 return False
239
240 # switch off raw mode
241 QThread.msleep(10)
242 self.__pasteOff()
243 self.__blockReadyRead = False
244
245 return True
246 117
247 def execute(self, commands, *, mode="raw", timeout=0): 118 def execute(self, commands, *, mode="raw", timeout=0):
248 """ 119 """
249 Public method to send commands to the connected device and return the 120 Public method to send commands to the connected device and return the
250 result. 121 result.
251 122
252 If no serial connection is available, empty results will be returned. 123 If no connection is available, empty results will be returned.
253 124
254 @param commands list of commands to be executed 125 @param commands list of commands to be executed
255 @type str or list of str 126 @type str or list of str
256 @keyparam mode submit mode to be used (one of 'raw' or 'paste') (defaults to 127 @keyparam mode submit mode to be used (one of 'raw' or 'paste') (defaults to
257 'raw') 128 'raw')
259 @keyparam timeout per command timeout in milliseconds (0 for configured default) 130 @keyparam timeout per command timeout in milliseconds (0 for configured default)
260 (defaults to 0) 131 (defaults to 0)
261 @type int (optional) 132 @type int (optional)
262 @return tuple containing stdout and stderr output of the device 133 @return tuple containing stdout and stderr output of the device
263 @rtype tuple of (bytes, bytes) 134 @rtype tuple of (bytes, bytes)
135 @exception NotImplementedError raised to indicate that this method needs to
136 be implemented in a derived class
264 @exception ValueError raised in case of an unsupported submit mode 137 @exception ValueError raised in case of an unsupported submit mode
265 """ 138 """
139 raise NotImplementedError(
140 "This method needs to be implemented in a derived class."
141 )
142
266 if mode not in ("paste", "raw"): 143 if mode not in ("paste", "raw"):
267 raise ValueError("Unsupported submit mode given ('{0}').".format(mode)) 144 raise ValueError("Unsupported submit mode given ('{0}').".format(mode))
268 145
269 if mode == "raw": 146 return b"", b""
270 return self.__execute_raw(commands, timeout=timeout)
271 elif mode == "paste":
272 return self.__execute_paste(commands, timeout=timeout)
273 else:
274 # just in case
275 return b"", b""
276
277 def __execute_raw(self, commands, timeout=0):
278 """
279 Private method to send commands to the connected device using 'raw REPL' mode
280 and return the result.
281
282 If no serial connection is available, empty results will be returned.
283
284 @param commands list of commands to be executed
285 @type str or list of str
286 @param timeout per command timeout in milliseconds (0 for configured default)
287 (defaults to 0)
288 @type int (optional)
289 @return tuple containing stdout and stderr output of the device
290 @rtype tuple of (bytes, bytes)
291 """
292 if not self.__serial:
293 return b"", b""
294
295 if not self.__serial.isConnected():
296 return b"", b"Device not connected or not switched on."
297
298 result = bytearray()
299 err = b""
300
301 if isinstance(commands, str):
302 commands = [commands]
303
304 # switch on raw mode
305 self.__blockReadyRead = True
306 ok = self.__rawOn()
307 if not ok:
308 self.__blockReadyRead = False
309 return (b"", b"Could not switch to raw mode. Is the device switched on?")
310
311 # send commands
312 QThread.msleep(10)
313 for command in commands:
314 if command:
315 commandBytes = command.encode("utf-8")
316 self.__serial.write(commandBytes + b"\x04")
317 QCoreApplication.processEvents(
318 QEventLoop.ProcessEventsFlag.ExcludeUserInputEvents
319 )
320 ok = self.__serial.readUntil(b"OK")
321 if ok != b"OK":
322 self.__blockReadyRead = False
323 return (
324 b"",
325 "Expected 'OK', got '{0}', followed by '{1}'".format(
326 ok, self.__serial.readAll()
327 ).encode("utf-8"),
328 )
329
330 # read until prompt
331 response = self.__serial.readUntil(b"\x04>", timeout=timeout)
332 if self.__serial.hasTimedOut():
333 self.__blockReadyRead = False
334 return b"", b"Timeout while processing commands."
335 if b"\x04" in response[:-2]:
336 # split stdout, stderr
337 out, err = response[:-2].split(b"\x04")
338 result += out
339 else:
340 err = b"invalid response received: " + response
341 if err:
342 result = b""
343 break
344
345 # switch off raw mode
346 QThread.msleep(10)
347 self.__rawOff()
348 self.__blockReadyRead = False
349
350 return bytes(result), err
351
352 def __execute_paste(self, commands, timeout=0):
353 """
354 Private method to send commands to the connected device using 'paste' mode
355 and return the result.
356
357 If no serial connection is available, empty results will be returned.
358
359 @param commands list of commands to be executed
360 @type str or list of str
361 @param timeout per command timeout in milliseconds (0 for configured default)
362 (defaults to 0)
363 @type int (optional)
364 @return tuple containing stdout and stderr output of the device
365 @rtype tuple of (bytes, bytes)
366 """
367 if not self.__serial:
368 return b"", b""
369
370 if not self.__serial.isConnected():
371 return b"", b"Device not connected or not switched on."
372
373 if isinstance(commands, list):
374 commands = "\n".join(commands)
375
376 # switch on paste mode
377 self.__blockReadyRead = True
378 ok = self.__pasteOn()
379 if not ok:
380 self.__blockReadyRead = False
381 return (b"", b"Could not switch to paste mode. Is the device switched on?")
382
383 # send commands
384 QThread.msleep(10)
385 for command in commands.splitlines(keepends=True):
386 # send the data as single lines
387 commandBytes = command.encode("utf-8")
388 self.__serial.write(commandBytes)
389 QCoreApplication.processEvents(
390 QEventLoop.ProcessEventsFlag.ExcludeUserInputEvents
391 )
392 QThread.msleep(10)
393 ok = self.__serial.readUntil(commandBytes)
394 if ok != commandBytes:
395 self.__blockReadyRead = False
396 return (
397 b"",
398 "Expected '{0}', got '{1}', followed by '{2}'".format(
399 commandBytes, ok, self.__serial.readAll()
400 ).encode("utf-8"),
401 )
402
403 # switch off paste mode causing the commands to be executed
404 self.__pasteOff()
405 QThread.msleep(10)
406 # read until Python prompt
407 result = (
408 self.__serial.readUntil(b">>> ", timeout=timeout)
409 .replace(b">>> ", b"")
410 .strip()
411 )
412 if self.__serial.hasTimedOut():
413 self.__blockReadyRead = False
414 return b"", b"Timeout while processing commands."
415
416 # get rid of any OSD string
417 if result.startswith(b"\x1b]0;"):
418 result = result.split(b"\x1b\\")[-1]
419
420 if self.TracebackMarker in result:
421 errorIndex = result.find(self.TracebackMarker)
422 out, err = result[:errorIndex], result[errorIndex:]
423 else:
424 out = result
425 err = b""
426
427 self.__blockReadyRead = False
428 return out, err
429 147
430 def executeAsync(self, commandsList, submitMode): 148 def executeAsync(self, commandsList, submitMode):
431 """ 149 """
432 Public method to execute a series of commands over a period of time 150 Public method to execute a series of commands over a period of time
433 without returning any result (asynchronous execution). 151 without returning any result (asynchronous execution).
434 152
435 @param commandsList list of commands to be execute on the device 153 @param commandsList list of commands to be execute on the device
436 @type list of str 154 @type list of str
437 @param submitMode mode to be used to submit the commands 155 @param submitMode mode to be used to submit the commands (one of 'raw'
438 @type str (one of 'raw' or 'paste') 156 or 'paste')
157 @type str
158 @exception NotImplementedError raised to indicate that this method needs to
159 be implemented in a derived class
439 @exception ValueError raised to indicate an unknown submit mode 160 @exception ValueError raised to indicate an unknown submit mode
440 """ 161 """
162 raise NotImplementedError(
163 "This method needs to be implemented in a derived class."
164 )
165
441 if submitMode not in ("raw", "paste"): 166 if submitMode not in ("raw", "paste"):
442 raise ValueError("Illegal submit mode given ({0})".format(submitMode)) 167 raise ValueError(
443 168 "Unsupported submit mode given ('{0}').".format(submitMode)
444 if submitMode == "raw":
445 startSequence = [ # sequence of commands to enter raw mode
446 b"\x02", # Ctrl-B: exit raw repl (just in case)
447 b"\r\x03\x03\x03", # Ctrl-C three times: interrupt any running program
448 b"\r\x01", # Ctrl-A: enter raw REPL
449 b'print("\\n")\r',
450 ]
451 endSequence = [
452 b"\r",
453 b"\x04",
454 ]
455 self.__executeAsyncRaw(
456 startSequence
457 + [c.encode("utf-8") + b"\r" for c in commandsList]
458 + endSequence
459 ) 169 )
460 elif submitMode == "paste":
461 self.__executeAsyncPaste(commandsList)
462
463 def __executeAsyncRaw(self, commandsList):
464 """
465 Private method to execute a series of commands over a period of time
466 without returning any result (asynchronous execution).
467
468 @param commandsList list of commands to be execute on the device
469 @type list of bytes
470 """
471 if commandsList:
472 command = commandsList.pop(0)
473 self.__serial.write(command)
474 QTimer.singleShot(2, lambda: self.__executeAsyncRaw(commandsList))
475 else:
476 self.__rawOff()
477 self.executeAsyncFinished.emit()
478
479 def __executeAsyncPaste(self, commandsList):
480 """
481 Private method to execute a series of commands over a period of time
482 without returning any result (asynchronous execution).
483
484 @param commandsList list of commands to be execute on the device
485 @type list of str
486 """
487 self.__blockReadyRead = True
488 self.__pasteOn()
489 command = b"\n".join(c.encode("utf-8)") for c in commandsList)
490 self.__serial.write(command)
491 self.__serial.readUntil(command)
492 self.__blockReadyRead = False
493 self.__pasteOff()
494 self.executeAsyncFinished.emit

Mercurial Repository: eric

eric ide

mercurial