โปรเจกต์ ส่งข้อมูล Arduino ผ่าน MQTT เพื่อทำการ Plot ใน Grafana

การบันทึกข้อมูล (Logging data) เป็นเรื่องที่สนุกมาก ผมต้องการบันทึกข้อมูลที่เก็บรวบรวมจาก Arduino MKR WIFI 1010 และต้องการแสดงผลข้อมูล (visualize) ผ่านบริการที่ผมสามารถตรวจสอบออนไลน์ได้จากทุกที่ ผมเลือกใช้ MQTT สำหรับการเก็บข้อมูลในเครื่อง (local storage) และใช้ Grafana Cloud สำหรับการแสดงผลข้อมูลบนเว็บ ผมได้พูดถึงรายละเอียดเกี่ยวกับการเก็บข้อมูลจริงไว้ใน project highlight ก่อนหน้านี้ วิธีการทำนี้ จำเป็นต้องใช้ Arduino ที่สามารถเชื่อมต่อ WIFI ได้

ผมทำสิ่งนี้โดยเฉพาะเพื่อบันทึกข้อมูลสภาพอากาศด้วย BME280 แต่แนวทางเดียวกันนี้ก็น่าจะใช้กับ Sensor ตัวอื่นๆ ได้เช่นกัน ผมได้พูดถึงการตั้งค่าทุกอย่างไว้ใน website ของผม และผมได้เก็บโค้ดไว้ใน github repository ในที่นี้ผมจะอธิบายขั้นตอนการกำหนดค่าทุกอย่างเพื่อส่งข้อมูลของผมไปยัง Grafana Cloud โดยเริ่มแรกผมได้รับแรงบันดาลใจมาจาก วิธีการทำนี้

สรุปสั้นๆ (TL;DR): ติดตั้ง MQTT และ Prometheus บนอุปกรณ์ที่คุณต้องการส่งข้อมูลไปให้, สร้างบัญชี Grafana Cloud ฟรี, ไปที่ github repository นี้ แล้วตั้งค่าโค้ดและบริการบันทึกข้อมูลเพื่อส่งข้อมูลจาก Arduino ของคุณไปยัง MQTT server ในเครื่อง และส่งต่อไปยัง Grafana Cloud

Part 1: Arduino MQTT Setup

ก่อนอื่นเราต้องส่งข้อมูลจาก Arduino ผ่าน MQTT ไปยังอุปกรณ์อีกเครื่องที่รัน MQTT เพื่อรับข้อมูล มีคู่มือการใช้งาน Arduino ที่ดีในเรื่องนี้สำหรับ Uno WIFI Rev 2 เริ่มต้นด้วยการ include client และสิ่งที่จำเป็นในการเชื่อมต่อ WIFI สำหรับผมจะมีลักษณะดังนี้:

#include "PubSubClient.h" 
#include <WiFiNINA.h>

ตอนนี้ให้ตั้งค่า Arduino เพื่อส่งข้อมูลที่ส่วนต้นของไฟล์ .ino ของคุณโดยการแก้ไขโค้ดด้านล่าง และกำหนด "topics" ซึ่งจะเป็นตัวระบุข้อมูลของคุณ

 // MQTT properties
const char* mqtt_server = "192.168.50.100";  // IP address ของ MQTT broker
const char* temp_topic = "outdoor/weather/temperature";
const char* humid_topic = "outdoor/weather/humidity";
const char* pressure_topic = "outdoor/weather/pressure";
const char* altitude_topic = "outdoor/weather/altitude";
const char* distance_topic = "outdoor/weather/distance";
const char* mqtt_username = MQTT_user; // MQTT username
const char* mqtt_password = MQTT_pass; // MQTT password
const char* clientID = "arduino"; // MQTT client ID

หลังจาก Arduino เชื่อมต่อกับ WIFI แล้ว เราจะเชื่อมต่อไปยัง MQTT server ที่ระบุไว้ด้านบน:

