mirror of
https://github.com/home-assistant/core
synced 2024-08-31 05:57:13 +02:00
Add alarm control panel support to ZHA (#49080)
* start implementation of IAS ACE * starting alarm control panel * enums * use new enums from zigpy * fix import * write state * fix registries after rebase * remove extra line * cleanup * fix deprecation warning * updates to catch up with codebase evolution * minor updates * cleanup * implement more ias ace functionality * cleanup * make config helper work for supplied section * connect to configuration * use ha async_create_task * add some tests * remove unused restore method * update tests * add tests from panel POV * dynamically include alarm control panel config * fix import Co-authored-by: Alexei Chetroi <lexoid@gmail.com>
This commit is contained in:
David F. Mulcahey
GitHub
parent
d4ed65e0f5
commit
a644c2e8ba
174
homeassistant/components/zha/alarm_control_panel.py
Normal file
174
homeassistant/components/zha/alarm_control_panel.py
Normal file
@ -0,0 +1,174 @@
|
||||
"""Alarm control panels on Zigbee Home Automation networks."""
|
||||
import functools
|
||||
import logging
|
||||
|
||||
from zigpy.zcl.clusters.security import IasAce
|
||||
|
||||
from homeassistant.components.alarm_control_panel import (
|
||||
ATTR_CHANGED_BY,
|
||||
ATTR_CODE_ARM_REQUIRED,
|
||||
ATTR_CODE_FORMAT,
|
||||
DOMAIN,
|
||||
FORMAT_TEXT,
|
||||
SUPPORT_ALARM_ARM_AWAY,
|
||||
SUPPORT_ALARM_ARM_HOME,
|
||||
SUPPORT_ALARM_ARM_NIGHT,
|
||||
SUPPORT_ALARM_TRIGGER,
|
||||
AlarmControlPanelEntity,
|
||||
)
|
||||
from homeassistant.components.zha.core.typing import ZhaDeviceType
|
||||
from homeassistant.const import (
|
||||
STATE_ALARM_ARMED_AWAY,
|
||||
STATE_ALARM_ARMED_HOME,
|
||||
STATE_ALARM_ARMED_NIGHT,
|
||||
STATE_ALARM_DISARMED,
|
||||
STATE_ALARM_TRIGGERED,
|
||||
)
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||
|
||||
from .core import discovery
|
||||
from .core.channels.security import (
|
||||
SIGNAL_ALARM_TRIGGERED,
|
||||
SIGNAL_ARMED_STATE_CHANGED,
|
||||
IasAce as AceChannel,
|
||||
)
|
||||
from .core.const import (
|
||||
CHANNEL_IAS_ACE,
|
||||
CONF_ALARM_ARM_REQUIRES_CODE,
|
||||
CONF_ALARM_FAILED_TRIES,
|
||||
CONF_ALARM_MASTER_CODE,
|
||||
DATA_ZHA,
|
||||
DATA_ZHA_DISPATCHERS,
|
||||
SIGNAL_ADD_ENTITIES,
|
||||
ZHA_ALARM_OPTIONS,
|
||||
)
|
||||
from .core.helpers import async_get_zha_config_value
|
||||
from .core.registries import ZHA_ENTITIES
|
||||
from .entity import ZhaEntity
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
STRICT_MATCH = functools.partial(ZHA_ENTITIES.strict_match, DOMAIN)
|
||||
|
||||
IAS_ACE_STATE_MAP = {
|
||||
IasAce.PanelStatus.Panel_Disarmed: STATE_ALARM_DISARMED,
|
||||
IasAce.PanelStatus.Armed_Stay: STATE_ALARM_ARMED_HOME,
|
||||
IasAce.PanelStatus.Armed_Night: STATE_ALARM_ARMED_NIGHT,
|
||||
IasAce.PanelStatus.Armed_Away: STATE_ALARM_ARMED_AWAY,
|
||||
IasAce.PanelStatus.In_Alarm: STATE_ALARM_TRIGGERED,
|
||||
}
|
||||
|
||||
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Set up the Zigbee Home Automation alarm control panel from config entry."""
|
||||
entities_to_create = hass.data[DATA_ZHA][DOMAIN]
|
||||
|
||||
unsub = async_dispatcher_connect(
|
||||
hass,
|
||||
SIGNAL_ADD_ENTITIES,
|
||||
functools.partial(
|
||||
discovery.async_add_entities, async_add_entities, entities_to_create
|
||||
),
|
||||
)
|
||||
hass.data[DATA_ZHA][DATA_ZHA_DISPATCHERS].append(unsub)
|
||||
|
||||
|
||||
@STRICT_MATCH(channel_names=CHANNEL_IAS_ACE)
|
||||
class ZHAAlarmControlPanel(ZhaEntity, AlarmControlPanelEntity):
|
||||
"""Entity for ZHA alarm control devices."""
|
||||
|
||||
def __init__(self, unique_id, zha_device: ZhaDeviceType, channels, **kwargs):
|
||||
"""Initialize the ZHA alarm control device."""
|
||||
super().__init__(unique_id, zha_device, channels, **kwargs)
|
||||
cfg_entry = zha_device.gateway.config_entry
|
||||
self._channel: AceChannel = channels[0]
|
||||
self._channel.panel_code = async_get_zha_config_value(
|
||||
cfg_entry, ZHA_ALARM_OPTIONS, CONF_ALARM_MASTER_CODE, "1234"
|
||||
)
|
||||
self._channel.code_required_arm_actions = async_get_zha_config_value(
|
||||
cfg_entry, ZHA_ALARM_OPTIONS, CONF_ALARM_ARM_REQUIRES_CODE, False
|
||||
)
|
||||
self._channel.max_invalid_tries = async_get_zha_config_value(
|
||||
cfg_entry, ZHA_ALARM_OPTIONS, CONF_ALARM_FAILED_TRIES, 3
|
||||
)
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
"""Run when about to be added to hass."""
|
||||
await super().async_added_to_hass()
|
||||
self.async_accept_signal(
|
||||
self._channel, SIGNAL_ARMED_STATE_CHANGED, self.async_set_armed_mode
|
||||
)
|
||||
self.async_accept_signal(
|
||||
self._channel, SIGNAL_ALARM_TRIGGERED, self.async_alarm_trigger
|
||||
)
|
||||
|
||||
@callback
|
||||
def async_set_armed_mode(self) -> None:
|
||||
"""Set the entity state."""
|
||||
self.async_write_ha_state()
|
||||
|
||||
@property
|
||||
def code_format(self):
|
||||
"""Regex for code format or None if no code is required."""
|
||||
return FORMAT_TEXT
|
||||
|
||||
@property
|
||||
def changed_by(self):
|
||||
"""Last change triggered by."""
|
||||
return None
|
||||
|
||||
@property
|
||||
def code_arm_required(self):
|
||||
"""Whether the code is required for arm actions."""
|
||||
return self._channel.code_required_arm_actions
|
||||
|
||||
async def async_alarm_disarm(self, code=None):
|
||||
"""Send disarm command."""
|
||||
self._channel.arm(IasAce.ArmMode.Disarm, code, 0)
|
||||
self.async_write_ha_state()
|
||||
|
||||
async def async_alarm_arm_home(self, code=None):
|
||||
"""Send arm home command."""
|
||||
self._channel.arm(IasAce.ArmMode.Arm_Day_Home_Only, code, 0)
|
||||
self.async_write_ha_state()
|
||||
|
||||
async def async_alarm_arm_away(self, code=None):
|
||||
"""Send arm away command."""
|
||||
self._channel.arm(IasAce.ArmMode.Arm_All_Zones, code, 0)
|
||||
self.async_write_ha_state()
|
||||
|
||||
async def async_alarm_arm_night(self, code=None):
|
||||
"""Send arm night command."""
|
||||
self._channel.arm(IasAce.ArmMode.Arm_Night_Sleep_Only, code, 0)
|
||||
self.async_write_ha_state()
|
||||
|
||||
async def async_alarm_trigger(self, code=None):
|
||||
"""Send alarm trigger command."""
|
||||
self.async_write_ha_state()
|
||||
|
||||
@property
|
||||
def supported_features(self) -> int:
|
||||
"""Return the list of supported features."""
|
||||
return (
|
||||
SUPPORT_ALARM_ARM_HOME
|
||||
| SUPPORT_ALARM_ARM_AWAY
|
||||
| SUPPORT_ALARM_ARM_NIGHT
|
||||
| SUPPORT_ALARM_TRIGGER
|
||||
)
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""Return the state of the entity."""
|
||||
return IAS_ACE_STATE_MAP.get(self._channel.armed_state)
|
||||
|
||||
@property
|
||||
def state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
state_attr = {
|
||||
ATTR_CODE_FORMAT: self.code_format,
|
||||
ATTR_CHANGED_BY: self.changed_by,
|
||||
ATTR_CODE_ARM_REQUIRED: self.code_arm_required,
|
||||
}
|
||||
return state_attr
|
||||
@ -9,6 +9,7 @@ from typing import Any
|
||||
import voluptuous as vol
|
||||
from zigpy.config.validators import cv_boolean
|
||||
from zigpy.types.named import EUI64
|
||||
from zigpy.zcl.clusters.security import IasAce
|
||||
import zigpy.zdo.types as zdo_types
|
||||
|
||||
from homeassistant.components import websocket_api
|
||||
@ -54,11 +55,13 @@ from .core.const import (
|
||||
WARNING_DEVICE_SQUAWK_MODE_ARMED,
|
||||
WARNING_DEVICE_STROBE_HIGH,
|
||||
WARNING_DEVICE_STROBE_YES,
|
||||
ZHA_ALARM_OPTIONS,
|
||||
ZHA_CHANNEL_MSG,
|
||||
ZHA_CONFIG_SCHEMAS,
|
||||
)
|
||||
from .core.group import GroupMember
|
||||
from .core.helpers import (
|
||||
async_input_cluster_exists,
|
||||
async_is_bindable_target,
|
||||
convert_install_code,
|
||||
get_matched_clusters,
|
||||
@ -894,6 +897,10 @@ async def websocket_get_configuration(hass, connection, msg):
|
||||
|
||||
data = {"schemas": {}, "data": {}}
|
||||
for section, schema in ZHA_CONFIG_SCHEMAS.items():
|
||||
if section == ZHA_ALARM_OPTIONS and not async_input_cluster_exists(
|
||||
hass, IasAce.cluster_id
|
||||
):
|
||||
continue
|
||||
data["schemas"][section] = voluptuous_serialize.convert(
|
||||
schema, custom_serializer=custom_serializer
|
||||
)
|
||||
|
||||
@ -8,13 +8,15 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from collections.abc import Coroutine
|
||||
import logging
|
||||
|
||||
from zigpy.exceptions import ZigbeeException
|
||||
import zigpy.zcl.clusters.security as security
|
||||
from zigpy.zcl.clusters.security import IasAce as AceCluster
|
||||
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.core import CALLABLE_T, callback
|
||||
|
||||
from .. import registries
|
||||
from .. import registries, typing as zha_typing
|
||||
from ..const import (
|
||||
SIGNAL_ATTR_UPDATED,
|
||||
WARNING_DEVICE_MODE_EMERGENCY,
|
||||
@ -25,11 +27,238 @@ from ..const import (
|
||||
)
|
||||
from .base import ChannelStatus, ZigbeeChannel
|
||||
|
||||
IAS_ACE_ARM = 0x0000 # ("arm", (t.enum8, t.CharacterString, t.uint8_t), False),
|
||||
IAS_ACE_BYPASS = 0x0001 # ("bypass", (t.LVList(t.uint8_t), t.CharacterString), False),
|
||||
IAS_ACE_EMERGENCY = 0x0002 # ("emergency", (), False),
|
||||
IAS_ACE_FIRE = 0x0003 # ("fire", (), False),
|
||||
IAS_ACE_PANIC = 0x0004 # ("panic", (), False),
|
||||
IAS_ACE_GET_ZONE_ID_MAP = 0x0005 # ("get_zone_id_map", (), False),
|
||||
IAS_ACE_GET_ZONE_INFO = 0x0006 # ("get_zone_info", (t.uint8_t,), False),
|
||||
IAS_ACE_GET_PANEL_STATUS = 0x0007 # ("get_panel_status", (), False),
|
||||
IAS_ACE_GET_BYPASSED_ZONE_LIST = 0x0008 # ("get_bypassed_zone_list", (), False),
|
||||
IAS_ACE_GET_ZONE_STATUS = (
|
||||
0x0009 # ("get_zone_status", (t.uint8_t, t.uint8_t, t.Bool, t.bitmap16), False)
|
||||
)
|
||||
NAME = 0
|
||||
SIGNAL_ARMED_STATE_CHANGED = "zha_armed_state_changed"
|
||||
SIGNAL_ALARM_TRIGGERED = "zha_armed_triggered"
|
||||
|
||||
@registries.ZIGBEE_CHANNEL_REGISTRY.register(security.IasAce.cluster_id)
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@registries.ZIGBEE_CHANNEL_REGISTRY.register(AceCluster.cluster_id)
|
||||
class IasAce(ZigbeeChannel):
|
||||
"""IAS Ancillary Control Equipment channel."""
|
||||
|
||||
def __init__(
|
||||
self, cluster: zha_typing.ZigpyClusterType, ch_pool: zha_typing.ChannelPoolType
|
||||
) -> None:
|
||||
"""Initialize IAS Ancillary Control Equipment channel."""
|
||||
super().__init__(cluster, ch_pool)
|
||||
self.command_map: dict[int, CALLABLE_T] = {
|
||||
IAS_ACE_ARM: self.arm,
|
||||
IAS_ACE_BYPASS: self._bypass,
|
||||
IAS_ACE_EMERGENCY: self._emergency,
|
||||
IAS_ACE_FIRE: self._fire,
|
||||
IAS_ACE_PANIC: self._panic,
|
||||
IAS_ACE_GET_ZONE_ID_MAP: self._get_zone_id_map,
|
||||
IAS_ACE_GET_ZONE_INFO: self._get_zone_info,
|
||||
IAS_ACE_GET_PANEL_STATUS: self._send_panel_status_response,
|
||||
IAS_ACE_GET_BYPASSED_ZONE_LIST: self._get_bypassed_zone_list,
|
||||
IAS_ACE_GET_ZONE_STATUS: self._get_zone_status,
|
||||
}
|
||||
self.arm_map: dict[AceCluster.ArmMode, CALLABLE_T] = {
|
||||
AceCluster.ArmMode.Disarm: self._disarm,
|
||||
AceCluster.ArmMode.Arm_All_Zones: self._arm_away,
|
||||
AceCluster.ArmMode.Arm_Day_Home_Only: self._arm_day,
|
||||
AceCluster.ArmMode.Arm_Night_Sleep_Only: self._arm_night,
|
||||
}
|
||||
self.armed_state: AceCluster.PanelStatus = AceCluster.PanelStatus.Panel_Disarmed
|
||||
self.invalid_tries: int = 0
|
||||
|
||||
# These will all be setup by the entity from zha configuration
|
||||
self.panel_code: str = "1234"
|
||||
self.code_required_arm_actions = False
|
||||
self.max_invalid_tries: int = 3
|
||||
|
||||
# where do we store this to handle restarts
|
||||
self.alarm_status: AceCluster.AlarmStatus = AceCluster.AlarmStatus.No_Alarm
|
||||
|
||||
@callback
|
||||
def cluster_command(self, tsn, command_id, args) -> None:
|
||||
"""Handle commands received to this cluster."""
|
||||
self.warning(
|
||||
"received command %s", self._cluster.server_commands.get(command_id)[NAME]
|
||||
)
|
||||
self.command_map[command_id](*args)
|
||||
|
||||
def arm(self, arm_mode: int, code: str, zone_id: int):
|
||||
"""Handle the IAS ACE arm command."""
|
||||
mode = AceCluster.ArmMode(arm_mode)
|
||||
|
||||
self.zha_send_event(
|
||||
self._cluster.server_commands.get(IAS_ACE_ARM)[NAME],
|
||||
{
|
||||
"arm_mode": mode.value,
|
||||
"arm_mode_description": mode.name,
|
||||
"code": code,
|
||||
"zone_id": zone_id,
|
||||
},
|
||||
)
|
||||
|
||||
zigbee_reply = self.arm_map[mode](code)
|
||||
self._ch_pool.hass.async_create_task(zigbee_reply)
|
||||
|
||||
if self.invalid_tries >= self.max_invalid_tries:
|
||||
self.alarm_status = AceCluster.AlarmStatus.Emergency
|
||||
self.armed_state = AceCluster.PanelStatus.In_Alarm
|
||||
self.async_send_signal(f"{self.unique_id}_{SIGNAL_ALARM_TRIGGERED}")
|
||||
else:
|
||||
self.async_send_signal(f"{self.unique_id}_{SIGNAL_ARMED_STATE_CHANGED}")
|
||||
self._send_panel_status_changed()
|
||||
|
||||
def _disarm(self, code: str):
|
||||
"""Test the code and disarm the panel if the code is correct."""
|
||||
if (
|
||||
code != self.panel_code
|
||||
and self.armed_state != AceCluster.PanelStatus.Panel_Disarmed
|
||||
):
|
||||
self.warning("Invalid code supplied to IAS ACE")
|
||||
self.invalid_tries += 1
|
||||
zigbee_reply = self.arm_response(
|
||||
AceCluster.ArmNotification.Invalid_Arm_Disarm_Code
|
||||
)
|
||||
else:
|
||||
self.invalid_tries = 0
|
||||
if (
|
||||
self.armed_state == AceCluster.PanelStatus.Panel_Disarmed
|
||||
and self.alarm_status == AceCluster.AlarmStatus.No_Alarm
|
||||
):
|
||||
self.warning("IAS ACE already disarmed")
|
||||
zigbee_reply = self.arm_response(
|
||||
AceCluster.ArmNotification.Already_Disarmed
|
||||
)
|
||||
else:
|
||||
self.warning("Disarming all IAS ACE zones")
|
||||
zigbee_reply = self.arm_response(
|
||||
AceCluster.ArmNotification.All_Zones_Disarmed
|
||||
)
|
||||
|
||||
self.armed_state = AceCluster.PanelStatus.Panel_Disarmed
|
||||
self.alarm_status = AceCluster.AlarmStatus.No_Alarm
|
||||
return zigbee_reply
|
||||
|
||||
def _arm_day(self, code: str) -> None:
|
||||
"""Arm the panel for day / home zones."""
|
||||
return self._handle_arm(
|
||||
code,
|
||||
AceCluster.PanelStatus.Armed_Stay,
|
||||
AceCluster.ArmNotification.Only_Day_Home_Zones_Armed,
|
||||
)
|
||||
|
||||
def _arm_night(self, code: str) -> None:
|
||||
"""Arm the panel for night / sleep zones."""
|
||||
return self._handle_arm(
|
||||
code,
|
||||
AceCluster.PanelStatus.Armed_Night,
|
||||
AceCluster.ArmNotification.Only_Night_Sleep_Zones_Armed,
|
||||
)
|
||||
|
||||
def _arm_away(self, code: str) -> None:
|
||||
"""Arm the panel for away mode."""
|
||||
return self._handle_arm(
|
||||
code,
|
||||
AceCluster.PanelStatus.Armed_Away,
|
||||
AceCluster.ArmNotification.All_Zones_Armed,
|
||||
)
|
||||
|
||||
def _handle_arm(
|
||||
self,
|
||||
code: str,
|
||||
panel_status: AceCluster.PanelStatus,
|
||||
armed_type: AceCluster.ArmNotification,
|
||||
) -> None:
|
||||
"""Arm the panel with the specified statuses."""
|
||||
if self.code_required_arm_actions and code != self.panel_code:
|
||||
self.warning("Invalid code supplied to IAS ACE")
|
||||
zigbee_reply = self.arm_response(
|
||||
AceCluster.ArmNotification.Invalid_Arm_Disarm_Code
|
||||
)
|
||||
else:
|
||||
self.warning("Arming all IAS ACE zones")
|
||||
self.armed_state = panel_status
|
||||
zigbee_reply = self.arm_response(armed_type)
|
||||
return zigbee_reply
|
||||
|
||||
def _bypass(self, zone_list, code) -> None:
|
||||
"""Handle the IAS ACE bypass command."""
|
||||
self.zha_send_event(
|
||||
self._cluster.server_commands.get(IAS_ACE_BYPASS)[NAME],
|
||||
{"zone_list": zone_list, "code": code},
|
||||
)
|
||||
|
||||
def _emergency(self) -> None:
|
||||
"""Handle the IAS ACE emergency command."""
|
||||
self._set_alarm(
|
||||
AceCluster.AlarmStatus.Emergency,
|
||||
IAS_ACE_EMERGENCY,
|
||||
)
|
||||
|
||||
def _fire(self) -> None:
|
||||
"""Handle the IAS ACE fire command."""
|
||||
self._set_alarm(
|
||||
AceCluster.AlarmStatus.Fire,
|
||||
IAS_ACE_FIRE,
|
||||
)
|
||||
|
||||
def _panic(self) -> None:
|
||||
"""Handle the IAS ACE panic command."""
|
||||
self._set_alarm(
|
||||
AceCluster.AlarmStatus.Emergency_Panic,
|
||||
IAS_ACE_PANIC,
|
||||
)
|
||||
|
||||
def _set_alarm(self, status: AceCluster.PanelStatus, event: str) -> None:
|
||||
"""Set the specified alarm status."""
|
||||
self.alarm_status = status
|
||||
self.armed_state = AceCluster.PanelStatus.In_Alarm
|
||||
self.async_send_signal(f"{self.unique_id}_{SIGNAL_ALARM_TRIGGERED}")
|
||||
self._send_panel_status_changed()
|
||||
|
||||
def _get_zone_id_map(self):
|
||||
"""Handle the IAS ACE zone id map command."""
|
||||
|
||||
def _get_zone_info(self, zone_id):
|
||||
"""Handle the IAS ACE zone info command."""
|
||||
|
||||
def _send_panel_status_response(self) -> None:
|
||||
"""Handle the IAS ACE panel status response command."""
|
||||
response = self.panel_status_response(
|
||||
self.armed_state,
|
||||
0x00,
|
||||
AceCluster.AudibleNotification.Default_Sound,
|
||||
self.alarm_status,
|
||||
)
|
||||
self._ch_pool.hass.async_create_task(response)
|
||||
|
||||
def _send_panel_status_changed(self) -> None:
|
||||
"""Handle the IAS ACE panel status changed command."""
|
||||
response = self.panel_status_changed(
|
||||
self.armed_state,
|
||||
0x00,
|
||||
AceCluster.AudibleNotification.Default_Sound,
|
||||
self.alarm_status,
|
||||
)
|
||||
self._ch_pool.hass.async_create_task(response)
|
||||
|
||||
def _get_bypassed_zone_list(self):
|
||||
"""Handle the IAS ACE bypassed zone list command."""
|
||||
|
||||
def _get_zone_status(
|
||||
self, starting_zone_id, max_zone_ids, zone_status_mask_flag, zone_status_mask
|
||||
):
|
||||
"""Handle the IAS ACE zone status command."""
|
||||
|
||||
|
||||
@registries.CHANNEL_ONLY_CLUSTERS.register(security.IasWd.cluster_id)
|
||||
@registries.ZIGBEE_CHANNEL_REGISTRY.register(security.IasWd.cluster_id)
|
||||
|
||||
@ -13,6 +13,7 @@ import zigpy_xbee.zigbee.application
|
||||
import zigpy_zigate.zigbee.application
|
||||
import zigpy_znp.zigbee.application
|
||||
|
||||
from homeassistant.components.alarm_control_panel import DOMAIN as ALARM
|
||||
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR
|
||||
from homeassistant.components.climate import DOMAIN as CLIMATE
|
||||
from homeassistant.components.cover import DOMAIN as COVER
|
||||
@ -83,6 +84,7 @@ CHANNEL_ELECTRICAL_MEASUREMENT = "electrical_measurement"
|
||||
CHANNEL_EVENT_RELAY = "event_relay"
|
||||
CHANNEL_FAN = "fan"
|
||||
CHANNEL_HUMIDITY = "humidity"
|
||||
CHANNEL_IAS_ACE = "ias_ace"
|
||||
CHANNEL_IAS_WD = "ias_wd"
|
||||
CHANNEL_IDENTIFY = "identify"
|
||||
CHANNEL_ILLUMINANCE = "illuminance"
|
||||
@ -106,6 +108,7 @@ CLUSTER_TYPE_IN = "in"
|
||||
CLUSTER_TYPE_OUT = "out"
|
||||
|
||||
PLATFORMS = (
|
||||
ALARM,
|
||||
BINARY_SENSOR,
|
||||
CLIMATE,
|
||||
COVER,
|
||||
@ -118,6 +121,10 @@ PLATFORMS = (
|
||||
SWITCH,
|
||||
)
|
||||
|
||||
CONF_ALARM_MASTER_CODE = "alarm_master_code"
|
||||
CONF_ALARM_FAILED_TRIES = "alarm_failed_tries"
|
||||
CONF_ALARM_ARM_REQUIRES_CODE = "alarm_arm_requires_code"
|
||||
|
||||
CONF_BAUDRATE = "baudrate"
|
||||
CONF_CUSTOM_QUIRKS_PATH = "custom_quirks_path"
|
||||
CONF_DATABASE = "database_path"
|
||||
@ -137,6 +144,14 @@ CONF_ZHA_OPTIONS_SCHEMA = vol.Schema(
|
||||
}
|
||||
)
|
||||
|
||||
CONF_ZHA_ALARM_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_ALARM_MASTER_CODE, default="1234"): cv.string,
|
||||
vol.Required(CONF_ALARM_FAILED_TRIES, default=3): cv.positive_int,
|
||||
vol.Required(CONF_ALARM_ARM_REQUIRES_CODE, default=False): cv.boolean,
|
||||
}
|
||||
)
|
||||
|
||||
CUSTOM_CONFIGURATION = "custom_configuration"
|
||||
|
||||
DATA_DEVICE_CONFIG = "zha_device_config"
|
||||
@ -191,8 +206,13 @@ POWER_BATTERY_OR_UNKNOWN = "Battery or Unknown"
|
||||
PRESET_SCHEDULE = "schedule"
|
||||
PRESET_COMPLEX = "complex"
|
||||
|
||||
ZHA_ALARM_OPTIONS = "zha_alarm_options"
|
||||
ZHA_OPTIONS = "zha_options"
|
||||
ZHA_CONFIG_SCHEMAS = {ZHA_OPTIONS: CONF_ZHA_OPTIONS_SCHEMA}
|
||||
|
||||
ZHA_CONFIG_SCHEMAS = {
|
||||
ZHA_OPTIONS: CONF_ZHA_OPTIONS_SCHEMA,
|
||||
ZHA_ALARM_OPTIONS: CONF_ZHA_ALARM_SCHEMA,
|
||||
}
|
||||
|
||||
|
||||
class RadioType(enum.Enum):
|
||||
|
||||
@ -65,6 +65,7 @@ from .const import (
|
||||
UNKNOWN,
|
||||
UNKNOWN_MANUFACTURER,
|
||||
UNKNOWN_MODEL,
|
||||
ZHA_OPTIONS,
|
||||
)
|
||||
from .helpers import LogMixin, async_get_zha_config_value
|
||||
|
||||
@ -396,7 +397,10 @@ class ZHADevice(LogMixin):
|
||||
async def async_configure(self):
|
||||
"""Configure the device."""
|
||||
should_identify = async_get_zha_config_value(
|
||||
self._zha_gateway.config_entry, CONF_ENABLE_IDENTIFY_ON_JOIN, True
|
||||
self._zha_gateway.config_entry,
|
||||
ZHA_OPTIONS,
|
||||
CONF_ENABLE_IDENTIFY_ON_JOIN,
|
||||
True,
|
||||
)
|
||||
self.debug("started configuration")
|
||||
await self._channels.async_configure()
|
||||
|
||||
@ -15,6 +15,7 @@ from homeassistant.helpers.entity_registry import async_entries_for_device
|
||||
|
||||
from . import const as zha_const, registries as zha_regs, typing as zha_typing
|
||||
from .. import ( # noqa: F401 pylint: disable=unused-import,
|
||||
alarm_control_panel,
|
||||
binary_sensor,
|
||||
climate,
|
||||
cover,
|
||||
|
||||
@ -31,7 +31,6 @@ from .const import (
|
||||
CUSTOM_CONFIGURATION,
|
||||
DATA_ZHA,
|
||||
DATA_ZHA_GATEWAY,
|
||||
ZHA_OPTIONS,
|
||||
)
|
||||
from .registries import BINDABLE_CLUSTERS
|
||||
from .typing import ZhaDeviceType, ZigpyClusterType
|
||||
@ -131,15 +130,27 @@ def async_is_bindable_target(source_zha_device, target_zha_device):
|
||||
|
||||
|
||||
@callback
|
||||
def async_get_zha_config_value(config_entry, config_key, default):
|
||||
def async_get_zha_config_value(config_entry, section, config_key, default):
|
||||
"""Get the value for the specified configuration from the zha config entry."""
|
||||
return (
|
||||
config_entry.options.get(CUSTOM_CONFIGURATION, {})
|
||||
.get(ZHA_OPTIONS, {})
|
||||
.get(section, {})
|
||||
.get(config_key, default)
|
||||
)
|
||||
|
||||
|
||||
def async_input_cluster_exists(hass, cluster_id):
|
||||
"""Determine if a device containing the specified in cluster is paired."""
|
||||
zha_gateway = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY]
|
||||
zha_devices = zha_gateway.devices.values()
|
||||
for zha_device in zha_devices:
|
||||
clusters_by_endpoint = zha_device.async_get_clusters()
|
||||
for clusters in clusters_by_endpoint.values():
|
||||
if cluster_id in clusters[CLUSTER_TYPE_IN]:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
async def async_get_zha_device(hass, device_id):
|
||||
"""Get a ZHA device for the given device registry id."""
|
||||
device_registry = await hass.helpers.device_registry.async_get_registry()
|
||||
|
||||
@ -9,6 +9,7 @@ import zigpy.profiles.zha
|
||||
import zigpy.profiles.zll
|
||||
import zigpy.zcl as zcl
|
||||
|
||||
from homeassistant.components.alarm_control_panel import DOMAIN as ALARM
|
||||
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR
|
||||
from homeassistant.components.climate import DOMAIN as CLIMATE
|
||||
from homeassistant.components.cover import DOMAIN as COVER
|
||||
@ -104,6 +105,7 @@ DEVICE_CLASS = {
|
||||
zigpy.profiles.zha.DeviceType.ON_OFF_PLUG_IN_UNIT: SWITCH,
|
||||
zigpy.profiles.zha.DeviceType.SHADE: COVER,
|
||||
zigpy.profiles.zha.DeviceType.SMART_PLUG: SWITCH,
|
||||
zigpy.profiles.zha.DeviceType.IAS_ANCILLARY_CONTROL: ALARM,
|
||||
},
|
||||
zigpy.profiles.zll.PROFILE_ID: {
|
||||
zigpy.profiles.zll.DeviceType.COLOR_LIGHT: LIGHT,
|
||||
|
||||
@ -54,6 +54,7 @@ from .core.const import (
|
||||
SIGNAL_ADD_ENTITIES,
|
||||
SIGNAL_ATTR_UPDATED,
|
||||
SIGNAL_SET_LEVEL,
|
||||
ZHA_OPTIONS,
|
||||
)
|
||||
from .core.helpers import LogMixin, async_get_zha_config_value
|
||||
from .core.registries import ZHA_ENTITIES
|
||||
@ -394,7 +395,10 @@ class Light(BaseLight, ZhaEntity):
|
||||
self._effect_list = effect_list
|
||||
|
||||
self._default_transition = async_get_zha_config_value(
|
||||
zha_device.gateway.config_entry, CONF_DEFAULT_LIGHT_TRANSITION, 0
|
||||
zha_device.gateway.config_entry,
|
||||
ZHA_OPTIONS,
|
||||
CONF_DEFAULT_LIGHT_TRANSITION,
|
||||
0,
|
||||
)
|
||||
|
||||
@callback
|
||||
@ -553,7 +557,10 @@ class LightGroup(BaseLight, ZhaGroupEntity):
|
||||
self._identify_channel = group.endpoint[Identify.cluster_id]
|
||||
self._debounced_member_refresh = None
|
||||
self._default_transition = async_get_zha_config_value(
|
||||
zha_device.gateway.config_entry, CONF_DEFAULT_LIGHT_TRANSITION, 0
|
||||
zha_device.gateway.config_entry,
|
||||
ZHA_OPTIONS,
|
||||
CONF_DEFAULT_LIGHT_TRANSITION,
|
||||
0,
|
||||
)
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
|
||||
@ -48,6 +48,15 @@ async def config_entry_fixture(hass):
|
||||
zigpy.config.CONF_DEVICE: {zigpy.config.CONF_DEVICE_PATH: "/dev/ttyUSB0"},
|
||||
zha_const.CONF_RADIO_TYPE: "ezsp",
|
||||
},
|
||||
options={
|
||||
zha_const.CUSTOM_CONFIGURATION: {
|
||||
zha_const.ZHA_ALARM_OPTIONS: {
|
||||
zha_const.CONF_ALARM_ARM_REQUIRES_CODE: False,
|
||||
zha_const.CONF_ALARM_MASTER_CODE: "4321",
|
||||
zha_const.CONF_ALARM_FAILED_TRIES: 2,
|
||||
}
|
||||
}
|
||||
},
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
return entry
|
||||
|
||||
245
tests/components/zha/test_alarm_control_panel.py
Normal file
245
tests/components/zha/test_alarm_control_panel.py
Normal file
@ -0,0 +1,245 @@
|
||||
"""Test zha alarm control panel."""
|
||||
from unittest.mock import AsyncMock, call, patch, sentinel
|
||||
|
||||
import pytest
|
||||
import zigpy.profiles.zha as zha
|
||||
import zigpy.zcl.clusters.security as security
|
||||
import zigpy.zcl.foundation as zcl_f
|
||||
|
||||
from homeassistant.components.alarm_control_panel import DOMAIN as ALARM_DOMAIN
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID,
|
||||
STATE_ALARM_ARMED_AWAY,
|
||||
STATE_ALARM_ARMED_HOME,
|
||||
STATE_ALARM_ARMED_NIGHT,
|
||||
STATE_ALARM_DISARMED,
|
||||
STATE_ALARM_TRIGGERED,
|
||||
STATE_UNAVAILABLE,
|
||||
)
|
||||
|
||||
from .common import async_enable_traffic, find_entity_id
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def zigpy_device(zigpy_device_mock):
|
||||
"""Device tracker zigpy device."""
|
||||
endpoints = {
|
||||
1: {
|
||||
"in_clusters": [security.IasAce.cluster_id],
|
||||
"out_clusters": [],
|
||||
"device_type": zha.DeviceType.IAS_ANCILLARY_CONTROL,
|
||||
}
|
||||
}
|
||||
return zigpy_device_mock(
|
||||
endpoints, node_descriptor=b"\x02@\x8c\x02\x10RR\x00\x00\x00R\x00\x00"
|
||||
)
|
||||
|
||||
|
||||
@patch(
|
||||
"zigpy.zcl.clusters.security.IasAce.client_command",
|
||||
new=AsyncMock(return_value=[sentinel.data, zcl_f.Status.SUCCESS]),
|
||||
)
|
||||
async def test_alarm_control_panel(hass, zha_device_joined_restored, zigpy_device):
|
||||
"""Test zha alarm control panel platform."""
|
||||
|
||||
zha_device = await zha_device_joined_restored(zigpy_device)
|
||||
cluster = zigpy_device.endpoints.get(1).ias_ace
|
||||
entity_id = await find_entity_id(ALARM_DOMAIN, zha_device, hass)
|
||||
assert entity_id is not None
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
|
||||
await async_enable_traffic(hass, [zha_device], enabled=False)
|
||||
# test that the panel was created and that it is unavailable
|
||||
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
|
||||
|
||||
# allow traffic to flow through the gateway and device
|
||||
await async_enable_traffic(hass, [zha_device])
|
||||
|
||||
# test that the state has changed from unavailable to STATE_ALARM_DISARMED
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
|
||||
|
||||
# arm_away from HA
|
||||
cluster.client_command.reset_mock()
|
||||
await hass.services.async_call(
|
||||
ALARM_DOMAIN, "alarm_arm_away", {ATTR_ENTITY_ID: entity_id}, blocking=True
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY
|
||||
assert cluster.client_command.call_count == 2
|
||||
assert cluster.client_command.await_count == 2
|
||||
assert cluster.client_command.call_args == call(
|
||||
4,
|
||||
security.IasAce.PanelStatus.Armed_Away,
|
||||
0,
|
||||
security.IasAce.AudibleNotification.Default_Sound,
|
||||
security.IasAce.AlarmStatus.No_Alarm,
|
||||
)
|
||||
|
||||
# disarm from HA
|
||||
await reset_alarm_panel(hass, cluster, entity_id)
|
||||
|
||||
# trip alarm from faulty code entry
|
||||
cluster.client_command.reset_mock()
|
||||
await hass.services.async_call(
|
||||
ALARM_DOMAIN, "alarm_arm_away", {ATTR_ENTITY_ID: entity_id}, blocking=True
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY
|
||||
cluster.client_command.reset_mock()
|
||||
await hass.services.async_call(
|
||||
ALARM_DOMAIN,
|
||||
"alarm_disarm",
|
||||
{ATTR_ENTITY_ID: entity_id, "code": "1111"},
|
||||
blocking=True,
|
||||
)
|
||||
await hass.services.async_call(
|
||||
ALARM_DOMAIN,
|
||||
"alarm_disarm",
|
||||
{ATTR_ENTITY_ID: entity_id, "code": "1111"},
|
||||
blocking=True,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
|
||||
assert cluster.client_command.call_count == 4
|
||||
assert cluster.client_command.await_count == 4
|
||||
assert cluster.client_command.call_args == call(
|
||||
4,
|
||||
security.IasAce.PanelStatus.In_Alarm,
|
||||
0,
|
||||
security.IasAce.AudibleNotification.Default_Sound,
|
||||
security.IasAce.AlarmStatus.Emergency,
|
||||
)
|
||||
|
||||
# reset the panel
|
||||
await reset_alarm_panel(hass, cluster, entity_id)
|
||||
|
||||
# arm_home from HA
|
||||
cluster.client_command.reset_mock()
|
||||
await hass.services.async_call(
|
||||
ALARM_DOMAIN, "alarm_arm_home", {ATTR_ENTITY_ID: entity_id}, blocking=True
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_HOME
|
||||
assert cluster.client_command.call_count == 2
|
||||
assert cluster.client_command.await_count == 2
|
||||
assert cluster.client_command.call_args == call(
|
||||
4,
|
||||
security.IasAce.PanelStatus.Armed_Stay,
|
||||
0,
|
||||
security.IasAce.AudibleNotification.Default_Sound,
|
||||
security.IasAce.AlarmStatus.No_Alarm,
|
||||
)
|
||||
|
||||
# arm_night from HA
|
||||
cluster.client_command.reset_mock()
|
||||
await hass.services.async_call(
|
||||
ALARM_DOMAIN, "alarm_arm_night", {ATTR_ENTITY_ID: entity_id}, blocking=True
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_NIGHT
|
||||
assert cluster.client_command.call_count == 2
|
||||
assert cluster.client_command.await_count == 2
|
||||
assert cluster.client_command.call_args == call(
|
||||
4,
|
||||
security.IasAce.PanelStatus.Armed_Night,
|
||||
0,
|
||||
security.IasAce.AudibleNotification.Default_Sound,
|
||||
security.IasAce.AlarmStatus.No_Alarm,
|
||||
)
|
||||
|
||||
# reset the panel
|
||||
await reset_alarm_panel(hass, cluster, entity_id)
|
||||
|
||||
# arm from panel
|
||||
cluster.listener_event(
|
||||
"cluster_command", 1, 0, [security.IasAce.ArmMode.Arm_All_Zones, "", 0]
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY
|
||||
|
||||
# reset the panel
|
||||
await reset_alarm_panel(hass, cluster, entity_id)
|
||||
|
||||
# arm day home only from panel
|
||||
cluster.listener_event(
|
||||
"cluster_command", 1, 0, [security.IasAce.ArmMode.Arm_Day_Home_Only, "", 0]
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_HOME
|
||||
|
||||
# reset the panel
|
||||
await reset_alarm_panel(hass, cluster, entity_id)
|
||||
|
||||
# arm night sleep only from panel
|
||||
cluster.listener_event(
|
||||
"cluster_command", 1, 0, [security.IasAce.ArmMode.Arm_Night_Sleep_Only, "", 0]
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_NIGHT
|
||||
|
||||
# disarm from panel with bad code
|
||||
cluster.listener_event(
|
||||
"cluster_command", 1, 0, [security.IasAce.ArmMode.Disarm, "", 0]
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_NIGHT
|
||||
|
||||
# disarm from panel with bad code for 2nd time trips alarm
|
||||
cluster.listener_event(
|
||||
"cluster_command", 1, 0, [security.IasAce.ArmMode.Disarm, "", 0]
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
|
||||
|
||||
# disarm from panel with good code
|
||||
cluster.listener_event(
|
||||
"cluster_command", 1, 0, [security.IasAce.ArmMode.Disarm, "4321", 0]
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
|
||||
|
||||
# panic from panel
|
||||
cluster.listener_event("cluster_command", 1, 4, [])
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
|
||||
|
||||
# reset the panel
|
||||
await reset_alarm_panel(hass, cluster, entity_id)
|
||||
|
||||
# fire from panel
|
||||
cluster.listener_event("cluster_command", 1, 3, [])
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
|
||||
|
||||
# reset the panel
|
||||
await reset_alarm_panel(hass, cluster, entity_id)
|
||||
|
||||
# emergency from panel
|
||||
cluster.listener_event("cluster_command", 1, 2, [])
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
|
||||
|
||||
# reset the panel
|
||||
await reset_alarm_panel(hass, cluster, entity_id)
|
||||
|
||||
|
||||
async def reset_alarm_panel(hass, cluster, entity_id):
|
||||
"""Reset the state of the alarm panel."""
|
||||
cluster.client_command.reset_mock()
|
||||
await hass.services.async_call(
|
||||
ALARM_DOMAIN,
|
||||
"alarm_disarm",
|
||||
{ATTR_ENTITY_ID: entity_id, "code": "4321"},
|
||||
blocking=True,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
|
||||
assert cluster.client_command.call_count == 2
|
||||
assert cluster.client_command.await_count == 2
|
||||
assert cluster.client_command.call_args == call(
|
||||
4,
|
||||
security.IasAce.PanelStatus.Panel_Disarmed,
|
||||
0,
|
||||
security.IasAce.AudibleNotification.Default_Sound,
|
||||
security.IasAce.AlarmStatus.No_Alarm,
|
||||
)
|
||||
cluster.client_command.reset_mock()
|
||||
Loading…
Reference in New Issue
Block a user