Refactor Tuya event platform to use DeviceWrapper (home-assistant#160366)

Author: epenet

homeassistant/components/tuya/event.py

@@ -21,80 +21,66 @@
 from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode
 from .entity import TuyaEntity
 from .models import (
+    DeviceWrapper,
     DPCodeEnumWrapper,
     DPCodeRawWrapper,
     DPCodeStringWrapper,
     DPCodeTypeInformationWrapper,
 )
 
 
-class _DPCodeEventWrapper(DPCodeTypeInformationWrapper):
-    """Base class for Tuya event wrappers."""
-
-    def __init__(self, dpcode: str, type_information: Any) -> None:
-        """Init _DPCodeEventWrapper."""
-        super().__init__(dpcode, type_information)
-        self.options = ["triggered"]
-
-    def get_event_type(
-        self, device: CustomerDevice, updated_status_properties: list[str] | None
-    ) -> str | None:
-        """Return the event type."""
-        if (
-            updated_status_properties is None
-            or self.dpcode not in updated_status_properties
-        ):
-            return None
-        return "triggered"
-
-    def get_event_attributes(self, device: CustomerDevice) -> dict[str, Any] | None:
-        """Return the event attributes."""
-        return None
-
-
-class _EventEnumWrapper(DPCodeEnumWrapper, _DPCodeEventWrapper):
+class _EventEnumWrapper(DPCodeEnumWrapper):
     """Wrapper for event enum DP codes."""
 
-    def get_event_type(
-        self, device: CustomerDevice, updated_status_properties: list[str] | None
-    ) -> str | None:
-        """Return the triggered event type."""
-        if (
-            updated_status_properties is None
-            or self.dpcode not in updated_status_properties
-        ):
+    def read_device_status(self, device: CustomerDevice) -> tuple[str, None] | None:
+        """Return the event details."""
+        if (raw_value := super().read_device_status(device)) is None:
             return None
-        return self.read_device_status(device)
+        return (raw_value, None)
 
 
-class _AlarmMessageWrapper(DPCodeStringWrapper, _DPCodeEventWrapper):
+class _AlarmMessageWrapper(DPCodeStringWrapper):
     """Wrapper for a STRING message on DPCode.ALARM_MESSAGE."""
 
-    def get_event_attributes(self, device: CustomerDevice) -> dict[str, Any] | None:
+    def __init__(self, dpcode: str, type_information: Any) -> None:
+        """Init _AlarmMessageWrapper."""
+        super().__init__(dpcode, type_information)
+        self.options = ["triggered"]
+
+    def read_device_status(
+        self, device: CustomerDevice
+    ) -> tuple[str, dict[str, Any]] | None:
         """Return the event attributes for the alarm message."""
-        if (raw_value := device.status.get(self.dpcode)) is None:
+        if (raw_value := super().read_device_status(device)) is None:
             return None
-        return {"message": b64decode(raw_value).decode("utf-8")}
+        return ("triggered", {"message": b64decode(raw_value).decode("utf-8")})
 
 
-class _DoorbellPicWrapper(DPCodeRawWrapper, _DPCodeEventWrapper):
+class _DoorbellPicWrapper(DPCodeRawWrapper):
     """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:
+    def __init__(self, dpcode: str, type_information: Any) -> None:
+        """Init _DoorbellPicWrapper."""
+        super().__init__(dpcode, type_information)
+        self.options = ["triggered"]
+
+    def read_device_status(
+        self, device: CustomerDevice
+    ) -> tuple[str, dict[str, Any]] | None:
         """Return the event attributes for the doorbell picture."""
         if (status := super().read_device_status(device)) is None:
             return None
-        return {"message": status.decode("utf-8")}
+        return ("triggered", {"message": status.decode("utf-8")})
 
 
 @dataclass(frozen=True)
 class TuyaEventEntityDescription(EventEntityDescription):
     """Describe a Tuya Event entity."""
 
-    wrapper_class: type[_DPCodeEventWrapper] = _EventEnumWrapper
+    wrapper_class: type[DPCodeTypeInformationWrapper] = _EventEnumWrapper
 
 
 # All descriptions can be found here. Mostly the Enum data types in the
@@ -220,7 +206,7 @@ def __init__(
         device: CustomerDevice,
         device_manager: Manager,
         description: EventEntityDescription,
-        dpcode_wrapper: _DPCodeEventWrapper,
+        dpcode_wrapper: DeviceWrapper[tuple[str, dict[str, Any] | None]],
     ) -> None:
         """Init Tuya event entity."""
         super().__init__(device, device_manager)
@@ -234,15 +220,11 @@ async def _handle_state_update(
         updated_status_properties: list[str] | None,
         dp_timestamps: dict | None = None,
     ) -> None:
-        if (
-            event_type := self._dpcode_wrapper.get_event_type(
-                self.device, updated_status_properties
-            )
-        ) is None:
+        if self._dpcode_wrapper.skip_update(
+            self.device, updated_status_properties
+        ) or not (event_data := self._dpcode_wrapper.read_device_status(self.device)):
             return
 
-        self._trigger_event(
-            event_type,
-            self._dpcode_wrapper.get_event_attributes(self.device),
-        )
+        event_type, event_attributes = event_data
+        self._trigger_event(event_type, event_attributes)
         self.async_write_ha_state()

homeassistant/components/tuya/models.py

@@ -30,6 +30,15 @@ class DeviceWrapper[T]:
 
     options: list[str]
 
+    def skip_update(
+        self, device: CustomerDevice, updated_status_properties: list[str] | None
+    ) -> bool:
+        """Determine if the wrapper should skip an update.
+
+        The default is to always skip, unless overridden in subclasses.
+        """
+        return True
+
     def read_device_status(self, device: CustomerDevice) -> T | None:
         """Read device status and convert to a Home Assistant value."""
         raise NotImplementedError
@@ -52,6 +61,19 @@ def __init__(self, dpcode: str) -> None:
         """Init DPCodeWrapper."""
         self.dpcode = dpcode
 
+    def skip_update(
+        self, device: CustomerDevice, updated_status_properties: list[str] | None
+    ) -> bool:
+        """Determine if the wrapper should skip an update.
+
+        By default, skip if updated_status_properties is given and
+        does not include this dpcode.
+        """
+        return (
+            updated_status_properties is None
+            or self.dpcode not in updated_status_properties
+        )
+
     def _convert_value_to_raw_value(self, device: CustomerDevice, value: Any) -> Any:
         """Convert a Home Assistant value back to a raw device value.