EnigmaIOTGatewayMQTT.cpp

Go to the documentation of this file.

 
#include <Arduino.h>
 
#include "GwOutput_mqtt.h"
 
#ifdef SECURE_MQTT
#include <WiFiClientSecure.h>
#else
#include <WiFiClient.h>
#endif // SECURE_MQTT
 
 
#ifdef ESP32
#include <WiFi.h>
#include "soc/soc.h"           // Disable brownout problems
#include "soc/rtc_cntl_reg.h"  // Disable brownout problems
#elif defined ESP8266
#include <ESP8266WiFi.h>
#endif // ESP32
 
#include <ArduinoOTA.h>
 
#include <CayenneLPP.h>
 
#include <EnigmaIOTGateway.h>
#include <helperFunctions.h>
#include <EnigmaIOTdebug.h>
#include <espnow_hal.h>
#include <ArduinoJson.h>
#include <ESPAsyncWiFiManager.h>
 
//#define MEAS_TEMP // Temperature measurement for Gateway monitoring using DS18B20
 
#ifdef MEAS_TEMP
#include <DallasTemperature.h>
#include <OneWire.h>
const time_t statusPeriod = 300 * 1000;
const int DS18B20_PIN = 16;
const int DS18B20_PREC = 12;
OneWire ow (DS18B20_PIN);
DallasTemperature ds18b20 (&ow);
DeviceAddress dsAddress;
float temperature;
#endif
 
#ifndef LED_BUILTIN
#define LED_BUILTIN 5
#endif // BUILTIN_LED
 
#define BLUE_LED LED_BUILTIN
#define RED_LED LED_BUILTIN
 
#ifdef ESP32
TimerHandle_t connectionLedTimer;
#elif defined(ESP8266)
ETSTimer connectionLedTimer;
#endif // ESP32
 
const int connectionLed = LED_BUILTIN;
bool connectionLedFlashing = false;
 
bool restartRequested = false;
time_t restartRequestTime;
 
 
void flashConnectionLed (void* led) {
    //digitalWrite (*(int*)led, !digitalRead (*(int*)led));
    digitalWrite (LED_BUILTIN, !digitalRead (LED_BUILTIN));
}
 
void startConnectionFlash (int period) {
#ifdef ESP32
    if (!connectionLedFlashing) {
        connectionLedFlashing = true;
        connectionLedTimer = xTimerCreate ("led_flash", pdMS_TO_TICKS (period), pdTRUE, (void*)0, flashConnectionLed);
        xTimerStart (connectionLedTimer, 0);
    }
#elif defined (ESP8266)
    ets_timer_disarm (&connectionLedTimer);
    if (!connectionLedFlashing) {
        connectionLedFlashing = true;
        ets_timer_arm_new (&connectionLedTimer, period, true, true);
    }
#endif // ESP32
}
 
void stopConnectionFlash () {
#ifdef ESP32
    if (connectionLedFlashing) {
        connectionLedFlashing = false;
        xTimerStop (connectionLedTimer, 0);
        xTimerDelete (connectionLedTimer, 0);
    }
#elif defined(ESP8266)
    if (connectionLedFlashing) {
        connectionLedFlashing = false;
        ets_timer_disarm (&connectionLedTimer);
        digitalWrite (connectionLed, LED_OFF);
    }
#endif // ESP32
}
 
