把树莓派变成智能家居中枢：用Python蓝牙连接温湿度传感器和手机APP
树莓派变身智能家居中枢：Python蓝牙温湿度监测系统实战
想象一下，清晨醒来时手机自动显示卧室温湿度，离家时远程查看宠物房环境数据，这些场景不再需要昂贵的商业设备——只需一个树莓派和几行Python代码就能实现。本文将带你从零构建一个基于蓝牙低功耗（BLE）的智能环境监测系统，实现传感器数据到手机的无线传输。
1. 硬件准备与环境配置
1.1 所需硬件组件清单
- 树莓派4B/3B+（内置蓝牙4.2+）
- DHT22温湿度传感器（精度±0.5°C，±2%RH）
- 4.7kΩ上拉电阻
- 面包板与杜邦线
- 智能手机（Android/iOS均可）
提示：DHT11成本更低但精度较差，DHT22虽然价格稍高但适合对数据精度有要求的场景
1.2 电路连接示意图
树莓派 GPIO 引脚布局： [USB端口侧] 3V3 (1) □ ■ □ (2) 5V GPIO2 (3) □ ■ □ (4) 5V GPIO3 (5) □ ■ □ (6) GND GPIO4 (7) □ ■ □ (8) GPIO14 GND (9) □ ■ □ (10) GPIO15 DHT22连接方式： VCC → 3V3 (引脚1) DATA → GPIO4 (引脚7) GND → GND (引脚9)在DATA与VCC之间连接4.7kΩ上拉电阻
1.3 系统环境准备
更新系统并安装必要组件：
# 更新软件源 sudo apt update && sudo apt upgrade -y # 安装Python环境 sudo apt install python3-pip python3-venv # 创建虚拟环境 python3 -m venv ble_env source ble_env/bin/activate # 安装依赖库 pip install Adafruit_DHT dbus-python pygobject2. 传感器数据采集模块
2.1 DHT22驱动实现
创建sensor_reader.py文件：
import Adafruit_DHT import time class DHT22Reader: def __init__(self, pin=4): self.sensor = Adafruit_DHT.DHT22 self.pin = pin def read_data(self): humidity, temperature = Adafruit_DHT.read_retry(self.sensor, self.pin) if humidity is not None and temperature is not None: return { 'temperature': round(temperature, 1), 'humidity': round(humidity, 1), 'timestamp': int(time.time()) } return None # 测试代码 if __name__ == "__main__": reader = DHT22Reader() while True: data = reader.read_data() if data: print(f"温度: {data['temperature']}°C, 湿度: {data['humidity']}%") time.sleep(2)2.2 数据稳定性优化
常见问题及解决方案：
| 问题现象 | 可能原因 | 解决方法 |
|---|---|---|
| 读取返回None | 接线松动/电源不稳 | 检查电路连接，确保上拉电阻正确安装 |
| 数据跳变过大 | 传感器响应延迟 | 增加读取重试次数，取多次测量平均值 |
| 周期性读取失败 | GPIO端口冲突 | 更换GPIO引脚，避免使用系统保留引脚 |
添加数据平滑处理算法：
from collections import deque class SmoothingReader(DHT22Reader): def __init__(self, pin=4, window_size=5): super().__init__(pin) self.temp_history = deque(maxlen=window_size) self.humid_history = deque(maxlen=window_size) def read_data(self): raw_data = super().read_data() if raw_data: self.temp_history.append(raw_data['temperature']) self.humid_history.append(raw_data['humidity']) return { 'temperature': sum(self.temp_history)/len(self.temp_history), 'humidity': sum(self.humid_history)/len(self.humid_history), 'timestamp': raw_data['timestamp'] } return None3. BLE服务架构设计
3.1 GATT服务规划
我们定义自定义服务UUID：0000AA10-0000-1000-8000-00805F9B34FB
服务包含三个特征：
- 温度数据（可读+通知）
- 湿度数据（可读+通知）
- 配置参数（可写）
特征UUID设计：
TEMPERATURE_CHAR_UUID = "0000AA11-0000-1000-8000-00805F9B34FB" HUMIDITY_CHAR_UUID = "0000AA12-0000-1000-8000-00805F9B34FB" CONFIG_CHAR_UUID = "0000AA13-0000-1000-8000-00805F9B34FB"3.2 服务端核心代码
创建ble_service.py：
import dbus import dbus.service import dbus.mainloop.glib from gi.repository import GLib import json from sensor_reader import DHT22Reader # UUID定义 SERVICE_UUID = "0000AA10-0000-1000-8000-00805F9B34FB" # ...其他UUID定义... class EnvironmentService(dbus.service.Object): def __init__(self, bus, path): super().__init__(bus, path) self.reader = DHT22Reader() self.update_interval = 2.0 # 默认2秒更新 @dbus.service.method(GATT_CHRC_IFACE, in_signature='a{sv}', out_signature='ay') def ReadValue(self, options): data = self.reader.read_data() if data: return dbus.Array( json.dumps(data).encode('utf-8'), signature='y' ) return dbus.Array(b'', signature='y') @dbus.service.method(GATT_CHRC_IFACE, in_signature='aya{sv}') def WriteValue(self, value, options): try: config = json.loads(bytes(value).decode('utf-8')) if 'interval' in config: self.update_interval = float(config['interval']) except Exception as e: print(f"配置写入错误: {str(e)}") class Advertisement(dbus.service.Object): # ...广告实现代码... def register_services(): # 初始化DBus dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) bus = dbus.SystemBus() # 注册广告 adv = Advertisement(bus, '/com/example/env_sensor') adapter = find_adapter(bus) adapter.RegisterAdvertisement(adv.get_path(), {}) # 注册GATT服务 service = EnvironmentService(bus, '/com/example/env_service') gatt_manager = find_adapter(bus, 'org.bluez.GattManager1') gatt_manager.RegisterApplication(service.get_path(), {}) return bus if __name__ == "__main__": bus = register_services() mainloop = GLib.MainLoop() print("环境监测服务已启动...") mainloop.run()4. 手机端应用开发
4.1 Android端实现方案
使用Android Studio创建基础应用，关键BLE操作代码：
// BluetoothGattCallback实现 private final BluetoothGattCallback gattCallback = new BluetoothGattCallback() { @Override public void onConnectionStateChange(BluetoothGatt gatt, int status, int newState) { if (newState == BluetoothProfile.STATE_CONNECTED) { gatt.discoverServices(); } } @Override public void onServicesDiscovered(BluetoothGatt gatt, int status) { BluetoothGattService envService = gatt.getService(UUID.fromString(SERVICE_UUID)); if (envService != null) { // 获取特征并启用通知 BluetoothGattCharacteristic tempChar = envService.getCharacteristic( UUID.fromString(TEMPERATURE_CHAR_UUID)); gatt.setCharacteristicNotification(tempChar, true); } } @Override public void onCharacteristicChanged(BluetoothGatt gatt, BluetoothGattCharacteristic characteristic) { // 处理实时数据更新 String jsonData = new String(characteristic.getValue()); runOnUiThread(() -> updateUI(jsonData)); } };4.2 iOS端快捷方案
对于不想开发原生应用的用户，可以使用现成工具：
- LightBlue Explorer：可视化BLE调试工具
- BLE Scanner：支持数据日志记录
- Pythonista（iOS） +bleak库：直接在iPhone上运行Python脚本
示例Pythonista代码片段：
import bleak import asyncio async def monitor_sensor(): async with bleak.BleakClient(DEVICE_ADDRESS) as client: svc = await client.get_services() temp_char = svc.get_characteristic(TEMP_UUID) def callback(sender, data): print(f"更新数据: {data.decode()}") await client.start_notify(temp_char, callback) while True: await asyncio.sleep(1) asyncio.run(monitor_sensor())5. 系统优化与扩展
5.1 功耗管理技巧
树莓派作为常驻设备时，可通过以下方式降低功耗：
- CPU降频：设置保守模式
echo "conservative" | sudo tee /sys/devices/system/cpu/cpu0/cpufreq/scaling_governor - 蓝牙优化：调整广播间隔
adv_properties = { 'Type': 'peripheral', 'Interval': dbus.UInt32(800), # 800ms 'Timeout': dbus.UInt32(0) # 无超时 } - 硬件方案：使用USB蓝牙适配器替代内置模块
5.2 数据持久化方案
添加SQLite存储支持：
import sqlite3 from contextlib import closing class DataLogger: def __init__(self, db_path='sensor_data.db'): self.conn = sqlite3.connect(db_path) with closing(self.conn.cursor()) as cur: cur.execute('''CREATE TABLE IF NOT EXISTS readings (timestamp INTEGER PRIMARY KEY, temperature REAL, humidity REAL)''') def add_reading(self, data): with closing(self.conn.cursor()) as cur: cur.execute('INSERT INTO readings VALUES (?,?,?)', (data['timestamp'], data['temperature'], data['humidity'])) self.conn.commit()5.3 多传感器组网
通过BLE Mesh扩展多个监测点：
- 设置每个树莓派为独立节点
- 使用相同的服务UUID但不同的设备名称
- 手机端轮询切换连接不同设备
设备识别方案：
adv_properties = { 'LocalName': f'EnvMonitor-{socket.gethostname()}', 'ManufacturerData': { 0xFFFF: [0x01, 0x02] # 自定义厂商代码 } }6. 故障排查指南
6.1 常见问题速查表
| 现象 | 检查步骤 | 解决方案 |
|---|---|---|
| 手机搜索不到设备 | 1. 确认树莓派蓝牙已启用 2. 检查广告是否成功注册 3. 验证设备是否在范围内 | sudo hciconfig hci0 up重启蓝牙服务 |
| 连接后无数据 | 1. 确认GATT服务注册成功 2. 检查特征属性配置 3. 验证通知是否启用 | 使用bluetoothctl交互命令检查服务 |
| 数据更新延迟 | 1. 检查传感器读取间隔 2. 确认BLE连接参数 3. 测试信号强度 | 调整广播间隔和连接参数 |
6.2 调试工具推荐
- bluetoothctl：内置命令行工具
sudo bluetoothctl scan on # 查看周边设备 list # 检查本地适配器 - Wireshark+BTSnoop：协议分析
- nRF Connect：跨平台BLE调试应用
对于持续运行的系统，建议添加看门狗监控：
import subprocess from threading import Timer def check_ble_status(): result = subprocess.run(['hciconfig', 'hci0'], capture_output=True, text=True) if 'RUNNING' not in result.stdout: subprocess.run(['sudo', 'systemctl', 'restart', 'bluetooth']) Timer(300, check_ble_status).start() # 每5分钟检查一次7. 项目扩展方向
7.1 家居自动化集成
将数据接入智能家居平台：
- Home Assistant：通过REST API添加传感器
# configuration.yaml sensor: - platform: rest name: Room Temperature resource: http://树莓派IP:5000/api/temperature unit_of_measurement: "°C" - IFTTT：设置温湿度触发条件
- MQTT桥接：发布到物联网消息总线
7.2 可视化仪表盘
使用Grafana展示历史数据：
from flask import Flask, jsonify import sqlite3 app = Flask(__name__) @app.route('/api/readings') def get_readings(): conn = sqlite3.connect('sensor_data.db') cur = conn.cursor() cur.execute('SELECT * FROM readings ORDER BY timestamp DESC LIMIT 100') return jsonify([dict(zip(['time','temp','humid'], row)) for row in cur.fetchall()]) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)7.3 安全增强措施
- 连接加密：启用BLE安全配对
adv_properties['Includes'] = dbus.Array(['tx-power', 'flags'], signature='s') - 数据校验：添加CRC校验字段
import zlib def add_checksum(data): crc = zlib.crc32(json.dumps(data).encode()) return {**data, 'checksum': crc} - 访问控制：白名单MAC地址过滤
实际部署中发现，将树莓派放置在金属盒中可能导致蓝牙信号衰减，最佳位置是距离地面1.5-2米且避开大型电器设备。对于多房间监测，建议每个节点间隔不超过8米以保证稳定的信号质量。