4、数据采集软件框架：基于Python的采集框架搭建、PySerial与Socket通信、数据帧协议设计

好，咱们进入第四章。这一章我打算聊聊数据采集的软件框架。说实话，这部分内容是我在实际项目中踩坑最多的地方。你想想看，硬件搭好了，传感器也接上了，结果软件一跑，数据丢了、帧错位了、通信超时了……这些我都经历过。

所以这一章，我会把我在多个项目中沉淀下来的采集框架思路，掰开揉碎了讲给你听。核心就三块：Python采集框架怎么搭、PySerial和Socket怎么用、数据帧协议怎么设计才靠谱。

4.1 采集框架的整体思路

我个人习惯把采集框架分成三层：

• 物理层：负责和硬件打交道。串口、网口、USB，都归这层管。
• 协议层：负责解析数据帧。校验、拆包、组包，都在这里完成。
• 应用层：负责数据存储、显示、转发。用户看到的东西，都在这层。

为什么要分层？说白了，就是为了解耦。你换了个传感器，只需要改物理层；你改了数据格式，只需要动协议层。我在一个项目中就吃过亏——所有代码揉在一起，换个串口波特率都要改半天。后来重构了，清爽多了。

下面这张图，是我常用的框架结构。你可以看到数据是怎么从硬件一路流到应用层的。

4.2 PySerial通信实战

串口通信，在嵌入式领域太常见了。我最早用Python做串口采集时，用的是pySerial库。这个库很成熟，但用不好也会出问题。

先看一个基础示例：

import serial
import time

# 初始化串口
ser = serial.Serial(
    port='COM3',        # Windows下用COM口，Linux下用/dev/ttyUSB0
    baudrate=115200,
    bytesize=8,
    parity='N',         # 无校验
    stopbits=1,
    timeout=0.5         # 超时0.5秒
)

# 发送查询指令
ser.write(b'\xAA\x55\x01\x00')

# 读取响应
response = ser.read(10)
print(response.hex())

ser.close()

这段代码看起来简单，对吧？但我告诉你，实际项目中往往没这么顺利。

正确的做法是：先读帧头，再根据帧头中的长度字段，读取剩余数据。我一般会写一个专门的接收函数：

def read_frame(ser, header_len=2, timeout=2.0):
    """读取完整的数据帧"""
    start_time = time.time()
    
    # 第一步：读取帧头
    header = b''
    while len(header) < header_len:
        chunk = ser.read(header_len - len(header))
        if not chunk:
            if time.time() - start_time > timeout:
                return None  # 超时
            continue
        header += chunk
    
    # 第二步：从帧头解析数据长度（假设第3个字节是长度）
    data_len = header[2]  # 这里根据你的协议调整
    data = b''
    
    # 第三步：读取剩余数据
    while len(data) < data_len:
        chunk = ser.read(data_len - len(data))
        if not chunk:
            if time.time() - start_time > timeout:
                return None
            continue
        data += chunk
    
    return header + data

嗯，这里要注意：timeout参数一定要设置。我曾经在一个项目中没设超时，结果硬件掉线了，程序直接卡死在那里。后来加了超时和重试机制，才稳定下来。

4.3 Socket通信要点

串口适合近距离、低速场景。但如果你要远程采集，或者数据量很大，Socket通信是更好的选择。

Python的socket库是标准库，不需要额外安装。我一般用TCP，因为它可靠。UDP虽然快，但丢包你得自己处理。

看一个TCP客户端的例子：

import socket

def tcp_client(host='192.168.1.100', port=5000):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(3.0)  # 设置超时
    
    try:
        sock.connect((host, port))
        print(f"已连接到 {host}:{port}")
        
        # 发送数据
        sock.send(b'\xAA\x55\x01\x00')
        
        # 接收数据
        data = sock.recv(1024)
        print(f"收到: {data.hex()}")
        
    except socket.timeout:
        print("连接超时")
    except ConnectionRefusedError:
        print("连接被拒绝")
    finally:
        sock.close()

另外，如果你要同时处理多个设备，可以考虑用select或asyncio。我在一个项目中同时采集8台设备的数据，就是用select做的多路复用，效果不错。

4.4 数据帧协议设计