void arduinoOTAConfigure () {
    // Port defaults to 3232
    // ArduinoOTA.setPort(3232);
 
    // Hostname defaults to esp3232-[MAC]
    ArduinoOTA.setHostname (EnigmaIOTGateway.getNetworkName ());
 
    // No authentication by default
    ArduinoOTA.setPassword (EnigmaIOTGateway.getNetworkKey (true));
 
    // Password can be set with it's md5 value as well
    // MD5(admin) = 21232f297a57a5a743894a0e4a801fc3
    // ArduinoOTA.setPasswordHash("21232f297a57a5a743894a0e4a801fc3");
 
    ArduinoOTA.onStart ([] () {
        if (ArduinoOTA.getCommand () == U_FLASH) {
            DEBUG_WARN ("Start updating sketch");
        } else {// U_SPIFFS
            DEBUG_WARN ("Start updating filesystem");
            // NOTE: if updating SPIFFS this would be the place to unmount SPIFFS using SPIFFS.end()
        }
                        });
    ArduinoOTA.onEnd ([] () {
        DEBUG_WARN ("OTA Finished");
                      });
    ArduinoOTA.onProgress ([] (unsigned int progress, unsigned int total) {
        static bool printed = false;
        unsigned int percent = progress / (total / 100);
        digitalWrite (BLUE_LED, !digitalRead (BLUE_LED));
        // if (!(percent % 1)) {
        //  //Serial.print ('.');
        // }
        if (!(percent % 20) && !printed && percent != 0) {
            DEBUG_WARN (" %d%%\n", percent);
            printed = true;
        } else if (percent % 20) {
            printed = false;
        }
        if (progress == total) {
            DEBUG_WARN ("OTA transfer finished");
        }
                           });
    ArduinoOTA.onError ([] (ota_error_t error) {
        DEBUG_WARN ("OTA Error[%u]: ", error);
        if (error == OTA_AUTH_ERROR) { DEBUG_WARN ("OTA Auth Failed"); }        else if (error == OTA_BEGIN_ERROR) { DEBUG_WARN ("OTA Begin Failed"); }         else if (error == OTA_CONNECT_ERROR) { DEBUG_WARN ("OTA Connect Failed"); }         else if (error == OTA_RECEIVE_ERROR) { DEBUG_WARN ("OTA Receive Failed"); }         else if (error == OTA_END_ERROR) { DEBUG_WARN ("OTA End Failed"); }
                        });
 
    ArduinoOTA.begin ();
}
 
void wifiManagerExit (boolean status) {
    GwOutput.configManagerExit (status);
}
 
void wifiManagerStarted () {
    GwOutput.configManagerStart (&EnigmaIOTGateway);
}
 
void processRxControlData (char* macStr, uint8_t* data, uint8_t length) {
    if (data) {
        if (data[0] == VERSION_ANS && length >= 4) {
            DEBUG_INFO ("Version message: %d.%d.%d", data[1], data[2], data[3]);
            Node* node = EnigmaIOTGateway.getNodes ()->getNodeFromName (macStr);
            if (node) {
                node->setVersion (data[1], data[2], data[3]);
            }
        }
        GwOutput.outputControlSend (macStr, data, length);
    }
}
 
void doRestart () {
    DEBUG_WARN ("Restart requested");
    const size_t capacity = JSON_OBJECT_SIZE (1);
    size_t len;
    char* payload;
 
    DynamicJsonDocument doc (capacity);
 
    doc["action"] = "restart";
 
    len = measureJson (doc) + 1;
    payload = (char*)malloc (len);
    serializeJson (doc, (char*)payload, len);
    char addr[] = "gateway";
    GwOutput.outputDataSend (addr, payload, len - 1);
    free (payload);
 
    restartRequested = true;
    restartRequestTime = millis ();
 
}
 
#if SUPPORT_HA_DISCOVERY
void processHADiscovery (const char* topic, char* message, size_t len) {
    DEBUG_INFO ("About to process HA discovery. Len: %d - %s --> %.*s", len, topic, len, message);
    GwOutput.rawMsgSend (topic, message, len, true);
}
#endif
 
