Python串口通信实战：高效接收与解析外部设备数据流34

在现代工业控制、物联网(IoT)设备互联、嵌入式系统调试以及各种智能硬件开发中，串口通信（Serial Communication）扮演着不可或缺的角色。它是一种简单、稳定且广泛支持的数据传输方式。对于专业的程序员而言，掌握如何利用Python高效地进行串口数据接收与处理，无疑是一项核心技能。

Python凭借其简洁的语法、丰富的第三方库以及强大的跨平台能力，成为了实现串口通信的理想选择。其中，pyserial库是Python进行串口操作的事实标准，它提供了全面且易于使用的API，让开发者能够轻松地与各种串口设备进行交互。

本文将深入探讨如何使用Python的pyserial库来接收串口数据。我们将从基础概念讲起，逐步讲解库的安装、基本使用、各种数据接收模式，直到高级的并发处理和数据解析技巧，旨在为读者提供一份全面且实用的指南。

一、 串口通信基础：理解你的“管道”

在开始编写代码之前，我们需要对串口通信的一些基本概念有所了解。这就像在铺设管道之前，你需要知道水压、管道尺寸和连接方式一样。

1.1 什么是串口通信？

串口通信是一种逐位（bit by bit）传输数据的通信方式。相对于并行通信（一次传输多个位），串口通信只需要更少的导线，因此成本更低，更适合长距离传输。

1.2 关键参数

配置串口时，有几个关键参数必须与外部设备保持一致，否则将无法正确通信：
波特率 (Baud Rate)：表示每秒传输的位数。常见的有9600、19200、38400、115200等。
数据位 (Data Bits)：表示每个数据帧中实际数据位的数量，通常为8位。
停止位 (Stop Bits)：用于表示一个数据帧的结束，通常为1位或2位。
奇偶校验位 (Parity Bit)：用于检测数据传输过程中是否发生错误。选项包括无校验(None)、奇校验(Odd)、偶校验(Even)等。
流控制 (Flow Control)：控制数据传输速率，防止数据溢出。分为硬件流控制(RTS/CTS)和软件流控制(XON/XOFF)，也可以选择无流控制。

1.3 串口设备标识

在不同的操作系统中，串口设备的标识方式不同：
Windows：通常是COMx，如COM1, COM2等。
Linux/macOS：通常是/dev/ttySx (物理串口) 或 /dev/ttyUSBx (USB转串口适配器)，如/dev/ttyS0, /dev/ttyUSB0等。

二、 pyserial库入门：搭建你的Python串口环境

pyserial是Python中用于串口通信的第三方库。首先，我们需要安装它。

2.1 安装pyserial

打开你的终端或命令提示符，运行以下命令：pip install pyserial

2.2 打开与关闭串口

使用pyserial的第一步是创建一个Serial对象，并指定串口名称和配置参数。推荐使用Python的with语句来管理串口资源，确保串口在操作完成后自动关闭，即使发生异常也能正常释放。import serial
import .list_ports
# 1. 列出所有可用的串口 (可选，但推荐用于查找串口)
def list_available_ports():
ports = ()
if not ports:
print("未找到任何串口设备。")
return []
print("可用串口设备列表:")
for port, desc, hwid in sorted(ports):
print(f" {port}: {desc} [{hwid}]")
return [ for port in ports]
# 假设你的设备连接在 'COM3' 或 '/dev/ttyUSB0'
# 请根据你的实际情况修改
SERIAL_PORT = 'COM3' # Windows 示例
# SERIAL_PORT = '/dev/ttyUSB0' # Linux/macOS 示例
BAUD_RATE = 9600
try:
# 使用 with 语句，确保串口资源自动释放
with (
port=SERIAL_PORT,
baudrate=BAUD_RATE,
bytesize=, # 数据位: 8
parity=serial.PARITY_NONE, # 校验位: 无
stopbits=serial.STOPBITS_ONE, # 停止位: 1
timeout=1 # 读取超时时间，单位秒
) as ser:
if ():
print(f"成功打开串口: {SERIAL_PORT}")
print(f"串口配置: {ser.get_settings()}")
# 在这里进行数据读取操作
# ...
else:
print(f"未能打开串口: {SERIAL_PORT}")
except as e:
print(f"串口操作异常: {e}")
except FileNotFoundError:
print(f"串口 '{SERIAL_PORT}' 不存在或无法访问。")
except Exception as e:
print(f"发生未知错误: {e}")
# 调用函数列出串口
# available_ports = list_available_ports()
# if available_ports:
# print(f"你可以尝试使用其中的一个串口，例如: {available_ports[0]}")

