From badebe0c7f02eef5b60e61725a40fb1f9fcf6246 Mon Sep 17 00:00:00 2001 From: epenet <6771947+epenet@users.noreply.github.com> Date: Tue, 6 Jan 2026 16:09:13 +0100 Subject: [PATCH] Refactor Tuya event platform to use DeviceWrapper (#160366) --- homeassistant/components/tuya/event.py | 90 ++++++++++--------------- homeassistant/components/tuya/models.py | 22 ++++++ 2 files changed, 58 insertions(+), 54 deletions(-) diff --git a/homeassistant/components/tuya/event.py b/homeassistant/components/tuya/event.py index 5cfc8424f61..de754f15a47 100644 --- a/homeassistant/components/tuya/event.py +++ b/homeassistant/components/tuya/event.py @@ -21,6 +21,7 @@ from . import TuyaConfigEntry from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode from .entity import TuyaEntity from .models import ( + DeviceWrapper, DPCodeEnumWrapper, DPCodeRawWrapper, DPCodeStringWrapper, @@ -28,73 +29,58 @@ from .models import ( ) -class _DPCodeEventWrapper(DPCodeTypeInformationWrapper): - """Base class for Tuya event wrappers.""" +class _EventEnumWrapper(DPCodeEnumWrapper): + """Wrapper for event enum DP codes.""" + + 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 (raw_value, None) + + +class _AlarmMessageWrapper(DPCodeStringWrapper): + """Wrapper for a STRING message on DPCode.ALARM_MESSAGE.""" def __init__(self, dpcode: str, type_information: Any) -> None: - """Init _DPCodeEventWrapper.""" + """Init _AlarmMessageWrapper.""" 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): - """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 - ): - return None - return self.read_device_status(device) - - -class _AlarmMessageWrapper(DPCodeStringWrapper, _DPCodeEventWrapper): - """Wrapper for a STRING message on DPCode.ALARM_MESSAGE.""" - - def get_event_attributes(self, device: CustomerDevice) -> dict[str, Any] | None: + 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 @@ class TuyaEventEntity(TuyaEntity, EventEntity): 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 @@ class TuyaEventEntity(TuyaEntity, EventEntity): 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() diff --git a/homeassistant/components/tuya/models.py b/homeassistant/components/tuya/models.py index 4feea69d797..a7234e36882 100644 --- a/homeassistant/components/tuya/models.py +++ b/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 @@ class DPCodeWrapper(DeviceWrapper): """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.