"""Support for Tuya event entities.""" from __future__ import annotations from base64 import b64decode from dataclasses import dataclass from typing import Any from tuya_sharing import CustomerDevice, Manager from homeassistant.components.event import ( EventDeviceClass, EventEntity, EventEntityDescription, ) from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from . import TuyaConfigEntry from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode from .entity import TuyaEntity from .models import ( DPCodeEnumWrapper, DPCodeRawWrapper, DPCodeStringWrapper, DPCodeTypeInformationWrapper, ) class _DPCodeEventWrapper(DPCodeTypeInformationWrapper): """Base class for Tuya event wrappers.""" options: list[str] 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): """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: """Return the event attributes for the alarm message.""" if (raw_value := device.status.get(self.dpcode)) is None: return None return {"message": b64decode(raw_value).decode("utf-8")} 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 (status := super().read_device_status(device)) is None: return None return {"message": status.decode("utf-8")} @dataclass(frozen=True) class TuyaEventEntityDescription(EventEntityDescription): """Describe a Tuya Event entity.""" wrapper_class: type[_DPCodeEventWrapper] = _EventEnumWrapper # All descriptions can be found here. Mostly the Enum data types in the # default status set of each category (that don't have a set instruction) # end up being events. EVENTS: dict[DeviceCategory, tuple[TuyaEventEntityDescription, ...]] = { DeviceCategory.SP: ( TuyaEventEntityDescription( key=DPCode.ALARM_MESSAGE, device_class=EventDeviceClass.DOORBELL, translation_key="doorbell_message", wrapper_class=_AlarmMessageWrapper, ), TuyaEventEntityDescription( key=DPCode.DOORBELL_PIC, device_class=EventDeviceClass.DOORBELL, translation_key="doorbell_picture", wrapper_class=_DoorbellPicWrapper, ), ), DeviceCategory.WXKG: ( TuyaEventEntityDescription( key=DPCode.SWITCH_MODE1, device_class=EventDeviceClass.BUTTON, translation_key="numbered_button", translation_placeholders={"button_number": "1"}, ), TuyaEventEntityDescription( key=DPCode.SWITCH_MODE2, device_class=EventDeviceClass.BUTTON, translation_key="numbered_button", translation_placeholders={"button_number": "2"}, ), TuyaEventEntityDescription( key=DPCode.SWITCH_MODE3, device_class=EventDeviceClass.BUTTON, translation_key="numbered_button", translation_placeholders={"button_number": "3"}, ), TuyaEventEntityDescription( key=DPCode.SWITCH_MODE4, device_class=EventDeviceClass.BUTTON, translation_key="numbered_button", translation_placeholders={"button_number": "4"}, ), TuyaEventEntityDescription( key=DPCode.SWITCH_MODE5, device_class=EventDeviceClass.BUTTON, translation_key="numbered_button", translation_placeholders={"button_number": "5"}, ), TuyaEventEntityDescription( key=DPCode.SWITCH_MODE6, device_class=EventDeviceClass.BUTTON, translation_key="numbered_button", translation_placeholders={"button_number": "6"}, ), TuyaEventEntityDescription( key=DPCode.SWITCH_MODE7, device_class=EventDeviceClass.BUTTON, translation_key="numbered_button", translation_placeholders={"button_number": "7"}, ), TuyaEventEntityDescription( key=DPCode.SWITCH_MODE8, device_class=EventDeviceClass.BUTTON, translation_key="numbered_button", translation_placeholders={"button_number": "8"}, ), TuyaEventEntityDescription( key=DPCode.SWITCH_MODE9, device_class=EventDeviceClass.BUTTON, translation_key="numbered_button", translation_placeholders={"button_number": "9"}, ), ), } async def async_setup_entry( hass: HomeAssistant, entry: TuyaConfigEntry, async_add_entities: AddConfigEntryEntitiesCallback, ) -> None: """Set up Tuya events dynamically through Tuya discovery.""" manager = entry.runtime_data.manager @callback def async_discover_device(device_ids: list[str]) -> None: """Discover and add a discovered Tuya binary sensor.""" entities: list[TuyaEventEntity] = [] for device_id in device_ids: device = manager.device_map[device_id] if descriptions := EVENTS.get(device.category): entities.extend( TuyaEventEntity( device, manager, description, dpcode_wrapper=dpcode_wrapper ) for description in descriptions if ( dpcode_wrapper := description.wrapper_class.find_dpcode( device, description.key ) ) ) async_add_entities(entities) async_discover_device([*manager.device_map]) entry.async_on_unload( async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device) ) class TuyaEventEntity(TuyaEntity, EventEntity): """Tuya Event Entity.""" entity_description: EventEntityDescription def __init__( self, device: CustomerDevice, device_manager: Manager, description: EventEntityDescription, dpcode_wrapper: _DPCodeEventWrapper, ) -> None: """Init Tuya event entity.""" super().__init__(device, device_manager) self.entity_description = description self._attr_unique_id = f"{super().unique_id}{description.key}" self._dpcode_wrapper = dpcode_wrapper self._attr_event_types = dpcode_wrapper.options async def _handle_state_update( self, 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: return self._trigger_event( event_type, self._dpcode_wrapper.get_event_attributes(self.device), ) self.async_write_ha_state()