diff --git a/tests/test_quirks.py b/tests/test_quirks.py index 1c3e3a878a..1833935172 100644 --- a/tests/test_quirks.py +++ b/tests/test_quirks.py @@ -15,7 +15,9 @@ import zigpy.profiles import zigpy.quirks as zq from zigpy.quirks import CustomDevice +from zigpy.quirks.v2 import QuirkBuilder import zigpy.types +from zigpy.zcl import foundation import zigpy.zdo.types import zhaquirks @@ -841,3 +843,40 @@ def check_for_duplicate_cluster_ids(clusters) -> None: for ep_id, ep_data in quirk.replacement[ENDPOINTS].items(): # noqa: B007 check_for_duplicate_cluster_ids(ep_data.get(INPUT_CLUSTERS, [])) check_for_duplicate_cluster_ids(ep_data.get(OUTPUT_CLUSTERS, [])) + + +async def test_local_data_cluster(zigpy_device_from_v2_quirk) -> None: + """Ensure reading attributes from a LocalDataCluster works as expected.""" + + class TestLocalCluster(zhaquirks.LocalDataCluster): + """Test cluster.""" + + cluster_id = 0x1234 + _CONSTANT_ATTRIBUTES = {1: 10} + _VALID_ATTRIBUTES = [2] + + ( + QuirkBuilder("manufacturer-local-test", "model") + .adds(TestLocalCluster) + .add_to_registry() + ) + device = zigpy_device_from_v2_quirk("manufacturer-local-test", "model") + assert isinstance(device.endpoints[1].in_clusters[0x1234], TestLocalCluster) + + # reading invalid attribute return unsupported attribute + assert await device.endpoints[1].in_clusters[0x1234].read_attributes([0]) == ( + {}, + {0: foundation.Status.UNSUPPORTED_ATTRIBUTE}, + ) + + # reading constant attribute works + assert await device.endpoints[1].in_clusters[0x1234].read_attributes([1]) == ( + {1: 10}, + {}, + ) + + # reading valid attribute returns None with success status + assert await device.endpoints[1].in_clusters[0x1234].read_attributes([2]) == ( + {2: None}, + {}, + ) diff --git a/zhaquirks/__init__.py b/zhaquirks/__init__.py index 5e62a96456..9b212bd21e 100644 --- a/zhaquirks/__init__.py +++ b/zhaquirks/__init__.py @@ -9,6 +9,7 @@ import pathlib import pkgutil import sys +import typing from typing import Any import zigpy.device @@ -60,9 +61,15 @@ def __init__(self, *args, **kwargs): class LocalDataCluster(CustomCluster): - """Cluster meant to prevent remote calls.""" + """Cluster meant to prevent remote calls. - _CONSTANT_ATTRIBUTES = {} + Set _CONSTANT_ATTRIBUTES to provide constant values for attribute ids. + Set _VALID_ATTRIBUTES to provide a list of valid attribute ids that will never be shown as unsupported. + These are attributes that should be populated later. + """ + + _CONSTANT_ATTRIBUTES: dict[int, typing.Any] = {} + _VALID_ATTRIBUTES: list[int] = [] async def bind(self): """Prevent bind.""" @@ -94,7 +101,10 @@ async def read_attributes_raw(self, attributes, manufacturer=None, **kwargs): record.value.value = self._CONSTANT_ATTRIBUTES[record.attrid] else: record.value.value = self._attr_cache.get(record.attrid) - if record.value.value is not None: + if ( + record.value.value is not None + or record.attrid in self._VALID_ATTRIBUTES + ): record.status = foundation.Status.SUCCESS return (records,) diff --git a/zhaquirks/tuya/__init__.py b/zhaquirks/tuya/__init__.py index ef9a734437..9c6f294cd0 100644 --- a/zhaquirks/tuya/__init__.py +++ b/zhaquirks/tuya/__init__.py @@ -864,7 +864,7 @@ class TuyaLocalCluster(LocalDataCluster): """ def update_attribute(self, attr_name: str, value: Any) -> None: - """Update attribute by attribute name.""" + """Update attribute by name and safeguard against unknown attributes.""" try: attr = self.attributes_by_name[attr_name] @@ -1497,8 +1497,32 @@ class TuyaNewManufCluster(CustomCluster): ), } + dp_to_attribute: dict[int, DPToAttributeMapping] = {} data_point_handlers: dict[int, str] = {} + def __init__(self, *args, **kwargs): + """Initialize the cluster and mark attributes as valid on LocalDataClusters.""" + super().__init__(*args, **kwargs) + for dp_map in self.dp_to_attribute.values(): + # get the endpoint that is being mapped to + endpoint = self.endpoint + if dp_map.endpoint_id: + endpoint = self.endpoint.device.endpoints.get(dp_map.endpoint_id) + + # the endpoint to be mapped to might not actually exist within all quirks + if not endpoint: + continue + + cluster = getattr(endpoint, dp_map.ep_attribute, None) + # the cluster to be mapped to might not actually exist within all quirks + if not cluster: + continue + + # mark mapped to attribute as valid if existing and if on a LocalDataCluster + attr = cluster.attributes_by_name.get(dp_map.attribute_name) + if attr and isinstance(cluster, LocalDataCluster): + cluster._VALID_ATTRIBUTES.append(attr.id) + def handle_cluster_request( self, hdr: foundation.ZCLHeader,