Filter ANSI escape codes from the output.
+112 def clear(self): -113 """Clear the receive and transmit buffers.""" -114 self._serial.clear() -115 self._print_data = "" +@@ -760,27 +752,27 @@113 def clear(self): +114 """Clear the receive and transmit buffers.""" +115 self._serial.clear() +116 self._print_data = ""Parameters
117 def log_to_file(self, filename: Path | str, add_datetime: bool = False) -> Path: -118 """ -119 Log the received data to `filename` or close the log file. -120 -121 :param filename: file to log into. If `None`, the file is closed and -122 logging is disabled. -123 :param add_datetime: appends the date and time to the filename. -124 See `emdbg.utils.add_datetime` -125 -126 :return: the actual filename with date and time (if selected). -127 """ -128 if self._logfile is not None: -129 self._logfile.close() -130 -131 if filename is None: -132 self._logfile = None -133 return None -134 -135 filename = add_dt(filename) if add_datetime else Path(filename) -136 self._logfile = filename.open("wt") -137 return filename +@@ -815,18 +807,18 @@118 def log_to_file(self, filename: Path | str, add_datetime: bool = False) -> Path: +119 """ +120 Log the received data to `filename` or close the log file. +121 +122 :param filename: file to log into. If `None`, the file is closed and +123 logging is disabled. +124 :param add_datetime: appends the date and time to the filename. +125 See `emdbg.utils.add_datetime` +126 +127 :return: the actual filename with date and time (if selected). +128 """ +129 if self._logfile is not None: +130 self._logfile.close() +131 +132 if filename is None: +133 self._logfile = None +134 return None +135 +136 filename = add_dt(filename) if add_datetime else Path(filename) +137 self._logfile = filename.open("wt") +138 return filenameReturns
140 def read_lines(self, timeout: float = _TIMEOUT) -> str | None: -141 """ -142 Return any lines received within `timeout`. -143 Note that any ANSI escape codes (for color or cursor position) are -144 filtered out. -145 -146 :param timeout: seconds to wait until new lines arrive. -147 -148 :return: received lines or None on timeout -149 """ -150 lines = self._filter(self._read_packets(Nsh._NEWLINE, timeout)) -151 return self._join(lines) +@@ -860,33 +852,33 @@141 def read_lines(self, timeout: float = _TIMEOUT) -> str | None: +142 """ +143 Return any lines received within `timeout`. +144 Note that any ANSI escape codes (for color or cursor position) are +145 filtered out. +146 +147 :param timeout: seconds to wait until new lines arrive. +148 +149 :return: received lines or None on timeout +150 """ +151 lines = self._filter(self._read_packets(Nsh._NEWLINE, timeout)) +152 return self._join(lines)Returns
154 def wait_for(self, pattern: str, timeout: float = _TIMEOUT) -> str | None: -155 """ -156 Waits for a regex pattern to appear in a line in the stream. -157 This function reads any new received lines and searches for `pattern` in -158 every line. If the line matches, all lines are returned. -159 Note that any ANSI escape codes (for color or cursor position) are -160 filtered out. -161 -162 :param pattern: regex pattern to search for via `re.search`. To match -163 line beginnings and ends you must use `^pattern$`. -164 :param timeout: seconds to wait until new lines arrive. -165 -166 :return: received lines until matched pattern or None on timeout -167 """ -168 lines = [] -169 start = time.time() -170 while True: -171 if time.time() - start > timeout: -172 break -173 new_lines = self.read_lines(0) -174 lines.append(new_lines) -175 for line in new_lines: -176 if re.search(pattern, line): -177 return self._join(lines) -178 time.sleep(0.1) -179 LOGGER.warning(f"Waiting for '{pattern}' timed out after {timeout:.1f}s!") -180 return None +@@ -924,21 +916,21 @@155 def wait_for(self, pattern: str, timeout: float = _TIMEOUT) -> str | None: +156 """ +157 Waits for a regex pattern to appear in a line in the stream. +158 This function reads any new received lines and searches for `pattern` in +159 every line. If the line matches, all lines are returned. +160 Note that any ANSI escape codes (for color or cursor position) are +161 filtered out. +162 +163 :param pattern: regex pattern to search for via `re.search`. To match +164 line beginnings and ends you must use `^pattern$`. +165 :param timeout: seconds to wait until new lines arrive. +166 +167 :return: received lines until matched pattern or None on timeout +168 """ +169 lines = [] +170 start = time.time() +171 while True: +172 if time.time() - start > timeout: +173 break +174 new_lines = self.read_lines(0) +175 lines.append(new_lines) +176 for line in new_lines: +177 if re.search(pattern, line): +178 return self._join(lines) +179 time.sleep(0.1) +180 _LOGGER.warning(f"Waiting for '{pattern}' timed out after {timeout:.1f}s!") +181 return NoneReturns
182 def wait_for_prompt(self, timeout: float = _TIMEOUT) -> list[str]: -183 """ -184 Waits to the `nsh> ` prompt to arrive in the stream. -185 Note that any ANSI escape codes (for color or cursor position) are -186 filtered out. -187 -188 :param timeout: seconds to wait until the prompt arrives. -189 -190 :return: all lines until the prompt arrives. -191 """ -192 if prompts := self._read_packets(Nsh._NEWLINE + Nsh._PROMPT, timeout): -193 prompt = Nsh._PROMPT + Nsh._PROMPT.join(prompts) -194 return self._join(self._filter(prompt.split(Nsh._NEWLINE))) -195 LOGGER.warning(f"Waiting for 'nsh> ' prompt timed out after {timeout:.1f}s!") -196 return None +@@ -972,22 +964,22 @@183 def wait_for_prompt(self, timeout: float = _TIMEOUT) -> list[str]: +184 """ +185 Waits to the `nsh> ` prompt to arrive in the stream. +186 Note that any ANSI escape codes (for color or cursor position) are +187 filtered out. +188 +189 :param timeout: seconds to wait until the prompt arrives. +190 +191 :return: all lines until the prompt arrives. +192 """ +193 if prompts := self._read_packets(Nsh._NEWLINE + Nsh._PROMPT, timeout): +194 prompt = Nsh._PROMPT + Nsh._PROMPT.join(prompts) +195 return self._join(self._filter(prompt.split(Nsh._NEWLINE))) +196 _LOGGER.warning(f"Waiting for 'nsh> ' prompt timed out after {timeout:.1f}s!") +197 return NoneReturns
199 def command(self, command: str, timeout: float = _TIMEOUT) -> str | None: -200 """ -201 Send a command and return all lines until the next prompt. -202 If the command is asynchronous, you need to poll for new lines separately. -203 Note that any ANSI escape codes (for color or cursor position) are -204 filtered out. -205 -206 :param command: command string to send to the NSH. -207 :param timeout: seconds to wait until the prompt arrives. -208 -209 :return: all lines from the command issue until the next prompt arrives. -210 """ -211 self._serial.clear() -212 self._write_line(command) -213 if timeout is not None: -214 return self.wait_for_prompt(timeout) +@@ -1023,13 +1015,13 @@200 def command(self, command: str, timeout: float = _TIMEOUT) -> str | None: +201 """ +202 Send a command and return all lines until the next prompt. +203 If the command is asynchronous, you need to poll for new lines separately. +204 Note that any ANSI escape codes (for color or cursor position) are +205 filtered out. +206 +207 :param command: command string to send to the NSH. +208 :param timeout: seconds to wait until the prompt arrives. +209 +210 :return: all lines from the command issue until the next prompt arrives. +211 """ +212 self._serial.clear() +213 self._write_line(command) +214 if timeout is not None: +215 return self.wait_for_prompt(timeout)Returns
216 def command_nowait(self, command: str): -217 """ -218 Send a command to the NSH without waiting for a response. -219 -220 :param command: command string to send to the NSH. -221 """ -222 self.command(command, None) +@@ -1055,15 +1047,15 @@217 def command_nowait(self, command: str): +218 """ +219 Send a command to the NSH without waiting for a response. +220 +221 :param command: command string to send to the NSH. +222 """ +223 self.command(command, None)Parameters
225 def reboot(self, timeout: int = 15) -> str | None: -226 """ -227 Send the reboot command and wait for the reboot to be completed. -228 -229 :param timeout: seconds to wait until the prompt arrives. -230 -231 :return: all lines from reboot until the next prompt. -232 """ -233 return self.command("reboot", timeout) +@@ -1095,24 +1087,24 @@226 def reboot(self, timeout: int = 15) -> str | None: +227 """ +228 Send the reboot command and wait for the reboot to be completed. +229 +230 :param timeout: seconds to wait until the prompt arrives. +231 +232 :return: all lines from reboot until the next prompt. +233 """ +234 return self.command("reboot", timeout)Returns
235 def is_alive(self, timeout: float = _TIMEOUT, attempts: int = 4) -> bool: -236 """ -237 Check if the NSH is responding to newline inputs with a `nsh> ` prompt. -238 The total timeout is `attempts * timeout`! -239 -240 :param timeout: seconds to wait until the prompt arrives. -241 :param attempts: number of times to send a newline and wait. -242 :return: `True` is NSH responds, `False` otherwise -243 """ -244 self._serial.clear() -245 attempt = 0 -246 timeout = timeout / attempts -247 while attempt < attempts: -248 self._write_line("") -249 if self.wait_for_prompt(timeout) is not None: -250 return True -251 attempt += 1 -252 return False +@@ -1142,45 +1134,51 @@236 def is_alive(self, timeout: float = _TIMEOUT, attempts: int = 4) -> bool: +237 """ +238 Check if the NSH is responding to newline inputs with a `nsh> ` prompt. +239 The total timeout is `attempts * timeout`! +240 +241 :param timeout: seconds to wait until the prompt arrives. +242 :param attempts: number of times to send a newline and wait. +243 :return: `True` is NSH responds, `False` otherwise +244 """ +245 self._serial.clear() +246 attempt = 0 +247 timeout = timeout / attempts +248 while attempt < attempts: +249 self._write_line("") +250 if self.wait_for_prompt(timeout) is not None: +251 return True +252 attempt += 1 +253 return FalseReturns
@contextmanagerdef - nsh(serial: str, baudrate: int = 57600): + nsh(serial_or_port: str, baudrate: int = 57600):
256@contextmanager -257def nsh(serial: str, baudrate: int = 57600): -258 """ -259 Opens a serial port with the `serial` number and closes it again. -260 -261 :param serial: the serial number of the port to connect to. -262 :param baudrate: the baudrate to use. -263 -264 :raises `SerialException`: if serial port is not found. +-257@contextmanager +258def nsh(serial_or_port: str, baudrate: int = 57600): +259 """ +260 Opens a serial port with the `serial` number or filepath and closes it again. +261 +262 :param serial_or_port: the serial number or the filepath of the port to +263 connect to. +264 :param baudrate: the baudrate to use. 265 -266 :return: yields an initialized `Nsh` object. -267 """ -268 nsh = None -269 port = find_serial_port(serial) -270 try: -271 LOGGER.info(f"Starting on port '{serial}'..." if serial else "Starting...") -272 device = Serial(port.device, baudrate=baudrate) -273 reader_thread = ReaderThread(device, lambda: _NshReader(device)) -274 with reader_thread as reader: -275 nsh = Nsh(reader_thread, reader) -276 yield nsh -277 finally: -278 if nsh is not None: nsh.log_to_file(None) -279 LOGGER.debug("Stopping.") +266 :raises `SerialException`: if serial port is not found. +267 +268 :return: yields an initialized `Nsh` object. +269 """ +270 nsh = None +271 if "/" in serial_or_port: +272 ttyDevice = serial_or_port +273 else: +274 ttyDevice = find_serial_port(serial_or_port).device +275 try: +276 _LOGGER.info(f"Starting on port '{serial_or_port}'..." +277 if serial_or_port else "Starting...") +278 device = Serial(ttyDevice, baudrate=baudrate) +279 reader_thread = ReaderThread(device, lambda: _NshReader(device)) +280 with reader_thread as reader: +281 nsh = Nsh(reader_thread, reader) +282 yield nsh +283 finally: +284 if nsh is not None: nsh.log_to_file(None) +285 _LOGGER.debug("Stopping.")Opens a serial port with the
+serialnumber and closes it again.- -Opens a serial port with the
serialnumber or filepath and closes it again.Parameters
-
@@ -1198,55 +1196,6 @@- serial: the serial number of the port to connect to.
+- serial_or_port: the serial number or the filepath of the port to +connect to.
- baudrate: the baudrate to use.
Returns
- - - - def - patched_run(self): - - - -- -- - - -285def patched_run(self): -286 from serial import SerialException -287 self.serial.timeout = 0.1 -288 self.protocol = self.protocol_factory() -289 try: -290 self.protocol.connection_made(self) -291 except Exception as e: -292 self.alive = False -293 self.protocol.connection_lost(e) -294 self._connection_made.set() -295 return -296 error = None -297 self._connection_made.set() -298 while self.alive and self.serial.is_open: -299 try: -300 data = self.serial.read(self.serial.in_waiting or 1) -301 except SerialException as e: -302 if self.alive and "readiness" in str(e): -303 # LOGGER.debug(e) -304 continue -305 error = e -306 break -307 else: -308 if data: -309 try: -310 self.protocol.data_received(data) -311 except Exception as e: -312 error = e -313 break -314 self.alive = False -315 self.protocol.connection_lost(error) -316 self.protocol = None -