4. 串口通信编程实战

串口通信是工业上位机最基础也最容易“踩坑”的一类通信方式。它的物理层简单，但应用层没有天然的消息边界，因此“打开端口、配置参数、收发字节流、处理粘包与断帧”这四件事必须一次性做对，否则后续所有协议解析都会变得不可靠。本节以一个典型的设备心跳读取任务为主线，把串口编程的关键环节串起来：上位机周期性向设备发送心跳请求帧，设备返回心跳应答帧，上位机从字节流中正确切出完整帧并解析出设备状态。

一、任务对象与通信约定

假设我们面对一台支持串口通信的工业采集模块，通信参数为：波特率 9600、数据位 8、停止位 1、无校验、无流控。心跳协议约定如下（示例参数，用于教学演示）：

• 请求帧：AA 01 00 00 55，其中 AA 为帧头，01 为设备地址，00 为心跳命令，00 为数据长度，55 为帧尾。
• 应答帧：AA 01 00 04 S1 S2 S3 S4 CS 55，其中 S1~S4 为 4 字节状态数据，CS 为从设备地址到数据末字节的累加和校验。
• 超时约定：单次请求等待应答不超过 300 ms，连续 3 次超时判定设备离线。

这里先明确一个参照约定：本文所有“发送/接收”均以上位机为观察者，即“发送”指上位机向设备写字节，“接收”指上位机从串口读字节。后续图示与代码均遵循这一约定。

二、打开端口与配置参数

串口编程的第一步不是写数据，而是正确打开端口并设置参数。以 Python 的 pyserial 为例，打开端口时应显式指定所有通信参数，避免依赖默认值：

import serial

ser = serial.Serial(
    port='COM3',          # Windows 示例；Linux 下形如 /dev/ttyUSB0
    baudrate=9600,
    bytesize=serial.EIGHTBITS,
    parity=serial.PARITY_NONE,
    stopbits=serial.STOPBITS_ONE,
    timeout=0.3,          # 读超时 300 ms
    write_timeout=0.3     # 写超时 300 ms
)

其中 timeout 是读操作的最长阻塞时间，write_timeout 是写操作的最长阻塞时间。很多初学者只设置 timeout，结果在设备未就绪时写操作长时间阻塞，导致上位机界面卡死。两个超时都应设置，且量级要与协议节奏匹配：心跳周期通常为 1 s，单次等待 300 ms 是合理值。

三、收发字节流与粘包、断帧

串口是字节流设备，read() 返回的字节数并不等于一帧的长度。可能出现三种情况：一次读到半帧（断帧）、一次读到一帧半或两帧（粘包）、读到含噪声的无效数据。因此不能假设“读一次就是一帧”，而应把接收数据放入缓冲区，再按帧结构切分。

针对本节协议，切帧规则可以这样定义：在缓冲区中查找帧头 AA，再确认帧尾 55 存在，且长度满足最小帧长；若校验通过则取出一帧，否则丢弃帧头继续查找。下面给出一个可运行的心跳读取实现：

import serial, time

FRAME_HEAD = 0xAA
FRAME_TAIL = 0x55
REQ = bytes([0xAA, 0x01, 0x00, 0x00, 0x55])

def checksum(data: bytes) -> int:
    return sum(data) & 0xFF

def build_ack_ok(frame: bytes) -> bool:
    # 应答帧：AA 01 00 04 S1 S2 S3 S4 CS 55，共 10 字节
    if len(frame) != 10:
        return False
    if frame[0] != FRAME_HEAD or frame[-1] != FRAME_TAIL:
        return False
    if frame[1] != 0x01 or frame[2] != 0x00 or frame[3] != 0x04:
        return False
    return checksum(frame[1:9]) == frame[9]

def read_one_frame(ser, buf: bytearray, timeout_s: float = 0.3):
    deadline = time.time() + timeout_s
    while time.time() < deadline:
        chunk = ser.read(64)
        if chunk:
            buf.extend(chunk)
        # 尝试从缓冲区切出一帧
        while len(buf) >= 10:
            if buf[0] != FRAME_HEAD:
                buf.pop(0)
                continue
            candidate = bytes(buf[:10])
            if candidate[-1] == FRAME_TAIL and build_ack_ok(candidate):
                del buf[:10]
                return candidate
            # 帧头位置不对或校验失败，丢弃一个字节继续找
            buf.pop(0)
    return None

ser = serial.Serial('COM3', 9600, timeout=0.05)
ser.reset_input_buffer()
buf = bytearray()
fail_count = 0

while True:
    ser.write(REQ)
    frame = read_one_frame(ser, buf, timeout_s=0.3)
    if frame is None:
        fail_count += 1
        print(f"心跳超时，连续失败 {fail_count} 次")
        if fail_count >= 3:
            print("设备离线")
            break
    else:
        fail_count = 0
        s1, s2, s3, s4 = frame[4], frame[5], frame[6], frame[7]
        print(f"心跳正常，状态字节：{s1:02X} {s2:02X} {s3:02X} {s4:02X}")
    time.sleep(1.0)

这段代码体现了三个关键点：第一，read(64) 只负责把字节搬进缓冲区，不承担切帧职责；第二，切帧循环以帧头为锚点，校验失败时只丢弃一个字节而不是整段缓冲区，避免误丢后续有效帧；第三，超时计数与心跳周期解耦，3 次失败才判离线，避免偶发干扰导致误判。

四、心跳读取任务的状态流转

把上述逻辑抽象成状态视角，可以更清楚地看到“正常—超时—离线”的迁移条件。下图展示了心跳任务在运行过程中的状态与迁移关系，其中超时计数是核心变量。

五、结果观察与判断依据

运行上述程序后，正常情况下的输出类似：

心跳正常，状态字节：00 01 00 00
心跳正常，状态字节：00 01 00 00
心跳正常，状态字节：00 01 00 00

如果设备断电或串口线松动，输出会变为：

心跳超时，连续失败 1 次
心跳超时，连续失败 2 次
心跳超时，连续失败 3 次
设备离线

判断依据可以归纳为三条：第一，是否能在 300 ms 内从缓冲区切出长度 10、帧头帧尾正确、校验通过的帧；第二，状态字节是否符合设备手册定义（例如 S1 为运行状态、S2 为告警标志）；第三，连续超时次数是否达到 3。若前两条满足而第三条未触发，说明通信链路正常；若频繁出现“校验失败但帧头帧尾正确”，则要检查波特率是否匹配、地线是否共地、是否存在电磁干扰。

六、可直接复用的心跳任务检查清单

把以上环节落实后，串口心跳读取就不再是“能通就行”的碰运气，而是一个参数明确、边界清晰、异常可判定的工程任务。后续更复杂的命令交互与控制框架，都可以建立在这套“打开—配置—收发—切帧—判定”的稳定基础之上。