简介这份资源面向智能家居、嵌入式与深度学习方向的开发者及学生提供一套基于深度神经网络的终身学习智能家居系统完整实现。系统以语音交互为核心通过采集用户日常调节指令存入数据库并持续训练实现个性化环境管理同时结合光照与温度传感器由智能管家自动调节灯具亮度及空调或地暖温度构建舒适居住环境。压缩包共30个文件约1.98MB包含17个png界面与效果图、8个py核心脚本、3个pyc编译文件及2个html网页模板覆盖语音处理、数据采集、线程调度与Web控制等模块目录结构清晰。目前已有54人学习下载适合希望研究终身学习机制、传感器联动与语音控制落地的读者参考可据此快速理解系统架构、复用关键脚本并搭建自己的智能家居实验环境。1. 拆开这个终身学习智能家居包它到底能跑通什么智能家居 Demo 满地都是但大多数是「手机点一下灯亮」的遥控器套壳。这个包不一样的地方在于它把「终身学习」四个字落到了代码里——语音指令不只是执行还会被采集、入库、再训练让系统慢慢摸清你的作息习惯。整套东西跑在树莓派上Python 写的后端前端是 Flask 模板传感器数据走数据库模型用深度神经网络做推理和增量更新。如果你手头正好有块树莓派、几个传感器想搭一套能自己「长脑子」的家居控制原型这个包值得拆开看。它适合嵌入式方向的学生做课程设计也适合想了解边缘端在线学习怎么落地的开发者。下面我按实际复现顺序把目录结构、数据流、训练逻辑和踩过的坑一条条讲清楚。2. 目录结构与数据流从 read.py 到 Web_Control.html 的完整链路拿到一个 zip 包我习惯先看目录树再动手。这个包的根目录叫raspberry-pie-master名字里带 pie 是树莓派的常见拼写变体不影响运行。核心文件分四类传感器采集、线程调度、Web 服务、模型推理。下面按数据从硬件到界面的流向拆。2.1 文件清单与职责划分先看根目录下的 Python 文件每个都有明确分工文件名职责关键依赖read.py读取传感器原始数据光照、温度GPIO 库、ADC 驱动cap.py摄像头/音频采集入口OpenCV、PyAudiovideo_handler.py视频帧处理与手势/动作识别OpenCV、NumPymyThread.py线程管理周期性采集任务threadingthread_run.py线程启动入口协调各采集线程threadingdata_timers.py定时器逻辑控制采集频率time、schedulea.py模型推理与训练主逻辑PyTorch 或 TensorFlowapp.pyFlask Web 服务路由与接口Flasktemplates/前端页面模板Jinja2static/静态资源CSS、JS、图片—images/采集的图片样本存储—__pycache__/Python 字节码缓存—__pycache__里能看到video_handler.cpython-39.pyc、myThread.cpython-39.pyc、cap.cpython-39.pyc说明原作者用的是 Python 3.9。如果你本地是 3.10 或 3.11重新编译即可不影响源码逻辑。2.2 数据采集到入库的链路整个系统的数据流是这样的传感器 →read.py→data_timers.py定时触发 →myThread.py写入数据库 →a.py读取训练 →app.py暴露接口 →Web_Control.html展示与控制。先看采集层。read.py负责从 GPIO 读取光照和温度传感器的模拟量常见做法是用 MCP3008 这类 ADC 芯片转成数字量。代码结构大致如下# read.py 传感器读取核心逻辑 import spidev import time class SensorReader: def __init__(self, bus0, device0): self.spi spidev.SpiDev() self.spi.open(bus, device) self.spi.max_speed_hz 1350000 # MCP3008 推荐速率 def read_channel(self, channel): 读取指定通道的 ADC 值channel 范围 0-7 if channel 0 or channel 7: raise ValueError(通道号必须在 0-7 之间) adc self.spi.xfer2([1, (8 channel) 4, 0]) value ((adc[1] 3) 8) adc[2] return value def read_light(self): 光照传感器接在通道 0 raw self.read_channel(0) # 将 0-1023 映射到 0-100 的百分比 return round(raw / 1023 * 100, 2) def read_temp(self): 温度传感器接在通道 1假设为 TMP36 raw self.read_channel(1) voltage raw * 3.3 / 1023 # TMP36 转换公式温度 (电压 - 0.5) * 100 return round((voltage - 0.5) * 100, 2)这里max_speed_hz设成 1350000 是 MCP3008 数据手册的推荐值设太高会读到乱码。read_light把 10 位 ADC 值归一化到百分比read_temp按 TMP36 的线性公式换算。如果你用的是 DHT11 这类数字传感器就不需要 ADC直接读 GPIO 高低电平时序即可但代码要换成对应的驱动库。采集频率由data_timers.py控制。常见做法是每 30 秒采一次太密会撑爆数据库太疏则训练样本不够。myThread.py把采集任务放到独立线程里避免阻塞 Flask 主线程# myThread.py 采集线程封装 import threading import time from read import SensorReader class CollectThread(threading.Thread): def __init__(self, db_writer, interval30): super().__init__() self.db_writer db_writer self.interval interval self._stop_event threading.Event() self.sensor SensorReader() def run(self): while not self._stop_event.is_set(): light self.sensor.read_light() temp self.sensor.read_temp() # 写入数据库表结构见 2.3 节 self.db_writer.insert(light, temp, time.time()) time.sleep(self.interval) def stop(self): self._stop_event.set()interval参数建议根据实际场景调光照变化慢60 秒也够温度如果开空调30 秒能捕捉到趋势。_stop_event是标准做法比while True加break更干净退出时不会残留线程。2.3 数据库表结构与训练数据准备数据库这块包里的代码没有指定具体数据库类型常见做法是用 SQLite因为树莓派上零配置。建表语句如下-- 传感器数据表存储原始采集记录 CREATE TABLE IF NOT EXISTS sensor_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, light REAL NOT NULL, temperature REAL NOT NULL, timestamp REAL NOT NULL, user_adjust_light REAL, -- 用户手动调节后的光照值 user_adjust_temp REAL -- 用户手动调节后的温度值 ); -- 语音指令记录表 CREATE TABLE IF NOT EXISTS voice_commands ( id INTEGER PRIMARY KEY AUTOINCREMENT, command_text TEXT NOT NULL, intent_label TEXT, -- 意图分类标签 timestamp REAL NOT NULL );user_adjust_light和user_adjust_temp这两列是关键。当用户觉得当前光照不合适、手动调了灯系统就把调节后的值记下来作为监督学习的标签。训练时输入是传感器读数加时间特征输出是用户偏好的目标值。这样模型学的是「什么时间、什么环境光下用户喜欢多亮」而不是简单的阈值判断。语音指令表里的intent_label需要预先定义几个类别比如「开灯」「关灯」「调亮」「调暗」「升温」「降温」。a.py里的模型会对语音转文字的结果做意图分类分类结果再映射到具体控制指令。3. 深度神经网络与终身学习a.py 里的模型怎么搭、怎么更新这个包的核心卖点在「终身学习」也就是模型不是一次训练完就固定而是随着新数据不断微调。下面拆a.py里的模型结构和更新策略。3.1 网络结构选型与输入特征工程输入特征不能只给光照和温度两个值那样模型学不到时间规律。常见做法是把时间编码进去小时、分钟、星期几做归一化或独热编码。光照和温度做标准化减均值除标准差。输出是用户偏好的光照值和温度值两个回归头。网络结构用全连接就够不需要上 CNN 或 RNN因为输入维度低、样本量不会太大。典型结构如下# a.py 模型定义 import torch import torch.nn as nn class PreferenceNet(nn.Module): def __init__(self, input_dim8, hidden_dim64): super().__init__() self.shared nn.Sequential( nn.Linear(input_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU() ) self.light_head nn.Linear(hidden_dim, 1) # 预测偏好光照 self.temp_head nn.Linear(hidden_dim, 1) # 预测偏好温度 def forward(self, x): feat self.shared(x) light self.light_head(feat) temp self.temp_head(feat) return light, tempinput_dim8对应光照、温度、小时 sin、小时 cos、星期 sin、星期 cos、光照变化率、温度变化率。用 sin/cos 编码时间比直接给 0-23 的整数更好模型能理解 23 点和 0 点是相邻的。hidden_dim64是经验值样本少的时候可以降到 32避免过拟合。3.2 终身学习的实现弹性权重巩固与回放缓冲终身学习最大的坑是「灾难性遗忘」——模型学了新数据旧知识全丢了。这个包里a.py的实现思路常见做法是两种一是弹性权重巩固EWC在损失函数里加一项惩罚限制重要权重变化太大二是回放缓冲把旧样本存一部分每次训练混着新数据一起喂。EWC 的损失函数写法# a.py 训练循环中的 EWC 惩罚项 def compute_ewc_loss(model, fisher_matrix, old_params, lambda_ewc0.1): 计算 EWC 惩罚项限制重要参数偏移 loss 0 for name, param in model.named_parameters(): if name in fisher_matrix: # Fisher 信息矩阵越大说明该参数越重要 loss (fisher_matrix[name] * (param - old_params[name]) ** 2).sum() return lambda_ewc * losslambda_ewc控制惩罚力度设太大模型学不动新数据设太小旧知识保不住。我一般从 0.1 开始试看新旧任务上的表现再调。fisher_matrix在每次任务训练完后计算存下来给下一次用。回放缓冲更简单粗暴维护一个固定大小的样本池新样本进来就替换掉最老的# a.py 回放缓冲实现 from collections import deque import random class ReplayBuffer: def __init__(self, capacity500): self.buffer deque(maxlencapacity) def add(self, features, light_label, temp_label): self.buffer.append((features, light_label, temp_label)) def sample(self, batch_size): batch random.sample(self.buffer, min(batch_size, len(self.buffer))) return zip(*batch)capacity500是权衡树莓派内存有限存太多会吃紧存太少回放效果差。实际跑的时候每次训练从缓冲里抽 32 条旧样本混着新样本一起算损失。3.3 训练触发条件与模型保存终身学习不是每来一条数据就训练一次那样计算量扛不住。常见做法是攒够一批再触发比如新增样本超过 50 条或者距离上次训练超过 24 小时。data_timers.py里可以加一个计数器# data_timers.py 训练触发逻辑 class TrainTrigger: def __init__(self, threshold50, max_interval_hours24): self.threshold threshold self.max_interval max_interval_hours * 3600 self.new_samples 0 self.last_train_time time.time() def should_train(self): if self.new_samples self.threshold: return True if time.time() - self.last_train_time self.max_interval: return True return False def on_new_sample(self): self.new_samples 1 def on_train_done(self): self.new_samples 0 self.last_train_time time.time()模型保存用torch.save文件名带时间戳方便回滚# a.py 模型保存 import os from datetime import datetime def save_model(model, save_dirmodels): os.makedirs(save_dir, exist_okTrue) timestamp datetime.now().strftime(%Y%m%d_%H%M%S) path os.path.join(save_dir, fpreference_net_{timestamp}.pth) torch.save(model.state_dict(), path) return path保留最近 5 个版本就行多了占空间。如果新模型在验证集上表现变差可以回滚到上一个版本。4. Flask 控制面板与语音链路app.py 和 Web_Control.html 怎么联动后端模型跑通后需要一个界面让用户看到当前状态、手动调节、发语音指令。app.py是 Flask 入口templates/Web_Control.html是主控制页。4.1 Flask 路由设计与接口定义app.py里至少要有这几个路由首页渲染、传感器数据查询、控制指令下发、语音上传。代码骨架# app.py Flask 路由 from flask import Flask, render_template, jsonify, request import sqlite3 app Flask(__name__) def get_db(): conn sqlite3.connect(smart_home.db) conn.row_factory sqlite3.Row return conn app.route(/) def index(): return render_template(Web_Control.html) app.route(/api/sensor/latest) def latest_sensor(): conn get_db() row conn.execute( SELECT * FROM sensor_data ORDER BY timestamp DESC LIMIT 1 ).fetchone() conn.close() if row: return jsonify(dict(row)) return jsonify({error: no data}), 404 app.route(/api/control, methods[POST]) def control(): data request.get_json() device data.get(device) # light 或 ac value data.get(value) # 目标值 # 这里调用 GPIO 控制函数同时记录用户调节 # control_device(device, value) # record_user_adjust(device, value) return jsonify({status: ok}) app.route(/api/voice, methods[POST]) def voice(): audio request.files.get(audio) if not audio: return jsonify({error: no audio}), 400 # 保存音频调用语音识别再走意图分类 # text speech_to_text(audio) # intent classify_intent(text) return jsonify({status: received})/api/sensor/latest返回最新一条传感器数据前端定时轮询刷新。/api/control接收 JSON 格式的控制指令同时把用户调节记录到数据库供后续训练用。/api/voice接收音频文件走语音识别和意图分类。4.2 前端轮询与语音采集Web_Control.html里用 JavaScript 定时拉数据并用 MediaRecorder API 采集语音// Web_Control.html 前端逻辑 async function fetchSensor() { const res await fetch(/api/sensor/latest); const data await res.json(); document.getElementById(light-value).textContent data.light %; document.getElementById(temp-value).textContent data.temperature °C; } setInterval(fetchSensor, 5000); // 每 5 秒刷新一次 // 语音采集 let mediaRecorder; let audioChunks []; async function startRecording() { const stream await navigator.mediaDevices.getUserMedia({ audio: true }); mediaRecorder new MediaRecorder(stream); audioChunks []; mediaRecorder.ondataavailable e audioChunks.push(e.data); mediaRecorder.onstop async () { const blob new Blob(audioChunks, { type: audio/wav }); const formData new FormData(); formData.append(audio, blob, command.wav); await fetch(/api/voice, { method: POST, body: formData }); }; mediaRecorder.start(); } function stopRecording() { mediaRecorder.stop(); }轮询间隔 5 秒是折中太频繁增加树莓派负担太慢界面看起来卡。语音采集用浏览器原生 API不需要额外插件。录完直接 POST 到后端后端再做识别。4.3 语音识别与意图分类的衔接语音识别这块包里没有指定具体引擎。常见做法是本地跑一个轻量模型比如用 Vosk 做离线识别或者调用系统自带的语音接口。识别出文字后用a.py里的意图分类模型打标签。意图分类可以复用同一个网络加一个分类头# a.py 意图分类头 class IntentNet(nn.Module): def __init__(self, vocab_size, embed_dim32, num_intents6): super().__init__() self.embedding nn.Embedding(vocab_size, embed_dim) self.fc nn.Sequential( nn.Linear(embed_dim, 64), nn.ReLU(), nn.Linear(64, num_intents) ) def forward(self, x): # x 是 token id 序列取平均池化 emb self.embedding(x).mean(dim1) return self.fc(emb)num_intents6对应前面定义的六种指令。训练数据来自voice_commands表用户每次语音控制后如果手动纠正了就把纠正后的标签更新回去形成闭环。5. 避坑与排查树莓派上跑这套系统的五个血泪教训这套系统在树莓派上跑坑不少。下面五条是我实际复现时踩过的按「现象 → 原因 → 解决」写。5.1 传感器读数跳变严重现象光照值在 20% 和 80% 之间反复横跳温度偶尔跳到 80°C。原因SPI 读取没有加去抖或者 ADC 参考电压不稳。树莓派 GPIO 的 3.3V 供电在接多个传感器时会波动。解决软件上做滑动平均连续读 5 次取中位数硬件上给 ADC 加一个 0.1uF 的去耦电容。代码里加def read_stable(sensor_func, times5): values [sensor_func() for _ in range(times)] values.sort() return values[times // 2] # 取中位数5.2 数据库写入锁死现象跑几个小时后采集线程报database is locked数据不再更新。原因SQLite 默认只允许一个写连接Flask 的请求线程和采集线程同时写就冲突。解决给数据库连接加超时或者用 WAL 模式。在get_db里加conn sqlite3.connect(smart_home.db, timeout10) conn.execute(PRAGMA journal_modeWAL)WAL 模式允许多读一写冲突概率大幅降低。如果还不行就加一个写队列所有写操作串行化。5.3 模型训练时内存溢出现象训练跑几分钟后进程被 killdmesg显示 OOM。原因回放缓冲存了太多样本或者 DataLoader 的num_workers设太大树莓派内存不够。解决把capacity降到 200batch_size降到 16num_workers设为 0主进程加载。树莓派 4B 只有 4GB 内存别按服务器的配置来。5.4 语音识别延迟高现象说完指令后要等 3-5 秒才响应。原因语音识别模型太大或者音频采样率太高。解决把采样率降到 16000Hz用轻量模型。如果还慢就在前端做端点检测只上传有效语音段别把静音也传上去。5.5 模型更新后旧指令失效现象训练完新模型后之前能识别的「开灯」指令突然不灵了。原因灾难性遗忘新数据把旧知识覆盖了。解决检查 EWC 的lambda_ewc是不是设太小或者回放缓冲里旧样本比例太低。我一般把新旧样本比例控制在 1:1lambda_ewc调到 0.5 试试。6. 进阶技巧用验证集守住模型底线终身学习系统最怕的是模型越训越差但你又不知道。我的习惯是每次训练前留出 10% 的旧数据做验证集训练完先跑验证指标下降就回滚。具体做法# a.py 训练后验证与回滚 def train_with_validation(model, new_data, val_data, old_model_path): # 保存当前模型作为备份 backup_path save_model(model, models/backup) # 训练 train(model, new_data) # 验证 old_loss evaluate(model, val_data) # 加载旧模型对比 model.load_state_dict(torch.load(old_model_path)) new_loss evaluate(model, val_data) if new_loss old_loss * 1.1: # 允许 10% 波动 # 回滚 model.load_state_dict(torch.load(backup_path)) return False return True1.1这个系数是容忍度设太紧会导致模型永远不更新设太松则回滚机制形同虚设。我一般根据验证集大小调样本少于 100 条时放宽到 1.2。另外模型文件命名带时间戳保留最近 5 个版本回滚时直接指定路径就行。数据库里也记一笔训练日志包括训练时间、样本数、验证损失方便回溯。从那以后我每次动模型之前都强制走一遍验证集哪怕只是微调一个参数。希望帮到你。本文还有配套的精品资源点击获取