patrick/OpenDTU
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Optimize MQTT subscription handling

Browse Source
This commit is contained in:
Thomas Basler 2024-09-23 21:59:43 +02:00
parent d6a5fef4e7
commit 93b6e5a885
2 changed files with 61 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
include/MqttHandleInverter.h
View File

@ -5,6 +5,8 @@
#include <Hoymiles.h> #include <Hoymiles.h>
#include <TaskSchedulerDeclarations.h> #include <TaskSchedulerDeclarations.h>
#include <espMqttClient.h> #include <espMqttClient.h>
#include <frozen/map.h>
#include <frozen/string.h>
class MqttHandleInverterClass { class MqttHandleInverterClass {
public: public:
@ -19,7 +21,6 @@ public:
private: private:
void loop(); void loop();
void publishField(std::shared_ptr<InverterAbstract> inv, const ChannelType_t type, const ChannelNum_t channel, const FieldId_t fieldId); void publishField(std::shared_ptr<InverterAbstract> inv, const ChannelType_t type, const ChannelNum_t channel, const FieldId_t fieldId);
void onMqttMessage(const espMqttClientTypes::MessageProperties& properties, const char* topic, const uint8_t* payload, const size_t len, const size_t index, const size_t total);
Task _loopTask; Task _loopTask;
@ -41,6 +42,27 @@ private:
FLD_IRR, FLD_IRR,
FLD_Q FLD_Q
}; };
enum class Topic : unsigned {
LimitPersistentRelative,
LimitPersistentAbsolute,
LimitNonPersistentRelative,
LimitNonPersistentAbsolute,
Power,
Restart,
};
static constexpr frozen::string _cmdtopic = "+/cmd/";
static constexpr frozen::map<frozen::string, Topic, 6> _subscriptions = {
{ "limit_persistent_relative", Topic::LimitPersistentRelative },
{ "limit_persistent_absolute", Topic::LimitPersistentAbsolute },
{ "limit_nonpersistent_relative", Topic::LimitNonPersistentRelative },
{ "limit_nonpersistent_absolute", Topic::LimitNonPersistentAbsolute },
{ "power", Topic::Power },
{ "restart", Topic::Restart },
};
void onMqttMessage(Topic t, const espMqttClientTypes::MessageProperties& properties, const char* topic, const uint8_t* payload, const size_t len, const size_t index, const size_t total);
}; };
extern MqttHandleInverterClass MqttHandleInverter; extern MqttHandleInverterClass MqttHandleInverter;

90
src/MqttHandleInverter.cpp
View File

