[SERIAL] Add the capability send info to Serial instead of/added MQTT (#2030)
* [SERIAL] Add the capability send info to Serial instead of/added MQTT Add the capability to send json data to Serial as a communication layer, added/instead of MQTT * Use delay rather than vTaskDelay --------- Co-authored-by: Florian <1technophile@users.noreply.github.com>
This commit is contained in:
FlorianandFlorian authored and GitHub committed
1 parent
3d6d9baeaf
commit
5492e334dc
7 files changed
+143
-89
No files matched your search
+7
-3
@@ -823,7 +823,8 @@ enum PowerMode { DEACTIVATED = -1,
|
||||
|
||||
/*----------------CONFIGURABLE PARAMETERS-----------------*/
|
||||
struct SYSConfig_s {
|
||||
bool XtoMQTT; // if true the gateway will publish the received data on the MQTT broker
|
||||
bool mqtt; // if true the gateway will publish the received data on the MQTT broker
|
||||
bool serial; // if true the gateway will publish the received data on the SERIAL
|
||||
bool offline;
|
||||
bool discovery; // HA discovery convention
|
||||
bool ohdiscovery; // OH discovery specificities
|
||||
@@ -833,8 +834,11 @@ struct SYSConfig_s {
|
||||
enum PowerMode powerMode;
|
||||
};
|
||||
|
||||
#ifndef DEFAULT_XtoMQTT
|
||||
# define DEFAULT_XtoMQTT true
|
||||
#ifndef DEFAULT_MQTT
|
||||
# define DEFAULT_MQTT true
|
||||
#endif
|
||||
#ifndef DEFAULT_SERIAL
|
||||
# define DEFAULT_SERIAL false
|
||||
#endif
|
||||
#ifndef DEFAULT_OFFLINE
|
||||
# define DEFAULT_OFFLINE false
|
||||
|
||||
+67
-49
@@ -37,6 +37,7 @@ SoftwareSerial SERIALSoftSerial(SERIAL_RX_GPIO, SERIAL_TX_GPIO); // RX, TX
|
||||
// use pointer to stream class for serial communication to make code
|
||||
// compatible with both softwareSerial as hardwareSerial.
|
||||
Stream* SERIALStream = NULL;
|
||||
unsigned long msgCount = 0;
|
||||
|
||||
void setupSERIAL() {
|
||||
//Initalize serial port
|
||||
@@ -125,67 +126,80 @@ void SERIALtoMQTT() {
|
||||
}
|
||||
}
|
||||
|
||||
# elif SERIALtoMQTTmode == 1 // Convert recievee JSON data to multiple MQTT topics based (nested) keys
|
||||
# elif SERIALtoMQTTmode == 1 // Convert received JSON data to one or multiple MQTT topics
|
||||
void SERIALtoMQTT() {
|
||||
// Assumes valid JSON data at SERIAL interface. Use (nested) keys to split JSON data in separate
|
||||
// sub-MQTT-topics up to the defined nesting level.
|
||||
if (SERIALStream->available()) {
|
||||
Log.trace(F("SERIALtoMQTT" CR));
|
||||
|
||||
// Allocate the JSON document
|
||||
StaticJsonDocument<JSON_MSG_BUFFER> SERIALBuffer;
|
||||
JsonObject SERIALdata = SERIALBuffer.to<JsonObject>();
|
||||
|
||||
// JSON validation
|
||||
StaticJsonDocument<0> emptyDoc;
|
||||
StaticJsonDocument<0> filter;
|
||||
DeserializationError validationError = deserializeJson(emptyDoc, *SERIALStream, DeserializationOption::Filter(filter));
|
||||
if (validationError == DeserializationError::Ok) {
|
||||
Log.trace(F("SERIAL data is valid JSON" CR));
|
||||
} else {
|
||||
Log.error(F("SERIAL data is not valid JSON" CR));
|
||||
return;
|
||||
}
|
||||
// Read the entire JSON string from the serial stream
|
||||
String jsonString = SERIALStream->readStringUntil('\n');
|
||||
Log.trace(F("SERIAL msg received: %s" CR), jsonString.c_str());
|
||||
|
||||
// Read the JSON document from the "link" serial port
|
||||
DeserializationError err = deserializeJson(SERIALBuffer, *SERIALStream);
|
||||
int startIndex = 0;
|
||||
while (startIndex < jsonString.length()) {
|
||||
int endIndex = jsonString.indexOf(SERIALPost, startIndex) + strlen(SERIALPost);
|
||||
if (endIndex == -1) {
|
||||
endIndex = jsonString.length();
|
||||
}
|
||||
// Extract the individual JSON message
|
||||
String singleJsonString = jsonString.substring(startIndex, endIndex);
|
||||
// Check if the JSON string contains the prefix and postfix
|
||||
if (singleJsonString.startsWith(SERIALPre) && singleJsonString.endsWith(SERIALPost)) {
|
||||
// Remove the prefix and postfix
|
||||
singleJsonString = singleJsonString.substring(strlen(SERIALPre), singleJsonString.length() - strlen(SERIALPost));
|
||||
} else {
|
||||
Log.error(F("Invalid SERIAL message format: %s" CR), singleJsonString.c_str());
|
||||
return;
|
||||
}
|
||||
|
||||
// Check if the JSON object is empty
|
||||
if (SERIALBuffer.isNull() || SERIALBuffer.size() == 0) {
|
||||
Log.error(F("SERIAL data is empty JSON object" CR));
|
||||
return;
|
||||
}
|
||||
// Check if the JSON string is complete (basic check for curly braces)
|
||||
if (singleJsonString.indexOf('{') == -1 || singleJsonString.indexOf('}') == -1) {
|
||||
Log.error(F("Incomplete JSON string: %s" CR), singleJsonString.c_str());
|
||||
return;
|
||||
}
|
||||
|
||||
if (err == DeserializationError::Ok) {
|
||||
// Read the JSON document from the JSON string
|
||||
DeserializationError err = deserializeJson(SERIALBuffer, singleJsonString);
|
||||
|
||||
if (err == DeserializationError::Ok) {
|
||||
// Check if the JSON object is empty
|
||||
if (SERIALBuffer.isNull() || SERIALBuffer.size() == 0) {
|
||||
Log.error(F("SERIAL data is empty JSON object: %s" CR), jsonString.c_str());
|
||||
return;
|
||||
}
|
||||
// JSON received
|
||||
# if jsonPublishing
|
||||
// send as json
|
||||
if (SERIALdata.containsKey("origin")) {
|
||||
if (SERIALdata["origin"] == "BT") {
|
||||
// send as json
|
||||
if (SERIALdata.containsKey("origin")) {
|
||||
if (SERIALdata["origin"] == "BT") {
|
||||
//Decode the BT data
|
||||
# ifdef ZgatewayBT
|
||||
SERIALdata.remove("origin");
|
||||
PublishDeviceData(SERIALdata);
|
||||
SERIALdata.remove("origin");
|
||||
PublishDeviceData(SERIALdata);
|
||||
# endif
|
||||
} else {
|
||||
handleJsonEnqueue(SERIALdata);
|
||||
}
|
||||
} else {
|
||||
SERIALdata["origin"] = subjectSERIALtoMQTT;
|
||||
handleJsonEnqueue(SERIALdata);
|
||||
}
|
||||
} else {
|
||||
SERIALdata["origin"] = subjectSERIALtoMQTT;
|
||||
handleJsonEnqueue(SERIALdata);
|
||||
}
|
||||
|
||||
# endif
|
||||
# if simplePublishing
|
||||
// send as MQTT topics
|
||||
char topic[mqtt_topic_max_size + 1] = subjectSERIALtoMQTT;
|
||||
sendMQTTfromNestedJson(SERIALBuffer.as<JsonVariant>(), topic, 0, SERIALmaxJSONlevel);
|
||||
// send as MQTT topics
|
||||
char topic[mqtt_topic_max_size + 1] = subjectSERIALtoMQTT;
|
||||
sendMQTTfromNestedJson(SERIALBuffer.as<JsonVariant>(), topic, 0, SERIALmaxJSONlevel);
|
||||
# endif
|
||||
} else {
|
||||
// Print error to serial log
|
||||
Log.error(F("Error in SERIALJSONtoMQTT, deserializeJson() returned %s" CR), err.c_str());
|
||||
|
||||
// Flush all bytes in the "link" serial port buffer
|
||||
while (SERIALStream->available() > 0)
|
||||
SERIALStream->read();
|
||||
} else {
|
||||
// Print error to serial log
|
||||
Log.error(F("Error in SERIALJSONtoMQTT, deserializeJson() returned %s" CR), err.c_str());
|
||||
}
|
||||
startIndex = endIndex;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -228,19 +242,23 @@ void sendMQTTfromNestedJson(JsonVariant obj, char* topic, int level, int maxLeve
|
||||
}
|
||||
# endif
|
||||
|
||||
void MQTTtoSERIAL(char* topicOri, JsonObject& SERIALdata) {
|
||||
Log.trace(F("json" CR));
|
||||
if (cmpToMainTopic(topicOri, subjectMQTTtoSERIAL)) {
|
||||
Log.trace(F("MQTTtoSERIAL json" CR));
|
||||
const char* data = SERIALdata["value"];
|
||||
void XtoSERIAL(const char* topicOri, JsonObject& SERIALdata) {
|
||||
if (cmpToMainTopic(topicOri, subjectXtoSERIAL) || SYSConfig.serial) {
|
||||
Log.trace(F(" MQTTtoSERIAL" CR));
|
||||
const char* prefix = SERIALdata["prefix"] | SERIALPre;
|
||||
const char* postfix = SERIALdata["postfix"] | SERIALPost;
|
||||
Log.trace(F("Value set: %s" CR), data);
|
||||
Log.trace(F("Prefix set: %s" CR), prefix);
|
||||
Log.trace(F("Postfix set: %s" CR), postfix);
|
||||
std::string data;
|
||||
SERIALStream->print(prefix);
|
||||
SERIALStream->print(data);
|
||||
if (SYSConfig.serial) {
|
||||
SERIALdata["msgcount"] = msgCount++;
|
||||
serializeJson(SERIALdata, data);
|
||||
} else if (SERIALdata.containsKey("value")) {
|
||||
data = SERIALdata["value"].as<std::string>();
|
||||
}
|
||||
Log.notice(F("[ OMG->SERIAL ] data: %s" CR), data.c_str());
|
||||
SERIALStream->print(data.c_str());
|
||||
SERIALStream->print(postfix);
|
||||
delay(10);
|
||||
}
|
||||
}
|
||||
#endif
|
||||
@@ -29,7 +29,6 @@
|
||||
extern void setupBT();
|
||||
extern bool BTtoMQTT();
|
||||
extern void MQTTtoBT(char* topicOri, JsonObject& RFdata);
|
||||
extern void pubMainCore(JsonObject& data);
|
||||
extern void launchBTDiscovery(bool overrideDiscovery);
|
||||
extern void stopProcessing();
|
||||
extern String stateBTMeasures(bool);
|
||||
|
||||
@@ -28,7 +28,7 @@
|
||||
|
||||
extern void setupSERIAL();
|
||||
extern void SERIALtoMQTT();
|
||||
extern void MQTTtoSERIAL(char* topicOri, JsonObject& SERIALdata);
|
||||
extern void XtoSERIAL(const char* topicOri, JsonObject& SERIALdata);
|
||||
|
||||
/*-------------------SERIAL topics & parameters----------------------*/
|
||||
|
||||
@@ -66,7 +66,7 @@ extern void MQTTtoSERIAL(char* topicOri, JsonObject& SERIALdata);
|
||||
#define SERIALmaxJSONlevel 2 // Max nested level in which JSON keys are converted to seperate sub-topics
|
||||
|
||||
// settings for MQTT to SERIAL
|
||||
#define subjectMQTTtoSERIAL "/commands/MQTTtoSERIAL"
|
||||
#define subjectXtoSERIAL "/commands/XtoSERIAL"
|
||||
#ifndef SERIALPre
|
||||
# define SERIALPre "00" // The prefix for the SERIAL message
|
||||
#endif
|
||||
|
||||
+44
-32
@@ -451,18 +451,31 @@ bool to_bool(String const& s) { // thanks Chris Jester-Young from stackoverflow
|
||||
}
|
||||
|
||||
/*
|
||||
* Publish a message depending on its origin
|
||||
* Dispatch json messages towards the communication layer
|
||||
*
|
||||
*/
|
||||
void pubMainCore(JsonObject& data) {
|
||||
void jsonDispatch(JsonObject& data) {
|
||||
if (data.containsKey("origin")) {
|
||||
pub((char*)data["origin"].as<const char*>(), data);
|
||||
#if message_UTCtimestamp == true
|
||||
data["UTCtime"] = UTCtimestamp();
|
||||
#endif
|
||||
#if message_unixtimestamp == true
|
||||
data["unixtime"] = unixtimestamp();
|
||||
#endif
|
||||
pubWebUI((char*)data["origin"].as<const char*>(), data);
|
||||
if (SYSConfig.mqtt && !SYSConfig.offline) {
|
||||
pub((char*)data["origin"].as<const char*>(), data);
|
||||
#ifdef ZgatewayBT
|
||||
if (data.containsKey("distance")) {
|
||||
String topic = String(mqtt_topic) + BTConfig.presenceTopic + String(gateway_name);
|
||||
Log.trace(F("Pub HA Presence %s" CR), topic.c_str());
|
||||
pub_custom_topic((char*)topic.c_str(), data, false);
|
||||
if (data.containsKey("distance")) {
|
||||
String topic = String(mqtt_topic) + BTConfig.presenceTopic + String(gateway_name);
|
||||
Log.trace(F("Pub HA Presence %s" CR), topic.c_str());
|
||||
pub_custom_topic((char*)topic.c_str(), data, false);
|
||||
}
|
||||
#endif
|
||||
}
|
||||
#ifdef ZgatewaySERIAL
|
||||
if (SYSConfig.serial)
|
||||
XtoSERIAL("", data);
|
||||
#endif
|
||||
} else {
|
||||
Log.error(F("No origin in JSON filtered" CR));
|
||||
@@ -480,7 +493,6 @@ void enqueueJsonObject(const StaticJsonDocument<JSON_MSG_BUFFER>& jsonDoc) {
|
||||
blockedMessages++;
|
||||
return;
|
||||
}
|
||||
Log.trace(F("Enqueue JSON" CR));
|
||||
JsonBundle bundle;
|
||||
bundle.doc = jsonDoc;
|
||||
jsonQueue.push(bundle);
|
||||
@@ -582,7 +594,7 @@ void emptyQueue() {
|
||||
bundle = jsonQueue.front();
|
||||
jsonQueue.pop();
|
||||
JsonObject obj = bundle.doc.as<JsonObject>();
|
||||
pubMainCore(obj);
|
||||
jsonDispatch(obj);
|
||||
queueLengthSum++;
|
||||
}
|
||||
|
||||
@@ -605,14 +617,7 @@ void pub(const char* topicori, const char* payload, bool retainFlag) {
|
||||
* @param data The Json Object that represents the message
|
||||
*/
|
||||
void pub(const char* topicori, JsonObject& data) {
|
||||
String dataAsString = "";
|
||||
bool ret = sensor_Retain;
|
||||
#if message_UTCtimestamp == true
|
||||
data["UTCtime"] = UTCtimestamp();
|
||||
#endif
|
||||
#if message_unixtimestamp == true
|
||||
data["unixtime"] = unixtimestamp();
|
||||
#endif
|
||||
if (data.containsKey("retain") && data["retain"].is<bool>()) {
|
||||
ret = data["retain"];
|
||||
data.remove("retain");
|
||||
@@ -624,8 +629,6 @@ void pub(const char* topicori, JsonObject& data) {
|
||||
Log.error(F("Empty JSON, not published" CR));
|
||||
return;
|
||||
}
|
||||
serializeJson(data, dataAsString);
|
||||
Log.notice(F("Send on %s msg %s" CR), topicori, dataAsString.c_str());
|
||||
String topic = String(mqtt_topic) + String(gateway_name) + String(topicori);
|
||||
#if valueAsATopic
|
||||
# ifdef ZgatewayPilight
|
||||
@@ -643,9 +646,9 @@ void pub(const char* topicori, JsonObject& data) {
|
||||
#endif
|
||||
|
||||
#if jsonPublishing
|
||||
Log.trace(F("jsonPubl - ON" CR));
|
||||
String dataAsString = "";
|
||||
serializeJson(data, dataAsString);
|
||||
pubMQTT(topic.c_str(), dataAsString.c_str(), ret);
|
||||
pubWebUI(topicori, data);
|
||||
#endif
|
||||
|
||||
#if simplePublishing
|
||||
@@ -714,10 +717,10 @@ void pubMQTT(const char* topic, const char* payload) {
|
||||
* @param retainFlag true if retain the retain Flag
|
||||
*/
|
||||
void pubMQTT(const char* topic, const char* payload, bool retainFlag) {
|
||||
if (SYSConfig.XtoMQTT && !SYSConfig.offline) {
|
||||
if (SYSConfig.mqtt && !SYSConfig.offline) {
|
||||
if (mqtt && mqtt->connected()) {
|
||||
SendReceiveIndicatorON();
|
||||
Log.trace(F("[ OMG->MQTT ] topic: %s msg: %s " CR), topic, payload);
|
||||
Log.notice(F("[ OMG->MQTT ] topic: %s msg: %s " CR), topic, payload);
|
||||
mqtt->publish(topic, payload, 0, retainFlag);
|
||||
} else {
|
||||
Log.warning(F("MQTT not connected, aborting the publication" CR));
|
||||
@@ -854,7 +857,8 @@ void delayWithOTA(long waitMillis) {
|
||||
}
|
||||
|
||||
void SYSConfig_init() {
|
||||
SYSConfig.XtoMQTT = DEFAULT_XtoMQTT;
|
||||
SYSConfig.mqtt = DEFAULT_MQTT;
|
||||
SYSConfig.serial = DEFAULT_SERIAL;
|
||||
SYSConfig.offline = DEFAULT_OFFLINE;
|
||||
#ifdef ZmqttDiscovery
|
||||
SYSConfig.discovery = DEFAULT_DISCOVERY;
|
||||
@@ -867,7 +871,8 @@ void SYSConfig_init() {
|
||||
}
|
||||
|
||||
void SYSConfig_fromJson(JsonObject& SYSdata) {
|
||||
Config_update(SYSdata, "xtomqtt", SYSConfig.XtoMQTT);
|
||||
Config_update(SYSdata, "mqtt", SYSConfig.mqtt);
|
||||
Config_update(SYSdata, "serial", SYSConfig.serial);
|
||||
Config_update(SYSdata, "offline", SYSConfig.offline);
|
||||
#ifdef ZmqttDiscovery
|
||||
Config_update(SYSdata, "disc", SYSConfig.discovery);
|
||||
@@ -883,7 +888,8 @@ void SYSConfig_fromJson(JsonObject& SYSdata) {
|
||||
void SYSConfig_save() {
|
||||
StaticJsonDocument<JSON_MSG_BUFFER> jsonBuffer;
|
||||
JsonObject SYSdata = jsonBuffer.to<JsonObject>();
|
||||
SYSdata["xtomqtt"] = SYSConfig.XtoMQTT;
|
||||
SYSdata["mqtt"] = SYSConfig.mqtt;
|
||||
SYSdata["serial"] = SYSConfig.serial;
|
||||
SYSdata["offline"] = SYSConfig.offline;
|
||||
SYSdata["powermode"] = SYSConfig.powerMode;
|
||||
# ifdef ZmqttDiscovery
|
||||
@@ -1044,11 +1050,11 @@ void setupMQTT() {
|
||||
}
|
||||
reconnection_count++; // Increment reconnection count
|
||||
Log.trace(F("MQTT connection count: %d" CR), reconnection_count);
|
||||
if (reconnection_count > reconnection_threshold && SYSConfig.XtoMQTT) {
|
||||
if (reconnection_count > reconnection_threshold && SYSConfig.mqtt) {
|
||||
// Detected instability in MQTT connection
|
||||
Log.warning(F("MQTT connection instability detected: %d reconnections in the last %d seconds" CR), reconnection_count, reconnection_window_millis / 1000);
|
||||
// Stop xtoMQTT to see if it helps and still enables to receive data
|
||||
SYSConfig.XtoMQTT = false;
|
||||
SYSConfig.mqtt = false;
|
||||
}
|
||||
}
|
||||
# endif
|
||||
@@ -2483,7 +2489,6 @@ void loop() {
|
||||
}
|
||||
timer_sys_checks = millis();
|
||||
}
|
||||
emptyQueue();
|
||||
}
|
||||
} else if (!SYSConfig.offline) { // disconnected from network
|
||||
#ifdef ESP32
|
||||
@@ -2622,6 +2627,9 @@ void loop() {
|
||||
launchRTL_433Discovery(false);
|
||||
# endif
|
||||
#endif
|
||||
// Empty the queue
|
||||
emptyQueue();
|
||||
// Sleep if ready
|
||||
if (ready_to_sleep) {
|
||||
sleep();
|
||||
}
|
||||
@@ -2939,7 +2947,7 @@ void receivingMQTT(char* topicOri, char* datacallback) {
|
||||
MQTTtoSomfy(topicOri, jsondata);
|
||||
# endif
|
||||
# ifdef ZgatewaySERIAL
|
||||
MQTTtoSERIAL(topicOri, jsondata);
|
||||
XtoSERIAL(topicOri, jsondata);
|
||||
# endif
|
||||
# ifdef MQTT_HTTPS_FW_UPDATE
|
||||
MQTTHttpsFWUpdate(topicOri, jsondata);
|
||||
@@ -3448,9 +3456,13 @@ void MQTTtoSYS(char* topicOri, JsonObject& SYSdata) { // json object decoding
|
||||
# endif
|
||||
#endif
|
||||
|
||||
if (SYSdata.containsKey("xtomqtt") && SYSdata["xtomqtt"].is<bool>()) {
|
||||
SYSConfig.XtoMQTT = SYSdata["xtomqtt"];
|
||||
Log.notice(F("xtomqtt: %T" CR), SYSConfig.XtoMQTT);
|
||||
if (SYSdata.containsKey("mqtt") && SYSdata["mqtt"].is<bool>()) {
|
||||
SYSConfig.mqtt = SYSdata["mqtt"];
|
||||
Log.notice(F("xtomqtt: %T" CR), SYSConfig.mqtt);
|
||||
}
|
||||
if (SYSdata.containsKey("serial") && SYSdata["serial"].is<bool>()) {
|
||||
SYSConfig.serial = SYSdata["serial"];
|
||||
Log.notice(F("SERIAL: %T" CR), SYSConfig.serial);
|
||||
}
|
||||
if (SYSdata.containsKey("offline") && SYSdata["offline"].is<bool>()) {
|
||||
SYSConfig.offline = SYSdata["offline"];
|
||||
|
||||
Reference in new issue
Block a user