Skip to content

Commit

Permalink
Update unsupported attribute on configure
Browse files Browse the repository at this point in the history
This allow us to only display available attributes.
  • Loading branch information
elupus committed Jul 26, 2024
1 parent 70b6e91 commit 42e57ff
Showing 1 changed file with 47 additions and 0 deletions.
47 changes: 47 additions & 0 deletions zha/zigbee/cluster_handlers/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,11 +11,14 @@
from typing import TYPE_CHECKING, Any, Final, ParamSpec, TypedDict

import zigpy.exceptions
import zigpy.types
import zigpy.util
import zigpy.zcl
from zigpy.zcl.foundation import (
CommandSchema,
ConfigureReportingResponseRecord,
DiscoverAttributesResponseRecord,
GeneralCommand,
Status,
ZCLAttributeDef,
)
Expand Down Expand Up @@ -441,6 +444,10 @@ async def async_configure(self) -> None:
if ch_specific_cfg:
self.debug("Performing cluster handler specific configuration")
await ch_specific_cfg()

self.debug("Discovering available attributes")
await self.discover_unsupported_attributes()

self.debug("finished cluster handler configuration")
else:
self.debug("skipping cluster handler configuration")
Expand Down Expand Up @@ -624,6 +631,46 @@ async def write_attributes_safe(
f"Failed to write attribute {name}={value}: {record.status}",
)

async def _discover_attributes_all(
self,
) -> list[DiscoverAttributesResponseRecord] | None:
discovery_complete = zigpy.types.Bool.false
start_attribute_id = 0
attribute_info = []
cluster = self.cluster
while discovery_complete != zigpy.types.Bool.true:
rsp = await cluster.discover_attributes(
start_attribute_id=start_attribute_id, max_attribute_ids=0xFF
)
assert rsp, "Must have a response to discover request"

if rsp.command.id == GeneralCommand.Default_Response:
self.debug(
"Ignoring attribute discovery due to unexpected default response"
)
return None

attribute_info.extend(rsp.attribute_info)
discovery_complete = rsp.discovery_complete
start_attribute_id = (
max((info.attrid for info in rsp.attribute_info), default=0) + 1
)
return attribute_info

async def discover_unsupported_attributes(self):
"""Discover the list of unsupported attributes from the device."""
attribute_info = await self._discover_attributes_all()
if attribute_info is None:
return
attr_ids = {info.attrid for info in attribute_info}

cluster = self.cluster
for attr_id in cluster.attributes:
if attr_id in attr_ids:
cluster.remove_unsupported_attribute(attr_id)
else:
cluster.add_unsupported_attribute(attr_id)

def log(self, level, msg, *args, **kwargs) -> None:
"""Log a message."""
msg = f"[%s:%s]: {msg}"
Expand Down

0 comments on commit 42e57ff

Please sign in to comment.