andry/EMS-ESP32
1
0
Fork 0
mirror of https://github.com/emsesp/EMS-ESP32.git synced 2025-12-06 07:49:52 +03:00
Code Issues Packages Projects Releases Wiki Activity

added mqtt_nestedjson

Browse Source
This commit is contained in:
Paul
2020-02-22 11:37:55 +01:00
parent 97aaff07ef
commit 9eda49b46a
5 changed files with 434 additions and 394 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
CHANGELOG.md
View File

@@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- `boiler wwonetime` command from Telnet
- `set bus_id <ID>` to support multiple EMS-ESP circuits. Default is 0x0B to mimic a service key.
- MQTT publish messages are queued and gracefully published every second to avoid TCP blocks
- Added `mqtt_nestedjson` option to disable multiple data records being nested into a single JSON string
### Fixed
- set boiler warm water temp on Junkers/Bosch HT3
@@ -29,7 +30,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Removed
- `autodetect scan`
- `mqttlog` and showing MQTT log in the web interface - no point showing history of previous mqtt publishes in ESP's precious memory. For debugging recommend using MQTT Explorer or another external tool.
- `mqttlog all` and showing MQTT log in the web interface - no point showing history of previous mqtt publishes in ESP's precious memory. For debugging I recommend using MQTT Explorer or another external tool.
## [1.9.4] 15-12-2019

17
src/MyESP.cpp
View File

