Python使用pyserial实现串口数据发送与接收

在嵌入式开发、工业物联网与仪器测控领域，串口通信（UART / RS-232 / RS-485）是最经典且广泛采用的物理层交互规范。在 Python 生态中，pyserial 库是处理串口通信的绝对标准库。本文将从串口发现、参数协商、字节流编码发送、多模式接收策略到多线程全双工架构，完整解析实现串口双向交互的核心技术要点与工程落地方法。

一、串口通信原理与全双工数据链路

串口通信属于字符或字节流串行传输协议，其物理底层主要依赖 TXD（发送数据引脚）、RXD（接收数据引脚）以及公共 GND（信号地线）形成双向回路。上位机与下位机之间在无需共享全局时钟信号的异步通信模式下，完全依赖预先约定的波特率（Baud Rate）来同步每个比特的时序周期。

在操作系统层面，串口由专门的驱动程序抽象为字符设备（Windows 下为 COMx，Linux 下为 /dev/ttyUSBx 或 /dev/ttySx）。Python 应用程序并不直接操作底层电气引脚，而是通过 pyserial 库调用操作系统内核提供的串口 API。操作系统内部维护了环形 FIFO 发送缓冲区与接收缓冲区，为全双工异步传输提供了弹性缓冲。

理解这一系统分层至关重要：当 Python 执行 ser.write() 时，数据实际上首先写入操作系统的发送缓冲区，再由硬件时钟逐步串行化移位到物理线上；同样，当外部设备送来数据时，硬件会自动将比特拼装为字节放入接收缓冲区，直到应用程序主动执行 ser.read() 读取。

二、环境准备与可用端口动态发现

使用前首先需要安装 pyserial。这里存在一个在初学者中极为普遍的命名陷阱：在 PyPI 软件源中，必须安装的包名是 pyserial，而在 Python 代码中导入的模块名是 serial。

在实际工程中，硬件设备的插拔可能会导致端口号动态变化，硬编码端口名容易引发找不到设备的异常。pyserial 提供了 serial.tools.list_ports 工具，能够跨平台枚举当前操作系统内挂载的所有串口及其硬件属性（包含 VID、PID 与描述信息）：

import serial.tools.list_ports

def scan_serial_ports():
    """扫描并列出系统中所有可用串口设备"""
    ports = serial.tools.list_ports.comports()
    if not ports:
        print("[提示] 未检测到任何可用串口设备，请检查数据线连接与驱动状态。")
        return []
    
    available_ports = []
    print(f"[发现] 共检测到 {len(ports)} 个串口设备:")
    for p in ports:
        port_info = {
            "device": p.device,          # 如 'COM3' 或 '/dev/ttyUSB0'
            "description": p.description,# 如 'USB-SERIAL CH340 (COM3)'
            "hwid": p.hwid               # 硬件 VID:PID 标识
        }
        available_ports.append(port_info)
        print(f"  - 端口: {p.device} | 描述: {p.description}")
    return available_ports

if __name__ == "__main__":
    scan_serial_ports()

示例输出展示了标准的设备枚举格式。通过匹配 description 或 hwid 中的芯片特征码（例如 CH340、CP2102、FT232），程序可以实现无须人工干预的串口自动连接。

三、串口初始化与连接参数配置

建立串口会话的核心是实例化 serial.Serial 类。一个成功的串口会话依赖于五个关键通信参数的严格匹配：端口号（port）、波特率（baudrate）、数据位（bytesize）、校验位（parity）和停止位（stopbits），工业上通常简记为“波特率 + 8-N-1”（即 8 个数据位、无校验、1 个停止位）。

1. 关键参数详解

2. 读超时（timeout）的三种工作机制

• timeout = None（默认阻塞模式）：调用 read(size) 时，如果缓冲区未满 size 字节，程序将无限期等待挂起，直到凑满所需字节数。此模式极易导致程序冻结。
• timeout = 0（非阻塞模式）：立即读取当前缓冲区内已有的全部或至多 size 字节，若缓冲区为空则瞬间返回空字节对象 b''。
• timeout = x（固定秒数超时）：在 x 秒内持续等待并收集数据；一旦时间到达设定阈值，即使尚未集齐 size 字节，也会立即截断并返回当前已收到的字节。这是绝大多数生产环境推荐的配置。

推荐使用 Python 的上下文管理器（with 语句）来操作串口，这样能确保在发生异常或程序退出时，操作系统的端口句柄被正确关闭与释放：

import serial

# 使用上下文管理器安全打开串口
try:
    with serial.Serial(
        port='COM3',
        baudrate=115200,
        bytesize=serial.EIGHTBITS,
        parity=serial.PARITY_NONE,
        stopbits=serial.STOPBITS_ONE,
        timeout=1.0
    ) as ser:
        if ser.is_open:
            print(f"[成功] 串口 {ser.name} 打开成功，波特率: {ser.baudrate}")
            # 可以在此执行发送与接收业务
except serial.SerialException as err:
    print(f"[错误] 无法打开串口: {err}")