WiFiClient wifiClient;
// เชื่อมต่อไปยังคอมพิวเตอร์ที่รัน MQTT, 
// ซึ่งกำลังรอรับข้อมูลอยู่ที่ port 1883 
PubSubClient client(mqtt_server, 1883, wifiClient);

ตอนนี้เราจะกำหนดฟังก์ชันการทำงานบางอย่างที่ใช้ในการเริ่ม WIFI และเชื่อมต่อกับ MQTT server

void connect_MQTT(){
 Serial.print("Connecting to ");
 Serial.println(ssid);
 WiFi.begin(ssid, pass);
 // รอจนกว่าการเชื่อมต่อจะได้รับการยืนยันก่อนจะดำเนินการต่อ
 while (WiFi.status() != WL_CONNECTED) {
   delay(500);
   Serial.print(".");
 }
 // สำหรับการ Debug - แสดง IP Address
 Serial.println("WiFi connected");
 Serial.print("IP address: ");
 Serial.println(WiFi.localIP());
 // เชื่อมต่อกับ MQTT Broker
 // client.connect จะคืนค่า boolean เพื่อให้เราทราบว่าการเชื่อมต่อสำเร็จหรือไม่
 if (client.connect(clientID, mqtt_username, mqtt_password)) {
   Serial.println("Connected to MQTT Broker!");
 }
 else {
   Serial.println("Connection to MQTT Broker failed...");
 }
}

ใน Loop หลัก ให้เชื่อมต่อกับ MQTT server และส่งข้อมูลที่รวบรวมมา ข้อมูลที่ส่งจะต้องอยู่ในรูปแบบ String

connect_MQTT();
 delay(10); // delay นี้เพื่อให้แน่ใจว่า client.publish จะไม่ไปขัดแย้งกับการเรียกใช้ client.connect
 // Publish ข้อมูลทั้งหมดไปยัง MQTT Broker
 if (client.publish(temp_topic, String(bmeData[0]).c_str())) {
   delay(10); // delay นี้เพื่อให้แน่ใจว่าการอัปโหลดข้อมูล BME เรียบร้อยดี
   client.publish(humid_topic, String(bmeData[3]).c_str());
   delay(10); // delay นี้เพื่อให้แน่ใจว่าการอัปโหลดข้อมูล BME เรียบร้อยดี
   client.publish(altitude_topic, String(bmeData[2]).c_str());
   delay(10); // delay นี้เพื่อให้แน่ใจว่าการอัปโหลดข้อมูล BME เรียบร้อยดี
   client.publish(pressure_topic, String(bmeData[1]).c_str());
   delay(10);
   client.publish(distance_topic, String(US100_distance).c_str());
   delay(10);
   Serial.println("Weather data sent!");
 }
 // client.publish จะคืนค่า boolean ขึ้นอยู่กับว่าส่งสำเร็จหรือไม่
 // หากข้อความส่งไม่สำเร็จ เราจะเข้าสู่โหมด sleep และลองใหม่อีกครั้งใน 5 นาที
 else {
   Serial.println("Weather data failed to send. Going to sleep and will just keep trying.");
 } 

ตัดการเชื่อมต่อและปิด WIFI หลังจากส่งข้อมูลผ่าน MQTT เรียบร้อยแล้ว ขั้นตอนนี้อาจไม่จำเป็น แต่ผมต้องการประหยัดพลังงานสำหรับงานของผม

client.disconnect();  // ตัดการเชื่อมต่อจาก MQTT broker
WiFi.end(); // ปิด wifi ก่อนเข้าสู่โหมด sleep 

Part 2: MQTT Server Setup

เราจำเป็นต้องส่งข้อมูลไปที่ไหนสักแห่ง ผมใช้ Raspberry Pi ที่เปิดทิ้งไว้ตลอดเวลาเพื่อทำเป็น MQTT server พร้อมที่เก็บข้อมูลเพิ่มเติมสำหรับบันทึกข้อมูล แต่ผมคิดว่าวิธีนี้ใช้ได้กับเครื่อง Linux อื่นๆ รวมถึง Mac และ Windows ด้วย ในที่นี้ผมจะใช้ Unix

