ha-integration

Home Assistant 集成开发

创建专业级的自定义 Home Assistant 集成，包含完整的配置流程和实体实现。

⚠️ 开始前须知

此技能可避免 8 个常见的集成错误，并节省约 40% 的实现时间。

指标	不使用此技能	使用此技能
设置时间	45 分钟	12 分钟
常见错误	8 个	0 个
配置流程问题	5 个以上	0 个
实体注册错误	4 个以上	0 个

此技能可避免的已知问题

1. manifest.json 缺少依赖项 - 忘记声明所需的 Home Assistant 组件
2. async/await 问题 - 未正确等待协调器更新和实体初始化
3. 实体状态类别不匹配 - 为传感器平台使用了错误的 STATE_CLASS（measurement 而非 total）
4. 配置流程架构错误 - 无效的 vol.Schema 定义导致验证失败
5. 设备信息未关联 - 创建的实体未正确连接到设备注册表
6. 协调器错误 - 未妥善处理数据更新失败
7. 平台导入时机错误 - 在组件初始化前加载平台文件
8. 缺少唯一 ID 生成 - 重启后创建重复实体

快速开始

第 1 步：创建 manifest.json

{
  "domain": "my_integration",
  "name": "My Integration",
  "codeowners": ["@username"],
  "config_flow": true,
  "documentation": "https://github.com/username/ha-my-integration",
  "requirements": [],
  "version": "0.0.1"
}

为何重要： manifest.json 定义集成元数据、声明依赖项，并在 Home Assistant 中启用配置流程 UI。

第 2 步：创建包含异步设置的 init.py

import asyncio
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady

from .coordinator import MyDataUpdateCoordinator

DOMAIN = "my_integration"

async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
    """Set up the integration from config entry."""
    hass.data.setdefault(DOMAIN, {})

    # Create coordinator
    coordinator = MyDataUpdateCoordinator(hass, entry)
    await coordinator.async_config_entry_first_refresh()

    hass.data[DOMAIN][entry.entry_id] = coordinator

    # Forward setup to platforms
    await hass.config_entries.async_forward_entry_setups(entry, ["sensor"])

    return True

async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
    """Unload the integration."""
    unload_ok = await hass.config_entries.async_unload_platforms(entry, ["sensor"])
    if unload_ok:
        hass.data[DOMAIN].pop(entry.entry_id)
    return unload_ok

为何重要： 正确的异步初始化可确保 Home Assistant 等待数据加载和平台设置完成后再继续。

第 3 步：创建包含验证逻辑的 config_flow.py

from typing import Any, Dict, Optional
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigEntry
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult

from .const import DOMAIN

class MyIntegrationConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle config flow for my_integration."""

    async def async_step_user(self, user_input: Optional[Dict[str, Any]] = None) -> FlowResult:
        """Handle user initiation of config flow."""
        errors = {}

        if user_input is not None:
            # Validate user input
            try:
                # Validate connection or API call
                pass
            except Exception as exc:
                errors["base"] = "invalid_auth"

            if not errors:
                # Create unique entry
                await self.async_set_unique_id(user_input.get("host"))
                self._abort_if_unique_id_configured()

                return self.async_create_entry(
                    title=user_input.get("name"),
                    data=user_input
                )

        # Show form
        return self.async_show_form(
            step_id="user",
            data_schema=vol.Schema({
                vol.Required("name"): str,
                vol.Required("host"): str,
            }),
            errors=errors
        )

    @staticmethod
    @callback
    def async_get_options_flow(config_entry: ConfigEntry):
        """Return options flow for this integration."""
        return MyIntegrationOptionsFlow(config_entry)

为何重要： 配置流程提供方便用户使用的设置 UI，并在创建配置条目之前验证输入。

关键规则

✅ 始终执行

• ✅ 全程使用 async/await（async_setup_entry、async_added_to_hass、async_update_data）
• ✅ 为每个实体生成 unique_id（防止重启后出现重复实体）
• ✅ 通过 device_info 属性将实体关联到设备
• ✅ 妥善处理协调器更新失败（记录日志、将实体标记为不可用）
• ✅ 在 manifest.json requirements 中声明所有外部依赖项
• ✅ 使用类型提示，以获得更好的 IDE 支持并符合 Home Assistant 要求
• ✅ 通过协调器模式（DataUpdateCoordinator）注册实体

❌ 绝不执行

• ❌ 使用同步网络调用（requests 库）——改用 aiohttp
• ❌ 在组件级别导入平台文件——让 Home Assistant 转发设置流程
• ❌ 创建没有 unique_id 的实体——会导致重启后出现重复实体
• ❌ 忽略协调器更新失败——应将实体标记为不可用
• ❌ 硬编码 API 端点——使用配置流程存储这些端点
• ❌ 实现多设备集成时忘记设置 device_info
• ❌ 错误使用 STATE_CLASS（measurement、total 或 total_increasing）

常见错误

❌ 错误：

# Synchronous network call - blocks event loop
import requests
data = requests.get("https://api.example.com/data").json()

# No unique_id - duplicate entities on restart
class MySensor(SensorEntity):
    pass

# Missing await
coordinator.async_refresh()

✅ 正确：

# Async network call - doesn't block
async with aiohttp.ClientSession() as session:
    async with session.get("https://api.example.com/data") as resp:
        data = await resp.json()

# Proper unique_id generation
class MySensor(SensorEntity):
    @property
    def unique_id(self) -> str:
        return f"{self.coordinator.data['id']}_sensor"

# Proper await
await coordinator.async_request_refresh()

原因： 同步调用会阻塞 Home Assistant 的事件循环，导致 UI 冻结。缺少 unique_id 会造成实体重复。缺少 await 意味着代码会在异步操作完成前继续执行。

已知问题预防

问题	根本原因	解决方案
重启后出现重复实体	未设置 unique_id	实现 unique_id 属性，并使用稳定标识符
配置流程验证静默失败	async_step_user 中缺少错误处理	使用 try/except 包装验证，并设置 errors 字典
实体状态未更新	Coordinator 未刷新，或实体未订阅	在更新监听器中使用 @callback 装饰器
设备未显示	缺少 device_info，或 device_identifier 不匹配	设置 device_info，并确保标识符与注册表中的一致
设置期间 UI 冻结	async_setup_entry 中有同步网络调用	所有异步网络操作均使用 aiohttp
平台导入失败	在 init.py 中导入平台文件	让 Home Assistant 通过 async_forward_entry_setups 处理

Manifest 配置参考

manifest.json

{
  "domain": "integration_name",
  "name": "Integration Display Name",
  "codeowners": ["@github_username"],
  "config_flow": true,
  "documentation": "https://github.com/username/repo",
  "homeassistant": "2024.1.0",
  "requirements": ["requests>=2.25.0"],
  "version": "1.0.0",
  "issue_tracker": "https://github.com/username/repo/issues"
}

关键设置：

• domain：唯一标识符（字母数字、下划线、小写）
• config_flow：设为 true 以启用配置界面
• requirements：所需 PyPI 软件包列表（例如，["requests>=2.25.0"]）
• homeassistant：所需的最低 Home Assistant 版本

配置流程模式

使用 vol.All 进行验证的 Schema

vol.Schema({
    vol.Required("host"): vol.All(str, vol.Length(min=5)),
    vol.Required("port", default=8080): int,
    vol.Optional("api_key"): str,
})

凭据过期时的重新认证流程

async def async_step_reauth(self, user_input: Dict[str, Any] | None = None) -> FlowResult:
    """Handle reauth upon an API authentication error."""
    config_entry = self.hass.config_entries.async_get_entry(
        self.context["entry_id"]
    )

    if user_input is not None:
        config_entry.data = {**config_entry.data, **user_input}
        self.hass.config_entries.async_update_entry(config_entry)
        return self.async_abort(reason="reauth_successful")

    return self.async_show_form(
        step_id="reauth",
        data_schema=vol.Schema({vol.Required("api_key"): str})
    )

实体实现模式

带状态类别的传感器

from homeassistant.components.sensor import SensorEntity, SensorStateClass
from homeassistant.const import UnitOfTemperature

class TemperatureSensor(SensorEntity):
    """Temperature sensor entity."""

    _attr_device_class = "temperature"
    _attr_state_class = SensorStateClass.MEASUREMENT
    _attr_native_unit_of_measurement = UnitOfTemperature.CELSIUS

    def __init__(self, coordinator, idx):
        """Initialize sensor."""
        self.coordinator = coordinator
        self._idx = idx

    @property
    def unique_id(self) -> str:
        """Return unique ID."""
        return f"{self.coordinator.data['id']}_temp_{self._idx}"

    @property
    def device_info(self) -> DeviceInfo:
        """Return device information."""
        return DeviceInfo(
            identifiers={(DOMAIN, self.coordinator.data['id'])},
            name=self.coordinator.data['name'],
            manufacturer="My Company",
        )

    @property
    def native_value(self) -> float | None:
        """Return sensor value."""
        try:
            return float(self.coordinator.data['temperature'])
        except (KeyError, TypeError):
            return None

    async def async_added_to_hass(self) -> None:
        """Connect to coordinator when added."""
        await super().async_added_to_hass()
        self.async_on_remove(
            self.coordinator.async_add_listener(self._handle_coordinator_update)
        )

    @callback
    def _handle_coordinator_update(self) -> None:
        """Update when coordinator updates."""
        self.async_write_ha_state()

二进制传感器

from homeassistant.components.binary_sensor import BinarySensorEntity, BinarySensorDeviceClass

class MotionSensor(BinarySensorEntity):
    """Motion detection sensor."""

    _attr_device_class = BinarySensorDeviceClass.MOTION

    @property
    def is_on(self) -> bool | None:
        """Return True if motion detected."""
        return self.coordinator.data.get('motion', False)

DataUpdateCoordinator 模式

from datetime import timedelta
from homeassistant.helpers.update_coordinator import (
    DataUpdateCoordinator,
    UpdateFailed,
)
import logging

_LOGGER = logging.getLogger(__name__)

class MyDataUpdateCoordinator(DataUpdateCoordinator):
    """Coordinator for fetching data."""

    def __init__(self, hass, entry):
        """Initialize coordinator."""
        super().__init__(
            hass,
            _LOGGER,
            name="My Integration",
            update_interval=timedelta(minutes=5),
        )
        self.entry = entry

    async def _async_update_data(self):
        """Fetch data from API."""
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(
                    f"https://api.example.com/data",
                    headers={"Authorization": f"Bearer {self.entry.data['api_key']}"}
                ) as resp:
                    if resp.status == 401:
                        raise ConfigEntryAuthFailed("Invalid API key")
                    return await resp.json()
        except asyncio.TimeoutError as err:
            raise UpdateFailed("API timeout") from err
        except Exception as err:
            raise UpdateFailed(f"API error: {err}") from err

设备注册表模式

使用标识符创建设备

from homeassistant.helpers.device_registry import DeviceInfo

device_info = DeviceInfo(
    identifiers={(DOMAIN, "device_unique_id")},
    name="Device Name",
    manufacturer="Manufacturer",
    model="Model Name",
    sw_version="1.0.0",
    via_device=(DOMAIN, "parent_device_id"),  # For child devices
)

序列号和连接

device_info = DeviceInfo(
    identifiers={(DOMAIN, device_id)},
    serial_number="SERIAL123",
    connections={(dr.CONNECTION_NETWORK_MAC, "aa:bb:cc:dd:ee:ff")},
)

常见模式

从配置条目加载配置

class MyIntegration:
    def __init__(self, hass: HomeAssistant, entry: ConfigEntry):
        self.hass = hass
        self.entry = entry
        self.api_key = entry.data.get("api_key")
        self.host = entry.data.get("host")

处理选项流程

async def async_step_init(self, user_input: Optional[Dict[str, Any]] = None) -> FlowResult:
    """Manage integration options."""
    if user_input is not None:
        return self.async_create_entry(
            title="",
            data=user_input
        )

    current_options = self.config_entry.options
    return self.async_show_form(
        step_id="init",
        data_schema=vol.Schema({
            vol.Optional("refresh_rate", default=current_options.get("refresh_rate", 5)): int,
        })
    )

随附资源

参考资料

位于 references/：

• manifest-reference.md - 完整的 manifest.json 字段参考
• entity-base-classes.md - 实体实现基类和属性
• config-flow-patterns.md - 高级配置流程模式和验证

模板

位于 assets/：

• manifest.json - manifest.json 入门模板
• config_flow.py - 基本配置流程样板代码
• __init__.py - 集成初始化模板
• coordinator.py - DataUpdateCoordinator 模板

注意： 如需深入了解特定主题，请参阅上述参考文件。

依赖项

必需

软件包	版本	用途
homeassistant	>=2024.1.0	Home Assistant 核心
voluptuous	>=0.13.0	配置验证 Schema

可选

软件包	版本	用途
aiohttp	>=3.8.0	异步 HTTP 请求（用于 API 集成）
pyyaml	>=5.4	YAML 解析（用于配置文件集成）

官方文档

• 创建组件 - Home Assistant 开发者
• 配置条目 - Home Assistant 开发者
• 实体索引 - Home Assistant 开发者
• 设备注册表 - Home Assistant 开发者

故障排除

重启后实体重复出现

症状： Home Assistant 重启后，同一传感器／开关／灯出现 2 个或更多

解决方案：

# Add unique_id property to entity class
@property
def unique_id(self) -> str:
    return f"{self.coordinator.data['id']}_{self.platform}_{self._attr_name}"

配置流程验证一直无法完成

症状： 提交时表单卡住，且不显示错误

解决方案：

# Ensure all async operations are awaited and errors caught
async def async_step_user(self, user_input=None):
    errors = {}
    if user_input is not None:
        try:
            await self._validate_input(user_input)  # ← Add await
        except Exception as e:
            errors["base"] = "validation_error"  # ← Set error

        if not errors:
            return self.async_create_entry(...)

更新后实体显示为不可用

症状： Coordinator 更新后，所有实体都变为不可用

解决方案：

# Handle coordinator errors gracefully
async def _async_update_data(self):
    try:
        return await self.api.fetch_data()
    except Exception as err:
        raise UpdateFailed(f"Error: {err}") from err  # ← Raises UpdateFailed, not Exception

设备未显示在设备注册表中

症状： 已创建设备，但在 Home Assistant 设备中不可见

解决方案：

# Ensure device_info is returned by ALL entities for the device
@property
def device_info(self) -> DeviceInfo:
    return DeviceInfo(
        identifiers={(DOMAIN, self.coordinator.data['id'])},  # ← Must be consistent
        name=self.coordinator.data['name'],
        manufacturer="Manufacturer",
    )

设置检查清单

实现新的集成之前，请确认：

• 域名唯一，并遵循小写字母加下划线的命名规范
• 已创建包含域名、名称和代码所有者的 manifest.json
• 已实现配置流程或手动配置方法
• 所有异步函数均已正确等待
• 已为所有实体生成唯一 ID（防止重复）
• 如果是多设备集成，已关联设备信息
• 已采用 DataUpdateCoordinator 或同等轮询模式
• 使用 UpdateFailed 异常处理错误
• 所有函数签名均包含类型提示
• 已为配置流程验证编写测试
• manifest 中的文档 URL 指向有效位置