166
167 :return: received lines until matched pattern or None on timeout
168 """
-169 lines = []
+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 None
-182
-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 None
+174 if (new_lines := self.read_lines(0)) is not None:
+175 lines += new_lines
+176 if re.search(pattern, new_lines):
+177 return lines
+178 time.sleep(0.1)
+179 _LOGGER.warning(f"Waiting for '{pattern}' timed out after {timeout:.1f}s!")
+180 return None
+181
+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
+197
198
-199
-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)
-216
-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)
+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)
+215
+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)
+223
224
-225
-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)
-235
-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 False
+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)
+234
+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
@@ -866,19 +864,18 @@ Returns
166
167 :return: received lines until matched pattern or None on timeout
168 """
-169 lines = []
+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 None
+174 if (new_lines := self.read_lines(0)) is not None:
+175 lines += new_lines
+176 if re.search(pattern, new_lines):
+177 return lines
+178 time.sleep(0.1)
+179 _LOGGER.warning(f"Waiting for '{pattern}' timed out after {timeout:.1f}s!")
+180 return None
@@ -916,21 +913,21 @@ Returns
- 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 None
+ 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
Returns
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)
+ 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)
Returns
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)
+ 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)
Parameters
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)
+ 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)
Returns
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 False
+ 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
Returns
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 :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.")
+ 256@contextmanager
+257def nsh(serial_or_port: str, baudrate: int = 57600):
+258 """
+259 Opens a serial port with the `serial` number or filepath and closes it again.
+260
+261 :param serial_or_port: the serial number or the filepath of the port to
+262 connect to.
+263 :param baudrate: the baudrate to use.
+264
+265 :raises `SerialException`: if serial port is not found.
+266
+267 :return: yields an initialized `Nsh` object.
+268 """
+269 nsh = None
+270 if "/" in serial_or_port:
+271 ttyDevice = serial_or_port
+272 else:
+273 ttyDevice = find_serial_port(serial_or_port).device
+274 try:
+275 _LOGGER.info(f"Starting on port '{serial_or_port}'..."
+276 if serial_or_port else "Starting...")
+277 device = Serial(ttyDevice, baudrate=baudrate)
+278 reader_thread = ReaderThread(device, lambda: _NshReader(device))
+279 with reader_thread as reader:
+280 nsh = Nsh(reader_thread, reader)
+281 yield nsh
+282 finally:
+283 if nsh is not None: nsh.log_to_file(None)
+284 _LOGGER.debug("Stopping.")