Developing IoT Mashups with Docker, MQTT, Node-RED, InfluxDB, Grafana

Page for the tutorial " Developping IoT Mashups with Docker, MQTT, Node-RED, InfluxDB, Grafana " at Eclipse IoT Days Grenoble 2016

COMING SOON

=Install Docker=

See Docker

=Install Node.js and NPM= On Raspian sudo apt-get install nodejs npm

=Install Node-RED=

On Raspian Jessie sudo apt-get update sudo apt-get install nodered Or sudo npm install -g --unsafe-perm node-red

The default directory is : ~/.node-red

Browse http://127.0.0.1:1880/

Now, you can write your own Node-RED flows and functions.

Install the admin cmdline tool

sudo npm install -g node-red-admin node-red-admin help

node-red-admin list
 * 1) installed modules

node-red-admin search serial node-red-admin search rfxcom node-red-admin search zwave node-red-admin search ble node-red-admin search bluetooth node-red-admin search sensortag
 * 1) search modules

node-red-admin search beaglebone node-red-admin search gpio

node-red-admin search mongodb node-red-admin search redis node-red-admin search cassandra node-red-admin search elastic node-red-admin search influxdb node-red-admin search kafka

Remark: node-red-admin install does not work ! Use

cd ~/.node-red sudo npm install -g node-red-node-redis sudo npm install -g node-red-contrib-kafka-consumer sudo npm install -g node-red-node-mongodb

Restart Node-Red

node-red-admin list

=Install extra nodes for Node-RED= Extra nodes are provided with the Node-RED community. There are listed here.

Serial for Geiger Counter


Install the serial node (node-red-node-serialport) npm install node-red-node-serialport or sudo npm install -g npm@2.x npm install node-red-node-serialport@0.0.5 if node.js prior to 4.x (ie v0.10.x and v0.12.x)

Restart Node-RED with.

Check the module in the list node-red-admin list | grep serial

Check available serial ports (/dev/tty.usbserial* on MacOS X, /dev/ttyUSB* ...) with.

Connect the Geiger counter to the host.

Check available serial ports (/dev/tty.usbserial* on MacOS X, ...) with.

Add a node serial "Geiger" with a Settings of 9600/8/N/1 and 'Split input into fixed lenghts of 1 chars'.

Add a node debug.

Connect "Geiger" to debug.

Deploy the flow.

The Geiger Counter sends a random sequence of 0 and 1.

Add a node function "Count Particles" with a flow-scoped variable (geiger/count): var COUNT='geiger/count';

// initialise the counter to 0 if it doesn't exist already var count = flow.get(COUNT)||0; count += 1; // store the value back flow.set(COUNT,count); // make it part of the outgoing msg object msg.count = count; msg.payload = {}; msg.payload.device = "geiger"; msg.payload.count = count;

node.log("Received particles : " + count);

return msg;

Connect node "Geiger" to node "Count Particles".

Deploy the new flow.

Edit the node "Count Particles" and add the 2 following statements in order to display the count into the node's status. ... setTimeout(function { node.status({}); }, 1000); // The shape property can be: ring or dot. // The fill property can be: red, green, yellow, blue or grey node.status({fill:"green",shape:"dot",text:"#"+count}); return msg;

Deploy the new flow.

Add a node inject "One minute timer" with a repeat interval of 1 minute.

Add a node function "Reset Particles Count" with a flow-scoped variable (geiger/count): var COUNT='geiger/count';

// initialise the counter to 0 if it doesn't exist already var count = flow.get(COUNT)||0;

msg.count = count; msg.payload = {}; msg.payload.device = "geiger"; msg.payload.ppm = count;

node.log("Reset counter at " + count);

// make it part of the outgoing msg object count = 0; // store the value back flow.set(COUNT,count);

return msg;

Connect node "One minute timer" to node "Reset Particles Count" and node "Reset Particles Count" to node debug.

Deploy the new flow.

The result is:

Install RFXCom for Oregon Sensors


Install the rfxcom node (node-red-contrib-rfxcom) cd ~/.node-red npm install -g node-red-contrib-rfxcom

Restart Node-RED with.

Check the module in the list node-red-admin list | grep rfx

Connect the RFXCom receiver to the host.

