智能家居搭建笔记：Home Assistant 与小智 AI 集成

开源地址：78/xiaozhi-esp32: An MCP-based chatbot | 一个基于 MCP 的聊天机器人

前端若需要连 M3.5 的扬声器接口，可换 PCM5102A 模块。

重新配置 OTA 地址，开机时按 boot 键。

小智 ESP32 后端服务器

开源地址：xinnan-tech/xiaozhi-esp32-server: Backend service for xiaozhi-esp32, helps you quickly build an ESP32 device control server.

连接 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，对音乐播放进行了一些修改。

Home Assistant

树莓派地址：[IP 地址]

公网访问

使用 FRP 加载项内网穿透至阿里云服务器，再 Nginx 反向代理出来。 感觉有一个 bug，重启 HAOS 之后，自启动 FRP Client，但是 FRP Server 会还在重新尝试原先的客户端，就会崩溃。因此必须开崩溃自动恢复。

网络存储

此处用部署在局域网的 Samba 服务，用户名：[用户名]。

Music Assistant

使用 Provider 模式，YouTube Music 等使用延迟较大，因此还是本地使用方案。

歌曲下载地址

SQUID.WTF - Home lucida | music at internet speed

使用 Filesystem provider。

Podcast 收听地址

发现 RSS 真的还挺方便。

使用 Podcast provider。

链接：weekend-project-space/top-rss-list: 订阅人数最多的 rss 源，中文优质 rss 源

也可以自己抓取节目，小宇宙的音频都可以抓比较方便。

抓取代码 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>

把自己的文件夹地址代理出来就可以。

Frigate

摄像头管理 NVR，目前跑在 I5-5500U，M840 上，其实无法硬解 H265 视频，并且做 detect 其实也比较吃力。 官方地址：Frigate 官方文档

视频硬解适配

Intel CPU:

NVIDIA:Video Encode and Decode Support Matrix | NVIDIA Developer

在我的配置中，用 VAAPI 对 H264 硬解，对 H265 软解。

hwaccel_args: preset-vaapi

有时候会报错，需要指定设备及视频格式。

hwaccel_args: -hwaccel vaapi -hwaccel_device /dev/dri/renderD128 -hwaccel_output_format yuv420p

视频推流

RTSP 推流，但是用 QSV 进行硬编码。

各种嵌入式设备

Marshall 音箱

因为该音箱是一个有源音箱，连接方式有蓝牙和 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

刷机页面

链接：Web - ESPHome

Intel Realsense 摄像头/astra_color 摄像头

采用 QSV 硬编码。

夜视模式根据灰度直方图判断自动切换，夜里用 IR 流和 RGB 流，同一个流输出。

小米 4C 摄像头

由于小米摄像头本身无提供 RTSP 视频流，使用小米官方开源的 Miloco 项目接受 WebSocket 流，之后三方的 Micam 将此流转换为 RTSP 视频流，官方用 Go2rtc 推流，我这边改用了 Mediamtx，另外小米 4C 为 H265 编码，Docker 运行。

Miloco

开源地址：XiaoMi/xiaomi-miloco: Xiaomi Miloco

Micam

开源地址：miiot/micam: 🎦 Micam 是一个专为小米摄像头设计的 RTSP 桥接服务

拓竹 A1 mini

使用插件：Integration Overview

根据官方配置，能进行的操作似乎蛮多。 开启局域网模式以后获取视频流方法。

链接：jtessler/bambucam: Linux command line tool to access camera feed from a Bambu 3D printer

但我是 A1 mini 似乎智能获取 HTTP 视频流，效果如下。

ESPHOME 排插/MQTT 插座/Zigbee 插座

淘宝购入 Zigbee 信号接收器插在跑 HA 的树莓派上。

在 Home Assistant/configuration.yaml 配置。

终端设备淘宝非常多，几种通信协议的设备都能够支持，参考店家说明配置没啥难度。

Tasmota 设置 Set'Option 为 OFF 时为 switch，设置为 ON 时为 light。