有没有办法“暂停”pyserial 的 ReaderThread 以允许直接读取串行端口答案

【问题标题】：Is there a way to 'pause' pyserial's ReaderThread to allow a direct read of a serial port有没有办法“暂停”pyserial 的 ReaderThread 以允许直接读取串行端口
【发布时间】：2022-10-07 00:12:58
【问题描述】：

我有一个我正在玩的 gui，它使用 pyserial。在其中，我使用 pyserial 的 ReaderThread 来监视我的串行设备的串行输出并将其打印在控制台窗口上。

我还使用 pyserial 的 Serial() 实现向串行设备发送命令。

通常我不需要获取对 ser.write() 的响应，只需让 ReaderThread 处理它。

但是，现在有些情况下，我希望暂停 ReaderThread 对变量执行 ser.read()，对变量执行操作，然后取消暂停 ReaderThread 以使其继续执行此操作。

尝试了 ReaderThread.stop()，但它似乎正在断开连接。

还尝试创建我自己的具有互斥锁的 readerThread.run() 函数并用它替换 run 方法，但这有点松鼠。

我错过了一个简单的方法吗？

我会创建一个LineReader （或类似的）来处理传入的流，并在handle_line 方法内，检查是否将行发送到默认路径或您的自定义处理程序的标志，即：handle_line(self, data): if flag: do_something(data); else: some_signal.emit(data)

标签： python pyserial

【解决方案1】：

通过猴子修补 ReaderThread 类想出了一个方法：

def localinit(self, serial_instance, protocol_factory):
    """\
    Initialize thread.

    Note that the serial_instance' timeout is set to one second!
    Other settings are not changed.
    """
    super(ReaderThread, self).__init__()
    self.daemon = True
    self.serial = serial_instance
    self.protocol_factory = protocol_factory
    self.alive = True
    self._lock = threading.Lock()
    self._connection_made = threading.Event()
    self.protocol = None


    self._stop_event = threading.Event()

    print("****************************************************")
    print("                 localinit                          ")
    print("****************************************************")


def localrun(self):
    """Reader loop"""
    print("****************************************************")
    print("                 localrun                           ")
    print("****************************************************")
    if not hasattr(self.serial, 'cancel_read'):
        self.serial.timeout = 1
    self.protocol = self.protocol_factory()
    try:
        self.protocol.connection_made(self)
    except Exception as e:
        self.alive = False
        self.protocol.connection_lost(e)
        self._connection_made.set()
        return
    error = None
    self._connection_made.set()

    while self.alive and self.serial.is_open:
        while self._stop_event.is_set():
            #print("local run while")
            time.sleep(1)
       try:

            data = self.serial.read(self.serial.in_waiting or 1)
            
    except serial.SerialException as e:
        # probably some I/O problem such as disconnected USB serial
        # adapters -> exit
        error = e
        break
    else:
        if data:
            # make a separated try-except for called user code
            try:
                self.protocol.data_received(data)
                
            except Exception as e:
                error = e
                break
    
self.alive = False
self.protocol.connection_lost(error)
self.protocol = None


def localpause(self):
    self._stop_event.set()

def localresume(self):
    self._stop_event.clear()

然后在我的主要代码中：

        ReaderThread.run = localrun
        ReaderThread.__init__ = localinit  
        ReaderThread.pause = localpause
        ReaderThread.resume = localresume
        self.reader = ReaderThread(serialPort, SerialReaderProtocolLine)
        self.reader.start()


    def write_read_cmd(self, cmd_str):
        if(serialPort.isOpen() == False):
            print("Serial port not yet open")
            return
    
        app.serialcom.reader.pause()
        serialPort.reset_input_buffer()  # flush the buffer
        serialPort.reset_input_buffer()  # flush the buffer
        serialPort.reset_input_buffer()  # flush the buffer
        serialPort.write(bytes(cmd_str, encoding='utf-8'))
        line = serialPort.readline()
        app.serialcom.reader.resume()
        line = line.decode("utf-8")
        return line

【讨论】：