重要提示：

timeout参数非常重要。如果设置为None，read()等函数将无限期阻塞，直到读取到数据。如果设置为0，则是非阻塞模式，没有数据立即返回空。设置为正数，则表示等待数据的时间上限。
在Linux下，如果遇到“Permission Denied”错误，可能需要将当前用户添加到dialout用户组：sudo usermod -a -G dialout $USER，然后重新登录。

三、 数据接收详解：多种读取模式

pyserial提供了多种方式来从串口读取数据，以适应不同的应用场景。理解这些方法的异同至关重要。

3.1 字节流与字符串编码

串口接收到的数据始终是字节（bytes）类型。在Python中，如果你需要将这些字节数据处理为可读的字符串，必须进行解码 (decode)。解码时需要指定正确的编码格式（如'utf-8', 'ascii', 'latin-1'等），这取决于发送设备的数据编码方式。

3.2 常见数据接收方法

3.2.1 (size=1)：读取指定字节数

这是最基本的读取方法，它尝试从串口缓冲区读取size个字节。如果设置了timeout，它会等待直到读取到足够多的字节或超时。如果超时，它会返回已读取到的字节（可能是空字节串）。# 假设串口已成功打开
# with (...) as ser:
# 读取10个字节
received_bytes = (10)
print(f"接收到 {len(received_bytes)} 字节数据: {received_bytes}")

# 尝试解码 (假设是ASCII编码)
try:
received_string = ('ascii')
print(f"解码后的字符串: {received_string}")
except UnicodeDecodeError:
print("无法解码为ASCII字符串，可能数据编码不匹配。")

3.2.2 (eol=b'')：读取一行数据

这个方法会一直读取，直到遇到换行符（默认是，即b''）或者超时。它非常适用于接收以换行符结尾的文本数据（如来自Arduino或传感器的日志信息）。返回的数据会包含换行符。# 假设串口已成功打开
# with (...) as ser:
print("等待接收一行数据...")
line_bytes = ()
if line_bytes:
print(f"接收到一行字节数据: {line_bytes}")
try:
line_string = ('utf-8').strip() # strip() 去除首尾空白，包括换行符
print(f"解码后的字符串: {line_string}")
except UnicodeDecodeError:
print("解码失败，尝试其他编码或检查数据。")
else:
print("超时，未接收到数据。")

注意： 如果你的设备使用其他行结束符（如\r），你需要将其指定为eol参数，例如 (eol=b'\r')。

3.2.3 ser.read_until(expected=b'', size=None)：读取直到指定序列

与readline()类似，但可以指定任意字节序列作为结束符。例如，你可以读取直到收到b'END'。size参数可以限制最大读取字节数。# 假设串口已成功打开
# with (...) as ser:
print("等待接收直到b'READY'...")
data_until_ready = ser.read_until(expected=b'READY', size=50) # 最多读取50字节
if data_until_ready:
print(f"接收到数据直到'READY': {data_until_ready}")
try:
print(f"解码后的字符串: {('utf-8')}")
except UnicodeDecodeError:
pass # 可能是二进制数据，不强行解码
else:
print("超时，未接收到'READY'序列。")

3.2.4 ser.in_waiting：检查缓冲区数据量

这个属性返回串口输入缓冲区中等待读取的字节数。它不会阻塞，可以用于非阻塞地检查是否有数据到来。# 假设串口已成功打开
# with (...) as ser:
print("等待数据中...")
while True:
if ser.in_waiting > 0:
# 缓冲区有数据，可以开始读取
data = (ser.in_waiting) # 一次性读取所有可用数据
print(f"接收到 {len(data)} 字节: {data}")
# 处理数据
# ...
else:
# 缓冲区为空，可以进行其他操作或短暂休眠
(0.1) # 避免CPU空转

四、 高级接收技巧与最佳实践

对于需要长时间运行、实时响应或处理复杂数据流的应用程序，仅仅依靠简单的读取方法是不够的。我们需要更高级的技巧。

4.1 多线程接收：保持主程序响应

