把树莓派变成智能家居中枢：用Python蓝牙连接温湿度传感器和手机APP

树莓派变身智能家居中枢Python蓝牙温湿度监测系统实战想象一下清晨醒来时手机自动显示卧室温湿度离家时远程查看宠物房环境数据这些场景不再需要昂贵的商业设备——只需一个树莓派和几行Python代码就能实现。本文将带你从零构建一个基于蓝牙低功耗BLE的智能环境监测系统实现传感器数据到手机的无线传输。1. 硬件准备与环境配置1.1 所需硬件组件清单树莓派4B/3B内置蓝牙4.2DHT22温湿度传感器精度±0.5°C±2%RH4.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, pin4): 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, pin4, window_size5): super().__init__(pin) self.temp_history deque(maxlenwindow_size) self.humid_history deque(maxlenwindow_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服务规划我们定义自定义服务UUID0000AA10-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-00805F9B34FB3.2 服务端核心代码创建ble_service.pyimport 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_signaturea{sv}, out_signatureay) def ReadValue(self, options): data self.reader.read_data() if data: return dbus.Array( json.dumps(data).encode(utf-8), signaturey ) return dbus.Array(b, signaturey) dbus.service.method(GATT_CHRC_IFACE, in_signatureaya{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_defaultTrue) 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支持数据日志记录PythonistaiOS 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_pathsensor_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: fEnvMonitor-{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 # 检查本地适配器WiresharkBTSnoop协议分析nRF Connect跨平台BLE调试应用对于持续运行的系统建议添加看门狗监控import subprocess from threading import Timer def check_ble_status(): result subprocess.run([hciconfig, hci0], capture_outputTrue, textTrue) 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: °CIFTTT设置温湿度触发条件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(host0.0.0.0, port5000)7.3 安全增强措施连接加密启用BLE安全配对adv_properties[Includes] dbus.Array([tx-power, flags], signatures)数据校验添加CRC校验字段import zlib def add_checksum(data): crc zlib.crc32(json.dumps(data).encode()) return {**data, checksum: crc}访问控制白名单MAC地址过滤实际部署中发现将树莓派放置在金属盒中可能导致蓝牙信号衰减最佳位置是距离地面1.5-2米且避开大型电器设备。对于多房间监测建议每个节点间隔不超过8米以保证稳定的信号质量。