一、項(xiàng)目概述
隨著物聯(lián)網(wǎng)（IoT）的快速發(fā)展，邊緣計(jì)算的應(yīng)用越來越廣泛。邊緣計(jì)算可以將數(shù)據(jù)處理和分析推向離數(shù)據(jù)源更近的地方，從而降低延遲，提高效率。本文將介紹如何利用樹莓派構(gòu)建一個(gè)多協(xié)議邊緣計(jì)算網(wǎng)關(guān)，以支持各種傳感器和設(shè)備的連接和數(shù)據(jù)傳輸。
該項(xiàng)目的主要目標(biāo)是實(shí)現(xiàn)一個(gè)邊緣網(wǎng)關(guān)，能夠通過不同的通信協(xié)議（如MQTT、HTTP、CoAP等）采集和處理來自傳感器的數(shù)據(jù)，并將這些數(shù)據(jù)發(fā)送至云端或本地?cái)?shù)據(jù)庫進(jìn)行存儲(chǔ)和分析。
二、系統(tǒng)架構(gòu)
1. 硬件選型
主控單元：樹莓派4B
傳感器：溫度傳感器（如DHT11）、濕度傳感器、運(yùn)動(dòng)傳感器等
其他模塊：Wi-Fi和藍(lán)牙模塊（樹莓派4B內(nèi)置）
2. 通信協(xié)議
MQTT：用于輕量級(jí)的設(shè)備間消息傳遞。
HTTP/HTTPS：用于與云服務(wù)或Web應(yīng)用程序的交互。
CoAP：用于低功耗設(shè)備的通信。
3. 技術(shù)棧
操作系統(tǒng)：Raspberry Pi OS
編程語言：Python
數(shù)據(jù)庫：SQLite（用于本地?cái)?shù)據(jù)存儲(chǔ)）
中間件：Eclipse Mosquitto（MQTT代理）
云平臺(tái)：AWS IoT（可選）
三、環(huán)境搭建
1. 安裝Raspberry Pi OS
下載Raspberry Pi Imager并安裝。
使用Raspberry Pi Imager將Raspberry Pi OS寫入SD卡。
將SD卡插入樹莓派，連接顯示器和電源，啟動(dòng)樹莓派。
2. 更新系統(tǒng)
運(yùn)行以下命令更新系統(tǒng)：
sudo apt updatesudo apt upgrade -y
3. 安裝Python和依賴庫
樹莓派通常預(yù)裝Python，但我們需要安裝一些額外的庫。
sudo apt install python3 python3-pip -ysudo pip3 install paho-mqtt flask
4. 安裝并配置Eclipse Mosquitto
sudo apt install mosquitto mosquitto-clients -y
啟動(dòng)Mosquitto服務(wù)：
sudo systemctl start mosquittosudo systemctl enable mosquitto
5. 安裝SQLite
sudo apt install sqlite3 libsqlite3-dev -y
四、代碼實(shí)現(xiàn)
1. 數(shù)據(jù)采集模塊
在上面的代碼中，我們已經(jīng)完成了讀取DHT11傳感器數(shù)據(jù)并將其發(fā)布到MQTT代理的功能。接下來，我們需要將此腳本保存并運(yùn)行。
保存并運(yùn)行數(shù)據(jù)采集腳本
-
在樹莓派上創(chuàng)建一個(gè)Python腳本文件，例如
sensor_data.py。nano sensor_data.py 將以下代碼粘貼到文件中：
import Adafruit_DHT
import paho.mqtt.client as mqtt
import time
\# 設(shè)置傳感器類型和GPIO引腳
sensor = Adafruit_DHT.DHT11
pin = 4 # GPIO引腳編號(hào)
\# MQTT設(shè)置
mqtt_broker = "localhost" # 本地Mosquitto代理
mqtt_topic = "sensor/data"
\# 創(chuàng)建MQTT客戶端
client = mqtt.Client()
client.connect(mqtt_broker)
while True:
# 讀取傳感器數(shù)據(jù)
humidity, temperature = Adafruit\_DHT.read\_retry(sensor, pin)
if humidity is not None and temperature is not None:
# 構(gòu)建消息
message = f"Temperature: {temperature}°C, Humidity: {humidity}%"
print(message)
# 發(fā)布到MQTT
client.publish(mqtt_topic, message)
else:
print("Failed to retrieve data from the sensor.")
# 每10秒讀取一次數(shù)據(jù)
time.sleep(10)
保存并退出編輯器（在nano中按
CTRL + X，然后按Y確認(rèn)保存）。-
運(yùn)行Python腳本：
python3 sensor_data.py
2. 創(chuàng)建Web服務(wù)模塊
接下來，我們將創(chuàng)建一個(gè)簡(jiǎn)單的Flask Web服務(wù)，用于展示傳感器數(shù)據(jù)。
創(chuàng)建Flask應(yīng)用
-
創(chuàng)建一個(gè)新的Python文件，例如
app.py：nano app.py 將以下代碼粘貼到文件中：
from flask import Flask, jsonify
import sqlite3
app = Flask(__name__)
# 數(shù)據(jù)庫連接函數(shù)
def get_db_connection():
conn = sqlite3.connect('sensor_data.db')
conn.row_factory = sqlite3.Row
return conn
@app.route('/data', methods=['GET'])
def get_data():
conn = get_db_connection()
data = conn.execute('SELECT * FROM readings ORDER BY id DESC').fetchall()
conn.close()
return jsonify([dict(row) for row in data])
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000)
- 保存并退出編輯器。
3. 數(shù)據(jù)存儲(chǔ)模塊
為了將傳感器數(shù)據(jù)存儲(chǔ)到SQLite數(shù)據(jù)庫中，我們需要在數(shù)據(jù)采集模塊中添加數(shù)據(jù)庫寫入的邏輯。
修改數(shù)據(jù)采集模塊以支持SQLite
在 sensor_data.py 中，添加數(shù)據(jù)庫寫入功能：
import Adafruit_DHT
import paho.mqtt.client as mqtt
import time
import sqlite3
# 設(shè)置傳感器類型和GPIO引腳
sensor = Adafruit_DHT.DHT11
pin = 4 # GPIO引腳編號(hào)
# MQTT設(shè)置
mqtt_broker = "localhost" # 本地Mosquitto代理
mqtt_topic = "sensor/data"
# 創(chuàng)建MQTT客戶端
client = mqtt.Client()
client.connect(mqtt_broker)
# 數(shù)據(jù)庫連接函數(shù)
def get_db_connection():
conn = sqlite3.connect('sensor_data.db')
return conn
# 創(chuàng)建表（如果不存在）
conn = get_db_connection()
conn.execute('''
CREATE TABLE IF NOT EXISTS readings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
temperature REAL,
humidity REAL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
''')
conn.close()
while True:
# 讀取傳感器數(shù)據(jù)
humidity, temperature = Adafruit_DHT.read_retry(sensor, pin)
if humidity is not None and temperature is not None:
# 保存數(shù)據(jù)到SQLite數(shù)據(jù)庫
conn = get_db_connection()
conn.execute('INSERT INTO readings (temperature, humidity) VALUES (?, ?)',
(temperature, humidity))
conn.commit()
conn.close()
# 構(gòu)建消息
message = f"Temperature: {temperature}°C, Humidity: {humidity}%"
print(message)
# 發(fā)布到MQTT
client.publish(mqtt_topic, message)
else:
print("Failed to retrieve data from the sensor.")
# 每10秒讀取一次數(shù)據(jù)
time.sleep(10)
四、代碼實(shí)現(xiàn)
4. 啟動(dòng)Web服務(wù)
在終端中啟動(dòng)Flask Web服務(wù)，以便我們可以通過瀏覽器訪問傳感器數(shù)據(jù)。
-
運(yùn)行Flask應(yīng)用：
python3 app.py -
在瀏覽器中訪問
http://<樹莓派的IP地址>:5000/data，你應(yīng)該能夠看到從傳感器采集到的溫度和濕度數(shù)據(jù)的JSON格式輸出。這些數(shù)據(jù)是從SQLite數(shù)據(jù)庫中讀取的。例如，輸出可能如下所示：
[ {"id": 1, "temperature": 22.0, "humidity": 45.0, "timestamp": "2024-08-04 12:01:22"}, {"id": 2, "temperature": 23.0, "humidity": 50.0, "timestamp": "2024-08-04 12:01:32"}, ... ]
5. 測(cè)試MQTT功能
為了驗(yàn)證MQTT功能的正常工作，可以使用MQTT客戶端訂閱傳感器數(shù)據(jù)主題。
-
在另一個(gè)終端中，使用Mosquitto客戶端進(jìn)行訂閱：
mosquitto_sub -h localhost -t sensor/data 你應(yīng)該能夠看到來自傳感器的數(shù)據(jù)實(shí)時(shí)打印在終端中。
五、項(xiàng)目總結(jié)
在本項(xiàng)目中，我們成功地使用樹莓派構(gòu)建了一個(gè)多協(xié)議的邊緣計(jì)算網(wǎng)關(guān)，涉及以下關(guān)鍵步驟：
硬件和軟件環(huán)境搭建：我們選擇了樹莓派4B作為控制單元，并安裝了Raspberry Pi OS。通過安裝Python及其相關(guān)庫，確保了系統(tǒng)的正常運(yùn)行。
數(shù)據(jù)采集模塊：通過DHT11傳感器實(shí)時(shí)采集溫度和濕度數(shù)據(jù)，并將這些數(shù)據(jù)通過MQTT協(xié)議發(fā)送到本地的Mosquitto代理。
數(shù)據(jù)存儲(chǔ)模塊：使用SQLite數(shù)據(jù)庫存儲(chǔ)傳感器數(shù)據(jù)，并確保數(shù)據(jù)持久性。
Web服務(wù)模塊：通過Flask框架創(chuàng)建Web服務(wù)，以便用戶可以通過HTTP請(qǐng)求獲取傳感器的歷史數(shù)據(jù)。
測(cè)試和驗(yàn)證：通過MQTT客戶端訂閱消息和通過瀏覽器訪問Web服務(wù)，驗(yàn)證了整個(gè)系統(tǒng)的功能和數(shù)據(jù)流的正確性。