四、串口数据发送规范与实战

在 pyserial 中发送数据，唯一的入参要求是字节对象（bytes 或 bytearray）。如果直接向 ser.write() 传入 Python 字符串（str），解释器将立即抛出 TypeError: unicode strings are not supported, please encode to bytes 异常。

1. 文本与 ASCII 指令发送

对于 AT 指令、CLI 控制台或可读文本协议，必须显式调用 .encode('utf-8') 或 .encode('ascii') 进行序列化。同时，绝大多数硬件固件依赖换行结束符来判断一条指令的终结，常见的有 （回车换行，CRLF）或单独的 （LF）：

# 1. 文本协议指令发送示例
command = "AT+VERSION
"
bytes_to_send = command.encode('utf-8')

# write() 返回成功写入发送缓冲区的字节总数
bytes_written = ser.write(bytes_to_send)
print(f"[发送] 已写入 {bytes_written} 字节数据: {bytes_to_send!r}")

# 强制排空系统发送缓冲区，确保数据即刻离开物理引脚
ser.flush()

2. 二进制协议与 Hex 数据包发送

在工业传感器（如 Modbus RTU）或自定义通信协议中，通常使用纯十六进制字节流进行通信。可以使用 bytes.fromhex() 将十六进制字符串快速转换为字节包，或者借助标准库 struct.pack() 组装具有特定字节序的二进制报文：

import struct

# 方式 A：直接构造纯 Hex 报文 (例如 Modbus 读保持寄存器 01 03 00 00 00 02 C4 0B)
hex_cmd = "01 03 00 00 00 02 C4 0B"
payload = bytes.fromhex(hex_cmd)
ser.write(payload)
ser.flush()

# 方式 B：使用 struct 打包结构化二进制报文 (大端序: 包头0xAA55, 命令码0x01, 浮点数参数25.4)
header = 0xAA55
msg_id = 0x01
temperature = 25.4
binary_packet = struct.pack(">H B f", header, msg_id, temperature)
ser.write(binary_packet)
ser.flush()

五、串口数据接收的三种核心策略

与发送相对，数据接收面临更复杂的时序与不确定性。由于串口通信是纯字节流传输，物理层不提供“消息边界”。pyserial 提供了多种接收 API，开发者需要根据通信协议的特征选用最合适的方案。

策略对比与场景剖析

1. 固定长度读取：ser.read(size)
   适合已知报文固定大小的协议（例如固定 8 字节的 Modbus 指令帧）。其运行机制是挂起当前线程，直到收到指定的 size 字节或触发 timeout。需要注意的是，当遇到干扰或硬件断开导致超时时，返回的字节数可能小于 size，因此必须校验 len(data) == size。
2. 按行文本读取：ser.readline()
   专门用于以 结尾的行文本数据（如 GPS NMEA 语句、调试 print 日志）。如果对方设备只发送 或没有发送换行符，readline() 将一直阻塞直到超时时间耗尽，这会导致严重的通信延迟。
3. 缓冲区探测读取：ser.in_waiting + ser.read()（高并发与流式首选）
   通过属性 ser.in_waiting 获取操作系统接收缓冲区当前已积攒的未读字节数。只有当 in_waiting > 0 时才调用 read(ser.in_waiting)，实现绝对零等待的非阻塞读取。收集到的字节流在 Python 应用层放入内存缓冲区进行滑动窗口解包。

接收数据解码与安全容错

在将接收到的字节解码为字符串时，外界电气噪声或下位机波特率瞬时抖动可能会引入非法字节。若直接使用 data.decode('utf-8')，遇到无法解析的字节序列时会直接抛出 UnicodeDecodeError 导致程序崩溃。生产代码应指定 errors='replace' 或 errors='ignore'：

# 接收与安全解码范式
if ser.in_waiting > 0:
    raw_bytes = ser.read(ser.in_waiting)
    
    # 方式 1：打印十六进制原始数据（排查通信协议底层的金标准）
    hex_str = raw_bytes.hex(' ').upper()
    print(f"[HEX 接收] {hex_str}")
    
    # 方式 2：解码为文本，遇到坏字节替换为  保证进程不崩溃
    text_content = raw_bytes.decode('utf-8', errors='replace')
    print(f"[文本 接收] {text_content.strip()}")

六、工程实践：多线程全双工双向交互完整实现

在真实的监控终端或人机交互系统中，发送（下发指令）与接收（持续回传）是完全异步且并行的。如果在主线程中简单使用死循环轮询，将导致界面卡死或无法及时响应用户键盘输入。

最经典的工业解耦方案是多线程架构：创建一个独立的守护线程专门负责监听串口接收队列，主线程则用于处理业务逻辑或接收用户输入。二者通过 threading.Event 实现优雅停机与资源回收。

import time
import threading
import serial

class SerialWorker:
    """工业级全双工串口通信管理类"""
    def __init__(self, port: str, baudrate: int = 115200, timeout: float = 0.5):
        self.port = port
        self.baudrate = baudrate
        self.timeout = timeout
        self.ser = None
        self.stop_event = threading.Event()
        self.recv_thread = None

    def connect(self) -> bool:
        """打开串口并启动后台监听线程"""
        try:
            self.ser = serial.Serial(
                port=self.port,
                baudrate=self.baudrate,
                timeout=self.timeout
            )
            print(f"[系统] 成功打开串口 {self.port}，波特率: {self.baudrate}")
            
            # 清理打开时残留的历史脏数据
            self.ser.reset_input_buffer()
            self.ser.reset_output_buffer()

            # 启动独立监听线程
            self.stop_event.clear()
            self.recv_thread = threading.Thread(target=self._receive_loop, daemon=True)
            self.recv_thread.start()
            return True
        except serial.SerialException as e:
            print(f"[系统] 串口连接失败: {e}")
            return False

    def send(self, data: str) -> bool:
        """向串口发送文本指令"""
        if not self.ser or not self.ser.is_open:
            print("[系统] 发送失败: 串口未连接")
            return False
        try:
            payload = (data + "
").encode('utf-8')
            self.ser.write(payload)
            self.ser.flush()
            print(f">>> [发送]: {data}")
            return True
        except serial.SerialException as e:
            print(f"[系统] 发送异常: {e}")
            return False

    def _receive_loop(self):
        """后台持续运行的接收工作循环"""
        print("[监听] 串口接收守护线程已就绪...")
        while not self.stop_event.is_set():
            try:
                if self.ser and self.ser.is_open and self.ser.in_waiting > 0:
                    # 探测到有就绪数据，即刻读出
                    raw_data = self.ser.read(self.ser.in_waiting)
                    if raw_data:
                        text = raw_data.decode('utf-8', errors='replace').strip()
                        print(f"<<< [接收]: {text} (HEX: {raw_data.hex(' ').upper()})")
                else:
                    # 缓冲无数据时让出 CPU 时间片，避免高频自旋
                    time.sleep(0.01)
            except serial.SerialException as e:
                print(f"[监听] 通信链路中断: {e}")
                break

    def disconnect(self):
        """安全终止线程并释放串口资源"""
        print("[系统] 正在关闭串口与释放资源...")
        self.stop_event.set()
        if self.recv_thread and self.recv_thread.is_alive():
            self.recv_thread.join(timeout=1.0)
        if self.ser and self.ser.is_open:
            self.ser.close()
        print("[系统] 串口已安全断开")

if __name__ == "__main__":
    # 使用示例：连接并执行双向交互
    # 请根据实际环境修改端口号（Windows: 'COM3', Linux: '/dev/ttyUSB0'）
    worker = SerialWorker(port="COM3", baudrate=115200)
    if worker.connect():
        try:
            # 模拟业务指令交互
            worker.send("PING")
            time.sleep(1.0)
            worker.send("GET_STATUS")
            time.sleep(2.0)
        finally:
            # 确保无论如何都能执行到优雅退出
            worker.disconnect()

七、串口通信常见故障排查与全景总结

在硬件联调过程中，串口通信往往容易遇到各类隐蔽的软硬件问题。建立系统的故障排查矩阵能够极大缩短调试周期。

高频核心问题排查指南

• 端口占用与权限被拒（PermissionError / Access Denied）
  在 Windows 下，串口属于系统级独占设备，若串口调试助手或另一个 Python 实例未退出，新实例将绝对无法打开该端口。在 Linux 下，普通用户默认无权读写 /dev/ttyUSB* 设备节点，可通过命令 sudo usermod -aG dialout $USER 将当前用户加入 dialout 组，注销后重新登录即可彻底解决。
• 通信乱码与帧错误（Framing Error）
  乱码的首要原因是波特率不匹配（例如设备固件输出 9600，而 Python 以 115200 打开）。其次是物理地线未共地：上位机与下位机之间除了连接 TXD 和 RXD 外，必须连接公共接地线（GND），否则两个电路板之间的电平参考点不一致，逻辑 0 与 1 的识别阈值发生漂移。
• 打开串口导致单片机意外重启（DTR / RTS 陷阱）
  许多主流单片机开发板（如 Arduino Uno、ESP32 DevKit）在板载 USB 转串口芯片设计中，将 DTR/RTS 信号通过电容偶联至主控的 RESET 硬件复位引脚，以便实现一键下载固件。而 pyserial 在 open() 时默认会将 DTR 拉低，导致开发板在 Python 刚连上的瞬间发生硬复位。若要抑制此行为，应在构造实例时设置 dsrdtr=False, rtscts=False，或在 open() 之前显式将属性置为 False。
• 数据粘包与断包处理
  任何基于字节流的通信协议都必须面对“粘包”（两个数据包被合并一次读出）与“断包”（一个数据包被切断为两次读出）。在设计通信协议时，严禁假设单次 read() 刚好获取完整数据包，推荐在应用层引入成熟的封包机制：固定包头标志（如 0xAA 0x55）+ 载荷长度字段 + 协议数据 + CRC16 校验码。