場景介紹
今天程序貓帶領大家如何實時獲取樹莓派傳感器溫濕度數據，并自動存儲到數據庫中。確保數據的持續性。
實現過程
硬件連接
樹莓派4b連接GPIO引腳與DHT11傳感器;
硬件只涉及樹莓派、DHT11傳感器。
DHT11的信號引腳連接樹莓派的GPIO17， DHT11的Vdd（+）和GND引腳（-）分別連接樹莓派的電源正極和GND引腳。
啟動MQTT服務
終端啟動Mosquitto服務
sudo systemctl start mosquitto
讀取傳感器數據存儲數據庫
在數據采集模塊中添加數據庫寫入的邏輯，編寫腳本代碼sensor_save.py
import time
import board
import adafruit_dht
from datetime import datetime
import paho.mqtt.client as mqtt
import json
import sqlite3
from pathlib import Path# MQTT配置
broker = "192.168.137.31"
port = 1883
topic = "DHT11_message"# 傳感器引腳配置
SENSOR_PIN = board.D17 # 使用CircuitPython標準命名（物理引腳11）# 數據庫配置
DB_NAME = "sensor_data.db"
DB_TABLE = "environment_data"# 初始化數據庫
def init_db():# 檢查數據庫文件是否存在，不存在則創建db_file = Path(DB_NAME)if not db_file.exists():conn = sqlite3.connect(DB_NAME)cursor = conn.cursor()cursor.execute(f"""CREATE TABLE {DB_TABLE} (id INTEGER PRIMARY KEY AUTOINCREMENT,timestamp TEXT NOT NULL,temperature REAL NOT NULL,humidity REAL NOT NULL)""")conn.commit()conn.close()print(f"數據庫 {DB_NAME} 和表 {DB_TABLE} 創建成功")else:print(f"使用現有數據庫 {DB_NAME}")# 寫入數據到數據庫
def write_to_db(timestamp, temperature, humidity):try:conn = sqlite3.connect(DB_NAME)cursor = conn.cursor()cursor.execute(f"""INSERT INTO {DB_TABLE} (timestamp, temperature, humidity)VALUES (?, ?, ?)""", (timestamp, temperature, humidity))conn.commit()conn.close()return Trueexcept sqlite3.Error as e:print(f"數據庫寫入錯誤: {e}")return False# 初始化MQTT客戶端
client = mqtt.Client()
client.connect(broker, port)# 初始化數據庫
init_db()try:# 添加 use_pulseio=False 參數dhtDevice = adafruit_dht.DHT11(SENSOR_PIN, use_pulseio=False)print("傳感器初始化成功，開始數據采集......")while True:try:# 讀取溫濕度數據temperature = dhtDevice.temperaturehumidity = dhtDevice.humidity# 檢查有效讀數if temperature is None or humidity is None:raise RuntimeError("讀取到無效值")# 獲取當前時間戳recDate = datetime.now().strftime('%Y-%m-%d %H:%M:%S')# 1. 寫入數據庫if write_to_db(recDate, temperature, humidity):print(f"? [{recDate}] 數據庫寫入成功: 溫度={temperature}°C, 濕度={humidity}%")else:print(f"? [{recDate}] 數據庫寫入失敗")# 2. 準備MQTT消息并發送msg = json.dumps({'Tem': temperature,'Hum': humidity,'Date': recDate})result = client.publish(topic, msg)if result[0] == mqtt.MQTT_ERR_SUCCESS:print(f"? [{recDate}] MQTT發送成功: {msg}")else:print(f"? MQTT發送失敗，錯誤碼: {result[0]}")except RuntimeError as e:print(f"傳感器讀取錯誤: {e}")except Exception as e:print(f"意外錯誤: {e}")time.sleep(10.0) # 采集間隔10sexcept KeyboardInterrupt:print("程序被用戶終止")
except RuntimeError as e:print(f"初始化失敗: {e}")exit(1)
finally:# 清理資源if 'dhtDevice' in locals():dhtDevice.exit()client.disconnect()print("資源已釋放")
進入虛擬環境
樹莓派終端輸入命令：
source adafruit_env/bin/activate
運行腳本
python3 sensor_save.py
可以看到傳感器數據采集寫入數據庫
安裝DB Browser for SQLite?工具
適用于Raspbian/Debian系統：
控制臺輸入命令：
sudo apt install sqlitebrowser
打開可視化工具，并打開sensor_data.db,查看表數據environment_data寫入成功。
