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
 9
10
11

components/<domain>/
├── manifest.json       # 集成清单（必需）
├── __init__.py         # 集成入口
├── config_flow.py      # 配置流（可选，支持 UI 配置时需要）
├── light.py            # 向 light 域提供实体（可选平台文件）
├── sensor.py           # 向 sensor 域提供实体（可选平台文件）
├── strings.json        # Core 集成英文翻译源；自定义集成通常使用 translations/en.json
├── icons.json          # 前端图标资源（可选）
├── translations/       # 自定义集成运行时翻译目录（可选）
│   └── en.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.9.0"],
  "iot_class": "local_push",
  "zeroconf": ["_hue._tcp.local."],
  "codeowners": ["@marcelveldt"]
}

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

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.async_initialize()             # 初始化配置对象的事件循环侧状态
        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
 2
 3
 4
 5
 6
 7
 8
 9
10

# 普通任务会参与 setup/unload 生命周期管理
entry.async_create_task(hass, do_one_shot_work(), name="refresh device")

# 常驻任务在 entry unload 时自动取消，也不会阻塞 HA 完成启动
entry.async_create_background_task(
    hass, run_push_listener(), name="device push listener"
)

# 真正阻塞的同步调用才进入 executor
result = await hass.async_add_executor_job(sync_library.read, address)

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

await self.async_set_unique_id(device_serial_or_mac)
self._abort_if_unique_id_configured()

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
17

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) -> ComponentProtocol  # 获取 __init__.py 模块
    async def async_get_platform(self, platform) -> ModuleType  # 获取平台模块
    async def async_get_platforms(self, platforms) -> dict[str, ModuleType]
    async def resolve_dependencies(self) -> set[str] | None

1
2
3
4
5
6
7

@dataclass(frozen=True, kw_only=True)
class ConfigSubentry:
    data: MappingProxyType[str, Any]
    subentry_id: str        # Core 分配的 ULID
    subentry_type: str
    title: str
    unique_id: str | None

1

async_add_entities(entities, config_subentry_id=subentry.subentry_id)

1
2
3

device = device_registry.async_get_device_by_identifier(
    (DOMAIN, serial_number), config_entry.entry_id
)

1
2
3
4
5
6
7

from homeassistant.helpers.device_registry import ChildDeviceInfo

