Use process_raw_value in Tuya RawTypeInformation (home-assistant#158521)

Author: epenet

homeassistant/components/tuya/alarm_control_panel.py

@@ -20,7 +20,7 @@
 from . import TuyaConfigEntry
 from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode
 from .entity import TuyaEntity
-from .models import DPCodeBase64Wrapper, DPCodeEnumWrapper
+from .models import DPCodeEnumWrapper, DPCodeRawWrapper
 
 ALARM: dict[DeviceCategory, tuple[AlarmControlPanelEntityDescription, ...]] = {
     DeviceCategory.MAL: (
@@ -32,7 +32,7 @@
 }
 
 
-class _AlarmChangedByWrapper(DPCodeBase64Wrapper):
+class _AlarmChangedByWrapper(DPCodeRawWrapper):
     """Wrapper for changed_by.
 
     Decode base64 to utf-16be string, but only if alarm has been triggered.
@@ -42,10 +42,10 @@ def read_device_status(self, device: CustomerDevice) -> str | None:
         """Read the device status."""
         if (
             device.status.get(DPCode.MASTER_STATE) != "alarm"
-            or (data := self.read_bytes(device)) is None
+            or (status := super().read_device_status(device)) is None
         ):
             return None
-        return data.decode("utf-16be")
+        return status.decode("utf-16be")
 
 
 class _AlarmModeWrapper(DPCodeEnumWrapper):
@@ -182,8 +182,6 @@ def alarm_state(self) -> AlarmControlPanelState | None:
     @property
     def changed_by(self) -> str | None:
         """Last change triggered by."""
-        if self._changed_by_wrapper is None:
-            return None
         return self._read_wrapper(self._changed_by_wrapper)
 
     async def async_alarm_disarm(self, code: str | None = None) -> None:

homeassistant/components/tuya/event.py

@@ -21,8 +21,8 @@
 from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode
 from .entity import TuyaEntity
 from .models import (
-    DPCodeBase64Wrapper,
     DPCodeEnumWrapper,
+    DPCodeRawWrapper,
     DPCodeStringWrapper,
     DPCodeTypeInformationWrapper,
 )
@@ -82,17 +82,17 @@ def get_event_attributes(self, device: CustomerDevice) -> dict[str, Any] | None:
         return {"message": b64decode(raw_value).decode("utf-8")}
 
 
-class _DoorbellPicWrapper(DPCodeBase64Wrapper, _DPCodeEventWrapper):
+class _DoorbellPicWrapper(DPCodeRawWrapper, _DPCodeEventWrapper):
     """Wrapper for a RAW message on DPCode.DOORBELL_PIC.
 
     It is expected that the RAW data is base64/utf8 encoded URL of the picture.
     """
 
     def get_event_attributes(self, device: CustomerDevice) -> dict[str, Any] | None:
         """Return the event attributes for the doorbell picture."""
-        if (raw_value := self._read_device_status_raw(device)) is None:
+        if (status := super().read_device_status(device)) is None:
             return None
-        return {"message": b64decode(raw_value).decode("utf-8")}
+        return {"message": status.decode("utf-8")}
 
 
 @dataclass(frozen=True)

homeassistant/components/tuya/models.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import base64
 from typing import Any, Self
 
 from tuya_sharing import CustomerDevice
@@ -114,20 +113,6 @@ def find_dpcode(
         return None
 
 
-class DPCodeBase64Wrapper(DPCodeTypeInformationWrapper[RawTypeInformation]):
-    """Wrapper to extract information from a RAW/binary value."""
-
-    _DPTYPE = RawTypeInformation
-
-    def read_bytes(self, device: CustomerDevice) -> bytes | None:
-        """Read the device value for the dpcode."""
-        if (raw_value := self._read_device_status_raw(device)) is None or (
-            len(decoded := base64.b64decode(raw_value)) == 0
-        ):
-            return None
-        return decoded
-
-
 class DPCodeBooleanWrapper(DPCodeTypeInformationWrapper[BooleanTypeInformation]):
     """Simple wrapper for boolean values.
 
@@ -198,6 +183,12 @@ def _convert_value_to_raw_value(self, device: CustomerDevice, value: Any) -> Any
         )
 
 
+class DPCodeRawWrapper(DPCodeTypeInformationWrapper[RawTypeInformation]):
+    """Wrapper to extract information from a RAW/binary value."""
+
+    _DPTYPE = RawTypeInformation
+
+
 class DPCodeStringWrapper(DPCodeTypeInformationWrapper[StringTypeInformation]):
     """Wrapper to extract information from a STRING value."""
 

homeassistant/components/tuya/sensor.py

@@ -39,10 +39,10 @@
 )
 from .entity import TuyaEntity
 from .models import (
-    DPCodeBase64Wrapper,
     DPCodeEnumWrapper,
     DPCodeIntegerWrapper,
     DPCodeJsonWrapper,
+    DPCodeRawWrapper,
     DPCodeTypeInformationWrapper,
     DPCodeWrapper,
 )
@@ -119,7 +119,7 @@ def read_device_status(self, device: CustomerDevice) -> float | None:
         return raw_value.get("voltage")
 
 
-class _RawElectricityDataWrapper(DPCodeBase64Wrapper):
+class _RawElectricityDataWrapper(DPCodeRawWrapper):
     """Custom DPCode Wrapper for extracting ElectricityData from base64."""
 
     def _convert(self, value: ElectricityData) -> float:
@@ -128,7 +128,7 @@ def _convert(self, value: ElectricityData) -> float:
 
     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 or (
+        if (raw_value := super().read_device_status(device)) is None or (
             value := ElectricityData.from_bytes(raw_value)
         ) is None:
             return None

homeassistant/components/tuya/type_information.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import base64
 from dataclasses import dataclass
 from typing import Any, ClassVar, Self, cast
 
@@ -292,11 +293,19 @@ class JsonTypeInformation(TypeInformation[Any]):
 
 
 @dataclass(kw_only=True)
-class RawTypeInformation(TypeInformation[Any]):
+class RawTypeInformation(TypeInformation[bytes]):
     """Raw type information."""
 
     _DPTYPE = DPType.RAW
 
+    def process_raw_value(
+        self, raw_value: Any | None, device: CustomerDevice
+    ) -> bytes | None:
+        """Read and process raw value against this type information."""
+        if raw_value is None:
+            return None
+        return base64.b64decode(raw_value)
+
 
 @dataclass(kw_only=True)
 class StringTypeInformation(TypeInformation[str]):