Use dpcode_wrapper in tuya alarm control panel platform (home-assistant#156306)

epenet · web-flow · commit 0ed342b43313 · 2025-11-11T09:36:09.000+01:00
diff --git a/homeassistant/components/tuya/alarm_control_panel.py b/homeassistant/components/tuya/alarm_control_panel.py
@@ -19,9 +19,9 @@
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
 from . import TuyaConfigEntry
-from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode, DPType
+from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode
 from .entity import TuyaEntity
-from .models import EnumTypeData, find_dpcode
+from .models import DPCodeEnumWrapper
 from .util import get_dpcode
 
 
@@ -85,9 +85,21 @@ def async_discover_device(device_ids: list[str]) -> None:
             device = manager.device_map[device_id]
             if descriptions := ALARM.get(device.category):
                 entities.extend(
-                    TuyaAlarmEntity(device, manager, description)
+                    TuyaAlarmEntity(
+                        device,
+                        manager,
+                        description,
+                        action_dpcode_wrapper=action_dpcode_wrapper,
+                        state_dpcode_wrapper=DPCodeEnumWrapper.find_dpcode(
+                            device, description.master_state
+                        ),
+                    )
                     for description in descriptions
-                    if description.key in device.status
+                    if (
+                        action_dpcode_wrapper := DPCodeEnumWrapper.find_dpcode(
+                            device, description.key, prefer_function=True
+                        )
+                    )
                 )
         async_add_entities(entities)
 
@@ -103,41 +115,31 @@ class TuyaAlarmEntity(TuyaEntity, AlarmControlPanelEntity):
 
     _attr_name = None
     _attr_code_arm_required = False
-    _master_state: EnumTypeData | None = None
     _alarm_msg_dpcode: DPCode | None = None
 
     def __init__(
         self,
         device: CustomerDevice,
         device_manager: Manager,
         description: TuyaAlarmControlPanelEntityDescription,
+        *,
+        action_dpcode_wrapper: DPCodeEnumWrapper,
+        state_dpcode_wrapper: DPCodeEnumWrapper | None,
     ) -> None:
         """Init Tuya Alarm."""
         super().__init__(device, device_manager)
         self.entity_description = description
         self._attr_unique_id = f"{super().unique_id}{description.key}"
+        self._action_dpcode_wrapper = action_dpcode_wrapper
+        self._state_dpcode_wrapper = state_dpcode_wrapper
 
         # Determine supported  modes
-        if supported_modes := find_dpcode(
-            self.device, description.key, dptype=DPType.ENUM, prefer_function=True
-        ):
-            if Mode.HOME in supported_modes.range:
-                self._attr_supported_features |= AlarmControlPanelEntityFeature.ARM_HOME
-
-            if Mode.ARM in supported_modes.range:
-                self._attr_supported_features |= AlarmControlPanelEntityFeature.ARM_AWAY
-
-            if Mode.SOS in supported_modes.range:
-                self._attr_supported_features |= AlarmControlPanelEntityFeature.TRIGGER
-
-        # Determine master state
-        if enum_type := find_dpcode(
-            self.device,
-            description.master_state,
-            dptype=DPType.ENUM,
-            prefer_function=True,
-        ):
-            self._master_state = enum_type
+        if Mode.HOME in action_dpcode_wrapper.type_information.range:
+            self._attr_supported_features |= AlarmControlPanelEntityFeature.ARM_HOME
+        if Mode.ARM in action_dpcode_wrapper.type_information.range:
+            self._attr_supported_features |= AlarmControlPanelEntityFeature.ARM_AWAY
+        if Mode.SOS in action_dpcode_wrapper.type_information.range:
+            self._attr_supported_features |= AlarmControlPanelEntityFeature.TRIGGER
 
         # Determine alarm message
         if dp_code := get_dpcode(self.device, description.alarm_msg):
@@ -149,8 +151,8 @@ def alarm_state(self) -> AlarmControlPanelState | None:
         # 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'.
         if (
-            self._master_state is not None
-            and self.device.status.get(self._master_state.dpcode) == State.ALARM
+            self._state_dpcode_wrapper is not None
+            and self.device.status.get(self._state_dpcode_wrapper.dpcode) == State.ALARM
         ):
             # Only report as triggered if NOT a battery warning
             if (
@@ -166,28 +168,26 @@ def alarm_state(self) -> AlarmControlPanelState | None:
     def changed_by(self) -> str | None:
         """Last change triggered by."""
         if (
-            self._master_state is not None
+            self._state_dpcode_wrapper is not None
             and self._alarm_msg_dpcode is not None
-            and self.device.status.get(self._master_state.dpcode) == State.ALARM
+            and self.device.status.get(self._state_dpcode_wrapper.dpcode) == State.ALARM
             and (encoded_msg := self.device.status.get(self._alarm_msg_dpcode))
         ):
             return b64decode(encoded_msg).decode("utf-16be")
         return None
 
-    def alarm_disarm(self, code: str | None = None) -> None:
+    async def async_alarm_disarm(self, code: str | None = None) -> None:
         """Send Disarm command."""
-        self._send_command(
-            [{"code": self.entity_description.key, "value": Mode.DISARMED}]
-        )
+        await self._async_send_dpcode_update(self._action_dpcode_wrapper, Mode.DISARMED)
 
-    def alarm_arm_home(self, code: str | None = None) -> None:
+    async def async_alarm_arm_home(self, code: str | None = None) -> None:
         """Send Home command."""
-        self._send_command([{"code": self.entity_description.key, "value": Mode.HOME}])
+        await self._async_send_dpcode_update(self._action_dpcode_wrapper, Mode.HOME)
 
-    def alarm_arm_away(self, code: str | None = None) -> None:
+    async def async_alarm_arm_away(self, code: str | None = None) -> None:
         """Send Arm command."""
-        self._send_command([{"code": self.entity_description.key, "value": Mode.ARM}])
+        await self._async_send_dpcode_update(self._action_dpcode_wrapper, Mode.ARM)
 
-    def alarm_trigger(self, code: str | None = None) -> None:
+    async def async_alarm_trigger(self, code: str | None = None) -> None:
         """Send SOS command."""
-        self._send_command([{"code": self.entity_description.key, "value": Mode.SOS}])
+        await self._async_send_dpcode_update(self._action_dpcode_wrapper, Mode.SOS)
diff --git a/homeassistant/components/tuya/models.py b/homeassistant/components/tuya/models.py
@@ -196,7 +196,7 @@ def __init__(self, dpcode: str, type_information: T) -> None:
     def find_dpcode(
         cls,
         device: CustomerDevice,
-        dpcodes: str | DPCode | tuple[DPCode, ...],
+        dpcodes: str | DPCode | tuple[DPCode, ...] | None,
         *,
         prefer_function: bool = False,
     ) -> Self | None:
diff --git a/tests/components/tuya/test_alarm_control_panel.py b/tests/components/tuya/test_alarm_control_panel.py
@@ -2,14 +2,22 @@
 
 from __future__ import annotations
 
+from typing import Any
 from unittest.mock import patch
 
 import pytest
 from syrupy.assertion import SnapshotAssertion
 from tuya_sharing import CustomerDevice, Manager
 
-from homeassistant.components.alarm_control_panel import AlarmControlPanelState
-from homeassistant.const import Platform
+from homeassistant.components.alarm_control_panel import (
+    DOMAIN as ALARM_DOMAIN,
+    SERVICE_ALARM_ARM_AWAY,
+    SERVICE_ALARM_ARM_HOME,
+    SERVICE_ALARM_DISARM,
+    SERVICE_ALARM_TRIGGER,
+    AlarmControlPanelState,
+)
+from homeassistant.const import ATTR_ENTITY_ID, Platform
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import entity_registry as er
 
@@ -33,6 +41,45 @@ async def test_platform_setup_and_discovery(
     await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
 
 
+@patch("homeassistant.components.tuya.PLATFORMS", [Platform.ALARM_CONTROL_PANEL])
+@pytest.mark.parametrize(
+    "mock_device_code",
+    ["mal_gyitctrjj1kefxp2"],
+)
+@pytest.mark.parametrize(
+    ("service", "command"),
+    [
+        (SERVICE_ALARM_ARM_AWAY, {"code": "master_mode", "value": "arm"}),
+        (SERVICE_ALARM_ARM_HOME, {"code": "master_mode", "value": "home"}),
+        (SERVICE_ALARM_DISARM, {"code": "master_mode", "value": "disarmed"}),
+        (SERVICE_ALARM_TRIGGER, {"code": "master_mode", "value": "sos"}),
+    ],
+)
+async def test_service(
+    hass: HomeAssistant,
+    mock_manager: Manager,
+    mock_config_entry: MockConfigEntry,
+    mock_device: CustomerDevice,
+    service: str,
+    command: dict[str, Any],
+) -> None:
+    """Test service."""
+    entity_id = "alarm_control_panel.multifunction_alarm"
+    await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
+
+    state = hass.states.get(entity_id)
+    assert state is not None, f"{entity_id} does not exist"
+    await hass.services.async_call(
+        ALARM_DOMAIN,
+        service,
+        {
+            ATTR_ENTITY_ID: entity_id,
+        },
+        blocking=True,
+    )
+    mock_manager.send_commands.assert_called_once_with(mock_device.id, [command])
+
+
 @pytest.mark.parametrize(
     "mock_device_code",
     ["mal_gyitctrjj1kefxp2"],
