当前位置：首页 > news >正文

基于IoT的智能温控空调系统设计与实现

news 2025/10/12 10:30:07

前言

随着物联网技术的快速发展，智能家居已经成为现代生活的重要组成部分。智能空调作为智能家居的核心设备之一，其自动感知环境温度并智能调节的能力，不仅提升了用户体验，还能有效节约能源。本文将详细介绍如何设计和实现一个能够自动感知环境温度并进行智能工作的空调控制系统。

一、系统架构设计

1.1 整体架构

智能温控空调系统采用分层架构设计，主要包含以下几个层次：

┌─────────────────────────────────────────┐
│           应用层（APP/Web）              │
├─────────────────────────────────────────┤
│           云平台服务层                   │
├─────────────────────────────────────────┤
│           通信层（MQTT/HTTP）            │
├─────────────────────────────────────────┤
│           边缘计算层                     │
├─────────────────────────────────────────┤
│     感知层（传感器）+ 执行层（空调）      │
└─────────────────────────────────────────┘

1.2 核心组件

感知模块

DHT22温湿度传感器：精度高，响应速度快
红外人体感应传感器：检测房间内是否有人
光照传感器：辅助判断环境状态

控制模块

ESP32微控制器：支持WiFi和蓝牙，处理能力强
红外发射模块：控制传统空调
继电器模块：直接控制空调电源

智能决策模块

基于规则的决策引擎
机器学习预测模型
用户习惯学习算法

二、核心算法实现

2.1 温度感知与数据采集

import time
import board
import adafruit_dht
import json
from datetime import datetimeclass TemperatureSensor:def __init__(self, pin=board.D4):"""初始化温湿度传感器"""self.dhtDevice = adafruit_dht.DHT22(pin)self.history_data = []def read_sensor_data(self):"""读取传感器数据"""try:temperature = self.dhtDevice.temperaturehumidity = self.dhtDevice.humidityif temperature is not None and humidity is not None:data = {"temperature": round(temperature, 2),"humidity": round(humidity, 2),"timestamp": datetime.now().isoformat()}self.history_data.append(data)return dataelse:return Noneexcept RuntimeError as error:print(f"读取错误: {error.args[0]}")return Nonedef get_average_temperature(self, minutes=5):"""获取指定时间内的平均温度"""if not self.history_data:return Nonecurrent_time = datetime.now()recent_data = []for data in reversed(self.history_data):data_time = datetime.fromisoformat(data["timestamp"])if (current_time - data_time).seconds <= minutes * 60:recent_data.append(data["temperature"])else:breakreturn sum(recent_data) / len(recent_data) if recent_data else None

2.2 智能控制算法

class SmartACController:def __init__(self):self.target_temperature = 25  # 目标温度self.comfort_range = 2  # 舒适区间±2度self.energy_saving_mode = Falseself.occupancy_detected = Falseself.learning_model = UserPreferenceModel()def calculate_optimal_settings(self, current_temp, humidity, occupancy, outdoor_temp=None):"""计算最优空调设置"""settings = {"power": "OFF","mode": "AUTO","temperature": self.target_temperature,"fan_speed": "AUTO"}# 无人模式if not occupancy:if self.energy_saving_mode:settings["power"] = "OFF"return settingselse:# 保持最低运行settings["temperature"] = self.target_temperature + 3settings["fan_speed"] = "LOW"# 温度控制逻辑temp_diff = current_temp - self.target_temperatureif abs(temp_diff) <= self.comfort_range:# 在舒适区间内settings["power"] = "ON"settings["fan_speed"] = "LOW"elif temp_diff > self.comfort_range:# 温度过高，需要制冷settings["power"] = "ON"settings["mode"] = "COOL"settings["temperature"] = self.target_temperature - 1# 根据温差调整风速if temp_diff > 5:settings["fan_speed"] = "HIGH"elif temp_diff > 3:settings["fan_speed"] = "MEDIUM"else:settings["fan_speed"] = "LOW"elif temp_diff < -self.comfort_range:# 温度过低，需要制热settings["power"] = "ON"settings["mode"] = "HEAT"settings["temperature"] = self.target_temperature + 1settings["fan_speed"] = "MEDIUM"# 湿度调节if humidity > 70:settings["mode"] = "DRY"return settingsdef apply_user_preferences(self, settings, user_id):"""应用用户偏好"""preferences = self.learning_model.get_user_preferences(user_id)if preferences:settings["temperature"] = preferences.get("preferred_temp", settings["temperature"])settings["fan_speed"] = preferences.get("preferred_fan_speed", settings["fan_speed"])return settings

