场景介绍

今天程序猫带领大家如何实时获取树莓派传感器温湿度数据，并自动存储到数据库中。确保数据的持续性。

实现过程

硬件连接

树莓派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("资源已释放")