Fix parsing of Tuya electricity RAW values (home-assistant#157039)

Author: abelyliu

homeassistant/components/tuya/raw_data_models.py

@@ -0,0 +1,60 @@
+"""Parsers for RAW (base64-encoded bytes) values."""
+
+from dataclasses import dataclass
+import struct
+from typing import Self
+
+
+@dataclass(kw_only=True)
+class ElectricityData:
+    """Electricity RAW value."""
+
+    current: float
+    power: float
+    voltage: float
+
+    @classmethod
+    def from_bytes(cls, raw: bytes) -> Self | None:
+        """Parse bytes and return an ElectricityValue object."""
+        # Format:
+        # - legacy: 8 bytes
+        # - v01: [ver=0x01][len=0x0F][data(15 bytes)]
+        # - v02: [ver=0x02][len=0x0F][data(15 bytes)][sign_bitmap(1 byte)]
+        # Data layout (big-endian):
+        # - voltage: 2B, unit 0.1 V
+        # - current: 3B, unit 0.001 A (i.e., mA)
+        # - active power: 3B, unit 0.001 kW (i.e., W)
+        # - reactive power: 3B, unit 0.001 kVar
+        # - apparent power: 3B, unit 0.001 kVA
+        # - power factor: 1B, unit 0.01
+        # Sign bitmap (v02 only, 1 bit means negative):
+        # - bit0 current
+        # - bit1 active power
+        # - bit2 reactive
+        # - bit3 power factor
+
+        is_v1 = len(raw) == 17 and raw[0:2] == b"\x01\x0f"
+        is_v2 = len(raw) == 18 and raw[0:2] == b"\x02\x0f"
+        if is_v1 or is_v2:
+            data = raw[2:17]
+
+            voltage = struct.unpack(">H", data[0:2])[0] / 10.0
+            current = struct.unpack(">L", b"\x00" + data[2:5])[0]
+            power = struct.unpack(">L", b"\x00" + data[5:8])[0]
+
+            if is_v2:
+                sign_bitmap = raw[17]
+                if sign_bitmap & 0x01:
+                    current = -current
+                if sign_bitmap & 0x02:
+                    power = -power
+
+            return cls(current=current, power=power, voltage=voltage)
+
+        if len(raw) >= 8:
+            voltage = struct.unpack(">H", raw[0:2])[0] / 10.0
+            current = struct.unpack(">L", b"\x00" + raw[2:5])[0]
+            power = struct.unpack(">L", b"\x00" + raw[5:8])[0]
+            return cls(current=current, power=power, voltage=voltage)
+
+        return None

homeassistant/components/tuya/sensor.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-import struct
 
 from tuya_sharing import CustomerDevice, Manager
 
@@ -49,6 +48,7 @@
     DPCodeWrapper,
     EnumTypeData,
 )
+from .raw_data_models import ElectricityData
 
 
 class _WindDirectionWrapper(DPCodeTypeInformationWrapper[EnumTypeData]):
@@ -120,42 +120,52 @@ def read_device_status(self, device: CustomerDevice) -> float | None:
         return raw_value.get("voltage")
 
 
-class _RawElectricityCurrentWrapper(DPCodeBase64Wrapper):
-    """Custom DPCode Wrapper for extracting electricity current from base64."""
+class _RawElectricityDataWrapper(DPCodeBase64Wrapper):
+    """Custom DPCode Wrapper for extracting ElectricityData from base64."""
 
-    native_unit = UnitOfElectricCurrent.MILLIAMPERE
-    suggested_unit = UnitOfElectricCurrent.AMPERE
+    def _convert(self, value: ElectricityData) -> float:
+        """Extract specific value from T."""
+        raise NotImplementedError
 
     def read_device_status(self, device: CustomerDevice) -> float | None:
         """Read the device value for the dpcode."""
-        if (raw_value := super().read_bytes(device)) is None:
+        if (raw_value := super().read_bytes(device)) is None or (
+            value := ElectricityData.from_bytes(raw_value)
+        ) is None:
             return None
-        return struct.unpack(">L", b"\x00" + raw_value[2:5])[0]
+        return self._convert(value)
+
+
+class _RawElectricityCurrentWrapper(_RawElectricityDataWrapper):
+    """Custom DPCode Wrapper for extracting electricity current from base64."""
+
+    native_unit = UnitOfElectricCurrent.MILLIAMPERE
+    suggested_unit = UnitOfElectricCurrent.AMPERE
+
+    def _convert(self, value: ElectricityData) -> float:
+        """Extract specific value from ElectricityData."""
+        return value.current
 
 
-class _RawElectricityPowerWrapper(DPCodeBase64Wrapper):
+class _RawElectricityPowerWrapper(_RawElectricityDataWrapper):
     """Custom DPCode Wrapper for extracting electricity power from base64."""
 
     native_unit = UnitOfPower.WATT
     suggested_unit = UnitOfPower.KILO_WATT
 
-    def read_device_status(self, device: CustomerDevice) -> float | None:
-        """Read the device value for the dpcode."""
-        if (raw_value := super().read_bytes(device)) is None:
-            return None
-        return struct.unpack(">L", b"\x00" + raw_value[5:8])[0]
+    def _convert(self, value: ElectricityData) -> float:
+        """Extract specific value from ElectricityData."""
+        return value.power
 
 
-class _RawElectricityVoltageWrapper(DPCodeBase64Wrapper):
+class _RawElectricityVoltageWrapper(_RawElectricityDataWrapper):
     """Custom DPCode Wrapper for extracting electricity voltage from base64."""
 
     native_unit = UnitOfElectricPotential.VOLT
 
-    def read_device_status(self, device: CustomerDevice) -> float | None:
-        """Read the device value for the dpcode."""
-        if (raw_value := super().read_bytes(device)) is None:
-            return None
-        return struct.unpack(">H", raw_value[0:2])[0] / 10.0
+    def _convert(self, value: ElectricityData) -> float:
+        """Extract specific value from ElectricityData."""
+        return value.voltage
 
 
 CURRENT_WRAPPER = (_RawElectricityCurrentWrapper, _JsonElectricityCurrentWrapper)

tests/components/tuya/snapshots/test_sensor.ambr

@@ -5630,7 +5630,7 @@
     'last_changed': <ANY>,
     'last_reported': <ANY>,
     'last_updated': <ANY>,
-    'state': '599.296',
+    'state': '0.072',
   })
 # ---
 # name: test_platform_setup_and_discovery[sensor.duan_lu_qi_ha_phase_a_power-entry]
@@ -5689,7 +5689,7 @@
     'last_changed': <ANY>,
     'last_reported': <ANY>,
     'last_updated': <ANY>,
-    'state': '18.432',
+    'state': '0.008',
   })
 # ---
 # name: test_platform_setup_and_discovery[sensor.duan_lu_qi_ha_phase_a_voltage-entry]
@@ -5745,7 +5745,7 @@
     'last_changed': <ANY>,
     'last_reported': <ANY>,
     'last_updated': <ANY>,
-    'state': '52.7',
+    'state': '234.1',
   })
 # ---
 # name: test_platform_setup_and_discovery[sensor.duan_lu_qi_ha_supply_frequency-entry]