智能家居系统基于 Home Assistant 核心，集成小智 AI 语音交互、Frigate 视频监控及各类 IoT 设备。采用树莓派运行 HAOS，通过 FRP 实现公网访问。前端使用 ESP32 连接小智服务器，后端支持 Zigbee、WiFi 及小米设备接入。包含 Music Assistant 本地音乐播放、自定义 Python 脚本抓取播客生成 RSS 源。监控部分利用 Intel CPU 硬解及 RTSP 推流，整合甲醛传感器、3D 打印机摄像头等设备，实现家庭自动化管理。
概述： 总体而言借用各种开源项目，致力做到好用可控。输入设备小智 AI 作为用户前端，后端用小智 AI 华南理工开源服务器。智能家庭中控采用树莓派 5 搭载 HAOS，用 Home Assistant，包括手机 APP。各种终端设备，支持 Zigbee 通信协议、WiFi 通信协议，小米设备支持 Miloco 的设备可以连接，其他类似美的海尔的设备也看 Home Assistant 的插件支持程度。
主机：ThinkPad S5 Yoga IP 地址：[IP 地址] 用户：[用户名]
| 服务 | 备注 | 服务端口 |
|---|---|---|
| frigate | 5000 | |
| samba | mnt/media/usbshare | |
| http | /mnt/usb_share/podcast | 10086 |
| mediamtx.service | [服务信息] | 8554 |
| napcat | ||
| koi | ||
| miloco | ||
| micam | [服务信息] | |
| astra-color.service | ||
| zaokafei-fetch.timer | /mnt/usb_share/podcast/zaokafei | |
| bambucam.service | Desktop/bambustudio/bambucam | 5004 |
| glances.service | 61208 |
主机：MacBook Air IP 地址：[IP 地址] 用户：[用户名]
| 服务 | 备注 | 服务端口 |
|---|---|---|
| intel-color | ||
| ir | ||
| switch | Desktop/swtich_rtsp/swtich_rtsp.py | |
| mediamtx | [服务信息] | 8554 |
主机：PC IP 地址：[IP 地址] 用户：[用户名]
前端若需要连 M3.5 的扬声器接口，可换 PCM5102A 模块。
重新配置 OTA 地址，开机时按 boot 键。
连接 Home Assistant 采用了很多尝试。 最先 MCP 连接方式：ha-mcp-for-xiaozhi/README.md at v0.1.1 · c1pher-cn/ha-mcp-for-xiaozhi 该方式能将多个 SSE 都连入 Home Assistant 封装为 function call 让 xiaozhi-server 调用，在 Home Assistant 端采用的是 Assist 提供的接口。
也尝试用 xiaozhi-server 贡献的 Home Assistant 代码，其实用了 Home Assistant 的 API，好处是可以精简 MCP，对音乐播放进行了一些修改。
树莓派地址：[IP 地址]
使用 FRP 加载项内网穿透至阿里云服务器，再 Nginx 反向代理出来。 感觉有一个 bug，重启 HAOS 之后，自启动 FRP Client，但是 FRP Server 会还在重新尝试原先的客户端，就会崩溃。因此必须开崩溃自动恢复。
此处用部署在局域网的 Samba 服务，用户名：[用户名]。
使用 Provider 模式，YouTube Music 等使用延迟较大，因此还是本地使用方案。
使用 Filesystem provider。
发现 RSS 真的还挺方便。
使用 Podcast provider。
也可以自己抓取节目，小宇宙的音频都可以抓比较方便。
抓取代码 Desktop/zaokafei
#!/usr/bin/env python3 # -*- coding: utf-8 -*-
import json
import os
import re
from datetime import datetime, timezone
from email.utils import format_datetime
from urllib.parse import quote
import requests
import xml.etree.ElementTree as ET
PODCAST_URL = "https://www.xiaoyuzhoufm.com/podcast/60de7c003dd577b40d5a40f3"
OUT_DIR = "/mnt/usb_share/podcast/zaokafei"
XML_PATH = os.path.join(OUT_DIR, "zaokafei.xml")
BASE_HTTP = "http://[本地 IP]:10086"
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
}
def get_next_data(url: str) -> dict:
html = requests.get(url, headers=HEADERS, timeout=30).text
m = re.search(r'<script type="application/json">(.*?)</script>', html, re.S,)
if not m:
raise RuntimeError("没找到 __NEXT_DATA__，小宇宙页面结构可能变了")
return json.loads(m.group(1))
def sanitize_filename(name: str) -> str:
name = re.sub(r'[\\/:*?"<>|]', "_", name).strip()
name = re.sub(r"\s+", " ", name)
return name[:160] if len(name) > 160 else name
def parse_pubdate_to_utc(pub: str) -> datetime:
try:
if pub.endswith("Z"):
dt = datetime.fromisoformat(pub.replace("Z", "+00:00"))
else:
dt = datetime.fromisoformat(pub)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(timezone.utc)
except Exception:
return datetime.now(timezone.utc)
def download_file(url: str, path: str):
r = requests.get(url, headers=HEADERS, stream=True, timeout=180)
r.raise_for_status()
tmp = path + ".part"
with open(tmp, "wb") as f:
for chunk in r.iter_content(chunk_size=1024 * 256):
if chunk:
f.write(chunk)
os.replace(tmp, path)
def ensure_xml_exists():
if os.path.exists(XML_PATH):
return
raise RuntimeError(f"找不到 {XML_PATH}\n请先把 zaokafei.xml 放到 {OUT_DIR} 下")
def xml_has_guid(root: ET.Element, guid: str) -> bool:
for item in root.findall("./channel/item"):
g = item.findtext("guid")
if g and g.strip() == guid:
return True
return False
def add_item_to_xml(title: str, guid: str, pub_dt_utc: datetime, enclosure_url: str, length_bytes: int):
ET.register_namespace("itunes", "http://www.itunes.com/dtds/podcast-1.0.dtd")
tree = ET.parse(XML_PATH)
root = tree.getroot()
channel = root.find("channel")
if channel is None:
raise RuntimeError("XML 里没找到 <channel>")
if xml_has_guid(root, guid):
print("XML 已存在该期 guid，跳过写入：", guid)
return
item = ET.Element("item")
t = ET.SubElement(item, "title")
t.text = title
pub = ET.SubElement(item, "pubDate")
pub.text = format_datetime(pub_dt_utc)
g = ET.SubElement(item, "guid", attrib={"isPermaLink": "false"})
g.text = guid
ET.SubElement(
item, "enclosure",
attrib={
"url": enclosure_url,
"length": str(length_bytes),
"type": "audio/mp4",
},
)
first_item = channel.find("item")
if first_item is None:
channel.append(item)
else:
children = list(channel)
idx = children.index(first_item)
channel.insert(idx, item)
tree.write(XML_PATH, encoding="UTF-8", xml_declaration=True)
print("已更新 XML：", XML_PATH)
def main():
os.makedirs(OUT_DIR, exist_ok=True)
ensure_xml_exists()
data = get_next_data(PODCAST_URL)
podcast = data["props"]["pageProps"]["podcast"]
latest = podcast["episodes"][0]
eid = latest["eid"]
title = latest["title"]
pub_raw = latest.get("pubDate", "")
audio_url = (
latest.get("media", {}).get("source", {}).get("url") or latest.get("enclosure", {}).get("url")
)
if not audio_url:
raise RuntimeError("没找到 media.source.url / enclosure.url")
pub_dt_utc = parse_pubdate_to_utc(pub_raw)
date_str = pub_dt_utc.strftime("%Y-%m-%d")
safe_title = sanitize_filename(title)
filename = f"{date_str} - {safe_title}.m4a"
out_path = os.path.join(OUT_DIR, filename)
print("最新一期：", title)
print("EID：", eid)
print("发布时间 (UTC):", pub_dt_utc.isoformat())
print("音频：", audio_url)
if os.path.exists(out_path):
print("文件已存在，跳过下载：", out_path)
else:
download_file(audio_url, out_path)
print("下载完成：", out_path)
enclosure_url = f"{BASE_HTTP}{quote('/' + filename)}"
length_bytes = os.path.getsize(out_path)
add_item_to_xml(
title=title,
guid=eid,
pub_dt_utc=pub_dt_utc,
enclosure_url=enclosure_url,
length_bytes=length_bytes,
)
if __name__ == "__main__":
main()
然后放一个 xml 文件
<?xml version='1.0' encoding='UTF-8'?>
<rss xmlns:itunes="http://www.itunes.com/dtds/podcast-1.0.dtd" version="2.0">
<channel>
<title>声动早咖啡</title>
<link>http://[本地 IP]:10086/</link>
<description>一个十五分钟的晨间仪式，轻松同步日常生活与商业世界。</description>
<itunes:image href="http://[本地 IP]:10086/cover.png" />
<image>
<url>http://[本地 IP]:10086/cover.png</url>
<title>声动早咖啡</title>
<link>http://[本地 IP]:10086/</link>
</image>
<!-- 示例条目 -->
</channel>
</rss>
把自己的文件夹地址代理出来就可以。
摄像头管理 NVR，目前跑在 I5-5500U，M840 上，其实无法硬解 H265 视频，并且做 detect 其实也比较吃力。 官方地址：Frigate 官方文档
Intel CPU:
在我的配置中，用 VAAPI 对 H264 硬解，对 H265 软解。
hwaccel_args: preset-vaapi
有时候会报错，需要指定设备及视频格式。
hwaccel_args: -hwaccel vaapi -hwaccel_device /dev/dri/renderD128 -hwaccel_output_format yuv420p
RTSP 推流，但是用 QSV 进行硬编码。
因为该音箱是一个有源音箱，连接方式有蓝牙和 3.5mm 连接两种。
Music Assistant 安装 Snapcast provider。
在终端设备安装 Snapcast 客户端。
使用 ZH08 CH2O 传感器，是电化学传感器。
ze08-ch2o.yaml
esphome:
name: ze08-ch2o
friendly_name: ZE08 CH2O
esp32:
board: esp32dev
framework:
type: arduino
wifi:
ssid: "[SSID]"
password: "[PASSWORD]"
logger:
baud_rate: 0
api:
ota:
- platform: esphome
external_components:
- source: type: local path: components
components: [ze08_ch2o]
uart:
id: uart_ze08
rx_pin: GPIO3
tx_pin: GPIO1
baud_rate: 9600
data_bits: 8
parity: NONE
stop_bits: 1
ze08_ch2o:
id: ze08_sensor
uart_id: uart_ze08
ch2o_ppb:
name: "CH2O (ppb)"
ch2o_mg_m3:
name: "CH2O (mg/m³)"
sensor:
- platform: template
name: dummy_sensor
id: dummy_sensor
lambda: |- return 0.0;
update_interval: 1h
internal: true
init.py
import esphome.codegen as cg
import esphome.config_validation as cv
from esphome.components import uart, sensor
from esphome.const import CONF_ID
ze08_ns = cg.esphome_ns.namespace("ze08_ch2o")
ZE08CH2OUart = ze08_ns.class_("ZE08CH2OUart", cg.Component, uart.UARTDevice)
CONF_PPB = "ch2o_ppb"
CONF_MGM3 = "ch2o_mg_m3"
CONFIG_SCHEMA = cv.Schema({
cv.GenerateID(): cv.declare_id(ZE08CH2OUart),
cv.Required(CONF_PPB): sensor.sensor_schema(unit_of_measurement="ppb", accuracy_decimals=0, icon="mdi:molecule"),
cv.Required(CONF_MGM3): sensor.sensor_schema(unit_of_measurement="mg/m³", accuracy_decimals=3, icon="mdi:molecule"),
}).extend(uart.UART_DEVICE_SCHEMA).extend(cv.COMPONENT_SCHEMA)
async def to_code(config):
var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config)
await uart.register_uart_device(var, config)
ppb = await sensor.new_sensor(config[CONF_PPB])
mgm3 = await sensor.new_sensor(config[CONF_MGM3])
cg.add(var.set_sensors(ppb, mgm3))
ze08_ch2o_uart.h
#pragma once
#include "esphome/core/component.h"
#include "esphome/components/uart/uart.h"
#include "esphome/components/sensor/sensor.h"
namespace esphome {
namespace ze08_ch2o {
class ZE08CH2OUart : public Component, public uart::UARTDevice {
public:
ZE08CH2OUart() : uart::UARTDevice(nullptr) {}
explicit ZE08CH2OUart(uart::UARTComponent *parent) : uart::UARTDevice(parent) {}
void set_sensors(sensor::Sensor *ppb, sensor::Sensor *mgm3) {
ch2o_ppb_ = ppb;
ch2o_mg_m3_ = mgm3;
}
void loop() override {
while (available()) {
uint8_t b = read();
if (idx_ == 0) {
if (b != 0xFF) continue;
buf_[idx_++] = b;
continue;
}
buf_[idx_++] = b;
if (idx_ >= 9) {
parse_frame_();
idx_ = 0;
}
}
}
protected:
sensor::Sensor *ch2o_ppb_{nullptr};
sensor::Sensor *ch2o_mg_m3_{nullptr};
uint8_t buf_[9]{0};
uint8_t idx_{0};
void parse_frame_() {
if (buf_[0] != 0xFF) return;
if (buf_[1] != 0x17) return;
if (buf_[2] != 0x04) return;
uint16_t sum = 0;
for (int i = 1; i <= 7; i++) sum += buf_[i];
uint8_t cs = (uint8_t)(~(sum & 0xFF) + 1);
if (cs != buf_[8]) return;
uint16_t ppb = ((uint16_t)buf_[4] << 8) | buf_[5];
float mgm3 = (ppb / 1000.0f) * 1.25f;
if (ch2o_ppb_) ch2o_ppb_->publish_state(ppb);
if (ch2o_mg_m3_) ch2o_mg_m3_->publish_state(mgm3);
}
};
} // namespace ze08_ch2o
} // namespace esphome
采用 QSV 硬编码。
夜视模式根据灰度直方图判断自动切换，夜里用 IR 流和 RGB 流，同一个流输出。
由于小米摄像头本身无提供 RTSP 视频流，使用小米官方开源的 Miloco 项目接受 WebSocket 流，之后三方的 Micam 将此流转换为 RTSP 视频流，官方用 Go2rtc 推流，我这边改用了 Mediamtx，另外小米 4C 为 H265 编码，Docker 运行。
使用插件：Integration Overview
根据官方配置，能进行的操作似乎蛮多。 开启局域网模式以后获取视频流方法。
但我是 A1 mini 似乎智能获取 HTTP 视频流，效果如下。
淘宝购入 Zigbee 信号接收器插在跑 HA 的树莓派上。
在 Home Assistant/configuration.yaml 配置。
终端设备淘宝非常多，几种通信协议的设备都能够支持，参考店家说明配置没啥难度。
Tasmota 设置 Set'Option 为 OFF 时为 switch，设置为 ON 时为 light。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。 在线工具，HTML转Markdown在线工具，online