Chapter 12 · 树莓派与嵌入式 Linux
综合实战：家庭自动化中枢
把前 11 章的技艺熔于一炉：传感器采集 → MQTT 通信 → 数据入库 → Web 面板可视化 → 常驻自启。从零搭一个真正跑得起来、断电重启也能自愈的家庭自动化中枢。
把前 11 章的技艺熔于一炉：传感器采集 → MQTT 通信 → 数据入库 → Web 面板可视化 → 常驻自启。从零搭一个真正跑得起来、断电重启也能自愈的家庭自动化中枢。
sqlite3 模块直接用，无需装服务器。适合树莓派存传感器时序数据（时间戳+读数）。数据量大或高频写入时可换 InfluxDB 等时序库，但入门 SQLite 足够，且省资源。SIGTERM 信号。捕获它做清理（断开 MQTT、关闭数据库、断电 GPIO），避免数据损坏或硬件异常。Python 用 signal 模块注册处理函数。生产级服务必须优雅退出，不能一杀了之。.env / config.yaml）或环境变量，而非硬编码在代码里。换环境改配置即可，密码不进 Git。这是可维护、可移植部署的基本功。把前 11 章的能力组装成一个完整的数据流水线——从传感器到手机屏幕：
整合第 5、11 章。读传感器，通过 MQTT 发布，配置与代码分离：
# sensor_node.py — 采集传感器并发布到 MQTT import os, json, time, signal, sys import paho.mqtt.client as mqtt import board, adafruit_bmp280 # 配置从环境变量读（配置与代码分离） BROKER = os.getenv("MQTT_BROKER", "localhost") USER = os.getenv("MQTT_USER", "piuser") PASSWD = os.getenv("MQTT_PASS", "") # 初始化传感器 i2c = board.I2C() sensor = adafruit_bmp280.Adafruit_BMP280_I2C(i2c, address=0x76) client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="sensor-node") client.username_pw_set(USER, PASSWD) client.will_set("home/livingroom/status", "offline", retain=True) client.connect(BROKER, 1883, 60) client.loop_start() client.publish("home/livingroom/status", "online", retain=True) # 优雅退出：收到 SIGTERM 时清理 def shutdown(sig, frame): client.publish("home/livingroom/status", "offline", retain=True) client.loop_stop(); client.disconnect() sys.exit(0) signal.signal(signal.SIGTERM, shutdown) signal.signal(signal.SIGINT, shutdown) while True: payload = json.dumps({ "temp": round(sensor.temperature, 1), "pressure": round(sensor.pressure, 1), "ts": int(time.time()), }) client.publish("home/livingroom/temp", payload, qos=1, retain=True) time.sleep(10)
中枢订阅数据，写入 SQLite，并执行自动化规则（超温发控制指令）：
# collector.py — 订阅、入库、跑规则 import sqlite3, json, os import paho.mqtt.client as mqtt DB = "/home/pi/hub/data.db" FAN_THRESHOLD = float(os.getenv("FAN_TEMP", "28")) # 建表（若不存在） conn = sqlite3.connect(DB, check_same_thread=False) conn.execute("""CREATE TABLE IF NOT EXISTS readings( ts INTEGER, room TEXT, temp REAL, pressure REAL)""") conn.commit() def on_connect(c, u, f, rc, p): c.subscribe("home/+/temp", qos=1) def on_message(c, u, msg): d = json.loads(msg.payload) room = msg.topic.split("/")[1] # 入库 conn.execute("INSERT INTO readings VALUES (?,?,?,?)", (d["ts"], room, d["temp"], d.get("pressure"))) conn.commit() # 规则引擎：超温 → 发开风扇指令 if d["temp"] > FAN_THRESHOLD: c.publish(f"home/{room}/cmd/fan", "ON", qos=1) print(f"{room} {d['temp']}°C 超阈值，开风扇") else: c.publish(f"home/{room}/cmd/fan", "OFF", qos=1) client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="hub-collector") client.username_pw_set(os.getenv("MQTT_USER"), os.getenv("MQTT_PASS")) client.on_connect = on_connect client.on_message = on_message client.connect(os.getenv("MQTT_BROKER", "localhost"), 1883, 60) client.loop_forever()
执行层节点（第 4/6 章）订阅 home/+/cmd/fan，收到 ON 就用 gpiozero 打开继电器/风扇，形成闭环。
用 Flask 读数据库，手机浏览器看实时数据：
# web.py — Flask 实时数据面板 from flask import Flask, jsonify, render_template_string import sqlite3 app = Flask(__name__) DB = "/home/pi/hub/data.db" PAGE = """<!DOCTYPE html><html><head><meta charset="utf-8"> <title>家庭中枢</title><meta http-equiv="refresh" content="10"></head> <body style="font-family:sans-serif;padding:2rem"> <h1>🏠 家庭自动化中枢</h1> <p>客厅温度：<b>{{temp}}°C</b> 气压：{{pressure}} hPa</p> <p>更新时间：{{ts}}</p></body></html>""" def latest(): conn = sqlite3.connect(DB) row = conn.execute( "SELECT temp,pressure,ts FROM readings ORDER BY ts DESC LIMIT 1" ).fetchone() conn.close() return row @app.route("/") def index(): t, p, ts = latest() or ("--", "--", "--") return render_template_string(PAGE, temp=t, pressure=p, ts=ts) @app.route("/api/latest") def api(): # 给前端图表/App 用的 JSON 接口 t, p, ts = latest() or (None,) * 3 return jsonify(temp=t, pressure=p, ts=ts) if __name__ == "__main__": # 0.0.0.0 让局域网可访问 app.run(host="0.0.0.0", port=8000)
用第 9 章 systemd 把三个进程都托管为自启服务。以中枢为例：
# /etc/systemd/system/hub-collector.service [Unit] Description=家庭中枢数据采集与规则引擎 After=network-online.target mosquitto.service Wants=network-online.target [Service] Type=simple User=pi WorkingDirectory=/home/pi/hub EnvironmentFile=/home/pi/hub/.env # 配置分离 Environment=PYTHONUNBUFFERED=1 ExecStart=/home/pi/hub/env/bin/python /home/pi/hub/collector.py Restart=always RestartSec=5 [Install] WantedBy=multi-user.target
# 三个服务全部启用自启（sensor_node / collector / web） sudo systemctl daemon-reload sudo systemctl enable --now hub-collector hub-web sensor-node # 查看整体运行状态 systemctl status hub-collector hub-web sensor-node # 实时看中枢日志 journalctl -u hub-collector -f # 手机浏览器打开 http://raspberrypi.local:8000 看面板
更优雅的方式是把 Mosquitto、collector、web 全写进一份 compose.yaml（第 10 章），docker compose up -d 一键拉起、统一管理、换机秒迁移。硬件访问节点（sensor_node）用 --device 映射 I2C/GPIO。两种方式都行：systemd 更贴近系统、Docker 更易迁移，按喜好选。
让中枢无人值守长期稳定运行，上线前逐项确认：
无人值守设备最怕三件事：① SD 卡损耗——频繁写入几个月就坏，务必把数据库/日志迁到 USB SSD，或开 log2ram；② 磁盘塞满——日志和数据无限增长最终撑爆，配日志轮转和数据清理；③ 过热降频——夏天或满载时 CPU 超 80°C 掉速甚至重启，加散热并用 vcgencmd measure_temp 监控。这三点做不好，Demo 能跑但撑不过一个季度。
十二章走下来，你把树莓派从一块陌生的电路板，变成了能采集、通信、存储、展示、自动响应的家庭自动化中枢。回顾这条路：认识单板计算机与 GPIO（1、4 章）→ 烧录系统并用 Linux 命令驾驭它（2、3 章）→ 接 I2C/SPI/1-Wire 传感器感知世界、用 PWM 驱动电机、摄像头视觉（5、6、7 章）→ SSH 远程掌控、systemd 常驻自启、Docker 容器化部署（8、9、10 章）→ MQTT 联网上云，最终综合实战把一切串成一套断电自愈、无人值守的系统（11、12 章）。树莓派的魅力正在于此——它既是学习 Linux 与嵌入式的最佳教具，也是能真正跑在你家角落、日夜工作的生产设备。接下来，把这套架构套用到你自己的点子上：气象站、宠物投喂器、NAS、家庭监控、边缘 AI 推理……硬件与软件的边界，就在你的想象力那里。愿你玩得尽兴。