@ -7,13 +7,6 @@
#include "MqttSettings.h" #include "MqttSettings.h"
#include <ctime> #include <ctime>
#define TOPIC_SUB_LIMIT_PERSISTENT_RELATIVE "limit_persistent_relative"
#define TOPIC_SUB_LIMIT_PERSISTENT_ABSOLUTE "limit_persistent_absolute"
#define TOPIC_SUB_LIMIT_NONPERSISTENT_RELATIVE "limit_nonpersistent_relative"
#define TOPIC_SUB_LIMIT_NONPERSISTENT_ABSOLUTE "limit_nonpersistent_absolute"
#define TOPIC_SUB_POWER "power"
#define TOPIC_SUB_RESTART "restart"
#define PUBLISH_MAX_INTERVAL 60000 #define PUBLISH_MAX_INTERVAL 60000
MqttHandleInverterClass MqttHandleInverter; MqttHandleInverterClass MqttHandleInverter;
@ -154,7 +147,7 @@ String MqttHandleInverterClass::getTopic(std::shared_ptr<InverterAbstract> inv,
return inv->serialString() + "/" + chanNum + "/" + chanName; return inv->serialString() + "/" + chanNum + "/" + chanName;
} }
void MqttHandleInverterClass::onMqttMessage(const espMqttClientTypes::MessageProperties& properties, const char* topic, const uint8_t* payload, const size_t len, const size_t index, const size_t total) void MqttHandleInverterClass::onMqttMessage(Topic t, const espMqttClientTypes::MessageProperties& properties, const char* topic, const uint8_t* payload, const size_t len, const size_t index, const size_t total)
{ {
const CONFIG_T& config = Configuration.get(); const CONFIG_T& config = Configuration.get();
@ -162,15 +155,11 @@ void MqttHandleInverterClass::onMqttMessage(const espMqttClientTypes::MessagePro
strncpy(token_topic, topic, MQTT_MAX_TOPIC_STRLEN + 40); // convert const char* to char* strncpy(token_topic, topic, MQTT_MAX_TOPIC_STRLEN + 40); // convert const char* to char*
char* serial_str; char* serial_str;
char* subtopic;
char* setting;
char* rest = &token_topic[strlen(config.Mqtt.Topic)]; char* rest = &token_topic[strlen(config.Mqtt.Topic)];
serial_str = strtok_r(rest, "/", &rest); serial_str = strtok_r(rest, "/", &rest);
subtopic = strtok_r(rest, "/", &rest);
setting = strtok_r(rest, "/", &rest);
if (serial_str == NULL || subtopic == NULL || setting == NULL) { if (serial_str == NULL) {
return; return;
} }
@ -183,33 +172,30 @@ void MqttHandleInverterClass::onMqttMessage(const espMqttClientTypes::MessagePro
return; return;
} }
// check if subtopic is unequal cmd std::string strValue(reinterpret_cast<const char*>(payload), len);
if (strcmp(subtopic, "cmd")) { float payload_val = -1;
try {
payload_val = std::stof(strValue);
} catch (std::invalid_argument const& e) {
MessageOutput.printf("MQTT handler: cannot parse payload of topic '%s' as float: %s\r\n",
topic, strValue.c_str());
return; return;
} }
char* strlimit = new char[len + 1]; switch (t) {
memcpy(strlimit, payload, len); case Topic::LimitPersistentRelative:
strlimit[len] = '\0';
const float payload_val = strtof(strlimit, NULL);
delete[] strlimit;
if (payload_val < 0) {
MessageOutput.printf("MQTT payload < 0 received --> ignoring\r\n");
return;
}
if (!strcmp(setting, TOPIC_SUB_LIMIT_PERSISTENT_RELATIVE)) {
// Set inverter limit relative persistent // Set inverter limit relative persistent
MessageOutput.printf("Limit Persistent: %.1f %%\r\n", payload_val); MessageOutput.printf("Limit Persistent: %.1f %%\r\n", payload_val);
inv->sendActivePowerControlRequest(payload_val, PowerLimitControlType::RelativPersistent); inv->sendActivePowerControlRequest(payload_val, PowerLimitControlType::RelativPersistent);
break;
} else if (!strcmp(setting, TOPIC_SUB_LIMIT_PERSISTENT_ABSOLUTE)) { case Topic::LimitPersistentAbsolute:
// Set inverter limit absolute persistent // Set inverter limit absolute persistent
MessageOutput.printf("Limit Persistent: %.1f W\r\n", payload_val); MessageOutput.printf("Limit Persistent: %.1f W\r\n", payload_val);
inv->sendActivePowerControlRequest(payload_val, PowerLimitControlType::AbsolutPersistent); inv->sendActivePowerControlRequest(payload_val, PowerLimitControlType::AbsolutPersistent);
break;
} else if (!strcmp(setting, TOPIC_SUB_LIMIT_NONPERSISTENT_RELATIVE)) { case Topic::LimitNonPersistentRelative:
// Set inverter limit relative non persistent // Set inverter limit relative non persistent
MessageOutput.printf("Limit Non-Persistent: %.1f %%\r\n", payload_val); MessageOutput.printf("Limit Non-Persistent: %.1f %%\r\n", payload_val);
if (!properties.retain) { if (!properties.retain) {
@ -217,8 +203,9 @@ void MqttHandleInverterClass::onMqttMessage(const espMqttClientTypes::MessagePro
} else { } else {
MessageOutput.println("Ignored because retained"); MessageOutput.println("Ignored because retained");
} }
break;
} else if (!strcmp(setting, TOPIC_SUB_LIMIT_NONPERSISTENT_ABSOLUTE)) { case Topic::LimitNonPersistentAbsolute:
// Set inverter limit absolute non persistent // Set inverter limit absolute non persistent
MessageOutput.printf("Limit Non-Persistent: %.1f W\r\n", payload_val); MessageOutput.printf("Limit Non-Persistent: %.1f W\r\n", payload_val);
if (!properties.retain) { if (!properties.retain) {
@ -226,13 +213,15 @@ void MqttHandleInverterClass::onMqttMessage(const espMqttClientTypes::MessagePro
} else { } else {
MessageOutput.println("Ignored because retained"); MessageOutput.println("Ignored because retained");
} }
break;
} else if (!strcmp(setting, TOPIC_SUB_POWER)) { case Topic::Power:
// Turn inverter on or off // Turn inverter on or off
MessageOutput.printf("Set inverter power to: %d\r\n", static_cast<int32_t>(payload_val)); MessageOutput.printf("Set inverter power to: %d\r\n", static_cast<int32_t>(payload_val));
inv->sendPowerControlRequest(static_cast<int32_t>(payload_val) > 0); inv->sendPowerControlRequest(static_cast<int32_t>(payload_val) > 0);
break;
} else if (!strcmp(setting, TOPIC_SUB_RESTART)) { case Topic::Restart:
// Restart inverter // Restart inverter
MessageOutput.printf("Restart inverter\r\n"); MessageOutput.printf("Restart inverter\r\n");
if (!properties.retain && payload_val == 1) { if (!properties.retain && payload_val == 1) {
@ -245,29 +234,26 @@ void MqttHandleInverterClass::onMqttMessage(const espMqttClientTypes::MessagePro
void MqttHandleInverterClass::subscribeTopics() void MqttHandleInverterClass::subscribeTopics()
{ {
using std::placeholders::_1; String const& prefix = MqttSettings.getPrefix();
using std::placeholders::_2;
using std::placeholders::_3;
using std::placeholders::_4;
using std::placeholders::_5;
using std::placeholders::_6;
const String topic = MqttSettings.getPrefix(); auto subscribe = [&prefix, this](char const* subTopic, Topic t) {
MqttSettings.subscribe(String(topic + "+/cmd/" + TOPIC_SUB_LIMIT_PERSISTENT_RELATIVE), 0, std::bind(&MqttHandleInverterClass::onMqttMessage, this, _1, _2, _3, _4, _5, _6)); String fullTopic(prefix + _cmdtopic.data() + subTopic);
MqttSettings.subscribe(String(topic + "+/cmd/" + TOPIC_SUB_LIMIT_PERSISTENT_ABSOLUTE), 0, std::bind(&MqttHandleInverterClass::onMqttMessage, this, _1, _2, _3, _4, _5, _6)); MqttSettings.subscribe(fullTopic.c_str(), 0,
MqttSettings.subscribe(String(topic + "+/cmd/" + TOPIC_SUB_LIMIT_NONPERSISTENT_RELATIVE), 0, std::bind(&MqttHandleInverterClass::onMqttMessage, this, _1, _2, _3, _4, _5, _6)); std::bind(&MqttHandleInverterClass::onMqttMessage, this, t,
MqttSettings.subscribe(String(topic + "+/cmd/" + TOPIC_SUB_LIMIT_NONPERSISTENT_ABSOLUTE), 0, std::bind(&MqttHandleInverterClass::onMqttMessage, this, _1, _2, _3, _4, _5, _6)); std::placeholders::_1, std::placeholders::_2,
MqttSettings.subscribe(String(topic + "+/cmd/" + TOPIC_SUB_POWER), 0, std::bind(&MqttHandleInverterClass::onMqttMessage, this, _1, _2, _3, _4, _5, _6)); std::placeholders::_3, std::placeholders::_4,
MqttSettings.subscribe(String(topic + "+/cmd/" + TOPIC_SUB_RESTART), 0, std::bind(&MqttHandleInverterClass::onMqttMessage, this, _1, _2, _3, _4, _5, _6)); std::placeholders::_5, std::placeholders::_6));
};
for (auto const& s : _subscriptions) {
subscribe(s.first.data(), s.second);
}
} }
void MqttHandleInverterClass::unsubscribeTopics() void MqttHandleInverterClass::unsubscribeTopics()
{ {
const String topic = MqttSettings.getPrefix(); String const& prefix = MqttSettings.getPrefix() + _cmdtopic.data();
MqttSettings.unsubscribe(String(topic + "+/cmd/" + TOPIC_SUB_LIMIT_PERSISTENT_RELATIVE)); for (auto const& s : _subscriptions) {
MqttSettings.unsubscribe(String(topic + "+/cmd/" + TOPIC_SUB_LIMIT_PERSISTENT_ABSOLUTE)); MqttSettings.unsubscribe(prefix + s.first.data());
MqttSettings.unsubscribe(String(topic + "+/cmd/" + TOPIC_SUB_LIMIT_NONPERSISTENT_RELATIVE)); }
MqttSettings.unsubscribe(String(topic + "+/cmd/" + TOPIC_SUB_LIMIT_NONPERSISTENT_ABSOLUTE));
MqttSettings.unsubscribe(String(topic + "+/cmd/" + TOPIC_SUB_POWER));
MqttSettings.unsubscribe(String(topic + "+/cmd/" + TOPIC_SUB_RESTART));
} }