在许多应用中（例如带有GUI的应用程序），你不能让串口读取操作阻塞主线程，否则程序会“卡死”。此时，使用多线程是最佳解决方案。一个单独的线程负责监听和读取串口数据，然后将数据传递给主线程进行处理。import serial
import .list_ports
import threading
import time
import queue # 用于线程间安全通信
class SerialReader():
def __init__(self, port, baudrate, data_queue):
super().__init__()
= port
= baudrate
self.data_queue = data_queue
= None
= True
def run(self):
try:
= (
port=,
baudrate=,
timeout=1 # 设置超时，让read()有机会返回，从而检查状态
)
if ():
print(f"[{}] 串口 {} 成功打开。")
while :
try:
# 尝试读取一行数据
data_bytes = ()
if data_bytes:
# 假设是UTF-8编码的字符串，并去除首尾空白
data_str = ('utf-8').strip()
(data_str) # 将数据放入队列
# print(f"[{}] 放入队列: {data_str}") # Debug

except :
# print(f"[{}] 读取超时，继续等待...") # Debug
continue # 超时是正常的，继续循环
except UnicodeDecodeError:
print(f"[{}] 解码错误，可能接收到非UTF-8数据: {data_bytes}")
except Exception as e:
print(f"[{}] 读取线程发生异常: {e}")
break # 发生严重错误，退出循环
else:
print(f"[{}] 无法打开串口 {}。")
except as e:
print(f"[{}] 串口初始化异常: {e}")
finally:
if and ():
()
print(f"[{}] 串口 {} 已关闭。")
def stop(self):
= False
print(f"[{}] 停止信号已发送。")
# 主程序
if __name__ == "__main__":
# 查找可用串口
print("正在查找可用串口...")
ports = ()
if not ports:
print("未找到任何串口设备。请检查设备连接和驱动。")
exit()
# 简单选择第一个找到的串口，或手动指定
# SERIAL_PORT = 'COM3' # 替换为你的实际串口
SERIAL_PORT = ports[0].device if ports else 'COM_UNKNOWN'
BAUD_RATE = 9600
print(f"选择串口: {SERIAL_PORT}")
# 创建一个线程安全的队列，用于线程间通信
data_queue = ()
# 创建并启动串口读取线程
reader_thread = SerialReader(SERIAL_PORT, BAUD_RATE, data_queue)
= True # 设置为守护线程，主程序退出时自动终止
()
print("主程序开始运行，等待接收串口数据 (按 Ctrl+C 停止)...")
try:
while True:
# 主程序从队列中获取数据
if not ():
received_data = ()
print(f"主程序收到数据: {received_data}")
# 在这里处理接收到的数据，例如：
# - 数据解析
# - 更新GUI界面
# - 存储到文件或数据库

# 示例：简单的数据解析
if ("TEMP:"):
try:
temp_value = float((":")[1])
print(f" --> 解析出温度: {temp_value}°C")
except ValueError:
print(" --> 温度数据格式错误。")
elif ("HUM:"):
try:
hum_value = float((":")[1])
print(f" --> 解析出湿度: {hum_value}%")
except ValueError:
print(" --> 湿度数据格式错误。")

(0.01) # 短暂休眠，避免CPU空转

except KeyboardInterrupt:
print("主程序收到终止信号。")
finally:
() # 通知读取线程停止
() # 等待读取线程结束
print("程序已退出。")

多线程注意事项：

使用进行线程间的数据传递是最佳实践，因为它本身是线程安全的。
设置SerialReader为守护线程（ = True）可以确保当主程序退出时，子线程也会被强制终止。
在SerialReader的run方法中，设置可以防止readline()无限期阻塞，从而有机会检查标志来优雅地停止线程。
主程序通过调用()来设置running标志，并通过()等待子线程完全结束。

4.2 数据解析与校验

接收到的原始数据通常需要进一步解析才能提取有意义的信息。常见的数据格式有：
文本格式 (ASCII/UTF-8)：如CSV（逗号分隔值）、JSON、自定义协议字符串。例如："TEMP,25.5,HUM,60.2"。
二进制格式：设备直接发送原始字节，需要根据协议文档进行位操作和字节转换。例如：Modbus、自定义封包。

在上面的多线程示例中，我们已经演示了简单的文本数据解析。对于更复杂的数据，你可能需要编写专门的解析函数，甚至利用结构体（如Python的struct模块）来处理二进制数据。import struct
# 假设接收到这样一串二进制数据：表示一个float和一个int
# 比如：b'\x40\x48\x00\x00\x0a\x00\x00\x00' (对应 float 3.14, int 10)
binary_data = b'\x40\x48\x00\x00\x0a\x00\x00\x00'
# 'f' 表示 float, 'i' 表示 int
# '' 表示大端字节序，这里假设是小端
try:
# 按照 'f' (float) 和 'i' (int) 的格式解包
parsed_data = ('

2026-03-02

上一篇：用Python玩转幻方：从原理到代码实现

下一篇：Python字符串分类统计与高效数据分析实战指南