MCU心跳机制

MCU设备心跳检测，用于检测MCU与服务器的通信连接状态，确保设备在线可响应。

说明：本文档中 {uuid} 等同于设备的 {imei}（4G通讯模块IMEI号）

指令码：0x30

协议参考：二进制协议格式（0xA8头码 + 补码校验）。校验码计算详见 校验码计算说明

心跳规则

1. MCU每60秒，发布一次心跳数据，自发自收一条相同的数据
2. 发布和订阅使用相同的主题
3. 如果MCU连续3次（即3分钟）未收到返回数据，MCU判断连接异常并尝试重新连接MQTT服务器
4. 心跳数据同时由EMQX通过Webhook转发至服务端，服务端解析信号强度(CSQ)和系统状态(ST)，用于维护设备在线状态

设备上报 ▲

发布主题：/mcu/{uuid}/user/heart

说明：心跳包通过 /user/heart 主题发布，设备订阅同一主题接收回传以验证MQTT链路连通性。同时EMQX通过规则将心跳消息转发至服务端Webhook，由服务端解析CSQ和ST用于设备状态管理。

设备需订阅以下主题以接收心跳回传：

• /mcu/{uuid}/user/heart — 接收自己的心跳回传

数据格式：字节码数据，Hex，16进制解析

扩展内容格式

格式：CSQ:xx;ST:xx
示例：CSQ:27;ST:00

系统状态（ST）

信号值说明（CSQ）

信号强度值范围 0 ~ 31，值越大表示信号强度越好

Hex数据示例

A8 00 20 30
38 36 30 36 30 32 30 36 39 31 36 35 33 35 32
43 53 51 3A 32 37 3B 53 54 3A 30 30
F7

数据包解析示例：

A8 ------------------------------------------------ Byte[0] 头码
00 20 --------------------------------------------- Byte[1~2] 包长度：32
30 ------------------------------------------------ Byte[3] 指令码：0x30 MCU心跳
38 36 30 36 30 32 30 36 39 31 36 35 33 35 32 ------- Byte[4~18] 设备IMEI："860602069165352"

-- Byte[19~n] 扩展内容，ASCII显示：
43 53 51 3A 32 37 3B 53 54 3A 30 30
 C  S  Q  :  2  7  ;  S  T  :  0  0

解析结果：
CSQ:27   → 信号强度27（较好）
ST:00    → 系统状态：待机

F7 ------------------------------------------------ Byte[n+1] 校验码

ASCII字符对照表

设备回复 △

发布主题：/mcu/{uuid}/user/heart

数据格式：字节码数据，Hex，16进制解析（与设备上报内容完全一致）

说明：MQTT Broker会将设备发布的消息转发给所有订阅者（包括设备自己），设备收到自己发布的消息即证明MQTT链路正常。此回传过程由MQTT Broker直接完成。

服务端处理说明

心跳包经EMQX规则转发至服务端Webhook，处理流程：

1. EMQX收到 /mcu/{uuid}/user/heart 的 message.publish 事件
2. EMQX通过规则转发至服务端 /emqx/webhook/http_event
3. 服务端识别 /mcu/ 前缀，委托MCU Webhook服务处理
4. 解析二进制协议：提取IMEI、CSQ信号强度、ST系统状态
5. 更新设备在线状态和信号质量

服务端可获取的信息：

异常告警建议

校验码计算

采用补码校验。详见 校验码计算说明

快速说明：

• 校验码 = 所有字节之和（不含校验码自身）的补码
• 验证：所有字节（含校验码）之和的低8位为0x00

MicroPython 实现示例

import time
from umqtt.simple import MQTTClient

# 配置参数
IMEI = "860602069165352"
MQTT_HOST = "emqxbms.gs.dogwof.com"
MQTT_PORT = 1885
MQTT_USER = "860602069165352"
MQTT_PASS = "your_password"
HEART_INTERVAL = 60  # 心跳间隔（秒）

# Topic
TOPIC_HEART = "/mcu/{}/user/heart".format(IMEI)
TOPIC_GET = "/mcu/{}/user/get".format(IMEI)


def calculate_checksum(data: bytes) -> int:
    """计算补码校验码"""
    total = sum(data)
    low_byte = total & 0xFF
    return 0 if low_byte == 0 else (0x100 - low_byte)


def get_csq() -> int:
    """
    获取信号强度（需根据实际硬件实现）

    Returns:
        int: 信号强度 0-31 或 99
    """
    return 27


def get_system_status() -> int:
    """
    获取系统状态（需根据实际硬件实现）

    Returns:
        int: 系统状态 0=待机 1=充电中 2=故障 3=升级中
    """
    return 0


def build_heartbeat_packet(imei: str, csq: int, status: int) -> bytes:
    """
    构建MCU心跳数据包（cmd=0x30）

    Args:
        imei: 设备IMEI号（15位ASCII字符串）
        csq: 信号强度（0-31或99）
        status: 系统状态（0~3）

    Returns:
        bytes: 完整数据包（含校验码）
    """
    content = "CSQ:{};ST:{:02d}".format(csq, status)

    packet = bytearray()
    packet.append(0xA8)  # 头码
    packet.append(0x00)  # 长度高字节（占位）
    packet.append(0x00)  # 长度低字节（占位）
    packet.append(0x30)  # 指令码：MCU心跳
    packet.extend(imei.encode('ascii'))    # IMEI（15字节）
    packet.extend(content.encode('ascii')) # 扩展内容

    # 回填长度
    total_length = len(packet) + 1  # +1 校验码
    packet[1] = (total_length >> 8) & 0xFF
    packet[2] = total_length & 0xFF

    # 校验码
    checksum = calculate_checksum(packet)
    packet.append(checksum)

    return bytes(packet)