void processRxData (uint8_t* mac, uint8_t* buffer, uint8_t length, uint16_t lostMessages, bool control, gatewayPayloadEncoding_t payload_type, char* nodeName = NULL) {
    uint8_t* addr = mac;
    size_t pld_size = 0;
    const int PAYLOAD_SIZE = 1024; // Max MQTT payload in PubSubClient library normal operation.
 
    char payload[PAYLOAD_SIZE];
 
    char mac_str[ENIGMAIOT_ADDR_LEN * 3];
    mac2str (addr, mac_str);
 
    if (control) {
        processRxControlData (nodeName ? nodeName : mac_str, buffer, length);
        return;
    }
    //char* netName = EnigmaIOTGateway.getNetworkName ();
    if (payload_type == CAYENNELPP) {
        DEBUG_INFO ("CayenneLPP message");
        const int capacity = JSON_ARRAY_SIZE (25) + 25 * JSON_OBJECT_SIZE (4);
        DynamicJsonDocument jsonBuffer (capacity);
        JsonArray root = jsonBuffer.createNestedArray ();
        CayenneLPP cayennelpp (MAX_DATA_PAYLOAD_SIZE);
 
        cayennelpp.decode ((uint8_t*)buffer, length, root);
        uint8_t error = cayennelpp.getError ();
        if (error != LPP_ERROR_OK) {
            DEBUG_ERROR ("Error decoding CayenneLPP data: %d", error);
            return;
        }
        pld_size = serializeJson (root, payload, PAYLOAD_SIZE);
    } else if (payload_type == MSG_PACK) {
        DEBUG_INFO ("MsgPack message");
        const int capacity = JSON_ARRAY_SIZE (25) + 25 * JSON_OBJECT_SIZE (4);
        DynamicJsonDocument jsonBuffer (capacity);
        DeserializationError error = deserializeMsgPack (jsonBuffer, buffer, length);
        if (error != DeserializationError::Ok) {
            DEBUG_ERROR ("Error decoding MSG Pack data: %s", error.c_str ());
            return;
        }
        pld_size = serializeJson (jsonBuffer, payload, PAYLOAD_SIZE);
    } else if (payload_type == RAW) {
        DEBUG_INFO ("RAW message");
        if (length <= PAYLOAD_SIZE) {
            memcpy (payload, buffer, length);
            pld_size = length;
        } else { // This will not happen but may lead to errors in case of using another physical transport
            memcpy (payload, buffer, PAYLOAD_SIZE);
            pld_size = PAYLOAD_SIZE;
        }
    }
 
    GwOutput.outputDataSend (nodeName ? nodeName : mac_str, payload, pld_size);
    DEBUG_INFO ("Published data message from %s, length %d: %s, Encoding 0x%02X", nodeName ? nodeName : mac_str, pld_size, payload, payload_type);
    if (lostMessages > 0) {
        pld_size = snprintf (payload, PAYLOAD_SIZE, "{\"lostMessages\":%u}", lostMessages);
        GwOutput.outputDataSend (nodeName ? nodeName : mac_str, payload, pld_size, GwOutput_data_type::lostmessages);
        DEBUG_INFO ("Published MQTT from %s: %s", nodeName ? nodeName : mac_str, payload);
    }
#if ENABLE_STATUS_MESSAGES
    pld_size = snprintf (payload, PAYLOAD_SIZE, "{\"per\":%e,\"lostmessages\":%u,\"totalmessages\":%u,\"packetshour\":%.2f}",
                         EnigmaIOTGateway.getPER ((uint8_t*)mac),
                         EnigmaIOTGateway.getErrorPackets ((uint8_t*)mac),
                         EnigmaIOTGateway.getTotalPackets ((uint8_t*)mac),
                         EnigmaIOTGateway.getPacketsHour ((uint8_t*)mac));
    GwOutput.outputDataSend (nodeName ? nodeName : mac_str, payload, pld_size, GwOutput_data_type::status);
    DEBUG_INFO ("Published MQTT from %s: %s", nodeName ? nodeName : mac_str, payload);
#endif
}
 
void onDownlinkData (uint8_t* address, char* nodeName, control_message_type_t msgType, char* data, unsigned int len) {
    uint8_t* buffer;
    unsigned int bufferLen = len;
    gatewayPayloadEncoding_t encoding = ENIGMAIOT;
 
    if (nodeName) {
        DEBUG_INFO ("DL Message for %s. Type 0x%02X", nodeName, msgType);
    } else {
        DEBUG_INFO ("DL Message for " MACSTR ". Type 0x%02X", MAC2STR (address), msgType);
    }
    DEBUG_DBG ("Data: %.*s Length: %d", len, data, len);
 
    if (msgType == USERDATA_GET || msgType == USERDATA_SET) {
        const int capacity = JSON_ARRAY_SIZE (25) + 25 * JSON_OBJECT_SIZE (4);
        DynamicJsonDocument json (capacity);
        DeserializationError error = deserializeJson (json, data, len, DeserializationOption::NestingLimit (3));
        if (error == DeserializationError::Ok) {
            DEBUG_INFO ("JSON Message. Result %s", error.c_str ());
            bufferLen = measureMsgPack (json) + 1; // Add place for \0
            buffer = (uint8_t*)malloc (bufferLen);
            bufferLen = serializeMsgPack (json, (char*)buffer, bufferLen);
            encoding = MSG_PACK;
        } else {
            DEBUG_INFO ("Not JSON Message. Error %s", error.c_str ());
            bufferLen++; // Add place for \0
            buffer = (uint8_t*)malloc (bufferLen);
            sprintf ((char*)buffer, "%.*s", len, data);
            encoding = RAW;
        }
    } else {
        bufferLen = len + 1;
        buffer = (uint8_t*)calloc (sizeof (uint8_t), bufferLen);
        memcpy (buffer, data, len);
    }
 
 
    if (!EnigmaIOTGateway.sendDownstream (address, buffer, bufferLen, msgType, encoding, nodeName)) {
        if (nodeName) {
            DEBUG_WARN ("Error sending esp_now message to %s", nodeName);
        } else {
            DEBUG_WARN ("Error sending esp_now message to " MACSTR, MAC2STR (address));
        }
    } else {
        DEBUG_DBG ("Esp-now message sent or queued correctly");
    }
 
    free (buffer);
}
 
