Added support for reading master settings from MQTT with no web server

fredilarsen · fredilarsen · commit 093dfcd8fb92 · 2020-12-20T18:05:41.000+01:00
diff --git a/src/MI/ModuleInterfaceMqttTransfer.h b/src/MI/ModuleInterfaceMqttTransfer.h
@@ -21,6 +21,7 @@
 #include <MI/MITransferBase.h>
 #include <utils/MITime.h>
 #include <utils/MIUptime.h>
+#include <utils/MIUtilities.h>
 #include <utils/BinaryBuffer.h>
 
 #include <ReconnectingMqttClient.h>
@@ -42,6 +43,35 @@ class MIMqttTransfer : public MITransferBase {
   // State
   ReconnectingMqttClient client;
 
+  // Debug print related
+  #if defined(DEBUG_PRINT) || defined(DEBUG_PRINT_SETTINGUPDATE_MQTT) || defined(DEBUG_PRINT_TIMES)
+  uint32_t last_settings_debug_print_ms = 0;
+  bool some_settings_missing = true;
+  #endif
+
+
+  void debug_print_missing_settings() {
+    #if defined(DEBUG_PRINT) || defined(DEBUG_PRINT_SETTINGUPDATE_MQTT) || defined(DEBUG_PRINT_TIMES)
+    if (some_settings_missing && mi_interval_elapsed(last_settings_debug_print_ms, 10000)) {
+      some_settings_missing = false;
+      for (uint8_t module_ix = 0; module_ix < interfaces.get_module_count(); module_ix++) {
+        uint8_t initialized = interfaces[module_ix]->settings.get_initialized_count(),
+          total = interfaces[module_ix]->settings.get_num_variables();
+        if (initialized < total) {
+          some_settings_missing = true;
+          printf("Still missing %d settings for %s: ", (total - initialized), interfaces[module_ix]->module_name);
+          for (uint8_t v = 0; v < total; v++) {
+            if (!interfaces[module_ix]->settings.get_module_variable(v).is_initialized()) {
+              printf("%s ", interfaces[module_ix]->settings.get_module_variable(v).name);
+            }
+          }
+          printf("\n");
+        }
+      }
+    }
+    #endif
+  }
+
 public:
   MIMqttTransfer(ModuleInterfaceSet &module_interface_set, const uint8_t *broker_address, const uint16_t broker_port = 1883) :
     MITransferBase(module_interface_set) {
@@ -56,7 +86,7 @@ class MIMqttTransfer : public MITransferBase {
     client.set_address(broker_ip, broker_port, "modulemaster"); 
   }
 
-  void start() {
+  virtual void start() {
     #ifdef MIMQTT_USE_JSON
     client.subscribe("moduleinterface/+/setting,moduleinterface/+/input", 1);
     #else
@@ -190,7 +220,11 @@ class MIMqttTransfer : public MITransferBase {
     #endif
   }
 
-  // Override this in derived classes to read master settings or other topics
+ // Override this in derived classes to read master settings
+  virtual void read_master_topic(const char *modulename, const char *category, 
+                                 const char *topic, const char *data, uint16_t len, uint8_t transfer_ix) {  }
+
+  // Override this in derived classes to read other topics
   virtual void read_nonmodule_topic(const char *modulename, const char *category, 
                                     const char *topic, const char *data, uint16_t len, uint8_t transfer_ix) {  }
 
@@ -228,9 +262,16 @@ class MIMqttTransfer : public MITransferBase {
         }
     }
 */
+
+    #ifdef DEBUG_PRINT_SETTINGUPDATE_MQTT
+    printf("Got topic '%s' for module '%s' ix %d\n", topic, modulename.c_str(), module_ix);
+    #endif
     if (!mi) {
       // Read settings not meant for a module, for example master settings
-      read_nonmodule_topic(modulename.c_str(), category.c_str(), topic, data, len, transfer_ix);
+      if (strncmp("master_", modulename.c_str(), 7) == 0)
+        read_master_topic(modulename.c_str(), category.c_str(), topic, data, len, transfer_ix);
+      else
+        read_nonmodule_topic(modulename.c_str(), category.c_str(), topic, data, len, transfer_ix);
       return;
     }
 
@@ -290,13 +331,19 @@ class MIMqttTransfer : public MITransferBase {
       if (mv.is_changed() && is_event) mv.set_event(true); // Trigger immediate transfer to modules
       #if defined(MASTER_MULTI_TRANSFER)
       mv.set_initialized();
-      if (mvs->is_initialized()) mvs->set_updated(); // All variables have been set, and one was just updated
-      #endif
-      #if defined(MASTER_MULTI_TRANSFER) && defined(DEBUG_PRINT_SETTINGSYNC)
+      if (mvs->is_initialized()) {
+        #if defined(DEBUG_PRINT) || defined(DEBUG_PRINT_SETTINGUPDATE_MQTT) || defined(DEBUG_PRINT_TIMES)
+        if (settings && !mvs->is_updated())
+          printf("GOT ALL settings for %s\n", modulename.c_str());
+        #endif
+        mvs->set_updated(); // All variables have been set, and one was just updated
+      }
+      #ifdef DEBUG_PRINT_SETTINGSYNC
       if (prev_bits != mv.change_bits) 
         printf("FROM MQTT '%s' ix: %d VALUE %ld cbits: %d->%d changed:%d->%d->%d\n", variable_name.c_str(), varpos, 
           mv.get_uint32(), prev_bits, mv.change_bits, prev_changed, mv_new.is_changed(), mv.is_changed());
       #endif
+      #endif
     }
     #endif
   }
diff --git a/src/MI/ModuleVariableSet.h b/src/MI/ModuleVariableSet.h
@@ -409,6 +409,16 @@ struct ModuleVariableSet {
 
   uint8_t get_num_variables() const { return num_variables; }
 
+  #if defined(IS_MASTER) && defined(MASTER_MULTI_TRANSFER)
+  uint8_t get_initialized_count() const {
+    uint8_t count = 0;
+    for (uint8_t i = 0; i < num_variables; i++) {
+      if (variables[i].is_initialized()) count++;
+    }
+    return count;
+  }
+  #endif
+
   uint8_t get_variable_ix(const char *variable_name) const {
     #ifdef IS_MASTER
     for (uint8_t i = 0; i < num_variables; i++)
diff --git a/src/MI_PJON/PJONModuleInterfaceMqttTransfer.h b/src/MI_PJON/PJONModuleInterfaceMqttTransfer.h
@@ -28,19 +28,98 @@ class PJONMIMqttTransfer : public MIMqttTransfer {
   // Whether to read master settings from MQTT or not
   void set_read_master_settings(bool read) { read_master_settings = read; }
 
+  virtual void start() {
+    // When starting based on master settings from MQTT, we are doing it in 3 phases:
+    // 1. Get master settings, create ModuleInterface objects based on module list.
+    // 2. Get contracts from all modules.
+    // 3. Reconnect to MQTT now subscribing to all settings and inputs, so that
+    //    these will be registered in objects and synced to modules.
+
+    // If we have all master settings then start normally (phase 3)
+    if (!read_master_settings || (got_settings() && got_contracts())) {
+      phase = PHASE_RUNNING;
+      MIMqttTransfer::start();
+    }
+    else {
+      // Subscribe to master settings only (phase 1)
+      phase = PHASE_WAIT_MQTT;
+      #if defined(DEBUG_PRINT) || defined(DEBUG_PRINT_SETTINGUPDATE_MQTT)
+      printf("%%%%%%%%%%%% PHASE_WAIT_MQTT starting at %dms.\n", millis());
+      #endif
+      String master_topic = "moduleinterface/master_"; master_topic += interfaces.get_prefix();
+      #ifdef MIMQTT_USE_JSON
+      master_topic += "/setting";
+      #else
+      master_topic += "/setting/+";
+      #endif
+      #if defined(DEBUG_PRINT) || defined(DEBUG_PRINT_SETTINGUPDATE_MQTT)
+      printf("Subscribing to topic '%s'\n", master_topic.c_str());
+      #endif
+      client.subscribe(master_topic.c_str(), 1);
+      client.start();
+    }
+  }
+
 private:
+  // Flags used in got_master_settings member
+  const uint8_t SETTING_MODULE_LIST = 1, SETTING_DEVID = 2, SETTING_INTERVAL = 4, SETTING_ALL = 7;
+
+  // Startup phases
+  enum Phase { PHASE_STOPPED, PHASE_WAIT_MQTT, PHASE_WAIT_CONTRACTS, PHASE_RUNNING };
+
   bool read_master_settings = false;
+  uint8_t got_master_settings = 0;  // Bitmasked, 
+  enum Phase phase = PHASE_STOPPED;
+  long contract_retrieval_start = 0;
+
+  bool got_settings() const { return (got_master_settings & SETTING_ALL) == SETTING_ALL; }
+  bool got_contracts() const { 
+    return interfaces.got_all_contracts() 
+      && ((interfaces.get_inactive_module_count() == 0) 
+        || (uint32_t(millis() - contract_retrieval_start) > MI_INACTIVE_TIME_THRESHOLD*1000l));
+  }
 
-  virtual void read_nonmodule_topic(const char *modulename, 
-                                    const char *category, 
-                                    const char *topic, 
-                                    const char *data, 
-                                    uint16_t    len, 
-                                    uint8_t     transfer_ix)
+  virtual void update() {
+    MIMqttTransfer::update();
+    check_phase();
+    if (read_master_settings) debug_print_missing_settings();
+  }
+
+  void check_phase() {
+    // Check if we are ready to transition to a new connection phase
+    switch(phase) {
+      case PHASE_STOPPED: break;
+      case PHASE_WAIT_MQTT: 
+        if (got_settings()) {
+          phase = PHASE_WAIT_CONTRACTS;
+          contract_retrieval_start = millis();
+          #if defined(DEBUG_PRINT) || defined(DEBUG_PRINT_SETTINGUPDATE_MQTT)
+          printf("%%%%%%%%%%%% PHASE_WAIT_CONTRACTS starting at %dms. Got all master settings.\n", millis());
+          #endif
+        }
+        break;
+      case PHASE_WAIT_CONTRACTS: 
+        if (got_contracts()) { 
+          // Reconnect to MQTT broker wil full subscription, not only master settings
+          #if defined(DEBUG_PRINT) || defined(DEBUG_PRINT_SETTINGUPDATE_MQTT)
+          printf("%%%%%%%%%%%% PHASE_RUNNING starting at %dms. Got %d (%d inactive) modules.\n", millis(), interfaces.get_module_count(), interfaces.get_inactive_module_count());
+          #endif
+          phase = PHASE_RUNNING;
+          stop(); 
+          start();
+        }
+        break;
+      case PHASE_RUNNING: break;
+    }
+  }
+
+  virtual void read_master_topic(const char *modulename, 
+                                 const char *category, 
+                                 const char *topic, 
+                                 const char *data, 
+                                 uint16_t    len, 
+                                 uint8_t     transfer_ix)
   {
-printf("Starting on read_nonmodule_topic '%s'\n", topic);
-if (strncmp(topic, "moduleinterface/master_m1/setting/modules", 33) == 0) printf("------> Modules: '%s'\n", data);
-printf("%d Category: %s\n", read_master_settings, category);
     if (!read_master_settings) return;
     bool for_me = false;
     {
@@ -50,32 +129,42 @@ printf("%d Category: %s\n", read_master_settings, category);
     if (for_me && strcmp(category, "setting")==0) {
       // A setting for this master
       #ifdef MIMQTT_USE_JSON
-      read_master_json_settings_from_buffer(interfaces, data, len);
+      bool ok = read_master_json_settings_from_buffer((PJONModuleInterfaceSet&) interfaces, data, len);
+      got_master_settings = ok ? SETTING_ALL : 0;
+      // Check if module list was changed and objects recreated
+      bool all_empty = true;
+      for (uint8_t m = 0; m < interfaces.get_module_count(); m++) {
+        if (interfaces[m]->settings.get_num_variables() > 0 || interfaces[m]->outputs.get_num_variables() > 0) {
+          all_empty = false; break;
+        }
+      }
+      if (all_empty) phase = PHASE_WAIT_MQTT; // Trigger exchange of contracts followed by reconnect to get all settings
       #else
       if (strstr(topic, "/modules")>0) {
-       if (((PJONModuleInterfaceSet&)interfaces).set_interface_list(data)) {
-         // The module list changed. Reconnect to get all settings from MQTT to new module objects.
-printf("--> GOT NEW MODULE LIST. Reconnecting to get settings.\n");
-          #ifdef DEBUG_PRINT
-         DPRINTLN("--> GOT NEW MODULE LIST. Reconnecting to get settings.");
+        if (((PJONModuleInterfaceSet&)interfaces).set_interface_list(data)) {
+          // The module list changed. Reconnect to get all settings from MQTT to new module objects.
+          got_master_settings |= SETTING_MODULE_LIST;
+          phase = PHASE_WAIT_MQTT;
+          #if defined(DEBUG_PRINT) || defined(DEBUG_PRINT_SETTINGUPDATE_MQTT)
+          if (phase != PHASE_WAIT_MQTT) DPRINTLN("--> GOT NEW MODULE LIST. Changing to PHASE_WAIT_MQTT.");
           #endif
-         stop();
-         start();
-       }
-        #ifdef DEBUG_PRINT
+        }
+        #if defined(DEBUG_PRINT) || defined(DEBUG_PRINT_SETTINGUPDATE_MQTT)
         DPRINT("Modules: '"); DPRINT(data); DPRINTLN("'");
         #endif
       } else if (strstr(topic, "/devid")>0) {
         // Set PJON id for master
+        got_master_settings |= SETTING_DEVID;
         uint8_t device_id = (uint8_t) atoi(data);
         if (device_id != 0) ((PJONModuleInterfaceSet&)interfaces).get_link()->set_id(device_id);
       } else if (strstr(topic, "/intsettings")>0) {
         // Set time interval between exchanges
+        got_master_settings |= SETTING_INTERVAL;
         uint32_t interval = atoi(data);
         if (interval > 100) ((PJONModuleInterfaceSet&)interfaces).set_transfer_interval(interval);
       }
       #endif
+      check_phase();
     }
-printf("Finished with read_nonmodule_topic");
   }
 };