อันดับแรก เราต้องติดตั้ง MQTT และ Prometheus:

• ติดตั้ง MQTT บนอุปกรณ์ที่คุณต้องการส่งข้อมูลไปให้ ในที่นี้ผมใช้ mosquitto

• ติดตั้ง Prometheus บนอุปกรณ์เครื่องเดียวกัน

ตรวจสอบให้แน่ใจว่า MQTT และ Prometheus client กำลังทำงานอยู่ ผมสามารถตั้งค่าให้ MQTT เริ่มทำงานอัตโนมัติทุกครั้งที่เปิดเครื่องได้ง่ายๆ แต่สำหรับ Prometheus ดูเหมือนจะซับซ้อนกว่า ในตอนนี้ผมจึงใช้วิธีเริ่มรัน Prometheus client ด้วยตนเองหาก Raspberry Pi ถูกรีเซ็ตด้วยเหตุผลบางประการ

เมื่อ Prometheus และ MQTT ทำงานบนอุปกรณ์แล้ว Arduino ก็สามารถส่งข้อมูลไปยังอุปกรณ์นั้นได้ สำเร็จ!

เราจำเป็นต้องบอกให้อุปกรณ์รอฟังข้อมูลด้วย MQTT และส่งต่อข้อมูลไปยัง Prometheus จากนั้น Grafana Cloud จะนำข้อมูลจาก Prometheus มาสร้างกราฟ

บนเครื่องที่เป็น MQTT server เราสามารถตั้งค่าให้ข้อมูลถูกส่งต่อไปยัง Prometheus ได้ นี่คือส่วนที่ โค้ด python และ ตัวอย่าง นี้จะเข้ามามีบทบาท

ผมเลือกที่จะทำสิ่งนี้ด้วย Python เพราะผมพบ MQTT และ Prometheus client สำหรับมันได้อย่างรวดเร็ว และผมสามารถสร้างตาราง SQLite เพื่อเก็บข้อมูลไว้ในเครื่องได้ แต่ผมคิดว่ายังมีตัวเลือกภาษาโปรแกรมอื่นๆ อีกมากมาย

เพื่อให้โค้ด Python ทำงานได้ อันดับแรกให้ติดตั้ง paho mqtt client และ python Prometheus client ตอนนี้เราสามารถเริ่มเก็บและส่งข้อมูลในสคริปต์ Python ได้แล้ว ผมต้องการข้อมูลสำรองไว้ในเครื่อง ดังนั้นผมจึงเขียนข้อมูล MQTT ลงในตาราง SQLite บนดิสก์ขนาดใหญ่ที่เชื่อมต่อกับ Raspberry Pi ของผม วิธีนี้จะทำให้ผมมีบันทึกข้อมูลในเครื่องที่เป็นอิสระจากระยะเวลาการเก็บรักษาข้อมูล (retention period) ของ Grafana Cloud คุณอาจจะรัน Grafana ในเครื่องเองก็ได้เพื่อไม่ต้องกังวลเรื่องการเก็บรักษาข้อมูล แต่ตอนนี้ผมชอบที่จะมีสำเนาถาวรไว้ในดิสก์มากกว่า

ตอนนี้ผมจะอธิบายโค้ด Python ที่ใช้บันทึกและส่งข้อมูล MQTT (นี่คือไฟล์ getWeatherData.py ใน github ของผมเพื่อใช้อ้างอิง แต่คุณสามารถสร้างไฟล์ของคุณเองได้ตามสะดวก) ขั้นแรก ให้ include library ต่างๆ: (ผมใช้ dotenv เพื่อเก็บพารามิเตอร์ในไฟล์ .env เฉพาะสำหรับการตั้งค่าของผม ซึ่งเป็นขั้นตอนเสริม)

