RFC: Tuya data refactoring (implicit conversion of tuya.Data)

tests/test_tuya.py

@@ -276,12 +276,12 @@ def test_ts0121_signature(assert_signature_matches_quirk):
 
 async def test_tuya_data_conversion():
     """Test tuya conversion from Data to ztype and reverse."""
-    assert Data([4, 0, 0, 1, 39]).to_value(t.uint32_t) == 295
-    assert Data([4, 0, 0, 0, 220]).to_value(t.uint32_t) == 220
-    assert Data([4, 255, 255, 255, 236]).to_value(t.int32s) == -20
-    assert Data.from_value(t.uint32_t(295)) == [4, 0, 0, 1, 39]
-    assert Data.from_value(t.uint32_t(220)) == [4, 0, 0, 0, 220]
-    assert Data.from_value(t.int32s(-20)) == [4, 255, 255, 255, 236]
+    assert t.uint32_t(Data([4, 0, 0, 1, 39])) == 295
+    assert t.uint32_t(Data([4, 0, 0, 0, 220])) == 220
+    assert t.int32s(Data([4, 255, 255, 255, 236])) == -20
+    assert Data(t.uint32_t(295)) == [4, 0, 0, 1, 39]
+    assert Data(t.uint32_t(220)) == [4, 0, 0, 0, 220]
+    assert Data(t.int32s(-20)) == [4, 255, 255, 255, 236]
 
 
 class TuyaTestManufCluster(TuyaManufClusterAttributes):

zhaquirks/tuya/__init__.py

@@ -231,22 +231,35 @@ def __init__(self, value=None, function=0, *args, **kwargs):
 class Data(t.List, item_type=t.uint8_t):
     """list of uint8_t."""
 
-    @classmethod
-    def from_value(cls, value):
+    def __init__(self, value=None):
         """Convert from a zigpy typed value to a tuya data payload."""
+        if value is None:
+            super().__init__()
+            return
+        if type(value) is list or type(value) is bytes:
+            super().__init__(value)
+            return
         # serialized in little-endian by zigpy
-        data = cls(value.serialize())
+        super().__init__(value.serialize())
         # we want big-endian, with length prepended
-        data.append(len(data))
-        data.reverse()
-        return data
+        self.append(len(self))
+        self.reverse()
 
-    def to_value(self, ztype):
-        """Convert from a tuya data payload to a zigpy typed value."""
+    def __int__(self):
+        """Convert from a tuya data payload to an int typed value."""
         # first uint8_t is the length of the remaining data
         # tuya data is in big endian whereas ztypes use little endian
-        value, _ = ztype.deserialize(bytes(reversed(self[1:])))
-        return value
+        ints = {1: t.int8s, 2: t.int16s, 3: t.int24s, 4: t.int32s}
+        return ints[self[0]].deserialize(bytes(reversed(self[1:])))[0]
+
+    def __iter__(self):
+        """Convert from a tuya data payload to a list typed value."""
+        return iter(reversed(self[1:]))
+
+    def serialize(self) -> bytes:
+        """Overload serialize to avoid prior implicit conversion to list."""
+        assert self._item_type is not None
+        return b"".join([self._item_type(i).serialize() for i in self[:]])
 
 
 class TuyaDatapointData(t.Struct):
@@ -469,7 +482,7 @@ def handle_cluster_request(
             return
 
         ztype = self.attributes[tuya_cmd].type
-        zvalue = tuya_data.to_value(ztype)
+        zvalue = ztype(tuya_data)
         self._update_attribute(tuya_cmd, zvalue)
 
     def read_attributes(
@@ -492,7 +505,7 @@ async def write_attributes(self, attributes, manufacturer=None):
             cmd_payload.tsn = self.endpoint.device.application.get_sequence()
             cmd_payload.command_id = record.attrid
             cmd_payload.function = 0
-            cmd_payload.data = Data.from_value(record.value.value)
+            cmd_payload.data = record.value.value
 
             await super().command(
                 TUYA_SET_DATA,