d21e7f1b3f
新增功能: - 配网管理模块 (provisioning.py): 支持设备扫描、配网、超时处理 - 配网配置步骤: UI 配置流程增加配网参数配置（Network Key, App Key 等） - 分组管理：支持 SIG 分组和 VENDOR 分组的加入/删除操作 - HA 服务调用：7 个配网和分组相关的服务 文件变更: - const.py: 添加配网相关常量（CONF_NETWORK_KEY, PROV_TIMEOUT 等） - config_flow.py: 增加 prov_config 配置步骤和 OptionsFlow 菜单 - provisioning.py: 新建配网管理器（ProvisioningManager 类） - coordinator.py: 集成配网管理器，添加配网状态管理方法 - services.py: 新建服务定义和注册 - services.yaml: HA 服务定义文件 - __init__.py: 集成服务注册和卸载 - PRD.md: 更新服务调用接口和配置参数文档 配网功能说明: - 首次使用需配置 Network Key, App Key, Network ID, IV Index - 配网超时时间：180 秒 - 组地址范围：0xC000 - 0xCFFF - 支持 SIG 标准分组和 VENDOR 自定义分组
342 lines
12 KiB
Python
342 lines
12 KiB
Python
"""SigMesh Gateway 配置流程."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
import voluptuous as vol
|
||
from homeassistant import config_entries
|
||
from homeassistant.core import callback
|
||
from homeassistant.data_entry_flow import FlowResult
|
||
from homeassistant.helpers import selector
|
||
|
||
from .const import (
|
||
CONF_APP_KEY,
|
||
CONF_BAUDRATE,
|
||
CONF_GROUP_ADDRESS,
|
||
CONF_IV_INDEX,
|
||
CONF_NETWORK_ID,
|
||
CONF_NETWORK_KEY,
|
||
CONF_SERIAL_DEVICE,
|
||
DEFAULT_APP_KEY,
|
||
DEFAULT_BAUDRATE,
|
||
DEFAULT_GROUP_ADDRESS_START,
|
||
DEFAULT_IV_INDEX,
|
||
DEFAULT_NAME,
|
||
DEFAULT_NETWORK_ID,
|
||
DEFAULT_NETWORK_KEY,
|
||
DOMAIN,
|
||
PROV_TIMEOUT,
|
||
)
|
||
|
||
|
||
class SigMeshGatewayConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||
"""SigMesh Gateway 配置流程."""
|
||
|
||
VERSION = 1
|
||
|
||
def __init__(self) -> None:
|
||
"""初始化配置流程."""
|
||
self._errors: dict[str, str] = {}
|
||
self._user_input: dict[str, Any] | None = None
|
||
self._prov_config: dict[str, Any] | None = None
|
||
|
||
async def async_step_user(
|
||
self, user_input: dict[str, Any] | None = None
|
||
) -> FlowResult:
|
||
"""处理用户配置步骤 - 串口配置."""
|
||
self._errors = {}
|
||
|
||
if user_input is not None:
|
||
# 验证串口连接
|
||
try:
|
||
import serial
|
||
|
||
test_serial = serial.Serial(
|
||
user_input[CONF_SERIAL_DEVICE],
|
||
baudrate=user_input.get(CONF_BAUDRATE, DEFAULT_BAUDRATE),
|
||
timeout=0.5,
|
||
)
|
||
test_serial.close()
|
||
self._user_input = user_input
|
||
# 进入配网配置步骤
|
||
return await self.async_step_prov_config()
|
||
except serial.SerialException as e:
|
||
self._errors[CONF_SERIAL_DEVICE] = f"无法打开串口：{e}"
|
||
except Exception as e:
|
||
self._errors["base"] = f"验证失败：{e}"
|
||
|
||
# 获取可用串口列表
|
||
try:
|
||
import serial.tools.list_ports
|
||
|
||
ports = serial.tools.list_ports.comports()
|
||
port_list = [
|
||
selector.SelectOptionDict(value=p.device, label=f"{p.device} - {p.description}")
|
||
for p in ports
|
||
]
|
||
except Exception:
|
||
port_list = [
|
||
selector.SelectOptionDict(value="/dev/ttyUSB0", label="/dev/ttyUSB0"),
|
||
selector.SelectOptionDict(value="/dev/ttyUSB1", label="/dev/ttyUSB1"),
|
||
selector.SelectOptionDict(value="/dev/ttyACM0", label="/dev/ttyACM0"),
|
||
]
|
||
|
||
return self.async_show_form(
|
||
step_id="user",
|
||
data_schema=vol.Schema(
|
||
{
|
||
vol.Required(CONF_SERIAL_DEVICE, default="/dev/ttyUSB0"): selector.SelectSelector(
|
||
selector.SelectSelectorConfig(options=port_list),
|
||
),
|
||
vol.Required(CONF_BAUDRATE, default=DEFAULT_BAUDRATE): vol.Coerce(int),
|
||
}
|
||
),
|
||
errors=self._errors,
|
||
)
|
||
|
||
async def async_step_prov_config(
|
||
self, user_input: dict[str, Any] | None = None
|
||
) -> FlowResult:
|
||
"""处理配网配置步骤 - 首次使用需要配置。
|
||
|
||
根据文档，USB dongle 首次使用需要设置配网参数：
|
||
- Network Key (16 字节网络密钥)
|
||
- App Key (16 字节应用密钥)
|
||
- Network ID (2 字节网络 ID)
|
||
- IV Index (4 字节)
|
||
"""
|
||
self._errors = {}
|
||
|
||
if user_input is not None:
|
||
self._prov_config = user_input
|
||
# 保存所有配置，完成配置流程
|
||
return self.async_create_entry(
|
||
title=self._user_input.get(CONF_SERIAL_DEVICE, DEFAULT_NAME),
|
||
data={
|
||
**self._user_input,
|
||
**user_input,
|
||
},
|
||
)
|
||
|
||
# 配网配置表单
|
||
return self.async_show_form(
|
||
step_id="prov_config",
|
||
data_schema=vol.Schema(
|
||
{
|
||
vol.Required(CONF_NETWORK_KEY, default=DEFAULT_NETWORK_KEY): str,
|
||
vol.Required(CONF_APP_KEY, default=DEFAULT_APP_KEY): str,
|
||
vol.Required(CONF_NETWORK_ID, default=DEFAULT_NETWORK_ID): str,
|
||
vol.Required(CONF_IV_INDEX, default=DEFAULT_IV_INDEX): int,
|
||
vol.Required(
|
||
CONF_GROUP_ADDRESS,
|
||
default=hex(DEFAULT_GROUP_ADDRESS_START),
|
||
): str,
|
||
}
|
||
),
|
||
description_placeholders={
|
||
"timeout": PROV_TIMEOUT,
|
||
},
|
||
errors=self._errors,
|
||
)
|
||
|
||
@callback
|
||
def async_get_options_flow(
|
||
self,
|
||
) -> SigMeshGatewayOptionsFlow:
|
||
"""获取选项流程."""
|
||
return SigMeshGatewayOptionsFlow()
|
||
|
||
|
||
class SigMeshGatewayOptionsFlow(config_entries.OptionsFlow):
|
||
"""SigMesh Gateway 选项流程."""
|
||
|
||
def __init__(self) -> None:
|
||
"""初始化选项流程."""
|
||
self._errors: dict[str, str] = {}
|
||
self._prov_action: str | None = None
|
||
|
||
async def async_step_init(
|
||
self, user_input: dict[str, Any] | None = None
|
||
) -> FlowResult:
|
||
"""管理选项."""
|
||
# 显示主菜单
|
||
return self.async_show_menu(
|
||
step_id="init",
|
||
menu_options=["poll_config", "prov_action", "group_config"],
|
||
)
|
||
|
||
async def async_step_poll_config(
|
||
self, user_input: dict[str, Any] | None = None
|
||
) -> FlowResult:
|
||
"""配置轮询间隔."""
|
||
if user_input is not None:
|
||
return self.async_create_entry(title="", data=user_input)
|
||
|
||
return self.async_show_form(
|
||
step_id="poll_config",
|
||
data_schema=vol.Schema(
|
||
{
|
||
vol.Required(
|
||
"poll_interval",
|
||
default=self.config_entry.options.get("poll_interval", 30),
|
||
): vol.Coerce(int),
|
||
}
|
||
),
|
||
)
|
||
|
||
async def async_step_prov_action(
|
||
self, user_input: dict[str, Any] | None = None
|
||
) -> FlowResult:
|
||
"""配网操作选择."""
|
||
if user_input is not None:
|
||
self._prov_action = user_input.get("action")
|
||
if self._prov_action == "start_scan":
|
||
return await self.async_step_start_scan()
|
||
elif self._prov_action == "stop_prov":
|
||
return await self.async_step_stop_prov()
|
||
elif self._prov_action == "bind_appkey":
|
||
return await self.async_step_bind_appkey()
|
||
|
||
return self.async_show_form(
|
||
step_id="prov_action",
|
||
data_schema=vol.Schema(
|
||
{
|
||
vol.Required("action"): selector.SelectSelector(
|
||
selector.SelectSelectorConfig(
|
||
options=[
|
||
selector.SelectOptionDict(value="start_scan", label="开始扫描设备"),
|
||
selector.SelectOptionDict(value="stop_prov", label="停止配网"),
|
||
selector.SelectOptionDict(value="bind_appkey", label="绑定 App Key"),
|
||
],
|
||
),
|
||
),
|
||
}
|
||
),
|
||
)
|
||
|
||
async def async_step_start_scan(
|
||
self, user_input: dict[str, Any] | None = None
|
||
) -> FlowResult:
|
||
"""开始扫描设备."""
|
||
if user_input is not None:
|
||
# TODO: 调用配网管理器开始扫描
|
||
return self.async_create_entry(title="", data={})
|
||
|
||
return self.async_show_form(
|
||
step_id="start_scan",
|
||
description_placeholders={"timeout": PROV_TIMEOUT},
|
||
data_schema=vol.Schema(
|
||
{
|
||
vol.Required("confirm"): bool,
|
||
}
|
||
),
|
||
)
|
||
|
||
async def async_step_stop_prov(
|
||
self, user_input: dict[str, Any] | None = None
|
||
) -> FlowResult:
|
||
"""停止配网."""
|
||
if user_input is not None:
|
||
# TODO: 调用配网管理器停止配网
|
||
return self.async_create_entry(title="", data={})
|
||
|
||
return self.async_show_form(
|
||
step_id="stop_prov",
|
||
data_schema=vol.Schema(
|
||
{
|
||
vol.Required("confirm"): bool,
|
||
}
|
||
),
|
||
)
|
||
|
||
async def async_step_bind_appkey(
|
||
self, user_input: dict[str, Any] | None = None
|
||
) -> FlowResult:
|
||
"""绑定 App Key."""
|
||
if user_input is not None:
|
||
# TODO: 调用配网管理器绑定 App Key
|
||
return self.async_create_entry(title="", data={})
|
||
|
||
return self.async_show_form(
|
||
step_id="bind_appkey",
|
||
data_schema=vol.Schema(
|
||
{
|
||
vol.Required("device_address"): str,
|
||
vol.Required("element_address", default=0): vol.Coerce(int),
|
||
}
|
||
),
|
||
)
|
||
|
||
async def async_step_group_config(
|
||
self, user_input: dict[str, Any] | None = None
|
||
) -> FlowResult:
|
||
"""分组配置."""
|
||
if user_input is not None:
|
||
action = user_input.get("group_action")
|
||
if action == "add_to_group":
|
||
return await self.async_step_add_to_group()
|
||
elif action == "remove_from_group":
|
||
return await self.async_step_remove_from_group()
|
||
|
||
return self.async_show_form(
|
||
step_id="group_config",
|
||
data_schema=vol.Schema(
|
||
{
|
||
vol.Required("group_action"): selector.SelectSelector(
|
||
selector.SelectSelectorConfig(
|
||
options=[
|
||
selector.SelectOptionDict(value="add_to_group", label="添加到组"),
|
||
selector.SelectOptionDict(value="remove_from_group", label="从组移除"),
|
||
],
|
||
),
|
||
),
|
||
}
|
||
),
|
||
)
|
||
|
||
async def async_step_add_to_group(
|
||
self, user_input: dict[str, Any] | None = None
|
||
) -> FlowResult:
|
||
"""添加设备到组."""
|
||
if user_input is not None:
|
||
# TODO: 调用配网管理器添加设备到组
|
||
return self.async_create_entry(title="", data={})
|
||
|
||
return self.async_show_form(
|
||
step_id="add_to_group",
|
||
data_schema=vol.Schema(
|
||
{
|
||
vol.Required("target_address"): str,
|
||
vol.Required("element_address", default=0): vol.Coerce(int),
|
||
vol.Required("group_address", default=hex(DEFAULT_GROUP_ADDRESS_START)): str,
|
||
vol.Required("model_id", default=4352): vol.Coerce(int), # 默认 0x1100
|
||
vol.Required("is_sig", default=True): bool,
|
||
}
|
||
),
|
||
description_placeholders={
|
||
"default_group": hex(DEFAULT_GROUP_ADDRESS_START),
|
||
},
|
||
)
|
||
|
||
async def async_step_remove_from_group(
|
||
self, user_input: dict[str, Any] | None = None
|
||
) -> FlowResult:
|
||
"""从组中移除设备."""
|
||
if user_input is not None:
|
||
# TODO: 调用配网管理器移除设备
|
||
return self.async_create_entry(title="", data={})
|
||
|
||
return self.async_show_form(
|
||
step_id="remove_from_group",
|
||
data_schema=vol.Schema(
|
||
{
|
||
vol.Required("target_address"): str,
|
||
vol.Required("element_address", default=0): vol.Coerce(int),
|
||
vol.Required("group_address", default=hex(DEFAULT_GROUP_ADDRESS_START)): str,
|
||
vol.Required("model_id", default=4352): vol.Coerce(int),
|
||
vol.Required("is_sig", default=True): bool,
|
||
}
|
||
),
|
||
)
|