import paho.mqtt.client as mqtt
from prometheus_client import start_http_server, Summary, Gauge
import time
import datetime
import sqlite3
import os
from dotenv import load_dotenv
from sqlite3 import Error
load_dotenv()

ตอนนี้ตั้งค่าข้อมูล Prometheus ที่จะส่งไปยัง Grafana Cloud ผมมีตัวแปร Gauge สี่ตัวเพื่อติดตามข้อมูลสภาพอากาศแบบอนุกรมเวลา (timeseries)

############ แก้ไขส่วนนี้สำหรับตัวแปร prometheus ของคุณ #############
# มีตัวแปรหลายประเภทที่ prometheus ยอมรับ,
# เนื่องจากข้อมูลเหล่านี้เป็นการวัดจุดข้อมูลในแต่ละช่วงเวลา Gauge จึงเป็นตัวเลือกที่เหมาะสม
temp = Gauge('temperature', 'Weather Station Temperature')
hum = Gauge('humidity', 'Weather Station Humidity')
alt = Gauge('altitude', 'Weather Station Altitude')
pres = Gauge('pressure', 'Weather Station Pressure')
#####################################################################

จากนั้นกำหนดค่าพารามิเตอร์ MQTT เพื่อรับข้อมูลที่มาจาก Arduino

############ แก้ไขส่วนนี้สำหรับการตั้งค่า mqtt ของคุณ ##############
MQTT_ADDRESS = os.environ.get("MQTT_ADDRESS") 
MQTT_USER = os.environ.get("MQTT_USER") 
MQTT_PASSWORD = os.environ.get("MQTT_PASSWORD") 
MQTT_TOPIC = 'outdoor/weather/temperature'
MQTT_REGEX = 'home/([^/]+)/([^/]+)'
MQTT_CLIENT_ID = 'raspberrypi'
########################################################  

ข้ามมาดูในส่วนของโปรแกรม "main" ซึ่งจะมีลักษณะดังนี้:

def main():
   # เริ่มการทำงานของ Prometheus server
   start_http_server(8000)
   # ตั้งค่าฐานข้อมูล SQLlite
   databasePath = setup_database()
   createTable(databasePath)
   # เริ่ม mqtt client
   mqtt_client = mqtt.Client(MQTT_CLIENT_ID)
   mqtt_client.username_pw_set(MQTT_USER, MQTT_PASSWORD)
   # ระบุโปรแกรมที่จะเรียกใช้เมื่อตรงตามเงื่อนไขของ mqtt
   mqtt_client.on_connect = on_connect
   mqtt_client.on_message = on_message
   # ตั้งค่า mqtt บน port
   mqtt_client.connect(MQTT_ADDRESS, 1883)
   # ให้ทำงานตลอดไป
   mqtt_client.loop_forever()
if __name__ == '__main__':
   main()

ดังนั้นเมื่อโปรแกรมเริ่มต้น main จะถูกเรียกใช้ จากนั้นจะเริ่ม Prometheus, สร้างฐานข้อมูล SQLite, เริ่มรอฟังข้อมูล MQTT แล้ว Loop ทำงานตลอดไป ผมจะเริ่มจากโค้ด MQTT ก่อน

อันดับแรก ฟังก์ชัน on_connect จะทำงานเมื่อ Arduino เชื่อมต่อกับ MQTT:

def on_connect(client, userdata, flags, rc):
   """ ทำงานเมื่อ client เชื่อมต่อ """
   # มีรหัสการเชื่อมต่อ mqtt ต่างๆ ซึ่งจำเป็นเฉพาะเมื่อต้องการ debugging เท่านั้น
   print('Connected with result code ' + str(rc)) 
   # Subscribe mqtt ตามตัวแปรต่อไปนี้
   client.subscribe([('outdoor/weather/temperature',1),('outdoor/weather/humidity',1),
                     ('outdoor/weather/altitude',1),('outdoor/weather/pressure',1)])