Check available serial ports (/dev/tty.usbserial* on MacOS X, ...) with.

Add a node rfxcom-sensors "RFXCom" with the correct serial port.

Add a node debug display the full message (not only msg.payload).

Connect "RFXCom" to debug.

Deploy the flow.

The flow loos like that:

Serial for Arduino + Weather Shield
Plug the Weather Shield into the Arduino Leonardo.

Install the Arduino IDE on your host (link)

Install 2 extra libraries (Menu Sketch > Include Library > Add ZIP Library)
 * https://codeload.github.com/sparkfun/SparkFun_HTU21D_Breakout_Arduino_Library/zip/master
 * https://codeload.github.com/sparkfun/SparkFun_MPL3115A2_Breakout_Arduino_Library/zip/master

Select the right board (Arduino Leonardo) with the menu Tool

Load and unzip the Weather_Shield sketch bundle https://codeload.github.com/sparkfun/Weather_Shield/zip/master

Load the sketch Weather_Shield/firmware/Weather_Shield/Weather_Shield.ino into the Arduino Leonardo

Check the output with the menu Tool > Serial Monitor

The output looks like this: $,winddir=-1,windspeedmph=3.1,windgustmph=117.7,windgustdir=-1,windspdmph_avg2m=3.1,winddir_avg2m=0,windgustmph_10m=117.7,windgustdir_10m=-1,humidity=33.6,tempf=75.2,rainin=3.04,dailyrainin=3.04,pressure=98257.75,batt_lvl=4.28,light_lvl=0.28,#

Install the serial node (node-red-node-serialport) npm install node-red-node-serialport or sudo npm install -g npm@2.x npm install node-red-node-serialport@0.0.5 if node.js prior to 4.x (ie v0.10.x and v0.12.x)

Restart Node-RED with.

Check the module in the list node-red-admin list | grep serial

Check available serial ports (/dev/tty.usbserial* on MacOS X, /dev/ttyUSB* ...) with.

Start Node-RED node-red -v

Add a node debug

Add a node function "Clean Data" with the following statements var m=msg.payload; var i=m.indexOf(",")+1; msg.payload=m.substr(i,m.lastIndexOf(",")-i); setTimeout(function { node.status({}); }, 500) node.status({fill:"green",shape:"dot",text:"updated"}); return msg;

Add a node serial "Weather Station" with settings 9600/8/N/1 and split input on charracter /n

Connect them together like that

The flow is : [{"id":"40c61a94.428894","type":"serial-port","z":"65e2c142.626cf8","serialport":"/dev/tty.usbmodem1451","serialbaud":"9600","databits":"8","parity":"none","stopbits":"1","newline":"\\n","bin":"false","out":"char","addchar":false},{"id":"e0626709.612a1","type":"serial in","z":"65e2c142.626cf8","name":"Weather Station","serial":"40c61a94.428894","x":200.5,"y":486.5,"wires":"576a5389.61c4e4"},{"id":"bf78bff1.63e8a","type":"debug","z":"65e2c142.626cf8","name":"","active":true,"console":"false","complete":"payload","x":802.5,"y":488,"wires":[]},{"id":"576a5389.61c4e4","type":"function","z":"65e2c142.626cf8","name":"Clean data","func":"var m=msg.payload;\nvar i=m.indexOf(\",\")+1;\nmsg.payload=m.substr(i,m.lastIndexOf(\",\")-i);\nsetTimeout(function { node.status({}); }, 500)\nnode.status({fill:\"green\",shape:\"dot\",text:\"updated\"});\nreturn msg;","outputs":1,"noerr":0,"x":498.5,"y":486.5,"wires":"bf78bff1.63e8a"}]

Serial for LoRa Libelium
Install the Waspmote PRO IDE from Libelium on your host (link)

Load the sketch SX_02a_TX_LoRa on the LoRa Waspmote. /* * -- [SX_02a] - TX LoRa * * Explanation: This example shows how to configure the semtech * module in LoRa mode and then send packets with plain-text payloads */ // Include this library to transmit with sx1272
 * 1) include 

// define the destination address to send packets uint8_t rx_address = 8;

// status variable int8_t e;

