Arduino - MQTT

Arduino is an open-source electronics platform based on easy-to-use hardware and software. Arduino boards are able to read inputs - light on a sensor, a finger on a button, or a Twitter message - and turn it into an output - activating a motor, turning on an LED, publishing something online. In this tutorial we show how an Arduino MKR1000 can be used to collect temperature and humididy data and store the data on an SD card. When a WIFI connection is available the data is sent to an MQTT server. The Arduino integration uses the standard MQTT import module. The application that is running on the board is written in the Arduino C++ implementation.

Hardware

The Arduino MKR has analog and digital ports. Power can be supplied by a USB or a LIPO battery and the battery is charged when USB power is present. The pin layout is displayed below.

First we connect a DHT11 temperature and humidity sensor. The signal is connected to the digital input D1.

Arduino MKR1000	DHT11
D1	Signal
VCC	VCC
GND	GND

We also include a micro SD card to store the sensor values locally.

Arduino MKR1000	MicroSD
D8 MOSI	MOSI
D9 SCK	SCK
D10 MISO	MISO
D4	CS
VCC	VCC
GND	GND

For testing purposes you can use a breadbord with jumpers. An example is displayed below.

Setup

Next connect the board to your computer using a USB cable use the Arduino IDE to upload the configuration to the board. In the headers and setup the libraries for the sensor are included and the WIFI and SD card are initialized. The time is set using an NTP time server. Make sure you enter the name of the WIFI network in WIFI_NETWORK_2G and the password in WIFI_PASSWORD.

#include <DHT.h>
#include <DHT_U.h>
#include <PubSubClient.h>
#include <SD.h>
#include <SPI.h>
#include <Time.h>
#include <TimeLib.h>
#include <WiFi101.h>
#include <WiFiClient.h>
#include <WiFiServer.h>
#include <WiFiUdp.h>

#include "dht.h"
#define DHTPIN 1
#define DHTTYPE DHT11

#define WIFI_NETWORK_2G "" 
#define WIFI_PASSWORD ""
#define TOPIC "arduino/room/325"

char mqttServer[] = "broker.emqx.io";
static const char ntpServerName[] = "us.pool.ntp.org";
const int timeZone = 1;  // Central European Time
const int chipSelect = 4;

DHT dht(DHTPIN, DHTTYPE);
WiFiClient net;
PubSubClient client(net);
time_t getNtpTime();
WiFiUDP Udp;
File dataFile;
unsigned int localPort = 8888;

void setup() {
    Serial.begin(9600);
    Serial.println("Connected");
    WiFi.hostname("Arduino");
    WiFi.begin(WIFI_NETWORK_2G, WIFI_PASSWORD);
    Serial.println("Begin...");
    client.setServer(mqttServer, 1883);
    Serial.println("Starting UDP");
    Udp.begin(localPort);
    Serial.print("Local port: ");
    Serial.println("waiting for sync");
    setSyncProvider(getNtpTime);
    setSyncInterval(300);
    SD.begin(chipSelect);
}

The loop includes a function that collects the data from the sensor, a function that logs the data to the SD card and a function that synchronizes the data with an MQTT server if there is a WIFI connection.

void loop() {    
    String payload = read();
    log(payload);
    sync();
    delay(5000);
}

Reading sensor data

The sensor data is read from the sensor and converted into a JSON payload:

String read() {
    int h = dht.readHumidity();
    int t = dht.readTemperature();

    if (isnan(h) || isnan(t)) {
        Serial.println("Failed to read from DHT sensor!");
        return "{}";
    }

    int hic = dht.computeHeatIndex(t, h, false);

    String date = "";
    date += year();
    date += "-";
    date += month() < 10 ? "0" : "";
    date += month();
    date += "-";
    date += day() < 10 ? "0" : "";
    date += day();
    date += hour() < 10 ? " 0" : " ";
    date += hour();
    date += ":";
    date += minute() < 10 ? "0" : "";
    date += minute();
    date += ":";
    date += second() < 10 ? "0" : "";
    date += second();

    String payload = "{";
    payload += "\"room\": \"325\",";
    payload += "\"time\":\"";
    payload += date;
    payload += "\",";
    payload += "\"temperature\":\"";
    payload += t;
    payload += "\",";
    payload += "\"humidity\":\"";
    payload += h;
    payload += "\"}";
    return payload;
}

Storing sensor data on SD card

The log function writes the data to two files. The log.txt file is for permanent storage and the cache.txt file is for temporary storage until the data has been streamed to an MQTT server.

void log(String payload) {
    dataFile = SD.open("cache.txt", FILE_WRITE);
    if (dataFile) {
        dataFile.println(payload);
        dataFile.close();
    } else {
        Serial.println("error opening datalog.txt");
    }
    dataFile = SD.open("log.txt", FILE_WRITE);
    if (dataFile) {
        dataFile.println(payload);
        dataFile.close();
    } else {
        Serial.println("error opening datalog.txt");
    }
    Serial.print("Log: ");
    Serial.println(payload);
}

