简介这是一份面向物联网初学者的基于Python与ESP8266的智能家居系统完整源码包整体围绕智能灯光与智能锅炉两大子系统展开适合作为课程设计、毕业设计或项目实训的参考。包内共15个文件主体是4份Arduino.ino固件和3个Python程序分别完成灯光开关、人体感应、锅炉控制、温度采集以及App界面、数据库、MQTT通信逻辑另外附带Kivy界面文件.kv、项目说明书.pdf/README、Wokwi仿真入口和打包好的SmartHomeApp压缩包整体仅3.3MB便于快速部署与二次开发。资源目录将设备端固件和App端程序分开便于按模块对照学习。系统具备手机App远程控制、设备状态实时监控、规则化自动开关和用户验证功能能直观理解从ESP8266传感器到云MQTT再到客户端界面的完整链路。资源目前已有100人学习下载适合希望快速跑通智能家居demo并继续扩展功能的开发者。1. 一套基于Python的智能家居系统源码Python到底管哪几层把“源码基于Python的智能家居系统.zip”解压之后你大概率看不到漂亮的控制面板最先看到的是一堆core、device、handlers目录。用 Python 做智能家居和用 App 做智能家居是两条路App 方案把逻辑缝在云端源码方案要你在本地把设备接入、消息路由、控制 API 这三层全部打通。换句话说这套源码真正值钱的地方不是某一个传感器代码而是它怎么用 Python 调度一台常驻运行的进程让灯、温湿度计、开关面板在局域网里彼此发现、上报、被控制。如果你已经跑过 Python 入门教程想找个比爬虫更能练长连接的项目这个方向就很合适它逼着你把多线程、队列、协议解析和数据库读写串在一起处理。2. 智能家居系统源码的整体结构与消息链路2.1 为什么第一层选 MQTT 而不是 HTTP 轮询设备端与网关的关系是“多对一、有长连接、要实时”。HTTP 轮询的痛点很明显传感器每 5 秒拉一次请求网关要处理大量无效报文设备侧也因为频繁建连把电量耗在握手阶段。MQTT 的发布/订阅模型把通信拆成 topic设备的动作和状态都变成一条条消息网关和开源 HA 系统只需要订阅关心的主题。下面是两类协议的对比表对比项HTTP 轮询MQTT 长连接连接方式请求-响应发布-订阅实时性取决于轮询间隔消息即达设备端资源消耗每次建 TCP 连接保持一个连接断线重连无状态重发较难QoS 与 session 可恢复局域网内可用性需要知道每个设备 IP只认 broker 和主题这也是为什么很多源码包的核心位置都会看到mqtt目录它承载的不只是控制指令还包括设备心跳和 OTA 升级的通知。Python 里做这一层选 paho-mqtt 就足够了它是事实标准API 稳定Home Assistant 自己也没重新造轮子而是把 MQTT 作为默认接入协议之一。2.2 源码里那五个跑不掉的 Python 模块一个能运行的基于 Python 的智能家居系统源码目录一般逃不出下面这些职责目录/文件职责core/设备注册表、事件总线、定时任务protocols/MQTT、Modbus、串口协议适配devices/具体设备驱动如 DHT22、继电器、人体红外api/Flask/FastAPI 的本地控制 REST 接口storage/SQLite 或文件存储历史数据与场景配置我习惯先读core/registry.py因为它定义了设备类型与内存表结构。源码不管用了什么设计模式最后都要解决两个问题一条上报数据从 MQTT 回调进入内存后走什么路径一条控制指令从 API 进入系统后如何路由到设备。把这个路径画清楚代码就读懂一半。2.3 用 dataclass pydantic 统一设备模型少写一半重复代码设备上报数据五花八门温度、湿度、电量、开关状态混在同一个 JSON 里。如果每个 handler 都自己解析后续加一个设备类型就要动多处逻辑。常见做法是先定义统一的设备模型让网络层只负责传输业务层只认模型。from dataclasses import dataclass, field from typing import Optional dataclass class DeviceInfo: device_id: str device_type: str # switch / sensor / hub name: str room: str 默认房间 ip: str online: bool False extra: dict field(default_factorydict) dataclass class SensorReport: device_id: str timestamp: int temperature: Optional[float] None humidity: Optional[float] None power: Optional[int] None # 0 或 1指示灯/继电器状态 battery: Optional[int] None逻辑说明SensorReport里的可选字段让同一个模型能承载温湿度计和开关面板两种数据DeviceInfo.extra用来放厂家私有字段避免把未知字段直接堆到顶层。参数说明timestamp用毫秒级 epoch不要用字符串时间否则后面 SQLite 存历史数据排序会很别扭。实际生产里更推荐把它改成pydantic.BaseModel这样在extra里还能做字段约束局域网流量不大的场景用 dataclass 是为了少引一个依赖。3. 用 Python 把设备接入层写出来3.1 基于 paho-mqtt 的最小客户端先写一个最小网关订阅程序把 MQTT 回调接到 Python 的处理函数里。这里要注意 paho-mqtt 的版本差异新版 2.x 要求显式指定CallbackAPIVersion如果你的环境仍是 1.x去掉mqtt.CallbackAPIVersion.VERSION2这个参数即可。import json import paho.mqtt.client as mqtt BROKER_HOST 0.0.0.0 # MQTT broker 地址一般就是网关主机 IP BROKER_PORT 1883 KEEPALIVE 60 CLIENT_ID gateway-main # 同一局域网下必须唯一 def on_connect(client, userdata, flags, reason_code, properties): if reason_code 0: client.subscribe(home/devices//report, qos1) else: print(fmqtt connect failed: {reason_code}) def on_message(client, userdata, msg): topic msg.topic payload json.loads(msg.payload.decode(utf-8)) print(freceived {topic} - {payload}) client mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_idCLIENT_ID) client.on_connect on_connect client.on_message on_message client.connect(BROKER_HOST, BROKER_PORT, KEEPALIVE) client.loop_forever()逻辑说明home/devices//report使用通配符匹配任意设备 ID这样新设备接入时网关无需改订阅逻辑。qos1保证消息至少到达一次设备重复上报的问题放在存储层去重而不是靠 QoS 2 去推高网络延迟。参数说明KEEPALIVE设 60 秒设备需要在这个时间窗口内发一次 PINGREQ如果局域网质量较好可以拉长到 90减少无谓心跳。CLIENT_ID不能重复否则后连接的客户端会把先连接的踢下线这是智能家居源码里最常见的“设备掉线”原因。3.2 用 MicroPython 在 ESP32 上复现同一份协议逻辑设备端如果是 ESP32固件一般走 MicroPython它提供的umqtt.simple与 paho 的 API 接近但少了自动重连。from umqtt.simple import MQTTClient import ujson, time CLIENT_ID esp32-livingroom SERVER 192.168.1.100 # 网关 IP TOPIC_REPORT home/devices/esp32-livingroom/report def read_temp(): # 这里替换成你现有的 DHT22 或 DS18B20 读取函数 return 25.6 def report(): c MQTTClient(CLIENT_ID, SERVER, keepalive60) c.connect() while True: payload ujson.dumps({ device_id: CLIENT_ID, timestamp: time.time_ns() // 1_000_000, temperature: read_temp(), }) c.publish(TOPIC_REPORT, payload, qos1) time.sleep(10) report()逻辑说明设备端只负责采集和上报不做业务判断把决策权留给网关服务。参数说明time.time_ns() // 1_000_000是毫秒时间戳与前面定义的模型对齐keepalive60必须和网关侧约定的范围一致太短会频繁断连太长会让网关误判设备离线。注意umqtt.simple断线后不会自动重连量产时需要自己包一层while not c.connect()的重试循环还要记录连续失败次数避免 broker 下线时所有设备同时打爆网关。3.3 指令下发与状态上报的状态机控制链路里最容易出错的地方是对“下发后没有回应”的处理。不能把指令发出去就认为设备执行了也不能在回调里直接改数据库。我一般会用一个全局队列承接所有下发指令。import queue, time, threading, json cmd_queue queue.Queue(maxsize200) def send_toggle(device_id: str): cmd_queue.put({ topic: fhome/devices/{device_id}/cmd, payload: {cmd: toggle, seq: int(time.time() * 1000)} }) def worker(mqtt_client): while True: item cmd_queue.get() mqtt_client.publish(item[topic], json.dumps(item[payload]), qos1)逻辑说明用队列把 API 线程与 MQTT 线程解耦即使某个设备暂时离线指令也在队列里等待不会因为网络抖动丢失。参数说明maxsize200是一个保守值家庭环境同时下发的指令很难超过这个数如果队列满了put会抛异常你可以捕获后返回 503而不是让网关进程卡死。设备收到指令后需要回一条ack消息网关收到 ack 才把状态标记为“已执行”否则在等待 3 秒后触发超时重发。4. 用 Flask 搭建本地控制 API 和可视化面板4.1 一个能给 MQTT 发指令的 REST 接口控制端点不需要引 FastAPIFlask 在局域网场景够用。下面的接口接收外部请求并把指令投递到上面的队列import time from flask import Flask, request, jsonify app Flask(__name__) app.route(/api/devices/string:device_id/command, methods[POST]) def command_device(device_id): data request.get_json(forceTrue) cmd data.get(cmd) if cmd not in {on, off, toggle}: return jsonify({code: 400, msg: invalid cmd}), 400 send_toggle(device_id) return jsonify({code: 0, msg: accepted}), 202逻辑说明接口层只做参数校验和入队不做 MQTT 直发这样 Web 服务重启不会丢正在排队中的指令。参数说明返回 202 而不是 200语义更准确“已接受不一定已执行”。forceTrue会让 Flask 在请求头缺少Content-Type时也尝试解析 JSON局域网内调试方便但暴露到外网时建议去掉并严格校验。4.2 用 SQLite 存温度湿度历史数据的几个参数上报数据落地用 SQLite 足够关键参数是 WAL 模式和批处理提交。import sqlite3 DB_PATH data/history.db def init_db(): conn sqlite3.connect(DB_PATH) conn.execute(PRAGMA journal_modeWAL;) conn.execute( CREATE TABLE IF NOT EXISTS sensor_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, device_id TEXT NOT NULL, ts INTEGER NOT NULL, temperature REAL, humidity REAL, battery INTEGER ); ) conn.commit() conn.close() def insert_report(report): conn sqlite3.connect(DB_PATH) conn.execute( INSERT INTO sensor_log(device_id, ts, temperature, humidity, battery) VALUES (?, ?, ?, ?, ?), (report.device_id, report.timestamp, report.temperature, report.humidity, report.battery) ) conn.commit() conn.close()逻辑说明PRAGMA journal_modeWAL让读和写可以并行历史数据查询体验会明显好于默认的 rollback journal。参数说明每个上报都单独connect并不优雅但家庭设备每分钟上报量很小这样能避免多线程之间共享连接带来的锁问题。设备多了以后再改成连接池或者把 SQLite 换成 TimescaleDB。ts列要建索引否则查询三个月历史数据会全表扫描。4.3 Dashboard 只读 API 与自动刷新面板的数据接口只做读操作推荐让前端轮询而不是在网关服务里做 WebSocket 推送。智能家居源码做成轮询有个好处页面重开或刷新时不需要重新建立连接状态。import time import sqlite3 from flask import jsonify app.route(/api/dashboard/latest) def dashboard_latest(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row rows conn.execute( SELECT device_id, temperature, humidity, ts FROM sensor_log WHERE ts ? AND id IN ( SELECT MAX(id) FROM sensor_log GROUP BY device_id ) , (int(time.time() * 1000) - 600_000,)).fetchall() conn.close() return jsonify([dict(r) for r in rows])逻辑说明id IN (SELECT MAX(id)...)取每个设备最近一条上报而不是直接取最新行避免同设备乱序插入时拿到旧数据。逻辑上带了 10 分钟时间窗口过滤设备掉线后这个设备不会出现在面板里。读操作放在 Flask 主线程会偶尔卡顿所以网关服务启动时要把threadedTrue打开多个浏览器标签同时刷新时不会互相等锁。5. 接入 Home Assistant 生态前要做的自检和一个去重技巧5.1 用 HA 的 REST API 同步设备状态如果你的源码要接入 Home Assistant不必让 HA 直接订阅 MQTT。更可控的做法是让 HA 周期拉取网关的本地 API然后通过 HA REST API 写入 entity 状态。import requests HA_URL http://homeassistant.local:8123 HA_TOKEN Bearer your-long-lived-token resp requests.get(http://127.0.0.1:5000/api/dashboard/latest, timeout3) for dev in resp.json(): entity_id fsensor.temperature_{dev[device_id]} requests.post( f{HA_URL}/api/states/{entity_id}, headers{Authorization: HA_TOKEN}, json{state: dev[temperature], attributes: {unit: °C}} )逻辑说明先由 Python 网关做本地数据聚合再把最终结果同步到 HA调试时只需抓网关的日志不用同时看两边。参数说明timeout3必须设置否则 HA 的同步任务会被一个假死网关拖住阻塞后续所有状态更新。5.2 从源码到跑通先验证这 5 个点验证点通过标准网关订阅正常用 mosquitto_pub 手动发一条消息回调能打印设备端重连拔掉设备电源再插上网关日志出现上线记录指令执行闭环Web API 下发后收到设备 ack数据库不丢数据重启网关后历史记录仍然完整HA 状态同步HA 面板出现温度实体且 10 秒内刷新5.3 一个具体技巧用 seq 去重处理 MQTT 重复上报QoS 1 下同一消息可能被投递多次如果直接写库历史数据会产生重复点。最简单的办法是在设备上报模型里增加seq字段网关内存里记录每个device_id最后处理的seq。last_seq: dict[str, int] {} def dedup_report(report: dict) - bool: device_id report[device_id] seq report.get(seq, 0) if seq last_seq.get(device_id, 0): return True last_seq[device_id] seq return False逻辑说明新上报的seq大于等于记录值才允许继续走业务链路旧消息直接丢弃。参数说明last_seq只保存在内存里网关重启后会清空所以数据库层面仍建议保留UNIQUE(device_id, seq)约束做第二道防线。把dedup_report加在on_message的第一个判断位置再观察设备上报日志重复插入就会消失。把这条逻辑跑顺后再给设备端加 OTA 升级时也不会受到重放上报的影响。本文还有配套的精品资源点击获取