Merge pull request letscontrolit#5009 from TD-er/feature/UpdateCULreader

[CUL Reader] Cherry pick code from ESPEasy_NOW pull request

src/src/DataStructs/mBusPacket.h

@@ -0,0 +1,120 @@
+#ifndef DATASTRUCTS_MBUSPACKET_H
+#define DATASTRUCTS_MBUSPACKET_H
+
+#include "../../ESPEasy_common.h"
+
+#include <vector>
+
+
+// 0 is sometimes used ("@@@")
+// 0xFFFF does not seem to be used ("___")
+#define mBus_packet_wildcard_manufacturer  0xFFFF
+
+// 0 is a valid meter type and 0xFF seems to be reserved
+#define mBus_packet_wildcard_metertype  0xFE
+
+// 0 is a valid serial and 0xFFFFFFFF seems to be reserved
+#define mBus_packet_wildcard_serial  0xFFFFFFFE
+
+
+typedef std::vector<uint8_t> mBusPacket_data;
+
+struct mBusPacket_header_t {
+  mBusPacket_header_t();
+
+  static String decodeManufacturerID(int id);
+  static int    encodeManufacturerID(const String& id_str);
+
+  String        getManufacturerId() const;
+
+  String        toString() const;
+
+  uint64_t      encode_toUInt64() const;
+
+  void          decode_fromUint64(uint64_t encodedValue);
+
+  bool          isValid() const;
+
+  bool          matchSerial(uint32_t serialNr) const;
+
+  void          clear();
+
+  // Use for stats as key:
+  union {
+    uint64_t _encodedValue{};
+    struct {
+      uint64_t _serialNr     : 32;
+      uint64_t _manufacturer : 16;
+      uint64_t _meterType    : 8;
+
+      // Use for filtering
+      uint64_t _length : 8;
+    };
+  };
+};
+
+struct mBusPacket_t {
+public:
+
+  bool                       parse(const String& payload);
+
+  // Get the header of the actual device, not the forwarding device (if present)
+  const mBusPacket_header_t* getDeviceHeader() const;
+
+  static  int16_t            decode_LQI_RSSI(uint16_t lqi_rssi,
+                                             uint8_t& LQI);
+
+  bool                       matchSerial(uint32_t serialNr) const;
+
+  uint32_t                   getDeviceSerial() const;
+
+  String                     toString() const;
+
+  // 32 bit value used to generate a map key for filtering
+  // Essentially the XOR of the first 32-bit with the second 32-bit of
+  // serial, manufacturer, metertype and length.
+  uint32_t deviceID_to_map_key() const;
+
+  uint32_t deviceID_to_map_key_no_length() const;
+
+private:
+
+  static uint32_t deviceID_to_map_key(uint64_t id1, uint64_t id2);
+
+  static uint8_t         hexToByte(const String& str,
+                                   size_t        index);
+
+  static mBusPacket_data removeChecksumsFrameA(const String& payload,
+                                               uint32_t    & checksum);
+  static mBusPacket_data removeChecksumsFrameB(const String& payload,
+                                               uint32_t    & checksum);
+
+  bool                   parseHeaders(const mBusPacket_data& payloadWithoutChecksums);
+
+public:
+
+  mBusPacket_header_t _deviceId1;
+  mBusPacket_header_t _deviceId2;
+  uint16_t            _lqi_rssi{};
+
+
+  /*
+     // Statistics:
+     // Key:
+     deviceID1:
+     - manufacturer
+     - metertype
+     - serialnr
+
+     // Value:
+     - message count
+     - rssi
+     - lqi???
+   */
+
+
+  // Checksum based on the XOR of all removed checksums from the message
+  uint32_t _checksum = 0;
+};
+
+#endif // ifndef DATASTRUCTS_MBUSPACKET_H

src/src/DataTypes/ESPEasy_plugin_functions.h

