Split action and state wrapper in Tuya alarm control panel (home-assistant#158532)

Author: epenet

homeassistant/components/tuya/alarm_control_panel.py

@@ -21,6 +21,7 @@
 from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode
 from .entity import TuyaEntity
 from .models import DPCodeEnumWrapper, DPCodeRawWrapper
+from .type_information import EnumTypeInformation
 
 ALARM: dict[DeviceCategory, tuple[AlarmControlPanelEntityDescription, ...]] = {
     DeviceCategory.MAL: (
@@ -48,21 +49,14 @@ def read_device_status(self, device: CustomerDevice) -> str | None:
         return status.decode("utf-16be")
 
 
-class _AlarmModeWrapper(DPCodeEnumWrapper):
-    """Wrapper for the alarm mode of a device.
+class _AlarmStateWrapper(DPCodeEnumWrapper):
+    """Wrapper for the alarm state of a device.
 
     Handles alarm mode enum values and determines the alarm state,
     including logic for detecting when the alarm is triggered and
     distinguishing triggered state from battery warnings.
     """
 
-    _ACTION_MAPPINGS = {
-        # Home Assistant action => Tuya device mode
-        "arm_home": "home",
-        "arm_away": "arm",
-        "disarm": "disarmed",
-        "trigger": "sos",
-    }
     _STATE_MAPPINGS = {
         # Tuya device mode => Home Assistant panel state
         "disarmed": AlarmControlPanelState.DISARMED,
@@ -71,7 +65,9 @@ class _AlarmModeWrapper(DPCodeEnumWrapper):
         "sos": AlarmControlPanelState.TRIGGERED,
     }
 
-    def read_panel_state(self, device: CustomerDevice) -> AlarmControlPanelState | None:
+    def read_device_status(
+        self, device: CustomerDevice
+    ) -> AlarmControlPanelState | None:
         """Read the device status."""
         # When the alarm is triggered, only its 'state' is changing. From 'normal' to 'alarm'.
         # The 'mode' doesn't change, and stays as 'arm' or 'home'.
@@ -84,10 +80,22 @@ def read_panel_state(self, device: CustomerDevice) -> AlarmControlPanelState | N
             ):
                 return AlarmControlPanelState.TRIGGERED
 
-        if (status := self.read_device_status(device)) is None:
+        if (status := super().read_device_status(device)) is None:
             return None
         return self._STATE_MAPPINGS.get(status)
 
+
+class _AlarmActionWrapper(DPCodeEnumWrapper):
+    """Wrapper for setting the alarm mode of a device."""
+
+    _ACTION_MAPPINGS = {
+        # Home Assistant action => Tuya device mode
+        "arm_home": "home",
+        "arm_away": "arm",
+        "disarm": "disarmed",
+        "trigger": "sos",
+    }
+
     def supports_action(self, action: str) -> bool:
         """Return if action is supported."""
         return (
@@ -123,14 +131,19 @@ def async_discover_device(device_ids: list[str]) -> None:
                         device,
                         manager,
                         description,
-                        mode_wrapper=mode_wrapper,
+                        action_wrapper=_AlarmActionWrapper(
+                            master_mode.dpcode, master_mode
+                        ),
                         changed_by_wrapper=_AlarmChangedByWrapper.find_dpcode(
                             device, DPCode.ALARM_MSG
                         ),
+                        state_wrapper=_AlarmStateWrapper(
+                            master_mode.dpcode, master_mode
+                        ),
                     )
                     for description in descriptions
                     if (
-                        mode_wrapper := _AlarmModeWrapper.find_dpcode(
+                        master_mode := EnumTypeInformation.find_dpcode(
                             device, DPCode.MASTER_MODE, prefer_function=True
                         )
                     )
@@ -156,28 +169,30 @@ def __init__(
         device_manager: Manager,
         description: AlarmControlPanelEntityDescription,
         *,
-        mode_wrapper: _AlarmModeWrapper,
+        action_wrapper: _AlarmActionWrapper,
         changed_by_wrapper: _AlarmChangedByWrapper | None,
+        state_wrapper: _AlarmStateWrapper,
     ) -> None:
         """Init Tuya Alarm."""
         super().__init__(device, device_manager)
         self.entity_description = description
         self._attr_unique_id = f"{super().unique_id}{description.key}"
-        self._mode_wrapper = mode_wrapper
+        self._action_wrapper = action_wrapper
         self._changed_by_wrapper = changed_by_wrapper
+        self._state_wrapper = state_wrapper
 
         # Determine supported modes
-        if mode_wrapper.supports_action("arm_home"):
+        if action_wrapper.supports_action("arm_home"):
             self._attr_supported_features |= AlarmControlPanelEntityFeature.ARM_HOME
-        if mode_wrapper.supports_action("arm_away"):
+        if action_wrapper.supports_action("arm_away"):
             self._attr_supported_features |= AlarmControlPanelEntityFeature.ARM_AWAY
-        if mode_wrapper.supports_action("trigger"):
+        if action_wrapper.supports_action("trigger"):
             self._attr_supported_features |= AlarmControlPanelEntityFeature.TRIGGER
 
     @property
     def alarm_state(self) -> AlarmControlPanelState | None:
         """Return the state of the device."""
-        return self._mode_wrapper.read_panel_state(self.device)
+        return self._read_wrapper(self._state_wrapper)
 
     @property
     def changed_by(self) -> str | None:
@@ -186,16 +201,16 @@ def changed_by(self) -> str | None:
 
     async def async_alarm_disarm(self, code: str | None = None) -> None:
         """Send Disarm command."""
-        await self._async_send_wrapper_updates(self._mode_wrapper, "disarm")
+        await self._async_send_wrapper_updates(self._action_wrapper, "disarm")
 
     async def async_alarm_arm_home(self, code: str | None = None) -> None:
         """Send Home command."""
-        await self._async_send_wrapper_updates(self._mode_wrapper, "arm_home")
+        await self._async_send_wrapper_updates(self._action_wrapper, "arm_home")
 
     async def async_alarm_arm_away(self, code: str | None = None) -> None:
         """Send Arm command."""
-        await self._async_send_wrapper_updates(self._mode_wrapper, "arm_away")
+        await self._async_send_wrapper_updates(self._action_wrapper, "arm_away")
 
     async def async_alarm_trigger(self, code: str | None = None) -> None:
         """Send SOS command."""
-        await self._async_send_wrapper_updates(self._mode_wrapper, "trigger")
+        await self._async_send_wrapper_updates(self._action_wrapper, "trigger")