引言
物联网（IoT）正在改变我们的生活方式，而树莓派作为最经典的开源硬件平台之一，凭借其出色的GPIO接口、丰富的社区资源和极低的成本，成为了智能家居项目的首选方案。本文将从零开始，带你搭建一个完整的家庭环境监测系统，涵盖温湿度、空气质量、光照度等核心参数，并通过Web界面实现远程监控和数据可视化。
系统架构设计
整个系统采用分层架构设计，自下而上分为感知层、边缘计算层和云服务层三层：
- 感知层：DHT22温湿度传感器、MQ-135空气质量传感器、BH1750光照传感器
- 边缘计算层：树莓派4B作为数据采集和本地处理中枢
- 云服务层：InfluxDB时序数据库 + Grafana可视化面板
系统工作流程为：传感器通过GPIO与树莓派连接，Python采集脚本每30秒读取一次数据，经本地预处理后写入InfluxDB，Grafana通过Web Dashboard展示实时趋势图，并支持历史数据查询和异常告警。
硬件选型与接线
本项目选用以下硬件组件，总计成本约350元人民币：
- 树莓派4B（2GB RAM）×1 —— 核心控制器
- DHT22温湿度传感器 ×1 —— 采集环境温度和湿度
- MQ-135空气质量传感器 ×1 —— 检测有害气体浓度
- BH1750光照传感器 ×1 —— 测量环境光照强度
- ADS1115模数转换器 ×1 —— 将MQ-135的模拟信号转换为数字量
- 杜邦线若干、面包板 ×1
接线说明：
- DHT22数据线接GPIO4（Pin 7），电源接3.3V（Pin 1），GND接Pin 9
- BH1750通过I2C通信，SDA接GPIO2（Pin 3），SCL接GPIO3（Pin 5）
- MQ-135模拟输出接ADS1115的A0通道，ADS1115通过I2C并联到BH1750总线
软件环境搭建
首先在树莓派上配置基础软件环境：
# 启用I2C接口
sudo raspi-config
# 选择 Interface Options I2C Enable
# 安装Python依赖
sudo apt update
sudo apt install -y python3-pip python3-smbus i2c-tools
pip3 install adafruit-circuitpython-dht adafruit-circuitpython-bh1750 adafruit-circuitpython-ads1x15 influxdb-client flask
# 安装InfluxDB（时序数据库）
wget -q https://repos.influxdata.com/influxdata-archive_compat.key
echo '393e8779c89ac8d958f81f942f9ad7fb82a25e133faddaf92e15b16e6ac9ce4c influxdata-archive_compat.key' | sha256sum -c && cat influxdata-archive_compat.key | gpg --dearmor | sudo tee /etc/apt/trusted.gpg.d/influxdata-archive_compat.gpg > /dev/null
echo "deb [signed-by=/etc/apt/trusted.gpg.d/influxdata-archive_compat.gpg] https://repos.influxdata.com/debian $(lsb_release -cs) stable" | sudo tee /etc/apt/sources.list.d/influxdata.list
sudo apt update && sudo apt install -y influxdb
sudo systemctl enable influxdb && sudo systemctl start influxdb
# 安装Grafana（数据可视化）
wget -q -O - https://packages.grafana.com/gpg.key | sudo apt-key add -
echo "deb https://packages.grafana.com/oss/deb stable main" | sudo tee /etc/apt/sources.list.d/grafana.list
sudo apt update && sudo apt install -y grafana
sudo systemctl enable grafana-server && sudo systemctl start grafana-server核心代码实现
1. 传感器数据采集模块
import time
import board
import digitalio
import adafruit_dht
import adafruit_bh1750
import adafruit_ads1x15.ads1115 as ADS
from adafruit_ads1x15.analog_in import AnalogIn
import busio
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
# I2C总线初始化
i2c = busio.I2C(board.SCL, board.SDA)
# 传感器初始化
dht = adafruit_dht.DHT22(board.D4)
light_sensor = adafruit_bh1750.BH1750(i2c)
ads = ADS.ADS1115(i2c)
air_sensor = AnalogIn(ads, ADS.P0)
# InfluxDB配置
bucket = "home_monitoring"
org = "ybb"
url = "http://localhost:8086"
token = "your-influxdb-token"
client = InfluxDBClient(url=url, token=token, org=org)
write_api = client.write_api(write_options=SYNCHRONOUS)
def read_sensors():
"""读取所有传感器数据"""
try:
temperature = dht.temperature
humidity = dht.humidity
lux = light_sensor.lux
air_quality = air_sensor.value # 原始ADC值
return {
"temperature": temperature,
"humidity": humidity,
"lux": round(lux, 2),
"air_quality_raw": air_quality,
"air_quality_ppm": round(air_quality * 0.1, 2)
}
except RuntimeError as e:
print(f"传感器读取错误: {e}")
return None
def write_to_db(data):
"""写入InfluxDB时间序列数据库"""
point = Point("environment") \
.tag("location", "living_room") \
.tag("device", "raspberry_pi_4b") \
.field("temperature", data["temperature"]) \
.field("humidity", data["humidity"]) \
.field("lux", data["lux"]) \
.field("air_quality", data["air_quality_ppm"])
write_api.write(bucket=bucket, org=org, record=point)
print(f"数据已写入: {data}")
if __name__ == "__main__":
print("家庭环境监测系统启动...")
while True:
data = read_sensors()
if data:
write_to_db(data)
time.sleep(30)2. Web远程监控面板
from flask import Flask, render_template, jsonify
from influxdb_client import InfluxDBClient
from datetime import datetime, timedelta
app = Flask(__name__)
client = InfluxDBClient(url="http://localhost:8086",
token="your-token", org="ybb")
query_api = client.query_api()
@app.route("/")
def dashboard():
return render_template("dashboard.html")
@app.route("/api/current")
def current_data():
"""获取最新环境数据"""
query = '''
from(bucket: "home_monitoring")
|> range(start: -5m)
|> filter(fn: (r) => r._measurement == "environment")
|> last()
'''
result = query_api.query(query)
data = {}
for table in result:
for record in table.records:
data[record.field] = record.value
return jsonify(data)
@app.route("/api/history//")
def history_data(field, hours=24):
"""获取历史趋势数据"""
query = f'''
from(bucket: "home_monitoring")
|> range(start: -{hours}h)
|> filter(fn: (r) => r._measurement == "environment")
|> filter(fn: (r) => r._field == "{field}")
|> aggregateWindow(every: 5m, fn: mean)
'''
result = query_api.query(query)
points = []
for table in result:
for record in table.records:
points.append({
"time": record.get_time().isoformat(),
"value": record.value
})
return jsonify(points)
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000, debug=False) 3. 异常告警机制
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
class AlertManager:
def __init__(self):
self.thresholds = {
"temperature": {"min": 10, "max": 35, "unit": "°C"},
"humidity": {"min": 30, "max": 70, "unit": "%"},
"lux": {"min": 50, "max": 2000, "unit": "lx"},
"air_quality": {"min": 0, "max": 300, "unit": "ppm"}
}
self.cooldown = {} # 告警冷却时间
def check_thresholds(self, data):
"""检查数据是否超出阈值"""
alerts = []
for metric, value in data.items():
if metric not in self.thresholds:
continue
threshold = self.thresholds[metric]
if value < threshold["min"] or value > threshold["max"]:
alert_key = f"{metric}_{int(time.time() / 3600)}"
if alert_key not in self.cooldown:
self.cooldown[alert_key] = True
alerts.append({
"metric": metric,
"value": value,
"threshold": threshold,
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
})
return alerts
def send_alert(self, alerts):
"""发送邮件告警"""
if not alerts:
return
msg = MIMEMultipart()
msg["Subject"] = "智能家居环境监测告警"
msg["From"] = "[email protected]"
msg["To"] = "[email protected]"
body = "检测到异常数据：\n\n"
for alert in alerts:
body += f"- {alert['metric']}: {alert['value']}{alert['threshold']['unit']} "
body += f"(安全范围: {alert['threshold']['min']}-{alert['threshold']['max']}{alert['threshold']['unit']})"
body += f"(时间: {alert['time']})\n"
msg.attach(MIMEText(body, "plain"))
try:
server = smtplib.SMTP("smtp.gmail.com", 587)
server.starttls()
server.login("[email protected]", "your-password")
server.send_message(msg)
server.quit()
print("告警邮件已发送")
except Exception as e:
print(f"邮件发送失败: {e}")
# 在主循环中集成告警
alert_manager = AlertManager()
def monitoring_loop():
while True:
data = read_sensors()
if data:
write_to_db(data)
alerts = alert_manager.check_thresholds(data)
if alerts:
alert_manager.send_alert(alerts)
time.sleep(30)Grafana可视化配置
访问 http://树莓派IP:3000 进入Grafana管理界面（默认账号 admin/admin），添加InfluxDB数据源后，创建以下监控面板：
- 温度趋势图：折线图展示24小时温度变化，设置30°C为高温警戒线
- 湿度仪表盘：环形仪表显示当前湿度百分比
- 空气质量热力图：以不同颜色标识空气质量等级
- 光照度面积图：展示每日光照周期变化
- 综合状态面板：汇总所有指标的一屏概览
通过Grafana的Alerting功能，还可以配置Slack、钉钉或企业微信告警推送，实现真正的智能通知。
性能优化与生产建议
在实际部署中，建议关注以下几个优化点：
- 数据降采样：InfluxDB配置保留策略，7天原始数据，1天精度聚合保留1年
- 断网续传：使用本地SQLite缓存数据，网络恢复后批量同步
- 看门狗机制：systemd服务配置自动重启，防止脚本卡死
- 安全加固：Flask配置HTTPS，使用Nginx反向代理，限制API访问频率
- OTA更新：配置Git版本管理，远程自动拉取最新代码
项目扩展方向
基础环境监测系统搭建完成后，可以进一步扩展到以下场景：
- 接入摄像头模块，实现安防监控和人脸识别门禁
- 增加CO2传感器和PM2.5检测器，构建空气质量监测站
- 连接继电器模块，实现空调、加湿器等设备的自动控制
- 接入HomeAssistant开源平台，实现全屋智能联动
- 扩展多节点部署，覆盖全屋各个房间的环境监测
结语
通过本项目，我们基于树莓派从零搭建了一个完整的物联网环境监测系统，涵盖了传感器数据采集、时序数据库存储、Web可视化展示和异常告警通知四大核心功能。整套方案成本不到400元，却实现了商业级环境监测系统的核心功能。树莓派 + 开源软件的组合，证明了个人开发者也能构建专业级的物联网应用，这正是开源硬件的魅力所在。
发表评论 取消回复