def main():
    """主循环"""
    client = MQTTClient(
        client_id=IMEI,
        server=MQTT_HOST,
        port=MQTT_PORT,
        user=MQTT_USER,
        password=MQTT_PASS,
        keepalive=120
    )
    client.connect()
    client.subscribe(TOPIC_GET)

    while True:
        try:
            packet = build_heartbeat_packet(IMEI, get_csq(), get_system_status())
            client.publish(TOPIC_HEART, packet, qos=1)
            print("心跳已发送: {}".format(packet.hex().upper()))

            time.sleep(HEART_INTERVAL)
            client.check_msg()

        except Exception as e:
            print("心跳发送失败: {}".format(e))
            try:
                client.connect()
            except:
                time.sleep(5)


if __name__ == "__main__":
    main()

C语言实现示例（ESP32/STM32）

#include <stdint.h>
#include <string.h>
#include <stdio.h>

#define PROTOCOL_HEAD       0xA8
#define CMD_MCU_HEARTBEAT   0x30

uint8_t calculate_checksum(const uint8_t *data, uint16_t len) {
    uint32_t sum = 0;
    for (uint16_t i = 0; i < len; i++) {
        sum += data[i];
    }
    uint8_t low_byte = sum & 0xFF;
    return (low_byte == 0) ? 0 : (0x100 - low_byte);
}

int mcu_build_heartbeat(uint8_t *buf, const char *imei, int csq, int status) {
    int idx = 0;

    buf[idx++] = PROTOCOL_HEAD;
    int len_idx = idx;
    buf[idx++] = 0x00;
    buf[idx++] = 0x00;
    buf[idx++] = CMD_MCU_HEARTBEAT;

    /* IMEI（15字节ASCII） */
    memcpy(&buf[idx], imei, 15);
    idx += 15;

    /* 扩展内容：CSQ:xx;ST:xx */
    idx += sprintf((char *)&buf[idx], "CSQ:%d;ST:%02d", csq, status);

    /* 回填长度 */
    uint16_t total_len = idx + 1;
    buf[len_idx]     = (total_len >> 8) & 0xFF;
    buf[len_idx + 1] = total_len & 0xFF;

    /* 校验码 */
    buf[idx] = calculate_checksum(buf, idx);
    return idx + 1;
}

/* 使用示例 */
void example_send_heartbeat(void) {
    uint8_t buf[64];
    /* csq=27, status=0(待机) */
    int len = mcu_build_heartbeat(buf, "860602069165352", 27, 0);
    /* mqtt_publish("/mcu/860602069165352/user/heart", buf, len, QOS_1); */
}

测试示例

Python 测试（paho-mqtt）

import paho.mqtt.client as mqtt
import time

IMEI = "860602069165352"
TOPIC_HEART = f"/mcu/{IMEI}/user/heart"


def calculate_checksum(data: bytes) -> int:
    total = sum(data)
    low_byte = total & 0xFF
    return 0 if low_byte == 0 else (0x100 - low_byte)


def build_heartbeat(imei: str, csq: int, status: int) -> bytes:
    content = f"CSQ:{csq};ST:{status:02d}"
    packet = bytearray()
    packet.append(0xA8)
    packet.append(0x00)
    packet.append(0x00)
    packet.append(0x30)  # MCU心跳指令码
    packet.extend(imei.encode('ascii'))
    packet.extend(content.encode('ascii'))

    total_length = len(packet) + 1
    packet[1] = (total_length >> 8) & 0xFF
    packet[2] = total_length & 0xFF

    checksum = calculate_checksum(packet)
    packet.append(checksum)
    return bytes(packet)


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code == 0:
        print("MQTT连接成功")

def on_publish(client, userdata, mid, reason_codes, properties):
    print(f"心跳发布确认, mid={mid}")

client = mqtt.Client(
    client_id=IMEI,
    callback_api_version=mqtt.CallbackAPIVersion.VERSION2
)
client.username_pw_set("860602069165352", "your_password")
client.on_connect = on_connect
client.on_publish = on_publish

client.connect("emqxbms.gs.dogwof.com", 1885, keepalive=120)
client.loop_start()

packet = build_heartbeat(IMEI, 27, 0)
print(f"心跳数据(Hex): {packet.hex().upper()}")

result = client.publish(TOPIC_HEART, packet, qos=1)

time.sleep(3)
client.loop_stop()
client.disconnect()

MQTTX 调试

使用 MQTTX 工具连接后，向以下 Topic 发送 Hex 数据进行测试：

Topic: /mcu/860602069165352/user/heart

Payload（Hex格式）:
A8 00 20 30 38 36 30 36 30 32 30 36 39 31 36 35 33 35 32 43 53 51 3A 32 37 3B 53 54 3A 30 30 F7

解析结果：CSQ:27（信号较好），ST:00（待机），校验码：F7