@@ -62,6 +62,9 @@ enum PluginFunctions_e {
    PLUGIN_PRIORITY_INIT_ALL           , // Pre-initialize all plugins that are set to PowerManager priority (not implemented in plugins)
    PLUGIN_PRIORITY_INIT               , // Pre-initialize a singe plugins that is set to PowerManager priority
    PLUGIN_WEBFORM_LOAD_ALWAYS         , // Loaded *after* PLUGIN_WEBFORM_LOAD, also shown for remote data-feed devices
+#ifdef USES_ESPEASY_NOW
+   PLUGIN_FILTEROUT_CONTROLLER_DATA   , // Can be called from the controller to query a task whether the data should be processed further.
+#endif
    PLUGIN_WEBFORM_PRE_SERIAL_PARAMS   , // Before serial parameters, convert additional parameters like baudrate or specific serial config
 
    PLUGIN_MAX_FUNCTION  // Leave as last one.

src/src/ESPEasyCore/Controller.cpp

@@ -36,15 +36,15 @@ constexpr pluginID_t PLUGIN_ID_MQTT_IMPORT(37);
 // ********************************************************************************
 // Interface for Sending to Controllers
 // ********************************************************************************
-void sendData(struct EventStruct *event)
+void sendData(struct EventStruct *event, bool sendEvents)
 {
   START_TIMER;
   #ifndef BUILD_NO_RAM_TRACKER
   checkRAM(F("sendData"));
   #endif // ifndef BUILD_NO_RAM_TRACKER
 //  LoadTaskSettings(event->TaskIndex);
 
-  if (Settings.UseRules) {
+  if (Settings.UseRules && sendEvents) {
     createRuleEvents(event);
   }
 

src/src/ESPEasyCore/Controller.h

@@ -9,7 +9,7 @@
 // ********************************************************************************
 // Interface for Sending to Controllers
 // ********************************************************************************
-void sendData(struct EventStruct *event);
+void sendData(struct EventStruct *event, bool sendEvents = true);
 
 bool validUserVar(struct EventStruct *event);
 

src/src/Globals/Plugins.cpp

@@ -854,6 +854,9 @@ bool PluginCall(uint8_t Function, struct EventStruct *event, String& str)
     case PLUGIN_I2C_HAS_ADDRESS:
     case PLUGIN_WEBFORM_SHOW_ERRORSTATE_OPT:
     case PLUGIN_INIT_VALUE_RANGES:
+    #ifdef USES_ESPEASY_NOW
+    case PLUGIN_FILTEROUT_CONTROLLER_DATA:
+    #endif
 
     // PLUGIN_MQTT_xxx functions are directly called from the scheduler.
     //case PLUGIN_MQTT_CONNECTION_STATE:

src/src/Helpers/CUL_interval_filter.cpp

@@ -0,0 +1,101 @@
+#include "../Helpers/CUL_interval_filter.h"
+
+
+#ifdef USES_P094
+
+# include "../ESPEasyCore/ESPEasy_Log.h"
+# include "../Globals/ESPEasy_time.h"
+# include "../Globals/TimeZone.h"
+# include "../Helpers/ESPEasy_time_calc.h"
+# include "../Helpers/StringConverter.h"
+
+
+CUL_time_filter_struct::CUL_time_filter_struct(uint32_t checksum, unsigned long UnixTimeExpiration)
+  : _checksum(checksum), _UnixTimeExpiration(UnixTimeExpiration) {}
+
+String CUL_interval_filter_getExpiration_log_str(const P094_filter& filter)
+{
+  const unsigned long expiration = filter.computeUnixTimeExpiration();
+
+  if ((expiration != 0) && (expiration != 0xFFFFFFFF)) {
+    struct tm exp_tm;
+    breakTime(time_zone.toLocal(expiration), exp_tm);
+
+    return concat(F(" Expiration: "), formatDateTimeString(exp_tm));
+  }
+  return EMPTY_STRING;
+}
+
+bool CUL_interval_filter::filter(const mBusPacket_t& packet, const P094_filter& filter)
+{
+  if (!enabled) {
+    return true;
+  }
+
+  if (filter.getFilterWindow() == P094_Filter_Window::None) {
+    // Will always be rejected, so no need to keep track of the message
+    return false;
+  }
+
+  if (filter.getFilterWindow() == P094_Filter_Window::All) {
+    // Will always be allowed, so no need to keep track of the message
+    return true;
+  }
+
+  const uint32_t key = packet.deviceID_to_map_key();
+  auto it            = _mBusFilterMap.find(key);
+
+  if (it != _mBusFilterMap.end()) {
+    // Already present
+    if (node_time.getUnixTime() < it->second._UnixTimeExpiration) {
+      if (loglevelActiveFor(LOG_LEVEL_INFO)) {
+        String log = concat(F("CUL   : Interval filtered: "), packet.toString());
+        log += CUL_interval_filter_getExpiration_log_str(filter);
+        addLogMove(LOG_LEVEL_INFO, log);
+      }
+      return false;
+    }
+
+    if (packet._checksum == it->second._checksum) {
+      if (loglevelActiveFor(LOG_LEVEL_INFO)) {
+        addLogMove(LOG_LEVEL_INFO, concat(F("CUL   : Interval Same Checksum: "), packet.toString()));
+      }
+      return false;
+    }
+
+    // Has expired, so remove from filter map
+    _mBusFilterMap.erase(it);
+  }
+
+  const unsigned long expiration = filter.computeUnixTimeExpiration();
+
+  CUL_time_filter_struct item(packet._checksum, expiration);
+
+  _mBusFilterMap[key] = item;
+
+  if (loglevelActiveFor(LOG_LEVEL_INFO)) {
+    String log = concat(F("CUL   : Add to IntervalFilter: "), packet.toString());
+    log += CUL_interval_filter_getExpiration_log_str(filter);
+
+    addLogMove(LOG_LEVEL_INFO, log);
+  }
+
+  return true;
+}
+
+void CUL_interval_filter::purgeExpired()
+{
+  auto it = _mBusFilterMap.begin();
+
+  const unsigned long currentTime = node_time.getUnixTime();
+
+  for (; it != _mBusFilterMap.end();) {
+    if (currentTime > it->second._UnixTimeExpiration) {
+      it = _mBusFilterMap.erase(it);
+    } else {
+      ++it;
+    }
+  }
+}
+
+#endif // ifdef USES_P094

src/src/Helpers/CUL_interval_filter.h

@@ -0,0 +1,42 @@
+#ifndef DATASTRUCTS_P094_CUL_TIME_FILTER_STRUCT_H
+#define DATASTRUCTS_P094_CUL_TIME_FILTER_STRUCT_H
+
+#include "../../ESPEasy_common.h"
+#ifdef USES_P094
+
+# include "../DataStructs/mBusPacket.h"
+# include "../PluginStructs/P094_Filter.h"
+
+# include <map>
+
+
+struct CUL_time_filter_struct {
+  CUL_time_filter_struct() = default;
+  CUL_time_filter_struct(uint32_t      checksum,
+                         unsigned long UnixTimeExpiration);
+
+  uint32_t      _checksum{};
+  unsigned long _UnixTimeExpiration{};
+};
+
+typedef uint32_t mBusSerial;
+
+typedef std::map<mBusSerial, CUL_time_filter_struct> mBusFilterMap;
+
+
+struct CUL_interval_filter {
+  // Return true when packet wasn't already present.
+  bool filter(const mBusPacket_t& packet,
+              const P094_filter & filter);
+
+  // Remove packets that have expired.
+  void purgeExpired();
+
+
+  mBusFilterMap _mBusFilterMap;
+
+  bool enabled = false;
+};
+
+#endif // ifdef USES_P094
+#endif // ifndef DATASTRUCTS_P094_CUL_TIME_FILTER_STRUCT_H