implement mqtt dynamic value publishing

include/ec.h

@@ -2,6 +2,12 @@
 #define _EC_H_
 
 #include <Arduino.h>
+mqttValueTiming timing_ec_adc;
+mqttValueTiming timing_ec_calibadc;
+mqttValueTiming timing_ec_adcadjusted;
+mqttValueTiming timing_ec_ec;
+mqttValueTiming timing_ec_sc;
+
 
 
 bool ec_flag_measurement_available=false;
@@ -29,7 +35,7 @@ uint16_t ec_calib_array_pos=0;
 uint16_t ec_array[EC_ARRAY_SIZE];
 uint16_t ec_array_pos=EC_ARRAY_SIZE;
 unsigned long last_measurement_ec=0;
-#define EC_MEASUREMENT_INTERVAL 30000 //complete filtered measurement every x ms
+#define EC_MEASUREMENT_INTERVAL 10*60*1000 //complete filtered measurement every x ms
   //One filtered measurement takes EC_READ_INTERVAL*EC_ARRAY_SIZE*4
 #define EC_READ_INTERVAL 10 //interval of reading adc value inside a measurement. one reading takes about 9-10ms
 
@@ -73,6 +79,33 @@ float ec_getECfromADC(float adc);
 float ec_calculateEC25(float pEC,float pTemp);
 
 void ec_setup() {
+  
+  timing_ec_adc.minchange=0.0;
+  timing_ec_adc.maxchange=250;
+  timing_ec_adc.mintime=10*000;
+  timing_ec_adc.maxtime=60*60*1000;
+
+  timing_ec_calibadc.minchange=0.0;
+  timing_ec_calibadc.maxchange=250;
+  timing_ec_calibadc.mintime=10*000;
+  timing_ec_calibadc.maxtime=60*60*1000;
+
+  timing_ec_adcadjusted.minchange=0.0;
+  timing_ec_adcadjusted.maxchange=2.0;
+  timing_ec_adcadjusted.mintime=10*000;
+  timing_ec_adcadjusted.maxtime=30*60*1000;
+
+  timing_ec_ec.minchange=0.0;
+  timing_ec_ec.maxchange=50;
+  timing_ec_ec.mintime=10*000;
+  timing_ec_ec.maxtime=60*60*1000;
+
+  timing_ec_sc.minchange=0.0;
+  timing_ec_sc.maxchange=50;
+  timing_ec_sc.mintime=10*000;
+  timing_ec_sc.maxtime=60*60*1000;
+
+
   ledcSetup(EC_PWM_CH, EC_FREQUENCY, EC_RESOLUTION);
   ledcAttachPin(EC_PIN_FREQ, EC_PWM_CH);
   ledcWrite(EC_PWM_CH, 127); //50% duty cycle
@@ -86,11 +119,13 @@ void ec_loop(unsigned long loopmillis) {
   
   static unsigned long last_read_ec=0;
 
+
   switch (ecstate) {
     case IDLE:
       
-      if (loopmillis>last_measurement_ec+EC_MEASUREMENT_INTERVAL && ecstate==IDLE) { //start measurement if idle
+      if (loopmillis>last_measurement_ec+EC_MEASUREMENT_INTERVAL || force_ec_measurement) { //start measurement if idle
         last_measurement_ec=loopmillis;
+        force_ec_measurement=false;
         ec_startMeasurement();
         ec_connectProbe(true);
         
@@ -135,6 +170,7 @@ void ec_loop(unsigned long loopmillis) {
         
         ec_array_pos++;
       }
+      
     }
   }else{ //measurement not running, then take calibration readings
     if (loopmillis>last_read_ec+EC_CALIB_READ_INTERVAL) { //take reading into array
@@ -148,7 +184,9 @@ void ec_loop(unsigned long loopmillis) {
         ec_calib_array_pos++;
         ec_calib_array_pos%=EC_CALIB_ARRAY_SIZE;
 
-        
+        if (isValueArrayOK(ec_calib_array,EC_CALIB_ARRAY_SIZE,EC_ADC_UNAVAILABLE)){
+          ec_calib_adc=getMean(ec_calib_array,EC_CALIB_ARRAY_SIZE);
+        }
       }
     }
   }
@@ -201,7 +239,12 @@ float ec_getECfromADC(float adc) {
 
     _ec=mapf(adc,x0,x1,y0,y1); //linear approximation    
   }
-  return _ec;
+  if (_ec>=0) {
+    return _ec;
+  }else{
+    return 0;
+  }
+  
 }
 
 float ec_calculateEC25(float pEC,float pTemp) 

include/flow.h

@@ -1,6 +1,8 @@
 #ifndef _FLOW_H_
 #define _FLOW_H_
 
+mqttValueTiming timing_flow;
+
 #define FLOW_PIN 19
 uint16_t flow_counter=0; //maximum counts/s measured with Eden 128 Pump was 171
 void IRAM_ATTR isr_flow();
@@ -15,8 +17,13 @@ uint32_t flow_counter_sum=0;
 
 void flow_setup() {
 
-    pinMode(FLOW_PIN, INPUT_PULLUP);
-    attachInterrupt(FLOW_PIN, isr_flow, CHANGE);
+  timing_flow.minchange=0.0;
+  timing_flow.maxchange=0.5;
+  timing_flow.mintime=1*000;
+  timing_flow.maxtime=30*60*1000;
+
+  pinMode(FLOW_PIN, INPUT_PULLUP);
+  attachInterrupt(FLOW_PIN, isr_flow, CHANGE);
 }
 
 void flow_loop(unsigned long loopmillis) {

include/soilmoisture.h

@@ -7,11 +7,14 @@
 
 #define READINTERVAL_SM 100
 
+mqttValueTiming timing_soilmoisture_sm1;
+mqttValueTiming timing_soilmoisture_sm2;
+mqttValueTiming timing_soilmoisture_sm3;
 
 //Calibration values
 //high=adc value for sensor in air
 //low=adc value for sensor in NaCl solution at 25°C with 12880 uS/cm
-float sm1_low=45555;
+float sm1_low=4555;
 float sm1_high=11799;
 float sm2_low=3235;
 float sm2_high=16050;
@@ -43,6 +46,23 @@ uint8_t sm_readchannel=0;
 
 
 void sm_setup() {
+
+  timing_soilmoisture_sm1.minchange=0.0;
+  timing_soilmoisture_sm1.maxchange=0.05;
+  timing_soilmoisture_sm1.mintime=2*1000;
+  timing_soilmoisture_sm1.maxtime=60*60*1000;
+
+  timing_soilmoisture_sm2.minchange=0.0;
+  timing_soilmoisture_sm2.maxchange=0.05;
+  timing_soilmoisture_sm2.mintime=2*1000;
+  timing_soilmoisture_sm2.maxtime=60*60*1000;
+
+  timing_soilmoisture_sm3.minchange=0.0;
+  timing_soilmoisture_sm3.maxchange=0.05;
+  timing_soilmoisture_sm3.mintime=2*1000;
+  timing_soilmoisture_sm3.maxtime=60*60*1000;
+
+
   for (uint16_t i=0;i<SM_SIZE;i++) {
     sm_mean1array[i]=SM_DISCONNECTED;
     sm_mean2array[i]=SM_DISCONNECTED;

include/temperature.h

@@ -2,6 +2,9 @@
 #define _TEMPERATURE_H_
 
 
+mqttValueTiming timing_temperature_reservoir;
+mqttValueTiming timing_temperature_air;
+
 #include <OneWire.h>
 #include <DallasTemperature.h>
 
@@ -38,6 +41,15 @@ float tempCmean_air=DEVICE_DISCONNECTED_C;
 
 
 void temperature_setup() {
+  timing_temperature_reservoir.minchange=0.0;
+  timing_temperature_reservoir.maxchange=0.5;
+  timing_temperature_reservoir.mintime=2*1000;
+  timing_temperature_reservoir.maxtime=60*60*1000;
+
+  timing_temperature_air.minchange=0.0;
+  timing_temperature_air.maxchange=0.5;
+  timing_temperature_air.mintime=2*1000;
+  timing_temperature_air.maxtime=60*60*1000;
 
   //initialize mean array
   for (uint16_t i=0;i<TEMPMEAN_SIZE;i++) {

include/wifi_functions.h

@@ -0,0 +1,121 @@
+
+#ifndef _WIFI_FUNCTIONS_H_
+#define _WIFI_FUNCTIONS_H_
+
+#include <WiFi.h>
+#include <MQTT.h>
+
+#include "wifi_settings.h"
+#include "helpfunctions.h"
+
+
+struct mqttValueTiming{
+    float minchange;
+    float maxchange;
+    unsigned long mintime;
+    unsigned long maxtime;
+    float lastvalue=0;
+    unsigned long lasttime=0;
+};
+
+
+WiFiClient net;
+MQTTClient client;
+
+bool sendallnext_flag=false;
+bool enableTiming=true;
+
+bool force_ec_measurement=false;
+
+void publishValueTimed(String topic,float value,uint8_t decimals,mqttValueTiming &mqttvt,unsigned long loopmillis);
+void publishValue(String topic,float value,uint8_t decimals);
+
+void connect() {
+  Serial.print("checking wifi...");
+  uint8_t timeout=0;
+  while (WiFi.status() != WL_CONNECTED && timeout<10) {
+    timeout++;
+    Serial.print(".");
+    delay(500);
+  }
+  
+  if (WiFi.status() == WL_CONNECTED)
+  {
+    Serial.print("\nconnecting...");
+    timeout=0;
+    #define CONNECT_TIMEOUT 5
+    while (!client.connect(client_id, "public", "public") && timeout<CONNECT_TIMEOUT) {
+        Serial.print(".");
+        delay(1000);
+        timeout++;
+    }
+    if (timeout>=CONNECT_TIMEOUT) {
+        Serial.println("\nconnection failed!");
+    }else{
+        Serial.println("\nconnected!");
+        client.subscribe((String)client_id+"/sendall");
+        client.subscribe((String)client_id+"/ec/trigger");
+        // client.unsubscribe("/hello");
+    }
+
+
+  }
+}
+
+
+void messageReceived(String &topic, String &payload) {
+  Serial.println("incoming: " + topic + " - " + payload);
+  if (topic==((String)client_id+"/sendall") && payload=="true") {
+    sendallnext_flag=true;
+    Serial.println("Send all values next time");
+  }
+  if (topic==((String)client_id+"/ec/trigger") && payload=="true") {
+    force_ec_measurement=true; 
+    Serial.println("Forced EC Measurement");
+  }
+}
+
+bool mqtt_loop(unsigned long loopmillis) {
+    static unsigned long last_client_loop=0;
+    #define CLIENT_LOOP_INTERVAL 10 //ms. fixes some wifi issues. from example https://registry.platformio.org/libraries/256dpi/MQTT/examples/ESP32DevelopmentBoard/ESP32DevelopmentBoard.ino
+    if (loopmillis >= last_client_loop+CLIENT_LOOP_INTERVAL) {
+        last_client_loop=loopmillis;
+        client.loop();
+
+        static unsigned long last_connection_try=0;
+        #define RETRY_CONNECTION 60000
+        if (!client.connected()) {
+            if (loopmillis-last_connection_try>RETRY_CONNECTION) {
+                connect();
+            }
+        }else{
+            return true;
+        }
+    }
+    return false;
+}
+
+void publishValueTimed(String topic,float value,uint8_t decimals,mqttValueTiming &mqttvt,unsigned long loopmillis) {
+    unsigned long timediff=loopmillis-mqttvt.lasttime;
+    float valuediff=abs(value-mqttvt.lastvalue);
+    valuediff=constrain(valuediff,mqttvt.minchange,mqttvt.maxchange);
+    unsigned long sendafter=mapf(valuediff,mqttvt.minchange,mqttvt.maxchange,mqttvt.maxtime,mqttvt.mintime); //map valuediff to time when to send
+    /*Serial.println();
+    Serial.print("timediff="); Serial.print(timediff);
+    Serial.print(" valuediff="); Serial.print(valuediff);
+    Serial.print(" sendafter="); Serial.println(sendafter);*/
+    if (timediff>=sendafter || !enableTiming) {
+        mqttvt.lasttime=loopmillis;
+        mqttvt.lastvalue=value;
+        publishValue(topic,value,decimals);
+    }
+}
+
+void publishValue(String topic,float value,uint8_t decimals) {
+    char buffer[16];
+    dtostrf(value,1,decimals,buffer);
+    client.publish((String)client_id+"/"+topic, buffer);
+    Serial.print("Publish Topic="); Serial.print((String)client_id+"/"+topic); Serial.print(" Message="); Serial.println(buffer);
+}
+
+#endif

include/wifi_settings.h

@@ -0,0 +1,6 @@
+
+const char ssid[] = "";
+const char pass[] = "";
+const char mqtt_host[] = "10.0.0.1";
+
+const char client_id[] = "hydroponic";

platformio.ini

@@ -20,3 +20,4 @@ lib_deps =
     d03n3rfr1tz3/HC-SR04@^1.1.2
     https://github.com/emilv/ArduinoSort/
     robtillaart/ADS1X15@^0.3.9
+    256dpi/MQTT@^2.5.1

src/main.cpp

@@ -1,7 +1,13 @@
 #include <Arduino.h>
 
+
+
+#include "wifi_functions.h"
+
 bool debug=true; //print Serial information
 
+
+
 #include "helpfunctions.h"
 #include "ADS1X15.h"
 
@@ -11,13 +17,16 @@ ADS1115 ADS(0x48);
 // ######## Temperature
 #include "temperature.h"
 
+
+
+
 // ######## EC
 #include "ec.h"
 
 
 
 // ######## Water Level
-#include "waterlevel.h"
+//#include "waterlevel.h"
 
 
 // ######## Flow Rate
@@ -27,6 +36,8 @@ ADS1115 ADS(0x48);
 #include "soilmoisture.h"
 
 
+
+
 unsigned long last_check=0;
 
 
@@ -44,6 +55,10 @@ void setup() {
   pinMode(PIN_LED,OUTPUT);
   digitalWrite(PIN_LED,LOW);
   Serial.begin(115200);
+  WiFi.begin(ssid, pass);
+  client.begin(mqtt_host, net);
+  client.onMessage(messageReceived);
+  connect();
 
   //init ADS1115
   if (!ADS.begin()) {
@@ -55,8 +70,8 @@ void setup() {
   Serial.println("Setup EC");
   ec_setup();
 
-  Serial.println("Setup Waterlevel");
-  waterlevel_setup();
+  //Serial.println("Setup Waterlevel");
+  //waterlevel_setup();
 
   Serial.println("Setup Temperature");
   temperature_setup();
@@ -64,6 +79,9 @@ void setup() {
   Serial.println("Setup Flow");
   flow_setup();
 
+  Serial.println("Setup Soilmoisture");
+  sm_setup();
+
   Serial.println("Finished Setup");
   delay(200);
 
@@ -82,18 +100,20 @@ void setup() {
 
 
   //Serial.println("time,tempReservoir,ECadcCalib,ECadc,ECadcAdjusted,EC,EC25");
-  Serial.println("time,tempReservoir,ECadcCalib,ECadc,ECadcAdjusted");
+  //Serial.println("time,tempReservoir,ECadcCalib,ECadc,ECadcAdjusted");
 
 }
 
 void loop() {
   unsigned long loopmillis=millis();
+  enableTiming=true; //reactivate
+  
   
   ec_loop(loopmillis);
 
   temperature_loop(loopmillis);
   
-  waterlevel_loop(loopmillis);
+  //waterlevel_loop(loopmillis);
 
   flow_loop(loopmillis);
 
@@ -103,18 +123,13 @@ void loop() {
   static bool getReading=false;
 
   if (!digitalRead(PIN_BUTTON)) {
-    last_check=loopmillis;
     valueError=false;
-    //Serial.println("Reset ValueError flag by user");
-    //digitalWrite(PIN_LED,valueError);
+    Serial.println("Reset ValueError flag by user");
+    digitalWrite(PIN_LED,valueError);
     delay(100);
-    last_measurement_ec=0; //force reading
-    getReading=true;
-    ec_flag_measurement_available=false;
-    digitalWrite(PIN_LED,HIGH);
-
   }
 
+  /*
   if (ec_flag_measurement_available && getReading) {
     ec_flag_measurement_available=false;
     getReading=false;
@@ -125,7 +140,7 @@ void loop() {
     Serial.print(ec_adc); Serial.print(",");
     Serial.print(ec_adc_adjusted);
     Serial.println();
-  }
+  }*/
   
   if (loopmillis>last_check+2000) {  //check values
   
@@ -137,9 +152,9 @@ void loop() {
     if (tempCmean_air==DEVICE_DISCONNECTED_C || tempCmean_reservoir==DEVICE_DISCONNECTED_C) {
       valueError=true;
     }
-    if (waterlevel==WATERLEVEL_UNAVAILABLE) {
-      valueError=true;
-    }
+    //if (waterlevel==WATERLEVEL_UNAVAILABLE) {
+    //  valueError=true;
+    //}
     if (sm_mean1==SM_DISCONNECTED || sm_mean2==SM_DISCONNECTED) {
       valueError=true;
     }
@@ -201,12 +216,57 @@ void loop() {
       Serial.print(ec25); 
       Serial.println();
 
-      Serial.print("Waterlevel distance,volume = "); Serial.print(waterlevel); Serial.print(","); Serial.print(watervolume);
+      /*Serial.print("Waterlevel distance,volume = "); Serial.print(waterlevel); Serial.print(","); Serial.print(watervolume);
       if (waterlevel_failcounter>0) {
         Serial.print(" fails="); Serial.print(waterlevel_failcounter);
-      }
+      }*/
       Serial.println();    
 
+
+
+      
+
+    }
+
+
+    if (mqtt_loop(loopmillis)) {
+      if (sendallnext_flag) {
+        sendallnext_flag=false;
+        enableTiming=false;
+      }
+
+      if (tempCmean_reservoir!=DEVICE_DISCONNECTED_C) {
+        publishValueTimed("temperature/reservoir",tempCmean_reservoir,2,timing_temperature_reservoir,loopmillis); 
+      }
+      if (tempCmean_air!=DEVICE_DISCONNECTED_C) {
+        publishValueTimed("temperature/air",tempCmean_air,2,timing_temperature_air,loopmillis); 
+      }
+      if (sm_mean1!=SM_DISCONNECTED) {
+        publishValueTimed("soilmoisture/sm1",sm_mean1,3,timing_soilmoisture_sm1,loopmillis); 
+      }
+      if (sm_mean2!=SM_DISCONNECTED) {
+        publishValueTimed("soilmoisture/sm2",sm_mean2,3,timing_soilmoisture_sm2,loopmillis); 
+      }
+      if (sm_mean3!=SM_DISCONNECTED) {
+        publishValueTimed("soilmoisture/sm3",sm_mean3,3,timing_soilmoisture_sm3,loopmillis); 
+      }
+      
+      publishValueTimed("flow/flow",flow,2,timing_flow,loopmillis); 
+      
+
+      if (ec_adc!=0) {
+        publishValueTimed("ec/adc",ec_adc,0,timing_ec_adc,loopmillis); 
+      }
+      if (ec_calib_adc!=0) {
+        publishValueTimed("ec/eccalibadc",ec_calib_adc,0,timing_ec_calibadc,loopmillis); 
+      }
+      if (ec_adc_adjusted!=0) {
+        publishValueTimed("ec/adcadjusted",ec_adc_adjusted,0,timing_ec_adcadjusted,loopmillis); 
+      }
+      if (ec!=EC_UNAVAILABLE){
+        publishValueTimed("ec/ec",ec,0,timing_ec_ec,loopmillis); 
+        publishValueTimed("ec/sc",ec25,0,timing_ec_sc,loopmillis); 
+      }
     }
     
 
@@ -214,64 +274,8 @@ void loop() {
 
     
 
-    /*
-    Serial.print(millis()/1000.0,2); Serial.print(",");
-    
-    Serial.print(getMeanf(tempCmean_reservoir_array,TEMPMEAN_SIZE)); Serial.print(",");
-    //Serial.print(getMean(sm_mean,SM_SIZE)); Serial.print(",");
-    
-    Serial.print(ec_calib_adc); Serial.print(",");
-    Serial.print(ec_adc); Serial.print(",");
-    Serial.print(ec_adc_adjusted); Serial.print(",");
-    Serial.print(ec); Serial.print(",");
-    Serial.print(ec25);
-    Serial.print(getMean(sm_mean,SM_SIZE));
-    
-
-    
-    
-    Serial.println();
-
-    
-
-    if (tempCmean_reservoir!=DEVICE_DISCONNECTED_C){
-      Serial.print("\t Treservoir="); Serial.print(tempCmean_reservoir); Serial.print("\t Tair="); Serial.print(tempCmean_air);
-    }else{
-      Serial.print("\t waiting for temperature");
-    }
-
-    
-    if (isValueArrayOKf(waterlevelMean,WATERLEVELMEAN_SIZE,-1.0)){
-      float _max=getMaxf(waterlevelMean,WATERLEVELMEAN_SIZE);
-      float _min=getMinf(waterlevelMean,WATERLEVELMEAN_SIZE);
-
-      float _filteredWaterlevel=getFilteredf(waterlevelMean,WATERLEVELMEAN_SIZE,8);
-      float _meanWaterlevel=getMeanf(waterlevelMean,WATERLEVELMEAN_SIZE);
-      
-
-      Serial.print("\t Dist="); Serial.print(_filteredWaterlevel); Serial.print("mm"); Serial.print("(+- "); Serial.print((_max-_min)/2.0); Serial.print(")"); Serial.print(" [mean="); Serial.print(_meanWaterlevel); Serial.print("]");
-      
-    }else{
-      Serial.print("\t waiting for distance");
-    }
-
-    Serial.print("\t Flow="); Serial.print(flow,2);
-    Serial.print("\t Flowsum="); Serial.print(flow_counter_sum);
-    
-    
-
-    Serial.println();   
-    */
-    
       
     
   }
   
 }
-
-
-
-/*
-TODO:
-- recalibrate EC. Note min max (shorted, open) values. Test screw terminal connection. Recalibrate EC Probe.
-*/