Decompose attribute_name tuple to new DP mapping list format

zhaquirks/tuya/__init__.py

@@ -1456,6 +1456,46 @@
     converter: Callable[[Any], Any] | None = None
     endpoint_id: int | None = None
 
+    def __init__(
+        self,
+        ep_attribute: str,
+        attribute_name: str | tuple[str, ...],
+        converter: Callable[[Any], Any] | None = None,
+        endpoint_id: int | None = None,
+    ):
+        """Init DPToAttributeMapping."""
+        self.ep_attribute = ep_attribute
+        self.attribute_name = attribute_name
+        self.converter = converter
+        self.endpoint_id = endpoint_id
+
+        if not isinstance(attribute_name, str):
+            _LOGGER.info(
+                "Using tuple attribute_name is deprecated, please multiple "
+                "DPToAttributeMapping instances instead. %s",
+                attribute_name,
+            )
+
+    def decompose_attributes(self) -> list[DPToAttributeMapping]:
+        """Decompose attributes into multiple mappings."""
+
+        def wrap_converter(converter: Callable[[Any], Any] | None, attr_index: int):
+            if converter is None:
+                return None
+            return lambda args: converter(args)[attr_index]
+
+        if isinstance(self.attribute_name, tuple):
+            return [
+                DPToAttributeMapping(
+                    self.ep_attribute,
+                    attr_name,
+                    wrap_converter(self.converter, attr_index),
+                    self.endpoint_id,
+                )
+                for attr_index, attr_name in enumerate(self.attribute_name)
+            ]
+        return [self]
+
 
 @dataclasses.dataclass
 class AttributeWithMask:
@@ -1531,8 +1571,10 @@
         super().__init__(*args, **kwargs)
 
         self._dp_to_attributes: dict[int, list[DPToAttributeMapping]] = {
-            dp: attr if isinstance(attr, list) else [attr]
-            for dp, attr in self.dp_to_attribute.items()
+            dp: [x for mapping in mappings for x in mapping.decompose_attributes()]
+            if isinstance(mappings, list)
+            else mappings.decompose_attributes()
+            for dp, mappings in self.dp_to_attribute.items()
         }
         for dp_map in self._dp_to_attributes.values():
             # get the endpoint that is being mapped to
@@ -1650,15 +1692,9 @@
             if mapped_attr.converter:
                 value = mapped_attr.converter(value)
 
-            if isinstance(mapped_attr.attribute_name, tuple):
-                for k, v in zip(mapped_attr.attribute_name, value):
-                    if isinstance(v, AttributeWithMask):
-                        v = cluster.get(k, 0) & (~v.mask) | v.value
-                    cluster.update_attribute(k, v)
-            else:
-                if isinstance(value, AttributeWithMask):
-                    value = (
-                        cluster.get(mapped_attr.attribute_name, 0) & (~value.mask)
-                        | value.value
-                    )
-                cluster.update_attribute(mapped_attr.attribute_name, value)
+            if isinstance(value, AttributeWithMask):
+                value = (
+                    cluster.get(mapped_attr.attribute_name, 0) & (~value.mask)
+                    | value.value
+                )
+            cluster.update_attribute(mapped_attr.attribute_name, value)

zhaquirks/tuya/builder/__init__.py

@@ -203,6 +203,7 @@ def __init__(
         """Init the TuyaQuirkBuilder."""
         self.tuya_data_point_handlers: dict[int, str] = {}
         self.tuya_dp_to_attribute: dict[int, list[DPToAttributeMapping]] = {}
+        self.tuya_attributes_to_dp_converters: dict[int, Callable[[Any], Any]] = {}
         self.new_attributes: set[foundation.ZCLAttributeDef] = set()
         super().__init__(manufacturer, model, registry)
         # quirk_file will point to the init call above if called from this QuirkBuilder,
@@ -532,11 +533,11 @@ def tuya_dp(
                     ep_attribute,
                     attribute_name,
                     converter=converter,
-                    dp_converter=dp_converter,
                     endpoint_id=endpoint_id,
                 )
             ],
             dp_handler,