self._attr_device_info = ChildDeviceInfo(
    identifiers={(DOMAIN, channel_id)},
    name="Channel 1",
    parent_device_id=parent_device_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

__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_CORE_CONFIG_UPDATE
                    │     ├── fire EVENT_HOMEASSISTANT_START
                    │     ├── wait startup tasks and startup jobs (with timeout)
                    │     ├── CoreState → RUNNING
                    │     ├── fire EVENT_CORE_CONFIG_UPDATE
                    │     └── 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

class ConfigEntries:
    def __init__(self, hass, hass_config):
        self.flow = ConfigEntriesFlowManager(hass, self, hass_config)
        self.options = OptionsFlowManager(hass)
        self.subentries = ConfigSubentryFlowManager(hass)

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

流程处理器（一次会话一个实例）
FlowHandler
  ├── ConfigEntryBaseFlow
  │     ├── ConfigFlow
  │     └── OptionsFlow
  ├── ConfigSubentryFlow
  ├── RepairsFlow
  └── LoginFlow / MFA setup flow

流程管理器（通常由 HA 长期持有，一个实例管理多次会话）
FlowManager
  ├── ConfigEntriesFlowManager
  ├── OptionsFlowManager
  ├── ConfigSubentryFlowManager
  ├── RepairsFlowManager
  ├── AuthManagerFlowManager
  └── MfaFlowManager

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

manager.async_init(handler, context=..., data=...)
  ├── manager.async_create_flow(handler, context, data)
  │     └── 专用 Manager 返回具体 FlowHandler
  ├── 注入 flow.hass / handler / flow_id / context / init_data
  ├── _async_add_flow_progress(flow)
  │     └── 写入 _progress、handler 和 init_data 类型索引
  └── _async_handle_step(flow, flow.init_step, data)
        ├── 检查 async_step_<step_id> 是否存在
        ├── await async_step_<step_id>(data)
        ├── 将 AbortFlow 异常规范化为 ABORT result
        ├── 管理 preview/progress task 和 result 类型
        ├── 未完成：保存 flow.cur_step，等待下一次输入
        └── 已完成：调用专用 manager.async_finish_flow(flow, result)
              ├── 专用副作用：创建条目/更新 options/增加 subentry/完成登录等
              ├── 若 Manager 返回 FORM：保存新的 cur_step，流程继续
              └── 否则 _async_remove_flow_progress(flow_id)
                    ├── 从全部索引删除
                    ├── 取消 progress task
                    └── 调用 flow.async_remove() 做清理

1
2
3
4
5

manager.async_configure(flow_id, user_input)
  ├── 从 _progress 取得 flow 和 flow.cur_step
  ├── 如果有 data_schema，先用 voluptuous 校验 user_input
  ├── MENU：转到 user_input["next_step_id"]
  └── 其他类型：再次调用当前 async_step_<step_id>(user_input)

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

subentries.async_init(
    (entry_id, subentry_type),
    context={"source": "user"},
)
  ├── 找到父 ConfigEntry
  ├── 加载父 entry.domain 的 ConfigFlow handler
  ├── async_get_supported_subentry_types(entry)
  ├── 实例化指定 subentry_type 的 ConfigSubentryFlow
  ├── 公共 FlowManager 状态机推进 async_step_user()
  └── CREATE_ENTRY
        └── ConfigSubentryFlowManager.async_finish_flow()
              └── ConfigEntries.async_add_subentry(parent_entry, new_subentry)

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

POST index {handler, ...}
  → FlowManagerIndexView._post_impl()
  → manager.async_init(handler, context=路由构造的上下文)
  → 返回第一个 FlowResult

POST resource/{flow_id} {user_input...}
  → FlowManagerResourceView.post()
  → manager.async_configure(flow_id, user_input)
  → 返回下一个或最终 FlowResult

GET resource/{flow_id}
  → manager.async_configure(flow_id)
  → 推进/轮询 external 或 progress step

DELETE resource/{flow_id}
  → manager.async_abort(flow_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

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

收到 user_input / discovery_info
  ├── 如果没有输入：返回 async_show_form()
  ├── 校验 schema 和连接能力
  ├── 获取稳定身份：serial / MAC / account id
  ├── await async_set_unique_id(stable_id)
  ├── _abort_if_unique_id_configured()
  └── async_create_entry(...) 或进入下一个 async_step_xxx

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

entry = ConfigEntry(
    data=result["data"],
    domain=result["handler"],
    source=flow.context["source"],
    title=result["title"],
    unique_id=flow.unique_id,
    version=result["version"],
    minor_version=result["minor_version"],
    options=result["options"],
    subentries_data=result["subentries"],
    discovery_keys=discovery_keys,
)
await self.config_entries.async_add(entry)

1
2

await self.async_set_unique_id(format_mac(discovery_info.address))
self._abort_if_unique_id_configured()

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

BLE advertisement
  ↓
bluetooth.manager._discover_service_info()
  ├── IntegrationMatcher.match_domains(service_info)
  │     └── 若同 address 的 name / manufacturer data / service data / service UUIDs 等字段此前都见过，返回空集合
  ↓
helpers.discovery_flow.async_create_flow(..., source="bluetooth", data=service_info)
  └── 若已有相同 domain/source/data 的发现 flow 正在进行，则不再 async_init
  ↓
ConfigEntriesFlowManager.async_init(...)
  └── 创建新的 ConfigFlow 实例，init_step="bluetooth"
        ↓
      async_step_bluetooth(discovery_info)

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

async def async_step_bluetooth(self, discovery_info: BluetoothServiceInfoBleak):
    await self.async_set_unique_id(format_mac(discovery_info.address))
    self._abort_if_unique_id_configured()

    self._discovered = {
        CONF_ADDRESS: discovery_info.address,
        CONF_NAME: discovery_info.name,
    }

    # 可选：探测设备能力、检查是否支持、读取加密要求
    if not await device_is_supported(discovery_info):
        return self.async_abort(reason="not_supported")

    self.context["title_placeholders"] = {"name": discovery_info.name}
    self._set_confirm_only()
    return self.async_show_form(step_id="bluetooth_confirm")

async def async_step_bluetooth_confirm(self, user_input=None):
    if user_input is not None:
        return self.async_create_entry(
            title=self._discovered[CONF_NAME],
            data={CONF_ADDRESS: self._discovered[CONF_ADDRESS]},
        )
    return self.async_show_form(step_id="bluetooth_confirm")

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

ESPHome proxy A ─┐
                  ├─ BluetoothServiceInfoBleak(address=外设 MAC, source=代理 MAC)
ESPHome proxy B ─┘
                             ↓
BluetoothManager.scanner_adv_received()
  ├─ 每个 scanner 仍保留自己看到的 BLEDevice/AdvertisementData
  ├─ _all_history[address]         # 全部路径中当前首选广播
  ├─ _connectable_history[address] # 可连接路径中当前首选广播
  └─ 只将接管成功且 payload 有变化的规范广播送入 HA discovery/callback
                             ↓
HomeAssistantBluetoothManager._discover_service_info()
  ├─ IntegrationMatcher 按 address 记录已见字段
  └─ DiscoveryKey(domain="bluetooth", key=address, version=1)
                             ↓
ConfigFlow.async_step_bluetooth()
  └─ async_set_unique_id(稳定设备 ID) + _abort_if_unique_id_configured()
                             ↓
Device Registry
  └─ connections={("bluetooth", address)}

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

integration: establish_connection(..., BLEDevice(address), ...)
                         ↓
HaBleakClientWrapper.connect()
                         ↓
BluetoothManager.async_scanner_devices_by_address(address, connectable=True)
  ├─ local adapter:  BLEDevice + AdvertisementData + local scanner
  ├─ ESPHome proxy A: BLEDevice + AdvertisementData + ESPHome scanner
  └─ ESPHome proxy B: BLEDevice + AdvertisementData + ESPHome scanner
                         ↓
BluetoothScannerDevice.score_connection_path()
                         ↓
选第一个可用 backend：本地 Bleak backend 或某个代理的 connector.client

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

被动广播：
BLE scan result
  -> ESPHome BLE hub
  -> raw advertisement 批次（默认每批最多 16 条）
  -> Native API
  -> HA Bluetooth manager

GATT：
HA 发起 connect request
  -> ESPHome BluetoothProxy 分配一个 connection slot
  -> BluetoothConnection + GATT backend
  -> 建连、服务发现、读写、notify、配对
  -> Native API 返回结果

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

class ConfigEntries:
    def __init__(self, hass, hass_config):
        self.flow = ConfigEntriesFlowManager(hass, self, hass_config)
        self.options = OptionsFlowManager(hass)

class ConfigFlow(ConfigEntryBaseFlow):
    @staticmethod
    @callback
    def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow:
        raise UnknownHandler

class OptionsFlow(ConfigEntryBaseFlow):
    @property
    def _config_entry_id(self) -> str: ...

    @property
    def config_entry(self) -> ConfigEntry: ...

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

前端
  │  hass.config_entries.options.async_init(entry.entry_id,
  │      context={"source": "user"})
  ↓
OptionsFlowManager.async_create_flow(entry_id)
  ├── 根据 entry_id 找到已有 ConfigEntry
  ├── 加载 entry.domain 的 config_flow.py
  └── handler.async_get_options_flow(entry) → OptionsFlow 实例
  ↓
FlowManager 设置 flow.hass / flow.handler / flow_id
  └── 默认进入 async_step_init(None)
  ↓
前端提交表单
  └── options.async_configure(flow_id, user_input)
        ├── 先用 data_schema 校验输入
        └── 反射调用 async_step_<当前 step>(user_input)
  ↓
async_create_entry(data=new_options)
  ↓
OptionsFlowManager.async_finish_flow()
  ├── ABORT：结束流程，不修改条目
  └── CREATE_ENTRY：async_update_entry(entry, options=new_options)
                    └── 持久化、触发 update listeners
                        └── OptionsFlowWithReload 时安排 reload

 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

from typing import Any

import voluptuous as vol

from homeassistant.config_entries import (
    ConfigEntry,
    ConfigFlow,
    ConfigFlowResult,
    OptionsFlow,
    OptionsFlowWithReload,
)
from homeassistant.core import callback

from .const import DOMAIN


class MyConfigFlow(ConfigFlow, domain=DOMAIN):
    """创建 My Integration 的 ConfigEntry。"""

    @staticmethod
    @callback
    def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow:
        """返回该条目的选项流。"""
        # 不要在这里给 flow.config_entry 赋值；Manager 会在初始化后注入关联关系。
        return MyOptionsFlow()


class MyOptionsFlow(OptionsFlowWithReload):
    """编辑不会改变设备身份的运行选项。"""

    async def async_step_init(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """显示并保存选项。"""
        if user_input is not None:
            # data 会被 OptionsFlowManager 写入 ConfigEntry.options。
            return self.async_create_entry(title="", data=user_input)

        options = self.config_entry.options
        return self.async_show_form(
            step_id="init",
            data_schema=vol.Schema(
                {
                    vol.Required(
                        "poll_interval",
                        default=options.get("poll_interval", 30),
                    ): vol.All(vol.Coerce(int), vol.Range(min=5)),
                    vol.Optional(
                        "enable_extra_entities",
                        default=options.get("enable_extra_entities", True),
                    ): bool,
                }
            ),
        )

 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

class EntityDescription(metaclass=FrozenOrThawed, frozen_or_thawed=True):
    """A class that describes Home Assistant entities."""

    key: str
    device_class: str | None = None
    entity_category: EntityCategory | None = None
    entity_registry_enabled_default: bool = True
    entity_registry_visible_default: bool = True
    force_update: bool = False
    icon: str | None = None
    has_entity_name: bool = False
    name: str | UndefinedType | None = UNDEFINED
    translation_key: str | None = None
    translation_placeholders: Mapping[str, str] | None = None
    unit_of_measurement: str | None = 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

from homeassistant.components.sensor import (
    SensorDeviceClass,
    SensorEntity,
    SensorEntityDescription,
    SensorStateClass,
)

SENSOR_DESCRIPTION = SensorEntityDescription(
    key="temperature",
    translation_key="temperature",
    translation_placeholders={"room": "Living room"},
    device_class=SensorDeviceClass.TEMPERATURE,
    native_unit_of_measurement="°C",
    state_class=SensorStateClass.MEASUREMENT,
)


class MySensor(SensorEntity):
    _attr_has_entity_name = True

    def __init__(self, description: SensorEntityDescription) -> None:
        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

component.<integration_domain>.entity.<platform_domain>.<translation_key>.name

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

{
  "entity": {
    "sensor": {
      "temperature": {
        "name": "{room} temperature",
        "state_attributes": {
          "signal": {
            "name": "Signal strength"
          }
        }
      }
    }
  }
}

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

# 写在描述对象上，适合多个同类实体复用同一实体类
SensorEntityDescription(
    key="battery",
    translation_key="battery",
    translation_placeholders={"device_name": "Door lock"},
)

# 写在实体类属性上，适合运行时动态决定
class MyBatterySensor(SensorEntity):
    _attr_has_entity_name = True
    _attr_translation_key = "battery"
    _attr_translation_placeholders = {"device_name": "Door lock"}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

class SensorEntityDescription(EntityDescription, frozen_or_thawed=True):
    """A class that describes sensor entities."""

    device_class: SensorDeviceClass | None = None
    last_reset: datetime | None = None
    native_unit_of_measurement: str | None = None
    options: list[str] | None = None
    state_class: SensorStateClass | None = None
    suggested_display_precision: int | None = None
    suggested_unit_of_measurement: str | None = None
    unit_of_measurement: None = None  # Type override, use native_unit_of_measurement