第四章 实时数据读取：串口通信基础与pyserial实战
各位同学，今天我们来聊聊心电数据采集中最关键的一环——怎么把硬件设备上的数据，稳稳当当地读到电脑里来。
说实话，我刚开始做心电项目时，以为串口通信就是简单的「插上线、读数据」。结果呢？数据丢包、帧错位、缓冲区溢出……各种问题轮番上阵。那段时间我几乎每天都在跟串口较劲。今天就把这些经验分享给你们，少走点弯路。
4.1 串口通信基础：别小看这根线
串口通信，说白了就是一根线一根线地传数据。一次传一个比特，像排队过安检一样。虽然现在USB、蓝牙满天飞，但在医疗设备、工业控制领域，串口依然是老大哥。
为什么？因为它简单、可靠、实时性好。你想想看，心电信号每秒才几百个采样点，串口完全够用。
串口通信三要素：
- 波特率：每秒传输的比特数。心电设备常用115200或57600
- 数据位：通常是8位，刚好一个字节
- 停止位：1位或2位，用来分隔数据包
我在项目中遇到过最坑的事：设备说明书上写的是115200波特率，结果实际是57600。读出来的数据全是乱码。嗯，后来我养成了一个习惯——拿到新设备，先用示波器量一下实际波特率。
4.2 pyserial库使用：Python的串口瑞士军刀
Python里操作串口，pyserial是首选。安装很简单：
pip install pyserial基本用法，我给你们写个模板：
import serial
import serial.tools.list_ports
# 第一步：找到设备
ports = serial.tools.list_ports.comports()
for port in ports:
print(f"发现设备：{port.device} - {port.description}")
# 第二步：打开串口
ser = serial.Serial(
port='COM3', # Windows下是COM口，Linux下是/dev/ttyUSB0
baudrate=115200, # 波特率
bytesize=8, # 数据位
parity='N', # 校验位：N无校验，E偶校验，O奇校验
stopbits=1, # 停止位
timeout=0.1 # 超时时间，单位秒
)
# 第三步：读取数据
data = ser.read(10) # 读取10个字节
print(data.hex()) # 打印十六进制
# 用完记得关
ser.close()我的小技巧：timeout参数别设太大。0.1秒足够，设大了程序会卡死。我曾经设了5秒，调试时等得想砸电脑。
4.3 数据帧解析：从字节流中提取心电信号
串口传过来的是一堆字节，怎么从中提取出心电数据？这就要看数据帧格式了。
典型的心电数据帧长这样：
帧头(2字节) | 数据长度(1字节) | 数据体(N字节) | 校验和(1字节)
0xAA 0x55 0x0A 10个数据字节 CRC解析代码：
def parse_frame(data):
"""解析心电数据帧"""
if len(data) < 4:
return None
# 检查帧头
if data[0] != 0xAA or data[1] != 0x55:
return None
# 获取数据长度
data_len = data[2]
# 检查数据完整性
if len(data) < 4 + data_len:
return None
# 提取数据体
payload = data[3:3+data_len]
# 校验和检查（简单累加校验）
checksum = sum(payload) & 0xFF
if checksum != data[-1]:
print("校验和错误，数据可能损坏")
return None
return payload注意：帧头千万别用0x00或0xFF。我见过有人用0x00做帧头，结果数据里出现0x00就乱套了。选帧头要选数据中不常见的值。
4.4 缓冲区设计：给数据找个临时仓库
串口数据不是一口气全来的，而是一点一点地到。所以我们需要缓冲区，先把数据存起来，等凑够一帧再处理。
我常用的环形缓冲区：
class RingBuffer:
"""环形缓冲区，固定大小，自动覆盖旧数据"""
def __init__(self, size=4096):
self.buffer = bytearray(size)
self.head = 0 # 写指针
self.tail = 0 # 读指针
self.size = size
def write(self, data):
"""写入数据"""
for byte in data:
self.buffer[self.head] = byte
self.head = (self.head + 1) % self.size
# 如果写指针追上读指针，说明缓冲区满了
if self.head == self.tail:
self.tail = (self.tail + 1) % self.size
def read(self, length):
"""读取数据"""
if self.available() < length:
return None
data = bytearray()
for _ in range(length):
data.append(self.buffer[self.tail])
self.tail = (self.tail + 1) % self.size
return data
def available(self):
"""可读数据量"""
return (self.head - self.tail) % self.size为什么用环形缓冲区？因为内存分配是性能杀手。你想想看，心电数据每秒几百个包，如果用列表不断append，Python的内存管理会疯掉的。
4.5 丢包处理策略：数据丢了怎么办？
串口通信不是100%可靠的。电磁干扰、线缆松动、缓冲区溢出……丢包是常态。
我总结了几种处理策略：
| 策略 | 原理 | 适用场景 |
|---|---|---|
| 重传机制 | 发现丢包后，请求设备重发 | 实时性要求不高的场景 |
| 插值补全 | 用前后数据估算丢失的数据 | 心电信号，短时丢包可接受 |
| 丢弃处理 | 直接丢掉损坏的数据包 | 数据量大，丢包率低 |
| 冗余编码 | 发送时加入冗余信息，接收端纠错 | 强干扰环境 |
心电项目中，我推荐「插值补全+丢弃处理」组合拳。为什么？因为心电信号是连续的，丢几个点用线性插值就能补上。但如果连续丢包超过5个，就别补了，直接标记为无效数据。
丢包检测代码：
def detect_packet_loss(packet_seq, expected_seq):
"""检测是否丢包"""
if packet_seq == expected_seq:
return 0 # 正常
elif packet_seq > expected_seq:
loss_count = packet_seq - expected_seq
print(f"检测到丢包：{loss_count}个包")
return loss_count
else:
# 序列号回绕，说明是新的周期
return 0避坑指南：我曾经在强电磁环境下调试心电设备，丢包率高达30%。后来发现是串口线没屏蔽。换了带屏蔽的线，丢包率降到1%以下。所以，硬件问题先排查，别一上来就写复杂算法。
4.6 实战：完整的数据读取流程
把上面的知识点串起来，就是一个完整的数据读取流程：
import serial
import threading
from collections import deque
class ECGDataReader:
"""心电数据读取器"""
def __init__(self, port, baudrate=115200):
self.ser = serial.Serial(port, baudrate, timeout=0.05)
self.buffer = RingBuffer(8192)
self.data_queue = deque(maxlen=1000)
self.running = False
def start(self):
"""启动读取线程"""
self.running = True
self.thread = threading.Thread(target=self._read_loop)
self.thread.daemon = True
self.thread.start()
def _read_loop(self):
"""后台读取循环"""
while self.running:
# 读取原始字节
raw_data = self.ser.read(256)
if raw_data:
# 写入缓冲区
self.buffer.write(raw_data)
# 尝试解析帧
self._try_parse_frame()
def _try_parse_frame(self):
"""尝试从缓冲区解析数据帧"""
# 查找帧头
while self.buffer.available() >= 4:
# 检查帧头
if self.buffer.peek(2) == b'\xAA\x55':
# 读取帧长度
data_len = self.buffer.peek(3, 1)[0]
# 检查是否够一帧
if self.buffer.available() >= 4 + data_len:
frame = self.buffer.read(4 + data_len)
payload = parse_frame(frame)
if payload:
self.data_queue.append(payload)
else:
break
else:
# 跳过无效字节
self.buffer.read(1)
def stop(self):
"""停止读取"""
self.running = False
self.ser.close()这个类封装了串口读取、缓冲区管理、帧解析的全部逻辑。你只需要调用start()，然后从data_queue里取数据就行。
好了，这一章的内容就到这里。下一章我们会讲怎么把读到的数据实时绘制成波形图。记住，串口通信是心电系统的「咽喉」，这里出问题，后面全是白搭。动手试试吧，有问题随时问我。