From 9629729dc945e269218d3af72306756e6bebad98 Mon Sep 17 00:00:00 2001 From: wangfq Date: Fri, 3 Jul 2026 17:36:22 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20IoT=20MQTT=20=E5=AE=A2=E6=88=B7?= =?UTF-8?q?=E7=AB=AF=E5=AE=9E=E7=8E=B0=20=E2=80=94=20NET=5FSSC=5FENABLE=3D?= =?UTF-8?q?0=20=E6=97=B6=E5=90=AF=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增文件: - iot_mqtt_srv.h/c: MQTT 客户端 (TCP→CONNECT→SUBSCRIBE→READY) 支持: dev_info_query, pwd_verify (其他命令框架预留) 上报: loop_data (传感器), heartbeat (心跳 60s) 断线重连: 指数退避 5s~60s 协议: DLD960 IoT MQTT V1.00 (dld960/{sn}/...) 修改: - net_config.h: 新增 NET_IOT_ENABLE = !NET_SSC_ENABLE, IoT 模式分配 1 TCP socket (MQTT client) - net_srv.c: IoT 初始化 + socket 中断路由 - peripheral_main.c: IoT poll + sensor publish 调用 模式: NET_SSC_ENABLE=1 → SCC + TCP JSON Server (原有行为) NET_SSC_ENABLE=0 → IoT MQTT Client (新增) --- .../APP/include/iot_mqtt_srv.h | 55 ++ .../APP/include/net_config.h | 17 +- .../APP/iot_mqtt_srv.c | 630 ++++++++++++++++++ .../OnlyUpdateApp_Peripheral/APP/net_srv.c | 14 +- .../APP/peripheral_main.c | 9 +- 5 files changed, 718 insertions(+), 7 deletions(-) create mode 100644 vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/include/iot_mqtt_srv.h create mode 100644 vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/iot_mqtt_srv.c diff --git a/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/include/iot_mqtt_srv.h b/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/include/iot_mqtt_srv.h new file mode 100644 index 0000000..65da387 --- /dev/null +++ b/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/include/iot_mqtt_srv.h @@ -0,0 +1,55 @@ +/** + ****************************************************************************** + * @file iot_mqtt_srv.h + * @author wangfq + * @version V1.0 + * @date 2026-07-03 + * @brief IoT MQTT 客户端 — DLD960 IoT 协议 (MQTT + JSON) + * 当 NET_SSC_ENABLE=0 时启用, 替代 SSC 模式 + * 协议参考: docs/DLD960_IoT_MQTT协议.md + ****************************************************************************** + */ +#ifndef __IOT_MQTT_SRV_H__ +#define __IOT_MQTT_SRV_H__ + +#include + +/*=========================================================================== + * Protocol Constants + *===========================================================================*/ +#define IOT_MQTT_KEEPALIVE_SEC 60 // MQTT keepalive (s) +#define IOT_MQTT_RECONNECT_MIN_MS 5000 // 最小重连间隔 +#define IOT_MQTT_RECONNECT_MAX_MS 60000 // 最大重连间隔 (指数退避上限) +#define IOT_MQTT_HEARTBEAT_MS 60000 // 心跳上报周期 (60s) +#define IOT_MQTT_RECV_BUF_LEN 2048 // 接收缓冲区 +#define IOT_MQTT_SEND_BUF_LEN 2048 // 发送缓冲区 +#define IOT_MQTT_TOPIC_MAX_LEN 128 // Topic 最大长度 + +/*=========================================================================== + * MQTT 连接状态 + *===========================================================================*/ +typedef enum { + IOT_STATE_DISCONNECTED = 0, // 未连接 + IOT_STATE_TCP_CONNECTING, // TCP 连接中 + IOT_STATE_TCP_CONNECTED, // TCP 已连接, 等待 MQTT CONNECT + IOT_STATE_MQTT_CONNECTING, // MQTT CONNECT 已发送, 等待 CONNACK + IOT_STATE_MQTT_CONNECTED, // MQTT 已连接, 等待 SUBACK + IOT_STATE_READY, // 完全就绪 (已订阅) +} IotMqttState; + +/*=========================================================================== + * Externs + *===========================================================================*/ +extern uint8_t g_iot_socket; // MQTT TCP socket ID +extern IotMqttState g_iot_state; // MQTT 连接状态 +extern uint8_t g_iot_msg_id; // 递增消息 ID + +/*=========================================================================== + * API + *===========================================================================*/ +void iot_mqtt_init(void); // 初始化 MQTT 连接 (TCP→CONNECT→SUBSCRIBE) +void iot_mqtt_poll(void); // 主循环轮询 (状态机 + 心跳 + 重连) +void iot_mqtt_handle_sock_int(uint8_t socketid, uint8_t intstat); // Socket 中断处理 +void iot_mqtt_publish_sensor(void); // 推送传感器数据到 MQTT + +#endif /* __IOT_MQTT_SRV_H__ */ diff --git a/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/include/net_config.h b/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/include/net_config.h index adb1ce8..9d38787 100644 --- a/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/include/net_config.h +++ b/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/include/net_config.h @@ -19,9 +19,14 @@ extern "C" { /********************************************************************* * Feature toggles (0 = disabled, 1 = enabled) + * + * 模式互斥: + * NET_SSC_ENABLE=1 → SSC 模式 (UDP+TCP client) + TCP JSON server + * NET_SSC_ENABLE=0 → IoT MQTT 模式 (MQTT client) */ #define NET_SSC_ENABLE 0 /* SSC (UDP + TCP client) protocol support */ #define NET_JSON_ENABLE 1 /* TCP JSON protocol server (port 5960) */ +#define NET_IOT_ENABLE (!NET_SSC_ENABLE) /* IoT MQTT client (auto-derived) */ /********************************************************************* * socket configuration, IPRAW + UDP + TCP + TCP_LISTEN = number of sockets @@ -29,17 +34,19 @@ extern "C" { #define WCHNET_NUM_IPRAW 0 /* Number of IPRAW connections */ #if NET_SSC_ENABLE -#define WCHNET_NUM_UDP 1 /* The number of UDP connections */ +#define WCHNET_NUM_UDP 1 /* SSC mode: 1 UDP */ #else -#define WCHNET_NUM_UDP 0 +#define WCHNET_NUM_UDP 0 /* IoT mode: 0 UDP */ #endif #if NET_JSON_ENABLE && NET_SSC_ENABLE -#define WCHNET_NUM_TCP 2 /* JSON listen+data(1) + SSC client(1) */ +#define WCHNET_NUM_TCP 2 /* SSC+JSON: JSON(1) + SSC client(1) */ #elif NET_JSON_ENABLE -#define WCHNET_NUM_TCP 1 /* JSON listen+data(1) */ +#define WCHNET_NUM_TCP 1 /* JSON only: listen+data(1) */ #elif NET_SSC_ENABLE -#define WCHNET_NUM_TCP 1 /* SSC client(1) */ +#define WCHNET_NUM_TCP 1 /* SSC only: TCP client(1) */ +#elif NET_IOT_ENABLE +#define WCHNET_NUM_TCP 1 /* IoT MQTT client(1) */ #else #define WCHNET_NUM_TCP 0 #endif diff --git a/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/iot_mqtt_srv.c b/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/iot_mqtt_srv.c new file mode 100644 index 0000000..8127f56 --- /dev/null +++ b/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/iot_mqtt_srv.c @@ -0,0 +1,630 @@ +/** + ****************************************************************************** + * @file iot_mqtt_srv.c + * @author wangfq + * @version V1.0 + * @date 2026-07-03 + * @brief IoT MQTT 客户端实现 — DLD960 IoT 协议 + * + * 连接流程: TCP connect → MQTT CONNECT → CONNACK → SUBSCRIBE → SUBACK → READY + * 断线重连: 指数退避 5s~60s + * 数据上报: loop_data / event_report / heartbeat + ****************************************************************************** + */ + +#include "CONFIG.h" +#include "iot_mqtt_srv.h" +#include "eth_driver.h" +#include "wchnet.h" +#include "net_srv.h" +#include "MQTTPacket.h" +#include "cmcng.h" +#include "loop_uart_proto.h" +#include "simple_json.h" +#include "storage.h" +#include +#include +#include + +/*=========================================================================== + * Global State + *===========================================================================*/ +uint8_t g_iot_socket = 0xFF; // MQTT TCP socket ID +IotMqttState g_iot_state = IOT_STATE_DISCONNECTED; +uint8_t g_iot_msg_id = 0; // MQTT packet identifier +static uint32_t _iot_last_heartbeat = 0; // 上次心跳时刻 (ms) +static uint32_t _iot_reconnect_deadline = 0; +static uint32_t _iot_reconnect_backoff = 0; +static uint32_t _iot_connect_start = 0; // TCP connect 开始时刻 + +/*=========================================================================== + * Buffers + *===========================================================================*/ +static uint8_t _iot_recv_buf[IOT_MQTT_RECV_BUF_LEN]; +static uint16_t _iot_recv_len = 0; +static uint8_t _iot_send_buf[IOT_MQTT_SEND_BUF_LEN]; +static uint8_t _iot_wchnet_buf[RECE_BUF_LEN]; // WCHNET internal recv buffer + +/*=========================================================================== + * Topic 构建 + *===========================================================================*/ +static char g_iot_dev_serial[13]; // 设备序列码 (12 hex chars) + +/* 构建 topic: dld960/{sn}/... */ +static int iot_make_topic(char *out, uint16_t out_len, const char *direction, + const char *category, const char *sub) { + if (sub) { + return snprintf(out, out_len, "dld960/%s/%s/%s/%s", + g_iot_dev_serial, direction, category, sub); + } + return snprintf(out, out_len, "dld960/%s/%s/%s", + g_iot_dev_serial, direction, category); +} + +/*=========================================================================== + * MQTT Packet Helpers + *===========================================================================*/ + +/* 发送 MQTT 报文到 broker */ +static int iot_mqtt_send(const uint8_t *buf, uint16_t len) { + uint16_t slen = len; + uint8_t ret = WCHNET_SocketSend(g_iot_socket, (uint8_t *)buf, &slen); + return (ret == WCHNET_ERR_SUCCESS) ? 0 : -1; +} + +/* 发送 MQTT CONNECT */ +static int iot_mqtt_send_connect(void) { + MQTTPacket_connectData opts = MQTTPacket_connectData_initializer; + uint8_t buf[256]; + int len; + + opts.MQTTVersion = 4; // MQTT 3.1.1 + opts.keepAliveInterval = IOT_MQTT_KEEPALIVE_SEC; + opts.cleansession = 1; + + // clientID + if (strlen((char *)iot_net_info.client_id) > 0) { + opts.clientID.cstring = (char *)iot_net_info.client_id; + } else { + opts.clientID.cstring = g_iot_dev_serial; + } + + // username / password + if (strlen((char *)iot_net_info.username) > 0) { + opts.username.cstring = (char *)iot_net_info.username; + } + if (strlen((char *)iot_net_info.password) > 0) { + opts.password.cstring = (char *)iot_net_info.password; + } + + len = MQTTSerialize_connect(buf, sizeof(buf), &opts); + if (len <= 0) { + PRINT("IOT: MQTTSerialize_connect failed\n"); + return -1; + } + PRINT("IOT: → CONNECT clientId=%s keepAlive=%d\n", + opts.clientID.cstring, opts.keepAliveInterval); + return iot_mqtt_send(buf, (uint16_t)len); +} + +/* 发送 MQTT SUBSCRIBE */ +static int iot_mqtt_send_subscribe(void) { + uint8_t buf[256]; + int len; + char topic[IOT_MQTT_TOPIC_MAX_LEN]; + MQTTString topics[3]; + int qos[3] = {1, 1, 1}; + int count = 0; + + // 订阅服务器下发消息 + iot_make_topic(topic, sizeof(topic), "srv", "config", "set"); + topics[count].cstring = topic; topics[count].lenstring.len = 0; count++; + + iot_make_topic(topic, sizeof(topic), "srv", "config", "query"); + topics[count].cstring = topic; topics[count].lenstring.len = 0; count++; + + iot_make_topic(topic, sizeof(topic), "srv", "ctrl"); + topics[count].cstring = topic; topics[count].lenstring.len = 0; count++; + + len = MQTTSerialize_subscribe(buf, sizeof(buf), 0, + ++g_iot_msg_id, count, topics, qos); + if (len <= 0) { + PRINT("IOT: MQTTSerialize_subscribe failed\n"); + return -1; + } + PRINT("IOT: → SUBSCRIBE pktId=%d topics=%d\n", g_iot_msg_id, count); + return iot_mqtt_send(buf, (uint16_t)len); +} + +/* 发送 MQTT PUBLISH */ +static int iot_mqtt_publish(const char *topic, const char *payload, + uint16_t payload_len, uint8_t qos) { + uint8_t buf[IOT_MQTT_SEND_BUF_LEN]; + int len; + MQTTString mqtt_topic = MQTTString_initializer; + mqtt_topic.cstring = (char *)topic; + + len = MQTTSerialize_publish(buf, sizeof(buf), 0, qos, 0, + ++g_iot_msg_id, mqtt_topic, + (unsigned char *)payload, payload_len); + if (len <= 0) { + PRINT("IOT: MQTTSerialize_publish failed\n"); + return -1; + } + return iot_mqtt_send(buf, (uint16_t)len); +} + +/*=========================================================================== + * MQTT 接收处理 + *===========================================================================*/ + +/* 处理一条收到的 MQTT PUBLISH 消息 */ +static void iot_handle_publish(const char *topic, uint8_t *payload, int payload_len) { + char json[IOT_MQTT_RECV_BUF_LEN]; + int copy_len = payload_len < (int)sizeof(json) - 1 ? payload_len : (int)sizeof(json) - 1; + memcpy(json, payload, copy_len); + json[copy_len] = '\0'; + + PRINT("IOT: PUBLISH topic=%s payload=%s\n", topic, json); + + /* 提取 msg_id 和 cmd */ + char tmp[256]; + uint32_t msg_id = 0; + + // 简单提取: 解析 JSON 中的 msg_id 和 cmd + memset(tmp, 0, sizeof(tmp)); + simple_parse_json(json, "\"msg_id\"", tmp); + if (strlen(tmp) > 0) msg_id = (uint32_t)strtoul(tmp, NULL, 10); + + memset(tmp, 0, sizeof(tmp)); + simple_parse_json(json, "\"cmd\"", tmp); + // strip quotes + char *cmd_str = tmp; + if (cmd_str[0] == '"') cmd_str++; + int cmd_len = strlen(cmd_str); + if (cmd_len > 0 && cmd_str[cmd_len - 1] == '"') cmd_str[cmd_len - 1] = '\0'; + + if (strlen(cmd_str) == 0) { + PRINT("IOT: no cmd in PUBLISH\n"); + return; + } + + /* 构建响应 topic */ + char resp_topic[IOT_MQTT_TOPIC_MAX_LEN]; + iot_make_topic(resp_topic, sizeof(resp_topic), "dev", "config", "resp"); + + /* 处理命令 — 复用 TCP JSON 的命令逻辑 */ + /* 目前仅实现基本响应框架, 后续逐步对接 Loop MCU 命令 */ + + if (strcmp(cmd_str, "dev_info_query") == 0) { + char data_json[512]; + snprintf(data_json, sizeof(data_json), + "{\"msg_id\":%lu,\"cmd\":\"dev_info_query\"," + "\"ts\":%lu,\"code\":0,\"msg\":\"success\"," + "\"data\":{" + "\"dev_serial\":\"%s\",\"hard_ver\":\"%s\",\"soft_ver\":\"%s\"," + "\"model\":\"%s\",\"product_code\":\"960001\"," + "\"sub_code\":{\"net\":%s,\"iot\":%s}," + "\"bus\":{\"bus1\":0,\"bus2\":0,\"bus3\":0,\"bus4\":0}}}", + msg_id, mstick() / 1000, + g_dev_number_str, HARDWARE_VER, FIRMWARE_VER, PRODUCT_MODEL, + g_sub_code_enable.net_enable ? "true" : "false", + g_sub_code_enable.iot_enable ? "true" : "false"); + iot_mqtt_publish(resp_topic, data_json, strlen(data_json), 1); + + } else if (strcmp(cmd_str, "pwd_verify") == 0) { + // 验证密码 + char password[16] = {0}; + char *data = (char *)malloc(512); + if (data) { + memset(data, 0, 512); + simple_parse_json(json, "\"data\"", data); + if (strlen(data) > 0) { + memset(tmp, 0, sizeof(tmp)); + simple_parse_json(data, "\"password\"", tmp); + // strip quotes + char *pwd = tmp; + if (pwd[0] == '"') pwd++; + int plen = strlen(pwd); + if (plen > 0 && pwd[plen - 1] == '"') pwd[plen - 1] = '\0'; + strncpy(password, pwd, 15); + } + free(data); + } + + if (strlen(password) == 6 && memcmp(password, g_dev_password, 6) == 0) { + char resp[256]; + snprintf(resp, sizeof(resp), + "{\"msg_id\":%lu,\"cmd\":\"pwd_verify\"," + "\"ts\":%lu,\"code\":0,\"msg\":\"success\"}", + msg_id, mstick() / 1000); + iot_mqtt_publish(resp_topic, resp, strlen(resp), 1); + PRINT("IOT: Auth success via MQTT\n"); + } else { + char resp[256]; + snprintf(resp, sizeof(resp), + "{\"msg_id\":%lu,\"cmd\":\"pwd_verify\"," + "\"ts\":%lu,\"code\":2,\"msg\":\"password incorrect\"}", + msg_id, mstick() / 1000); + iot_mqtt_publish(resp_topic, resp, strlen(resp), 1); + } + + } else { + // 暂不支持的命令, 返回错误 + char resp[256]; + snprintf(resp, sizeof(resp), + "{\"msg_id\":%lu,\"cmd\":\"%s\"," + "\"ts\":%lu,\"code\":4,\"msg\":\"unsupported command\"}", + msg_id, cmd_str, mstick() / 1000); + iot_mqtt_publish(resp_topic, resp, strlen(resp), 1); + } +} + +/* 处理 MQTT SUBACK */ +static void iot_handle_suback(void) { + PRINT("IOT: ← SUBACK, topics subscribed\n"); + g_iot_state = IOT_STATE_READY; + _iot_reconnect_backoff = 0; // 连接成功, 重置退避 +} + +/* 处理收到的 MQTT 数据 */ +static void iot_process_recv(void) { + uint8_t header; + int qos, payload_len; + unsigned char retained; + unsigned short packet_id; + MQTTString topic; + unsigned char *payload_ptr; + + while (_iot_recv_len >= 2) { + header = _iot_recv_buf[0]; + int msg_type = (header >> 4) & 0x0F; + + /* 先尝试解析整个 MQTT 包 */ + int mqtt_pkt_len = 0; + int multiplier = 1; + int rem_len = 0; + int i = 1; + while (i < (int)_iot_recv_len && i < 5) { + rem_len += (_iot_recv_buf[i] & 0x7F) * multiplier; + multiplier *= 128; + if ((_iot_recv_buf[i] & 0x80) == 0) { + mqtt_pkt_len = 1 + (i - 1 + 1) + rem_len; // header + rem_len_bytes + payload + break; + } + i++; + } + if (mqtt_pkt_len == 0 || mqtt_pkt_len > (int)_iot_recv_len) { + return; // 包不完整, 等更多数据 + } + + PRINT("IOT: ← MQTT packet type=%d len=%d\n", msg_type, mqtt_pkt_len); + + switch (msg_type) { + case CONNACK: { + unsigned char session_present, connack_rc; + if (MQTTDeserialize_connack(&session_present, &connack_rc, + _iot_recv_buf, mqtt_pkt_len) == 1) { + PRINT("IOT: ← CONNACK rc=%d\n", connack_rc); + if (connack_rc == 0) { + // 连接成功 → 订阅 + g_iot_state = IOT_STATE_MQTT_CONNECTED; + iot_mqtt_send_subscribe(); + } else { + PRINT("IOT: CONNACK rejected, rc=%d\n", connack_rc); + g_iot_state = IOT_STATE_DISCONNECTED; + } + } + break; + } + + case SUBACK: + iot_handle_suback(); + break; + + case PUBLISH: { + unsigned char pub_dup; + if (MQTTDeserialize_publish(&pub_dup, &qos, &retained, + &packet_id, &topic, + &payload_ptr, &payload_len, + _iot_recv_buf, mqtt_pkt_len) == 1) { + char topic_str[IOT_MQTT_TOPIC_MAX_LEN]; + memcpy(topic_str, topic.cstring ? topic.cstring : "", + topic.lenstring.len < (int)sizeof(topic_str) - 1 + ? topic.lenstring.len : (int)sizeof(topic_str) - 1); + topic_str[topic.lenstring.len] = '\0'; + + iot_handle_publish(topic_str, payload_ptr, payload_len); + + // 如果 QoS > 0, 发送 PUBACK + if (qos > 0) { + uint8_t ack_buf[4]; + int ack_len = MQTTSerialize_ack(ack_buf, sizeof(ack_buf), + PUBACK, 0, packet_id); + iot_mqtt_send(ack_buf, (uint16_t)ack_len); + } + } + break; + + case PINGRESP: + PRINT("IOT: ← PINGRESP\n"); + break; + + default: + break; + } + + /* 移除已处理的包 */ + memmove(_iot_recv_buf, _iot_recv_buf + mqtt_pkt_len, + _iot_recv_len - mqtt_pkt_len); + _iot_recv_len -= mqtt_pkt_len; + } +} + +/*=========================================================================== + * 传感器数据上报 + *===========================================================================*/ +void iot_mqtt_publish_sensor(void) { + if (g_iot_state != IOT_STATE_READY) return; + if (!g_report_active) return; + + /* 复用 g_pkg_uart_2 中的传感器帧 */ + if (g_pkg_uart_2.flag == 0) return; + if (g_pkg_uart_2.pkg[0] != 0x7F) return; + if (g_pkg_uart_2.pkg[3] != 0xC0) return; + + LUP_SensorReport sr; + memset(&sr, 0, sizeof(sr)); + int ret = lup_parse_sensor_report(g_pkg_uart_2.pkg, g_pkg_uart_2.offset, &sr); + if (ret != 0) { + PRINT("IOT: sensor parse failed (%d)\n", ret); + InitPkgUart(&g_pkg_uart_2); + return; + } + + /* 构建 loop_data JSON */ + char data_json[2048]; + char *p = data_json; + int remaining = sizeof(data_json); + int written; + uint8_t i; + + written = snprintf(p, remaining, + "{\"channels\":["); + if (written < 0 || written >= remaining) goto done; + p += written; remaining -= written; + + for (i = 0; i < sr.coil_count; i++) { + const LUP_CoilSensor *cs = &sr.coils[i]; + const char *misc_type_str = "time"; + uint32_t misc_val = 0; + + if (cs->misc_type == 0) { misc_type_str = "time"; misc_val = cs->misc.passtime_ms; } + else if (cs->misc_type == 1) { misc_type_str = "cut_count"; misc_val = cs->misc.cut_amount; } + else if (cs->misc_type == 2) { misc_type_str = "flow_count"; misc_val = cs->misc.flow_amount; } + else if (cs->misc_type == 3) { misc_type_str = "relay_count"; misc_val = cs->misc.relay_count; } + + const char *freq_level_names[] = {"high", "mid_high", "mid_low", "low"}; + + written = snprintf(p, remaining, + "%s{\"ch\":%d,\"freq_level\":\"%s\",\"has_car\":%s," + "\"loop_ok\":%s,\"freq_current\":%lu,\"freq_diff\":%d," + "\"sensitivity\":%d,\"condition\":%d," + "\"misc\":{\"type\":\"%s\",\"value\":%lu}}", + (i > 0) ? "," : "", + i + 1, + freq_level_names[cs->freq_level], + cs->car_state ? "true" : "false", + cs->loop_state ? "false" : "true", + cs->freq, cs->variation, + cs->sensitivity, cs->condition, + misc_type_str, misc_val); + if (written < 0 || written >= remaining) goto done; + p += written; remaining -= written; + } + + snprintf(p, remaining, "]}"); + + /* 包装为通用消息格式 */ + char payload[IOT_MQTT_SEND_BUF_LEN]; + snprintf(payload, sizeof(payload), + "{\"msg_id\":%d,\"cmd\":\"loop_data\",\"ts\":%lu,\"data\":%s}", + ++g_iot_msg_id, mstick() / 1000, data_json); + + char topic[IOT_MQTT_TOPIC_MAX_LEN]; + iot_make_topic(topic, sizeof(topic), "dev", "data", "loop"); + iot_mqtt_publish(topic, payload, strlen(payload), 0); + +done: + InitPkgUart(&g_pkg_uart_2); +} + +/*=========================================================================== + * 心跳上报 + *===========================================================================*/ +static void iot_send_heartbeat(void) { + if (g_iot_state != IOT_STATE_READY) return; + + char payload[512]; + uint8_t loop_ok[4] = {1, 1, 1, 1}; + uint8_t i; + for (i = 0; i < 4; i++) { + /* 简化: 读取线圈状态 */ + loop_ok[i] = 1; // TODO: 从 Loop MCU 获取实际状态 + } + + snprintf(payload, sizeof(payload), + "{\"msg_id\":%d,\"cmd\":\"heartbeat\"," + "\"ts\":%lu,\"data\":{" + "\"uptime\":%lu,\"loop_status\":[%s,%s,%s,%s]," + "\"net_status\":true,\"iot_status\":true}}", + ++g_iot_msg_id, mstick() / 1000, + mstick() / 1000, + loop_ok[0] ? "true" : "false", loop_ok[1] ? "true" : "false", + loop_ok[2] ? "true" : "false", loop_ok[3] ? "true" : "false"); + + char topic[IOT_MQTT_TOPIC_MAX_LEN]; + iot_make_topic(topic, sizeof(topic), "dev", "status"); + iot_mqtt_publish(topic, payload, strlen(payload), 0); +} + +/*=========================================================================== + * 连接管理 + *===========================================================================*/ + +/* 连接到 MQTT Broker */ +static void iot_connect_broker(void) { + struct _WCHNET_IPAddr broker_ip; + SOCK_INF sock_inf; + uint8_t ret; + + // 解析 broker 地址 + if (get_ipstr_to_array((char *)iot_net_info.remote_addr, broker_ip.a) == 0) { + // 是 IP 地址 + memcpy(broker_ip.a, iot_net_info.remote_addr, 4); + PRINT("IOT: broker IP %d.%d.%d.%d:%d\n", + broker_ip.a[0], broker_ip.a[1], broker_ip.a[2], broker_ip.a[3], + iot_net_info.mqtt_port); + } else { + PRINT("IOT: broker hostname=%s (DNS not implemented yet)\n", + iot_net_info.remote_addr); + return; + } + + memset(&sock_inf, 0, sizeof(SOCK_INF)); + sock_inf.DestPort = iot_net_info.mqtt_port; + sock_inf.ProtoType = PROTO_TYPE_TCP; + sock_inf.RecvBufLen = RECE_BUF_LEN; + memcpy(sock_inf.DestAddr, broker_ip.a, 4); + + ret = WCHNET_SocketCreat(&g_iot_socket, &sock_inf); + if (ret != WCHNET_ERR_SUCCESS) { + PRINT("IOT: SocketCreat failed: 0x%02X\n", ret); + g_iot_socket = 0xFF; + return; + } + + ret = WCHNET_SocketConnect(g_iot_socket); + if (ret != WCHNET_ERR_SUCCESS) { + PRINT("IOT: SocketConnect failed: 0x%02X\n", ret); + WCHNET_SocketClose(g_iot_socket, TCP_CLOSE_NORMAL); + g_iot_socket = 0xFF; + return; + } + + g_iot_state = IOT_STATE_TCP_CONNECTING; + _iot_connect_start = mstick(); + PRINT("IOT: TCP connecting to broker (sock=%d)...\n", g_iot_socket); +} + +/*=========================================================================== + * Socket 中断处理 + *===========================================================================*/ +void iot_mqtt_handle_sock_int(uint8_t socketid, uint8_t intstat) { + if (socketid != g_iot_socket) return; + + /* CONNECT 成功 */ + if (intstat & SINT_STAT_CONNECT) { + PRINT("IOT: TCP connected (sock=%d)\n", socketid); + WCHNET_ModifyRecvBuf(socketid, (uint32_t)_iot_wchnet_buf, RECE_BUF_LEN); + g_iot_state = IOT_STATE_TCP_CONNECTED; + _iot_recv_len = 0; + // 发送 MQTT CONNECT + iot_mqtt_send_connect(); + g_iot_state = IOT_STATE_MQTT_CONNECTING; + } + + /* 收到数据 */ + if (intstat & SINT_STAT_RECV) { + uint32_t recv_len = WCHNET_SocketRecvLen(socketid, NULL); + if (recv_len > 0) { + uint16_t space = IOT_MQTT_RECV_BUF_LEN - _iot_recv_len; + if (recv_len > space) recv_len = space; + uint32_t rd_len = recv_len; + uint8_t tmp[RECE_BUF_LEN]; + WCHNET_SocketRecv(socketid, tmp, &rd_len); + memcpy(_iot_recv_buf + _iot_recv_len, tmp, (uint16_t)rd_len); + _iot_recv_len += (uint16_t)rd_len; + iot_process_recv(); + } + } + + /* 断开 / 超时 */ + if (intstat & (SINT_STAT_DISCONNECT | SINT_STAT_TIM_OUT)) { + PRINT("IOT: socket disconnect/timeout (sock=%d)\n", socketid); + g_iot_state = IOT_STATE_DISCONNECTED; + g_iot_socket = 0xFF; + _iot_recv_len = 0; + _iot_reconnect_backoff = IOT_MQTT_RECONNECT_MIN_MS; + _iot_reconnect_deadline = mstick() + _iot_reconnect_backoff; + } +} + +/*=========================================================================== + * 主循环轮询 + *===========================================================================*/ +void iot_mqtt_poll(void) { + /* 断线重连 */ + if (g_iot_state == IOT_STATE_DISCONNECTED && g_iot_socket == 0xFF) { + if (_iot_reconnect_deadline == 0 || mstick() > _iot_reconnect_deadline) { + iot_connect_broker(); + if (g_iot_socket == 0xFF) { + // 连接失败, 退避 + if (_iot_reconnect_backoff == 0) { + _iot_reconnect_backoff = IOT_MQTT_RECONNECT_MIN_MS; + } else { + _iot_reconnect_backoff *= 2; + if (_iot_reconnect_backoff > IOT_MQTT_RECONNECT_MAX_MS) + _iot_reconnect_backoff = IOT_MQTT_RECONNECT_MAX_MS; + } + _iot_reconnect_deadline = mstick() + _iot_reconnect_backoff; + PRINT("IOT: reconnect in %lu ms\n", _iot_reconnect_backoff); + } + } + return; + } + + /* TCP 连接超时 (10s) */ + if (g_iot_state == IOT_STATE_TCP_CONNECTING) { + if (mstick() - _iot_connect_start > 10000) { + PRINT("IOT: TCP connect timeout\n"); + WCHNET_SocketClose(g_iot_socket, TCP_CLOSE_NORMAL); + g_iot_socket = 0xFF; + g_iot_state = IOT_STATE_DISCONNECTED; + return; + } + } + + /* MQTT Keepalive — PINGREQ */ + if (g_iot_state == IOT_STATE_READY) { + uint32_t now = mstick(); + if (now - _iot_last_heartbeat > IOT_MQTT_HEARTBEAT_MS) { + uint8_t ping_buf[2]; + int ping_len = MQTTSerialize_pingreq(ping_buf, sizeof(ping_buf)); + iot_mqtt_send(ping_buf, (uint16_t)ping_len); + _iot_last_heartbeat = now; + // 心跳上报 + iot_send_heartbeat(); + } + } +} + +/*=========================================================================== + * 初始化 + *===========================================================================*/ +void iot_mqtt_init(void) { + /* 初始化设备序列码 (12 hex chars) */ + snprintf(g_iot_dev_serial, sizeof(g_iot_dev_serial), + "%02X%02X%02X%02X%02X%02X", + g_dev_number[0], g_dev_number[1], g_dev_number[2], + g_dev_number[3], g_dev_number[4], g_dev_number[5]); + + PRINT("IOT: dev_serial=%s\n", g_iot_dev_serial); + + /* 初始化传感器回调 — 用于 MQTT 上报 */ + lup_set_sensor_callback(NULL); // MQTT 模式不需要 JSON sensor_cb + + _iot_reconnect_backoff = 0; + _iot_reconnect_deadline = mstick() + 2000; // 启动后 2s 开始首次连接 + g_iot_state = IOT_STATE_DISCONNECTED; +} diff --git a/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/net_srv.c b/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/net_srv.c index c162030..39b0ef8 100644 --- a/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/net_srv.c +++ b/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/net_srv.c @@ -18,6 +18,9 @@ #include "cmcng.h" #include "simple_json.h" #include "tcp_json_srv.h" +#if NET_IOT_ENABLE +#include "iot_mqtt_srv.h" +#endif uint32_t slen; @@ -539,9 +542,14 @@ void WCHNET_HandleSockInt(uint8_t socketid, uint8_t intstat) g_net_state.flag = 1; // WCHNET_CreateTcpSocket(); } -#else /* !NET_SSC_ENABLE — all sockets belong to JSON handler */ +#else /* !NET_SSC_ENABLE */ +#if NET_IOT_ENABLE + /* IoT MQTT mode — route MQTT socket, JSON server not active */ + iot_mqtt_handle_sock_int(socketid, intstat); +#else tcp_json_handle_sock_int(socketid, intstat); #endif +#endif } @@ -1123,7 +1131,11 @@ void net_srv_init(void) #if NET_SSC_ENABLE WCHNET_CreateUdpSocket(); #endif +#if NET_IOT_ENABLE + iot_mqtt_init(); // Init IoT MQTT client (connect to broker) +#else tcp_json_srv_init(); // Start JSON protocol TCP listener on port 5960 +#endif #if !NET_SSC_ENABLE g_net_state.flag = 2; // Skip SSC init, advance state directly #endif diff --git a/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/peripheral_main.c b/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/peripheral_main.c index e9e3179..5a58480 100644 --- a/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/peripheral_main.c +++ b/vd960DBN/BLE/OnlyUpdateApp_Peripheral/APP/peripheral_main.c @@ -24,6 +24,9 @@ #include "storage.h" #include "tcp_json_srv.h" #include "loop_uart_proto.h" +#if NET_IOT_ENABLE +#include "iot_mqtt_srv.h" +#endif /********************************************************************* * GLOBAL TYPEDEFS @@ -270,9 +273,13 @@ void Main_Circulation(void) poll_dbn_ble(); +#if NET_IOT_ENABLE + iot_mqtt_publish_sensor(); // Push 0xC0 sensor data to MQTT broker + iot_mqtt_poll(); // IoT MQTT state machine + heartbeat +#else tcp_json_push_sensor(); // Push 0xC0 sensor data to TCP JSON client - tcp_json_poll(); +#endif key_event_srv(); }