Add google_assistant alarm_control_panel (#26249)

* add alarm_control_panel to google_assistant

* add cancel arming option

* raise error if requested state is same as current

* rework executing command logic

* Add tests

* fixed tests

* fixed level synonyms

homeassistant/components/google_assistant/const.py

@@ -15,6 +15,7 @@ from homeassistant.components import (
     sensor,
     switch,
     vacuum,
+    alarm_control_panel,
 )
 
 DOMAIN = "google_assistant"
@@ -48,6 +49,7 @@ DEFAULT_EXPOSED_DOMAINS = [
     "lock",
     "binary_sensor",
     "sensor",
+    "alarm_control_panel",
 ]
 
 PREFIX_TYPES = "action.devices.types."
@@ -66,6 +68,7 @@ TYPE_SENSOR = PREFIX_TYPES + "SENSOR"
 TYPE_DOOR = PREFIX_TYPES + "DOOR"
 TYPE_TV = PREFIX_TYPES + "TV"
 TYPE_SPEAKER = PREFIX_TYPES + "SPEAKER"
+TYPE_ALARM = PREFIX_TYPES + "SECURITYSYSTEM"
 
 SERVICE_REQUEST_SYNC = "request_sync"
 HOMEGRAPH_URL = "https://homegraph.googleapis.com/"
@@ -81,6 +84,9 @@ ERR_PROTOCOL_ERROR = "protocolError"
 ERR_UNKNOWN_ERROR = "unknownError"
 ERR_FUNCTION_NOT_SUPPORTED = "functionNotSupported"
 
+ERR_ALREADY_DISARMED = "alreadyDisarmed"
+ERR_ALREADY_ARMED = "alreadyArmed"
+
 ERR_CHALLENGE_NEEDED = "challengeNeeded"
 ERR_CHALLENGE_NOT_SETUP = "challengeFailedNotSetup"
 ERR_TOO_MANY_FAILED_ATTEMPTS = "tooManyFailedAttempts"
@@ -106,6 +112,7 @@ DOMAIN_TO_GOOGLE_TYPES = {
     script.DOMAIN: TYPE_SCENE,
     switch.DOMAIN: TYPE_SWITCH,
     vacuum.DOMAIN: TYPE_VACUUM,
+    alarm_control_panel.DOMAIN: TYPE_ALARM,
 }
 
 DEVICE_CLASS_TO_GOOGLE_TYPES = {

homeassistant/components/google_assistant/trait.py

@@ -16,6 +16,7 @@ from homeassistant.components import (
     sensor,
     switch,
     vacuum,
+    alarm_control_panel,
 )
 from homeassistant.components.climate import const as climate
 from homeassistant.const import (
@@ -31,6 +32,20 @@ from homeassistant.const import (
     ATTR_SUPPORTED_FEATURES,
     ATTR_TEMPERATURE,
     ATTR_ASSUMED_STATE,
+    SERVICE_ALARM_DISARM,
+    SERVICE_ALARM_ARM_HOME,
+    SERVICE_ALARM_ARM_AWAY,
+    SERVICE_ALARM_ARM_NIGHT,
+    SERVICE_ALARM_ARM_CUSTOM_BYPASS,
+    SERVICE_ALARM_TRIGGER,
+    STATE_ALARM_ARMED_HOME,
+    STATE_ALARM_ARMED_AWAY,
+    STATE_ALARM_ARMED_NIGHT,
+    STATE_ALARM_ARMED_CUSTOM_BYPASS,
+    STATE_ALARM_DISARMED,
+    STATE_ALARM_TRIGGERED,
+    STATE_ALARM_PENDING,
+    ATTR_CODE,
     STATE_UNKNOWN,
 )
 from homeassistant.core import DOMAIN as HA_DOMAIN
@@ -43,6 +58,8 @@ from .const import (
     CHALLENGE_ACK_NEEDED,
     CHALLENGE_PIN_NEEDED,
     CHALLENGE_FAILED_PIN_NEEDED,
+    ERR_ALREADY_DISARMED,
+    ERR_ALREADY_ARMED,
 )
 from .error import SmartHomeError, ChallengeNeeded
 
@@ -62,6 +79,7 @@ TRAIT_FANSPEED = PREFIX_TRAITS + "FanSpeed"
 TRAIT_MODES = PREFIX_TRAITS + "Modes"
 TRAIT_OPENCLOSE = PREFIX_TRAITS + "OpenClose"
 TRAIT_VOLUME = PREFIX_TRAITS + "Volume"
+TRAIT_ARMDISARM = PREFIX_TRAITS + "ArmDisarm"
 
 PREFIX_COMMANDS = "action.devices.commands."
 COMMAND_ONOFF = PREFIX_COMMANDS + "OnOff"
@@ -85,6 +103,7 @@ COMMAND_MODES = PREFIX_COMMANDS + "SetModes"
 COMMAND_OPENCLOSE = PREFIX_COMMANDS + "OpenClose"
 COMMAND_SET_VOLUME = PREFIX_COMMANDS + "setVolume"
 COMMAND_VOLUME_RELATIVE = PREFIX_COMMANDS + "volumeRelative"
+COMMAND_ARMDISARM = PREFIX_COMMANDS + "ArmDisarm"
 
 TRAITS = []
 
@@ -873,6 +892,98 @@ class LockUnlockTrait(_Trait):
         )
 
 
+@register_trait
+class ArmDisArmTrait(_Trait):
+    """Trait to Arm or Disarm a Security System.
+
+    https://developers.google.com/actions/smarthome/traits/armdisarm
+    """
+
+    name = TRAIT_ARMDISARM
+    commands = [COMMAND_ARMDISARM]
+
+    state_to_service = {
+        STATE_ALARM_ARMED_HOME: SERVICE_ALARM_ARM_HOME,
+        STATE_ALARM_ARMED_AWAY: SERVICE_ALARM_ARM_AWAY,
+        STATE_ALARM_ARMED_NIGHT: SERVICE_ALARM_ARM_NIGHT,
+        STATE_ALARM_ARMED_CUSTOM_BYPASS: SERVICE_ALARM_ARM_CUSTOM_BYPASS,
+        STATE_ALARM_TRIGGERED: SERVICE_ALARM_TRIGGER,
+    }
+
+    @staticmethod
+    def supported(domain, features, device_class):
+        """Test if state is supported."""
+        return domain == alarm_control_panel.DOMAIN
+
+    @staticmethod
+    def might_2fa(domain, features, device_class):
+        """Return if the trait might ask for 2FA."""
+        return True
+
+    def sync_attributes(self):
+        """Return ArmDisarm attributes for a sync request."""
+        response = {}
+        levels = []
+        for state in self.state_to_service:
+            # level synonyms are generated from state names
+            # 'armed_away' becomes 'armed away' or 'away'
+            level_synonym = [state.replace("_", " ")]
+            if state != STATE_ALARM_TRIGGERED:
+                level_synonym.append(state.split("_")[1])
+
+            level = {
+                "level_name": state,
+                "level_values": [{"level_synonym": level_synonym, "lang": "en"}],
+            }
+            levels.append(level)
+        response["availableArmLevels"] = {"levels": levels, "ordered": False}
+        return response
+
+    def query_attributes(self):
+        """Return ArmDisarm query attributes."""
+        if "post_pending_state" in self.state.attributes:
+            armed_state = self.state.attributes["post_pending_state"]
+        else:
+            armed_state = self.state.state
+        response = {"isArmed": armed_state in self.state_to_service}
+        if response["isArmed"]:
+            response.update({"currentArmLevel": armed_state})
+        return response
+
+    async def execute(self, command, data, params, challenge):
+        """Execute an ArmDisarm command."""
+        if params["arm"] and not params.get("cancel"):
+            if self.state.state == params["armLevel"]:
+                raise SmartHomeError(ERR_ALREADY_ARMED, "System is already armed")
+            if self.state.attributes["code_arm_required"]:
+                _verify_pin_challenge(data, self.state, challenge)
+            service = self.state_to_service[params["armLevel"]]
+        # disarm the system without asking for code when
+        # 'cancel' arming action is received while current status is pending
+        elif (
+            params["arm"]
+            and params.get("cancel")
+            and self.state.state == STATE_ALARM_PENDING
+        ):
+            service = SERVICE_ALARM_DISARM
+        else:
+            if self.state.state == STATE_ALARM_DISARMED:
+                raise SmartHomeError(ERR_ALREADY_DISARMED, "System is already disarmed")
+            _verify_pin_challenge(data, self.state, challenge)
+            service = SERVICE_ALARM_DISARM
+
+        await self.hass.services.async_call(
+            alarm_control_panel.DOMAIN,
+            service,
+            {
+                ATTR_ENTITY_ID: self.state.entity_id,
+                ATTR_CODE: data.config.secure_devices_pin,
+            },
+            blocking=True,
+            context=data.context,
+        )
+
+
 @register_trait
 class FanSpeedTrait(_Trait):
     """Trait to control speed of Fan.
@@ -1343,7 +1454,6 @@ def _verify_pin_challenge(data, state, challenge):
     """Verify a pin challenge."""
     if not data.config.should_2fa(state):
         return
-
     if not data.config.secure_devices_pin:
         raise SmartHomeError(ERR_CHALLENGE_NOT_SETUP, "Challenge is not set up")
 

tests/components/google_assistant/__init__.py

@@ -230,4 +230,11 @@ DEMO_DEVICES = [
         "type": "action.devices.types.LOCK",
         "willReportState": False,
     },
+    {
+        "id": "alarm_control_panel.alarm",
+        "name": {"name": "Alarm"},
+        "traits": ["action.devices.traits.ArmDisarm"],
+        "type": "action.devices.types.SECURITYSYSTEM",
+        "willReportState": False,
+    },
 ]

tests/components/google_assistant/test_google_assistant.py

@@ -7,7 +7,15 @@ from aiohttp.hdrs import CONTENT_TYPE, AUTHORIZATION
 import pytest
 
 from homeassistant import core, const, setup
-from homeassistant.components import fan, cover, light, switch, lock, media_player
+from homeassistant.components import (
+    fan,
+    cover,
+    light,
+    switch,
+    lock,
+    media_player,
+    alarm_control_panel,
+)
 from homeassistant.components.climate import const as climate
 from homeassistant.const import CLOUD_NEVER_EXPOSED_ENTITIES
 from homeassistant.components import google_assistant as ga
@@ -98,6 +106,14 @@ def hass_fixture(loop, hass):
         setup.async_setup_component(hass, lock.DOMAIN, {"lock": [{"platform": "demo"}]})
     )
 
+    loop.run_until_complete(
+        setup.async_setup_component(
+            hass,
+            alarm_control_panel.DOMAIN,
+            {"alarm_control_panel": [{"platform": "demo"}]},
+        )
+    )
+
     return hass
 
 

tests/components/google_assistant/test_trait.py

@@ -1,6 +1,6 @@
 """Tests for the Google Assistant traits."""
 from unittest.mock import patch, Mock
-
+import logging
 import pytest
 
 from homeassistant.components import (
@@ -18,12 +18,16 @@ from homeassistant.components import (
     switch,
     vacuum,
     group,
+    alarm_control_panel,
 )
 from homeassistant.components.climate import const as climate
 from homeassistant.components.google_assistant import trait, helpers, const, error
 from homeassistant.const import (
     STATE_ON,
     STATE_OFF,
+    STATE_ALARM_ARMED_AWAY,
+    STATE_ALARM_DISARMED,
+    STATE_ALARM_PENDING,
     ATTR_ENTITY_ID,
     SERVICE_TURN_ON,
     SERVICE_TURN_OFF,
@@ -40,6 +44,7 @@ from homeassistant.util import color
 from tests.common import async_mock_service, mock_coro
 from . import BASIC_CONFIG, MockConfig
 
+_LOGGER = logging.getLogger(__name__)
 
 REQ_ID = "ff36a3cc-ec34-11e6-b1a0-64510650abcf"
 
@@ -816,6 +821,336 @@ async def test_lock_unlock_unlock(hass):
     assert len(calls) == 2
 
 
+async def test_arm_disarm_arm_away(hass):
+    """Test ArmDisarm trait Arming support for alarm_control_panel domain."""
+    assert helpers.get_google_type(alarm_control_panel.DOMAIN, None) is not None
+    assert trait.ArmDisArmTrait.supported(alarm_control_panel.DOMAIN, 0, None)
+    assert trait.ArmDisArmTrait.might_2fa(alarm_control_panel.DOMAIN, 0, None)
+
+    trt = trait.ArmDisArmTrait(
+        hass,
+        State(
+            "alarm_control_panel.alarm",
+            STATE_ALARM_ARMED_AWAY,
+            {alarm_control_panel.ATTR_CODE_ARM_REQUIRED: True},
+        ),
+        PIN_CONFIG,
+    )
+    assert trt.sync_attributes() == {
+        "availableArmLevels": {
+            "levels": [
+                {
+                    "level_name": "armed_home",
+                    "level_values": [
+                        {"level_synonym": ["armed home", "home"], "lang": "en"}
+                    ],
+                },
+                {
+                    "level_name": "armed_away",
+                    "level_values": [
+                        {"level_synonym": ["armed away", "away"], "lang": "en"}
+                    ],
+                },
+                {
+                    "level_name": "armed_night",
+                    "level_values": [
+                        {"level_synonym": ["armed night", "night"], "lang": "en"}
+                    ],
+                },
+                {
+                    "level_name": "armed_custom_bypass",
+                    "level_values": [
+                        {
+                            "level_synonym": ["armed custom bypass", "custom"],
+                            "lang": "en",
+                        }
+                    ],
+                },
+                {
+                    "level_name": "triggered",
+                    "level_values": [{"level_synonym": ["triggered"], "lang": "en"}],
+                },
+            ],
+            "ordered": False,
+        }
+    }
+
+    assert trt.query_attributes() == {
+        "isArmed": True,
+        "currentArmLevel": STATE_ALARM_ARMED_AWAY,
+    }
+
+    assert trt.can_execute(
+        trait.COMMAND_ARMDISARM, {"arm": True, "armLevel": STATE_ALARM_ARMED_AWAY}
+    )
+
+    calls = async_mock_service(
+        hass, alarm_control_panel.DOMAIN, alarm_control_panel.SERVICE_ALARM_ARM_AWAY
+    )
+
+    # Test with no secure_pin configured
+
+    with pytest.raises(error.SmartHomeError) as err:
+        trt = trait.ArmDisArmTrait(
+            hass,
+            State(
+                "alarm_control_panel.alarm",
+                STATE_ALARM_DISARMED,
+                {alarm_control_panel.ATTR_CODE_ARM_REQUIRED: True},
+            ),
+            BASIC_CONFIG,
+        )
+        await trt.execute(
+            trait.COMMAND_ARMDISARM,
+            BASIC_DATA,
+            {"arm": True, "armLevel": STATE_ALARM_ARMED_AWAY},
+            {},
+        )
+    assert len(calls) == 0
+    assert err.value.code == const.ERR_CHALLENGE_NOT_SETUP
+
+    trt = trait.ArmDisArmTrait(
+        hass,
+        State(
+            "alarm_control_panel.alarm",
+            STATE_ALARM_DISARMED,
+            {alarm_control_panel.ATTR_CODE_ARM_REQUIRED: True},
+        ),
+        PIN_CONFIG,
+    )
+    # No challenge data
+    with pytest.raises(error.ChallengeNeeded) as err:
+        await trt.execute(
+            trait.COMMAND_ARMDISARM,
+            PIN_DATA,
+            {"arm": True, "armLevel": STATE_ALARM_ARMED_AWAY},
+            {},
+        )
+    assert len(calls) == 0
+    assert err.value.code == const.ERR_CHALLENGE_NEEDED
+    assert err.value.challenge_type == const.CHALLENGE_PIN_NEEDED
+
+    # invalid pin
+    with pytest.raises(error.ChallengeNeeded) as err:
+        await trt.execute(
+            trait.COMMAND_ARMDISARM,
+            PIN_DATA,
+            {"arm": True, "armLevel": STATE_ALARM_ARMED_AWAY},
+            {"pin": 9999},
+        )
+    assert len(calls) == 0
+    assert err.value.code == const.ERR_CHALLENGE_NEEDED
+    assert err.value.challenge_type == const.CHALLENGE_FAILED_PIN_NEEDED
+
+    # correct pin
+    await trt.execute(
+        trait.COMMAND_ARMDISARM,
+        PIN_DATA,
+        {"arm": True, "armLevel": STATE_ALARM_ARMED_AWAY},
+        {"pin": "1234"},
+    )
+
+    assert len(calls) == 1
+
+    # Test already armed
+    with pytest.raises(error.SmartHomeError) as err:
+        trt = trait.ArmDisArmTrait(
+            hass,
+            State(
+                "alarm_control_panel.alarm",
+                STATE_ALARM_ARMED_AWAY,
+                {alarm_control_panel.ATTR_CODE_ARM_REQUIRED: True},
+            ),
+            PIN_CONFIG,
+        )
+        await trt.execute(
+            trait.COMMAND_ARMDISARM,
+            PIN_DATA,
+            {"arm": True, "armLevel": STATE_ALARM_ARMED_AWAY},
+            {},
+        )
+    assert len(calls) == 1
+    assert err.value.code == const.ERR_ALREADY_ARMED
+
+    # Test with code_arm_required False
+    trt = trait.ArmDisArmTrait(
+        hass,
+        State(
+            "alarm_control_panel.alarm",
+            STATE_ALARM_DISARMED,
+            {alarm_control_panel.ATTR_CODE_ARM_REQUIRED: False},
+        ),
+        PIN_CONFIG,
+    )
+    await trt.execute(
+        trait.COMMAND_ARMDISARM,
+        PIN_DATA,
+        {"arm": True, "armLevel": STATE_ALARM_ARMED_AWAY},
+        {},
+    )
+    assert len(calls) == 2
+
+
+async def test_arm_disarm_disarm(hass):
+    """Test ArmDisarm trait Disarming support for alarm_control_panel domain."""
+    assert helpers.get_google_type(alarm_control_panel.DOMAIN, None) is not None
+    assert trait.ArmDisArmTrait.supported(alarm_control_panel.DOMAIN, 0, None)
+    assert trait.ArmDisArmTrait.might_2fa(alarm_control_panel.DOMAIN, 0, None)
+
+    trt = trait.ArmDisArmTrait(
+        hass,
+        State(
+            "alarm_control_panel.alarm",
+            STATE_ALARM_DISARMED,
+            {alarm_control_panel.ATTR_CODE_ARM_REQUIRED: True},
+        ),
+        PIN_CONFIG,
+    )
+    assert trt.sync_attributes() == {
+        "availableArmLevels": {
+            "levels": [
+                {
+                    "level_name": "armed_home",
+                    "level_values": [
+                        {"level_synonym": ["armed home", "home"], "lang": "en"}
+                    ],
+                },
+                {
+                    "level_name": "armed_away",
+                    "level_values": [
+                        {"level_synonym": ["armed away", "away"], "lang": "en"}
+                    ],
+                },
+                {
+                    "level_name": "armed_night",
+                    "level_values": [
+                        {"level_synonym": ["armed night", "night"], "lang": "en"}
+                    ],
+                },
+                {
+                    "level_name": "armed_custom_bypass",
+                    "level_values": [
+                        {
+                            "level_synonym": ["armed custom bypass", "custom"],
+                            "lang": "en",
+                        }
+                    ],
+                },
+                {
+                    "level_name": "triggered",
+                    "level_values": [{"level_synonym": ["triggered"], "lang": "en"}],
+                },
+            ],
+            "ordered": False,
+        }
+    }
+
+    assert trt.query_attributes() == {"isArmed": False}
+
+    assert trt.can_execute(trait.COMMAND_ARMDISARM, {"arm": False})
+
+    calls = async_mock_service(
+        hass, alarm_control_panel.DOMAIN, alarm_control_panel.SERVICE_ALARM_DISARM
+    )
+
+    # Test without secure_pin configured
+    with pytest.raises(error.SmartHomeError) as err:
+        trt = trait.ArmDisArmTrait(
+            hass,
+            State(
+                "alarm_control_panel.alarm",
+                STATE_ALARM_ARMED_AWAY,
+                {alarm_control_panel.ATTR_CODE_ARM_REQUIRED: True},
+            ),
+            BASIC_CONFIG,
+        )
+        await trt.execute(trait.COMMAND_ARMDISARM, BASIC_DATA, {"arm": False}, {})
+
+    assert len(calls) == 0
+    assert err.value.code == const.ERR_CHALLENGE_NOT_SETUP
+
+    trt = trait.ArmDisArmTrait(
+        hass,
+        State(
+            "alarm_control_panel.alarm",
+            STATE_ALARM_ARMED_AWAY,
+            {alarm_control_panel.ATTR_CODE_ARM_REQUIRED: True},
+        ),
+        PIN_CONFIG,
+    )
+
+    # No challenge data
+    with pytest.raises(error.ChallengeNeeded) as err:
+        await trt.execute(trait.COMMAND_ARMDISARM, PIN_DATA, {"arm": False}, {})
+    assert len(calls) == 0
+    assert err.value.code == const.ERR_CHALLENGE_NEEDED
+    assert err.value.challenge_type == const.CHALLENGE_PIN_NEEDED
+
+    # invalid pin
+    with pytest.raises(error.ChallengeNeeded) as err:
+        await trt.execute(
+            trait.COMMAND_ARMDISARM, PIN_DATA, {"arm": False}, {"pin": 9999}
+        )
+    assert len(calls) == 0
+    assert err.value.code == const.ERR_CHALLENGE_NEEDED
+    assert err.value.challenge_type == const.CHALLENGE_FAILED_PIN_NEEDED
+
+    # correct pin
+    await trt.execute(
+        trait.COMMAND_ARMDISARM, PIN_DATA, {"arm": False}, {"pin": "1234"}
+    )
+
+    assert len(calls) == 1
+
+    # Test already disarmed
+    with pytest.raises(error.SmartHomeError) as err:
+        trt = trait.ArmDisArmTrait(
+            hass,
+            State(
+                "alarm_control_panel.alarm",
+                STATE_ALARM_DISARMED,
+                {alarm_control_panel.ATTR_CODE_ARM_REQUIRED: True},
+            ),
+            PIN_CONFIG,
+        )
+        await trt.execute(trait.COMMAND_ARMDISARM, PIN_DATA, {"arm": False}, {})
+    assert len(calls) == 1
+    assert err.value.code == const.ERR_ALREADY_DISARMED
+
+    # Cancel arming after already armed will require pin
+    with pytest.raises(error.SmartHomeError) as err:
+        trt = trait.ArmDisArmTrait(
+            hass,
+            State(
+                "alarm_control_panel.alarm",
+                STATE_ALARM_ARMED_AWAY,
+                {alarm_control_panel.ATTR_CODE_ARM_REQUIRED: False},
+            ),
+            PIN_CONFIG,
+        )
+        await trt.execute(
+            trait.COMMAND_ARMDISARM, PIN_DATA, {"arm": True, "cancel": True}, {}
+        )
+    assert len(calls) == 1
+    assert err.value.code == const.ERR_CHALLENGE_NEEDED
+    assert err.value.challenge_type == const.CHALLENGE_PIN_NEEDED
+
+    # Cancel arming while pending to arm doesn't require pin
+    trt = trait.ArmDisArmTrait(
+        hass,
+        State(
+            "alarm_control_panel.alarm",
+            STATE_ALARM_PENDING,
+            {alarm_control_panel.ATTR_CODE_ARM_REQUIRED: False},
+        ),
+        PIN_CONFIG,
+    )
+    await trt.execute(
+        trait.COMMAND_ARMDISARM, PIN_DATA, {"arm": True, "cancel": True}, {}
+    )
+    assert len(calls) == 2
+
+
 async def test_fan_speed(hass):
     """Test FanSpeed trait speed control support for fan domain."""
     assert helpers.get_google_type(fan.DOMAIN, None) is not None