Python串口数据采集后通过MQTT上传至物联网平台

一、系统整体架构与数据流转

在典型的边缘物联网拓扑中，数据流通由物理感知层、边缘计算层以及云端物联网平台层三部分协同完成。各层之间通过不同的传输介质与通信协议完成解耦。

硬件传感器或控制器通常以轮询（Polled）或主动上报（Continuous Stream）模式发送原始字节流；运行Python的边缘主机充当本地网关，通过串口监听获取数据包，完成CRC校验、物理量数值换算、状态提取与时间戳打标，最后转换为结构化的遥测消息（Telemetry），交由MQTT客户端推送到远端服务器。

图 1：Python 串口数据采集与 MQTT 物联网平台上传系统架构图

二、环境准备与依赖安装

在开始编写代码前，需确认Python环境（推荐使用 Python 3.8 及以上版本），并安装两个关键的第三方库：用于操作底层串口通信的 pyserial，以及用于实现标准MQTT协议客户端的 paho-mqtt。

# 在终端或虚拟环境中执行安装命令
pip install pyserial paho-mqtt

在调试阶段，如果没有物理传感器或硬件开发板，可以采用以下测试方案：

• 虚拟串口工具：Windows环境下可使用 VSPDs (Virtual Serial Port Driver) 或开源的 com0com 成对创建虚拟串口（如 COM1 与 COM2 互联）；Linux/macOS环境下可使用 socat -d -d pty,raw,echo=0 pty,raw,echo=0 命令生成一对伪终端设备。
• 公共或本地MQTT Broker：推荐在本地使用 Docker 快速拉起一个 EMQX 容器（docker run -d --name emqx -p 1883:1883 -p 18083:18083 emqx/emqx:latest），或使用公共测试服务器（如 broker.emqx.io:1883）。

三、核心实现步骤与程序执行流程

整个程序的运行逻辑遵循严格的初始化、事件循环与容错处理流程。主逻辑需确保即使底层串口发生偶发拔插，或网络波动导致MQTT断连，采集脚本也不会异常崩溃退出。

图 2：Python 采集程序生命周期与数据处理循环执行流程

1. 串口数据读取与协议解析

工业现场的串口数据格式主要分为两类：ASCII文本行协议（常见于自研单片机、GPS模块如NMEA-0183等，以回车换行符结尾）与二进制协议（如标准 Modbus RTU）。

对于ASCII协议，可使用 ser.readline() 读取单行字符串并做字符串分割；而对于二进制协议，必须严格按照数据帧长读取字节流（bytes），并利用 struct.unpack 提取整型或浮点数据。

图 3：原始串口报文向 MQTT 标准遥测 JSON 负载转换示意

2. MQTT 客户端关键机制配置

在物联网数据上报中，使用 paho-mqtt 需要注意以下几个关键配置参数：

• Client ID（客户端标识）：每个连接到 Broker 的客户端必须具有唯一标识。在多网关部署时，若两台网关配置了相同的 Client ID，会导致双方互相踢下线并进入死循环重连。通常使用网关 MAC 地址或固定前缀 + UUID。
• QoS 等级（服务质量）：对于高频采集传感器（如每秒上报一次的温湿度），通常选用 QoS 0（最多交付一次），减少握手开销；对于关键事件、报警或累计电量等关键度量，建议选用 QoS 1（至少交付一次）。
• Clean Session（全新会话）：遥测上报型边缘设备通常设置为 clean_session=True，断开时不保留离线排队消息，防止重新上线时突发拥塞。
• 后台网络线程：推荐调用 client.loop_start() 启动独立的后台网络线程，自动处理心跳 Ping 请求（Keepalive）与网络断开重连，避免主线程因等待串口而阻塞 MQTT 心跳包发送导致超时断开。

四、完整端到端实现代码

下面提供一份完整的工程级代码。代码将串口监听、解析、MQTT发布集成在健壮的循环中，包含连接状态回调、自动重连逻辑与异常保护。