ฟังก์ชันนี้จะทำหน้าที่ Subscribe ข้อมูลจาก Arduino ที่ตั้งค่าไว้ก่อนหน้านี้ จากนั้นเมื่อได้รับข้อมูลผ่าน MQTT ฟังก์ชัน on_message จะทำงาน:

def on_message(client, userdata, msg):
   """ ทำงานเมื่อได้รับข้อความ MQTT """ 
   process_request(msg)

ซึ่งจะเรียกฟังก์ชันเพื่อประมวลผลข้อความ นี่คือส่วนสำคัญของโค้ด:

 def process_request(msg):
   """ฟังก์ชันสำหรับอ่านข้อมูลที่เผยแพร่ผ่าน mqtt"""
   timeVal = datetime.datetime.now()
   # แสดงเวลาเพื่อตรวจสอบว่ากำลังทำงานอยู่ 
   print("Current Time:",datetime.datetime.now())
   # แสดงข้อความ 
   msgStr = str(msg.payload)
   goodMsg = msgStr[2:-1]
   print(msg.topic + ' ' + goodMsg)
   # ตรวจสอบให้แน่ใจว่าเราเชื่อมโยง logs ของ prometheus กับตัวแปร mqtt ที่ถูกต้อง
   # สิ่งนี้จะเผยแพร่ตัวแปร mqtt ไปยัง prometheus gauge  
   # และบันทึกข้อมูลลงในตาราง SQLite ด้วย 
   if msg.topic ==  'outdoor/weather/temperature':
       temp.set(msg.payload)
       sqlMsg = (str(timeVal),str(goodMsg),None,None,None);
       insert_database(sqlMsg)
   elif msg.topic == 'outdoor/weather/humidity':
       hum.set(msg.payload)
       sqlMsg = (str(timeVal),None,str(goodMsg),None,None);
       insert_database(sqlMsg)
   elif msg.topic == 'outdoor/weather/altitude':
       alt.set(msg.payload)
       sqlMsg = (str(timeVal),None,None,None,str(goodMsg));
       insert_database(sqlMsg)
   elif msg.topic == 'outdoor/weather/pressure':
       pres.set(msg.payload)
       sqlMsg = (str(timeVal),None,None,str(goodMsg),None);
       insert_database(sqlMsg)
   elif msg.topic == 'outdoor/weather/distance':
       dist.set(msg.payload)
   else:
       print('Incorrect topic')

ผมมีการจัดรูปแบบ String เล็กน้อยที่นี่ แต่สุดท้ายผมก็ไล่ดู Topic ต่างๆ ที่ส่งมาจาก Arduino ผ่าน MQTT และส่งข้อมูลไปยัง Prometheus gauge ที่ตรงกันตามลำดับ ข้อความ MQTT แต่ละข้อความจะมี Topic และ Payload โดย Topic คือ Tag ที่เราสร้างไว้บน Arduino และ Payload คือค่าจริงๆ โดยการตรวจสอบ Topic ของแต่ละข้อความ ผมจะใช้คำสั่ง .set เพื่อส่งข้อมูลไปยัง Prometheus gauge ที่กำหนดไว้ก่อนหน้านี้ และผมจะใส่ข้อมูลลงในคอลัมน์ที่ถูกต้องในตาราง SQLite

สุดท้าย ผมทำขั้นตอนเสริมในการบันทึกข้อมูลลงในตาราง SQLite ด้วยสิ่งนี้:

def createTable(databasePath):
   """สร้างตาราง SQLite หากยังไม่มี"""
   sql_create_weatherData_table = 'CREATE TABLE IF NOT EXISTS weatherData (id integer PRIMARY KEY,timedat text NOT NULL, temperature text, humidity text, pressure text, altitude text);'
   conn = setup_database()
   # สร้างตาราง
   if conn is not None:
       c = conn.cursor()
       c.execute(sql_create_weatherData_table)
   else:
       print("Error! cannot create the database connection.")
