4c3eb62dfb
新增 Web UI 组件: - web_ui.py: RESTful API 端点（状态、扫描、配网、分组、设备） - sigmesh-gateway-panel.js: Lovelace Dashboard 自定义卡片 - 设备扫描和发现 - 配网操作（开始/停止/绑定 App Key） - 分组管理（添加/移除） - 实时状态监控 配置更新: - __init__.py: 集成 Web UI 和服务注册 - const.py: 添加服务常量定义 - services.py: 保留服务调用用于向后兼容 - README.md: 添加 Web UI 配置说明 - docs/UI 使用指南.md: 详细的 UI 使用文档 使用方式: 1. 配置 frontend.extra_module_url 加载 JS 面板 2. 在 Lovelace Dashboard 添加 custom:sigmesh-gateway-panel 卡片 3. 通过 UI 完成所有配网和分组操作 API 端点: - GET /api/sigmesh_gateway/status - 获取配网状态 - POST /api/sigmesh_gateway/scan - 开始扫描 - POST /api/sigmesh_gateway/provisioning - 配网操作 - POST /api/sigmesh_gateway/group - 分组管理 - GET /api/sigmesh_gateway/devices - 设备列表
295 lines
10 KiB
Python
295 lines
10 KiB
Python
"""SigMesh Gateway Web UI 面板."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import Any
|
||
|
||
import voluptuous as vol
|
||
from aiohttp import web
|
||
from homeassistant.components.http import HomeAssistantView
|
||
from homeassistant.core import HomeAssistant
|
||
from homeassistant.helpers import config_validation as cv
|
||
|
||
from .const import DOMAIN
|
||
from .coordinator import SigMeshGatewayCoordinator
|
||
|
||
_LOGGER = logging.getLogger(__name__)
|
||
|
||
|
||
def setup_web_ui(hass: HomeAssistant, coordinators: dict[str, SigMeshGatewayCoordinator]) -> None:
|
||
"""设置 Web UI 路由。
|
||
|
||
Args:
|
||
hass: Home Assistant 实例
|
||
coordinators: 协调器字典
|
||
"""
|
||
|
||
# 注册 Web 视图
|
||
hass.http.register_view(SigMeshGatewayStatusView(coordinators))
|
||
hass.http.register_view(SigMeshGatewayScanView(coordinators))
|
||
hass.http.register_view(SigMeshGatewayProvView(coordinators))
|
||
hass.http.register_view(SigMeshGatewayGroupView(coordinators))
|
||
hass.http.register_view(SigMeshGatewayDevicesView(coordinators))
|
||
|
||
|
||
class SigMeshGatewayStatusView(HomeAssistantView):
|
||
"""获取配网状态视图."""
|
||
|
||
requires_auth = True
|
||
url = "/api/sigmesh_gateway/status"
|
||
name = "api:sigmesh_gateway:status"
|
||
|
||
def __init__(self, coordinators: dict[str, SigMeshGatewayCoordinator]) -> None:
|
||
self.coordinators = coordinators
|
||
|
||
async def get(self, request: web.Request) -> web.Response:
|
||
"""获取配网状态。
|
||
|
||
返回示例:
|
||
{
|
||
"state": "idle",
|
||
"devices": [
|
||
{"mac": "AA:BB:CC:DD:EE:FF", "elements": 2, "address": "0001"}
|
||
],
|
||
"groups": [
|
||
{"address": "C001", "devices": ["0001"], "model_id": 4352}
|
||
]
|
||
}
|
||
"""
|
||
result = {
|
||
"state": "idle",
|
||
"devices": [],
|
||
"groups": [],
|
||
}
|
||
|
||
for coordinator in self.coordinators.values():
|
||
prov_devices = coordinator.get_prov_devices()
|
||
result["devices"].extend([
|
||
{
|
||
"mac": dev.mac_address,
|
||
"elements": dev.element_count,
|
||
"address": hex(dev.unicast_address) if dev.unicast_address else None,
|
||
}
|
||
for dev in prov_devices.values()
|
||
])
|
||
|
||
# 获取组配置
|
||
for group_addr, configs in coordinator.prov_manager._group_configs.items():
|
||
result["groups"].extend([
|
||
{
|
||
"address": hex(group_addr),
|
||
"element_address": hex(cfg.element_address),
|
||
"model_id": hex(cfg.model_id),
|
||
}
|
||
for cfg in configs
|
||
])
|
||
|
||
if coordinator.prov_state.value != "idle":
|
||
result["state"] = coordinator.prov_state.value
|
||
|
||
return self.json(result)
|
||
|
||
|
||
class SigMeshGatewayScanView(HomeAssistantView):
|
||
"""开始扫描设备视图."""
|
||
|
||
requires_auth = True
|
||
url = "/api/sigmesh_gateway/scan"
|
||
name = "api:sigmesh_gateway:scan"
|
||
|
||
def __init__(self, coordinators: dict[str, SigMeshGatewayCoordinator]) -> None:
|
||
self.coordinators = coordinators
|
||
|
||
async def post(self, request: web.Request) -> web.Response:
|
||
"""开始扫描设备。
|
||
|
||
请求体: {} 或 {"timeout": 60}
|
||
"""
|
||
try:
|
||
data = await request.json() if request.can_read_body else {}
|
||
timeout = data.get("timeout", 60)
|
||
|
||
for coordinator in self.coordinators.values():
|
||
await coordinator.start_scanning()
|
||
|
||
return self.json({
|
||
"success": True,
|
||
"message": f"开始扫描设备，超时时间：{timeout}秒",
|
||
})
|
||
except Exception as e:
|
||
_LOGGER.error("扫描设备失败：%s", e)
|
||
return self.json({"success": False, "error": str(e)}, status=400)
|
||
|
||
|
||
class SigMeshGatewayProvView(HomeAssistantView):
|
||
"""配网操作视图."""
|
||
|
||
requires_auth = True
|
||
url = "/api/sigmesh_gateway/provisioning"
|
||
name = "api:sigmesh_gateway:provisioning"
|
||
|
||
def __init__(self, coordinators: dict[str, SigMeshGatewayCoordinator]) -> None:
|
||
self.coordinators = coordinators
|
||
|
||
async def post(self, request: web.Request) -> web.Response:
|
||
"""开始配网设备。
|
||
|
||
请求体:
|
||
{
|
||
"action": "start", // start, stop, bind
|
||
"device_address": "001A",
|
||
"element_address": 0, // 可选，bind 时需要
|
||
"timeout": 180 // 可选，超时时间（秒）
|
||
}
|
||
"""
|
||
try:
|
||
data = await request.json() if request.can_read_body else {}
|
||
action = data.get("action")
|
||
device_address = data.get("device_address")
|
||
element_address = data.get("element_address", 0)
|
||
timeout = data.get("timeout", 180)
|
||
|
||
if action == "start":
|
||
if not device_address:
|
||
return self.json({"success": False, "error": "需要设备地址"}, status=400)
|
||
for coordinator in self.coordinators.values():
|
||
await coordinator.start_provisioning(device_address)
|
||
return self.json({
|
||
"success": True,
|
||
"message": f"开始配网设备 {device_address}，超时时间：{timeout}秒",
|
||
})
|
||
elif action == "stop":
|
||
for coordinator in self.coordinators.values():
|
||
await coordinator.stop_provisioning()
|
||
return self.json({"success": True, "message": "停止配网"})
|
||
elif action == "bind":
|
||
if not device_address:
|
||
return self.json({"success": False, "error": "需要设备地址"}, status=400)
|
||
for coordinator in self.coordinators.values():
|
||
await coordinator.bind_app_key(device_address, element_address)
|
||
return self.json({
|
||
"success": True,
|
||
"message": f"绑定 App Key: {device_address}",
|
||
})
|
||
else:
|
||
return self.json({"success": False, "error": "未知操作"}, status=400)
|
||
|
||
except Exception as e:
|
||
_LOGGER.error("配网操作失败：%s", e)
|
||
return self.json({"success": False, "error": str(e)}, status=400)
|
||
|
||
|
||
class SigMeshGatewayGroupView(HomeAssistantView):
|
||
"""分组管理视图."""
|
||
|
||
requires_auth = True
|
||
url = "/api/sigmesh_gateway/group"
|
||
name = "api:sigmesh_gateway:group"
|
||
|
||
def __init__(self, coordinators: dict[str, SigMeshGatewayCoordinator]) -> None:
|
||
self.coordinators = coordinators
|
||
|
||
async def post(self, request: web.Request) -> web.Response:
|
||
"""管理设备分组。
|
||
|
||
请求体:
|
||
{
|
||
"action": "add", // add, remove
|
||
"target_address": "001A",
|
||
"element_address": 0,
|
||
"group_address": "C001",
|
||
"model_id": 4352,
|
||
"is_sig": true
|
||
}
|
||
"""
|
||
try:
|
||
data = await request.json() if request.can_read_body else {}
|
||
action = data.get("action")
|
||
target_address = data.get("target_address")
|
||
element_address = data.get("element_address", 0)
|
||
group_address = data.get("group_address")
|
||
model_id = data.get("model_id", 4352)
|
||
is_sig = data.get("is_sig", True)
|
||
|
||
# 解析组地址
|
||
if isinstance(group_address, str):
|
||
group_address = int(group_address, 16)
|
||
|
||
if action == "add":
|
||
for coordinator in self.coordinators.values():
|
||
await coordinator.add_device_to_group(
|
||
target_address, element_address, group_address, model_id, is_sig
|
||
)
|
||
return self.json({
|
||
"success": True,
|
||
"message": f"添加设备 {target_address} 到组 {hex(group_address)}",
|
||
})
|
||
elif action == "remove":
|
||
for coordinator in self.coordinators.values():
|
||
await coordinator.remove_device_from_group(
|
||
target_address, element_address, group_address, model_id, is_sig
|
||
)
|
||
return self.json({
|
||
"success": True,
|
||
"message": f"从组 {hex(group_address)} 移除设备 {target_address}",
|
||
})
|
||
else:
|
||
return self.json({"success": False, "error": "未知操作"}, status=400)
|
||
|
||
except Exception as e:
|
||
_LOGGER.error("分组操作失败：%s", e)
|
||
return self.json({"success": False, "error": str(e)}, status=400)
|
||
|
||
|
||
class SigMeshGatewayDevicesView(HomeAssistantView):
|
||
"""获取设备列表视图."""
|
||
|
||
requires_auth = True
|
||
url = "/api/sigmesh_gateway/devices"
|
||
name = "api:sigmesh_gateway:devices"
|
||
|
||
def __init__(self, coordinators: dict[str, SigMeshGatewayCoordinator]) -> None:
|
||
self.coordinators = coordinators
|
||
|
||
async def get(self, request: web.Request) -> web.Response:
|
||
"""获取所有设备列表。
|
||
|
||
返回示例:
|
||
{
|
||
"prov_devices": [...],
|
||
"mesh_devices": [...]
|
||
}
|
||
"""
|
||
prov_devices = []
|
||
mesh_devices = []
|
||
|
||
for coordinator in self.coordinators.values():
|
||
# 获取配网设备
|
||
prov_devices.extend([
|
||
{
|
||
"mac": dev.mac_address,
|
||
"elements": dev.element_count,
|
||
"unicast_address": dev.unicast_address,
|
||
"joined_at": dev.joined_at.isoformat() if dev.joined_at else None,
|
||
}
|
||
for dev in coordinator.get_prov_devices().values()
|
||
])
|
||
|
||
# 获取 Mesh 设备状态
|
||
if coordinator.data:
|
||
mesh_devices.extend([
|
||
{
|
||
"mac": mac,
|
||
"model_id": hex(device.model_id) if device.model_id else None,
|
||
"states": device.states,
|
||
"last_update": device.last_update,
|
||
}
|
||
for mac, device in coordinator.data.items()
|
||
])
|
||
|
||
return self.json({
|
||
"prov_devices": prov_devices,
|
||
"mesh_devices": mesh_devices,
|
||
})
|