cacto
a645737927
Реализованы sensor-1 (DHT22, телеметрия temperature/humidity) и
реле-актуаторы light-1/pump-1/fan-1 (turn_on/turn_off/set_level) по
протоколу MQTT платформы home_automatization (devices/{id}/telemetry|commands|ack).
100 lines
2.5 KiB
C++
100 lines
2.5 KiB
C++
#include "mqtt_link.h"
|
|
|
|
#include <ArduinoJson.h>
|
|
#include <PubSubClient.h>
|
|
#include <WiFi.h>
|
|
|
|
#include "config.h"
|
|
#include "devices.h"
|
|
#include "time_sync.h"
|
|
|
|
static WiFiClient wifiClient;
|
|
static PubSubClient mqttClient(wifiClient);
|
|
|
|
// devices/{device_id}/commands -> device_id, или "" если топик не подходит.
|
|
static String deviceIdFromCommandTopic(const String& topic) {
|
|
if (!topic.startsWith("devices/") || !topic.endsWith("/commands")) {
|
|
return "";
|
|
}
|
|
return topic.substring(8, topic.length() - 9); // strip "devices/" and "/commands"
|
|
}
|
|
|
|
static void publishAck(const String& deviceId, JsonDocument& state) {
|
|
JsonDocument ackDoc;
|
|
ackDoc["state"] = state.as<JsonObject>();
|
|
|
|
char buf[256];
|
|
size_t len = serializeJson(ackDoc, buf, sizeof(buf));
|
|
|
|
String topic = "devices/" + deviceId + "/ack";
|
|
mqttClient.publish(topic.c_str(), (const uint8_t*)buf, len, false);
|
|
}
|
|
|
|
static void onMessage(char* topic, byte* payload, unsigned int length) {
|
|
String deviceId = deviceIdFromCommandTopic(String(topic));
|
|
if (deviceId.isEmpty()) {
|
|
return;
|
|
}
|
|
|
|
JsonDocument cmd;
|
|
DeserializationError err = deserializeJson(cmd, payload, length);
|
|
if (err) {
|
|
Serial.print("[mqtt] invalid command payload on ");
|
|
Serial.println(topic);
|
|
return;
|
|
}
|
|
|
|
String action = cmd["action"] | "";
|
|
double level = cmd["level"] | 0.0;
|
|
if (action.isEmpty()) {
|
|
Serial.print("[mqtt] command missing action on ");
|
|
Serial.println(topic);
|
|
return;
|
|
}
|
|
|
|
JsonDocument state;
|
|
if (devicesApplyCommand(deviceId, action, level, state)) {
|
|
publishAck(deviceId, state);
|
|
}
|
|
}
|
|
|
|
void mqttBegin() {
|
|
mqttClient.setServer(MQTT_BROKER_HOST, MQTT_BROKER_PORT);
|
|
mqttClient.setCallback(onMessage);
|
|
}
|
|
|
|
void mqttEnsureConnected() {
|
|
if (mqttClient.connected()) {
|
|
return;
|
|
}
|
|
|
|
Serial.print("[mqtt] connecting to ");
|
|
Serial.print(MQTT_BROKER_HOST);
|
|
if (mqttClient.connect(MQTT_CLIENT_ID)) {
|
|
Serial.println(" connected");
|
|
mqttClient.subscribe("devices/+/commands", 1);
|
|
} else {
|
|
Serial.print(" failed, rc=");
|
|
Serial.println(mqttClient.state());
|
|
}
|
|
}
|
|
|
|
void mqttLoop() {
|
|
mqttClient.loop();
|
|
}
|
|
|
|
void mqttPublishTelemetry(const String& deviceId, const String& sensorType, double value) {
|
|
JsonDocument doc;
|
|
doc["device_id"] = deviceId;
|
|
doc["zone_id"] = ZONE_ID;
|
|
doc["sensor_type"] = sensorType;
|
|
doc["value"] = value;
|
|
doc["timestamp"] = nowRFC3339();
|
|
|
|
char buf[256];
|
|
size_t len = serializeJson(doc, buf, sizeof(buf));
|
|
|
|
String topic = "devices/" + deviceId + "/telemetry";
|
|
mqttClient.publish(topic.c_str(), (const uint8_t*)buf, len, false);
|
|
}
|