From 17cbec048e2586c55dfcd05d099f1c0e1d924546 Mon Sep 17 00:00:00 2001 From: lingyunxsh Date: Sun, 25 May 2025 15:28:16 +0800 Subject: [PATCH] =?UTF-8?q?=E6=95=B4=E7=90=86=E6=96=87=E6=A1=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- mqtt_esp32_client/mqtt_esp32_client.ino | 266 ++++++++++++++++++++++++ 1 file changed, 266 insertions(+) create mode 100644 mqtt_esp32_client/mqtt_esp32_client.ino diff --git a/mqtt_esp32_client/mqtt_esp32_client.ino b/mqtt_esp32_client/mqtt_esp32_client.ino new file mode 100644 index 0000000..b7e1ff7 --- /dev/null +++ b/mqtt_esp32_client/mqtt_esp32_client.ino @@ -0,0 +1,266 @@ +#include +#include +#include // 用于JSON操作 + +// ----------- 设备配置 (需要为您自己的设备和环境修改) ----------- +// WiFi +const char *ssid = "xxxxx"; // 请输入您的 Wi-Fi 名称 +const char *password = "xxxxx"; // 请输入您的 Wi-Fi 密码 + +// MQTT Broker +const char *mqtt_broker = "broker.emqx.io"; // 您的 MQTT Broker 地址 +const char *mqtt_username = "emqx"; // 您的 MQTT 用户名 (如果需要) +const char *mqtt_password = "public"; // 您的 MQTT 密码 (如果需要) +const int mqtt_port = 1883; // 您的 MQTT 端口 + +// 设备唯一标识符 (非常重要, 必须与后端注册的一致) +const char *spotUid = "ESP32_SPOT_001"; // 例如: "SPOT001", "P005-A1" 等 + +// ----------- MQTT 主题定义 ----------- +// 上行 (ESP32 -> 后端) +String topic_status_update; // 状态上报: charging/spot/{spotUid}/status +String topic_command_ack; // 指令回执: charging/spot/{spotUid}/command/ack +String topic_heartbeat; // 心跳: charging/spot/{spotUid}/heartbeat + +// 下行 (后端 -> ESP32, ESP32需要订阅这些) +String topic_command_start; // 启动充电: charging/spot/{spotUid}/command/start +String topic_command_stop; // 停止充电: charging/spot/{spotUid}/command/stop +// String topic_command_query_status; // 查询状态 (可选): charging/spot/{spotUid}/command/query_status + +// ----------- 全局变量 ----------- +WiFiClient espClient; +PubSubClient client(espClient); + +// 模拟硬件状态 (实际项目中需要从传感器或硬件逻辑获取) +const char* currentDeviceStatus = "IDLE"; // 设备当前状态: IDLE, CHARGING, FAULTED 等 +float currentVoltage = 220.0; +float currentCurrent = 0.0; +float currentPower = 0.0; +float currentEnergyConsumed = 0.0; +int currentErrorCode = 0; +String currentSessionId = ""; // 当前充电会话ID + +// 定时发送相关 +unsigned long lastStatusUpdateTime = 0; +unsigned long lastHeartbeatTime = 0; +const long statusUpdateInterval = 30000; // 状态上报间隔 (例如: 30秒) +const long heartbeatInterval = 60000; // 心跳间隔 (例如: 60秒) + +void setup_mqtt_topics() { + String baseTopic = "charging/spot/" + String(spotUid) + "/"; + topic_status_update = baseTopic + "status"; + topic_command_ack = baseTopic + "command/ack"; + topic_heartbeat = baseTopic + "heartbeat"; + + topic_command_start = baseTopic + "command/start"; + topic_command_stop = baseTopic + "command/stop"; + // topic_command_query_status = baseTopic + "command/query_status"; + Serial.println("MQTT 主题初始化完成:"); + Serial.println(" 状态上报: " + topic_status_update); + Serial.println(" 指令回执: " + topic_command_ack); + Serial.println(" 心跳: " + topic_heartbeat); + Serial.println(" 订阅启动指令: " + topic_command_start); + Serial.println(" 订阅停止指令: " + topic_command_stop); +} + +void connect_wifi() { + Serial.println("正在连接 Wi-Fi..."); + WiFi.begin(ssid, password); + while (WiFi.status() != WL_CONNECTED) { + delay(500); + Serial.print("."); + } + Serial.println("\nWi-Fi 连接成功!"); + Serial.print("IP 地址: "); + Serial.println(WiFi.localIP()); +} + +void reconnect_mqtt() { + while (!client.connected()) { + Serial.print("尝试连接 MQTT Broker: "); + Serial.println(mqtt_broker); + String client_id = "esp32-client-" + String(spotUid) + "-"; + client_id += String(WiFi.macAddress()); + Serial.printf("客户端 ID: %s \n", client_id.c_str()); + + if (client.connect(client_id.c_str(), mqtt_username, mqtt_password)) { + Serial.println("MQTT Broker 连接成功!"); + // 重新订阅主题 + client.subscribe(topic_command_start.c_str()); + Serial.println("已订阅: " + topic_command_start); + client.subscribe(topic_command_stop.c_str()); + Serial.println("已订阅: " + topic_command_stop); + // if (topic_command_query_status.length() > 0) client.subscribe(topic_command_query_status.c_str()); + + // (可选) 连接成功后立即发送一次状态 + publish_status_update(); + } else { + Serial.print("连接失败, rc="); + Serial.print(client.state()); + Serial.println(" 2秒后重试..."); + delay(2000); + } + } +} + +void callback(char *topic, byte *payload, unsigned int length) { + Serial.println("-----------------------"); + Serial.print("消息抵达, 主题: "); + Serial.println(topic); + + char message[length + 1]; + memcpy(message, payload, length); + message[length] = \'\\0\'; // 添加字符串结束符 + Serial.print("消息内容: "); + Serial.println(message); + + StaticJsonDocument<256> doc; // 适当调整JSON文档大小 + DeserializationError error = deserializeJson(doc, message); + + if (error) { + Serial.print("JSON 解析失败: "); + Serial.println(error.f_str()); + return; + } + + const char* commandId = doc["commandId"]; // 提取 commandId 用于回执 + + // 根据主题处理指令 + if (String(topic) == topic_command_start) { + Serial.println("接收到 [启动充电] 指令"); + // 实际硬件操作: 启动充电 + // 例如: digitalWrite(RELAY_PIN, HIGH); + currentDeviceStatus = "CHARGING"; + if (doc.containsKey("sessionId")) { + currentSessionId = String(doc["sessionId"].as()); + } + Serial.println("模拟: 充电已启动。会话ID: " + currentSessionId); + publish_command_ack(commandId, true, "Charging started successfully"); + publish_status_update(); // 立即更新状态 + + } else if (String(topic) == topic_command_stop) { + Serial.println("接收到 [停止充电] 指令"); + // 实际硬件操作: 停止充电 + // 例如: digitalWrite(RELAY_PIN, LOW); + currentDeviceStatus = "COMPLETED"; // 或者 "IDLE"，取决于逻辑 + Serial.println("模拟: 充电已停止。"); + publish_command_ack(commandId, true, "Charging stopped successfully"); + currentSessionId = ""; // 清除会话ID + publish_status_update(); // 立即更新状态 + } + // else if (String(topic) == topic_command_query_status) { + // Serial.println("接收到 [查询状态] 指令"); + // publish_status_update(); // 回复当前状态 + // publish_command_ack(commandId, true, "Status reported"); + // } + else { + Serial.println("未知指令主题"); + } + Serial.println("-----------------------"); +} + +void publish_message(const String& topic, const JsonDocument& doc, const char* message_type) { + String jsonBuffer; + serializeJson(doc, jsonBuffer); + Serial.print("发送 "); + Serial.print(message_type); + Serial.print(" 到主题 ["); + Serial.print(topic); + Serial.print("]: "); + Serial.println(jsonBuffer); + + if (client.publish(topic.c_str(), jsonBuffer.c_str())) { + Serial.println(String(message_type) + " 发送成功"); + } else { + Serial.println(String(message_type) + " 发送失败"); + } +} + +void publish_status_update() { + StaticJsonDocument<512> doc; // 适当调整JSON文档大小 + doc["spotUid"] = spotUid; + doc["timestamp"] = String(WiFi.getTime()); // 需要NTP同步时间才能获取正确UTC时间，此处仅为示例 + doc["status"] = currentDeviceStatus; + doc["voltage"] = currentVoltage; + doc["current"] = currentCurrent; + doc["power"] = currentPower; + doc["energyConsumed"] = currentEnergyConsumed; + doc["errorCode"] = currentErrorCode; + if (currentSessionId.length() > 0) { + doc["sessionId"] = currentSessionId; + } else { + doc["sessionId"] = nullptr; // 或者不包含此字段 + } + + publish_message(topic_status_update, doc, "状态更新"); + lastStatusUpdateTime = millis(); +} + +void publish_heartbeat() { + StaticJsonDocument<256> doc; // 适当调整JSON文档大小 + doc["spotUid"] = spotUid; + doc["timestamp"] = String(WiFi.getTime()); // 同上，NTP时间问题 + doc["status"] = currentDeviceStatus; // 心跳中也带上当前状态 + + publish_message(topic_heartbeat, doc, "心跳"); + lastHeartbeatTime = millis(); +} + +void publish_command_ack(const char* commandId, bool success, const char* message) { + if (!commandId || strlen(commandId) == 0) { + Serial.println("无法发送ACK: commandId 为空"); + return; + } + StaticJsonDocument<256> doc; // 适当调整JSON文档大小 + doc["commandId"] = commandId; + doc["spotUid"] = spotUid; + doc["status"] = success ? "SUCCESS" : "FAILURE"; + doc["message"] = message; + + publish_message(topic_command_ack, doc, "指令回执"); +} + +void setup() { + Serial.begin(115200); + Serial.println("\nESP32 充电桩模拟客户端启动..."); + + setup_mqtt_topics(); // 初始化MQTT主题 + connect_wifi(); // 连接Wi-Fi + + client.setServer(mqtt_broker, mqtt_port); + client.setCallback(callback); // 设置消息回调函数 +} + +void loop() { + if (!client.connected()) { + reconnect_mqtt(); // 如果MQTT未连接，则重连 + } + client.loop(); // 维持MQTT连接和处理消息 + + unsigned long currentTime = millis(); + + // 定时发送状态更新 + if (currentTime - lastStatusUpdateTime > statusUpdateInterval) { + // 在实际项目中，这里应该先更新 currentVoltage, currentCurrent 等状态值 + // 例如: currentVoltage = readVoltageSensor(); + publish_status_update(); + } + + // 定时发送心跳 + if (currentTime - lastHeartbeatTime > heartbeatInterval) { + publish_heartbeat(); + } + + // 模拟充电过程中的电量和功率变化 (仅为演示) + if (String(currentDeviceStatus) == "CHARGING") { + currentEnergyConsumed += 0.01; // 假设每秒消耗0.01kWh (不精确，仅为演示) + currentCurrent = 5.0; // 假设充电电流5A + currentPower = (currentVoltage * currentCurrent) / 1000.0; // kW + // 注意：实际项目中这些值应来自传感器或充电控制器 + // 这里为了演示，每隔一段时间简单更新一下 + } else { + currentCurrent = 0.0; + currentPower = 0.0; + } + delay(100); // 短暂延时，避免loop过于频繁，给其他任务一点时间 (可选) +}