From 1e21850a2b07b2ceb7cb2f4418e633bdb32f7bd2 Mon Sep 17 00:00:00 2001 From: John Date: Sat, 22 Jan 2022 15:28:32 +0100 Subject: [PATCH] extend MQTT integration support --- src/ebusd/mqtthandler.cpp | 143 +++++++++++++++++++------- src/ebusd/mqtthandler.h | 20 +++- src/lib/ebus/datatype.h | 9 +- src/lib/ebus/filereader.cpp | 92 +++++++++++++++++ src/lib/ebus/filereader.h | 14 +++ src/lib/ebus/message.cpp | 3 + src/lib/ebus/test/test_filereader.cpp | 70 +++++++++++++ 7 files changed, 311 insertions(+), 40 deletions(-) diff --git a/src/ebusd/mqtthandler.cpp b/src/ebusd/mqtthandler.cpp index 8a3bbfc2..bdbf65ea 100755 --- a/src/ebusd/mqtthandler.cpp +++ b/src/ebusd/mqtthandler.cpp @@ -24,6 +24,7 @@ #include #include #include "lib/utils/log.h" +#include "lib/ebus/symbol.h" namespace ebusd { @@ -573,16 +574,26 @@ MqttReplacer& MqttReplacers::get(const string& key) { return m_replacers[key]; } -string MqttReplacers::get(const string& key, bool untilFirstEmpty, bool onlyAlphanum) const { +string MqttReplacers::get(const string& key, bool untilFirstEmpty, bool onlyAlphanum, const string& fallbackKey) const { auto itc = m_constants.find(key); if (itc!=m_constants.end()) { return itc->second; } auto itv = m_replacers.find(key); - if (itv==m_replacers.end()) { - return ""; + if (itv!=m_replacers.end()) { + return itv->second.get(m_constants, untilFirstEmpty, onlyAlphanum); } - return itv->second.get(m_constants, untilFirstEmpty, onlyAlphanum); + if (!fallbackKey.empty()) { + itc = m_constants.find(fallbackKey); + if (itc!=m_constants.end()) { + return itc->second; + } + itv = m_replacers.find(fallbackKey); + if (itv!=m_replacers.end()) { + return itv->second.get(m_constants, untilFirstEmpty, onlyAlphanum); + } + } + return ""; } bool MqttReplacers::set(const string& key, const string& value, bool removeReplacer) { @@ -590,7 +601,7 @@ bool MqttReplacers::set(const string& key, const string& value, bool removeRepla if (removeReplacer) { m_replacers.erase(key); } - if (key.find('_')!=string::npos) { + if (key.find_first_of("-_")!=string::npos) { return false; } string upper = key; @@ -779,10 +790,10 @@ MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap* m_replacers.reduce(); } } - m_hasDefinitionTopic = !m_replacers.get("definition_topic", false, false).empty(); + m_hasDefinitionTopic = !m_replacers.get("definition-topic", false, false).empty(); m_hasDefinitionFieldsPayload = m_replacers.uses("fields_payload"); - m_subscribeConfigRestartTopic = m_replacers.get("config_restart_topic", false, false); - m_subscribeConfigRestartPayload = m_replacers.get("config_restart_payload", false, false); + m_subscribeConfigRestartTopic = m_replacers.get("config_restart-topic", false, false); + m_subscribeConfigRestartPayload = m_replacers.get("config_restart-payload", false, false); m_globalTopic = getTopic(nullptr, "global/"); m_subscribeTopic = getTopic(nullptr, "#"); if (check(mosquitto_lib_init(), "unable to initialize")) { @@ -1047,50 +1058,95 @@ void MqttHandler::run() { updates << dec << static_cast(uptime); publishTopic(uptimeTopic, updates.str()); } + if (m_connected && m_definitionsSince == 0) { + publishDefinition(m_replacers, "def_global_running-", m_globalTopic + "running", "global", "running", "def_global-"); + publishDefinition(m_replacers, "def_global_version-", m_globalTopic + "version", "global", "version", "def_global-"); + publishDefinition(m_replacers, "def_global_signal-", signalTopic, "global", "signal", "def_global-"); + publishDefinition(m_replacers, "def_global_uptime-", uptimeTopic, "global", "uptime", "def_global-"); + publishDefinition(m_replacers, "def_global_updatecheck-", m_globalTopic + "updatecheck", "global", "updatecheck", "def_global-"); + publishDefinition(m_replacers, "def_global_scan-", m_globalTopic + "scan", "global", "scan", "def_global-"); + m_definitionsSince = 1; + } if (m_connected && m_hasDefinitionTopic) { deque messages; + result_t result = RESULT_OK; + unsigned int filterPriority = parseInt(m_replacers["filter-priority"].c_str(), 10, 0, 9, &result); + if (result != RESULT_OK) { + filterPriority = 0; + } + string filterCircuit = m_replacers["filter-circuit"]; + FileReader::tolower(&filterCircuit); + string filterName = m_replacers["filter-name"]; + FileReader::tolower(&filterName); + string filterLevel = m_replacers["filter-level"]; + FileReader::tolower(&filterLevel); + string filterField = m_replacers["filter-field"]; + FileReader::tolower(&filterField); m_messages->findAll("", "", "", false, true, true, true, true, true, 0, 0, false, &messages); for (const auto& message : messages) { if (message->getCreateTime() <= m_definitionsSince) { // only newer defined continue; } + if ((filterPriority>0 && (message->getPollPriority()==0 || message->getPollPriority()>filterPriority)) + || !FileReader::matches(message->getCircuit(), filterCircuit, true, true) + || !FileReader::matches(message->getName(), filterName, true, true) + || !FileReader::matches(message->getLevel(), filterLevel, true, true)) { + continue; + } MqttReplacers msgValues = m_replacers; // need a copy here as the contents are manipulated msgValues.set("circuit", message->getCircuit()); msgValues.set("name", message->getName()); + msgValues.set("priority", static_cast(message->getPollPriority())); + msgValues.set("level", message->getLevel()); + msgValues.set("direction", message->isWrite() ? (message->isPassive() ? "uw" : "w") : (message->isPassive() ? "r" : "u")); if (!m_publishByField) { msgValues.set("topic", getTopic(message, "", "")); // TODO already present? } msgValues.reduce(); ostringstream fields; - for (size_t index = 0; index < message->getFieldCount(); index++) { + size_t fieldCount = message->getFieldCount(); + for (size_t index = 0; index < fieldCount; index++) { const SingleDataField* field = message->getField(index); if (!field || field->isIgnored()) { continue; } + string fieldName = message->getFieldName(index); + if (fieldName.empty() && fieldCount == 1) { + fieldName = "0"; // TODO should not occur + } + if (!FileReader::matches(fieldName, filterField, true, true)) { + continue; + } const DataType* dataType = field->getDataType(); string typeSuffix; if (dataType->isNumeric()) { - typeSuffix = "number"; + typeSuffix = dataType->getBitCount()<8 ? "bits" : "number"; + } else if (dataType->hasFlag(DAT)) { + auto dt = dynamic_cast(dataType); + if (dt->hasDate()) { + typeSuffix = dt->hasDate() ? "datetime" : "date"; + } else { + typeSuffix = "time"; + } } else { typeSuffix = "string"; } - //TODO valuelists => binary_sensor, date/time => device_class=date/timestamp, energy, power, current, gas, humidity, power_factor, pressure, temperature, voltage - string str = msgValues.get("type_"+typeSuffix, false, false); + //TODO energy, power, current, gas, humidity, power_factor, pressure, temperature, voltage + string str = msgValues.get("type-"+typeSuffix, false, false); if (str.empty()) { continue; } MqttReplacers values = msgValues; // need a copy here as the contents are manipulated values.set("type", str); values.set("index", static_cast(index)); - string fieldName = message->getFieldName(index); values.set("field", fieldName); values.set("fieldcomment", field->getAttribute("comment")); values.set("unit", field->getAttribute("unit")); - if (dataType->isNumeric()) { - auto dt = dynamic_cast(dataType); - values.set("min", static_cast(dt->getMinValue())); - values.set("max", static_cast(dt->getMaxValue())); - } +// if (dataType->isNumeric()) { +// auto dt = dynamic_cast(dataType); +// values.set("min", static_cast(dt->getMinValue())); +// values.set("max", static_cast(dt->getMaxValue())); +// } values.reduce(); str = values.get("type_part_"+typeSuffix, false, false); values.set("type_part", str); @@ -1106,29 +1162,14 @@ void MqttHandler::run() { fields << values["field_separator"]; } fields << value; - // TODO str << ",\"value_template\":\"{{value_json." << fieldName << ".value}}\""; } continue; } - string topic = values.get("definition_topic", false, false); - if (!topic.empty()) { - values.set("definition_topic", topic); - string payload = values.get("definition_payload", false, false); - string retainStr = values.get("definition_retain", false, false); - bool retain = !retainStr.empty() && !(retainStr=="0" || retainStr=="no" || retainStr=="false"); - publishTopic(topic, payload, retain); - } + publishDefinition(values); } if (fields.tellp()>0) { msgValues.set("fields_payload", fields.str()); - string topic = msgValues.get("definition_topic", false, false); - if (!topic.empty()) { - msgValues.set("definition_topic", topic); - string payload = msgValues.get("definition_payload", false, false); - string retainStr = msgValues.get("definition_retain", false, false); - bool retain = !retainStr.empty() && !(retainStr == "0" || retainStr == "no" || retainStr == "false"); - publishTopic(topic, payload, retain); - } + publishDefinition(msgValues); } } time(&m_definitionsSince); @@ -1181,6 +1222,38 @@ void MqttHandler::run() { publishTopic(m_globalTopic+"scan", "", true); // clear retain of scan status } +void MqttHandler::publishDefinition(MqttReplacers values, const string& prefix, const string& topic, + const string& circuit, const string& name, const string& fallbackPrefix) { + bool reduce = false; + if (!topic.empty()) { + values.set("topic", topic); + reduce = true; + } + if (!circuit.empty()) { + values.set("circuit", circuit); + reduce = true; + } + if (!name.empty()) { + values.set("name", name); + reduce = true; + } + if (reduce) { + values.reduce(); + } + bool noFallback = fallbackPrefix.empty(); + string defTopic = values.get(prefix+"topic", false, false, + noFallback ? "" : fallbackPrefix+"topic"); + if (defTopic.empty()) { + return; + } + string payload = values.get(prefix+"payload", false, false, + noFallback ? "" : fallbackPrefix+"payload"); + string retainStr = values.get(prefix+"retain", false, false, + noFallback ? "" : fallbackPrefix+"retain"); + bool retain = !retainStr.empty() && !(retainStr=="0" || retainStr=="no" || retainStr=="false"); + publishTopic(defTopic, payload, retain); +} + bool MqttHandler::handleTraffic(bool allowReconnect) { if (!m_mosquitto) { return false; diff --git a/src/ebusd/mqtthandler.h b/src/ebusd/mqtthandler.h index f9bf7bc7..52e35096 100644 --- a/src/ebusd/mqtthandler.h +++ b/src/ebusd/mqtthandler.h @@ -180,9 +180,12 @@ class MqttReplacers { /** * Get the variable or constant value of the specified key. * @param key the key for which to get the value. - * @param value the value string or empty. + * @param untilFirstEmpty true to only return the prefix before the first empty field. + * @param onlyAlphanum whether to only allow alpha numeric characters plus underscore. + * @param fallbackKey optional fallback key to use when key value is undefined. + * @return the value string or empty. */ - string get(const string& key, bool untilFirstEmpty, bool onlyAlphanum = false) const; + string get(const string& key, bool untilFirstEmpty, bool onlyAlphanum = false, const string& fallbackKey = "") const; /** * Set the constant value of the specified key and additionally normalized with uppercase key only (if the key does @@ -264,6 +267,19 @@ class MqttHandler : public DataSink, public DataSource, public WaitThread { private: + /** + * Publish a definition topic as specified in the given values. + * @param values the values with the message specification. + * @param prefix the prefix for picking the message specification from the values (before "topic", "payload", and + * "retain"). + * @param topic optional data topic (not the definition topic) to set before building the topic/payload, or empty. + * @param circuit optional circuit to set before building the topic/payload, or empty. + * @param name optional name to set before building the topic/payload, or empty. + * @param fallbackPrefix optional fallback prefix to use when topic/payload/retain with prefix above is not defined. + */ + void publishDefinition(MqttReplacers values, const string& prefix = "definition-", const string& topic = "", + const string& circuit = "", const string& name = "", const string& fallbackPrefix = ""); + /** * Called regularly to handle MQTT traffic. * @param allowReconnect true when reconnecting to the broker is allowed. diff --git a/src/lib/ebus/datatype.h b/src/lib/ebus/datatype.h index d4e76fec..c24f3043 100755 --- a/src/lib/ebus/datatype.h +++ b/src/lib/ebus/datatype.h @@ -175,11 +175,14 @@ enum PartType { /** bit flag for @a DataType: numeric type with base class @a NumberDataType. */ #define NUM 0x400 +/** bit flag for @a DataType: numeric type with base class @a DateTimeDataType. */ +#define DAT 0x800 + /** bit flag for @a DataType: special marker for certain types. */ -#define SPE 0x800 +#define SPE 0x1000 /** bit flag for @a DataType: stored duplicate for backwards compatibility, not to be traversed in lists any more. */ -#define DUP 0x1000 +#define DUP 0x2000 /** * Base class for all kinds of data types. @@ -364,7 +367,7 @@ class DateTimeDataType : public DataType { */ DateTimeDataType(const string& id, size_t bitCount, uint16_t flags, unsigned int replacement, bool hasDate, bool hasTime, int16_t resolution) - : DataType(id, bitCount, flags, replacement), m_hasDate(hasDate), m_hasTime(hasTime), + : DataType(id, bitCount, flags|DAT, replacement), m_hasDate(hasDate), m_hasTime(hasTime), m_resolution(resolution == 0 ? 1 : resolution) {} /** diff --git a/src/lib/ebus/filereader.cpp b/src/lib/ebus/filereader.cpp index aded44dc..de9bf963 100755 --- a/src/lib/ebus/filereader.cpp +++ b/src/lib/ebus/filereader.cpp @@ -118,6 +118,98 @@ void FileReader::tolower(string* str) { transform(str->begin(), str->end(), str->begin(), ::tolower); } +bool FileReader::matches(const string& input, const string& search, bool ignoreCase, bool searchIsLower) { + if (search.empty()) { + return true; // empty pattern matches everything + } + if (ignoreCase) { + string inputSub = input; + tolower(&inputSub); + if (searchIsLower) { + return matches(inputSub, search, false, true); + } + string searchSub = search; + tolower(&inputSub); + return matches(inputSub, searchSub, false, false); + } + // walk through alternatives + size_t from = 0; + bool found = false; + do { + size_t to = search.find('|', from); + if (to == string::npos) { + to = search.length(); + } else { + found = true; + } + if (from==to) { + return true; // empty pattern matches everything + } + size_t nextStart = to+1; + bool matchStart = search[from] == '^'; + if (matchStart) { + from++; + } + bool matchEnd = from POLL_PRIORITY_CONDITION)) { usePriority = POLL_PRIORITY_CONDITION; } + if (m_pollPriority != usePriority) { + time(&m_createTime); // mis-use creation time for this update + } bool ret = m_pollPriority == 0 && usePriority > 0; m_pollPriority = usePriority; if (ret || m_pollOrder > g_lastPollOrder+(unsigned int)m_pollPriority) { diff --git a/src/lib/ebus/test/test_filereader.cpp b/src/lib/ebus/test/test_filereader.cpp index 552424aa..3c4b0a17 100755 --- a/src/lib/ebus/test/test_filereader.cpp +++ b/src/lib/ebus/test/test_filereader.cpp @@ -47,6 +47,50 @@ void verify(bool expectFailMatch, string type, string input, } } +string matchInputs[] = { + // expected result (+=true, -=false), pattern, test strings + "+", "", "", ".", "*", + "+", "*", ".", "*", "**", + "+", "a", "hallo", "a", + "-", "a", "hi", "b", + "+", "^a", "aber", + "-", "^a", "reba", + "+", "^a*", "aber", + "-", "^a*", "reba", + "+", "^*a", "reba", + "-", "^*a", "rebx", + "+", "a$", "reba", + "-", "a$", "aber", + "+", "*a$", "reba", + "-", "*a$", "aber", + "+", "a*$", "aber", + "-", "a*$", "xber", + "+", "|.", "", ".", "*", + "+", ".|", "", ".", "*", + "+", "*|", ".", "*", "**", + "+", "a|z", "hallo", "a", + "-", "a|z", "hi", "b", + "+", "^a|z", "aber", + "-", "^a|z", "reba", + "+", "^a*|z", "aber", + "-", "^a*|z", "reba", + "+", "^*a|z", "reba", + "-", "^*a|z", "rebx", + "+", "a$|z", "reba", + "-", "a$|z", "aber", + "+", "*a$|z", "reba", + "-", "*a$|z", "aber", + "+", "a*$|z", "aber", + "-", "a*$|z", "xber", + "+", "a|*", "hi", "b", + "+", "^a|e*a", "reba", + "+", "^a*|r*a", "reba", + "+", "^*a|^r*x$", "rebx", + "+", "a$|^*$", "aber", + "+", "*a$|^*", "aber", + "+", "a*$|*$", "xber", +}; + string resultlines[][3] = { {"col 1", "col 2", "col 3"}, {"line 2 col 1 de", "line 2 col 2", "line 2 \"col 3\";default of col 3"}, @@ -215,6 +259,32 @@ int main(int argc, char** argv) { } return error ? 1 : 0; } + + bool expectResult = true; + bool nextPattern = true; + string pattern = ""; + for (auto& str : matchInputs) { + if (str == "+" || str == "-") { + expectResult = str == "+"; + nextPattern = true; + continue; + } + if (nextPattern) { + pattern = str; + nextPattern = false; + continue; + } + bool result = FileReader::matches(str, pattern); + cout << "matches(\"" << str << "\", \"" << pattern << "\") = " << (result ? "true" : "false"); + if (result==expectResult) { + cout << ": OK"; + } else { + cout << ": wrong"; + error = true; + } + cout << endl; + } + baseLine = __LINE__+1; istringstream ifs( "col 1.en,col 1.de,col 2,col 3\n"