Add _VALID_ATTRIBUTES to LocalDataCluster, fix Tuya datapoint mappings on LocalDataClusters

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},
+        {},
+    )

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

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,