From debc1f78d49e463f85aeee1c0b400d103c1d9e44 Mon Sep 17 00:00:00 2001 From: "David F. Mulcahey" <david.mulcahey@me.com> Date: Sat, 4 Apr 2020 08:40:55 -0400 Subject: [PATCH] Add zigbee information to ZHA device information (#33612) * add zigbee signature to zha device info * add typing * use props and sort clusters * review comment --- .../components/zha/core/channels/__init__.py | 31 ++++++++++++++++++- homeassistant/components/zha/core/const.py | 5 +++ homeassistant/components/zha/core/device.py | 13 ++++++++ 3 files changed, 48 insertions(+), 1 deletion(-) diff --git a/homeassistant/components/zha/core/channels/__init__.py b/homeassistant/components/zha/core/channels/__init__.py index 91a23e17f12..18eb2a6c1cc 100644 --- a/homeassistant/components/zha/core/channels/__init__.py +++ b/homeassistant/components/zha/core/channels/__init__.py @@ -1,7 +1,7 @@ """Channels module for Zigbee Home Automation.""" import asyncio import logging -from typing import Any, Dict, List, Optional, Union +from typing import Any, Dict, List, Optional, Tuple, Union from homeassistant.core import callback from homeassistant.helpers.dispatcher import async_dispatcher_send @@ -92,6 +92,14 @@ class Channels: """Return the unique id for this channel.""" return self._unique_id + @property + def zigbee_signature(self) -> Dict[int, Dict[str, Any]]: + """Get the zigbee signatures for the pools in channels.""" + return { + signature[0]: signature[1] + for signature in [pool.zigbee_signature for pool in self.pools] + } + @classmethod def new(cls, zha_device: zha_typing.ZhaDeviceType) -> "Channels": """Create new instance.""" @@ -231,6 +239,27 @@ class ChannelPool: """Return the unique id for this channel.""" return self._unique_id + @property + def zigbee_signature(self) -> Tuple[int, Dict[str, Any]]: + """Get the zigbee signature for the endpoint this pool represents.""" + return ( + self.endpoint.endpoint_id, + { + const.ATTR_PROFILE_ID: self.endpoint.profile_id, + const.ATTR_DEVICE_TYPE: f"0x{self.endpoint.device_type:04x}" + if self.endpoint.device_type is not None + else "", + const.ATTR_IN_CLUSTERS: [ + f"0x{cluster_id:04x}" + for cluster_id in sorted(self.endpoint.in_clusters) + ], + const.ATTR_OUT_CLUSTERS: [ + f"0x{cluster_id:04x}" + for cluster_id in sorted(self.endpoint.out_clusters) + ], + }, + ) + @classmethod def new(cls, channels: Channels, ep_id: int) -> "ChannelPool": """Create new channels for an endpoint.""" diff --git a/homeassistant/components/zha/core/const.py b/homeassistant/components/zha/core/const.py index da151f67dbb..055b5627bb1 100644 --- a/homeassistant/components/zha/core/const.py +++ b/homeassistant/components/zha/core/const.py @@ -22,8 +22,10 @@ ATTR_COMMAND = "command" ATTR_COMMAND_TYPE = "command_type" ATTR_DEVICE_IEEE = "device_ieee" ATTR_DEVICE_TYPE = "device_type" +ATTR_ENDPOINTS = "endpoints" ATTR_ENDPOINT_ID = "endpoint_id" ATTR_IEEE = "ieee" +ATTR_IN_CLUSTERS = "in_clusters" ATTR_LAST_SEEN = "last_seen" ATTR_LEVEL = "level" ATTR_LQI = "lqi" @@ -32,8 +34,11 @@ ATTR_MANUFACTURER_CODE = "manufacturer_code" ATTR_MEMBERS = "members" ATTR_MODEL = "model" ATTR_NAME = "name" +ATTR_NODE_DESCRIPTOR = "node_descriptor" ATTR_NWK = "nwk" +ATTR_OUT_CLUSTERS = "out_clusters" ATTR_POWER_SOURCE = "power_source" +ATTR_PROFILE_ID = "profile_id" ATTR_QUIRK_APPLIED = "quirk_applied" ATTR_QUIRK_CLASS = "quirk_class" ATTR_RSSI = "rssi" diff --git a/homeassistant/components/zha/core/device.py b/homeassistant/components/zha/core/device.py index ad3d1ff18ad..716ed5040ae 100644 --- a/homeassistant/components/zha/core/device.py +++ b/homeassistant/components/zha/core/device.py @@ -5,6 +5,7 @@ from enum import Enum import logging import random import time +from typing import Any, Dict from zigpy import types import zigpy.exceptions @@ -31,6 +32,7 @@ from .const import ( ATTR_COMMAND_TYPE, ATTR_DEVICE_TYPE, ATTR_ENDPOINT_ID, + ATTR_ENDPOINTS, ATTR_IEEE, ATTR_LAST_SEEN, ATTR_LQI, @@ -38,11 +40,13 @@ from .const import ( ATTR_MANUFACTURER_CODE, ATTR_MODEL, ATTR_NAME, + ATTR_NODE_DESCRIPTOR, ATTR_NWK, ATTR_POWER_SOURCE, ATTR_QUIRK_APPLIED, ATTR_QUIRK_CLASS, ATTR_RSSI, + ATTR_SIGNATURE, ATTR_VALUE, CLUSTER_COMMAND_SERVER, CLUSTER_COMMANDS_CLIENT, @@ -267,6 +271,14 @@ class ZHADevice(LogMixin): """Return True if sensor is available.""" return self._available + @property + def zigbee_signature(self) -> Dict[str, Any]: + """Get zigbee signature for this device.""" + return { + ATTR_NODE_DESCRIPTOR: str(self._zigpy_device.node_desc), + ATTR_ENDPOINTS: self._channels.zigbee_signature, + } + def set_available(self, available): """Set availability from restore and prevent signals.""" self._available = available @@ -366,6 +378,7 @@ class ZHADevice(LogMixin): ATTR_LAST_SEEN: update_time, ATTR_AVAILABLE: self.available, ATTR_DEVICE_TYPE: self.device_type, + ATTR_SIGNATURE: self.zigbee_signature, } async def async_configure(self): -- GitLab