Home Assistant Integration knowledge by home-assistant/core
npx skills add https://github.com/home-assistant/core --skill 'Home Assistant Integration knowledge'./homeassistant/components/<integration_domain>/./tests/components/<integration_domain>/homeassistant/components/my_integration/
├── __init__.py # 包含 async_setup_entry 的入口点
├── manifest.json # 集成元数据和依赖项
├── const.py # 域和常量
├── config_flow.py # UI 配置流程
├── coordinator.py # 数据更新协调器(如果需要)
├── entity.py # 基础实体类(如果有共享模式)
├── sensor.py # 传感器平台
├── strings.json # 面向用户的文本和翻译
├── services.yaml # 服务定义(如果适用)
└── quality_scale.yaml # 质量等级规则状态
集成可以根据需要包含平台(例如 sensor.py、switch.py 等)。以下平台有额外的指南:
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
platform-diagnostics.md 用于诊断数据收集platform-repairs.md 用于用户可操作的修复问题manifest.jsonasync_setup_entry 和 async_unload_entry 的 __init__.pyconfig_flow.pyDOMAIN 常量的 const.pystrings.jsonsensor.py 等)quality_scale.yamlHome Assistant 使用集成质量等级来确保代码质量和一致性。质量等级决定了哪些规则适用:
manifest.json:查找 "quality_scale" 键以确定集成等级quality_scale.yaml 以了解:
done: 规则已实现exempt: 规则不适用(在注释中说明原因)todo: 规则需要实现quality_scale.yaml 结构rules:
# 青铜(强制性)
config-flow: done
entity-unique-id: done
action-setup:
status: exempt
comment: Integration does not register custom actions.
# 白银(如果目标是白银+)
entity-unavailable: done
parallel-updates: done
# 黄金(如果目标是黄金+)
devices: done
diagnostics: done
# 铂金(如果目标是铂金)
strict-typing: done
在审查/创建代码时:在应用规则之前,始终检查集成的质量等级和豁免状态。
homeassistant/const.py(使用这些而不是硬编码)homeassistant/components/{domain}/const.py - 常量homeassistant/components/{domain}/models.py - 数据模型homeassistant/components/{domain}/coordinator.py - 更新协调器homeassistant/components/{domain}/config_flow.py - 配置流程homeassistant/components/{domain}/{platform}.py - 平台实现coordinator.py : 集中数据获取逻辑
class MyCoordinator(DataUpdateCoordinator[MyData]):
def __init__(self, hass: HomeAssistant, client: MyClient, config_entry: ConfigEntry) -> None:
super().__init__(
hass,
logger=LOGGER,
name=DOMAIN,
update_interval=timedelta(minutes=1),
config_entry=config_entry, # ✅ 传递 config_entry - 这是被接受且推荐的
)
entity.py : 基础实体定义以减少重复
class MyEntity(CoordinatorEntity[MyCoordinator]):
_attr_has_entity_name = True
使用 ConfigEntry.runtime_data : 存储非持久性运行时数据
type MyIntegrationConfigEntry = ConfigEntry[MyClient]
async def async_setup_entry(hass: HomeAssistant, entry: MyIntegrationConfigEntry) -> bool: client = MyClient(entry.data[CONF_HOST]) entry.runtime_data = client
domain、name、codeowners、integration_type、documentation、requirementsdevice、hub、service、system、helpercloud_polling、local_polling、local_push)zeroconf、dhcp、bluetooth、ssdp、usbapplication_credentials、bluetooth_adapters)版本控制 : 始终设置 VERSION = 1 和 MINOR_VERSION = 1
唯一 ID 管理 :
await self.async_set_unique_id(device_unique_id)
self._abort_if_unique_id_configured()
错误处理 : 在 strings.json 的 config.error 下定义错误
步骤方法 : 使用标准命名(async_step_user、async_step_discovery 等)
manifest.json : 将 GitHub 用户名添加到 codeowners:
{
"domain": "my_integration", "name": "My Integration", "codeowners": ["@me"] }
传递 WebSession : 支持将 Web 会话传递给依赖项
async def async_setup_entry(hass: HomeAssistant, entry: MyConfigEntry) -> bool:
"""从配置条目设置集成。"""
client = MyClient(entry.data[CONF_HOST], async_get_clientsession(hass))
对于 cookies:使用 async_create_clientsession (aiohttp) 或 create_async_httpx_client (httpx)
标准模式 : 用于高效的数据管理
class MyCoordinator(DataUpdateCoordinator):
def __init__(self, hass: HomeAssistant, client: MyClient, config_entry: ConfigEntry) -> None:
super().__init__(
hass,
logger=LOGGER,
name=DOMAIN,
update_interval=timedelta(minutes=5),
config_entry=config_entry, # ✅ 传递 config_entry - 这是被接受且推荐的
)
self.client = client
async def _async_update_data(self):
try:
return await self.client.fetch_data()
except ApiError as err:
raise UpdateFailed(f"API 通信错误: {err}")
错误类型 : 对 API 错误使用 UpdateFailed,对身份验证问题使用 ConfigEntryAuthFailed
配置条目 : 始终将 config_entry 参数传递给协调器 - 这是被接受且推荐的
需要 UI 设置 : 所有集成必须支持通过 UI 进行配置
清单 : 在 manifest.json 中设置 "config_flow": true
数据存储 :
ConfigEntry.data 中ConfigEntry.options 中验证 : 在创建条目之前始终验证用户输入
配置条目命名 :
连接测试 : 在配置流程期间测试设备/服务连接:
try:
await client.get_data()
except MyException: errors["base"] = "cannot_connect"
重复预防 : 防止重复配置:
# 使用唯一 ID
await self.async_set_unique_id(identifier) self._abort_if_unique_id_configured()
self._async_abort_entries_match({CONF_HOST: user_input[CONF_HOST]})
必需方法 : 在配置流程中实现 async_step_reauth
凭据更新 : 允许用户更新凭据而无需重新添加
验证 : 验证账户是否与现有唯一 ID 匹配:
await self.async_set_unique_id(user_id)
self._abort_if_unique_id_mismatch(reason="wrong_account") return self.async_update_reload_and_abort( self._get_reauth_entry(), data_updates={CONF_API_TOKEN: user_input[CONF_API_TOKEN]} )
async_step_reconfigure 方法_abort_if_unique_id_mismatch 防止更改底层账户清单配置 : 添加发现方法(zeroconf、dhcp 等)
{
"zeroconf": ["_mydevice._tcp.local."] }
发现处理程序 : 实现适当的 async_step_* 方法:
async def async_step_zeroconf(self, discovery_info):
"""处理 zeroconf 发现。"""
await self.async_set_unique_id(discovery_info.properties["serialno"])
self._abort_if_unique_id_configured(updates={CONF_HOST: discovery_info.host})
网络更新 : 使用发现来更新动态 IP 地址
Zeroconf/mDNS : 使用异步实例
aiozc = await zeroconf.async_get_async_instance(hass)
SSDP 发现 : 注册带有清理的回调
entry.async_on_unload(
ssdp.async_register_callback(
hass, _async_discovered_device,
{"st": "urn:schemas-upnp-org:device:ZonePlayer:1"}
)
)
清单依赖项 : 将 bluetooth_adapters 添加到依赖项中
可连接 : 对于需要连接的设备,设置 "connectable": true
扫描器使用 : 始终使用共享扫描器实例
scanner = bluetooth.async_get_scanner()
entry.async_on_unload( bluetooth.async_register_callback( hass, _async_discovered_device, {"service_uuid": "example_uuid"}, bluetooth.BluetoothScanningMode.ACTIVE ) )
连接处理 : 切勿重用 BleakClient 实例,使用 10 秒以上的超时时间
async_setup_entry 中验证集成是否可以设置ConfigEntryNotReady: 设备离线或临时故障ConfigEntryAuthFailed: 身份验证问题ConfigEntryError: 无法解决的设置问题必需 : 实现 async_unload_entry 以进行运行时移除/重新加载
平台卸载 : 使用 hass.config_entries.async_unload_platforms
清理 : 使用 entry.async_on_unload 注册回调:
async def async_unload_entry(hass: HomeAssistant, entry: MyConfigEntry) -> bool:
"""卸载配置条目。"""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
entry.runtime_data.listener() # 清理资源
return unload_ok
注册 : 在 async_setup 中注册所有服务操作,而不是在 async_setup_entry 中
验证 : 检查配置条目是否存在以及是否已加载:
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
async def service_action(call: ServiceCall) -> ServiceResponse:
if not (entry := hass.config_entries.async_get_entry(call.data[ATTR_CONFIG_ENTRY_ID])):
raise ServiceValidationError("未找到条目")
if entry.state is not ConfigEntryState.LOADED:
raise ServiceValidationError("条目未加载")
异常处理 : 引发适当的异常:
# 对于无效输入
if end_date < start_date: raise ServiceValidationError("结束日期必须在开始日期之后")
try: await client.set_schedule(start_date, end_date) except MyConnectionError as err: raise HomeAssistantError("无法连接到计划") from err
实体服务 : 在平台设置时注册
platform.async_register_entity_service(
"my_entity_service",
{vol.Required("parameter"): cv.string},
"handle_service_method"
)
服务模式 : 始终验证输入
SERVICE_SCHEMA = vol.Schema({
vol.Required("entity_id"): cv.entity_ids,
vol.Required("parameter"): cv.string,
vol.Optional("timeout", default=30): cv.positive_int,
})
服务文件 : 创建包含描述和字段定义的 services.yaml
尽可能使用更新协调器模式
轮询间隔不可由用户配置:切勿在配置流程或配置条目中添加 scan_interval、update_interval 或轮询频率选项
集成决定间隔:根据集成逻辑以编程方式设置 update_interval,而不是根据用户输入
最小间隔:
并行更新:指定并发更新数量:
PARALLEL_UPDATES = 1 # 序列化更新以防止设备过载
PARALLEL_UPDATES = 0 # 无限制(适用于基于协调器的或只读的)
必需 : 每个实体必须有一个唯一 ID 用于注册表跟踪
必须每个平台唯一(而不是每个集成)
不要在 ID 中包含集成域或平台
实现:
class MySensor(SensorEntity):
def __init__(self, device_id: str) -> None:
self._attr_unique_id = f"{device_id}_temperature"
可接受的 ID 来源:
format_mac 格式化)f"{entry.entry_id}-battery"切勿使用:
Lambda/匿名函数 : 常用于 EntityDescription 中进行值转换
多行 Lambda : 当 lambda 超过行长度时,用括号包裹以提高可读性
不良模式:
SensorEntityDescription(
key="temperature",
name="Temperature",
value_fn=lambda data: round(data["temp_value"] * 1.8 + 32, 1) if data.get("temp_value") is not None else None, # ❌ 太长
)
良好模式:
SensorEntityDescription(
key="temperature",
name="Temperature",
value_fn=lambda data: ( # ✅ 括号与 lambda 在同一行
round(data["temp_value"] * 1.8 + 32, 1)
if data.get("temp_value") is not None
else None
),
)
使用 has_entity_name : 设置 _attr_has_entity_name = True
对于特定字段:
class MySensor(SensorEntity):
_attr_has_entity_name = True
def __init__(self, device: Device, field: str) -> None:
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, device.id)},
name=device.name,
)
self._attr_name = field # 例如:"temperature", "humidity"
对于设备本身 : 设置 _attr_name = None
在async_added_to_hass中订阅:
async def async_added_to_hass(self) -> None:
"""订阅事件。"""
self.async_on_remove(
self.client.events.subscribe("my_event", self._handle_event)
)
如果未使用async_on_remove,则在async_will_remove_from_hass中取消订阅
切勿在 __init__ 或其他方法中订阅
None(而不是 "unknown" 或 "unavailable")available() 属性而不是使用 "unavailable" 状态标记为不可用 : 当无法从设备/服务获取数据时
协调器模式:
@property
def available(self) -> bool: """返回实体是否可用。""" return super().available and self.identifier in self.coordinator.data
直接更新模式:
async def async_update(self) -> None:
"""更新实体。"""
try:
data = await self.client.get_data()
except MyException:
self._attr_available = False
else:
self._attr_available = True
self._attr_native_value = data.value
None创建设备 : 将相关实体分组到设备下
设备信息 : 提供全面的元数据:
_attr_device_info = DeviceInfo(
connections={(CONNECTION_NETWORK_MAC, device.mac)},
identifiers={(DOMAIN, device.id)},
name=device.name,
manufacturer="My Company",
model="My Sensor",
sw_version=device.version,
)
对于服务:添加 entry_type=DeviceEntryType.SERVICE
自动检测新设备 : 初始设置之后
实现模式:
def _check_device() -> None:
current_devices = set(coordinator.data)
new_devices = current_devices - known_devices
if new_devices:
known_devices.update(new_devices)
async_add_entities([MySensor(coordinator, device_id) for device_id in new_devices])
entry.async_on_unload(coordinator.async_add_listener(_check_device))
自动移除 : 当设备从集线器/账户中消失时
设备注册表更新:
device_registry.async_update_device(
device_id=device.id,
remove_config_entry_id=self.config_entry.entry_id,
)
手动删除 : 需要时实现 async_remove_config_entry_device
必需 : 为实体分配适当的类别
实现 : 设置 _attr_entity_category
class MySensor(SensorEntity):
_attr_entity_category = EntityCategory.DIAGNOSTIC
类别包括:用于系统/技术信息的 DIAGNOSTIC
可用时使用 : 为实体类型设置适当的设备类别
class MyTemperatureSensor(SensorEntity):
_attr_device_class = SensorDeviceClass.TEMPERATURE
为以下提供上下文:单位转换、语音控制、UI 表示
禁用嘈杂/不太受欢迎的实体 : 减少资源使用
class MySignalStrengthSensor(SensorEntity):
_attr_entity_registry_enabled_default = False
目标:频繁变化的状态、技术诊断
使用 has_entity_name 时必需 : 支持国际用户
实现:
class MySensor(SensorEntity):
_attr_has_entity_name = True
_attr_translation_key = "phase_voltage"
创建包含翻译的 strings.json:
{
"entity": { "sensor": { "phase_voltage": { "name": "Phase voltage" } } } }
可翻译错误 : 对面向用户的异常使用翻译键
实现:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="end_date_before_start_date",
)
添加到 strings.json:
{
"exceptions": { "end_date_before_start_date": { "message": "The end date cannot be before the start date." } } }
动态图标 : 支持基于状态和范围的图标选择
基于状态的图标:
{
"entity": { "sensor": { "tree_pollen": { "default": "mdi:tree", "state": { "high": "mdi:tree-outline" } } } } }
基于范围的图标(对于数值):
{
"entity": { "sensor": { "battery_level": { "default": "mdi:battery-unknown", "range": { "0": "mdi:battery-outline", "90": "mdi:battery-90", "100": "mdi:battery" } } } } }
tests/components/{domain}/tests.common 中的 pytest fixturesconfig_flow.py 中定义的方法;端到端地执行流程逻辑。集成特定测试(推荐):
pytest ./tests/components/<integration_domain> \
--cov=homeassistant.components.<integration_domain>
--cov-report term-missing
--durations-min=1
--durations=0
--numprocesses=auto
hass.data - 改用 fixtures 和适当的集成设置async def test_user_flow_success(hass, mock_api):
"""测试成功的用户流程。"""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
# 测试表单提交
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=TEST_USER_INPUT
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "My Device"
assert result["data"] == TEST_USER_INPUT
async def test_flow_connection_error(hass, mock_api_error):
"""测试连接错误处理。"""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=TEST_USER_INPUT
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
@pytest.fixture
def platforms() -> list[Platform]:
"""重写的 fixture 以指定要测试的平台。"""
return [Platform.SENSOR] # 或者根据需要指定另一个特定平台。
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "init_integration")
async def test_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
mock_config_entry: MockConfigEntry,
) -> None:
"""测试传感器实体。"""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
# 确保实体正确分配给设备
device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, "device_unique_id")}
)
assert device_entry
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
for entity_entry in entity_entries:
assert entity_entry.device_id == device_entry.id
# 现代集成 fixture 设置
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""返回默认的模拟配置条目。"""
return MockConfigEntry(
title="My Integration",
domain=DOMAIN,
data={CONF_HOST: "127.0.0.1", CONF_API_KEY: "test_key"},
unique_id="device_unique_id",
)
@pytest.fixture
def mock_device_api() -> Generator[MagicMock]:
"""返回一个模拟的设备 API。"""
with patch("homeassistant.components.my_integration.MyDeviceAPI", autospec=True) as api_mock:
api = api_mock.return_value
api.get_data.return_value = MyDeviceData.from_json(
load_fixture("device_data.json", DOMAIN)
)
yield api
@pytest.fixture
def platforms() -> list[Platform]:
"""Fixture 以指定要测试的平台。"""
return PLATFORMS
@pytest.fixture
async def init_integration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_device_api: MagicMock,
platforms: list[Platform],
) -> MockConfigEntry:
"""为测试设置集成。"""
mock_config_entry.add_to_hass(hass)
with patch("homeassistant.components.my_integration.PLATFORMS", platforms):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
return mock_config_entry
manifest.json 语法和必填字段unique_id 和 has_entity_name 的实现strings.json 条目和错误处理# 在测试中启用调试日志
caplog.set_level(logging.DEBUG, logger="my_integration")
# 在集成代码中 - 使用适当的日志记录
_LOGGER = logging.getLogger(__name__)
_LOGGER.debug("Processing data: %s", data) # 使用惰性日志记录
# 检查特定集成
python -m script.hassfest --integration-path homeassistant/components/my_integration
# 验证质量等级
# 根据当前规则检查 quality_scale.yaml
# 运行集成测试并覆盖
pytest ./tests/components/my_integration \
--cov=homeassistant.components.my_integration \
--cov-report term-missing
每周安装次数
–
仓库
GitHub 星标数
85.8K
首次出现时间
–
安全审计
./homeassistant/components/<integration_domain>/./tests/components/<integration_domain>/homeassistant/components/my_integration/
├── __init__.py # Entry point with async_setup_entry
├── manifest.json # Integration metadata and dependencies
├── const.py # Domain and constants
├── config_flow.py # UI configuration flow
├── coordinator.py # Data update coordinator (if needed)
├── entity.py # Base entity class (if shared patterns)
├── sensor.py # Sensor platform
├── strings.json # User-facing text and translations
├── services.yaml # Service definitions (if applicable)
└── quality_scale.yaml # Quality scale rule status
An integration can have platforms as needed (e.g., sensor.py, switch.py, etc.). The following platforms have extra guidelines:
platform-diagnostics.md for diagnostic data collectionplatform-repairs.md for user-actionable repair issuesmanifest.json with required fields (domain, name, codeowners, etc.)__init__.py with async_setup_entry and async_unload_entryconfig_flow.py with UI configuration supportconst.py with DOMAIN constantstrings.json with at least config flow textsensor.py, etc.) as neededquality_scale.yaml with rule status trackingHome Assistant uses an Integration Quality Scale to ensure code quality and consistency. The quality level determines which rules apply:
manifest.json: Look for "quality_scale" key to determine integration levelquality_scale.yaml in integration folder for:
done: Rule implementedexempt: Rule doesn't apply (with reason in comment)todo: Rule needs implementationquality_scale.yaml Structurerules:
# Bronze (mandatory)
config-flow: done
entity-unique-id: done
action-setup:
status: exempt
comment: Integration does not register custom actions.
# Silver (if targeting Silver+)
entity-unavailable: done
parallel-updates: done
# Gold (if targeting Gold+)
devices: done
diagnostics: done
# Platinum (if targeting Platinum)
strict-typing: done
When Reviewing/Creating Code : Always check the integration's quality scale level and exemption status before applying rules.
homeassistant/const.py (use these instead of hardcoding)homeassistant/components/{domain}/const.py - Constantshomeassistant/components/{domain}/models.py - Data modelshomeassistant/components/{domain}/coordinator.py - Update coordinatorhomeassistant/components/{domain}/config_flow.py - Configuration flowhomeassistant/components/{domain}/{platform}.py - Platform implementationscoordinator.py : Centralize data fetching logic
class MyCoordinator(DataUpdateCoordinator[MyData]):
def __init__(self, hass: HomeAssistant, client: MyClient, config_entry: ConfigEntry) -> None:
super().__init__(
hass,
logger=LOGGER,
name=DOMAIN,
update_interval=timedelta(minutes=1),
config_entry=config_entry, # ✅ Pass config_entry - it's accepted and recommended
)
entity.py : Base entity definitions to reduce duplication
class MyEntity(CoordinatorEntity[MyCoordinator]):
_attr_has_entity_name = True
Use ConfigEntry.runtime_data : Store non-persistent runtime data
type MyIntegrationConfigEntry = ConfigEntry[MyClient]
async def async_setup_entry(hass: HomeAssistant, entry: MyIntegrationConfigEntry) -> bool: client = MyClient(entry.data[CONF_HOST]) entry.runtime_data = client
domain, name, codeowners, integration_type, documentation, requirementsdevice, hub, service, system, helperVersion Control : Always set VERSION = 1 and MINOR_VERSION = 1
Unique ID Management :
await self.async_set_unique_id(device_unique_id)
self._abort_if_unique_id_configured()
Error Handling : Define errors in strings.json under config.error
Step Methods : Use standard naming (async_step_user, async_step_discovery, etc.)
manifest.json : Add GitHub usernames to codeowners:
{
"domain": "my_integration", "name": "My Integration", "codeowners": ["@me"] }
Pass WebSession : Support passing web sessions to dependencies
async def async_setup_entry(hass: HomeAssistant, entry: MyConfigEntry) -> bool:
"""Set up integration from config entry."""
client = MyClient(entry.data[CONF_HOST], async_get_clientsession(hass))
For cookies: Use async_create_clientsession (aiohttp) or create_async_httpx_client (httpx)
Standard Pattern : Use for efficient data management
class MyCoordinator(DataUpdateCoordinator):
def __init__(self, hass: HomeAssistant, client: MyClient, config_entry: ConfigEntry) -> None:
super().__init__(
hass,
logger=LOGGER,
name=DOMAIN,
update_interval=timedelta(minutes=5),
config_entry=config_entry, # ✅ Pass config_entry - it's accepted and recommended
)
self.client = client
async def _async_update_data(self):
try:
return await self.client.fetch_data()
except ApiError as err:
raise UpdateFailed(f"API communication error: {err}")
Error Types : Use UpdateFailed for API errors, ConfigEntryAuthFailed for auth issues
Config Entry : Always pass config_entry parameter to coordinator - it's accepted and recommended
UI Setup Required : All integrations must support configuration via UI
Manifest : Set "config_flow": true in manifest.json
Data Storage :
ConfigEntry.dataConfigEntry.optionsValidation : Always validate user input before creating entries
Config Entry Naming :
Connection Testing : Test device/service connection during config flow:
try:
await client.get_data()
Required Method : Implement async_step_reauth in config flow
Credential Updates : Allow users to update credentials without re-adding
Validation : Verify account matches existing unique ID:
await self.async_set_unique_id(user_id)
self._abort_if_unique_id_mismatch(reason="wrong_account") return self.async_update_reload_and_abort( self._get_reauth_entry(), data_updates={CONF_API_TOKEN: user_input[CONF_API_TOKEN]} )
async_step_reconfigure method_abort_if_unique_id_mismatchManifest Configuration : Add discovery method (zeroconf, dhcp, etc.)
{
"zeroconf": ["_mydevice._tcp.local."] }
Discovery Handler : Implement appropriate async_step_* method:
async def async_step_zeroconf(self, discovery_info):
"""Handle zeroconf discovery."""
await self.async_set_unique_id(discovery_info.properties["serialno"])
self._abort_if_unique_id_configured(updates={CONF_HOST: discovery_info.host})
Network Updates : Use discovery to update dynamic IP addresses
Zeroconf/mDNS : Use async instances
aiozc = await zeroconf.async_get_async_instance(hass)
SSDP Discovery : Register callbacks with cleanup
entry.async_on_unload(
ssdp.async_register_callback(
hass, _async_discovered_device,
{"st": "urn:schemas-upnp-org:device:ZonePlayer:1"}
)
)
Manifest Dependencies : Add bluetooth_adapters to dependencies
Connectable : Set "connectable": true for connection-required devices
Scanner Usage : Always use shared scanner instance
scanner = bluetooth.async_get_scanner()
entry.async_on_unload( bluetooth.async_register_callback( hass, _async_discovered_device, {"service_uuid": "example_uuid"}, bluetooth.BluetoothScanningMode.ACTIVE ) )
Connection Handling : Never reuse BleakClient instances, use 10+ second timeouts
async_setup_entryConfigEntryNotReady: Device offline or temporary failureConfigEntryAuthFailed: Authentication issuesConfigEntryError: Unresolvable setup problemsRequired : Implement async_unload_entry for runtime removal/reload
Platform Unloading : Use hass.config_entries.async_unload_platforms
Cleanup : Register callbacks with entry.async_on_unload:
async def async_unload_entry(hass: HomeAssistant, entry: MyConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
entry.runtime_data.listener() # Clean up resources
return unload_ok
Registration : Register all service actions in async_setup, NOT in async_setup_entry
Validation : Check config entry existence and loaded state:
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
async def service_action(call: ServiceCall) -> ServiceResponse:
if not (entry := hass.config_entries.async_get_entry(call.data[ATTR_CONFIG_ENTRY_ID])):
raise ServiceValidationError("Entry not found")
if entry.state is not ConfigEntryState.LOADED:
raise ServiceValidationError("Entry not loaded")
Exception Handling : Raise appropriate exceptions:
# For invalid input
if end_date < start_date: raise ServiceValidationError("End date must be after start date")
try: await client.set_schedule(start_date, end_date) except MyConnectionError as err: raise HomeAssistantError("Could not connect to the schedule") from err
Entity Services : Register on platform setup
platform.async_register_entity_service(
"my_entity_service",
{vol.Required("parameter"): cv.string},
"handle_service_method"
)
Service Schema : Always validate input
SERVICE_SCHEMA = vol.Schema({
vol.Required("entity_id"): cv.entity_ids,
vol.Required("parameter"): cv.string,
vol.Optional("timeout", default=30): cv.positive_int,
})
Services File : Create services.yaml with descriptions and field definitions
Use update coordinator pattern when possible
Polling intervals are NOT user-configurable : Never add scan_interval, update_interval, or polling frequency options to config flows or config entries
Integration determines intervals : Set update_interval programmatically based on integration logic, not user input
Minimum Intervals :
Parallel Updates : Specify number of concurrent updates:
PARALLEL_UPDATES = 1 # Serialize updates to prevent overwhelming device
PARALLEL_UPDATES = 0 # Unlimited (for coordinator-based or read-only)
Required : Every entity must have a unique ID for registry tracking
Must be unique per platform (not per integration)
Don't include integration domain or platform in ID
Implementation :
class MySensor(SensorEntity):
def __init__(self, device_id: str) -> None:
self._attr_unique_id = f"{device_id}_temperature"
Acceptable ID Sources :
format_mac from device registry)f"{entry.entry_id}-battery"Never Use :
Lambda/Anonymous Functions : Often used in EntityDescription for value transformation
Multiline Lambdas : When lambdas exceed line length, wrap in parentheses for readability
Bad pattern :
SensorEntityDescription(
key="temperature",
name="Temperature",
value_fn=lambda data: round(data["temp_value"] * 1.8 + 32, 1) if data.get("temp_value") is not None else None, # ❌ Too long
)
Good pattern :
SensorEntityDescription(
key="temperature",
name="Temperature",
value_fn=lambda data: ( # ✅ Parenthesis on same line as lambda
round(data["temp_value"] * 1.8 + 32, 1)
if data.get("temp_value") is not None
else None
),
)
Use has_entity_name : Set _attr_has_entity_name = True
For specific fields :
class MySensor(SensorEntity):
_attr_has_entity_name = True
def __init__(self, device: Device, field: str) -> None:
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, device.id)},
name=device.name,
)
self._attr_name = field # e.g., "temperature", "humidity"
For device itself : Set _attr_name = None
Subscribe inasync_added_to_hass:
async def async_added_to_hass(self) -> None:
"""Subscribe to events."""
self.async_on_remove(
self.client.events.subscribe("my_event", self._handle_event)
)
Unsubscribe inasync_will_remove_from_hass if not using async_on_remove
Never subscribe in __init__ or other methods
None (not "unknown" or "unavailable")available() property instead of using "unavailable" stateMark Unavailable : When data cannot be fetched from device/service
Coordinator Pattern :
@property
def available(self) -> bool: """Return if entity is available.""" return super().available and self.identifier in self.coordinator.data
Direct Update Pattern :
async def async_update(self) -> None:
"""Update entity."""
try:
data = await self.client.get_data()
except MyException:
self._attr_available = False
else:
self._attr_available = True
self._attr_native_value = data.value
NoneCreate Devices : Group related entities under devices
Device Info : Provide comprehensive metadata:
_attr_device_info = DeviceInfo(
connections={(CONNECTION_NETWORK_MAC, device.mac)},
identifiers={(DOMAIN, device.id)},
name=device.name,
manufacturer="My Company",
model="My Sensor",
sw_version=device.version,
)
For services: Add entry_type=DeviceEntryType.SERVICE
Auto-detect New Devices : After initial setup
Implementation Pattern :
def _check_device() -> None:
current_devices = set(coordinator.data)
new_devices = current_devices - known_devices
if new_devices:
known_devices.update(new_devices)
async_add_entities([MySensor(coordinator, device_id) for device_id in new_devices])
entry.async_on_unload(coordinator.async_add_listener(_check_device))
Auto-remove : When devices disappear from hub/account
Device Registry Update :
device_registry.async_update_device(
device_id=device.id,
remove_config_entry_id=self.config_entry.entry_id,
)
Manual Deletion : Implement async_remove_config_entry_device when needed
Required : Assign appropriate category to entities
Implementation : Set _attr_entity_category
class MySensor(SensorEntity):
_attr_entity_category = EntityCategory.DIAGNOSTIC
Categories include: DIAGNOSTIC for system/technical information
Use When Available : Set appropriate device class for entity type
class MyTemperatureSensor(SensorEntity):
_attr_device_class = SensorDeviceClass.TEMPERATURE
Provides context for: unit conversion, voice control, UI representation
Disable Noisy/Less Popular Entities : Reduce resource usage
class MySignalStrengthSensor(SensorEntity):
_attr_entity_registry_enabled_default = False
Target: frequently changing states, technical diagnostics
Required with has_entity_name : Support international users
Implementation :
class MySensor(SensorEntity):
_attr_has_entity_name = True
_attr_translation_key = "phase_voltage"
Create strings.json with translations:
{
"entity": { "sensor": { "phase_voltage": { "name": "Phase voltage" } } } }
Translatable Errors : Use translation keys for user-facing exceptions
Implementation :
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="end_date_before_start_date",
)
Add to strings.json:
{
"exceptions": { "end_date_before_start_date": { "message": "The end date cannot be before the start date." } } }
Dynamic Icons : Support state and range-based icon selection
State-based Icons :
{
"entity": { "sensor": { "tree_pollen": { "default": "mdi:tree", "state": { "high": "mdi:tree-outline" } } } } }
Range-based Icons (for numeric values):
{
"entity": { "sensor": { "battery_level": { "default": "mdi:battery-unknown", "range": { "0": "mdi:battery-outline", "90": "mdi:battery-90", "100": "mdi:battery" } } } } }
tests/components/{domain}/tests.commonconfig_flow.py; exercise the flow logic end-to-end.Integration-specific tests (recommended):
pytest ./tests/components/<integration_domain> \
--cov=homeassistant.components.<integration_domain>
--cov-report term-missing
--durations-min=1
--durations=0
--numprocesses=auto
hass.data directly - Use fixtures and proper integration setup insteadasync def test_user_flow_success(hass, mock_api):
"""Test successful user flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
# Test form submission
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=TEST_USER_INPUT
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "My Device"
assert result["data"] == TEST_USER_INPUT
async def test_flow_connection_error(hass, mock_api_error):
"""Test connection error handling."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=TEST_USER_INPUT
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
@pytest.fixture
def platforms() -> list[Platform]:
"""Overridden fixture to specify platforms to test."""
return [Platform.SENSOR] # Or another specific platform as needed.
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "init_integration")
async def test_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test the sensor entities."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
# Ensure entities are correctly assigned to device
device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, "device_unique_id")}
)
assert device_entry
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
for entity_entry in entity_entries:
assert entity_entry.device_id == device_entry.id
# Modern integration fixture setup
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
title="My Integration",
domain=DOMAIN,
data={CONF_HOST: "127.0.0.1", CONF_API_KEY: "test_key"},
unique_id="device_unique_id",
)
@pytest.fixture
def mock_device_api() -> Generator[MagicMock]:
"""Return a mocked device API."""
with patch("homeassistant.components.my_integration.MyDeviceAPI", autospec=True) as api_mock:
api = api_mock.return_value
api.get_data.return_value = MyDeviceData.from_json(
load_fixture("device_data.json", DOMAIN)
)
yield api
@pytest.fixture
def platforms() -> list[Platform]:
"""Fixture to specify platforms to test."""
return PLATFORMS
@pytest.fixture
async def init_integration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_device_api: MagicMock,
platforms: list[Platform],
) -> MockConfigEntry:
"""Set up the integration for testing."""
mock_config_entry.add_to_hass(hass)
with patch("homeassistant.components.my_integration.PLATFORMS", platforms):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
return mock_config_entry
manifest.json syntax and required fieldsunique_id and has_entity_name implementationstrings.json entries and error handling# Enable debug logging in tests
caplog.set_level(logging.DEBUG, logger="my_integration")
# In integration code - use proper logging
_LOGGER = logging.getLogger(__name__)
_LOGGER.debug("Processing data: %s", data) # Use lazy logging
# Check specific integration
python -m script.hassfest --integration-path homeassistant/components/my_integration
# Validate quality scale
# Check quality_scale.yaml against current rules
# Run integration tests with coverage
pytest ./tests/components/my_integration \
--cov=homeassistant.components.my_integration \
--cov-report term-missing
Weekly Installs
–
Repository
GitHub Stars
85.8K
First Seen
–
Security Audits
agent-browser 浏览器自动化工具 - Vercel Labs 命令行网页操作与测试
147,400 周安装
cloud_polling, local_polling, local_push)zeroconf, dhcp, bluetooth, ssdp, usbapplication_credentials, bluetooth_adapters)except MyException: errors["base"] = "cannot_connect"
Duplicate Prevention : Prevent duplicate configurations:
# Using unique ID
await self.async_set_unique_id(identifier) self._abort_if_unique_id_configured()
self._async_abort_entries_match({CONF_HOST: user_input[CONF_HOST]})