void newNodeConnected (uint8_t* mac, uint16_t node_id, char* nodeName = nullptr) {
 
    //Serial.printf ("New node connected: %s\n", macstr);
 
    if (nodeName) {
        if (!GwOutput.newNodeSend (nodeName, node_id)) {
            DEBUG_WARN ("Error sending new node %s", nodeName);
        } else {
            DEBUG_DBG ("New node %s message sent", nodeName);
        }
    } else {
        char macstr[ENIGMAIOT_ADDR_LEN * 3];
        mac2str (mac, macstr);
        if (!GwOutput.newNodeSend (macstr, node_id)) {
            DEBUG_WARN ("Error sending new node %s", macstr);
        } else {
            DEBUG_DBG ("New node %s message sent", macstr);
        }
    }
 
}
 
void nodeDisconnected (uint8_t* mac, gwInvalidateReason_t reason) {
    char macstr[ENIGMAIOT_ADDR_LEN * 3];
    mac2str (mac, macstr);
    //Serial.printf ("Node %s disconnected. Reason %u\n", macstr, reason);
    if (!GwOutput.nodeDisconnectedSend (macstr, reason)) {
        DEBUG_WARN ("Error sending node disconnected %s reason %d", macstr, reason);
    } else {
        DEBUG_DBG ("Node %s disconnected message sent. Reason %d", macstr, reason);
    }
}
 
//#ifdef ESP32
//void EnigmaIOTGateway_handle (void * param) {
//  for (;;) {
//      EnigmaIOTGateway.handle ();
//      vTaskDelay (0);
//  }
//}
//
//void GwOutput_handle (void* param) {
//  for (;;) {
//      GwOutput.loop ();
//      vTaskDelay (0);
//  }
//}
//
//TaskHandle_t xEnigmaIOTGateway_handle = NULL;
//TaskHandle_t gwoutput_handle = NULL;
//#endif // ESP32
 
