I have a Pi-3 running as the MQTT broker on my home network, this part is working.
I'm trying to setup some ESP-01 modules to relay simple data on a closed network. All the online examples I've found use a cloud type hosting service.
Are there any local host examples I can learn/leverage?
I have a RPi as my MQTT Broker and 13 ESP32's as MQTT clients, what issues are you having?
this is what I have open at the moment.
/*
https://github.com/G6EJD/ESP32-e-Paper-Weather-Display/blob/master/examples/Waveshare_4_2/Waveshare_4_2.ino
Stole some code from that guy.
*/
#include "MyBitmap.h"
#include <WiFi.h>
#include <PubSubClient.h>
#include "certs.h"
#include "sdkconfig.h"
#include "esp_system.h"
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include <SPI.h>
#include <Adafruit_Sensor.h>
#include "Adafruit_BME680.h"
#include <GxEPD2_BW.h>
#include <U8g2_for_Adafruit_GFX.h> // Select u8g2 font from here: https://github.com/olikraus/u8g2/wiki/fntlistall
#include <Fonts/FreeMonoBold9pt7b.h> //https://learn.adafruit.com/adafruit-gfx-graphics-library/using-fonts
#include <Fonts/FreeMono9pt7b.h>
#include <HardwareSerial.h>
#include <SimpleKalmanFilter.h>
#include "MHZ19.h"
#include <ESP32Time.h>
#include <SolarCalculator.h>
#include <driver/adc.h>
////
ESP32Time rtc;
MHZ19 myMHZ19;
GxEPD2_BW<GxEPD2_420, GxEPD2_420::HEIGHT> display(GxEPD2_420(/*CS=5*/ SS, /*DC=*/ 17, /*RST=*/ 16, /*BUSY=*/ 4)); // GDEW042T2
Adafruit_BME680 bme( GPIO_NUM_15 );
WiFiClient wifiClient; // do the WiFi instantiation thing
PubSubClient MQTTclient( mqtt_server, mqtt_port, wifiClient );
U8G2_FOR_ADAFRUIT_GFX u8g2Fonts;
////
#define evtStoreAirPressure ( 1 << 0 )
#define evtWaitForBME ( 1 << 1 )
#define evtParseMQTT ( 1 << 3 )
#define evtDisplayUpdate ( 1 << 4 )
#define evtDoBME ( 1 << 5 )
#define evtDewPoint ( 1 << 6 )
#define OneMinuteGroup ( evtDoBME )
EventGroupHandle_t eg;
//////
QueueHandle_t xQ_WindChillDewPoint;
QueueHandle_t xQ_eData;
struct stu_eData
{
float oTemperature = 0.0f;
float oHumidity = 0.0f;
float oPressure = 0.0f;
float Temperature = 0.0f;
float Pressure = 0.0f;
float Humidity = 0.0f;
float IAQ = 0.0f; // Index Air Quality
float RM0 = 0.0f; // Remaining Moisture from sensor 0
//float PM2 = 0.0f; // particles in air
float WS = 0.0f; // wind speed
String WD = ""; // wind direction
float RF = 0.0f; // rainfall
float WSV = 0.0f; // weather station volts
//float WSC = 0.0f; // weather station current
//float WSP = 0.0f; // weather station power
float WindChill = 0.0f; //windchill
float DewPoint = 0.0f; //dew point or dew index
int SunRiseHr = 0; // sunrise hour
int SunRiseMin = 0; //sunrise minute
int SunSetHr = 0; //sunset hour
int SunSetMin = 0; //sunset minute
int DuskHr = 0; //dusk
int DuskMin = 0; //dusk
int DawnHr = 0; // dawn
int DawnMin = 0; // dawn
int TransitHr = 0; // 'noon' time
int TransitMin = 0; // 'noon' time
double azimuth = 0.0f; // Sun's azimuth, in degrees
double elevation = 0.0f; // Sun's elevation, in degrees
float CO2 = 0.0f;
float PressureH = 0.0f;
float PressureL = 10000.0f;
int cngPress = 0; // pressure change 0= no change, -1 slow change +1 fast change
} x_eData; // environmental data
QueueHandle_t xQ_Message; // payload and topic queue of MQTT payload and topic
const int payloadSize = 100;
struct stu_message
{
char payload [payloadSize] = {'\0'};
String topic ;
} x_message;
////
const float oGasResistanceBaseLine = 149598.0f;
int mqttOK = 0;
volatile bool TimeSet = false;
const int BufferCount = 60;
float CollectionPressure[BufferCount] = {0.0f};
boolean LargeIcon = true, SmallIcon = false;
//bool CollectionDone = false;
#define Large 11 // For icon drawing, needs to be odd number for best effect
#define Small 5 // For icon drawing, needs to be odd number for best effect
enum alignment {LEFT, RIGHT, CENTER};
SemaphoreHandle_t sema_MQTT_KeepAlive;
SemaphoreHandle_t sema_PublishPM;
SemaphoreHandle_t sema_mqttOK;
SemaphoreHandle_t sema_CollectPressure;
SemaphoreHandle_t sema_eData;
////
//serial(2) = pin25 RX, pin26 TX
HardwareSerial co2Serial ( 2 );
////
void IRAM_ATTR onTimer()
{
BaseType_t xHigherPriorityTaskWoken;
xEventGroupSetBitsFromISR(eg, OneMinuteGroup, &xHigherPriorityTaskWoken);
} // void IRAM_ATTR onTimer()
// interrupt service routine for WiFi events put into IRAM
void IRAM_ATTR WiFiEvent(WiFiEvent_t event)
{
switch (event) {
break;
default: break;
}
} // void IRAM_ATTR WiFiEvent(WiFiEvent_t event)
//
void IRAM_ATTR mqttCallback(char* topic, byte * payload, unsigned int length)
{
memset( x_message.payload, '\0', payloadSize ); // clear payload char buffer
x_message.topic = ""; //clear topic string buffer
x_message.topic = topic; //store new topic
memcpy( x_message.payload, payload, length );
xQueueOverwrite( xQ_Message, (void *) &x_message );// send data to queue
} // void mqttCallback(char* topic, byte* payload, unsigned int length)
////
void setup()
{
gpio_config_t io_cfg = {}; // initialize the gpio configuration structure
io_cfg.mode = GPIO_MODE_OUTPUT;
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);
gpio_set_level( GPIO_NUM_0, LOW); //
//
adc1_config_width(ADC_WIDTH_12Bit);
adc1_config_channel_atten(ADC1_CHANNEL_6, ADC_ATTEN_DB_11);// using GPIO 34
// hardware timer 4 set for one minute alarm
hw_timer_t * timer = NULL;
timer = timerBegin( 3, 80, true );
timerAttachInterrupt( timer, &onTimer, true );
timerAlarmWrite(timer, 60000000, true);
timerAlarmEnable(timer);
///
co2Serial.begin( 9600 , SERIAL_8N1, 25, 26 ); // pin25 RX, pin26 TX
//
x_eData.WD.reserve(50);
x_message.topic.reserve( payloadSize );
//
xQ_WindChillDewPoint = xQueueCreate( 1, sizeof(stu_eData) );
xQ_Message = xQueueCreate( 1, sizeof(stu_message) );
xQ_eData = xQueueCreate( 1, sizeof(stu_eData) ); // sends a queue copy of the structure
//
sema_PublishPM = xSemaphoreCreateBinary();
xSemaphoreGive( sema_PublishPM );
sema_mqttOK = xSemaphoreCreateBinary();
xSemaphoreGive( sema_mqttOK );
sema_CollectPressure = xSemaphoreCreateBinary();
xSemaphoreGive( sema_CollectPressure );
sema_eData = xSemaphoreCreateBinary();
xSemaphoreGive ( sema_eData );
//
eg = xEventGroupCreate(); // get an event group handle
//
xTaskCreatePinnedToCore( fparseMQTT, "fparseMQTT", 7000, NULL, 5, NULL, 1 );
xTaskCreatePinnedToCore( MQTTkeepalive, "MQTTkeepalive", 5000, NULL, 6, NULL, 1 );
xTaskCreatePinnedToCore( DoTheBME680Thing, "DoTheBME280Thing", 20000, NULL, 5, NULL, 1);
xTaskCreatePinnedToCore( fmqttWatchDog, "fmqttWatchDog", 5000, NULL, 3, NULL, 1 );
xTaskCreatePinnedToCore( fDoTheDisplayThing, "fDoTheDisplayThing", 30000, NULL, 3, NULL, 1 );
xTaskCreatePinnedToCore( fGetCO2, "fGetCO2", 4500, NULL, 2, NULL, 1 );
xTaskCreatePinnedToCore( fParseDewPointWindChill, "fParseDewPointWindChill", 4500, NULL, 2, NULL, 1 );
xTaskCreatePinnedToCore( fSolarCalculations, "fSolarCalculations", 10000, NULL, 2, NULL, 1 );
xTaskCreatePinnedToCore( fProcessAirPressure, "fProcessAirPressure", 5000, NULL, 2, NULL, 1 );
xTaskCreatePinnedToCore( fFindDewPointWithHumidity, "fFindDewPointWithHumidity", 5000, NULL, 2, NULL, 1 );
xTaskCreatePinnedToCore( fLowSideSwitchTest, "fLowSideSwitchTest", 2000, NULL, 5, NULL, 1 );
} //void setup()
////
void fLowSideSwitchTest( void *pvParameters )
{
float adcValue = 0.0f;
for(;;)
{
gpio_set_level( GPIO_NUM_0, LOW); //
vTaskDelay( 1000 );
gpio_set_level( GPIO_NUM_0, HIGH); //
adcValue = float( adc1_get_raw(ADC1_CHANNEL_6) ); //
log_i( " low side measurement %f, adcValue );
//vTaskDelay( 10 );
}
vTaskDelete( NULL );
} //void fLowSideSwitchTest
////
void fFindDewPointWithHumidity( void *pvParameters )
{
float temperature = 0.0f;
for ( ;; )
{
xEventGroupWaitBits (eg, evtDewPoint, pdTRUE, pdTRUE, portMAX_DELAY );
temperature = (x_eData.oTemperature - 32) / 1.8f; // Celsius (°C) = (Fahrenheit - 32) / 1.8 convert to C
xSemaphoreTake( sema_eData, portMAX_DELAY );
x_eData.DewPoint = log(x_eData.oHumidity / 100) + (17.62 * temperature) / (243.12 + temperature);
x_eData.DewPoint = 243.12 * x_eData.DewPoint / (17.62 - x_eData.DewPoint);
x_eData.DewPoint = (x_eData.DewPoint * 1.8f) + 32.0f; // convert back to F
xSemaphoreGive ( sema_eData );
}
vTaskDelete( NULL );
}
////
String PressureRateOfChange ()
{
String s = "";
s.reserve( 10 );
xSemaphoreTake ( sema_eData, portMAX_DELAY );
switch ( x_eData.cngPress )
{
case 0:
s = "Slow";
break;
case 1:
s = "Unsettled";
break;
case 2:
s = "Fast";
break;
case 3:
s = "Steady";
break;
}
xSemaphoreGive ( sema_eData );
return s;
}
/*
Approaching storms and wind cause barometric pressure to decrease.
Rising pressure indicates fair weather.
The longer it takes barometric pressure to change,
the longer the coming weather pattern can be expected to last.
It is possible that a small weather event, such as a passing shower, may trigger no change in barometric pressure.
slow rate of change .0762mmHg - 1.016mmHg < 3 hours
fast rate of change 4.572mmHg in < 3 hours
steady <=.0762Hg in <= 3 hours
*/
int CalculatePressureFactors( float pastPressure, float currentPressure )
{
float rateChange = abs(pastPressure - currentPressure);
//log_i ( " pressure rate of change %f", rateChange );
if ( (rateChange >= .0762f) && (rateChange <= 1.016f ) )
{
return 0; //slow
}
if ( (rateChange > 1.016f) && (rateChange < 4.572f) )
{
return 1;
}
if ( rateChange >= 4.572f )
{
return 2;//fast
}
if ( rateChange <= .0762 )
{
return 3;//steady
}
} //void CalculatePressureFactors()
////
void fProcessAirPressure ( void *pvParemeters )
{
int Ticks = 118; // Tick counter
bool Filled = false; // array has been filled before?
float *ptr = CollectionPressure; // pointer to the array
const int ticksTrigger = 120; // triggered at 1 minute intervals
for (;;)
{
//triggered by BME which is triggered by the 1 minute hardware timer.
xEventGroupWaitBits (eg, evtStoreAirPressure, pdTRUE, pdTRUE, portMAX_DELAY );
xSemaphoreTake( sema_CollectPressure, portMAX_DELAY );
xSemaphoreTake ( sema_eData, portMAX_DELAY );
if ( !Filled )
{
//if array has not been filled before, fill array with the same base value
for ( int j = 0; j < BufferCount; j++ )
{
*( ptr + j ) = x_eData.oPressure;
}
Filled = true;// array has been initially filled
} else {
if ( Ticks == ticksTrigger )
{
//when tick counter reaches the trigger level
//shift contents left and insert new value at the end
for ( int i = 0; i <= BufferCount - 2; i++ )
{
*( ptr + i ) = *( ptr + (i + 1) );
}
}
*( ptr + (BufferCount - 1) ) = x_eData.oPressure;//new value to be inserted
}
// find and store highest and lowest value
if ( x_eData.oPressure > x_eData.PressureH )
{
x_eData.PressureH = x_eData.oPressure;
}
if ( x_eData.oPressure < x_eData.PressureL )
{
x_eData.PressureL = x_eData.oPressure;
}
Ticks++;
if ( Ticks > ticksTrigger )
{
Ticks = 1;
}
//log_i( "ticks %d" , Ticks );
x_eData.cngPress = CalculatePressureFactors( *( ptr + 57), *( ptr + 59) ); // going back 4 hours
xSemaphoreGive( sema_eData );
xSemaphoreGive( sema_CollectPressure );
//
//log_i( " high watermark % d", uxTaskGetStackHighWaterMark( NULL ) );
} //for (;;)
vTaskDelete( NULL );
} //void fStoreAirPressure ( void *pvParemeters )
////
void fSolarCalculations ( void *pvParameters )
{
double sunrise; // Sunrise, in hours (UTC)
double transit; // Solar noon, in hours (UTC)
double sunset; // Sunset, in hours (UTC)
double dawn; // Civil dawn, in hours (UTC)
double dusk; // Civil dusk, in hours (UTC)
//double rt_ascension; // Sun's right ascension, in degrees
//double declination; // Sun's declination, in degrees
const float time_zone = -7.0f;
TickType_t xLastWakeTime = xTaskGetTickCount();
const TickType_t xFrequency = 1000; //delay for mS
int count = 3590;
int monthX = 1;
int dayX = 1;
for (;;)
{
if ( count % 60 == 0 )
{
if ( (rtc.getHour(true) >= 12) & (rtc.getHour(true) <= 23) )
{
dayX = 0;
} else {
dayX = 1;
}
calcSunriseSunset( rtc.getYear(), (rtc.getMonth() + monthX) , (rtc.getDay() + dayX), latitude, longitude, transit, sunrise, sunset ); // Calculate the times of sunrise, transit and sunset
sunrise += time_zone;
sunset += time_zone;
SolarTimeFormat( sunrise, 0 );
SolarTimeFormat( sunset, 1 );
calcCivilDawnDusk( rtc.getYear(), (rtc.getMonth() + monthX) , rtc.getDay(), latitude, longitude, transit, dawn, dusk); // Calculate the times of civil dawn and dusk (UTC)
transit += time_zone;
dawn += time_zone;
dusk += time_zone;
SolarTimeFormat( dawn, 2 );
SolarTimeFormat( dusk, 3 );
SolarTimeFormat( transit, 4 );
xSemaphoreTake( sema_eData, portMAX_DELAY );
calcHorizontalCoordinates( rtc.getYear(), (rtc.getMonth() + monthX) , rtc.getDay(), rtc.getHour(true) , rtc.getMinute(), rtc.getSecond(), latitude, longitude, x_eData.azimuth, x_eData.elevation );
xSemaphoreGive( sema_eData );
if (count >= 3600 )
{
SolarTimeFormat( 0.0f, 5 ); // publish MQTT
//log_i( "Hour:%d Azimuth %f, elevation %f, transit %dhr %dmin, dawn %dhr %dmin, dusk %dhr %dmin", rtc.getHour(true), x_eData.azimuth, x_eData.elevation, x_eData.TransitHr, x_eData.TransitMin, x_eData.DawnHr, x_eData.DawnMin, x_eData.DuskHr, x_eData.DuskMin );
count = 0;
}
}
//log_i( " high watermark % d", uxTaskGetStackHighWaterMark( NULL ) );
xLastWakeTime = xTaskGetTickCount();
vTaskDelayUntil( &xLastWakeTime, xFrequency );
count++;
} //for (;;)
vTaskDelete( NULL );
} //void fSolarCalculations ( )
////
void SolarTimeFormat( double h, int i )
{
int hours = 0;
int minutes = 0;
if ( h != 0.0f )
{
int m = int(round(h * 60));
hours = (m / 60) % 24;
minutes = m % 60;
}
switch ( i )
{
case 0:
xSemaphoreTake( sema_eData, portMAX_DELAY );
x_eData.SunRiseHr = hours;
x_eData.SunRiseMin = minutes;
xSemaphoreGive( sema_eData );
break;
case 1:
xSemaphoreTake( sema_eData, portMAX_DELAY );
x_eData.SunSetHr = hours;
x_eData.SunSetMin = minutes;
xSemaphoreGive( sema_eData );
break;
case 2:
xSemaphoreTake( sema_eData, portMAX_DELAY );
x_eData.DawnHr = hours;
x_eData.DawnMin = minutes;
xSemaphoreGive( sema_eData );
break;
case 3:
xSemaphoreTake( sema_eData, portMAX_DELAY );
x_eData.DuskHr = hours;
x_eData.DawnMin = minutes;
xSemaphoreGive( sema_eData );
break;
case 4:
xSemaphoreTake( sema_eData, portMAX_DELAY );
x_eData.TransitHr = hours;
x_eData.TransitMin = minutes;
xSemaphoreGive( sema_eData );
break;
case 5:
xSemaphoreTake( sema_eData, portMAX_DELAY );
struct stu_eData px_eData = x_eData;
xSemaphoreGive( sema_eData );
String sTopic = "";
sTopic.reserve( 35 );
sTopic.concat( String(px_eData.SunRiseHr) + "," );
sTopic.concat( String(px_eData.SunRiseMin) + "," );
sTopic.concat( String(px_eData.SunSetHr) + "," );
sTopic.concat( String(px_eData.SunSetMin) + "," );
sTopic.concat( String(px_eData.DawnHr) + "," );
sTopic.concat( String(px_eData.DawnMin) + "," );
sTopic.concat( String(px_eData.TransitHr) + "," );
sTopic.concat( String(px_eData.TransitMin) );
xSemaphoreTake( sema_MQTT_KeepAlive, portMAX_DELAY );
MQTTclient.publish( topicSRSSDDT, sTopic.c_str() );
xSemaphoreGive( sema_MQTT_KeepAlive );
sTopic = "";
px_eData.azimuth = double(round(px_eData.azimuth * 100)) / 100; // Round to two decimal places
px_eData.elevation = double(round(px_eData.elevation * 100)) / 100;
sTopic.concat( String(px_eData.azimuth) + "," + String(px_eData.elevation) );
xSemaphoreTake( sema_MQTT_KeepAlive, portMAX_DELAY );
MQTTclient.publish( topicAzEle, sTopic.c_str() );
xSemaphoreGive( sema_MQTT_KeepAlive );
sTopic = "";
break;
} // switch ( i ) {
} // void SolarTimeFormat( double h, int i )
/*
250-400ppm Normal background concentration in outdoor ambient air
400-1,000ppm Concentrations typical of occupied indoor spaces with good air exchange
1,000-2,000ppm Complaints of drowsiness and poor air.
2,000-5,000 ppm Headaches, sleepiness and stagnant, stale, stuffy air. Poor concentration, loss of attention, increased heart rate and slight nausea may also be present.
5,000 Workplace exposure limit (as 8-hour TWA) in most jurisdictions.
>40,000 ppm Exposure may lead to serious oxygen deprivation resulting in permanent brain damage, coma, even death.
*/
void fParseDewPointWindChill( void *pvParameters )
{
while ( !MQTTclient.connected() )
{
vTaskDelay( 250 );
}
struct stu_message px_message;
String sDewPoint = "";
String sWindChill = "";
sDewPoint.reserve( payloadSize );
sWindChill.reserve( payloadSize );
for (;;)
{
if ( xQueueReceive(xQ_WindChillDewPoint, &px_message, portMAX_DELAY) == pdTRUE )
{
sDewPoint = px_message.payload;
int commaIndex = sDewPoint.indexOf(',');
sWindChill.concat ( sDewPoint.substring(0, commaIndex) );
sDewPoint.remove( 0, (commaIndex + 1) );
xSemaphoreTake ( sema_eData, portMAX_DELAY );
x_eData.WindChill = sWindChill.toFloat();
xSemaphoreGive( sema_eData );
xEventGroupSetBits( eg, evtDewPoint );
sDewPoint = "";
sWindChill = "";
}
//log_i( " high watermark % d", uxTaskGetStackHighWaterMark( NULL ) );
}
vTaskDelete( NULL );
}
////
void fGetCO2 ( void *pvParameters )
{
uint64_t TimePastKalman = esp_timer_get_time();
myMHZ19.begin( co2Serial );
myMHZ19.autoCalibration();
TickType_t xLastWakeTime = xTaskGetTickCount();
const TickType_t xFrequency = 1000; //delay for mS
SimpleKalmanFilter KF_CO2( 1.0f, 1.0f, .01f );
float temp = 0.0f;
for ( ;; )
{
KF_CO2.setProcessNoise( (esp_timer_get_time() - TimePastKalman) / 1000000.0f );
temp = KF_CO2.updateEstimate( myMHZ19.getCO2() ); // apply simple Kalman filter
TimePastKalman = esp_timer_get_time();
temp = round(temp);
xSemaphoreTake ( sema_eData, portMAX_DELAY );
x_eData.CO2 = temp;
xSemaphoreGive ( sema_eData );
xSemaphoreTake( sema_MQTT_KeepAlive, portMAX_DELAY );
MQTTclient.publish( topicCO2, String(temp).c_str() );
xSemaphoreGive( sema_MQTT_KeepAlive );
xLastWakeTime = xTaskGetTickCount();
vTaskDelayUntil( &xLastWakeTime, xFrequency );
//log_i( " high watermark % d", uxTaskGetStackHighWaterMark( NULL ) );
}
vTaskDelete( NULL );
} //void fMHZ19B ( void *pvParameters )
////
void fparseMQTT( void *pvParameters )
{
struct stu_message px_message;
for (;;)
{
if ( xQueueReceive(xQ_Message, &px_message, portMAX_DELAY) == pdTRUE )
{
xSemaphoreTake( sema_mqttOK, portMAX_DELAY );
mqttOK = 0;
xSemaphoreGive( sema_mqttOK );
if (px_message.topic == topicAQIndex )
{
}
if ( px_message.topic == topicPower )
{
log_i( "%s", String(px_message.payload) );
}
if ( px_message.topic == topicOutsidePressure )
{
xSemaphoreTake( sema_eData, portMAX_DELAY );
x_eData.oPressure = String(px_message.payload).toFloat();
xSemaphoreGive ( sema_eData );
}
if ( px_message.topic == topicOutsideHumidity )
{
xSemaphoreTake( sema_eData, portMAX_DELAY );
x_eData.oHumidity = String(px_message.payload).toFloat();
xSemaphoreGive ( sema_eData );
}
if ( px_message.topic == topicOutsideTemperature )
{
xSemaphoreTake( sema_eData, portMAX_DELAY );
x_eData.oTemperature = String(px_message.payload).toFloat();
xSemaphoreGive ( sema_eData );
}
if ( px_message.topic == topicRemainingMoisture_0 )
{
xSemaphoreTake( sema_eData, portMAX_DELAY );
x_eData.RM0 = String(px_message.payload).toFloat();
xSemaphoreGive ( sema_eData );
}
if ( px_message.topic == topicWindSpeed )
{
xSemaphoreTake( sema_eData, portMAX_DELAY );
x_eData.WS = String(px_message.payload).toFloat();
xSemaphoreGive ( sema_eData );
}
if ( px_message.topic == topicWindDirection )
{
xSemaphoreTake( sema_eData, portMAX_DELAY );
x_eData.WD = "";
x_eData.WD = String(px_message.payload);
xSemaphoreGive ( sema_eData );
}
if ( px_message.topic == topicRainfall )
{
xSemaphoreTake( sema_eData, portMAX_DELAY );
x_eData.RF = String(px_message.payload).toFloat();
xSemaphoreGive ( sema_eData );
}
if ( px_message.topic == topicWSVolts )
{
x_eData.WSV = String(px_message.payload).toFloat();
//log_i( "%f", x_eData.WSV );
}
// if ( px_message.topic == topicWSCurrent )
// {
// x_eData.WSC = String(px_message.payload).toFloat();
// }
// if ( px_message.topic == topicWSPower )
// {
// x_eData.WSP = String(px_message.payload).toFloat();
// }
if ( px_message.topic == topicDPnWI )
{
xQueueSend( xQ_WindChillDewPoint, (void *) &px_message, 1 );
}
if ( String(px_message.topic) == topicOK )
{
if ( !TimeSet)
{
String temp = "";
temp.reserve(50);
temp.concat( String(px_message.payload[0]) );
temp.concat( String(px_message.payload[1]) );
temp.concat( String(px_message.payload[2]) );
temp.concat( String(px_message.payload[3]) );
int year = temp.toInt();
temp = "";
temp.concat( String(px_message.payload[5]) + String(px_message.payload[6]) );
int month = temp.toInt();
temp = "";
temp.concat(String(px_message.payload[8]) + String(px_message.payload[9]) );
int day = temp.toInt();
temp = "";
temp.concat( String(px_message.payload[11]) + String(px_message.payload[12]) );
int hour = temp.toInt();
temp = "";
temp.concat( String(px_message.payload[14]) + String(px_message.payload[15]) );
int min = temp.toInt();
rtc.setTime( 0, min, hour, day, month, year );
log_i( "rtc %s Year %d month %d day %d", rtc.getTime(), rtc.getYear(), (rtc.getMonth() + 1), rtc.getDay() );
TimeSet = true;
}
}
} //if ( xQueueReceive(xQ_Message, &px_message, portMAX_DELAY) == pdTRUE )
} //for(;;)
vTaskDelete( NULL );
} // void fparseMQTT( void *pvParameters )
////
void fDoTheDisplayThing( void * parameter )
{
float *ptr = CollectionPressure;
int yIncrement = 18;
int CurrentY = 20;
int CurrentX = 5;
String temp1 = "";
temp1.reserve(10);
String temp2 = "";
temp2.reserve(10);
int boxSpacing = 80;
size_t item_size;
for (;;)
{
xEventGroupWaitBits (eg, evtDisplayUpdate, pdTRUE, pdTRUE, portMAX_DELAY );
xSemaphoreTake( sema_eData, portMAX_DELAY );
struct stu_eData px_eData = x_eData;
xSemaphoreGive ( sema_eData );
CurrentY = 20;
display.init();
//display.setFont(&FreeMonoBold9pt7b);
display.setFont(&FreeMono9pt7b);
//u8g2Fonts.setFont(u8g2_font_helvB08_tf);
display.setTextColor(GxEPD_BLACK);
display.setFullWindow();
display.fillScreen(GxEPD_WHITE); // set the background to white (fill the buffer with value for white)
display.setCursor( CurrentX, CurrentY );
display.drawRect( CurrentX, CurrentY , 70, 55, GxEPD_BLACK);
display.drawBitmap( CurrentX + 10, CurrentY + 5, temperature_icon16x16, 16, 16, GxEPD_BLACK);
display.setCursor( CurrentX + 30, CurrentY + 15 );
display.print( "F" );
display.setCursor( CurrentX + 10, CurrentY + 40);
display.print( String(px_eData.oTemperature) );
display.drawRect( CurrentX + boxSpacing, CurrentY , 70, 55, GxEPD_BLACK);
display.setCursor( CurrentX + 90, CurrentY + 15 );
display.print( "R.H.");
display.setCursor( CurrentX + 90, CurrentY + 35 );
display.print( String((int)px_eData.oHumidity) + "%" );
display.setCursor( CurrentX, CurrentY + 40);
display.drawRect( CurrentX + (boxSpacing * 2 ), CurrentY , 70, 55, GxEPD_BLACK);
display.setCursor( CurrentX + 163, CurrentY + 15 );
display.print( "Dew Pt" );
display.setCursor( CurrentX + 165, CurrentY + 35 );
display.print( String(px_eData.DewPoint) );
display.drawRect( CurrentX + (boxSpacing * 3 ), CurrentY , 70, 55, GxEPD_BLACK);
display.setCursor( CurrentX + 246, CurrentY + 15 );
display.print( "AQI" );
display.setCursor( CurrentX + 246, CurrentY + 35 );
display.print( String(int(px_eData.IAQ)) + "%" );
display.drawRect( CurrentX + (boxSpacing * 4 ), CurrentY , 70, 55, GxEPD_BLACK);
display.setCursor( CurrentX + 327, CurrentY + 15 );
display.print( "R.M." );
display.setCursor( CurrentX + 327, CurrentY + 35 );
display.print( String(int(px_eData.RM0)) + "%" );
// end of first line
if ( px_eData.SunRiseMin < 10 )
{
temp1.concat( "0" + String(px_eData.SunRiseMin) );
} else {
temp1.concat( String(px_eData.SunRiseMin) );
}
if ( px_eData.SunSetMin < 10 )
{
temp2.concat( "0" + String(px_eData.SunSetMin) );
} else {
temp2.concat( String(px_eData.SunSetMin) );
}
CurrentY += yIncrement;
CurrentY += yIncrement;
CurrentY += yIncrement;
CurrentY += yIncrement;
display.setCursor( CurrentX, CurrentY );
display.print( "Wind: " );
CurrentY += yIncrement;
display.setCursor( CurrentX, CurrentY );
display.print( "Speed " + String(px_eData.WS) + "KPH, Dir " + String(px_eData.WD) + " Chill " + String(px_eData.WindChill) + "F" );
CurrentY += yIncrement;
display.drawRect( CurrentX, CurrentY , 70, 55, GxEPD_BLACK);
addsun( 35, CurrentY + 30 , Small, SmallIcon );
display.setCursor( CurrentX + 5, CurrentY + 15 );
display.print( "0" + String(px_eData.SunRiseHr) + ":" + temp1 );
display.setCursor( CurrentX + 5, CurrentY + 50 );
display.print( String(px_eData.SunSetHr) + ":" + temp2 );
display.drawRect( CurrentX + boxSpacing, CurrentY , 70, 55, GxEPD_BLACK);
addraindrop(CurrentX + 110, CurrentY + 15, 7);
display.setCursor( CurrentX + 90, CurrentY + 35 );
display.print( String(px_eData.RF) );
display.setCursor( CurrentX + 100, CurrentY + 50 );
display.print( "mm" );
display.drawRect( CurrentX + (boxSpacing * 2 ), CurrentY , 70, 55, GxEPD_BLACK);
display.setCursor( CurrentX + 177, CurrentY + 15 );
display.print( "C02" );
display.setCursor( CurrentX + 165, CurrentY + 35 );
display.print( String(int(px_eData.CO2)) );
display.setCursor( CurrentX + 165, CurrentY + 50 );
display.print( "PPM" );
display.drawRect( CurrentX + (boxSpacing * 3 ), CurrentY , 70, 55, GxEPD_BLACK);
display.setCursor( CurrentX + 246, CurrentY + 50 );
display.print( String(px_eData.WSV) );
//make graph
xSemaphoreTake( sema_CollectPressure, portMAX_DELAY );
CurrentY += yIncrement * 6;
display.setCursor( CurrentX, CurrentY); //set cursor position
//display.drawLine( CurrentX, CurrentY, CurrentX + 200, CurrentY, GxEPD_BLACK);
//int BaseLine = (int)CollectionPressure[0];
int BaseLine = (int) * ptr;
int offsetX = 0;
for ( int j = 0; j < BufferCount; j++ )
{
if ( *(ptr + j) != 0.0f )
{
int yAdj = BaseLine - (int) * (ptr + j);
display.setCursor( CurrentX + offsetX, CurrentY + yAdj );
display.print( "-" );
offsetX += 5;
// log_i( "pressure %f item %d", CollectionPressure[j], j );
}
}
CurrentY += yIncrement;
display.setCursor( CurrentX, CurrentY );
display.print( String(px_eData.oPressure) + "mmHg" );
int Xone = 48;
int Yone = 59;
CurrentY += yIncrement;
display.setCursor( CurrentX, CurrentY );
display.print( PressureRateOfChange() );
xSemaphoreGive( sema_CollectPressure );
temp2 = "";
temp1 = "";
//
display.display(false); // full update
display.hibernate();
//log_i( "DoTheBME280Thing high watermark % d", uxTaskGetStackHighWaterMark( NULL ) );
} //for (;;)
vTaskDelete( NULL );
} //void fDoTheDisplayTHing( void * parameter )
////
void fmqttWatchDog( void * paramater )
{
int UpdateImeTrigger = 86400; //seconds in a day
int UpdateTimeInterval = 86300; // 1st time update in 100 counts
int maxNonMQTTresponse = 120;
for (;;)
{
vTaskDelay( 1000 );
if ( mqttOK >= maxNonMQTTresponse )
{
ESP.restart();
}
xSemaphoreTake( sema_mqttOK, portMAX_DELAY );
mqttOK++;
xSemaphoreGive( sema_mqttOK );
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 );
}
////
float fCalulate_IAQ_Index( int gasResistance, float Humidity)
{
float hum_baseline = 40.0f;
float hum_weighting = 0.25f;
float gas_offset = 0.0f;
float hum_offset = 0.0f;
float hum_score = 0.0f;
float gas_score = 0.0f;
gas_offset = oGasResistanceBaseLine - float( gasResistance );
hum_offset = float( Humidity ) - hum_baseline;
// calculate hum_score as distance from hum_baseline
if ( hum_offset > 0.0f )
{
hum_score = 100.0f - hum_baseline - hum_offset;
hum_score /= ( 100.0f - hum_baseline );
hum_score *= ( hum_weighting * 100.0f );
} else {
hum_score = hum_baseline + hum_offset;
hum_score /= hum_baseline;
hum_score *= ( 100.0f - (hum_weighting * 100.0f) );
}
//calculate gas score as distance from baseline
if ( gas_offset > 0.0f )
{
gas_score = float( gasResistance ) / oGasResistanceBaseLine;
gas_score *= ( 100.0f - (hum_weighting * 100.0f ) );
} else {
gas_score = 100.0f - ( hum_weighting * 100.0f );
}
return ( hum_score + gas_score );
} //void fCalulate_IAQ_Index( int gasResistance, float Humidity):
////
void DoTheBME680Thing( void *pvParameters )
{
SPI.begin(); // initialize the SPI library
vTaskDelay( 10 );
if (!bme.begin()) {
log_i("Could not find a valid BME680 sensor, check wiring!");
while (1);
}
// Set up oversampling and filter initialization
bme.setTemperatureOversampling(BME680_OS_8X);
bme.setHumidityOversampling(BME680_OS_2X);
bme.setPressureOversampling(BME680_OS_4X);
bme.setIIRFilterSize(BME680_FILTER_SIZE_3);
bme.setGasHeater(320, 150); // 320*C for 150 ms
//wait for a mqtt connection
while ( !MQTTclient.connected() )
{
vTaskDelay( 250 );
}
xEventGroupSetBits( eg, evtWaitForBME );
String bmeInfo = "";
bmeInfo.reserve( 100 );
for (;;)
{
xEventGroupWaitBits (eg, evtDoBME, pdTRUE, pdTRUE, portMAX_DELAY );
xSemaphoreTake ( sema_eData, portMAX_DELAY );
x_eData.Temperature = bme.readTemperature();
x_eData.Temperature = ( x_eData.Temperature * 1.8f ) + 32.0f; // (Celsius x 1.8) + 32
x_eData.Pressure = bme.readPressure();
x_eData.Pressure = x_eData.Pressure / 133.3223684f; //converts to mmHg
x_eData.Humidity = bme.readHumidity();
x_eData.IAQ = fCalulate_IAQ_Index( bme.readGas(), x_eData.Humidity );
bmeInfo.concat( String(x_eData.Temperature, 2) );
bmeInfo.concat( "," );
bmeInfo.concat( String(x_eData.Pressure, 2) );
bmeInfo.concat( "," );
bmeInfo.concat( String(x_eData.Humidity, 2) );
bmeInfo.concat( "," );
bmeInfo.concat( String(x_eData.IAQ, 2) );
xSemaphoreGive ( sema_eData );
xSemaphoreTake( sema_MQTT_KeepAlive, portMAX_DELAY );
if ( MQTTclient.connected() )
{
MQTTclient.publish( topicInsideInfo, bmeInfo.c_str() );
}
xSemaphoreGive( sema_MQTT_KeepAlive );
xSemaphoreGive( sema_PublishPM ); // release publish of dust density
xSemaphoreTake( sema_mqttOK, portMAX_DELAY );
mqttOK ++;
xSemaphoreGive( sema_mqttOK );
bmeInfo = ""; // empty the string buffer
xEventGroupSetBits( eg, evtStoreAirPressure );
vTaskDelay( 50 );
xEventGroupSetBits( eg, evtDisplayUpdate );
// log_i( "DoTheBME280Thing high watermark % d", uxTaskGetStackHighWaterMark( NULL ) );
}
vTaskDelete ( NULL );
}
////
/*
Important to not set vTaskDelay/vTaskDelayUntil to less then 10. Errors begin to develop with the MQTT and network connection.
makes the initial wifi/mqtt connection and works to keeps those connections open.
*/
void MQTTkeepalive( void *pvParameters )
{
sema_MQTT_KeepAlive = xSemaphoreCreateBinary();
xSemaphoreGive( sema_MQTT_KeepAlive ); // found keep alive can mess with a publish, stop keep alive during publish
MQTTclient.setKeepAlive( 90 ); // setting keep alive to 90 seconds makes for a very reliable connection, must be set before the 1st connection is made.
TickType_t xLastWakeTime = xTaskGetTickCount();
const TickType_t xFrequency = 250; //delay for ms
for (;;)
{
//check for a is-connected and if the WiFi 'thinks' its connected, found checking on both is more realible than just a single check
if ( (wifiClient.connected()) && (WiFi.status() == WL_CONNECTED) )
{
xSemaphoreTake( sema_MQTT_KeepAlive, portMAX_DELAY ); // whiles MQTTlient.loop() is running no other mqtt operations should be in process
MQTTclient.loop();
xSemaphoreGive( sema_MQTT_KeepAlive );
}
else {
log_i( "MQTT keep alive found MQTT status % s WiFi status % s", String(wifiClient.connected()), String(WiFi.status()) );
if ( !(wifiClient.connected()) || !(WiFi.status() == WL_CONNECTED) )
{
connectToWiFi();
}
connectToMQTT();
}
//log_i( " high watermark % d", uxTaskGetStackHighWaterMark( NULL ) );
xLastWakeTime = xTaskGetTickCount();
vTaskDelayUntil( &xLastWakeTime, xFrequency );
}
vTaskDelete ( NULL );
}
////
void connectToMQTT()
{
byte mac[5]; // create client ID from mac address
WiFi.macAddress(mac); // get mac address
String clientID = String(mac[0]) + String(mac[4]) ; // use mac address to create clientID
while ( !MQTTclient.connected() )
{
MQTTclient.connect( clientID.c_str(), mqtt_username, mqtt_password );
vTaskDelay( 250 );
}
MQTTclient.setCallback ( mqttCallback );
MQTTclient.subscribe ( topicOK );
MQTTclient.subscribe ( topicRemainingMoisture_0 );
MQTTclient.subscribe ( topicWindSpeed );
MQTTclient.subscribe ( topicWindDirection );
MQTTclient.subscribe ( topicDPnWI );
MQTTclient.subscribe ( topicOutsideTemperature );
MQTTclient.subscribe ( topicOutsideHumidity );
MQTTclient.subscribe ( topicOutsidePressure );
MQTTclient.subscribe ( topicRainfall );
//MQTTclient.subscribe ( topicWSVolts );
MQTTclient.subscribe ( topicPower );
} //void connectToMQTT()
void connectToWiFi()
{
int TryCount = 0;
while ( WiFi.status() != WL_CONNECTED )
{
TryCount++;
WiFi.disconnect();
WiFi.begin( SSID, PASSWORD );
vTaskDelay( 4000 );
if ( TryCount == 10 )
{
ESP.restart();
}
}
WiFi.onEvent( WiFiEvent );
}
////
void addraindrop(int x, int y, int scale)
{
display.fillCircle(x, y, scale / 2, GxEPD_BLACK);
display.fillTriangle(x - scale / 2, y, x, y - scale * 1.2, x + scale / 2, y , GxEPD_BLACK);
x = x + scale * 1.6; y = y + scale / 3;
display.fillCircle(x, y, scale / 2, GxEPD_BLACK);
display.fillTriangle(x - scale / 2, y, x, y - scale * 1.2, x + scale / 2, y , GxEPD_BLACK);
}
////
void addsun(int x, int y, int scale, bool IconSize)
{
int linesize = 3;
if (IconSize == SmallIcon) linesize = 1;
display.fillRect(x - scale * 2, y, scale * 4, linesize, GxEPD_BLACK);
display.fillRect(x, y - scale * 2, linesize, scale * 4, GxEPD_BLACK);
display.drawLine(x - scale * 1.3, y - scale * 1.3, x + scale * 1.3, y + scale * 1.3, GxEPD_BLACK);
display.drawLine(x - scale * 1.3, y + scale * 1.3, x + scale * 1.3, y - scale * 1.3, GxEPD_BLACK);
if (IconSize == LargeIcon) {
display.drawLine(1 + x - scale * 1.3, y - scale * 1.3, 1 + x + scale * 1.3, y + scale * 1.3, GxEPD_BLACK);
display.drawLine(2 + x - scale * 1.3, y - scale * 1.3, 2 + x + scale * 1.3, y + scale * 1.3, GxEPD_BLACK);
display.drawLine(3 + x - scale * 1.3, y - scale * 1.3, 3 + x + scale * 1.3, y + scale * 1.3, GxEPD_BLACK);
display.drawLine(1 + x - scale * 1.3, y + scale * 1.3, 1 + x + scale * 1.3, y - scale * 1.3, GxEPD_BLACK);
display.drawLine(2 + x - scale * 1.3, y + scale * 1.3, 2 + x + scale * 1.3, y - scale * 1.3, GxEPD_BLACK);
display.drawLine(3 + x - scale * 1.3, y + scale * 1.3, 3 + x + scale * 1.3, y - scale * 1.3, GxEPD_BLACK);
}
display.fillCircle(x, y, scale * 1.3, GxEPD_WHITE);
display.fillCircle(x, y, scale, GxEPD_BLACK);
display.fillCircle(x, y, scale - linesize, GxEPD_WHITE);
}
////
void loop() { }
a MQTT client.
Why not use one of the "cloud type" examples and simply change the address of the broker in the code ?
That's where I started. I'm a novice so I'm not sure I'm getting all the syntax correct (Arduino).
Pi code is working sub & pub. Its getting the ESP8266 to publish that's a struggle.
Post one of the cloud based examples, using code tags when you do
Help us help you.
Below is the example I started with with some tweaks.
#include <ESP8266WiFi.h>
#include <PubSubClient.h>
#define wifi_ssid "ssid"
#define wifi_password "password"
#define mqtt_server "10.0.0.150" // Pi-3 Mosquito server static IP
#define mqtt_port 1883
#define mqtt_user "espWiFi"
#define mqtt_password "12345"
//#define in_topic "/light/in"
//#define out_topic "/light/out"
#define in_topic "/Esp01/in"
#define out_topic "/Esp01/out"
// Replace by 2 if you aren't enable to use Serial Monitor... Don't forget to Rewire R1 to GPIO2!
//#define in_led 0
#define in_led 2
WiFiClient espClient;
PubSubClient client;
void setup() {
// Serial.begin(115200);
Serial.begin(9600);
setup_wifi();
client.setClient(espClient);
client.setServer(mqtt_server, mqtt_port);
client.setCallback(callback);
// initialize digital pin LED_BUILTIN as an output.
pinMode(in_led, OUTPUT);
digitalWrite(in_led, HIGH);
}
void setup_wifi() {
delay(10);
// We start by connecting to a WiFi network
Serial.println();
Serial.print("Connecting to ");
Serial.println(wifi_ssid);
WiFi.begin(wifi_ssid, wifi_password);
while (WiFi.status() != WL_CONNECTED) {
delay(500);
Serial.print(".");
}
Serial.println("");
Serial.println("WiFi connected");
Serial.println("IP address: ");
Serial.println(WiFi.localIP());
}
void reconnect() {
// Loop until we're reconnected
while (!client.connected()) {
Serial.print("Attempting MQTT connection...");
// Attempt to connect
// If you do not want to use a username and password, change next line to
if (client.connect("ESP8266Client")) {
// if (client.connect("ESP8266Client", mqtt_user, mqtt_password)) {
Serial.println("connected");
} else {
Serial.print("failed, rc=");
Serial.print(client.state());
Serial.println(" try again in 5 seconds");
// Wait 5 seconds before retrying
delay(5000);
}
}
}
void callback(char* topic, byte* payload, unsigned int length) {
Serial.print("Message arrived [");
Serial.print(topic);
Serial.print("] ");
for (int i = 0; i < length; i++) {
char receivedChar = (char)payload[i];
Serial.print(receivedChar);
if (receivedChar == '0')
digitalWrite(in_led, LOW);
if (receivedChar == '1')
digitalWrite(in_led, HIGH);
}
Serial.println();
}
void loop() {
if (!client.connected()) {
reconnect();
}
client.loop
();
// Publishes a random 0 and 1 like someone switching off and on randomly (random(2))
client.publish(out_topic, String(random(2)).c_str(), true);
delay(1000);
client.subscribe(in_topic);
delay(1000);
}
I'm not using Usr, Pwd, or ID with my implementation, at least not on the Pi end. I'm really looking for the bare minimum for the ESP8266 to publish a simple message. Then Once I know that works I'll expand.
Have you been able to use something like MQTT.fx to connect to the MQTT Broker?
By using MQTT.fx the connection can be worked out then transferred to your code.
Did you try using the server name of the broker instead of the IP address?
delay(1000);
client.subscribe(in_topic);
delay(1000);
subscription should be done once and done with the mqtt connection.
For example ~line 66
if (client.connect("ESP8266Client")) {
I can't find "ESP8266Client", is it a method, instance, class, built in variable? This is just one line that has me confused.
I'm not sure what MQTT.fx is. I'm using MQTT Explorer from the PC and command line from the Pi to monitor traffic.
Your reconnect() function assumes a proper network connection.
here is how I maintain the mqtt connection
void MQTTkeepalive( void *pvParameters )
{
sema_MQTT_KeepAlive = xSemaphoreCreateBinary();
xSemaphoreGive( sema_MQTT_KeepAlive ); // found keep alive can mess with a publish, stop keep alive during publish
MQTTclient.setKeepAlive( 90 ); // setting keep alive to 90 seconds makes for a very reliable connection, must be set before the 1st connection is made.
TickType_t xLastWakeTime = xTaskGetTickCount();
const TickType_t xFrequency = 250; //delay for ms
for (;;)
{
//check for a is-connected and if the WiFi 'thinks' its connected, found checking on both is more realible than just a single check
if ( (wifiClient.connected()) && (WiFi.status() == WL_CONNECTED) )
{
xSemaphoreTake( sema_MQTT_KeepAlive, portMAX_DELAY ); // whiles MQTTlient.loop() is running no other mqtt operations should be in process
MQTTclient.loop();
xSemaphoreGive( sema_MQTT_KeepAlive );
}
else {
log_i( "MQTT keep alive found MQTT status % s WiFi status % s", String(wifiClient.connected()), String(WiFi.status()) );
if ( !(wifiClient.connected()) || !(WiFi.status() == WL_CONNECTED) )
{
connectToWiFi();
}
connectToMQTT();
}
//log_i( " high watermark % d", uxTaskGetStackHighWaterMark( NULL ) );
xLastWakeTime = xTaskGetTickCount();
vTaskDelayUntil( &xLastWakeTime, xFrequency );
}
vTaskDelete ( NULL );
}
notice how I check for a valid network connection before doing client loop()?
I copied your code, changed the WiFi details and the broker address to that of my Pi 4 hosting the broker and the code is happily publishing to the /Esp01/out topic using a Wemos D1 Mini
To make things more interesting I also increased the range of random numbers published from 1 (a waste of time) to 20
Interesting it works for you. Thank-you for taking the time.
I know I'm doing something wrong. Just not sure where.
What do you see on the Serial monitor when you upload the code ?
Unfortunately I don't see any data over the serial monitor, even for sketches that work. I believe my FTDI serial adapter does not have the power to run the 8266 wifi, just enough to flash the 8266. Or I'm doing something wrong there too 
If the FTDI adapter cannot supply enough current to run the WiFi then how is the ESP ever going to make a connection to the broker ?
Do you see anything on the Serial monitor if you leave it open and reset the ESP8266 ?
I have a breadboard with 5 to 3.3VDC regulator to power the 8266 and any connected sensors. So I flash then move to the breadboard.
I almost never see anthing in the serial monitor. The last command in the output window is
Leaving...
Hard resetting via RTS pin...
I can't make sense of your description
The the ESP is on the breadboard how is it connected to the PC ?
The ESP is only connected to the PC via FTDI serial adaptor during the sketch upload (flash). After that the ESP has to communicate wireless to the broker on the Pi-3. From my PC I run MQTT-Explorer to view broker topics.
Is it possible the includes and libraries I'm using are not compatible with the ESP-01 but are with a newer D1 Mini?
For example I see some tutorials address the server "XX.X.X.XXX" and others (XX,X,X,XXX)
I see the ESP on the AP client list so the WiFi code seems to work. It is the Pub and Sub code for some reason.