mqtt initial implementation

.travis.yml

@@ -22,6 +22,11 @@ before_install:
   - wget https://github.com/lewisxhe/AXP202X_Library/archive/v1.0.zip
   - unzip v1.0.zip
   - sudo mv AXP202X_Library-1.0 /usr/local/share/arduino/libraries/
+
+  - wget https://github.com/dx168b/async-mqtt-client/archive/master.zip
+  - unzip master.zip
+  - sudo mv async-mqtt-client-master /usr/local/share/arduino/libraries/
+
   # Trying to get rid of mDNS warnings (1000s of them...)
   # as suggested by https://forum.arduino.cc/index.php?topic=469428.0
   # Arduino IDE adds a lot of noise caused by network traffic, trying to firewall it off

RX_FSK/RX_FSK.ino

@@ -21,6 +21,8 @@
 #include "geteph.h"
 #include "rs92gps.h"
 
+#include "mqtt.h"
+
 #ifdef TTGO_V2
 // platformio currently fails to build with board v2 so ve override v1 pins instead
 #define OLED_SDA 4
@@ -57,6 +59,10 @@ WiFiClient client;
 WiFiServer tncserver(14580);
 WiFiClient tncclient;
 
+unsigned long lastMqttUptime = 0;
+boolean mqttEnabled;
+MQTT mqttclient;
+
 boolean forceReloadScreenConfig = false;
 
 enum KeyPress { KP_NONE = 0, KP_SHORT, KP_DOUBLE, KP_MID, KP_LONG };
