1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
from .console import Console, ConsolePosix
from .transport import TransportError
def do_repl_main_loop(
state, console_in, console_out_write, *, escape_non_printable, code_to_inject, file_to_inject
):
while True:
try:
console_in.waitchar(state.transport.serial)
c = console_in.readchar()
if c:
if c in (b"\x1d", b"\x18"): # ctrl-] or ctrl-x, quit
break
elif c == b"\x04": # ctrl-D
# special handling needed for ctrl-D if filesystem is mounted
state.transport.write_ctrl_d(console_out_write)
elif c == b"\x0a" and code_to_inject is not None: # ctrl-j, inject code
state.transport.serial.write(code_to_inject)
elif c == b"\x0b" and file_to_inject is not None: # ctrl-k, inject script
console_out_write(bytes("Injecting %s\r\n" % file_to_inject, "utf8"))
state.transport.enter_raw_repl(soft_reset=False)
with open(file_to_inject, "rb") as f:
pyfile = f.read()
try:
state.transport.exec_raw_no_follow(pyfile)
except TransportError as er:
console_out_write(b"Error:\r\n")
console_out_write(er)
state.transport.exit_raw_repl()
else:
state.transport.serial.write(c)
n = state.transport.serial.inWaiting()
if n > 0:
dev_data_in = state.transport.serial.read(n)
if dev_data_in is not None:
if escape_non_printable:
# Pass data through to the console, with escaping of non-printables.
console_data_out = bytearray()
for c in dev_data_in:
if c in (8, 9, 10, 13, 27) or 32 <= c <= 126:
console_data_out.append(c)
else:
console_data_out.extend(b"[%02x]" % c)
else:
console_data_out = dev_data_in
console_out_write(console_data_out)
except OSError as er:
if _is_disconnect_exception(er):
return True
else:
raise
return False
def do_repl(state, args):
state.ensure_friendly_repl()
state.did_action()
escape_non_printable = args.escape_non_printable
capture_file = args.capture
code_to_inject = args.inject_code
file_to_inject = args.inject_file
print("Connected to MicroPython at %s" % state.transport.device_name)
print("Use Ctrl-] or Ctrl-x to exit this shell")
if escape_non_printable:
print("Escaping non-printable bytes/characters by printing their hex code")
if capture_file is not None:
print('Capturing session to file "%s"' % capture_file)
capture_file = open(capture_file, "wb")
if code_to_inject is not None:
code_to_inject = bytes(code_to_inject.replace("\\n", "\r\n"), "utf8")
print("Use Ctrl-J to inject", code_to_inject)
if file_to_inject is not None:
print('Use Ctrl-K to inject file "%s"' % file_to_inject)
console = Console()
console.enter()
def console_out_write(b):
console.write(b)
if capture_file is not None:
capture_file.write(b)
capture_file.flush()
try:
return do_repl_main_loop(
state,
console,
console_out_write,
escape_non_printable=escape_non_printable,
code_to_inject=code_to_inject,
file_to_inject=file_to_inject,
)
finally:
console.exit()
if capture_file is not None:
capture_file.close()
def _is_disconnect_exception(exception):
"""
Check if an exception indicates device disconnect.
Returns True if the exception indicates the device has disconnected,
False otherwise.
"""
if isinstance(exception, OSError):
if hasattr(exception, "args") and len(exception.args) > 0:
# IO error, device disappeared
if exception.args[0] == 5:
return True
# Check for common disconnect messages in the exception string
exception_str = str(exception)
disconnect_indicators = ["Write timeout", "Device disconnected", "ClearCommError failed"]
return any(indicator in exception_str for indicator in disconnect_indicators)
return False
|
