Add zigbee information to ZHA device information (#33612)

* add zigbee signature to zha device info

* add typing

* use props and sort clusters

* review comment
This commit is contained in:
David F. Mulcahey 2020-04-04 08:40:55 -04:00 committed by GitHub
parent 05192b7164
commit debc1f78d4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 48 additions and 1 deletions

View File

@ -1,7 +1,7 @@
"""Channels module for Zigbee Home Automation.""" """Channels module for Zigbee Home Automation."""
import asyncio import asyncio
import logging 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.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_send from homeassistant.helpers.dispatcher import async_dispatcher_send
@ -92,6 +92,14 @@ class Channels:
"""Return the unique id for this channel.""" """Return the unique id for this channel."""
return self._unique_id 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 @classmethod
def new(cls, zha_device: zha_typing.ZhaDeviceType) -> "Channels": def new(cls, zha_device: zha_typing.ZhaDeviceType) -> "Channels":
"""Create new instance.""" """Create new instance."""
@ -231,6 +239,27 @@ class ChannelPool:
"""Return the unique id for this channel.""" """Return the unique id for this channel."""
return self._unique_id 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 @classmethod
def new(cls, channels: Channels, ep_id: int) -> "ChannelPool": def new(cls, channels: Channels, ep_id: int) -> "ChannelPool":
"""Create new channels for an endpoint.""" """Create new channels for an endpoint."""

View File

@ -22,8 +22,10 @@ ATTR_COMMAND = "command"
ATTR_COMMAND_TYPE = "command_type" ATTR_COMMAND_TYPE = "command_type"
ATTR_DEVICE_IEEE = "device_ieee" ATTR_DEVICE_IEEE = "device_ieee"
ATTR_DEVICE_TYPE = "device_type" ATTR_DEVICE_TYPE = "device_type"
ATTR_ENDPOINTS = "endpoints"
ATTR_ENDPOINT_ID = "endpoint_id" ATTR_ENDPOINT_ID = "endpoint_id"
ATTR_IEEE = "ieee" ATTR_IEEE = "ieee"
ATTR_IN_CLUSTERS = "in_clusters"
ATTR_LAST_SEEN = "last_seen" ATTR_LAST_SEEN = "last_seen"
ATTR_LEVEL = "level" ATTR_LEVEL = "level"
ATTR_LQI = "lqi" ATTR_LQI = "lqi"
@ -32,8 +34,11 @@ ATTR_MANUFACTURER_CODE = "manufacturer_code"
ATTR_MEMBERS = "members" ATTR_MEMBERS = "members"
ATTR_MODEL = "model" ATTR_MODEL = "model"
ATTR_NAME = "name" ATTR_NAME = "name"
ATTR_NODE_DESCRIPTOR = "node_descriptor"
ATTR_NWK = "nwk" ATTR_NWK = "nwk"
ATTR_OUT_CLUSTERS = "out_clusters"
ATTR_POWER_SOURCE = "power_source" ATTR_POWER_SOURCE = "power_source"
ATTR_PROFILE_ID = "profile_id"
ATTR_QUIRK_APPLIED = "quirk_applied" ATTR_QUIRK_APPLIED = "quirk_applied"
ATTR_QUIRK_CLASS = "quirk_class" ATTR_QUIRK_CLASS = "quirk_class"
ATTR_RSSI = "rssi" ATTR_RSSI = "rssi"

View File

@ -5,6 +5,7 @@ from enum import Enum
import logging import logging
import random import random
import time import time
from typing import Any, Dict
from zigpy import types from zigpy import types
import zigpy.exceptions import zigpy.exceptions
@ -31,6 +32,7 @@ from .const import (
ATTR_COMMAND_TYPE, ATTR_COMMAND_TYPE,
ATTR_DEVICE_TYPE, ATTR_DEVICE_TYPE,
ATTR_ENDPOINT_ID, ATTR_ENDPOINT_ID,
ATTR_ENDPOINTS,
ATTR_IEEE, ATTR_IEEE,
ATTR_LAST_SEEN, ATTR_LAST_SEEN,
ATTR_LQI, ATTR_LQI,
@ -38,11 +40,13 @@ from .const import (
ATTR_MANUFACTURER_CODE, ATTR_MANUFACTURER_CODE,
ATTR_MODEL, ATTR_MODEL,
ATTR_NAME, ATTR_NAME,
ATTR_NODE_DESCRIPTOR,
ATTR_NWK, ATTR_NWK,
ATTR_POWER_SOURCE, ATTR_POWER_SOURCE,
ATTR_QUIRK_APPLIED, ATTR_QUIRK_APPLIED,
ATTR_QUIRK_CLASS, ATTR_QUIRK_CLASS,
ATTR_RSSI, ATTR_RSSI,
ATTR_SIGNATURE,
ATTR_VALUE, ATTR_VALUE,
CLUSTER_COMMAND_SERVER, CLUSTER_COMMAND_SERVER,
CLUSTER_COMMANDS_CLIENT, CLUSTER_COMMANDS_CLIENT,
@ -267,6 +271,14 @@ class ZHADevice(LogMixin):
"""Return True if sensor is available.""" """Return True if sensor is available."""
return self._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): def set_available(self, available):
"""Set availability from restore and prevent signals.""" """Set availability from restore and prevent signals."""
self._available = available self._available = available
@ -366,6 +378,7 @@ class ZHADevice(LogMixin):
ATTR_LAST_SEEN: update_time, ATTR_LAST_SEEN: update_time,
ATTR_AVAILABLE: self.available, ATTR_AVAILABLE: self.available,
ATTR_DEVICE_TYPE: self.device_type, ATTR_DEVICE_TYPE: self.device_type,
ATTR_SIGNATURE: self.zigbee_signature,
} }
async def async_configure(self): async def async_configure(self):