@@ -468,6 +474,16 @@ struct st_configitems config_list[] = {
   {"tcp.port", "APRS TCP Port", 0, &sonde.config.tcpfeed.port},
   {"tcp.idformat", "DFM ID Format", -2, &sonde.config.tcpfeed.idformat},
   {"tcp.highrate", "Rate limit", 0, &sonde.config.tcpfeed.highrate},
+  
+  /* MQTT */
+  {"mqtt.active", "MQTT Active (needs reboot)", 0, &sonde.config.mqtt.active},
+  {"mqtt.id", "MQTT client ID", 63, &sonde.config.mqtt.id},
+  {"mqtt.host", "MQTT server IP address", 63, &sonde.config.mqtt.host},
+  {"mqtt.port", "MQTT Port", 0, &sonde.config.mqtt.port},
+  {"mqtt.username", "MQTT Username", 63, &sonde.config.mqtt.username},
+  {"mqtt.password", "MQTT Password", 63, &sonde.config.mqtt.password},
+  {"mqtt.prefix", "MQTT Prefix", 63, &sonde.config.mqtt.prefix},
+
   /* Hardware dependeing settings */
   {"", "Hardware configuration (requires reboot)", -5, NULL},
   {"disptype", "Display type (0=OLED/SSD1306, 1=TFT/ILI9225, 2=OLED/SH1106)", 0, &sonde.config.disptype},
@@ -487,6 +503,7 @@ struct st_configitems config_list[] = {
   {"gps_rxd", "GPS RXD pin (-1 to disable)", 0, &sonde.config.gps_rxd},
   {"gps_txd", "GPS TXD pin (not really needed)", 0, &sonde.config.gps_txd},
   {"mdnsname", "mDNS name", 14, &sonde.config.mdnsname},
+
 };
 const static int N_CONFIG = (sizeof(config_list) / sizeof(struct st_configitems));
 
@@ -1691,6 +1708,7 @@ void loopDecoder() {
     //Send a packet with position information
     // first check if ID and position lat+lonis ok
     SondeInfo *s = &sonde.sondeList[rxtask.receiveSonde];
+
     if (s->validID && ((s->validPos & 0x03) == 0x03)) {
       const char *str = aprs_senddata(s, sonde.config.call, sonde.config.udpfeed.symbol);
       if (connected)  {
@@ -1710,6 +1728,13 @@ void loopDecoder() {
         tncclient.write(raw, rawlen);
       }
     }
+
+    // send to MQTT if enabled
+    if (connected && mqttEnabled) {
+      Serial.println("Sending sonde info via MQTT");
+      mqttclient.publishPacket(s);
+    }
+
     // also send to web socket
     //TODO
   }
@@ -1795,6 +1820,7 @@ String translateEncryptionType(wifi_auth_mode_t encryptionType) {
   }
 }
 
+
 void enableNetwork(bool enable) {
   if (enable) {
     MDNS.begin(sonde.config.mdnsname);
@@ -1804,6 +1830,12 @@ void enableNetwork(bool enable) {
     if (sonde.config.kisstnc.active) {
       tncserver.begin();
     }
+    
+    if (sonde.config.mqtt.active && strlen(sonde.config.mqtt.host) > 0) {
+      mqttEnabled = true;
+      mqttclient.init(sonde.config.mqtt.host, sonde.config.mqtt.port, sonde.config.mqtt.id, sonde.config.mqtt.username, sonde.config.mqtt.password, sonde.config.mqtt.prefix);
+    }
+
     connected = true;
   } else {
     MDNS.end();
@@ -2367,5 +2399,12 @@ void loop() {
     sonde.updateDisplay();
     lastDisplay = currentDisplay;
   }
+
+  int now = millis();
+  if (mqttEnabled && (lastMqttUptime == 0 || (lastMqttUptime + 60000 < now) || (lastMqttUptime > now))) {
+    mqttclient.publishUptime();
+    lastMqttUptime = now;
+  }
+
   Serial.printf("Unused stack: %d\n", uxTaskGetStackHighWaterMark(0));
 }

RX_FSK/data/config.txt

@@ -96,5 +96,16 @@ tcp.symbol=/O
 tcp.highrate=20
 tcp.idformat=0
 #-------------------------------#
+# mqtt settings
+#-------------------------------#
+# data not sanitized / quality checked, outliers not filtered out
+mqtt.active=0
+mqtt.id=rdz_sonde_server
+mqtt.ip=/0
+mqtt.port=1883
+mqtt.username=/0
+mqtt.password=/0
+mqtt.prefix=rdz_sonde_server/
+#-------------------------------#
 # EOF
 #-------------------------------#

Setup.md

@@ -49,6 +49,9 @@ of your Arduino IDE, and rename main folder to AsyncTCP
 From https://github.com/lewisxhe/AXP202X_Library select "Download ZIP", extract to the libraries
 folder of your Arduino IDE, and rename main folder to AXP202X_Library-1.0
 
+From https://github.com/dx168b/async-mqtt-client select "Download ZIP", extract to the libraries
+folder of your Arduino IDE, and rename main folder to async-mqtt-client
+
 ## Additional libraries, part 3
 
 Copy the SX1278FSK, SondeLib and fonts folders from libraries of this project to your Arduino IDE's libraries

libraries/SondeLib/Sonde.cpp

@@ -189,6 +189,13 @@ void Sonde::defaultConfig() {
 	config.tcpfeed.highrate = 10;
 	config.tcpfeed.idformat = ID_DFMDXL;
 	config.kisstnc.active = 0;
+
+	config.mqtt.active = 0;
+	strcpy(config.mqtt.id, "rdz_sonde_server");
+	config.mqtt.port = 1883;
+	strcpy(config.mqtt.username, "/0");
+	strcpy(config.mqtt.password, "/0");
+	strcpy(config.mqtt.prefix, "rdz_sonde_server/");
 }
 
 void Sonde::setConfig(const char *cfg) {
@@ -314,6 +321,22 @@ void Sonde::setConfig(const char *cfg) {
 		config.tcpfeed.highrate = atoi(val);
 	} else if(strcmp(cfg,"tcp.idformat")==0) {
 		config.tcpfeed.idformat = atoi(val);
+	
+	} else if(strcmp(cfg,"mqtt.active")==0) {
+		config.mqtt.active = atoi(val)>0;
+	} else if(strcmp(cfg,"mqtt.id")==0) {
+		strncpy(config.mqtt.id, val, 63);
+	} else if(strcmp(cfg,"mqtt.host")==0) {
+		strncpy(config.mqtt.host, val, 63);
+	} else if(strcmp(cfg,"mqtt.port")==0) {
+		config.mqtt.port = atoi(val);
+	} else if(strcmp(cfg,"mqtt.username")==0) {
+		strncpy(config.mqtt.username, val, 63);
+	} else if(strcmp(cfg,"mqtt.password")==0) {
+		strncpy(config.mqtt.password, val, 63);
+	} else if(strcmp(cfg,"mqtt.prefix")==0) {
+		strncpy(config.mqtt.prefix, val, 63);
+
 	} else {
 		Serial.printf("Invalid config option '%s'=%s \n", cfg, val);
 	}

libraries/SondeLib/Sonde.h

@@ -144,6 +144,15 @@ struct st_kisstnc {
         int idformat;
 };
 
+struct st_mqtt {
+	bool active;
+	char id[64];
+	char host[64];
+	int port;
+	char username[64];
+	char password[64];
+	char prefix[64];
+};
 
 typedef struct st_rdzconfig {
 	// hardware configuration
@@ -188,6 +197,7 @@ typedef struct st_rdzconfig {
 	struct st_feedinfo udpfeed;	// target for AXUDP messages
 	struct st_feedinfo tcpfeed;	// target for APRS-IS TCP connections
 	struct st_kisstnc kisstnc;	// target for KISS TNC (via TCP, mainly for APRSdroid)
+	struct st_mqtt mqtt;
 } RDZConfig;
 
 

libraries/SondeLib/mqtt.cpp

@@ -0,0 +1,122 @@
+#include <Arduino.h>
+#include "mqtt.h"
+#include <WiFi.h>
+#include <AsyncMqttClient.h>
+
+void mqttCallback(char* topic, byte* payload, unsigned int length) {
+  Serial.print("Message arrived [");
+  Serial.print(topic);
+  Serial.print("] ");
+  for (int i=0;i<length;i++) {
+    Serial.print((char)payload[i]);
+  }
+  Serial.println();
+}
+
+
+void MQTT::init(const char* ip, uint16_t port, const char* id, const char *username, const char *password, const char *prefix)
+{
+    this->ip = this->ip.fromString(ip);
+    this->port = port;
+    this->username = username;
+    this->password = password;
+    this->prefix = prefix;
+    
+    char buffer[20];
+    snprintf(buffer, 20, "%s%d", id, random(0, 1000));
+    this->id = buffer;
+
+    Serial.println("[MQTT] pubsub client");
+    mqttClient.setServer(ip, port);
+    if (strlen(password) > 0) {
+        mqttClient.setCredentials(username, password);
+    }
+    mqttClient.connect();
+    
+    //mqttReconnectTimer = xTimerCreate("mqttTimer", pdMS_TO_TICKS(2000), pdFALSE, (void*)0, connectToMqtt);
+}
+
+void MQTT::connectToMqtt() {
+  Serial.println("Connecting to MQTT...");
+  mqttClient.connect();
+}
+
+void MQTT::publishUptime()
+{
+    Serial.println("[MQTT] writing");
+    char payload[12];
+    snprintf(payload, 12, "%lu", millis());
+    char topic[128];
+    snprintf(topic, 128, "%s%s", this->prefix, "uptime");
+    mqttClient.publish(topic, 1, 1, payload);
+}
+
+void MQTT::publishPacket(SondeInfo *s)
+{
+    char payload[1024];
+    snprintf(payload, 1024, "{"
+        "\"active\": %d,"
+        "\"freq\": %.2f,"
+        "\"id\": \"%s\","
+        "\"ser\": \"%s\","
+        "\"validId\": %d,"
+        "\"launchsite\": \"%s\","
+        "\"lat\": %.5f,"
+        "\"lon\": %.5f,"
+        "\"alt\": %.1f,"
+        "\"vs\": %.1f,"
+        "\"hs\": %.1f,"
+        "\"dir\": %.1f,"
+        "\"sats\": %d,"
+        "\"validPos\": %d,"
+        "\"time\": %d,"
+        "\"sec\": %d,"
+        "\"frame\": %d,"
+        "\"validTime\": %d,"
+        "\"rssi\": %d,"
+        "\"afc\": %d,"
+        "\"rxStat\": \"%s\","
+        "\"rxStart\": %d,"
+        "\"norxStart\": %d,"
+        "\"viewStart\": %d,"
+        "\"lastState\": %d,"
+        "\"launchKT\": %d,"
+        "\"burstKT\": %d,"
+        "\"countKT\": %d,"
+        "\"crefKT\": %d,"
+        "}",
+        (int)s->active,
+        s->freq,
+        s->id,
+        s->ser,
+        (int)s->validID,
+        s->launchsite,
+        s->lat,
+        s->lon,
+        s->alt,
+        s->vs,
+        s->hs,
+        s->dir,
+        s->sats,
+        s->validPos,
+        s->time,
+        s->sec,
+        s->frame,
+        (int)s->validTime,
+        s->rssi,
+        s->afc,
+        s->rxStat,
+        s->rxStart,
+        s->norxStart,
+        s->viewStart,
+        s->lastState,
+        s->launchKT,
+        s->burstKT,
+        s->countKT,
+        s->crefKT
+    );
+
+    char topic[128];
+    snprintf(topic, 128, "%s%s", this->prefix, "packet");
+    mqttClient.publish(topic, 1, 1, payload);
+}

libraries/SondeLib/mqtt.h

@@ -0,0 +1,28 @@
+#ifndef MQTT_h
+#define MQTT_h
+
+#include <WiFi.h>
+#include <AsyncMqttClient.h>
+#include "Sonde.h"
+
+class MQTT
+{
+public:
+    WiFiClient mqttWifiClient;
+    AsyncMqttClient mqttClient;
+    TimerHandle_t mqttReconnectTimer;
+    IPAddress ip;
+    uint16_t port;
+    const char *id;
+    const char *username;
+    const char *password;
+    const char *prefix;
+
+    void init(const char *ip, uint16_t port, const char *id, const char *username, const char *password, const char *prefix);
+    void publishPacket(SondeInfo *s);
+    void publishUptime();
+private:
+    void connectToMqtt();
+};
+
+#endif

platformio.ini

@@ -24,18 +24,10 @@ lib_deps_external =
     stevemarple/MicroNMEA @ ^2.0.3
     nkawu/TFT 22 ILI9225 @ ^1.4.4
     me-no-dev/ESP Async WebServer @ ^1.2.3
-
+    https://github.com/dx168b/async-mqtt-client
-[env:ttgo-lora32-v1]
+    
-platform = espressif32
+[env:ttgo-lora32]
-board = ttgo-lora32-v1
+platform = espressif32@1.12.4
-framework = arduino
-monitor_speed = 115200
-lib_deps =
-  ${extra.lib_deps_builtin}
-  ${extra.lib_deps_external}
-
-[env:ttgo-lora32-v2]
-platform = espressif32
 board = ttgo-lora32-v1
 framework = arduino
 monitor_speed = 115200