2.3 机器学习预测模型

import numpy as np
from sklearn.ensemble import RandomForestRegressor
from sklearn.preprocessing import StandardScalerclass TemperaturePredictionModel:def __init__(self):self.model = RandomForestRegressor(n_estimators=100, random_state=42)self.scaler = StandardScaler()self.is_trained = Falsedef prepare_features(self, data):"""准备特征数据"""features = []for record in data:hour = datetime.fromisoformat(record["timestamp"]).hourminute = datetime.fromisoformat(record["timestamp"]).minutefeature = [record["temperature"],record["humidity"],hour,minute,record.get("outdoor_temp", 30),record.get("occupancy", 1)]features.append(feature)return np.array(features)def train(self, historical_data):"""训练模型"""if len(historical_data) < 100:return FalseX = self.prepare_features(historical_data[:-1])y = [d["temperature"] for d in historical_data[1:]]X_scaled = self.scaler.fit_transform(X)self.model.fit(X_scaled, y)self.is_trained = Truereturn Truedef predict_next_temperature(self, current_data, minutes_ahead=30):"""预测未来温度"""if not self.is_trained:return None# 准备当前特征current_time = datetime.now()future_time = current_time + timedelta(minutes=minutes_ahead)feature = [[current_data["temperature"],current_data["humidity"],future_time.hour,future_time.minute,current_data.get("outdoor_temp", 30),current_data.get("occupancy", 1)]]feature_scaled = self.scaler.transform(feature)prediction = self.model.predict(feature_scaled)[0]return round(prediction, 2)

2.4 PID控制器实现

class PIDController:def __init__(self, kp=1.0, ki=0.1, kd=0.05):"""PID控制器初始化"""self.kp = kp  # 比例系数self.ki = ki  # 积分系数self.kd = kd  # 微分系数self.last_error = 0self.integral = 0self.last_time = time.time()def update(self, setpoint, measured_value):"""更新PID控制器"""current_time = time.time()dt = current_time - self.last_timeif dt <= 0.0:dt = 0.01# 计算误差error = setpoint - measured_value# 计算PID各项p_term = self.kp * errorself.integral += error * dti_term = self.ki * self.integralderivative = (error - self.last_error) / dtd_term = self.kd * derivative# 计算输出output = p_term + i_term + d_term# 更新状态self.last_error = errorself.last_time = current_time# 限制输出范围output = max(-100, min(100, output))return outputdef reset(self):"""重置PID控制器"""self.last_error = 0self.integral = 0self.last_time = time.time()

三、系统集成与通信

3.1 MQTT通信实现

import paho.mqtt.client as mqtt
import json
import threadingclass ACCommunicator:def __init__(self, broker_address="broker.emqx.io", port=1883):self.client = mqtt.Client("smart_ac_controller")self.broker_address = broker_addressself.port = portself.is_connected = False# 设置回调函数self.client.on_connect = self.on_connectself.client.on_message = self.on_messageself.client.on_disconnect = self.on_disconnectdef on_connect(self, client, userdata, flags, rc):"""连接回调"""if rc == 0:print("成功连接到MQTT服务器")self.is_connected = True# 订阅控制主题client.subscribe("home/ac/control")client.subscribe("home/ac/settings")else:print(f"连接失败，返回码: {rc}")def on_message(self, client, userdata, msg):"""消息回调"""try:topic = msg.topicpayload = json.loads(msg.payload.decode())if topic == "home/ac/control":self.handle_control_command(payload)elif topic == "home/ac/settings":self.update_settings(payload)except Exception as e:print(f"处理消息时出错: {e}")def publish_sensor_data(self, data):"""发布传感器数据"""if self.is_connected:topic = "home/ac/sensor"self.client.publish(topic, json.dumps(data))def publish_status(self, status):"""发布空调状态"""if self.is_connected:topic = "home/ac/status"self.client.publish(topic, json.dumps(status))

