feat: reconnect after serial lost connection

Author: horw

pytest-embedded-serial/pytest_embedded_serial/serial.py

@@ -189,20 +189,65 @@ def _event_loop(self):
                 return
 
             if _e == 'read':
-                if time.time() - last_data_time > 5.0:
-                    logging.warning('No data received from serial port for over 5 seconds.')
-                    last_data_time = time.time()
-
                 if self._block_reading:
                     continue
 
                 try:
                     s = self._s.read_all()
+
                     if s:
                         last_data_time = time.time()
 
+                    if time.time() - last_data_time > 15.0:
+                        logging.warning('No data received from serial port for over 15 seconds.')
+                        last_data_time = time.time()
+                except OSError as e:
+                    logging.error(f'OSError detected: {e}. Serial connection may be lost.')
+                    if self._s.closed:
+                        logging.error('Serial port is already closed. Exiting event loop.')
+                        return
+
+                    port = self._s.port
+                    port_config = {
+                        'baudrate': self._s.baudrate,
+                        'bytesize': self._s.bytesize,
+                        'parity': self._s.parity,
+                        'stopbits': self._s.stopbits,
+                        'timeout': self._s.timeout,
+                        'xonxoff': self._s.xonxoff,
+                        'rtscts': self._s.rtscts,
+                    }
+                    for attempt in range(1, 4):
+                        delay = attempt * 1.5
+                        logging.warning(
+                            f'Attempting to reconnect to serial port {port} (try {attempt}/3) after {delay}s...'
+                        )
+                        time.sleep(delay)
+                        try:
+                            self._s.close()
+                            self._s = pyserial.serial_for_url(port, **port_config)
+                            logging.info(f'Successfully reconnected to serial port {port}.')
+                            break
+                        except Exception as e:
+                            logging.warning(f'Reconnection attempt {attempt} failed: {e}')
+                    else:
+                        logging.error(
+                            f'Failed to reconnect to serial port {port} after 3 attempts. Exiting event loop.'
+                        )
+                        return
+
+                    continue
+
+                except Exception as e:
+                    logging.warning(
+                        'unknown error: %s.\nRecommend to close the serial process by `dut.serial.close()`', str(e)
+                    )
+                    return
+
+                try:
                     self._q.put(s)
-                except OSError:
+                except OSError as e:
+                    logging.warning(f'OSError. Error msg: {e}')
                     return
                 except Exception as e:
                     logging.warning(