def setup_database():
   """ตั้งค่าฐานข้อมูลสำหรับเก็บข้อมูลที่ส่งโดย mqtt"""
   databasePath =  os.environ.get("SQL_PATH")
   databasePath = str(databasePath)+"/mqtt.sqlite"
   conn = None
   try:
       conn = sqlite3.connect(databasePath)
   except Error as e:
       print(e)
   return conn
def insert_database(sqlMsg):
   """บันทึกข้อมูลสภาพอากาศลงในฐานข้อมูล"""
   databasePath =  os.environ.get("SQL_PATH")
   databasePath = str(databasePath)+"/mqtt.sqlite"
   conn = sqlite3.connect(databasePath)
   sql  ='INSERT INTO weatherData (timedat, temperature, humidity, pressure, altitude) values(?,?,?,?,?)'
   cur = conn.cursor()
   cur.execute(sql, sqlMsg) 
   conn.commit()

ฟังก์ชันเหล่านี้ถูกเรียกใช้ใน main() และ process_request(msg) เพื่อสร้างหรือเชื่อมต่อกับฐานข้อมูลในเครื่องอย่างปลอดภัย แล้วจึงจัดเก็บข้อมูล MQTT

ตอนนี้เราก็ได้รับข้อมูล MQTT และส่งมันออกมาในรูปแบบ Prometheus gauge แล้ว

Part 3: Configure Grafana

ส่วนที่ยากที่สุดผ่านไปแล้ว ตอนนี้เราแค่ต้องบอก Grafana Cloud ว่ากำลังมีการส่งข้อมูลไปให้

ในการส่งข้อมูล Prometheus ไปยัง Grafana Cloud ให้สร้างบัญชี Grafana Cloud ฟรีเพื่อรับข้อมูลที่บันทึกไว้ และตั้งค่าเพื่อรับข้อมูล ในการทำสิ่งนี้ ให้ไปที่ตัวเลือก Prometheus ใน Grafana stack ของคุณ แล้วจดรายละเอียดการตั้งค่า (remote_write, user name, password) สำหรับการส่งข้อมูล

ในไฟล์ติดตั้ง Prometheus (สำหรับผมจะอยู่ในไฟล์ติดตั้งที่แตกซิปออกมา) คุณต้องตั้งค่า prometheus.yml เพื่อส่งข้อมูลไปยัง Grafana Cloud ผ่าน remote_write

เมื่อเสร็จเรียบร้อยแล้วและ Prometheus กับ MQTT กำลังทำงานอยู่ ให้รันโค้ด Python ของคุณ สำหรับผมคือการพิมพ์ python3 getWeatherData.py ใน Unix terminal ใน Directory ที่โค้ดอยู่ ตอนนี้คุณควรเริ่มเห็นข้อมูล MQTT ถูกพิมพ์ออกมาในหน้าต่าง Terminal นี้ Arduino กำลังส่งข้อมูล Sensor ผ่าน MQTT! และหวังว่า Prometheus จะส่งข้อมูลไปยังบัญชี Grafana Cloud ของคุณด้วยเช่นกัน

ตอนนี้เราจะตั้งค่าข้อมูลบน Grafana Cloud เพื่อแสดงผลข้อมูลของคุณใน Dashboard ที่คุณสามารถเข้าถึงได้จากทุกที่หลังจากเข้าสู่ระบบ ผมทำสิ่งนี้โดยการล็อกอินเข้า Grafana แล้วไปที่การสร้าง Dashboard, เพิ่ม Panel เปล่า (empty panel) จากนั้นคุณควรจะสามารถค้นหาตัวแปรของคุณที่ติดป้ายชื่อเป็น gauge ในโค้ด Python ก่อนหน้านี้ได้ ภายใต้ Prometheus ในฐานะแหล่งข้อมูล (data source) และภายใต้ metrics

ตอนนี้คุณสามารถสร้าง Dashboard เหมือนกับที่ผมมีสำหรับข้อมูลสภาพอากาศได้แล้ว:

สำเร็จ!