如何使用 pyserial 通过多线程写入/读取串行端口

Question

我目前在尝试使用 pySerial 写入串行设备时遇到问题。我希望能够通过读取端口来不断更新我的终端，并在单独的线程上处理串行设备写入，同时也能够通过主线程上的用户输入发送命令。一切都按预期运行，除了当我发送命令之一（cmdA 或 cmdB）时，我正在读取的串行输出不会改变（这是预期的行为，因为正在发送的命令会改变设备的状态，这反过来改变串行端口正在读取的设备输出）。尽管如此，设备似乎没有接收到我发送的命令，即使代码继续毫无例外地运行并且所有函数似乎都按照编写的方式执行。

这是我当前的代码：

一个 SerialMonitor 类，一旦找到一组“同步字节”，就可以读取串行端口并打印出特定数量的字节

# SerialMonitorTool.py

import threading
import time

import serial


class SerialMonitor(threading.Thread):
    SYNC_BYTES = b'\x90\xeb'

    def __init__(self, device='/dev/ttyUSB0', baudrate=115200, timeout=5):
        print("Initializing Serial Monitor")
        self._running = False
        self._name = 'SerialMonitorThread-{}'.format(device)
        self._device = serial.Serial(device, baudrate=baudrate, timeout=timeout)
        self._write_lock = threading.Lock()

        super().__init__(name=self._name)

    def write(self, user_input, encode=False, terminator=None):
        print("Locking for CMD Write...")
        self._write_lock.acquire()
        tx = user_input + terminator if terminator else user_input
        print(f"Writing CMD to device: {tx}")
        self._device.write(tx.encode() if encode else tx)
        print("CMD Written...")
        self._write_lock.release()
        print("CMD Write Lock Released...")

    def stop(self):
        self._running = False
        print('stop thread: ' + threading.current_thread().getName())
        self.join()

    def run(self):
        print('starting thread: ' + threading.current_thread().getName())
        self._running = True
        try:
            while self._running:
                self._device.reset_input_buffer()
                self._device.read_until(self.SYNC_BYTES)
                ser_bytes = self._device.read(35)
                print(f'\r{ser_bytes}', end='', flush=True)
                time.sleep(0.25)
        finally:
            self._device.close()

和主线程

# SerialMain.py

from SerialMonitorTool import *

cmdA = b'\x90\xeb\x01'
cmdB = b'\x90\xeb\x02'

monitor: SerialMonitor()


def print_help():
    print('Usage: cmd [ a | b ]')


def send_cmd(cmd):
    monitor.write(cmd)


def main():
    monitor.start()

    while True:
        try:
            user_input = input()

            if user_input == '?' or user_input == 'h' or user_input == 'help':
                print_help()

            elif user_input == 'q' or user_input == 'quit':
                break

            elif user_input.startswith('cmd '):
                cmd_type = user_input[len('cmd '):].split(' ')
                if cmd_type[0] == 'a':
                    send_cmd(cmdA)
                elif cmd_type[0] == 'b':
                    send_cmd(cmdB)

        except Exception as e:
            print(e)

    monitor.stop()


def process_args():
    # process arguments
    import argparse
    parser = argparse.ArgumentParser(description='Serial Test Tool')
    parser.add_argument(
        '-D', '--device',
        help='Use the specified serial device.',
        default='/dev/ttyUSB0',
        type=str
    )

    global monitor
    monitor = SerialMonitor()


if __name__ == "__main__":
    process_args()
    main()

Answer 1

您的 write 方法似乎有问题，尝试在 write 方法中注释所有与锁相关的代码或将锁语法按以下顺序排列。

    def write(self, user_input, encode=False, terminator=None):
        tx = user_input + terminator if terminator else user_input
        print(f"Writing CMD to device: {tx}")
        self._device.write(tx.encode() if encode else tx)
        print("CMD Written...")

        print("Locking for CMD Write...")
        self._write_lock.acquire()
        self._write_lock.release()
        print("CMD Write Lock Released...")