04e942992b
问题：集成配置选项面板中字体不显示 修复内容: - 移除 vol.Coerce 包装器，直接使用 int/bool 类型 - 添加错误处理和显示 - 修复 self._hass 为 self.hass (OptionsFlow 中通过属性访问) - 添加实际操作调用（扫描、配网、绑定、分组） - 添加错误时返回上一步的逻辑 修改的方法: - async_step_poll_config: 简化类型定义 - async_step_prov_action: 添加错误处理 - async_step_start_scan: 添加实际调用和错误处理 - async_step_stop_prov: 添加实际调用和错误处理 - async_step_bind_appkey: 添加实际调用和错误处理 - async_step_add_to_group: 添加实际调用和错误处理 - async_step_remove_from_group: 添加实际调用和错误处理
412 lines
15 KiB
Python
412 lines
15 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),
|
||
): int,
|
||
}
|
||
),
|
||
)
|
||
|
||
async def async_step_prov_action(
|
||
self, user_input: dict[str, Any] | None = None
|
||
) -> FlowResult:
|
||
"""配网操作选择."""
|
||
errors = {}
|
||
|
||
if user_input is not None:
|
||
action = user_input.get("action")
|
||
if action == "start_scan":
|
||
return await self.async_step_start_scan()
|
||
elif action == "stop_prov":
|
||
return await self.async_step_stop_prov()
|
||
elif action == "bind_appkey":
|
||
return await self.async_step_bind_appkey()
|
||
errors["base"] = "无效的操作"
|
||
|
||
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"),
|
||
],
|
||
),
|
||
),
|
||
}
|
||
),
|
||
errors=errors,
|
||
)
|
||
|
||
async def async_step_start_scan(
|
||
self, user_input: dict[str, Any] | None = None
|
||
) -> FlowResult:
|
||
"""开始扫描设备."""
|
||
if user_input is not None:
|
||
# 调用配网管理器开始扫描
|
||
try:
|
||
for coordinator in self.hass.data.get(DOMAIN, {}).values():
|
||
if hasattr(coordinator, 'start_scanning'):
|
||
await coordinator.start_scanning()
|
||
except Exception as e:
|
||
self._errors["base"] = str(e)
|
||
return await self.async_step_prov_action()
|
||
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,
|
||
}
|
||
),
|
||
errors=self._errors,
|
||
)
|
||
|
||
async def async_step_stop_prov(
|
||
self, user_input: dict[str, Any] | None = None
|
||
) -> FlowResult:
|
||
"""停止配网."""
|
||
if user_input is not None:
|
||
# 调用配网管理器停止配网
|
||
try:
|
||
for coordinator in self.hass.data.get(DOMAIN, {}).values():
|
||
if hasattr(coordinator, 'stop_provisioning'):
|
||
await coordinator.stop_provisioning()
|
||
except Exception as e:
|
||
self._errors["base"] = str(e)
|
||
return await self.async_step_prov_action()
|
||
return self.async_create_entry(title="", data={})
|
||
|
||
return self.async_show_form(
|
||
step_id="stop_prov",
|
||
data_schema=vol.Schema(
|
||
{
|
||
vol.Required("confirm"): bool,
|
||
}
|
||
),
|
||
errors=self._errors,
|
||
)
|
||
|
||
async def async_step_bind_appkey(
|
||
self, user_input: dict[str, Any] | None = None
|
||
) -> FlowResult:
|
||
"""绑定 App Key."""
|
||
if user_input is not None:
|
||
# 调用配网管理器绑定 App Key
|
||
try:
|
||
device_address = user_input.get("device_address")
|
||
element_address = user_input.get("element_address", 0)
|
||
for coordinator in self.hass.data.get(DOMAIN, {}).values():
|
||
if hasattr(coordinator, 'bind_app_key'):
|
||
await coordinator.bind_app_key(device_address, element_address)
|
||
except Exception as e:
|
||
self._errors["base"] = str(e)
|
||
return await self.async_step_prov_action()
|
||
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): int,
|
||
}
|
||
),
|
||
errors=self._errors,
|
||
)
|
||
|
||
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:
|
||
# 调用配网管理器添加设备到组
|
||
try:
|
||
target_address = user_input.get("target_address")
|
||
element_address = user_input.get("element_address", 0)
|
||
group_address = user_input.get("group_address")
|
||
model_id = user_input.get("model_id", 4352)
|
||
is_sig = user_input.get("is_sig", True)
|
||
|
||
# 解析组地址
|
||
if isinstance(group_address, str):
|
||
group_address = int(group_address, 16)
|
||
|
||
for coordinator in self.hass.data.get(DOMAIN, {}).values():
|
||
if hasattr(coordinator, 'add_device_to_group'):
|
||
await coordinator.add_device_to_group(
|
||
target_address, element_address, group_address, model_id, is_sig
|
||
)
|
||
except Exception as e:
|
||
self._errors["base"] = str(e)
|
||
return await self.async_step_group_config()
|
||
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): int,
|
||
vol.Required("group_address", default=hex(DEFAULT_GROUP_ADDRESS_START)): str,
|
||
vol.Required("model_id", default=4352): int,
|
||
vol.Required("is_sig", default=True): bool,
|
||
}
|
||
),
|
||
description_placeholders={
|
||
"default_group": hex(DEFAULT_GROUP_ADDRESS_START),
|
||
},
|
||
errors=self._errors,
|
||
)
|
||
|
||
async def async_step_remove_from_group(
|
||
self, user_input: dict[str, Any] | None = None
|
||
) -> FlowResult:
|
||
"""从组中移除设备."""
|
||
if user_input is not None:
|
||
# 调用配网管理器移除设备
|
||
try:
|
||
target_address = user_input.get("target_address")
|
||
element_address = user_input.get("element_address", 0)
|
||
group_address = user_input.get("group_address")
|
||
model_id = user_input.get("model_id", 4352)
|
||
is_sig = user_input.get("is_sig", True)
|
||
|
||
# 解析组地址
|
||
if isinstance(group_address, str):
|
||
group_address = int(group_address, 16)
|
||
|
||
for coordinator in self.hass.data.get(DOMAIN, {}).values():
|
||
if hasattr(coordinator, 'remove_device_from_group'):
|
||
await coordinator.remove_device_from_group(
|
||
target_address, element_address, group_address, model_id, is_sig
|
||
)
|
||
except Exception as e:
|
||
self._errors["base"] = str(e)
|
||
return await self.async_step_group_config()
|
||
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): int,
|
||
vol.Required("group_address", default=hex(DEFAULT_GROUP_ADDRESS_START)): str,
|
||
vol.Required("model_id", default=4352): int,
|
||
vol.Required("is_sig", default=True): bool,
|
||
}
|
||
),
|
||
errors=self._errors,
|
||
)
|