随着物联网技术的快速发展,智能家居系统正在成为现代生活中不可或缺的一部分。本文介绍了一个基于STM32微控制器和Linux系统的智能家居解决方案,涵盖了硬件设计、软件架构、通信协议以及云平台集成等方面。
该系统具有以下特点:
系统的硬件架构如下图所示:
系统的软件架构采用分层设计,如下图所示:
#include "stm32f4xx_hal.h" #include "sensors.h" #include "zigbee.h" // 定义传感器数据结构 typedef struct { float temperature; float humidity; uint16_t light; } SensorData; // 主循环 void main(void) { // 初始化外设 HAL_Init(); Sensors_Init(); Zigbee_Init(); SensorData data; while(1) { // 读取传感器数据 data.temperature = Sensors_ReadTemperature(); data.humidity = Sensors_ReadHumidity(); data.light = Sensors_ReadLight(); // 通过Zigbee发送数据 Zigbee_SendData(&data, sizeof(SensorData)); // 检查是否有控制命令 if (Zigbee_CommandAvailable()) { uint8_t command = Zigbee_GetCommand(); ExecuteCommand(command); } // 延时1秒 HAL_Delay(1000); } } // 执行控制命令 void ExecuteCommand(uint8_t command) { switch(command) { case CMD_LIGHT_ON: GPIO_SetPin(LED_GPIO_Port, LED_Pin); break; case CMD_LIGHT_OFF: GPIO_ResetPin(LED_GPIO_Port, LED_Pin); break; case CMD_FAN_ON: GPIO_SetPin(FAN_GPIO_Port, FAN_Pin); break; case CMD_FAN_OFF: GPIO_ResetPin(FAN_GPIO_Port, FAN_Pin); break; default: // 未知命令,不做处理 break; } }
这段代码实现了STM32终端设备的主要功能:
接下来,让我们实现Raspberry Pi网关的部分代码:
import paho.mqtt.client as mqtt import json from flask import Flask, request, jsonify from zigbee_handler import ZigbeeHandler from database import Database app = Flask(__name__) zigbee = ZigbeeHandler() db = Database() mqtt_client = mqtt.Client() # MQTT配置 MQTT_BROKER = "iot.eclipse.org" MQTT_PORT = 1883 MQTT_TOPIC = "home/sensors" @app.route('/api/sensors', methods=['GET']) def get_sensor_data(): data = db.get_latest_sensor_data() return jsonify(data) @app.route('/api/control', methods=['POST']) def control_device(): command = request.json['command'] device_id = request.json['device_id'] zigbee.send_command(device_id, command) return jsonify({"status": "success"}) def on_mqtt_connect(client, userdata, flags, rc): print(f"Connected with result code {rc}") client.subscribe(MQTT_TOPIC) def on_mqtt_message(client, userdata, msg): payload = json.loads(msg.payload) db.save_sensor_data(payload) # 将数据转发到AWS IoT aws_iot_client.publish("aws/home/sensors", json.dumps(payload)) def zigbee_data_callback(data): # 处理从Zigbee接收到的数据 mqtt_client.publish(MQTT_TOPIC, json.dumps(data)) db.save_sensor_data(data) if __name__ == '__main__': # 设置MQTT客户端 mqtt_client.on_connect = on_mqtt_connect mqtt_client.on_message = on_mqtt_message mqtt_client.connect(MQTT_BROKER, MQTT_PORT, 60) mqtt_client.loop_start() # 设置Zigbee回调 zigbee.set_data_callback(zigbee_data_callback) # 启动Flask应用 app.run(host='0.0.0.0', port=5000)
这段代码实现了Raspberry Pi网关的主要功能:
下面是一个简单的React组件,用于显示传感器数据和控制设备:
import React, { useState, useEffect } from 'react'; import axios from 'axios'; const SmartHome = () => { const [sensorData, setSensorData] = useState(null); useEffect(() => { const fetchData = async () => { const result = await axios.get('/api/sensors'); setSensorData(result.data); }; fetchData(); const interval = setInterval(fetchData, 5000); // 每5秒更新一次数据 return () => clearInterval(interval); }, []); const controlDevice = async (deviceId, command) => { await axios.post('/api/control', { device_id: deviceId, command }); }; if (!sensorData) return Loading...; return ( Smart Home Dashboard
Sensor Data
Temperature: {sensorData.temperature}°C
Humidity: {sensorData.humidity}%
Light: {sensorData.light} lux
Device Control
); }; export default SmartHome;
以下是在Raspberry Pi网关上与AWS IoT云平台集成的Python代码示例:
import boto3 import json from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient # AWS IoT配置 AWS_IOT_ENDPOINT = "xxxxxxxxxxxxxxx.iot.us-west-2.amazonaws.com" AWS_IOT_CLIENT_ID = "raspberry_pi_gateway" AWS_IOT_TOPIC = "home/sensors" AWS_IOT_CERT_PATH = "/path/to/certificate.pem.crt" AWS_IOT_KEY_PATH = "/path/to/private.pem.key" AWS_IOT_ROOT_CA_PATH = "/path/to/root-CA.crt" # 初始化AWS IoT MQTT客户端 mqtt_client = AWSIoTMQTTClient(AWS_IOT_CLIENT_ID) mqtt_client.configureEndpoint(AWS_IOT_ENDPOINT, 8883) mqtt_client.configureCredentials(AWS_IOT_ROOT_CA_PATH, AWS_IOT_KEY_PATH, AWS_IOT_CERT_PATH) # 连接回调 def on_connect(self, params, rc): if rc == 0: print("Connected to AWS IoT") else: print(f"Connection failed with error code {rc}") # 消息回调 def on_message(client, userdata, message): payload = json.loads(message.payload.decode('utf-8')) print(f"Received message from AWS IoT: {payload}") # 处理来自云平台的命令 if 'command' in payload: execute_command(payload['command']) # 连接到AWS IoT mqtt_client.connect() mqtt_client.subscribe(AWS_IOT_TOPIC, 1, on_message) # 发送数据到AWS IoT def send_to_aws_iot(data): message = json.dumps(data) mqtt_client.publish(AWS_IOT_TOPIC, message, 1) print(f"Sent message to AWS IoT: {message}") # 使用AWS SDK创建IoT客户端 iot_client = boto3.client('iot-data') # 更新设备影子 def update_device_shadow(device_id, state): payload = json.dumps({ "state": { "reported": state } }) iot_client.update_thing_shadow( thingName=device_id, payload=payload ) # 示例:发送传感器数据并更新设备影子 sensor_data = { "temperature": 25.5, "humidity": 60, "light": 500 } send_to_aws_iot(sensor_data) update_device_shadow("living_room_sensor", sensor_data)
这段代码实现了以下功能:
send_to_aws_iot
。update_device_shadow
。在实际应用中,你需要将这些函数集成到之前的Raspberry Pi网关代码中,例如:
send_to_aws_iot
函数将数据发送到AWS IoT平台。update_device_shadow
函数更新相应设备的影子状态。以下是集成后的Raspberry Pi网关代码示例:
import paho.mqtt.client as mqtt import json from flask import Flask, request, jsonify from zigbee_handler import ZigbeeHandler from database import Database from aws_iot_handler import send_to_aws_iot, update_device_shadow, mqtt_client as aws_mqtt_client app = Flask(__name__) zigbee = ZigbeeHandler() db = Database() local_mqtt_client = mqtt.Client() # MQTT配置 MQTT_BROKER = "iot.eclipse.org" MQTT_PORT = 1883 MQTT_TOPIC = "home/sensors" @app.route('/api/sensors', methods=['GET']) def get_sensor_data(): data = db.get_latest_sensor_data() return jsonify(data) @app.route('/api/control', methods=['POST']) def control_device(): command = request.json['command'] device_id = request.json['device_id'] zigbee.send_command(device_id, command) # 更新设备影子 update_device_shadow(device_id, {"status": command}) return jsonify({"status": "success"}) def on_local_mqtt_connect(client, userdata, flags, rc): print(f"Connected to local MQTT broker with result code {rc}") client.subscribe(MQTT_TOPIC) def on_local_mqtt_message(client, userdata, msg): payload = json.loads(msg.payload) db.save_sensor_data(payload) # 将数据发送到AWS IoT send_to_aws_iot(payload) # 更新设备影子 update_device_shadow(payload['device_id'], payload) def zigbee_data_callback(data): # 处理从Zigbee接收到的数据 local_mqtt_client.publish(MQTT_TOPIC, json.dumps(data)) db.save_sensor_data(data) # 将数据发送到AWS IoT send_to_aws_iot(data) # 更新设备影子 update_device_shadow(data['device_id'], data) def aws_iot_command_callback(client, userdata, message): payload = json.loads(message.payload.decode('utf-8')) if 'command' in payload: device_id = payload.get('device_id') command = payload['command'] zigbee.send_command(device_id, command) # 更新设备影子 update_device_shadow(device_id, {"status": command}) if __name__ == '__main__': # 设置本地MQTT客户端 local_mqtt_client.on_connect = on_local_mqtt_connect local_mqtt_client.on_message = on_local_mqtt_message local_mqtt_client.connect(MQTT_BROKER, MQTT_PORT, 60) local_mqtt_client.loop_start() # 设置Zigbee回调 zigbee.set_data_callback(zigbee_data_callback) # 设置AWS IoT MQTT客户端回调 aws_mqtt_client.subscribe("home/commands", 1, aws_iot_command_callback) # 启动Flask应用 app.run(host='0.0.0.0', port=5000)
这个完整的Raspberry Pi网关代码实现了以下功能:
本智能家居系统项目成功地整合了多种技术,包括:
通过这个项目,我们实现了一个功能完整的智能家居系统,具有以下特点: