1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

E:\pyproject\core\
├── homeassistant/          # 核心源代码主目录
│   ├── core.py             # HomeAssistant 主类、EventBus、StateMachine、ServiceRegistry
│   ├── config_entries.py   # ConfigEntry、ConfigFlow、ConfigEntries 管理器
│   ├── bootstrap.py        # 启动引导，分阶段加载
│   ├── loader.py           # Integration 类，manifest 解析，自定义集成管理
│   ├── setup.py            # async_setup_component，依赖解析
│   ├── data_entry_flow.py  # FlowHandler/FlowManager 基类
│   ├── config.py           # 配置文件处理
│   ├── const.py            # 全局常量、Platform 枚举
│   ├── runner.py           # 运行器，事件循环设置
│   ├── __main__.py         # CLI 入口
│   ├── auth/               # 认证系统
│   ├── components/         # 所有内置集成（2000+个）
│   ├── generated/          # 自动生成的代码（蓝牙、DHCP、SSDP 数据等）
│   ├── helpers/            # 辅助模块（entity、entity_registry、template 等）
│   └── util/               # 工具模块
├── tests/                  # 测试代码
├── custom_components/      # 自定义集成目录
└── pyproject.toml          # 项目配置

1
2
3
4
5
6
7
8

components/<domain>/
├── manifest.json       # 集成清单（必需）
├── __init__.py         # 集成入口
├── config_flow.py      # 配置流（可选，支持 UI 配置时需要）
├── light.py            # 向 light 域提供实体（可选平台文件）
├── sensor.py           # 向 sensor 域提供实体（可选平台文件）
├── strings.json        # 前端字符串
└── services.yaml       # 服务定义

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

{
  "domain": "hue",
  "name": "Philips Hue",
  "integration_type": "hub",
  "config_flow": true,
  "dependencies": [],
  "after_dependencies": [],
  "requirements": ["aiohue==4.8.1"],
  "iot_class": "local_push",
  "zeroconf": ["_hue._tcp.local."],
  "codeowners": ["@marcelveldt"]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

class HomeAssistant:
    def __init__(self, config_dir: str):
        self.data = HassDict()                    # 全局数据存储字典
        self.loop = asyncio.get_running_loop()    # 事件循环
        self.bus = EventBus(self)                 # 事件总线
        self.services = ServiceRegistry(self)     # 服务注册表
        self.states = StateMachine(self.bus, self.loop)  # 状态机
        self.config = Config(self, config_dir)     # 配置
        self.config_entries: ConfigEntries         # 配置条目管理器
        self.auth: AuthManager                     # 认证管理器
        self.state: CoreState = CoreState.not_running
        self.exit_code: int = 0
        self.timeout = TimeoutManager()
        self.import_executor = InterruptibleThreadPoolExecutor(...)

1

NOT_RUNNING → STARTING → RUNNING → STOPPING → FINAL_WRITE → STOPPED

1
2
3

class EventBus:
    def async_listen(self, event_type, listener) -> Callable:  # 注册监听器
    def async_fire(self, event_type, event_data) -> None:       # 发射事件

1
2
3
4
5

class StateMachine:
    def async_set(self, entity_id, state, attributes) -> None  # 设置状态
    def async_get(self, entity_id) -> State | None              # 获取状态
    def async_all(self) -> list[State]                          # 获取所有状态
    def async_remove(self, entity_id) -> bool                   # 移除状态

1
2
3
4

class ServiceRegistry:
    def async_register(self, domain, service, service_func, schema) -> None
    def async_call(self, domain, service, service_data, blocking) -> Any
    def async_remove(self, domain, service) -> None

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

class Entity(metaclass=ABCCachedProperties, cached_properties=...):
    entity_id: str = None
    hass: HomeAssistant = None
    platform: EntityPlatform = None
    entity_description: EntityDescription

    # 核心属性
    @property
    def state(self) -> StateType                        # 实体状态值
    @property
    def capability_attributes(self) -> dict             # 能力属性
    @property
    def name(self) -> str | None                        # 名称
    @property
    def icon(self) -> str | None                        # 图标
    @property
    def device_class(self) -> str | None                # 设备类
    @property
    def unit_of_measurement(self) -> str | None         # 计量单位
    @property
    def supported_features(self) -> int                 # 支持的特性标志位
    @property
    def available(self) -> bool                         # 是否可用
    @property
    def should_poll(self) -> bool                       # 是否需要轮询（默认 True）

    # 核心方法
    def async_write_ha_state(self) -> None              # 写入状态到状态机（推荐）
    def async_update_ha_state(self, force_refresh) -> None  # 更新状态
    def async_device_update(self) -> None               # 从设备拉取最新数据
    def async_added_to_hass(self) -> None               # 添加到 HA 后的钩子
    def async_will_remove_from_hass(self) -> None       # 从 HA 移除前的钩子
    def async_on_remove(self, func) -> None             # 注册移除时的清理回调

1
2
3

class MyEntity(Entity):
    _attr_should_poll = False     # 等价于 should_poll = False
    _attr_icon = "mdi:lightbulb"  # 等价于 icon = "mdi:lightbulb"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

class ToggleEntity(Entity):
    _attr_is_on: bool | None = None

    @property
    @final
    def state(self) -> Literal["on", "off"] | None:
        if (is_on := self.is_on) is None: return None
        return STATE_ON if is_on else STATE_OFF

    async def async_turn_on(self, **kwargs) -> None: ...   # 子类必须实现
    async def async_turn_off(self, **kwargs) -> None: ...   # 子类必须实现
    async def async_toggle(self, **kwargs) -> None: ...     # 默认实现调用 turn_on/off

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

class ConfigEntry[_DataT = Any]:
    entry_id: str           # 唯一 ID (ULID)
    domain: str             # 所属集成域名
    title: str              # 显示标题
    data: MappingProxyType  # 配置数据（如 host, api_key）
    runtime_data: _DataT    # 运行时数据（泛型，集成可自定义类型）
    options: MappingProxyType  # 用户选项
    unique_id: str | None   # 去重 ID
    state: ConfigEntryState # 状态
    version: int            # 配置迁移版本
    minor_version: int      # 配置迁移次版本号
    source: str             # 来源: user, bluetooth, dhcp, zeroconf, ...
    subentries: MappingProxyType[str, ConfigSubentry]  # 子条目
    disabled_by: ConfigEntryDisabler | None

1
2
3
4
5
6

NOT_LOADED → SETUP_IN_PROGRESS → LOADED
                              → SETUP_ERROR
                              → SETUP_RETRY
                              → MIGRATION_ERROR
LOADED → UNLOAD_IN_PROGRESS → NOT_LOADED
                            → FAILED_UNLOAD

1
2
3
4
5
6
7

type HueConfigEntry = ConfigEntry[HueBridge]

# 在 async_setup_entry 中：
entry.runtime_data = bridge

# 在平台文件中：
bridge = config_entry.runtime_data  # 自动具有 HueBridge 类型

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

class Integration:
    domain: str
    name: str
    integration_type: str
    dependencies: list[str]
    after_dependencies: list[str]
    requirements: list[str]
    disabled: str | None
    manifest: dict

    @classmethod
    def resolve_from_root(cls, hass, root_module, domain) -> Integration | None

    async def async_get_component(self) -> ModuleType     # 获取 __init__.py 模块
    async def async_get_platform(self, platform) -> ModuleType  # 获取平台模块
    async def resolve_dependencies(self) -> bool | None

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

__main__.py::main()
  ├── validate_python()              # 校验 Python 版本
  ├── validate_os()                 # 校验操作系统
  ├── ensure_config_path()          # 确保配置目录存在
  ├── restore_backup()              # 检查是否需要从备份恢复
  └── runner.run(runtime_config)
        ├── HassEventLoopPolicy 设置  # 自定义事件循环策略
        │     ├── 64 workers 线程池
        │     ├── 调试模式
        │     └── loop.time = monotonic
        ├── asyncio.new_event_loop()
        └── loop.run_until_complete(setup_and_run_hass())
              ├── bootstrap.async_setup_hass()
              │     ├── create_hass()
              │     │     ├── core.HomeAssistant(config_dir)
              │     │     │     ├── EventBus
              │     │     │     ├── ServiceRegistry
              │     │     │     └── StateMachine
              │     │     ├── loader.async_setup(hass)
              │     │     └── async_enable_logging()
              │     ├── conf_util.async_ensure_config_exists()
              │     ├── conf_util.process_ha_config_upgrade()
              │     ├── conf_util.async_hass_config_yaml()
              │     └── async_from_config_dict(config_dict, hass)
              │           ├── ConfigEntries(hass, config)
              │           ├── loader.async_get_custom_components()
              │           ├── async_load_base_functionality()
              │           ├── async_setup_component("homeassistant")
              │           ├── async_setup_component("persistent_notification")
              │           ├── async_process_ha_core_config()
              │           └── _async_set_up_integrations()
              │                 ├── 解析依赖关系
              │                 ├── Stage 0: 核心基础设施
              │                 ├── Stage 1: 发现和云服务
              │                 └── Stage 2: 所有其余集成
              └── hass.async_run()
                    ├── hass.async_start()
                    │     ├── CoreState → STARTING
                    │     ├── fire EVENT_HOMEASSISTANT_START
                    │     ├── CoreState → RUNNING
                    │     └── fire EVENT_HOMEASSISTANT_STARTED
                    └── await _stopped.wait()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

await asyncio.gather(
    entity.async_setup(hass),        # 实体源
    frame.async_setup(hass),          # 帧追踪
    template.async_setup(hass),       # 模板引擎
    translation.async_setup(hass),    # 翻译
    device_registry.async_load(),     # 设备注册表
    area_registry.async_load(),       # 区域注册表
    entity_registry.async_load(),     # 实体注册表
    floor_registry.async_load(),      # 楼层注册表
    issue_registry.async_load(),      # 问题注册表
    label_registry.async_load(),      # 标签注册表
    restore_state.async_load(),       # 状态恢复
    hass.config_entries.async_initialize(),  # 配置条目
    ...
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

async_setup_component(hass, domain, config)
  │
  ├── 1. 检查 hass.config.components — 已加载则返回 True
  ├── 2. 检查 _DATA_SETUP — 正在加载则 await 已有 Future
  ├── 3. 创建新 Future 注册到 _DATA_SETUP
  │
  └── 4. 调用 _async_setup_component(hass, domain, config)
        │
        ├── 阶段1: 获取 Integration 定义
        │     └── loader.async_get_integration(hass, domain)
        │           ├── 先查 DATA_INTEGRATIONS 缓存
        │           ├── 内置集成: Integration.resolve_from_root(hass, components, domain)
        │           └── 自定义集成: 从 custom_components 包解析
        │
        ├── 阶段2: 检查集成是否被禁用
        │
        ├── 阶段3: 预加载翻译（异步启动，稍后等待）
        │
        ├── 阶段4: 解析依赖关系
        │     └── integration.resolve_dependencies()
        │           检查依赖是否存在 + 循环依赖检测
        │
        ├── 阶段5: 处理依赖和需求
        │     └── async_process_deps_reqs(hass, config, integration)
        │           ├── _async_process_dependencies() — 并行加载 dependencies + after_dependencies
        │           │     ├── dependencies: 强制加载（async_setup_component）
        │           │     └── after_dependencies: 仅等待已计划安装的依赖
        │           └── requirements.async_get_integration_with_requirements() — 安装 pip 包
        │
        ├── 阶段6: 导入组件模块
        │     └── integration.async_get_component() → importlib.import_module()
        │
        ├── 阶段7: 校验配置
        │     ├── conf_util.async_process_component_config() — 用 CONFIG_SCHEMA 校验
        │     ├── conf_util.async_handle_component_errors()
        │     └── conf_util.async_drop_config_annotations()
        │
        ├── 阶段8: 执行安装
        │     ├── 如果有 component.async_setup → task = component.async_setup(hass, config)
        │     ├── 否则如果有 component.setup → 线程池执行 component.setup(hass, config)
        │     ├── 否则如果没有 async_setup_entry → 记录错误返回 False
        │     ├── async_timeout(SLOW_SETUP_MAX_WAIT=300s) 等待结果
        │     └── 等待翻译加载完成
        │
        ├── 阶段9: 等待 Config Flow 导入完成
        │     └── hass.config_entries.flow.async_wait_import_flow_initialized(domain)
        │
        └── 阶段10: 设置 Config Entry + 标记完成
              ├── hass.config.components.add(domain)  ← 先标记，防止死锁
              ├── 并行安装所有 ConfigEntry:
              │     await asyncio.gather(*(
              │       entry.async_setup_locked(hass, integration=integration)
              │       for entry in hass.config_entries.async_entries(domain)
              │     ))
              └── 触发 EVENT_COMPONENT_LOADED 事件

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# 伪代码
for dep in integration.dependencies:
    if dep not in setup and dep not in components:
        create_task(async_setup_component(hass, dep, config))  # 强制安装

for dep in integration.after_dependencies:
    if dep in setup_done_tasks:  # 只等待已计划的
        wait_tasks.append(setup_done_tasks[dep])

await asyncio.gather(*all_tasks)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

entry.async_setup_locked(hass, integration=integration)
  │
  ├── 检查 entry.state (必须是 NOT_LOADED)
  ├── entry.state = SETUP_IN_PROGRESS
  │
  └── component.async_setup_entry(hass, entry)
        │  ← 集成 __init__.py 中定义的 async_setup_entry 函数
        │
        ├── 初始化连接/API
        ├── 创建运行时对象（如 HueBridge）
        ├── entry.runtime_data = 运行时对象
        ├── hass.config_entries.async_forward_entry_setups(entry, platforms)
        │     └── 转发到各平台设置
        └── 注册更新监听器等

1
2
3
4

FlowHandler (data_entry_flow.py)
  └── ConfigEntryBaseFlow (config_entries.py)
        ├── ConfigFlow   — 新设备配置流程
        └── OptionsFlow  — 已配置设备的选项流程

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

class FlowHandler:
    # 结果返回方法
    def async_show_form(self, *, step_id, data_schema, errors, ...) → FORM
    def async_create_entry(self, *, title, data, ...) → CREATE_ENTRY
    def async_abort(self, *, reason, ...) → ABORT
    def async_external_step(self, *, step_id, url, ...) → EXTERNAL_STEP
    def async_external_step_done(self, *, next_step_id) → EXTERNAL_STEP_DONE
    def async_show_progress(self, *, step_id, progress_action, progress_task, ...) → SHOW_PROGRESS
    def async_show_progress_done(self, *, next_step_id) → SHOW_PROGRESS_DONE
    def async_show_menu(self, *, step_id, menu_options, ...) → MENU

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

FlowManager.async_init(handler, *, context, data)
  │
  ├── async_create_flow(handler, context, data) → 创建 FlowHandler 实例
  ├── 分配 flow_id (UUID)
  ├── 注册到 _progress 索引
  └── _async_handle_step(flow, flow.init_step, data)
        │  ← init_step 默认为 "init"，但 ConfigFlow 中被覆盖为 context["source"]
        │
        ├── getattr(flow, f"async_step_{step_id}")(user_input)
        │     ← 反射调用步骤方法
        │
        ├── 如果结果在 FLOW_NOT_COMPLETE_STEPS 中 → 设置 cur_step，流程继续
        └── 否则 → async_finish_flow()，流程结束

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

class ConfigFlow(ConfigEntryBaseFlow):
    # 自动注册：子类声明 domain 时自动注册到 HANDLERS 注册表
    def __init_subclass__(cls, *, domain=None, **kwargs):
        if domain is not None:
            HANDLERS.register(domain)(cls)

    # unique_id 管理
    async def async_set_unique_id(self, unique_id, *, raise_on_progress=True) -> ConfigEntry | None
    def _abort_if_unique_id_configured(self, updates=None, *, error="already_configured") -> None

    # 发现步骤方法（全部默认委托到 _async_step_discovery_without_unique_id → async_step_user）
    async def async_step_user(self, user_input)               # 用户手动添加
    async def async_step_bluetooth(self, discovery_info)      # 蓝牙发现
    async def async_step_dhcp(self, discovery_info)           # DHCP 发现
    async def async_step_zeroconf(self, discovery_info)       # mDNS/Zeroconf 发现
    async def async_step_homekit(self, discovery_info)        # HomeKit 发现
    async def async_step_hassio(self, discovery_info)         # Hass.io 发现
    async def async_step_ssdp(self, discovery_info)           # SSDP 发现
    async def async_step_usb(self, discovery_info)            # USB 发现
    async def async_step_mqtt(self, discovery_info)            # MQTT 发现
    async def async_step_integration_discovery(self, discovery_info)  # 集成发现
    async def async_step_import(self, import_data)            # YAML 导入
    async def async_step_ignore(self, user_input)             # 忽略发现
    async def async_step_reauth(self, user_input)             # 重新认证
    async def async_step_reconfigure(self, user_input)        # 重新配置

    # 选项流支持
    @staticmethod
    def async_get_options_flow(config_entry) -> OptionsFlow  # 子类覆盖以支持选项

    # 结果方法
    def async_create_entry(self, *, title, data, options, ...) -> ConfigFlowResult
    def async_abort(self, *, reason, next_flow, ...) -> ConfigFlowResult
    def async_update_and_abort(self, entry, *, unique_id, title, data, ...) -> ConfigFlowResult
    def async_update_reload_and_abort(self, entry, ...) -> ConfigFlowResult

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

1. 用户点击"添加集成"
   ↓
2. ConfigEntriesFlowManager.async_init("my_integration", context={"source": "user"})
   ↓
3. async_create_flow() → 实例化 MyConfigFlow, 设置 init_step="user"
   ↓
4. _async_handle_step(flow, "user", None)
   ↓
5. MyConfigFlow.async_step_user(None)
   → 返回 async_show_form(step_id="user", data_schema=SCHEMA)
   ↓
6. 用户填写表单，前端调用 async_configure(flow_id, user_input)
   ↓
7. Schema 校验 user_input
   ↓
8. _async_handle_step(flow, "user", validated_input)
   ↓
9. MyConfigFlow.async_step_user(validated_input)
   → 调用 async_set_unique_id(device_id)
   → 返回 async_create_entry(title="My Device", data={...})
   ↓
10. async_finish_flow() → 创建 ConfigEntry → async_add() → 完成

1
2
3
4
5
6
7

1. 蓝牙发现设备
   → async_init("my_integration", context={"source": "bluetooth"}, data=BluetoothServiceInfoBleak(...))
   ↓
2. init_step="bluetooth" → 调用 async_step_bluetooth(discovery_info)
   ↓
3. 如果未覆盖: 默认实现调用 _async_step_discovery_without_unique_id() → async_step_user()
   如果已覆盖: 通常先 async_set_unique_id()，再返回表单或直接创建条目

1
2
3
4
5
6
7

class OptionsFlow(ConfigEntryBaseFlow):
    @property
    def config_entry(self) -> ConfigEntry: ...  # 关联的 ConfigEntry

# 常用变体
class OptionsFlowWithReload(OptionsFlow):
    automatic_reload: bool = True  # 选项变更后自动重新加载 ConfigEntry

1
2
3
4
5

class MyConfigFlow(ConfigFlow, domain="my_integration"):
    @staticmethod
    @callback
    def async_get_options_flow(config_entry) -> OptionsFlow:
        return MyOptionsFlow(config_entry)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

创建实体实例
  │
  ↓ EntityPlatform.async_add_entities() → _async_add_entity()
  │
  ├── entity.add_to_platform_start()
  │     ├── 设置 entity.hass = hass
  │     ├── 设置 entity.platform = entity_platform
  │     └── 设置 parallel_updates 信号量
  │
  ├── 可选: entity.async_device_update()  (update_before_add=True 时)
  │
  ├── 生成/查找 entity_id
  │     ├── 有 unique_id → 通过 entity_registry 查找或创建
  │     └── 无 unique_id → 根据 suggested_object_id 生成
  │
  ├── hass.states.async_reserve(entity_id)  防止并发冲突
  │
  ├── 注册移除回调
  │
  └── entity.add_to_platform_finish()
        ├── 调用 async_internal_added_to_hass()
        ├── 调用 async_added_to_hass()  ← 子类覆盖的钩子
        └── async_write_ha_state()  写入初始状态

1
2
3
4
5
6
7
8

# 推荐方式 — 立即写入状态
entity.async_write_ha_state()

# 传统方式 — 先更新再写入
entity.async_update_ha_state(force_refresh=False)

# 从设备拉取最新数据
entity.async_device_update()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

class LightEntity(ToggleEntity):
    # 核心属性
    brightness: int | None                      # 亮度 0..255
    color_mode: ColorMode | None                # 当前颜色模式
    supported_color_modes: set[ColorMode] | None # 支持的颜色模式
    hs_color: tuple[float, float] | None        # HS 颜色
    rgb_color: tuple[int, int, int] | None      # RGB 颜色
    color_temp_kelvin: int | None               # 色温(开尔文)
    min_color_temp_kelvin: int                  # 最小色温
    max_color_temp_kelvin: int                  # 最大色温
    effect: str | None                          # 当前效果
    effect_list: list[str] | None               # 效果列表

    # 核心方法
    async def async_turn_on(self, **kwargs)     # 开灯
    async def async_turn_off(self, **kwargs)     # 关灯
    async def async_toggle(self, **kwargs)       # 切换（重写自 ToggleEntity）

    # 色彩模式 ColorMode
    # ONOFF, BRIGHTNESS, HS, XY, RGB, RGBW, RGBWW, COLOR_TEMP, WHITE, UNKNOWN

    # 特性标志 LightEntityFeature (IntFlag)
    # EFFECT = 4, FLASH = 8, TRANSITION = 32

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# light/__init__.py
async def async_setup(hass, config):
    component = EntityComponent[LightEntity](_LOGGER, DOMAIN, hass, SCAN_INTERVAL)
    await component.async_setup(config)
    # 注册实体服务
    component.async_register_entity_service(SERVICE_TURN_ON, ..., async_handle_light_on_service)
    component.async_register_entity_service(SERVICE_TURN_OFF, ..., async_handle_light_off_service)
    component.async_register_entity_service(SERVICE_TOGGLE, ..., async_handle_toggle_service)

async def async_setup_entry(hass, entry):
    return await hass.data[DATA_COMPONENT].async_setup_entry(entry)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

class SensorEntity(Entity):
    native_value: StateType | date | datetime | Decimal  # 传感器原始值
    native_unit_of_measurement: str | None               # 原始计量单位
    device_class: SensorDeviceClass                       # 设备类
    state_class: SensorStateClass                         # 状态类 (MEASUREMENT/TOTAL/TOTAL_INCREASING)
    suggested_display_precision: int | None               # 建议显示精度
    suggested_unit_of_measurement: str | None              # 建议显示单位
    options: list[str] | None                              # 枚举选项

    # state 属性是 @final，不可重写
    # 内部自动处理: 单位转换、精度、校验

1
2
3
4
5

class BinarySensorEntity(Entity):
    is_on: bool | None              # 核心属性
    device_class: BinarySensorDeviceClass  # 设备类 (battery, motion, door, ...)

    # state 属性是 @final，根据 is_on 返回 "on" 或 "off"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

from homeassistant.helpers.entity import EntityDescription

# 定义描述
SENSOR_DESCRIPTION = SensorEntityDescription(
    key="temperature",
    device_class=SensorDeviceClass.TEMPERATURE,
    native_unit_of_measurement="°C",
    state_class=SensorStateClass.MEASUREMENT,
)

# 使用描述
class MySensor(SensorEntity):
    def __init__(self, description):
        self.entity_description = description
        self._attr_unique_id = f"{description.key}_sensor"
        self._native_value = None

    @property
    def native_value(self):
        return self._native_value

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

from homeassistant.helpers.update_coordinator import CoordinatorEntity, DataUpdateCoordinator

coordinator = DataUpdateCoordinator(
    hass, LOGGER,
    update_method=async_fetch_data,
    update_interval=timedelta(seconds=30),
)

class MyEntity(CoordinatorEntity, LightEntity):
    def __init__(self, coordinator, light_data):
        super().__init__(coordinator)
        self.light_data = light_data

    @property
    def brightness(self):
        return self.coordinator.data[self.light_data.id]["brightness"]

    # CoordinatorEntity 自动在 coordinator 刷新时调用 async_write_ha_state()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# 域级服务注册
hass.services.async_register(
    domain="light",
    service="turn_on",
    service_func=async_handle_turn_on,
    schema=TURN_ON_SCHEMA,
)

# 实体服务注册（更常用）
component.async_register_entity_service(
    SERVICE_TURN_ON,
    TURN_ON_SCHEMA,
    "async_turn_on",  # 字符串，映射到实体方法名
)

# 管理员服务注册
async_register_admin_service(hass, domain, service, service_func, schema)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

用户/自动化调用 light.turn_on(entity_id="light.living_room", brightness=255)
  │
  ├── ServiceRegistry.async_call("light", "turn_on", service_data)
  │     ├── 查找 handler = _services["light"]["turn_on"]
  │     ├── 校验 return_response 与 supports_response
  │     ├── Schema 校验 service_data
  │     ├── 创建 ServiceCall 对象
  │     ├── bus.fire(EVENT_CALL_SERVICE)
  │     └── 执行服务
  │
  └── 实体服务调用链:
        entity_service_call(hass, entities, service_func, call)
          ├── _resolve_entity_service_call_entities()  # 解析目标实体
          │     ├── 根据 entity_id/area_id/device_id 解析
          │     ├── 过滤不可用的实体
          │     ├── 过滤不满足 required_features 的实体
          │     └── 过滤不满足 entity_device_classes 的实体
          │
          └── _async_handle_entity_calls([(entity, _handle_single_entity_call(...))])
                ├── 单实体: entity.async_request_call(coro)
                ├── 多实体: asyncio.gather 并发调用
                └── _handle_single_entity_call()
                      ├── func 是字符串 → getattr(entity, func)(**data)
                      │                   如 entity.async_turn_on(brightness=255)
                      └── func 是 HassJob → hass.async_run_hass_job(func, entity, data)

1
2
3
4
5

component.async_register_batched_entity_service(
    SERVICE_GET_FORECASTS,
    GET_FORECASTS_SCHEMA,
    async_get_forecasts,  # func(entities, service_call)
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

class EntityComponent[_EntityT = Entity]:
    def __init__(self, logger, domain, hass, scan_interval):
        # 创建根平台（与域同名的 "catch-all" 平台）
        self._platforms: dict[str | tuple, EntityPlatform] = {}
        self._entities: dict[str, Entity]  # 引用根平台的 domain_entities
        hass.data[DATA_INSTANCES][domain] = self  # 全局注册

    async def async_setup(self, config):
        # 遍历配置中的平台，为每个平台创建异步任务

    async def async_setup_entry(self, config_entry):
        # 导入平台模块，创建 EntityPlatform，调用 EntityPlatform.async_setup_entry()

    async def async_unload_entry(self, config_entry):
        # 弹出平台，调用 EntityPlatform.async_reset()

    async_register_entity_service(...)     # 注册逐实体服务
    async_register_batched_entity_service(...)  # 注册批量服务

1
2
3
4
5
6
7

class EntityPlatform:
    def __init__(self, *, hass, logger, domain, platform_name, platform, scan_interval, entity_namespace):
        self.entities: dict[str, Entity] = {}        # 当前平台的实体
        self.domain_entities: dict[str, Entity]        # 按域索引的所有实体
        self.domain_platform_entities: dict[str, Entity]  # 按(域,平台名)索引的实体
        self.config_entry: ConfigEntry | None = None
        self._setup_complete: bool = False

1
2
3
4
5
6
7
8
9

路径 A: YAML 配置
  EntityPlatform.async_setup(platform_config, discovery_info)
    ├── 检查平台模块是否有 async_setup_platform / setup_platform
    └── 调用 platform.async_setup_platform(hass, config, async_add_entities, discovery_info)

路径 B: ConfigEntry
  EntityPlatform.async_setup_entry(config_entry)
    ├── 保存 self.config_entry = config_entry
    └── 调用 platform.async_setup_entry(hass, config_entry, async_add_entities)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

集成代码调用 async_add_entities(entities)
  │  ← 这是 EntityPlatform 传给平台的回调
  │
  ├── _async_schedule_add_entities() / _async_schedule_add_entities_for_entry()
  │     └── 创建 task → async_add_entities()
  │
  └── async_add_entities(new_entities, update_before_add, *, config_subentry_id=None)
        ├── 验证 config_subentry_id
        ├── update_before_add=True → _async_add_and_update_entities() (并发)
        └── update_before_add=False → _async_add_entities() (顺序)
              │
              └── 逐个调用 _async_add_entity(entity, ...)
                    ├── entity.add_to_platform_start()  # 设置 hass/platform 引用
                    ├── 可选: entity.async_device_update()
                    ├── 生成/查找 entity_id
                    │     ├── 有 unique_id → entity_registry 查找/创建
                    │     └── 无 unique_id → suggested_object_id 生成
                    ├── 禁用检查 → 已禁用则跳过
                    ├── 注册到三个字典:
                    │     self.entities[entity_id] = entity
                    │     self.domain_entities[entity_id] = entity
                    │     self.domain_platform_entities[entity_id] = entity
                    ├── hass.states.async_reserve(entity_id)
                    └── entity.add_to_platform_finish()
                          ├── async_internal_added_to_hass()
                          ├── async_added_to_hass()  ← 子类钩子
                          └── async_write_ha_state()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

_async_handle_interval_callback()
  ├── 重新调度自身（scan_interval 秒后）
  └── 创建后台任务 → _async_update_entity_states()
                        ├── 少量实体: 顺序更新
                        └── 多量实体: 并发 gather 更新

并行度控制: PARALLEL_UPDATES 常量
  ├── 平台模块可定义 PARALLEL_UPDATES 常量
  ├── 未定义且有同步 update 方法 → 默认为 1
  └── 设为 0 则不限制并行

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

HomeAssistant (hass)
  └─ EntityComponent (如 "light")
       │  管理 domain 级别的逻辑
       │
       ├─ EntityPlatform (platform_name="light")  ← 根平台，catch-all
       ├─ EntityPlatform (platform_name="hue")    ← hue 集成的 light 平台
       ├─ EntityPlatform (platform_name="tplink") ← tplink 集成的 light 平台
       └─ EntityPlatform (platform_name="zwave")  ← zwave 集成的 light 平台
            │
            └─ entities: {entity_id: Entity}  ← 该平台管理的所有实体

1
2
3
4
5

用户发现/配置桥接器
    → ConfigFlow 创建 ConfigEntry
    → async_setup_entry 创建 HueBridge 实例
    → HueBridge 初始化 API 并转发平台设置
    → 各平台 (light/sensor/...) 的 async_setup_entry 创建实体

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

DOMAIN = "hue"
type HueConfigEntry = ConfigEntry[HueBridge]

async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
    async_setup_services(hass)  # 注册全局服务
    return True

async def async_setup_entry(hass: HomeAssistant, entry: HueConfigEntry) -> bool:
    await check_migration(hass, entry)
    bridge = HueBridge(hass, entry)
    if not await bridge.async_initialize_bridge():
        return False
    # unique_id 修正 & 设备注册
    return True

async def async_unload_entry(hass: HomeAssistant, entry: HueConfigEntry) -> bool:
    return await entry.runtime_data.async_reset()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

class HueBridge:
    def __init__(self, hass, config_entry):
        self.config_entry = config_entry
        if self.api_version == 1:
            self.api = HueBridgeV1(self.host, app_key, session)
        else:
            self.api = HueBridgeV2(self.host, app_key)
        self.config_entry.runtime_data = self  # 自注册

    async def async_initialize_bridge(self) -> bool:
        # 1. 连接 API
        try:
            async with asyncio.timeout(10):
                await self.api.initialize()
        except Unauthorized:
            create_config_flow(self.hass, self.host)  # 重新触发配对
            return False
        except (TimeoutError, ...):
            raise ConfigEntryNotReady(...)  # HA 会自动重试

        # 2. 根据版本转发平台设置
        if self.api_version == 1:
            await self.hass.config_entries.async_forward_entry_setups(
                self.config_entry, PLATFORMS_v1
            )  # [BINARY_SENSOR, LIGHT, SENSOR]
        else:
            await self.hass.config_entries.async_forward_entry_setups(
                self.config_entry, PLATFORMS_v2
            )  # [BINARY_SENSOR, EVENT, LIGHT, SCENE, SENSOR, SWITCH]

        # 3. 注册选项更新监听器
        self.reset_jobs.append(
            self.config_entry.add_update_listener(_update_listener)
        )
        return True

    async def async_reset(self):
        # 卸载所有平台 + 清理
        while self.reset_jobs:
            self.reset_jobs.pop()()
        unload_success = await self.hass.config_entries.async_unload_platforms(
            self.config_entry, PLATFORMS_v1 if ... else PLATFORMS_v2
        )
        if unload_success:
            delattr(self.config_entry, "runtime_data")
        return unload_success

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

class HueFlowHandler(ConfigFlow, domain=DOMAIN):
    VERSION = 1

    async def async_step_user(self, user_input=None):
        # 展示桥接器列表或手动输入
        if user_input is None:
            bridges = await discover_nupnp()
            return self.async_show_form(step_id="init", ...)

    async def async_step_link(self, user_input=None):
        # 请求用户按 Link 按钮
        if user_input is None:
            return self.async_show_form(step_id="link")

        try:
            app_key = await create_app_key(bridge.host, ...)
        except LinkButtonNotPressed:
            errors["base"] = "register_failed"

        return self.async_create_entry(
            title=f"Hue Bridge {bridge.id}",
            data={CONF_HOST: bridge.host, CONF_API_KEY: app_key, CONF_API_VERSION: 2},
        )

    # 三种发现入口
    async def async_step_zeroconf(self, discovery_info): ...  # mDNS 发现
    async def async_step_homekit(self, discovery_info): ...  # HomeKit 发现
    async def async_step_import(self, import_data): ...      # YAML 导入

1
2
3
4
5
6

async def async_setup_entry(hass, config_entry, async_add_entities):
    bridge = config_entry.runtime_data
    if bridge.api_version == 1:
        await setup_entry_v1(hass, config_entry, async_add_entities)
    else:
        await setup_entry_v2(hass, config_entry, async_add_entities)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

async def setup_entry_v1(hass, config_entry, async_add_entities):
    bridge = config_entry.runtime_data

    # 创建 DataUpdateCoordinator（5秒轮询）
    coordinator = DataUpdateCoordinator(
        hass, LOGGER, name="light",
        update_method=partial(async_safe_fetch, bridge, bridge.api.lights.update),
        update_interval=timedelta(seconds=5),
    )
    await coordinator.async_refresh()
    if not coordinator.last_update_success:
        raise PlatformNotReady

    # 创建实体
    async_add_entities(HueLight(coordinator, light) for light in bridge.api.lights.values())

class HueLight(CoordinatorEntity, LightEntity):
    _attr_should_poll = False  # 由 Coordinator 管理更新

    async def async_turn_on(self, **kwargs):
        await bridge.async_request_call(self.light.set_state, **command)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

async def setup_entry_v2(hass, config_entry, async_add_entities):
    bridge = config_entry.runtime_data
    controller = bridge.api.lights

    # 立即添加所有当前灯
    async_add_entities(make_light_entity(light) for light in controller)

    # 注册事件监听器，动态添加新灯
    config_entry.async_on_unload(
        controller.subscribe(async_add_light, event_filter=EventType.RESOURCE_ADDED)
    )

class HueLight(HueBaseEntity, LightEntity):
    _attr_should_poll = False  # V2 事件驱动，不需要轮询

class HueBaseEntity(Entity):
    _attr_should_poll = False

    async def async_added_to_hass(self):
        # 订阅资源更新和删除事件
        self.async_on_remove(
            self.controller.subscribe(
                self._handle_event, self.resource.id,
                (EventType.RESOURCE_UPDATED, EventType.RESOURCE_DELETED),
            )
        )

    def _handle_event(self, event_type, resource):
        if event_type == EventType.RESOURCE_DELETED:
            # 自动移除实体
        else:
            self.async_write_ha_state()  # 推送状态更新

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

┌─────────────────────────────────────────────────────────────────┐
│  1. 发现阶段 (config_flow.py)                                    │
│                                                                 │
│  Zeroconf/HomeKit/手动 → HueFlowHandler                          │
│    → async_step_link() 请求用户按 Link 按钮                     │
│    → create_app_key() 获取 API 密钥                             │
│    → async_create_entry(data={host, api_key, api_version})      │
└───────────────────────────┬─────────────────────────────────────┘
                            │ 创建 ConfigEntry
                            ▼
┌─────────────────────────────────────────────────────────────────┐
│  2. 集成初始化 (__init__.py)                                     │
│                                                                 │
│  async_setup_entry(hass, entry):                                │
│    → bridge = HueBridge(hass, entry)                            │
│    → bridge.async_initialize_bridge()                           │
│        → api.initialize()  连接 API                              │
│        → async_forward_entry_setups()  转发到各平台               │
│    → entry.runtime_data = bridge                                │
└───────────────────────────┬─────────────────────────────────────┘
                            │ 转发到平台
                            ▼
┌─────────────────────────────────────────────────────────────────┐
│  3. 平台初始化 (light.py → v1/light.py 或 v2/light.py)         │
│                                                                 │
│  V1: DataUpdateCoordinator(5秒轮询) → HueLight(CoordinatorEntity)│
│  V2: 事件驱动 → HueLight(HueBaseEntity) → async_write_ha_state  │
└───────────────────────────┬─────────────────────────────────────┤
                            │ 实体运行
                            ▼
┌─────────────────────────────────────────────────────────────────┐
│  4. 实体运行时                                                   │
│                                                                 │
│  V1: Coordinator 定时轮询 → CoordinatorEntity 自动更新状态      │
│  V2: SSE 事件推送 → HueBaseEntity._handle_event → async_write   │
│  控制: bridge.async_request_call(api_method, **params)          │
│  卸载: bridge.async_reset() → unload_platforms → 清理          │
└─────────────────────────────────────────────────────────────────┘

1
2
3
4

custom_components/my_integration/
├── manifest.json
├── __init__.py
└── sensor.py          # 向 sensor 域提供实体（可选）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

{
  "domain": "my_integration",
  "name": "My Integration",
  "codeowners": ["@your_github"],
  "config_flow": true,
  "documentation": "https://www.example.com",
  "integration_type": "device",
  "iot_class": "local_polling",
  "requirements": ["my-library==1.0.0"],
  "zeroconf": ["_myservice._tcp.local."]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.const import Platform

DOMAIN = "my_integration"
type MyConfigEntry = ConfigEntry[MyRuntimeData]

PLATFORMS = [Platform.SENSOR, Platform.BINARY_SENSOR]


async def async_setup_entry(hass: HomeAssistant, entry: MyConfigEntry) -> bool:
    """从 ConfigEntry 设置集成。"""
    # 1. 初始化运行时数据
    runtime_data = MyRuntimeData(hass, entry)
    entry.runtime_data = runtime_data

    # 2. 初始化连接
    try:
        await runtime_data.async_connect()
    except ConnectionError as err:
        raise ConfigEntryNotReady(f"Connection failed: {err}") from err

    # 3. 转发平台设置
    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)

    # 4. 注册选项更新监听器
    entry.async_on_unload(entry.add_update_listener(_async_update_listener))

    return True


async def async_unload_entry(hass: HomeAssistant, entry: MyConfigEntry) -> bool:
    """卸载 ConfigEntry。"""
    # 卸载所有平台
    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
    if unload_ok:
        await entry.runtime_data.async_disconnect()
    return unload_ok


async def _async_update_listener(hass: HomeAssistant, entry: MyConfigEntry) -> None:
    """选项变更时重新加载。"""
    await hass.config_entries.async_reload(entry.entry_id)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

from homeassistant import config_entries
from homeassistant.core import callback
import voluptuous as vol

from .const import DOMAIN, CONF_HOST, CONF_API_KEY


class MyConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
    """My Integration 配置流。"""

    VERSION = 1

    async def async_step_user(self, user_input=None):
        """用户手动配置步骤。"""
        errors = {}

        if user_input is not None:
            # 验证连接
            try:
                await validate_input(self.hass, user_input)
            except CannotConnect:
                errors["base"] = "cannot_connect"
            except InvalidAuth:
                errors["base"] = "invalid_auth"
            else:
                # 设置唯一 ID 防止重复配置
                await self.async_set_unique_id(user_input[CONF_HOST])
                self._abort_if_unique_id_configured()
                return self.async_create_entry(
                    title=f"My Device ({user_input[CONF_HOST]})",
                    data=user_input,
                )

        return self.async_show_form(
            step_id="user",
            data_schema=vol.Schema({
                vol.Required(CONF_HOST): str,
                vol.Required(CONF_API_KEY): str,
            }),
            errors=errors,
        )

    async def async_step_zeroconf(self, discovery_info):
        """Zeroconf 自动发现步骤。"""
        host = discovery_info.host
        await self.async_set_unique_id(host)
        self._abort_if_unique_id_configured()

        # 展示确认表单
        return self.async_show_form(
            step_id="user",
            data_schema=vol.Schema({
                vol.Required(CONF_HOST, default=host): str,
                vol.Required(CONF_API_KEY): str,
            }),
        )

    @staticmethod
    @callback
    def async_get_options_flow(config_entry):
        """选项流。"""
        return MyOptionsFlow(config_entry)


class MyOptionsFlow(config_entries.OptionsFlowWithReload):
    """选项流。"""

    def __init__(self, config_entry):
        self.config_entry = config_entry

    async def async_step_init(self, user_input=None):
        if user_input is not None:
            return self.async_create_entry(data=user_input)

        return self.async_show_form(
            step_id="init",
            data_schema=vol.Schema({
                vol.Required("poll_interval", default=30): int,
            }),
        )