void setup { // Init USB port USB.ON; USB.println(F("SX_02a example")); USB.println(F("Semtech SX1272 module TX in LoRa"));

USB.println(F("")); USB.println(F("Setting configuration:")); USB.println(F(""));

// Init sx1272 module sx1272.ON;

// Select frequency channel e = sx1272.setChannel(CH_10_868); USB.print(F("Setting Channel CH_10_868.\t state ")); USB.println(e);

// Select implicit (off) or explicit (on) header mode e = sx1272.setHeaderON; USB.print(F("Setting Header ON.\t\t state ")); USB.println(e);

// Select mode: from 1 to 10 e = sx1272.setMode(1); USB.print(F("Setting Mode '1'.\t\t state ")); USB.println(e);

// Select CRC on or off e = sx1272.setCRC_ON; USB.print(F("Setting CRC ON.\t\t\t state ")); USB.println(e);

// Select output power (Max, High or Low) e = sx1272.setPower('L'); USB.print(F("Setting Power to 'L'.\t\t state ")); USB.println(e);

// Select the node address value: from 2 to 255 e = sx1272.setNodeAddress(2); USB.print(F("Setting Node Address to '2'.\t state ")); USB.println(e); USB.println; delay(1000); USB.println(F("")); USB.println(F("Sending:")); USB.println(F("")); }

void loop { // Sending packet before ending a timeout e = sx1272.sendPacketTimeout( rx_address, "This_is_a_new_message");

// Check sending status if( e == 0 ) {   USB.println(F("Packet sent OK")); } else {   USB.println(F("Error sending the packet")); USB.print(F("state: ")); USB.println(e, DEC); }  delay(2500); }

Check with the IDE Serial monitor the output of the Waspmote

Load the sketch SX_02b_RX_LoRa on the LoRa Dongle. /*  *  -- [SX_02b] - RX LoRa * Explanation: This example shows how to configure the semtech * module in LoRa mode and then receive packets with plain-text payloads */ // Include this library for transmit with sx1272
 * 1) include 

// status variable int8_t e;

void setup { // Init USB port USB.ON; USB.println(F("SX_02b example")); USB.println(F("Semtech SX1272 module RX in LoRa"));

USB.println(F("")); USB.println(F("Setting configuration:")); USB.println(F(""));

// Init sx1272 module sx1272.ON;

// Select frequency channel e = sx1272.setChannel(CH_10_868); USB.print(F("Setting Channel CH_10_868.\t state ")); USB.println(e);

// Select implicit (off) or explicit (on) header mode e = sx1272.setHeaderON; USB.print(F("Setting Header ON.\t\t state ")); USB.println(e);

// Select mode: from 1 to 10 e = sx1272.setMode(1); USB.print(F("Setting Mode '1'.\t\t state ")); USB.println(e);

// Select CRC on or off e = sx1272.setCRC_ON; USB.print(F("Setting CRC ON.\t\t\t state ")); USB.println(e);

// Select output power (Max, High or Low) e = sx1272.setPower('L'); USB.print(F("Setting Power to 'L'.\t\t state ")); USB.println(e);

// Select the node address value: from 2 to 255 e = sx1272.setNodeAddress(8); USB.print(F("Setting Node Address to '8'.\t state ")); USB.println(e);

delay(1000);

USB.println(F("")); USB.println(F("Receiving:")); USB.println(F("")); }

void loop { // receive packet e = sx1272.receivePacketTimeout(10000);

// check rx status if( e == 0 ) {   USB.println(F("\nShow packet received: "));

// show packet received sx1272.showReceivedPacket; } else {   USB.print(F("\nReceiving packet TIMEOUT, state ")); USB.println(e, DEC); } }

Check with the IDE Serial monitor the output of the USB Dongle

Install the serial node (node-red-node-serialport) npm install node-red-node-serialport or sudo npm install -g npm@2.x npm install node-red-node-serialport@0.0.5 if node.js prior to 4.x (ie v0.10.x and v0.12.x)

Restart Node-RED with.

Check the module in the list node-red-admin list | grep serial

Check available serial ports (/dev/tty.usbserial* on MacOS X, /dev/ttyUSB* ...) with.

Serial for LoRa Nucleo
Install the serial node (node-red-node-serialport) npm install node-red-node-serialport or sudo npm install -g npm@2.x npm install node-red-node-serialport@0.0.5 if node.js prior to 4.x (ie v0.10.x and v0.12.x)

Restart Node-RED with.

Check the module in the list node-red-admin list | grep serial

Check available serial ports (/dev/tty.usbserial* on MacOS X, /dev/ttyUSB* ...) with.

TO BE CONTINUED

UDP for LoRaWAN
Add a node 'udp in' listening the port 8123.

Add a node 'function' "toString" with the following statement: msg.payload=msg.payload.toString; return msg;

Add a node 'debug'.

Connect them together.

Deploy the new flow.

Test the flow with the shell command echo -n -e '{"device"="123456","temperature"=37.2}' > /dev/udp/127.0.0.1/8123

The flow looks like that:

Add a node 'json' into the flow.

Now you can filter and send the output of a process to the flow througth an 'udp in' node.

TO BE CONTINUED

UDP for ESP8266 + PowerSensor


Install the Arduino IDE on your host (link)

Add the "ESP8266 Modules" support with the Tools menu > Board Manager.

Select "NodeMCU 0.9 (ESP8266-12 Module)" in the Tools menu.

Load the following sketch on the ESP8266. /* From http://www.iotfrog.com/en/articles/article/227


 * 1) include 
 * 2) include 

const char* ssid    = "iotdays"; const char* password = "eclipse2016";

//IPAddress ipBroadCast(192,168,1,255); IPAddress ipBroadCast;

unsigned int udpRemotePort = 8123; unsigned int udplocalPort = 8124; const int UDP_PACKET_SIZE = 48;

char udpBuffer[ UDP_PACKET_SIZE]; WiFiUDP udp;

// Setup the Wifi connection void connectWifi { Serial.print("Connecting to "); Serial.println(ssid);

// Try to connect to wifi access point WiFi.begin(ssid, password); while (WiFi.status != WL_CONNECTED) { delay(500); Serial.print("."); } Serial.println(""); Serial.println("WiFi connected"); Serial.print("IP address: "); Serial.println(WiFi.localIP);

ipBroadCast = ~WiFi.subnetMask | WiFi.gatewayIP; }

// Send udp message void udpSend { // TODO read the current value of an analog pin strcpy(udpBuffer, "{\"device\"=\"12348266\",\"load\"=10.2}"); udp.beginPacket(ipBroadCast, udpRemotePort); udp.write(udpBuffer, sizeof(udpBuffer)); udp.endPacket; Serial.print("Broadcast: "); Serial.println(udpBuffer); }

// Setup hardware, serial port, and connect to wifi. void setup { Serial.begin(115200); delay(10); // We start by connecting to a WiFi network connectWifi; Serial.println("Starting UDP"); // set udp port for listen udp.begin(udplocalPort); Serial.print("Local port: "); Serial.println(udp.localPort); }

// LOOP MAIN // Send udp packet each 10 secconds void loop { udpSend; delay (10000); }

Add a node 'udp in' listening the port 8123 with a Buffer output.

Add a node 'function' "toStr" with the following statement: msg.payload=msg.payload.toString; return msg;

Add a node 'debug'.

Connect them together.

Deploy the new flow.

The flow looks like that:

Remark: you can test the flow with this shell command echo -n -e '{"device"="12348266","load"=10}' > /dev/udp/127.0.0.1/8123

The flow is: [{"id":"b32d523b.fd02d8","type":"udp in","z":"a994a424.14415","name":"","iface":"","port":"8123","ipv":"udp4","multicast":"false","group":"","datatype":"utf8","x":147,"y":94,"wires":"1658af1a.ebfd11"},{"id":"f732d1c6.93623","type":"debug","z":"a994a424.14415","name":"","active":true,"console":"false","complete":"payload","x":535,"y":94.5,"wires":[]},{"id":"1658af1a.ebfd11","type":"function","z":"a994a424.14415","name":"toStr","func":"setTimeout(function { node.status({}); }, 1000);\n// The shape property can be: ring or dot.\n// The fill property can be: red, green, yellow, blue or grey\nnode.status({fill:\"green\",shape:\"dot\",text:\"receiving\"}); \nmsg.payload=msg.payload.toString;\nreturn msg;","outputs":1,"noerr":0,"x":336,"y":94,"wires":"f732d1c6.93623"}]

ACR122U NFC Reader
Plug the ACR122U NFC reader in the host and check it with

Install LibNFC (on Debian) sudo apt-get install libnfc-dev libnfc-bin libnfc-examples

sudo nfc-list

Launch nfc-poll and put a NFC card/tag on the reader sudo nfc-poll -v

The ID of the NFC card/tag is prefixed by the string.

Execute the following script :

while true do nfc-poll | grep "UID (NFCID1):" > /tmp/nfc.log done
 * 1) !/bin/sh

Into the Node-RED dashboard, create a new flow.

Add a node tail for reading the tail of the log file containing the card/tags identifiers.

Add a node debug.

Connect the node tail to the node debug.

Deploy the new flow.

Add a node function "Build message" with the following statements: var m=msg.payload; var i=m.indexOf(":"); if(i>=0) { var tid = m.substr(i+1).replace(/\s+/g, ''); msg.payload={} msg.payload.device="acr122u"; msg.payload.nfc=tid; return msg; } return null;

Connect the node tail to the node "Build message" and the node "Build message" to the node debug.

Deploy the new flow.

Edit the node "Build message" and add the 2 following statements in order to display the last tid into the node's status during one second. ...   setTimeout(function { node.status({}); }, 1000); // The shape property can be: ring or dot. // The fill property can be: red, green, yellow, blue or grey node.status({fill:"green",shape:"dot",text:"#"+tid}); return msg; } return null;

Deploy the new flow.

=Install Mosquitto=

On MacOS X brew install mosquitto

On Debian sudo apt-get install mosquitto sudo service mosquitto status sudo apt-get install mosquitto-clients

Test local Mosquitto broker BROKER=localhost mosquitto_sub -d -h $BROKER -t 'iotdays/sensors/#' & mosquitto_pub -d -h $BROKER -t 'iotdays/sensors/gw0001' -m '{"device"="123456","temperature"=37.2}' mosquitto_pub -d -h $BROKER -t 'iotdays/sensors/gw0001' -m '{"device"="345678","temperature"=23.2,"humidity"=50}' mosquitto_pub -d -h $BROKER -t 'iotdays/sensors/gw0002' -m '{"device"="acr122u","nfc"="04ddf0f9232580"}' sleep 1 pkill mosquitto_sub

Test public Mosquitto broker BROKER=test.mosquitto.org mosquitto_sub -d -h $BROKER -t 'iotdays/sensors/#' & mosquitto_pub -d -h $BROKER -t 'iotdays/sensors/gw0001' -m '{"device"="123456","temperature"=37.2}' mosquitto_pub -d -h $BROKER -t 'iotdays/sensors/gw0001' -m '{"device"="345678","temperature"=23.2,"humidity"=50}' mosquitto_pub -d -h $BROKER -t 'iotdays/sensors/gw0002' -m '{"device"="acr122u","nfc"="04ddf0f9232580"}' pkill mosquitto_sub

=Using MQTT for collecting sensors data=

In the flows defined above, add a node mqtt out with a new MQTT broker localhost:1883 and a topic iotdays/sensors/gwXXXX where gwXXXX is the name of the host.

Check the publishing with BROKER=localhost mosquitto_sub -d -h $BROKER -t 'iotdays/sensors/#' &

Create a new flow for subscribing messages from the Mosquitto MQTT broker.

Add a node debug for displaying the full message.

Add a node mqtt in with with a new MQTT broker localhost:1883 and a topic iotdays/sensors#

Connect the node mqtt in to the node debug.

Deploy the new flow.

=Install InfluxDB=

=Populate InfluxDB from Node-RED=

influx

Enter the following InfluxDB statements create database iotdb use iotdb show measurements show series select * from temperature select * from humidity select * from nfc select * from ppm

In the Node-RED dashboard, create a new flow to collect sensors data from the MQTT server.

Install the Node-RED InfluxDB node.

Restart Node-RED.

Add a node mqtt in "MQTT Broker" for subscribing to the MQTT server (broker is test.mosquitto.org or localhost topic is iotdays/sensors/#)

Add a node function "Tranform into time series" to format the JSON messages into time series messages.

Add a node influxdb out "IoT Database" with the InfluxDB server.

Connect the node "MQTT Broker" to node "Tranform into time series" and to the node "Tranform into time series" to node "IoT Database".

The flow looks like than:

Extra for Arduino and Weather Shield
Add a node influxdb out

Add a node function with the following statements: var result = {}; msg.payload.split(',').forEach(function(x){   var arr = x.split('=');    arr[1] && (result[arr[0]] = parseFloat(arr[1])); }); msg.payload=result; return msg;

Connect the 2 additionnal nodes like that:

The flow is: [{"id":"57400c2a.74f84c","type":"influxdb","z":"65e2c142.626cf8","hostname":"localhost","port":"8086","database":"Weather","name":""},{"id":"40c61a94.428894","type":"serial-port","z":"65e2c142.626cf8","serialport":"/dev/tty.usbmodem1451","serialbaud":"9600","databits":"8","parity":"none","stopbits":"1","newline":"\\n","bin":"false","out":"char","addchar":false},{"id":"e0626709.612a1","type":"serial in","z":"65e2c142.626cf8","name":"Weather Station","serial":"40c61a94.428894","x":200.5,"y":486.5,"wires":"576a5389.61c4e4"},{"id":"bf78bff1.63e8a","type":"debug","z":"65e2c142.626cf8","name":"","active":true,"console":"false","complete":"payload","x":802.5,"y":488,"wires":[]},{"id":"576a5389.61c4e4","type":"function","z":"65e2c142.626cf8","name":"Clean data","func":"var m=msg.payload;\nvar i=m.indexOf(\",\")+1;\nmsg.payload=m.substr(i,m.lastIndexOf(\",\")-i);\nsetTimeout(function { node.status({}); }, 500)\nnode.status({fill:\"green\",shape:\"dot\",text:\"updated\"});\nreturn msg;","outputs":1,"noerr":0,"x":498.5,"y":486.5,"wires":"86b78cd4.e2d548","bf78bff1.63e8a"},{"id":"86b78cd4.e2d548","type":"function","z":"65e2c142.626cf8","name":"Format as InfluxDB line","func":"var result = {};\nmsg.payload.split(',').forEach(function(x){\n   var arr = x.split('=');\n    arr[1] && (result[arr[0]] = parseFloat(arr[1]));\n});\nmsg.payload=result;\nreturn msg;","outputs":1,"noerr":0,"x":667.5,"y":610,"wires":"1a959fff.cd4668","bf78bff1.63e8a"},{"id":"1a959fff.cd4668","type":"influxdb out","z":"65e2c142.626cf8","influxdb":"57400c2a.74f84c","name":"","measurement":"weather","x":908.5,"y":704,"wires":[]}]

Start the InfluxDB shell influx

Create the database create database Weather use Weather show measurements select * from weather

=Install Grafana (2.6)=

On Debian
wget https://grafanarel.s3.amazonaws.com/builds/grafana_2.6.0_amd64.deb sudo dpkg -i xzvf grafana_2.6.0_amd64.deb sudo service grafana-server start

=Visualize sensors data into the Grafana dashboard=

Browse http://localhost:3000

Login with username:password admin:admin

Create a new dashboard

Add a data source (InfluxDB 0.9) iotdb

Enable dashboard edition

Add a row

Add a graph in the row

Edit the graph

Edit the query for the graph for the data source

Duplicate the graph and change the measurement in the query

Duplicate the graph and aggregate (mean) all measurements of the same type (temperature, humidity).

Set the timeline and the refresh time.

See Grafana

The result should look like that

=Install Apache Spark=

See Spark

=Process a realtime stream of sensors data=

import org.eclipse.paho.client.mqttv3._ import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence

import org.apache.spark.storage.StorageLevel import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.mqtt._ import org.apache.spark.SparkConf

val brokerUrl = "tcp://localhost:1883" val topic = "iotdays/sensors/#" val ssc = new StreamingContext(sc, Seconds(60)) val sensordatas = MQTTUtils.createStream(ssc, brokerUrl, topic, StorageLevel.MEMORY_ONLY_SER_2)

TODO

ssc.start ssc.awaitTermination