"""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, })