# -*- coding: utf-8 -*-
"""
Python 串口采集并上传至 MQTT 物联网平台完整示例
依赖: pip install pyserial paho-mqtt
"""

import time
import json
import logging
import serial
from serial.serialutil import SerialException
import paho.mqtt.client as mqtt

# 配置控制台日志格式
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger("Gateway")

# ==================== 配置区 ====================
# 串口通信配置
SERIAL_PORT = "/dev/ttyUSB0"   # Windows 平台换为 "COM3" 等
SERIAL_BAUDRATE = 9600         # 波特率
SERIAL_TIMEOUT = 1.0           # 读取超时时间(秒)

# MQTT Broker 配置
MQTT_BROKER = "broker.emqx.io" # 测试用公共Broker，生产环境替换为私有IP或域名
MQTT_PORT = 1883               # 默认非加密端口，TLS可配置为8883
MQTT_KEEPALIVE = 60            # 心跳间隔
MQTT_CLIENT_ID = "gateway_edge_node_001"
MQTT_USERNAME = None           # 若平台需要鉴权，填写对应用户名
MQTT_PASSWORD = None           # 若平台需要鉴权，填写对应密码

# 数据上报主题模板
TELEMETRY_TOPIC_FMT = "devices/{device_id}/telemetry"


# ==================== MQTT 回调函数 ====================
def on_connect(client, userdata, flags, rc):
    """连接建立回调函数"""
    if rc == 0:
        logger.info("MQTT Broker 连接成功 (RC=0)")
    else:
        logger.warning(f"MQTT Broker 连接失败，返回码: {rc}")

def on_disconnect(client, userdata, rc):
    """连接断开回调函数"""
    if rc != 0:
        logger.warning(f"MQTT 意外断开连接 (RC={rc})，后台线程将自动尝试重连")
    else:
        logger.info("MQTT 连接正常关闭")

def on_publish(client, userdata, mid):
    """消息发布确认回调"""
    logger.debug(f"消息已发送，Message ID: {mid}")


# ==================== 核心业务逻辑 ====================
def parse_serial_line(raw_line: str) -> dict:
    """
    解析串口数据行
    假设设备输出格式形如: #SN01,26.4,65.2,12.0\r\n
    """
    clean_line = raw_line.strip()
    if not clean_line.startswith("#"):
        raise ValueError(f"帧头不匹配，无效数据: {clean_line}")

    parts = clean_line[1:].split(",")
    if len(parts) < 4:
        raise ValueError(f"字段数量不足: {clean_line}")

    device_id = parts[0]
    temperature = float(parts[1])
    humidity = float(parts[2])
    voltage = float(parts[3])

    return {
        "device_id": device_id,
        "temperature": round(temperature, 2),
        "humidity": round(humidity, 2),
        "voltage": round(voltage, 2),
        "timestamp": int(time.time())
    }