3.2 RESTful API接口

from flask import Flask, jsonify, request
from flask_cors import CORSapp = Flask(__name__)
CORS(app)# 全局控制器实例
ac_controller = SmartACController()
sensor = TemperatureSensor()@app.route('/api/status', methods=['GET'])
def get_status():"""获取当前状态"""current_data = sensor.read_sensor_data()if current_data:settings = ac_controller.calculate_optimal_settings(current_data["temperature"],current_data["humidity"],True  # 假设有人)return jsonify({"sensor_data": current_data,"ac_settings": settings,"target_temperature": ac_controller.target_temperature})return jsonify({"error": "无法读取传感器数据"}), 500@app.route('/api/temperature', methods=['POST'])
def set_temperature():"""设置目标温度"""data = request.jsontarget_temp = data.get('temperature')if target_temp and 16 <= target_temp <= 30:ac_controller.target_temperature = target_tempreturn jsonify({"success": True, "target_temperature": target_temp})return jsonify({"error": "温度设置无效"}), 400@app.route('/api/mode', methods=['POST'])
def set_mode():"""设置运行模式"""data = request.jsonmode = data.get('mode')valid_modes = ['AUTO', 'COOL', 'HEAT', 'DRY', 'FAN']if mode in valid_modes:# 设置模式逻辑return jsonify({"success": True, "mode": mode})return jsonify({"error": "模式无效"}), 400@app.route('/api/schedule', methods=['POST'])
def set_schedule():"""设置定时计划"""data = request.jsonschedule = data.get('schedule')# 处理定时逻辑return jsonify({"success": True, "schedule": schedule})

四、优化策略

4.1 能效优化

class EnergyOptimizer:def __init__(self):self.electricity_rates = {"peak": {"hours": [(8, 11), (18, 22)], "rate": 1.2},"normal": {"hours": [(6, 8), (11, 18), (22, 23)], "rate": 0.8},"valley": {"hours": [(23, 6)], "rate": 0.5}}def get_current_rate(self):"""获取当前电价"""current_hour = datetime.now().hourfor period, info in self.electricity_rates.items():for start, end in info["hours"]:if start <= current_hour < end:return info["rate"]return 0.8  # 默认电价def optimize_temperature_setting(self, current_temp, target_temp, outdoor_temp):"""优化温度设置以节能"""current_rate = self.get_current_rate()# 峰值电价时期，适当调整目标温度if current_rate > 1.0:if outdoor_temp > target_temp:  # 夏天optimized_temp = min(target_temp + 2, 28)else:  # 冬天optimized_temp = max(target_temp - 2, 18)else:optimized_temp = target_tempreturn optimized_tempdef calculate_energy_consumption(self, settings, duration_minutes):"""计算能耗"""base_consumption = 1.5  # 基础功率 kW# 根据设置调整功率power_multiplier = 1.0if settings["power"] == "OFF":return 0if settings["mode"] == "COOL":power_multiplier = 1.2elif settings["mode"] == "HEAT":power_multiplier = 1.5elif settings["mode"] == "DRY":power_multiplier = 0.8# 风速影响fan_multipliers = {"LOW": 0.9, "MEDIUM": 1.0, "HIGH": 1.2}power_multiplier *= fan_multipliers.get(settings["fan_speed"], 1.0)# 计算总能耗consumption_kwh = base_consumption * power_multiplier * (duration_minutes / 60)return round(consumption_kwh, 2)

4.2 用户习惯学习

class UserPreferenceModel:def __init__(self):self.user_data = {}def record_user_action(self, user_id, action):"""记录用户操作"""if user_id not in self.user_data:self.user_data[user_id] = {"actions": [],"preferences": {}}action_record = {"timestamp": datetime.now().isoformat(),"action": action,"context": {"time_of_day": datetime.now().hour,"day_of_week": datetime.now().weekday()}}self.user_data[user_id]["actions"].append(action_record)self.update_preferences(user_id)def update_preferences(self, user_id):"""更新用户偏好"""if user_id not in self.user_data:returnactions = self.user_data[user_id]["actions"]# 分析最近100次操作recent_actions = actions[-100:] if len(actions) > 100 else actions# 统计温度偏好temp_settings = [a["action"]["temperature"] for a in recent_actions if "temperature" in a["action"]]if temp_settings:avg_temp = sum(temp_settings) / len(temp_settings)self.user_data[user_id]["preferences"]["preferred_temp"] = round(avg_temp, 1)# 统计时间段偏好time_preferences = {}for action in recent_actions:hour = action["context"]["time_of_day"]time_slot = f"{hour:02d}:00-{(hour+1):02d}:00"if time_slot not in time_preferences:time_preferences[time_slot] = []if "temperature" in action["action"]:time_preferences[time_slot].append(action["action"]["temperature"])self.user_data[user_id]["preferences"]["time_preferences"] = {slot: round(sum(temps) / len(temps), 1)for slot, temps in time_preferences.items()if temps}def get_user_preferences(self, user_id):"""获取用户偏好"""if user_id in self.user_data:return self.user_data[user_id]["preferences"]return None

五、实际应用案例

5.1 主程序实现

import asyncio
import logging
from datetime import datetime, timedeltaclass SmartACSystem:def __init__(self):self.sensor = TemperatureSensor()self.controller = SmartACController()self.pid = PIDController()self.communicator = ACCommunicator()self.energy_optimizer = EnergyOptimizer()self.prediction_model = TemperaturePredictionModel()# 配置日志logging.basicConfig(level=logging.INFO)self.logger = logging.getLogger(__name__)async def main_loop(self):"""主控制循环"""self.logger.info("智能空调系统启动")# 连接MQTTself.communicator.client.connect(self.communicator.broker_address,self.communicator.port)self.communicator.client.loop_start()while True:try:# 读取传感器数据sensor_data = self.sensor.read_sensor_data()if sensor_data:# 预测未来温度predicted_temp = Noneif self.prediction_model.is_trained:predicted_temp = self.prediction_model.predict_next_temperature(sensor_data, minutes_ahead=30)# 计算最优设置settings = self.controller.calculate_optimal_settings(sensor_data["temperature"],sensor_data["humidity"],occupancy=True,  # 实际应用中从传感器获取outdoor_temp=30  # 实际应用中从API获取)# 能效优化optimized_temp = self.energy_optimizer.optimize_temperature_setting(sensor_data["temperature"],self.controller.target_temperature,outdoor_temp=30)settings["temperature"] = optimized_temp# PID微调pid_output = self.pid.update(settings["temperature"],sensor_data["temperature"])# 应用PID调整if abs(pid_output) > 10:if pid_output > 0:settings["fan_speed"] = "HIGH"else:settings["fan_speed"] = "LOW"# 发布数据self.communicator.publish_sensor_data(sensor_data)self.communicator.publish_status(settings)# 记录日志self.logger.info(f"温度: {sensor_data['temperature']}°C, "f"湿度: {sensor_data['humidity']}%, "f"设置: {settings}")if predicted_temp:self.logger.info(f"预测30分钟后温度: {predicted_temp}°C")# 等待下一个循环await asyncio.sleep(60)  # 每分钟更新一次except Exception as e:self.logger.error(f"主循环错误: {e}")await asyncio.sleep(5)def run(self):"""运行系统"""asyncio.run(self.main_loop())if __name__ == "__main__":system = SmartACSystem()system.run()