在嵌入式系统、测试仪表和传感器网络等场景中，PySerial 是 Python 与串口设备通信的桥梁。本文聚焦于 pySerial在Python串口通信中的数据接收与常见问题解决实战指南，从基础配置到高效的数据读取，结合实战代码示例，帮助开发者快速定位与解决接收过程中的各种问题。
一、基础配置与端口发现
1.1 选择合适的串口与波特率
在实际设备中，正确的端口名称和波特率是第一步。正确的端口名称可以避免打开失败，在 Windows 可用 COMx，在 Linux/macOS 常见 /dev/ttyUSBx /dev/ttyACMx。波特率、数据位、停止位、校验位都要和设备规范一致。
为快速定位可用端口，可以使用 pySerial 提供的工具。移步扫描端口列表有助于确认连接状态。
from serial.tools import list_ports
ports = [p.device for p in list_ports.comports()]
print("可用端口:", ports)
1.2 打开端口并初步测试
打开端口时，timeout 设置很关键，它决定了阻塞行为并影响数据响应时间。
初始测试可以只打开端口并输出状态，确保 串口对象已创建且处于打开状态。
import serial
port = '/dev/ttyUSB0' # Linux/macOS 示例；Windows 为 'COM3' 等
ser = serial.Serial(port, baudrate=115200, timeout=1) # 带超时的阻塞模式
print("端口打开:", ser.is_open)
ser.close()
二、数据接收的基本流程
2.1 阻塞、带超时、非阻塞读取
理解三种读取模式的差异对稳定数据接收至关重要。timeout 参数决定阻塞时间，非 0 时将在达到超时后返回数据（可能为空）。
选择适合场景的模式：持续实时化数据常用带超时的阻塞模式；后台采集可用非阻塞模式结合轮询。
import serial
ser = serial.Serial('/dev/ttyUSB0', 115200, timeout=1) # 阻塞带超时
data = ser.read(100) # 最多读取 100 字节，若无数据则等待 1 秒
print(data)
ser.close()
2.2 持续读取与缓冲管理
长期运行的接收通常需要一个缓冲区来拼接分段数据。in_waiting 表示当前缓冲中的字节数，可以据此决定是否读取。
结果通常以行或帧形式呈现，构建缓冲区并按定界符提取是常用做法。
import serial, time
ser = serial.Serial('/dev/ttyUSB0', 115200, timeout=0) # 非阻塞
buffer = bytearray()
while True:if ser.in_waiting:chunk = ser.read(ser.in_waiting)buffer.extend(chunk)if b'\n' in buffer:line, _, rest = buffer.partition(b'\n')print("收到行:", line.decode(errors='ignore').strip())buffer = resttime.sleep(0.01)
三、常见问题与排错实战
3.1 无法接收到数据/端口未打开
常见原因包括 端口未打开、设备未连接、权限不足 等。先确认设备是否正确连接、端口名称是否正确，再检查操作系统权限。
排错步骤往往从最简单的检查开始：确保设备存在、端口能够打开、没有占用该端口的其他应用。
import serial
try:ser = serial.Serial('COM3', 9600, timeout=1)print("打开成功:", ser.is_open)
except serial.SerialException as e:print("打开失败:", e)
3.2 波特率或帧格式不匹配导致数据错乱
如果输出看起来乱序或包含特殊字符，很可能是波特率、数据位、停止位或校验位不匹配。也有可能接收到半包数据。
解决办法通常是和设备端参数逐项对照，使用一组已知好数据进行验证，并确保软硬件握手设置一致。
import serial
# 尝试一组常用组合
configs = [(115200, serial.EIGHTBITS, serial.STOPBITS_ONE, serial.PARITY_NONE),(9600, serial.EIGHTBITS, serial.STOPBITS_ONE, serial.PARITY_NONE),(19200, serial.SEVENBITS, serial.STOPBITS_TWO, serial.PARITY_EVEN),
]
for baud, bits, stop, parity in configs:try:s = serial.Serial('/dev/ttyUSB0', baudrate=baud, bytesize=bits,stopbits=stop, parity=parity, timeout=1)print("测试波特率组合:", baud)s.close()except serial.SerialException:pass
四、性能与鲁棒性优化
4.1 最小化阻塞、提高吞吐
对高频数据流，尽量使用非阻塞或极短的超时，并且按块读取以降低调用开销。
此外，对缓冲区进行按行或按帧处理，避免逐字节处理，有助于提升稳定性与吞吐。
import serial, time
ser = serial.Serial('/dev/ttyUSB0', 115200, timeout=0) # 非阻塞
buf = bytearray()
start = time.time()
while time.time() - start < 5: # 5 秒示例if ser.in_waiting:data = ser.read(ser.in_waiting)buf.extend(data)# 简单帧处理while b'\n' in buf:line, _, rest = buf.partition(b'\n')print(line.decode(errors='ignore').strip())buf = resttime.sleep(0.01)
4.2 数据完整性与错误处理
在串口通信中，保护数据完整性至关重要，应对解码错误、丢包、以及设备断连等异常情况。
实现思路包括：对解码过程使用 errors='ignore' 或 errors='replace'，在断连时优雅关闭并重连。
import serial, time
ser = serial.Serial('/dev/ttyUSB0', 115200, timeout=0.5)
while True:data = ser.read(64)if data:try:text = data.decode('utf-8')print(text.strip())except UnicodeDecodeError:# 忽略非法字节print(data.decode('utf-8', errors='ignore').strip())else:# 处理空数据情况time.sleep(0.01)
五、实战案例与帧解析
5.1 使用定界符或长度字段提取数据帧
在实际应用中，设备往往以定界符（如换行符）或固定帧长度发送数据。使用缓冲区拼接和分帧解析可以提高鲁棒性。
下面示例演示如何按行读取并解析简单帧。
import serial, time
ser = serial.Serial('/dev/ttyUSB0', 115200, timeout=0.2)
buffer = bytearray()
while True:data = ser.read(1024)if data:buffer.extend(data)while b'\n' in buffer:line, _, rest = buffer.partition(b'\n')payload = line.strip().decode('utf-8', errors='ignore')print("帧:", payload)buffer = resttime.sleep(0.01)
5.2 将二进制帧解析为结构化数据
对于二进制帧，需要根据协议定义字段长度和字节序，才能正确解码。
示例：假设设备发送 8 字节的帧，其中前 2 字节为帧头，接着 4 字节为数据，最后 2 字节为校验和。
import serial
ser = serial.Serial('/dev/ttyUSB0', 115200, timeout=1)
def parse_frame(frame_bytes):if len(frame_bytes) != 8:return Nonehead = frame_bytes[0:2]payload = frame_bytes[2:6]checksum = int.from_bytes(frame_bytes[6:8], 'big')return {'head': head, 'payload': payload, 'checksum': checksum}buf = bytearray()
while True:data = ser.read(1024)if data:buf.extend(data)while len(buf) >= 8:frame = buf[:8]parsed = parse_frame(frame)if parsed:print(parsed)buf = buf[8:]else:time.sleep(0.01)