void setup () {
    Serial.begin (921600); Serial.println (); Serial.println ();
 
#ifdef ESP32
// Turn-off the 'brownout detector' to avoid random restarts during wake up,
// normally due to bad quality regulator on board
    WRITE_PERI_REG (RTC_CNTL_BROWN_OUT_REG, 0);
#endif
 
#ifdef ESP8266
    ets_timer_setfn (&connectionLedTimer, flashConnectionLed, (void*)&connectionLed);
#elif defined ESP32
 
#endif
    pinMode (LED_BUILTIN, OUTPUT);
    digitalWrite (LED_BUILTIN, LED_ON);
 
#ifdef MEAS_TEMP
    ds18b20.begin ();
    
    DEBUG_WARN ("Found %u sensors", ds18b20.getDeviceCount ());
    
    if (ds18b20.getAddress (dsAddress, 0)) {
        DEBUG_WARN ("DS18B20 address: %02X %02X %02X %02X %02X %02X %02X %02X",
                    dsAddress[0], dsAddress[1], dsAddress[2], dsAddress[3],
                    dsAddress[4], dsAddress[5], dsAddress[6], dsAddress[7]);
    } else {
        DEBUG_WARN ("No DS18B20 found");
    }
    ds18b20.setWaitForConversion (false);
    ds18b20.setResolution (DS18B20_PREC);
#endif // MEAS_TEMP
 
    startConnectionFlash (100);
    
    if (!GwOutput.loadConfig ()) {
        DEBUG_WARN ("Error reading config file");
    }
 
    EnigmaIOTGateway.setRxLed (BLUE_LED);
    EnigmaIOTGateway.setTxLed (RED_LED);
    EnigmaIOTGateway.onNewNode (newNodeConnected);
    EnigmaIOTGateway.onNodeDisconnected (nodeDisconnected);
    EnigmaIOTGateway.onWiFiManagerStarted (wifiManagerStarted);
    EnigmaIOTGateway.onWiFiManagerExit (wifiManagerExit);
    EnigmaIOTGateway.onDataRx (processRxData);
#if SUPPORT_HA_DISCOVERY
    EnigmaIOTGateway.onHADiscovery (processHADiscovery);
#endif
    EnigmaIOTGateway.onGatewayRestartRequested (doRestart);
 
    EnigmaIOTGateway.begin (&Espnow_hal);
 
    WiFi.mode (WIFI_AP_STA);
    WiFi.begin ();
 
    EnigmaIOTGateway.configWiFiManager ();
 
    WiFi.softAP (EnigmaIOTGateway.getNetworkName (), EnigmaIOTGateway.getNetworkKey (true));
    stopConnectionFlash ();
 
    DEBUG_INFO ("STA MAC Address: %s", WiFi.macAddress ().c_str ());
    DEBUG_INFO ("AP MAC Address: %s", WiFi.softAPmacAddress ().c_str ());
    DEBUG_INFO ("BSSID Address: %s", WiFi.BSSIDstr ().c_str ());
 
    DEBUG_INFO ("IP address: %s", WiFi.localIP ().toString ().c_str ());
    DEBUG_INFO ("WiFi Channel: %d", WiFi.channel ());
    DEBUG_INFO ("WiFi SSID: %s", WiFi.SSID ().c_str ());
    DEBUG_INFO ("Network Name: %s", EnigmaIOTGateway.getNetworkName ());
 
    GwOutput.setDlCallback (onDownlinkData);
    GwOutput.begin ();
 
    arduinoOTAConfigure ();
 
#ifdef ESP32
    //xTaskCreate (EnigmaIOTGateway_handle, "handle", 10000, NULL, 1, &xEnigmaIOTGateway_handle);
    //xTaskCreatePinnedToCore (EnigmaIOTGateway_handle, "handle", 4096, NULL, 0, &xEnigmaIOTGateway_handle, 1);
    //xTaskCreatePinnedToCore (GwOutput_handle, "gwoutput", 10000, NULL, 2, &gwoutput_handle, 1);
#endif
}
 
#ifdef MEAS_TEMP
void sendStatus (float temperature) {
    const size_t capacity = JSON_OBJECT_SIZE (1) + JSON_OBJECT_SIZE (3) + 30;;
    size_t len;
    char* payload;
 
    DynamicJsonDocument doc (capacity);
 
    JsonObject status = doc.createNestedObject ("status");
    if (temperature > -100) {
        status["temp"] = temperature;
    }
    status["nodes"] = EnigmaIOTGateway.getActiveNodesNumber ();
    status["mem"] = ESP.getFreeHeap ();
 
    len = measureJson (doc) + 1;
    payload = (char*)malloc (len);
    serializeJson (doc, (char*)payload, len);
    char addr[] = "gateway";
    GwOutput.outputDataSend (addr, payload, len - 1);
    free (payload);
}
#endif // MEAS_TEMP
 
void loop () {
    GwOutput.loop ();
    EnigmaIOTGateway.handle ();
    ArduinoOTA.handle ();
 
#ifdef MEAS_TEMP
    static bool tempRequested = false;
    static time_t lastTempTime = 0;
 
    if (ds18b20.validAddress (dsAddress)) {
        if ((millis () - lastTempTime > statusPeriod && !tempRequested) || !lastTempTime) {
            if (ds18b20.requestTemperaturesByIndex (0)) {
                DEBUG_INFO ("Temperature requested");
                lastTempTime = millis ();
                tempRequested = true;
            } else {
                DEBUG_WARN ("Temperature request error");
            }
        }
        if (tempRequested) {
            if (ds18b20.isConversionComplete ()) {
                temperature = ds18b20.getTempC (dsAddress);
                sendStatus (temperature);
                DEBUG_WARN ("Temperature: %f", temperature);
                tempRequested = false;
            }
        }
    }
#endif // MEAS_TEMP
 
    if (restartRequested) {
        if (millis () - restartRequestTime > 100) {
            ESP.restart ();
        }
    }
}