+            dp_converter,
         )
         return self
 
@@ -545,6 +546,7 @@ def tuya_dp_multi(
         dp_id: int,
         attribute_mapping: list[DPToAttributeMapping],
         dp_handler: str = "_dp_2_attr_update",
+        dp_converter: Callable[[Any], Any] | None = None,
     ) -> QuirkBuilder:  # fmt: skip
         """Add Tuya DP Converter that maps to multiple attributes."""
 
@@ -553,6 +555,8 @@ def tuya_dp_multi(
 
         self.tuya_dp_to_attribute.update({dp_id: attribute_mapping})
         self.tuya_data_point_handlers.update({dp_id: dp_handler})
+        if dp_converter:
+            self.tuya_attributes_to_dp_converters.update({dp_id: dp_converter})
         return self
 
     def tuya_dp_attribute(
@@ -845,6 +849,7 @@ def add_to_registry(
             self.new_attributes
             or self.tuya_data_point_handlers
             or self.tuya_dp_to_attribute
+            or self.tuya_attributes_to_dp_converters
             or force_add_cluster
         ):
 
@@ -865,7 +870,9 @@ class AttributeDefs(NewAttributeDefs):
 
             TuyaReplacementCluster.data_point_handlers = self.tuya_data_point_handlers
             TuyaReplacementCluster.dp_to_attribute = self.tuya_dp_to_attribute
-
+            TuyaReplacementCluster.attributes_to_dp_converters = (
+                self.tuya_attributes_to_dp_converters
+            )
             TuyaReplacementCluster.mcu_write_command = mcu_write_command
 
             self.replaces(TuyaReplacementCluster)

zhaquirks/tuya/mcu/__init__.py

@@ -4,6 +4,7 @@
 
 from collections.abc import Callable
 import datetime
+import logging
 from typing import Any
 
 import zigpy.types as t
@@ -28,6 +29,8 @@
     TuyaTimePayload,
 )
 
+_LOGGER = logging.getLogger(__name__)
+
 # New manufacturer attributes
 ATTR_MCU_VERSION = 0xEF00
 
@@ -49,6 +52,12 @@
         """Init method for compatibility with previous quirks using positional arguments."""
         super().__init__(ep_attribute, attribute_name, converter, endpoint_id)
         self.dp_converter = dp_converter
+        if dp_converter:
+            _LOGGER.info(
+                "DPToAttributeMapping with dp_converter is deprecated, use TuyaQuirksBuilder "
+                "(or TuyaMCUCluster.attributes_to_dp_converters) instead. attribute_name: %s",
+                attribute_name,
+            )
 
 
 class TuyaClusterData(t.Struct):
@@ -121,6 +130,7 @@
 
     set_time_offset = 1970  # MCU timestamp from 1/1/1970
     set_time_local_offset = None
+    attributes_to_dp_converters: dict[int, Callable[[Any], Any]] = {}
 
     class AttributeDefs(TuyaNewManufCluster.AttributeDefs):
         """Attribute Definitions."""
@@ -199,6 +209,21 @@
     def __init__(self, *args, **kwargs):
         """Init."""
         super().__init__(*args, **kwargs)
+
+        self._attributes_to_dp_converters: dict[int, Callable[[Any], Any]]
+        if self.attributes_to_dp_converters:
+            self._attributes_to_dp_converters = self.attributes_to_dp_converters
+        else:
+            # convert from legacy DP2AttributeMapping with attribute_name tuple to new
+            # DP2AttributeMapping with single attribute_name
+            self._attributes_to_dp_converters = {}
+            for dp, mappings in self.dp_to_attribute.items():
+                if not isinstance(mappings, list):
+                    mappings = [mappings]
+                for dp_mapping in mappings:
+                    if hasattr(dp_mapping, "dp_converter") and dp_mapping.dp_converter:
+                        self._attributes_to_dp_converters[dp] = dp_mapping.dp_converter
+
         # Cluster for endpoint: 1 (listen MCU commands)
         self.endpoint.device.command_bus = Bus()
         self.endpoint.device.command_bus.add_listener(self)
@@ -223,20 +248,19 @@
             cmd_payload.tsn = self.endpoint.device.application.get_sequence()
 
             val = data.attr_value
-            if mapping.dp_converter:
+
+            if attr_to_dp_converter := self._attributes_to_dp_converters.get(dp):
                 args = []
-                if isinstance(mapping.attribute_name, tuple):
+                for dp_attr in self._dp_to_attributes[dp]:
+                    if dp_attr.attribute_name == data.cluster_attr:
+                        args.append(val)
+                        continue
                     endpoint = self.endpoint
-                    if mapping.endpoint_id:
+                    if dp_attr.endpoint_id:
                         endpoint = endpoint.device.endpoints[mapping.endpoint_id]
-                    cluster = getattr(endpoint, mapping.ep_attribute)
-                    for attr in mapping.attribute_name:
-                        args.append(
-                            val if attr == data.cluster_attr else cluster.get(attr)
-                        )
-                else:
-                    args.append(val)
-                val = mapping.dp_converter(*args)
+                    cluster = getattr(endpoint, dp_attr.ep_attribute)
+                    args.append(cluster.get(dp_attr.attribute_name))
+                val = attr_to_dp_converter(*args)
             self.debug("value: %s", val)
 
             dpd = TuyaDatapointData(dp, val)
@@ -285,21 +309,20 @@
         result: dict[int, DPToAttributeMapping] = {}
         for dp, dp_mapping in self._dp_to_attributes.items():
             for mapped_attr in dp_mapping:
-                if (
-                    attribute_name == mapped_attr.attribute_name
-                    or (
-                        isinstance(mapped_attr.attribute_name, tuple)
-                        and attribute_name in mapped_attr.attribute_name
-                    )
-                ) and (
+                if attribute_name != mapped_attr.attribute_name:
+                    continue
+                if not (
                     (
                         mapped_attr.endpoint_id is None
                         and endpoint_id == self.endpoint.endpoint_id
                     )
                     or (endpoint_id == mapped_attr.endpoint_id)
                 ):
-                    self.debug("get_dp_mapping --> found DP: %s", dp)
-                    result[dp] = mapped_attr
+                    continue
+                self.debug("get_dp_mapping --> found DP: %s", dp)
+                result[dp] = mapped_attr
+                break
+
         return result
 
     def handle_mcu_version_response(self, payload: MCUVersion) -> foundation.Status:
@@ -315,7 +338,7 @@
         self.debug("handle_set_time_request payload: %s", payload)
         payload_rsp = TuyaTimePayload()
 
         utc_now = datetime.datetime.utcnow()  # noqa: DTZ003
         now = datetime.datetime.now()
 
         offset_time = datetime.datetime(self.set_time_offset, 1, 1)

zhaquirks/tuya/tuya_trv.py

@@ -696,17 +696,17 @@ async def write_attributes(
                     TuyaPresetMode.Heat: Thermostat.SystemMode.Heat,
                     TuyaPresetMode.Off: Thermostat.SystemMode.Off,
                 }[x],
-                dp_converter=lambda x: {
-                    Thermostat.SystemMode.Auto: TuyaPresetMode.Auto,
-                    Thermostat.SystemMode.Heat: TuyaPresetMode.Heat,
-                    Thermostat.SystemMode.Off: TuyaPresetMode.Off,
-                }[x],
             ),
             DPToAttributeMapping(
                 ep_attribute=TuyaMCUCluster.ep_attribute,
                 attribute_name="preset_mode",
             ),
         ],
+        dp_converter=lambda x, _: {
+            Thermostat.SystemMode.Auto: TuyaPresetMode.Auto,
+            Thermostat.SystemMode.Heat: TuyaPresetMode.Heat,
+            Thermostat.SystemMode.Off: TuyaPresetMode.Off,
+        }[x],
     )
     .tuya_attribute(
         dp_id=2,