here is some code written for the esp32 that does the subscribe and publish thing:
#include <ESP32Time.h>
#include <WiFi.h>
#include <PubSubClient.h>
#include "certs.h" // include the connection infor for WiFi and MQTT
#include "sdkconfig.h" // used for log printing
#include "esp_system.h"
#include "freertos/FreeRTOS.h" //freeRTOS items to be used
#include "freertos/task.h"
#include <driver/adc.h>
#include <SimpleKalmanFilter.h>
////
WiFiClient wifiClient; // do the WiFi instantiation thing
PubSubClient MQTTclient( mqtt_server, mqtt_port, wifiClient ); //do the MQTT instantiation thing
ESP32Time rtc;
////
#define evtDoParticleRead ( 1 << 0 ) // declare an event
#define evtADCreading ( 1 << 3 )
EventGroupHandle_t eg; // variable for the event group handle
////
SemaphoreHandle_t sema_MQTT_KeepAlive;
SemaphoreHandle_t sema_mqttOK;
////
QueueHandle_t xQ_RemainingMoistureMQTT;
QueueHandle_t xQ_RM;
QueueHandle_t xQ_Message;
////
struct stu_message
{
char payload [150] = {'\0'};
String topic;
} x_message;
////
int mqttOK = 0;
bool TimeSet = false;
bool manualPumpOn = false;
////
// interrupt service routine for WiFi events put into IRAM
void IRAM_ATTR WiFiEvent(WiFiEvent_t event)
{
switch (event) {
case SYSTEM_EVENT_STA_CONNECTED:
break;
case SYSTEM_EVENT_STA_DISCONNECTED:
log_i("Disconnected from WiFi access point");
break;
case SYSTEM_EVENT_AP_STADISCONNECTED:
log_i("WiFi client disconnected");
break;
default: break;
}
} // void IRAM_ATTR WiFiEvent(WiFiEvent_t event)
////
void IRAM_ATTR mqttCallback(char* topic, byte * payload, unsigned int length)
{
// clear locations
memset( x_message.payload, '\0', 150 );
x_message.topic = ""; //clear string buffer
x_message.topic = topic;
int i = 0;
for ( i; i < length; i++)
{
x_message.payload[i] = ((char)payload[i]);
}
// x_message.payload[i] = '\0';
xQueueOverwrite( xQ_Message, (void *) &x_message );// send data
} // void mqttCallback(char* topic, byte* payload, unsigned int length)
////
void setup()
{
x_message.topic.reserve(150);
//
xQ_Message = xQueueCreate( 1, sizeof(stu_message) );
xQ_RemainingMoistureMQTT = xQueueCreate( 1, sizeof(float) ); // sends a queue copy
xQ_RM = xQueueCreate( 1, sizeof(float) );
//
eg = xEventGroupCreate(); // get an event group handle
//
sema_mqttOK = xSemaphoreCreateBinary();
xSemaphoreGive( sema_mqttOK );
//
gpio_config_t io_cfg = {}; // initialize the gpio configuration structure
io_cfg.mode = GPIO_MODE_INPUT; // set gpio mode. GPIO_NUM_0 input from water level sensor
io_cfg.pull_down_en = GPIO_PULLDOWN_ENABLE; // enable pull down
io_cfg.pin_bit_mask = ( (1ULL << GPIO_NUM_0)); //bit mask of the pins to set, assign gpio number to be configured
gpio_config(&io_cfg); // configure the gpio based upon the parameters as set in the configuration structure
//
io_cfg = {}; //set configuration structure back to default values
io_cfg.mode = GPIO_MODE_OUTPUT;
io_cfg.pin_bit_mask = ( 1ULL << GPIO_NUM_4 ); //bit mask of the pins to set, assign gpio number to be configured
gpio_config(&io_cfg);
gpio_set_level( GPIO_NUM_4, LOW); // deenergize relay module
// set up A:D channels https://dl.espressif.com/doc/esp-idf/latest/api-reference/peripherals/adc.html
adc1_config_width(ADC_WIDTH_12Bit);
adc1_config_channel_atten(ADC1_CHANNEL_3, ADC_ATTEN_DB_11);// using GPIO 39
//
xTaskCreatePinnedToCore( MQTTkeepalive, "MQTTkeepalive", 10000, NULL, 6, NULL, 1 );
xTaskCreatePinnedToCore( fparseMQTT, "fparseMQTT", 10000, NULL, 5, NULL, 1 ); // assign all to core 1, WiFi in use.
xTaskCreatePinnedToCore( fPublish, "fPublish", 9000, NULL, 3, NULL, 1 );
xTaskCreatePinnedToCore( fReadAD, "fReadAD", 9000, NULL, 3, NULL, 1 );
xTaskCreatePinnedToCore( fDoMoistureDetector, "fDoMoistureDetector", 70000, NULL, 4, NULL, 1 );
xTaskCreatePinnedToCore( fmqttWatchDog, "fmqttWatchDog", 3000, NULL, 2, NULL, 1 );
} //void setup()
////
void fReadAD( void * parameter )
{
float ADbits = 4095.0f;
float uPvolts = 3.3f;
float adcValue_b = 0.0f; //plant in yellow pot
uint64_t TimePastKalman = esp_timer_get_time(); // used by the Kalman filter UpdateProcessNoise, time since last kalman calculation
float WetValue = 1.07f; // value found by putting sensor in water
float DryValue = 2.732f; // value of probe when held in air
float Range = DryValue - WetValue;
float RemainingMoisture = 100.0f;
SimpleKalmanFilter KF_ADC_b( 1.0f, 1.0f, .01f );
for (;;)
{
xEventGroupWaitBits (eg, evtADCreading, pdTRUE, pdTRUE, portMAX_DELAY ); //
adcValue_b = float( adc1_get_raw(ADC1_CHANNEL_3) ); //take a raw ADC reading
adcValue_b = ( adcValue_b * uPvolts ) / ADbits; //calculate voltage
KF_ADC_b.setProcessNoise( (esp_timer_get_time() - TimePastKalman) / 1000000.0f ); //get time, in microsecods, since last readings
adcValue_b = KF_ADC_b.updateEstimate( adcValue_b ); // apply simple Kalman filter
TimePastKalman = esp_timer_get_time(); // time of update complete
RemainingMoisture = 100.0f * (1 - ((adcValue_b - WetValue) / (DryValue - WetValue))); //remaining moisture = 1-(xTarget - xMin) / (xMax - xMin) as a percentage of the sensor wet dry volatges
xQueueOverwrite( xQ_RM, (void *) &RemainingMoisture );
log_i( "adcValue_b = %f remaining moisture %f%", adcValue_b, RemainingMoisture );
}
vTaskDelete( NULL );
}
////
void fPublish( void * parameter )
{
float RemainingMoisture = 100.0f;
for (;;)
{
if ( xQueueReceive(xQ_RemainingMoistureMQTT, &RemainingMoisture, portMAX_DELAY) == pdTRUE )
{
xSemaphoreTake( sema_MQTT_KeepAlive, portMAX_DELAY ); // whiles MQTTlient.loop() is running no other mqtt operations should be in process
MQTTclient.publish( topicRemainingMoisture_0, String(RemainingMoisture).c_str() );
xSemaphoreGive( sema_MQTT_KeepAlive );
}
} // for (;;)
vTaskDelete( NULL );
} //void fPublish( void * parameter )
////
void WaterPump0_off()
{
gpio_set_level( GPIO_NUM_4, LOW); //denergize relay module
}
////
void WaterPump0_on()
{
gpio_set_level( GPIO_NUM_4, HIGH); //energize relay module
}
////
void fmqttWatchDog( void * paramater )
{
int UpdateImeTrigger = 86400; //seconds in a day
int UpdateTimeInterval = 85000; // get another reading when = UpdateTimeTrigger
int maxNonMQTTresponse = 12;
TickType_t xLastWakeTime = xTaskGetTickCount();
const TickType_t xFrequency = 5000; //delay for mS
for (;;)
{
xLastWakeTime = xTaskGetTickCount();
vTaskDelayUntil( &xLastWakeTime, xFrequency );
xSemaphoreTake( sema_mqttOK, portMAX_DELAY ); // update mqttOK
mqttOK++;
xSemaphoreGive( sema_mqttOK );
if ( mqttOK >= maxNonMQTTresponse )
{
ESP.restart();
}
UpdateTimeInterval++; // trigger new time get
if ( UpdateTimeInterval >= UpdateImeTrigger )
{
TimeSet = false; // sets doneTime to false to get an updated time after a days count of seconds
UpdateTimeInterval = 0;
}
}
vTaskDelete( NULL );
} //void fmqttWatchDog( void * paramater )
////
void fDoMoistureDetector( void * parameter )
{
//wait for a mqtt connection
while ( !MQTTclient.connected() )
{