Sending sensor data to MQTT server

The sync function checks if there is a WIFI connection and a connection to the MQTT server. If this is the case the data is read from the cache.txt file and sent to the MQTT server. Each line is a separate message. After all lines have been processed the cache.txt is deleted.

// Connect to WIFI and MQTT
void sync() {
    if (WiFi.status() != WL_CONNECTED) {
        WiFi.begin(WIFI_NETWORK_2G, WIFI_PASSWORD);
    }
    if (WiFi.status() != WL_CONNECTED) {
        Serial.println("Wifi: disconnected");
        return;
    }
    Serial.println("Wifi: connected");
    client.setServer(mqttServer, 1883);
    client.disconnect();
    client.connect("Arduino Uno Device", "", "");
    if (!client.connected()) {
        Serial.println("MQTT: disconnected");
        return;
    }
    Serial.println("MQTT: connected");

    dataFile = SD.open("cache.txt", FILE_READ);

    char line[100];
    int index = 0;
    bool published = false;
    while (dataFile.available()) {
        char c = dataFile.read();
        if (c != '\n' && c != '\r') {
            line[index] = c;
            index++;
        } else {
            if (index > 0) {
                Serial.print("Publish ");
                published = client.publish(TOPIC, line, true);
                Serial.print(line);
                Serial.println();
            }
            index = 0;
        }
    }
    dataFile.close();
    if (published) {
        bool removed = SD.remove("cache.txt");
        Serial.println("Cleared cache");
    }
}

Connect to an NTP time server

The time is retreived from an NTP server using the functions below.

// Get time from NTP server
const int NTP_PACKET_SIZE = 48;      // NTP time is in the first 48 bytes of message
byte packetBuffer[NTP_PACKET_SIZE];  //buffer to hold incoming & outgoing packets
time_t getNtpTime() {
    IPAddress ntpServerIP;  // NTP server's ip address

    while (Udp.parsePacket() > 0)
        ;  // discard any previously received packets
    Serial.println("Transmit NTP Request");
    // get a random server from the pool
    WiFi.hostByName(ntpServerName, ntpServerIP);
    Serial.print(ntpServerName);
    Serial.print(": ");
    Serial.println(ntpServerIP);
    sendNTPpacket(ntpServerIP);
    uint32_t beginWait = millis();
    while (millis() - beginWait < 1500) {
        int size = Udp.parsePacket();
        if (size >= NTP_PACKET_SIZE) {
            Serial.println("Receive NTP Response");
            Udp.read(packetBuffer, NTP_PACKET_SIZE);  // read packet into the buffer
            unsigned long secsSince1900;
            // convert four bytes starting at location 40 to a long integer
            secsSince1900 = (unsigned long)packetBuffer[40] << 24;
            secsSince1900 |= (unsigned long)packetBuffer[41] << 16;
            secsSince1900 |= (unsigned long)packetBuffer[42] << 8;
            secsSince1900 |= (unsigned long)packetBuffer[43];
            return secsSince1900 - 2208988800UL + timeZone * SECS_PER_HOUR;
        }
    }
    Serial.println("No NTP Response :-(");
    return 0;  // return 0 if unable to get the time
}


// Send an NTP request to the time server at the given address
void sendNTPpacket(IPAddress &address) {
    memset(packetBuffer, 0, NTP_PACKET_SIZE);
    packetBuffer[0] = 0b11100011;  // LI, Version, Mode
    packetBuffer[1] = 0;           // Stratum, or type of clock
    packetBuffer[2] = 6;           // Polling Interval
    packetBuffer[3] = 0xEC;        // Peer Clock Precision
    packetBuffer[12] = 49;
    packetBuffer[13] = 0x4E;
    packetBuffer[14] = 49;
    packetBuffer[15] = 52;
    Udp.beginPacket(address, 123);  //NTP requests are to port 123
    Udp.write(packetBuffer, NTP_PACKET_SIZE);
    Udp.endPacket();
}

When the code is loaded on the card you can use the serial monitor to view the output. You can check if the data is received by the MQTT server by subscribing to the TOPIC defined in the header.

mosquitto_sub -h broker.emqx.io -v -t 'arduino/room/325/#'

Configuration

Create a new secret to store the MQTT server and topic:

name: Arduino
fields:
  - name: mqtt
    value: broker.emqx.io
  - name: topic
    value: arduino/room/325

Create an importer to map the MQTT topic to an internal topic:

name: Arduino
context: Arduino
data: `
object, topic
arduino/room/325, room325
`
cron: '* * * * * *'