@@ -83,6 +83,7 @@ MyESP::MyESP() {
_mqtt_heartbeat = false;
_mqtt_keepalive = MQTT_KEEPALIVE;
_mqtt_qos = MQTT_QOS;
_mqtt_nestedjson = false;
_mqtt_retain = MQTT_RETAIN;
_mqtt_will_topic = strdup(MQTT_WILL_TOPIC);
_mqtt_will_online_payload = strdup(MQTT_WILL_ONLINE_PAYLOAD);
@@ -527,9 +528,8 @@ void MyESP::_mqttPublishQueue() {
#endif
if (packet_id == 0) {
// it failed
// if we retried 3 times, give up. remove from queue
if (element.retry_count == 2) {
// it failed. if we retried n times, give up. remove from queue
if (element.retry_count == (MQTT_PUBLISH_MAX_RETRY - 1)) {
myDebug_P(PSTR("[MQTT] Failed to publish to %s with payload %s"), _mqttTopic(element.topic), element.payload);
_mqtt_publish_fails++; // increment failure counter
_mqttRemoveLastPublish();
@@ -879,6 +879,7 @@ void MyESP::_printSetCommands() {
myDebug_P(PSTR(" set mqtt_qos [0-3]"));
myDebug_P(PSTR(" set mqtt_keepalive [seconds]"));
myDebug_P(PSTR(" set mqtt_retain [on | off]"));
myDebug_P(PSTR(" set mqtt_nestedjson [on | off]"));
myDebug_P(PSTR(" set ntp_enabled <on | off>"));
myDebug_P(PSTR(" set ntp_interval [minutes]"));
myDebug_P(PSTR(" set ntp_timezone [n]"));
@@ -941,6 +942,7 @@ void MyESP::_printSetCommands() {
myDebug_P(PSTR(" mqtt_retain=%s"), (_mqtt_retain) ? "on" : "off");
myDebug_P(PSTR(" mqtt_qos=%d"), _mqtt_qos);
myDebug_P(PSTR(" mqtt_heartbeat=%s"), (_mqtt_heartbeat) ? "on" : "off");
myDebug_P(PSTR(" mqtt_nestedjson=%s"), (_mqtt_nestedjson) ? "on" : "off");
#ifdef FORCE_SERIAL
myDebug_P(PSTR(" serial=%s (this is always when compiled with -DFORCE_SERIAL)"), (_general_serial) ? "on" : "off");
@@ -1045,6 +1047,8 @@ bool MyESP::_changeSetting(uint8_t wc, const char * setting, const char * value)
restart = save_config;
} else if (strcmp(setting, "mqtt_heartbeat") == 0) {
save_config = fs_setSettingValue(&_mqtt_heartbeat, value, false);
} else if (strcmp(setting, "mqtt_nestedjson") == 0) {
save_config = fs_setSettingValue(&_mqtt_nestedjson, value, false);
} else if (strcmp(setting, "ntp_enabled") == 0) {
save_config = fs_setSettingValue(&_ntp_enabled, value, false);
} else if (strcmp(setting, "ntp_interval") == 0) {
@@ -1876,6 +1880,7 @@ bool MyESP::_fs_loadConfig() {
_mqtt_keepalive = mqtt["keepalive"] | MQTT_KEEPALIVE;
_mqtt_retain = mqtt["retain"];
_mqtt_qos = mqtt["qos"] | MQTT_QOS;
_mqtt_nestedjson = mqtt["nestedjson"] | true; // default to on
_mqtt_password = strdup(mqtt["password"] | "");
_mqtt_base = strdup(mqtt["base"] | MQTT_BASE_DEFAULT);
@@ -2091,6 +2096,7 @@ bool MyESP::_fs_writeConfig() {
mqtt["retain"] = _mqtt_retain;
mqtt["password"] = _mqtt_password;
mqtt["base"] = _mqtt_base;
mqtt["nestedjson"] = _mqtt_nestedjson;
JsonObject ntp = doc.createNestedObject("ntp");
ntp["server"] = _ntp_server;
@@ -2192,6 +2198,11 @@ void MyESP::_calculateLoad() {
}
}
// returns true if nested JSON setting is enabled
bool MyESP::mqttUseNestedJson() {
return _mqtt_nestedjson;
}
// returns true is MQTT is alive
bool MyESP::isMQTTConnected() {
return mqttClient.connected();

19
src/MyESP.h
View File

@@ -97,9 +97,11 @@ extern struct rst_info resetInfo;
#define MQTT_WILL_TOPIC "status" // for last will & testament topic name
#define MQTT_MAX_TOPIC_SIZE 50 // max length of MQTT topic
#define MQTT_MAX_PAYLOAD_SIZE 700 // max size of a JSON object. See https://arduinojson.org/v6/assistant/
#define MQTT_MAX_PAYLOAD_SIZE_LARGE 2000 // max size of a large JSON object, like for sending MQTT log
#define MQTT_MAX_PAYLOAD_SIZE_LARGE 2000 // max size of a large JSON object
#define MQTT_MAX_PAYLOAD_SIZE_SMALL 200
#define MQTT_QUEUE_MAX_SIZE 20 // Size of the MQTT queue
#define MQTT_PUBLISH_WAIT 1000 // every 2 seconds check MQTT queue
#define MQTT_PUBLISH_WAIT 1000 // every 1 second check MQTT queue
#define MQTT_PUBLISH_MAX_RETRY 4 // max retries for giving up on publishing
// Internal MQTT events
#define MQTT_CONNECT_EVENT 0
@@ -285,6 +287,7 @@ class MyESP {
void mqttPublish(const char * topic, const char * payload);
void mqttPublish(const char * topic, const char * payload, bool retain);
void setMQTT(mqtt_callback_f callback);
bool mqttUseNestedJson();
// OTA
void setOTA(ota_callback_f OTACallback_pre, ota_callback_f OTACallback_post);
@@ -329,7 +332,6 @@ class MyESP {
uint32_t getSystemLoadAverage();
uint32_t getSystemResetReason();
uint8_t getSystemBootStatus();
bool _have_ntp_time;
unsigned long getSystemTime();
void heartbeatPrint();
void heartbeatCheck(bool force = false);
@@ -365,6 +367,7 @@ class MyESP {
uint32_t _mqtt_last_connection;
bool _mqtt_connecting;
bool _mqtt_heartbeat;
bool _mqtt_nestedjson;
uint16_t _mqtt_publish_fails;
// wifi
@@ -417,10 +420,8 @@ class MyESP {
bool _fs_sendConfig();
size_t _fs_validateConfigFile(const char * filename, size_t maxsize, JsonDocument & doc);
size_t _fs_validateLogFile(const char * filename);
fs_loadsave_callback_f _fs_loadsave_callback_f;
fs_setlist_callback_f _fs_setlist_callback_f;
void _printSetCommands();
// general
@@ -447,26 +448,19 @@ class MyESP {
bool _rtcmem_status;
bool _rtcmemStatus();
bool _getRtcmemStatus();
void _rtcmemInit();
void _rtcmemSetup();
void _deferredReset(unsigned long delay, uint8_t reason);
uint8_t _getSystemStabilityCounter();
void _setSystemStabilityCounter(uint8_t counter);
uint8_t _getSystemDropoutCounter();
void _setSystemDropoutCounter(uint8_t counter);
void _increaseSystemDropoutCounter();
void _setSystemResetReason(uint8_t reason);
uint8_t _getCustomResetReason();
void _setCustomResetReason(uint8_t reason);
uint8_t _getSystemResetReason();
void _setSystemBootStatus(uint8_t status);
bool _systemStable;
void _bootupSequence();
bool _getSystemCheck();
@@ -497,6 +491,7 @@ class MyESP {
uint16_t _ntp_interval;
bool _ntp_enabled;
uint8_t _ntp_timezone;
bool _have_ntp_time;
};
extern MyESP myESP;

217
src/ems-esp.cpp
View File

@@ -582,13 +582,34 @@ void publishSensorValues() {
bool hasdata = false;
char buffer[128] = {0}; // temp string buffer
// if we're not using nested JSON, send each sensor out seperately
if (!myESP.mqttUseNestedJson()) {
for (uint8_t i = 0; i < EMSESP_Settings.dallas_sensors; i++) {
float sensorValue = ds18.getValue(i);
if (sensorValue != DS18_DISCONNECTED) {
hasdata = true;
char topic[30]; // sensors{1-n}
strlcpy(topic, TOPIC_EXTERNAL_SENSORS, sizeof(topic)); // create topic
strlcat(topic, _int_to_char(buffer, i + 1), sizeof(topic));
sensors[PAYLOAD_EXTERNAL_SENSOR_ID] = ds18.getDeviceID(buffer, i); // add ID
sensors[PAYLOAD_EXTERNAL_SENSOR_TEMP] = sensorValue; // add temp value
char data[100] = {0};
serializeJson(doc, data, sizeof(data)); // convert to string
myESP.mqttPublish(topic, data); // and publish
}
}
if (hasdata) {
myDebugLog("Publishing external sensor data via MQTT");
}
return; // exit
}
// see if the sensor values have changed, if so send it on
for (uint8_t i = 0; i < EMSESP_Settings.dallas_sensors; i++) {
float sensorValue = ds18.getValue(i);
if (sensorValue != DS18_DISCONNECTED) {
hasdata = true;
// create a nested object
// https://github.com/proddy/EMS-ESP/issues/327
// create a nested object - https://github.com/proddy/EMS-ESP/issues/327
char sensorID[10]; // sensor{1-n}
strlcpy(sensorID, PAYLOAD_EXTERNAL_SENSOR_NUM, sizeof(sensorID));
strlcat(sensorID, _int_to_char(buffer, i + 1), sizeof(sensorID));
@@ -598,44 +619,20 @@ void publishSensorValues() {
}
}
/* test code - https://github.com/proddy/EMS-ESP/issues/326
float sensorValue = 23.43;
hasdata = true;
char sensorID[10]; // sensor{1-n}
for (uint8_t i = 0; i < 10; i++) {
strlcpy(sensorID, PAYLOAD_EXTERNAL_SENSOR_NUM, sizeof(sensorID));
strlcat(sensorID, _int_to_char(buffer, i + 1), sizeof(sensorID));
JsonObject dataSensor = sensors.createNestedObject(sensorID);
dataSensor[PAYLOAD_EXTERNAL_SENSOR_ID] = "28D45A79A2190310";
dataSensor[PAYLOAD_EXTERNAL_SENSOR_TEMP] = sensorValue;
}
*/
if (!hasdata) {
return; // nothing to send
}
char data[DS18_MQTT_PAYLOAD_MAXSIZE] = {0};
serializeJson(doc, data, sizeof(data));
myDebugLog("Publishing external sensor data via MQTT");
myESP.mqttPublish(TOPIC_EXTERNAL_SENSORS, data);
if (hasdata) {
myDebugLog("Publishing external sensor data via MQTT");
}
}
// send values via MQTT
// a json object is created for each device type
void publishEMSValues(bool force) {
// don't send if MQTT is not connected or EMS bus is not connected
if (!myESP.isMQTTConnected() || (!ems_getBusConnected()) || (EMSESP_Settings.publish_time == -1)) {
return;
}
// publish Boiler data via MQTT
void publishEMSValues_boiler() {
char s[20] = {0}; // for formatting strings
StaticJsonDocument<MQTT_MAX_PAYLOAD_SIZE> doc;
char data[MQTT_MAX_PAYLOAD_SIZE] = {0};
// do we have boiler changes?
if (ems_getBoilerEnabled() && (ems_Device_has_flags(EMS_DEVICE_UPDATE_FLAG_BOILER) || force)) {
JsonObject rootBoiler = doc.to<JsonObject>();
if (EMS_Boiler.wWComfort == EMS_VALUE_UBAParameterWW_wwComfort_Hot) {
@@ -739,20 +736,20 @@ void publishEMSValues(bool force) {
// see if the heating or hot tap water has changed, if so send
// last_boilerActive stores heating in bit 1 and tap water in bit 2
static uint8_t last_boilerActive = 0xFF; // for remembering last setting of the tap water or heating on/off
if ((last_boilerActive != ((EMS_Boiler.tapwaterActive << 1) + EMS_Boiler.heatingActive)) || force) {
if (last_boilerActive != ((EMS_Boiler.tapwaterActive << 1) + EMS_Boiler.heatingActive)) {
myDebugLog("Publishing hot water and heating states via MQTT");
myESP.mqttPublish(TOPIC_BOILER_TAPWATER_ACTIVE, EMS_Boiler.tapwaterActive == 1 ? "1" : "0");
myESP.mqttPublish(TOPIC_BOILER_HEATING_ACTIVE, EMS_Boiler.heatingActive == 1 ? "1" : "0");
last_boilerActive = ((EMS_Boiler.tapwaterActive << 1) + EMS_Boiler.heatingActive); // remember last state
}
}
ems_Device_remove_flags(EMS_DEVICE_UPDATE_FLAG_BOILER); // unset flag
}
// handle the thermostat values
if (ems_getThermostatEnabled() && (ems_Device_has_flags(EMS_DEVICE_UPDATE_FLAG_THERMOSTAT) || force)) {
doc.clear();
// handle the thermostat values
void publishEMSValues_thermostat() {
char s[20] = {0}; // for formatting strings
StaticJsonDocument<MQTT_MAX_PAYLOAD_SIZE> doc;
char data[MQTT_MAX_PAYLOAD_SIZE] = {0};
JsonObject rootThermostat = doc.to<JsonObject>();
for (uint8_t hc_v = 1; hc_v <= EMS_THERMOSTAT_MAXHC; hc_v++) {
@@ -814,16 +811,16 @@ void publishEMSValues(bool force) {
}
}
data[0] = '\0'; // reset data for next package
serializeJson(doc, data, sizeof(data));
myDebugLog("Publishing thermostat data via MQTT");
myESP.mqttPublish(TOPIC_THERMOSTAT_DATA, data);
ems_Device_remove_flags(EMS_DEVICE_UPDATE_FLAG_THERMOSTAT); // unset flag
}
}
// handle the mixing values
if (ems_getMixingModuleEnabled() && (ems_Device_has_flags(EMS_DEVICE_UPDATE_FLAG_MIXING) || force)) {
doc.clear();
// publish mixing data
void publishEMSValues_mixing() {
char s[20] = {0}; // for formatting strings
StaticJsonDocument<MQTT_MAX_PAYLOAD_SIZE> doc;
char data[MQTT_MAX_PAYLOAD_SIZE] = {0};
JsonObject rootMixing = doc.to<JsonObject>();
for (uint8_t hc_v = 1; hc_v <= EMS_MIXING_MAXHC; hc_v++) {
@@ -863,17 +860,16 @@ void publishEMSValues(bool force) {
}
}
data[0] = '\0'; // reset data for next package
serializeJson(doc, data, sizeof(data));
myDebugLog("Publishing mixing data via MQTT");
myESP.mqttPublish(TOPIC_MIXING_DATA, data);
ems_Device_remove_flags(EMS_DEVICE_UPDATE_FLAG_MIXING); // unset flag
}
}
// For SM10 and SM100/SM200 Solar Modules
if (ems_getSolarModuleEnabled() && (ems_Device_has_flags(EMS_DEVICE_UPDATE_FLAG_SOLAR) || force)) {
// build new json object
doc.clear();
// For SM10 and SM100/SM200 Solar Modules
void publishEMSValues_solar() {
char s[20] = {0}; // for formatting strings
StaticJsonDocument<MQTT_MAX_PAYLOAD_SIZE> doc;
char data[MQTT_MAX_PAYLOAD_SIZE] = {0};
JsonObject rootSM = doc.to<JsonObject>();
if (EMS_SolarModule.collectorTemp > EMS_VALUE_SHORT_NOTSET)
@@ -902,36 +898,31 @@ void publishEMSValues(bool force) {
if (EMS_SolarModule.EnergyTotal < EMS_VALUE_USHORT_NOTSET)
rootSM[SM_ENERGYTOTAL] = (float)EMS_SolarModule.EnergyTotal / 10;
data[0] = '\0'; // reset data for next package
serializeJson(doc, data, sizeof(data));
myDebugLog("Publishing SM data via MQTT");
myESP.mqttPublish(TOPIC_SM_DATA, data);
ems_Device_remove_flags(EMS_DEVICE_UPDATE_FLAG_SOLAR); // unset flag
}
}
// handle HeatPump
if (ems_getHeatPumpEnabled() && (ems_Device_has_flags(EMS_DEVICE_UPDATE_FLAG_HEATPUMP) || force)) {
// build new json object
doc.clear();
JsonObject rootSM = doc.to<JsonObject>();
// handle HeatPump
void publishEMSValues_heatpump() {
StaticJsonDocument<MQTT_MAX_PAYLOAD_SIZE> doc;
char data[MQTT_MAX_PAYLOAD_SIZE] = {0};
JsonObject rootHP = doc.to<JsonObject>();
if (EMS_HeatPump.HPModulation != EMS_VALUE_INT_NOTSET)
rootSM[HP_PUMPMODULATION] = EMS_HeatPump.HPModulation;
rootHP[HP_PUMPMODULATION] = EMS_HeatPump.HPModulation;
if (EMS_HeatPump.HPSpeed != EMS_VALUE_INT_NOTSET)
rootSM[HP_PUMPSPEED] = EMS_HeatPump.HPSpeed;
rootHP[HP_PUMPSPEED] = EMS_HeatPump.HPSpeed;
data[0] = '\0'; // reset data for next package
serializeJson(doc, data, sizeof(data));
myDebugLog("Publishing HeatPump data via MQTT");
myESP.mqttPublish(TOPIC_HP_DATA, data);
ems_Device_remove_flags(EMS_DEVICE_UPDATE_FLAG_HEATPUMP); // unset flag
}
}
// Publish shower data
void do_publishShowerData() {
StaticJsonDocument<200> doc;
StaticJsonDocument<MQTT_MAX_PAYLOAD_SIZE_SMALL> doc;
JsonObject rootShower = doc.to<JsonObject>();
rootShower[TOPIC_SHOWER_TIMER] = EMSESP_Settings.shower_timer ? "1" : "0";
rootShower[TOPIC_SHOWER_ALERT] = EMSESP_Settings.shower_alert ? "1" : "0";
@@ -956,17 +947,58 @@ void do_publishShowerData() {
myESP.mqttPublish(TOPIC_SHOWER_DATA, data, false);
}
// call PublishValues with forcing forcing
void do_publishValues() {
// send values via MQTT
// a json object is created for each device type
void publishEMSValues(bool force) {
// don't send if MQTT is not connected or EMS bus is not connected
if (!myESP.isMQTTConnected() || (!ems_getBusConnected()) || (EMSESP_Settings.publish_time == -1)) {
return;
}
if (ems_getBoilerEnabled() && (ems_Device_has_flags(EMS_DEVICE_UPDATE_FLAG_BOILER) || force)) {
publishEMSValues_boiler();
ems_Device_remove_flags(EMS_DEVICE_UPDATE_FLAG_BOILER); // unset flag
}
if (ems_getThermostatEnabled() && (ems_Device_has_flags(EMS_DEVICE_UPDATE_FLAG_THERMOSTAT) || force)) {
publishEMSValues_thermostat();
ems_Device_remove_flags(EMS_DEVICE_UPDATE_FLAG_THERMOSTAT); // unset flag
}
if (ems_getMixingModuleEnabled() && (ems_Device_has_flags(EMS_DEVICE_UPDATE_FLAG_MIXING) || force)) {
publishEMSValues_mixing();
ems_Device_remove_flags(EMS_DEVICE_UPDATE_FLAG_MIXING); // unset flag
}
if (ems_getSolarModuleEnabled() && (ems_Device_has_flags(EMS_DEVICE_UPDATE_FLAG_SOLAR) || force)) {
publishEMSValues_solar();
ems_Device_remove_flags(EMS_DEVICE_UPDATE_FLAG_SOLAR); // unset flag
}
if (ems_getHeatPumpEnabled() && (ems_Device_has_flags(EMS_DEVICE_UPDATE_FLAG_HEATPUMP) || force)) {
publishEMSValues_heatpump();
ems_Device_remove_flags(EMS_DEVICE_UPDATE_FLAG_HEATPUMP); // unset flag
}
}
// publishes value via MQTT
void publishValues(bool force, bool send_sensor) {
if (EMSESP_Settings.publish_time == -1) {
myDebugLog("Publishing is disabled.");
return;
}
myDebugLog("Starting scheduled MQTT publish...");
publishEMSValues(false);
publishEMSValues(force);
if (send_sensor) {
publishSensorValues();
myESP.heartbeatCheck(true);
}
// myESP.heartbeatCheck(true);
}
// calls publishValues fron the Ticker loop, also sending sensor data
// but not using false for force so only data that has changed will be sent
void do_publishValues() {
publishValues(false, true);
}
// callback to light up the LED, called via Ticker every second
@@ -2096,6 +2128,10 @@ void initEMSESP() {
* Shower Logic
*/
void showerCheck() {
if (!EMSESP_Settings.shower_timer) {
return;
}
uint32_t time_now = millis();
// if already in cold mode, ignore all this logic until we're out of the cold blast
if (!EMSESP_Shower.doingColdShot) {
@@ -2231,7 +2267,7 @@ void setup() {
if (EMSESP_Settings.publish_time > 0) {
publishValuesTimer.attach(EMSESP_Settings.publish_time, do_publishValues); // post MQTT EMS values
} else if (EMSESP_Settings.publish_time == 0) {
// automatic mode. use this Ticker to send out sensor values
// automatic mode. use this Ticker to send out sensor values only. the EMS ones are done in the loop.
publishValuesTimer.attach(DEFAULT_SENSOR_PUBLISHTIME, publishSensorValues);
}
@@ -2256,33 +2292,30 @@ void setup() {
void loop() {
myESP.loop(); // handle telnet, mqtt, wifi etc
// to prevent load, only run checks every second
// get Dallas Sensor readings every 2 seconds
static uint32_t last_check = 0;
if (millis() - last_check < 1000) {
return;
}
last_check = millis();
// get Dallas Sensor readings
uint32_t time_now = millis();
if (time_now - last_check > 2000) {
last_check = time_now;
if (EMSESP_Settings.dallas_sensors) {
ds18.loop();
}
}
// if we have an EMS bus connection go and fetch some data and MQTT publish it
// if we just have an EMS bus connection go and fetch the data and MQTT publish it to get started
if (_need_first_publish) {
publishEMSValues(false);
publishSensorValues();
_need_first_publish = false; // reset flag
} else {
// check if we're on auto mode for publishing
// then send EMS values, only if its been flagged to update
if (EMSESP_Settings.publish_time == 0) {
publishEMSValues(false);
}
publishValues(true, true);
_need_first_publish = false;
return;
}
// do shower logic, if enabled
if (EMSESP_Settings.shower_timer) {
showerCheck();
// check if we're on auto mode for publishing
// then send EMS values, only if its been flagged to update.
// Don't send sensor data as this is done by the Ticker
if (EMSESP_Settings.publish_time == 0) {
publishValues(false, false);
}
// do shower logic
showerCheck();
}

2
src/version.h
View File

@@ -1 +1 @@
#define APP_VERSION "1.9.5b38"
#define APP_VERSION "1.9.5b39"