这是本章的重头戏。数据帧协议，说白了就是你和硬件之间约定好的「对话格式」。格式设计得好，后面解析就轻松；设计得不好，调试起来想砸电脑。

我常用的帧格式是这样的：

字段	长度（字节）	说明
帧头	2	固定值，如 0xAA 0x55
数据长度	1	数据域的长度（不含帧头和校验）
命令字	1	0x01=查询，0x02=设置，0x03=响应
数据域	N	实际数据，长度由数据长度字段决定
校验	2	CRC16 或 累加和

为什么要有帧头？说白了，就是为了同步。硬件可能在上电时发送乱码，或者你中途加入监听，帧头能帮你找到一帧的起始位置。

为什么要有数据长度？因为数据域长度可变。没有这个字段，你根本不知道一帧什么时候结束。

为什么要有校验？因为通信信道可能受干扰。我在一个工厂项目里，串口线走的是强电桥架旁边，数据经常出错。加了CRC校验后，错误帧全部被过滤掉了。

下面是一个简单的帧解析函数：

def parse_frame(data):
    """解析数据帧，返回 (命令字, 数据域) 或 None"""
    if len(data) < 6:  # 至少要有帧头+长度+命令+校验
        return None
    
    # 检查帧头
    if data[0] != 0xAA or data[1] != 0x55:
        return None
    
    # 解析数据长度
    data_len = data[2]
    
    # 检查帧长度是否完整
    expected_len = 2 + 1 + 1 + data_len + 2  # 帧头+长度+命令+数据+校验
    if len(data) != expected_len:
        return None
    
    # 校验（这里用累加和，实际项目建议用CRC16）
    checksum = sum(data[:-2]) & 0xFFFF
    received_checksum = (data[-2] << 8) | data[-1]
    if checksum != received_checksum:
        return None  # 校验失败
    
    cmd = data[3]
    payload = data[4:4+data_len]
    
    return (cmd, payload)

4.5 完整采集流程示例

把上面这些串起来，一个完整的采集流程大概是这样的：

1. 初始化串口或Socket连接
2. 发送查询指令（组帧）
3. 接收响应数据（拆帧）
4. 校验数据完整性
5. 解析数据域，提取有效信息
6. 存储或转发
7. 循环执行步骤2-6

我一般会把这个流程封装成一个类，方便复用：

class DataCollector:
    def __init__(self, port, baudrate=115200):
        self.ser = serial.Serial(port, baudrate, timeout=0.5)
        self.buffer = b''  # 接收缓冲区
    
    def send_query(self, cmd=0x01, payload=b''):
        """发送查询指令"""
        frame = self._build_frame(cmd, payload)
        self.ser.write(frame)
    
    def receive_response(self):
        """接收并解析响应"""
        while True:
            chunk = self.ser.read(64)
            if not chunk:
                break
            self.buffer += chunk
            
            # 尝试从缓冲区中提取完整帧
            frame = self._extract_frame()
            if frame:
                return frame
        return None
    
    def _build_frame(self, cmd, payload):
        """组帧"""
        data_len = len(payload)
        frame = bytes([0xAA, 0x55, data_len, cmd]) + payload
        checksum = sum(frame) & 0xFFFF
        frame += checksum.to_bytes(2, 'big')
        return frame
    
    def _extract_frame(self):
        """从缓冲区提取完整帧"""
        # 查找帧头
        idx = self.buffer.find(b'\xAA\x55')
        if idx == -1:
            return None
        
        # 跳过帧头前的垃圾数据
        if idx > 0:
            self.buffer = self.buffer[idx:]
        
        # 检查长度是否足够
        if len(self.buffer) < 6:
            return None
        
        data_len = self.buffer[2]
        expected_len = 6 + data_len
        
        if len(self.buffer) < expected_len:
            return None
        
        frame = self.buffer[:expected_len]
        self.buffer = self.buffer[expected_len:]
        
        return frame

这个类虽然简单，但已经包含了采集框架的核心逻辑。你可以在它的基础上扩展，比如加日志、加多线程、加数据库存储。

好了，这一章的内容就到这里。数据采集框架这东西，说起来不难，但真正做好需要积累。我的建议是：先跑通一个最简单的串口收发，然后慢慢加功能。别一上来就想搞个大而全的框架，容易把自己绕进去。

---

公众号:蓝海资料掘金营，微信deep3321