262 lines
9.1 KiB
Python
262 lines
9.1 KiB
Python
"""
|
|
Alexa Devices Alarm Control Panel using Guard Mode.
|
|
|
|
SPDX-License-Identifier: Apache-2.0
|
|
|
|
For more details about this platform, please refer to the documentation at
|
|
https://community.home-assistant.io/t/echo-devices-alexa-as-media-player-testers-needed/58639
|
|
"""
|
|
|
|
from asyncio import sleep
|
|
import logging
|
|
from typing import Optional
|
|
|
|
from alexapy import hide_email, hide_serial
|
|
from homeassistant.components.alarm_control_panel import AlarmControlPanelEntity
|
|
from homeassistant.const import CONF_EMAIL, STATE_UNAVAILABLE
|
|
from homeassistant.exceptions import ConfigEntryNotReady
|
|
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
|
|
|
from .alexa_entity import parse_guard_state_from_coordinator
|
|
from .alexa_media import AlexaMedia
|
|
from .const import (
|
|
CONF_EXCLUDE_DEVICES,
|
|
CONF_INCLUDE_DEVICES,
|
|
CONF_QUEUE_DELAY,
|
|
DATA_ALEXAMEDIA,
|
|
DEFAULT_QUEUE_DELAY,
|
|
DOMAIN as ALEXA_DOMAIN,
|
|
)
|
|
from .helpers import _catch_login_errors, add_devices, safe_get
|
|
|
|
try:
|
|
from homeassistant.components.alarm_control_panel import AlarmControlPanelState
|
|
|
|
STATE_ALARM_ARMED_AWAY = AlarmControlPanelState.ARMED_AWAY
|
|
STATE_ALARM_DISARMED = AlarmControlPanelState.DISARMED
|
|
except ImportError:
|
|
from homeassistant.const import STATE_ALARM_ARMED_AWAY, STATE_ALARM_DISARMED
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
DEPENDENCIES = [ALEXA_DOMAIN]
|
|
|
|
|
|
async def async_setup_platform(
|
|
hass, config, add_devices_callback, discovery_info=None
|
|
) -> bool:
|
|
"""Set up the Alexa alarm control panel platform."""
|
|
devices: list[AlexaAlarmControlPanel] = []
|
|
account = None
|
|
if config:
|
|
account = config.get(CONF_EMAIL)
|
|
if account is None and discovery_info:
|
|
account = safe_get(discovery_info, ["config", CONF_EMAIL])
|
|
if account is None:
|
|
raise ConfigEntryNotReady
|
|
include_filter = config.get(CONF_INCLUDE_DEVICES, [])
|
|
exclude_filter = config.get(CONF_EXCLUDE_DEVICES, [])
|
|
account_dict = hass.data[DATA_ALEXAMEDIA]["accounts"][account]
|
|
guard_media_players = {}
|
|
for key, device in account_dict["devices"]["media_player"].items():
|
|
if key not in account_dict["entities"]["media_player"]:
|
|
_LOGGER.debug(
|
|
"%s: Media player %s not loaded yet; delaying load",
|
|
hide_email(account),
|
|
hide_serial(key),
|
|
)
|
|
raise ConfigEntryNotReady
|
|
if "GUARD_EARCON" in device["capabilities"]:
|
|
guard_media_players[key] = account_dict["entities"]["media_player"][key]
|
|
if "alarm_control_panel" not in (account_dict["entities"]):
|
|
(
|
|
hass.data[DATA_ALEXAMEDIA]["accounts"][account]["entities"][
|
|
"alarm_control_panel"
|
|
]
|
|
) = {}
|
|
alexa_client: Optional[AlexaAlarmControlPanel] = None
|
|
guard_entities = safe_get(account_dict, ["devices", "guard"], [])
|
|
if guard_entities:
|
|
alexa_client = AlexaAlarmControlPanel(
|
|
account_dict["login_obj"],
|
|
account_dict["coordinator"],
|
|
guard_entities[0],
|
|
guard_media_players,
|
|
)
|
|
else:
|
|
_LOGGER.debug("%s: No Alexa Guard entity found", hide_email(account))
|
|
if not (alexa_client and alexa_client.unique_id):
|
|
_LOGGER.debug(
|
|
"%s: Skipping creation of uninitialized device: %s",
|
|
hide_email(account),
|
|
alexa_client,
|
|
)
|
|
elif alexa_client.unique_id not in (
|
|
account_dict["entities"]["alarm_control_panel"]
|
|
):
|
|
devices.append(alexa_client)
|
|
(
|
|
hass.data[DATA_ALEXAMEDIA]["accounts"][account]["entities"][
|
|
"alarm_control_panel"
|
|
][alexa_client.unique_id]
|
|
) = alexa_client
|
|
else:
|
|
_LOGGER.debug(
|
|
"%s: Skipping already added device: %s", hide_email(account), alexa_client
|
|
)
|
|
return await add_devices(
|
|
hide_email(account),
|
|
devices,
|
|
add_devices_callback,
|
|
include_filter,
|
|
exclude_filter,
|
|
)
|
|
|
|
|
|
async def async_setup_entry(hass, config_entry, async_add_devices):
|
|
"""Set up the Alexa alarm control panel platform by config_entry."""
|
|
return await async_setup_platform(
|
|
hass, config_entry.data, async_add_devices, discovery_info=None
|
|
)
|
|
|
|
|
|
async def async_unload_entry(hass, entry) -> bool:
|
|
"""Unload a config entry."""
|
|
account = entry.data[CONF_EMAIL]
|
|
_LOGGER.debug("Attempting to unload alarm control panel")
|
|
account_dict = hass.data[DATA_ALEXAMEDIA]["accounts"][account]
|
|
for device in account_dict["entities"]["alarm_control_panel"].values():
|
|
_LOGGER.debug("Removing %s", device)
|
|
await device.async_remove()
|
|
return True
|
|
|
|
|
|
class AlexaAlarmControlPanel(AlarmControlPanelEntity, AlexaMedia, CoordinatorEntity):
|
|
"""Implementation of Alexa Media Player alarm control panel."""
|
|
|
|
def __init__(self, login, coordinator, guard_entity, media_players=None) -> None:
|
|
"""Initialize the Alexa device."""
|
|
AlexaMedia.__init__(self, None, login)
|
|
CoordinatorEntity.__init__(self, coordinator)
|
|
_LOGGER.debug("%s: Initiating alarm control panel", hide_email(login.email))
|
|
# AlexaAPI requires a AlexaClient object, need to clean this up
|
|
|
|
# Guard info
|
|
self._appliance_id = guard_entity["appliance_id"]
|
|
self._guard_entity_id = guard_entity["id"]
|
|
self._friendly_name = "Alexa Guard " + self._appliance_id[-5:]
|
|
self._media_players = {} or media_players
|
|
self._attrs: dict[str, str] = {}
|
|
_LOGGER.debug(
|
|
"%s: Guard Discovered %s: %s %s",
|
|
self.account,
|
|
self._friendly_name,
|
|
hide_serial(self._appliance_id),
|
|
hide_serial(self._guard_entity_id),
|
|
)
|
|
|
|
@_catch_login_errors
|
|
async def _async_alarm_set(
|
|
self,
|
|
command: str = "",
|
|
code=None, # pylint: disable=unused-argument
|
|
) -> None:
|
|
"""Send command."""
|
|
try:
|
|
if not self.enabled:
|
|
return
|
|
except AttributeError:
|
|
pass
|
|
if command not in (STATE_ALARM_ARMED_AWAY, STATE_ALARM_DISARMED):
|
|
_LOGGER.error("Invalid command: %s", command)
|
|
return
|
|
command_map = {STATE_ALARM_ARMED_AWAY: "AWAY", STATE_ALARM_DISARMED: "HOME"}
|
|
available_media_players = list(
|
|
filter(lambda x: x.state != STATE_UNAVAILABLE, self._media_players.values())
|
|
)
|
|
if available_media_players:
|
|
_LOGGER.debug("Sending guard command to: %s", available_media_players[0])
|
|
available_media_players[0].check_login_changes()
|
|
# Extract appliance ID safely to prevent IndexError if format is unexpected
|
|
appliance_parts = self._appliance_id.split("_")
|
|
appliance_id = (
|
|
appliance_parts[2] if len(appliance_parts) > 2 else self._appliance_id
|
|
)
|
|
await available_media_players[0].alexa_api.set_guard_state(
|
|
appliance_id,
|
|
command_map[command],
|
|
queue_delay=self.hass.data[DATA_ALEXAMEDIA]["accounts"][self.email][
|
|
"options"
|
|
].get(CONF_QUEUE_DELAY, DEFAULT_QUEUE_DELAY),
|
|
)
|
|
await sleep(2) # delay
|
|
else:
|
|
_LOGGER.debug("Performing static guard command")
|
|
await self.alexa_api.static_set_guard_state(
|
|
self._login, self._guard_entity_id, command
|
|
)
|
|
await self.coordinator.async_request_refresh()
|
|
|
|
async def async_alarm_disarm(
|
|
self,
|
|
code=None, # pylint:disable=unused-argument
|
|
) -> None:
|
|
"""Send disarm command."""
|
|
await self._async_alarm_set(STATE_ALARM_DISARMED)
|
|
|
|
async def async_alarm_arm_away(
|
|
self,
|
|
code=None, # pylint:disable=unused-argument
|
|
) -> None:
|
|
"""Send arm away command."""
|
|
await self._async_alarm_set(STATE_ALARM_ARMED_AWAY)
|
|
|
|
@property
|
|
def unique_id(self):
|
|
"""Return the unique ID."""
|
|
return self._guard_entity_id
|
|
|
|
@property
|
|
def name(self):
|
|
"""Return the name of the device."""
|
|
return self._friendly_name
|
|
|
|
@property
|
|
def state(self):
|
|
"""Return the state of the device."""
|
|
_state = parse_guard_state_from_coordinator(
|
|
self.coordinator, self._guard_entity_id
|
|
)
|
|
if _state == "ARMED_AWAY":
|
|
return STATE_ALARM_ARMED_AWAY
|
|
return STATE_ALARM_DISARMED
|
|
|
|
@property
|
|
def supported_features(self) -> int:
|
|
"""Return the list of supported features."""
|
|
# pylint: disable=import-outside-toplevel
|
|
try:
|
|
from homeassistant.components.alarm_control_panel import (
|
|
AlarmControlPanelEntityFeature,
|
|
)
|
|
except ImportError:
|
|
return 0
|
|
return AlarmControlPanelEntityFeature.ARM_AWAY
|
|
|
|
@property
|
|
def assumed_state(self) -> bool:
|
|
"""Return assumed state.
|
|
|
|
Returns
|
|
bool: Whether the state is assumed
|
|
|
|
"""
|
|
last_refresh_success = (
|
|
self.coordinator.data and self._guard_entity_id in self.coordinator.data
|
|
)
|
|
return not last_refresh_success
|
|
|
|
@property
|
|
def extra_state_attributes(self):
|
|
"""Return the state attributes."""
|
|
return self._attrs
|