def main():
    # 1. 初始化并连接 MQTT 客户端
    mqtt_client = mqtt.Client(client_id=MQTT_CLIENT_ID, clean_session=True)
    mqtt_client.on_connect = on_connect
    mqtt_client.on_disconnect = on_disconnect
    mqtt_client.on_publish = on_publish

    if MQTT_USERNAME and MQTT_PASSWORD:
        mqtt_client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)

    logger.info(f"正在连接 MQTT 服务器: {MQTT_BROKER}:{MQTT_PORT} ...")
    try:
        mqtt_client.connect(MQTT_BROKER, MQTT_PORT, keepalive=MQTT_KEEPALIVE)
    except Exception as e:
        logger.error(f"初始连接 MQTT 失败: {e}，将在后台重试")

    # 启动后台独立网络循环线程
    mqtt_client.loop_start()

    # 2. 串口采集主循环（带断线重试保护）
    ser = None
    logger.info(f"启动串口监听任务，目标端口: {SERIAL_PORT}...")

    try:
        while True:
            # 确保串口处于开启状态
            if ser is None or not ser.is_open:
                try:
                    ser = serial.Serial(
                        port=SERIAL_PORT,
                        baudrate=SERIAL_BAUDRATE,
                        bytesize=serial.EIGHTBITS,
                        parity=serial.PARITY_NONE,
                        stopbits=serial.STOPBITS_ONE,
                        timeout=SERIAL_TIMEOUT
                    )
                    logger.info(f"串口 {SERIAL_PORT} 打开成功")
                except SerialException as e:
                    logger.warning(f"打开串口失败: {e}，2秒后重试...")
                    time.sleep(2)
                    continue

            try:
                # 读取一行数据（以换行符分隔）
                line_bytes = ser.readline()
                if not line_bytes:
                    # 串口超时无数据，继续等待
                    continue

                raw_str = line_bytes.decode("utf-8", errors="ignore").strip()
                if not raw_str:
                    continue

                logger.info(f"收到串口原始数据: {raw_str}")

                # 数据解析
                parsed_data = parse_serial_line(raw_str)
                device_id = parsed_data.pop("device_id")

                # 构建上传的主题与负载
                topic = TELEMETRY_TOPIC_FMT.format(device_id=device_id)
                payload_str = json.dumps(parsed_data, ensure_ascii=False)

                # MQTT 上报 (选用 QoS 1 确保数据送达)
                result = mqtt_client.publish(topic, payload=payload_str, qos=1)
                if result.rc == mqtt.MQTT_ERR_SUCCESS:
                    logger.info(f"成功推送到主题 [{topic}]: {payload_str}")
                else:
                    logger.warning(f"MQTT 推送失败，错误代码: {result.rc}")

            except (UnicodeDecodeError, ValueError) as parse_err:
                logger.warning(f"数据格式错误，跳过该帧: {parse_err}")
            except SerialException as ser_err:
                logger.error(f"串口通信发生故障: {ser_err}，准备重建连接")
                if ser:
                    ser.close()
                ser = None
                time.sleep(2)

    except KeyboardInterrupt:
        logger.info("用户手动终止程序")
    finally:
        # 退出资源释放
        if ser and ser.is_open:
            ser.close()
            logger.info("串口已关闭")
        mqtt_client.loop_stop()
        mqtt_client.disconnect()
        logger.info("MQTT 连接已断开，程序退出完成")


if __name__ == "__main__":
    main()

五、现场测试与故障排查决策

在实际部署与联调时，常因物理接线松动、操作系统串口权限限制或网络安全组策略导致采集受阻。遇到无数据或断连时，建议按照以下排查决策流程逐级定位。

图 4：串口采集与 MQTT 上报常见故障排查决策树

六、生产环境长期稳定运行的建议

当系统从实验室验证走向工矿企业等恶劣环境的长期运行（7×24小时）时，还需要考虑以下工程化细节：

1. 引入环形缓存与粘包处理：如果传感器返回的是连续二进制流，单纯的 readline() 将无法工作。应在内存中维护一个 bytearray 接收缓冲区，每次从串口追加读取到的原始字节，并在缓冲区内检索协议特征头（如 0xAA 0x55）与校验码，完成完整的粘包/断包剥离。
2. 消息本地缓存（离线存储）：若现场网络环境脆弱，建议在网关本地增加轻量级存储（如 SQLite 或本地磁盘缓冲文件）。当检测到 MQTT 断网时，将采集的数据帧持久化落盘，待 MQTT 恢复连接后再批量补发，避免数据丢失。
3. 守护进程化（Systemd / Supervisor）：将 Python 脚本配置为操作系统系统服务（如 Linux 下的 systemd unit），配置 Restart=always 与 RestartSec=5s，确保主机意外重启或不可抗力导致程序异常崩溃时能自动拉起。