jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-25 09:47:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add device automation support to ZHA (#26821)

Browse Source 
* beginning ZHA device automations

* add cube

* load triggers from zigpy devices

* cleanup

* add face_any for aqara cube

* add endpoint id to events

* add cluster id to events

* cleanup

* add device tilt

* add test

* fix copy paste error

* add event trigger test

* add test for no triggers for device

* add exception test

* better exception tests
This commit is contained in:
David F. Mulcahey 2019-09-24 11:54:41 -04:00 committed by Paulus Schoutsen
parent 161c8aada6
commit 18873d202d
6 changed files with 524 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
homeassistant/components/zha/.translations/en.json
View File

@ -16,5 +16,48 @@
}
},
"title": "ZHA"
},
"device_automation": {
"trigger_type": {
"remote_button_short_press": "\"{subtype}\" button pressed",
"remote_button_short_release": "\"{subtype}\" button released",
"remote_button_long_press": "\"{subtype}\" button continuously pressed",
"remote_button_long_release": "\"{subtype}\" button released after long press",
"remote_button_double_press": "\"{subtype}\" button double clicked",
"remote_button_triple_press": "\"{subtype}\" button triple clicked",
"remote_button_quadruple_press": "\"{subtype}\" button quadruple clicked",
"remote_button_quintuple_press": "\"{subtype}\" button quintuple clicked",
"device_rotated": "Device rotated \"{subtype}\"",
"device_shaken": "Device shaken",
"device_slid": "Device slid \"{subtype}\"",
"device_tilted": "Device tilted",
"device_knocked": "Device knocked \"{subtype}\"",
"device_dropped": "Device dropped",
"device_flipped": "Device flipped \"{subtype}\""
},
"trigger_subtype": {
"turn_on": "Turn on",
"turn_off": "Turn off",
"dim_up": "Dim up",
"dim_down": "Dim down",
"left": "Left",
"right": "Right",
"open": "Open",
"close": "Close",
"both_buttons": "Both buttons",
"button_1": "First button",
"button_2": "Second button",
"button_3": "Third button",
"button_4": "Fourth button",
"button_5": "Fifth button",
"button_6": "Sixth button",
"face_any": "With any/specified face(s) activated",
"face_1": "With face 1 activated",
"face_2": "With face 2 activated",
"face_3": "With face 3 activated",
"face_4": "With face 4 activated",
"face_5": "With face 5 activated",
"face_6": "With face 6 activated"
}
}
}

2
homeassistant/components/zha/core/channels/__init__.py
View File

@ -240,6 +240,8 @@ class ZigbeeChannel(LogMixin):
{
"unique_id": self._unique_id,
"device_ieee": str(self._zha_device.ieee),
"endpoint_id": cluster.endpoint.endpoint_id,
"cluster_id": cluster.cluster_id,
"command": command,
"args": args,
},

7
homeassistant/components/zha/core/device.py
View File

@ -187,6 +187,13 @@ class ZHADevice(LogMixin):
"""Return cluster channels and relay channels for device."""
return self._all_channels
@property
def device_automation_triggers(self):
"""Return the device automation triggers for this device."""
if hasattr(self._zigpy_device, "device_automation_triggers"):
return self._zigpy_device.device_automation_triggers
return None
@property
def available_signal(self):
"""Signal to use to subscribe to device availability changes."""

89
homeassistant/components/zha/device_automation.py Normal file
View File

@ -0,0 +1,89 @@
"""Provides device automations for ZHA devices that emit events."""
import voluptuous as vol
import homeassistant.components.automation.event as event
from homeassistant.components.device_automation.exceptions import (
InvalidDeviceAutomationConfig,
)
from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF_TYPE
from . import DOMAIN
from .core.const import DATA_ZHA, DATA_ZHA_GATEWAY
from .core.helpers import convert_ieee
CONF_SUBTYPE = "subtype"
DEVICE = "device"
DEVICE_IEEE = "device_ieee"
ZHA_EVENT = "zha_event"
TRIGGER_SCHEMA = vol.All(
vol.Schema(
{
vol.Required(CONF_DEVICE_ID): str,
vol.Required(CONF_DOMAIN): DOMAIN,
vol.Required(CONF_PLATFORM): DEVICE,
vol.Required(CONF_TYPE): str,
vol.Required(CONF_SUBTYPE): str,
}
)
)
async def async_trigger(hass, config, action, automation_info):
"""Listen for state changes based on configuration."""
config = TRIGGER_SCHEMA(config)
trigger = (config[CONF_TYPE], config[CONF_SUBTYPE])
zha_device = await _async_get_zha_device(hass, config[CONF_DEVICE_ID])
if (
zha_device.device_automation_triggers is None
or trigger not in zha_device.device_automation_triggers
):
raise InvalidDeviceAutomationConfig
trigger = zha_device.device_automation_triggers[trigger]
state_config = {
event.CONF_EVENT_TYPE: ZHA_EVENT,
event.CONF_EVENT_DATA: {DEVICE_IEEE: str(zha_device.ieee), **trigger},
}
return await event.async_trigger(hass, state_config, action, automation_info)
async def async_get_triggers(hass, device_id):
"""List device triggers.
Make sure the device supports device automations and
if it does return the trigger list.
"""
zha_device = await _async_get_zha_device(hass, device_id)
if not zha_device.device_automation_triggers:
return
triggers = []
for trigger, subtype in zha_device.device_automation_triggers.keys():
triggers.append(
{
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_PLATFORM: DEVICE,
CONF_TYPE: trigger,
CONF_SUBTYPE: subtype,
}
)
return triggers
async def _async_get_zha_device(hass, device_id):
device_registry = await hass.helpers.device_registry.async_get_registry()
registry_device = device_registry.async_get(device_id)
zha_gateway = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY]
ieee_address = list(list(registry_device.identifiers)[0])[1]
ieee = convert_ieee(ieee_address)
zha_device = zha_gateway.devices[ieee]
if not zha_device:
raise InvalidDeviceAutomationConfig
return zha_device

43
homeassistant/components/zha/strings.json
View File

@ -16,5 +16,48 @@
"abort": {
"single_instance_allowed": "Only a single configuration of ZHA is allowed."
}
},
"device_automation": {
"trigger_type": {
"remote_button_short_press": "\"{subtype}\" button pressed",
"remote_button_short_release": "\"{subtype}\" button released",
"remote_button_long_press": "\"{subtype}\" button continuously pressed",
"remote_button_long_release": "\"{subtype}\" button released after long press",
"remote_button_double_press": "\"{subtype}\" button double clicked",
"remote_button_triple_press": "\"{subtype}\" button triple clicked",
"remote_button_quadruple_press": "\"{subtype}\" button quadruple clicked",
"remote_button_quintuple_press": "\"{subtype}\" button quintuple clicked",
"device_rotated": "Device rotated \"{subtype}\"",
"device_shaken": "Device shaken",
"device_slid": "Device slid \"{subtype}\"",
"device_tilted": "Device tilted",
"device_knocked": "Device knocked \"{subtype}\"",
"device_dropped": "Device dropped",
"device_flipped": "Device flipped \"{subtype}\""
},
"trigger_subtype": {
"turn_on": "Turn on",
"turn_off": "Turn off",
"dim_up": "Dim up",
"dim_down": "Dim down",
"left": "Left",
"right": "Right",
"open": "Open",
"close": "Close",
"both_buttons": "Both buttons",
"button_1": "First button",
"button_2": "Second button",
"button_3": "Third button",
"button_4": "Fourth button",
"button_5": "Fifth button",
"button_6": "Sixth button",
"face_any": "With any/specified face(s) activated",
"face_1": "with face 1 activated",
"face_2": "with face 2 activated",
"face_3": "with face 3 activated",
"face_4": "with face 4 activated",
"face_5": "with face 5 activated",
"face_6": "with face 6 activated"
}
}
}

308
tests/components/zha/test_device_automation.py Normal file
View File

@ -0,0 +1,308 @@
"""ZHA device automation tests."""
from unittest.mock import patch
import pytest
import homeassistant.components.automation as automation
from homeassistant.components.device_automation import (
_async_get_device_automations as async_get_device_automations,
)
from homeassistant.components.switch import DOMAIN
from homeassistant.components.zha.core.const import CHANNEL_ON_OFF
from homeassistant.helpers.device_registry import async_get_registry
from homeassistant.setup import async_setup_component
from .common import async_enable_traffic, async_init_zigpy_device
from tests.common import async_mock_service
ON = 1
OFF = 0
SHAKEN = "device_shaken"
COMMAND = "command"
COMMAND_SHAKE = "shake"
COMMAND_HOLD = "hold"
COMMAND_SINGLE = "single"
COMMAND_DOUBLE = "double"
DOUBLE_PRESS = "remote_button_double_press"
SHORT_PRESS = "remote_button_short_press"
LONG_PRESS = "remote_button_long_press"
LONG_RELEASE = "remote_button_long_release"
def _same_lists(list_a, list_b):
if len(list_a) != len(list_b):
return False
for item in list_a:
if item not in list_b:
return False
return True
@pytest.fixture
def calls(hass):
"""Track calls to a mock serivce."""
return async_mock_service(hass, "test", "automation")
async def test_triggers(hass, config_entry, zha_gateway):
"""Test zha device triggers."""
from zigpy.zcl.clusters.general import OnOff, Basic
# create zigpy device
zigpy_device = await async_init_zigpy_device(
hass, [Basic.cluster_id], [OnOff.cluster_id], None, zha_gateway
)
zigpy_device.device_automation_triggers = {
(SHAKEN, SHAKEN): {COMMAND: COMMAND_SHAKE},
(DOUBLE_PRESS, DOUBLE_PRESS): {COMMAND: COMMAND_DOUBLE},
(SHORT_PRESS, SHORT_PRESS): {COMMAND: COMMAND_SINGLE},
(LONG_PRESS, LONG_PRESS): {COMMAND: COMMAND_HOLD},
(LONG_RELEASE, LONG_RELEASE): {COMMAND: COMMAND_HOLD},
}
await hass.config_entries.async_forward_entry_setup(config_entry, DOMAIN)
await hass.async_block_till_done()
hass.config_entries._entries.append(config_entry)
zha_device = zha_gateway.get_device(zigpy_device.ieee)
ieee_address = str(zha_device.ieee)
ha_device_registry = await async_get_registry(hass)
reg_device = ha_device_registry.async_get_device({("zha", ieee_address)}, set())
triggers = await async_get_device_automations(
hass, "async_get_triggers", reg_device.id
)
expected_triggers = [
{
"device_id": reg_device.id,
"domain": "zha",
"platform": "device",
"type": SHAKEN,
"subtype": SHAKEN,
},
{
"device_id": reg_device.id,
"domain": "zha",
"platform": "device",
"type": DOUBLE_PRESS,
"subtype": DOUBLE_PRESS,
},
{
"device_id": reg_device.id,
"domain": "zha",
"platform": "device",
"type": SHORT_PRESS,
"subtype": SHORT_PRESS,
},
{
"device_id": reg_device.id,
"domain": "zha",
"platform": "device",
"type": LONG_PRESS,
"subtype": LONG_PRESS,
},
{
"device_id": reg_device.id,
"domain": "zha",
"platform": "device",
"type": LONG_RELEASE,
"subtype": LONG_RELEASE,
},
]
assert _same_lists(triggers, expected_triggers)
async def test_no_triggers(hass, config_entry, zha_gateway):
"""Test zha device with no triggers."""
from zigpy.zcl.clusters.general import OnOff, Basic
# create zigpy device
zigpy_device = await async_init_zigpy_device(
hass, [Basic.cluster_id], [OnOff.cluster_id], None, zha_gateway
)
await hass.config_entries.async_forward_entry_setup(config_entry, DOMAIN)
await hass.async_block_till_done()
hass.config_entries._entries.append(config_entry)
zha_device = zha_gateway.get_device(zigpy_device.ieee)
ieee_address = str(zha_device.ieee)
ha_device_registry = await async_get_registry(hass)
reg_device = ha_device_registry.async_get_device({("zha", ieee_address)}, set())
triggers = await async_get_device_automations(
hass, "async_get_triggers", reg_device.id
)
assert triggers == []
async def test_if_fires_on_event(hass, config_entry, zha_gateway, calls):
"""Test for remote triggers firing."""
from zigpy.zcl.clusters.general import OnOff, Basic
# create zigpy device
zigpy_device = await async_init_zigpy_device(
hass, [Basic.cluster_id], [OnOff.cluster_id], None, zha_gateway
)
zigpy_device.device_automation_triggers = {
(SHAKEN, SHAKEN): {COMMAND: COMMAND_SHAKE},
(DOUBLE_PRESS, DOUBLE_PRESS): {COMMAND: COMMAND_DOUBLE},
(SHORT_PRESS, SHORT_PRESS): {COMMAND: COMMAND_SINGLE},
(LONG_PRESS, LONG_PRESS): {COMMAND: COMMAND_HOLD},
(LONG_RELEASE, LONG_RELEASE): {COMMAND: COMMAND_HOLD},
}
await hass.config_entries.async_forward_entry_setup(config_entry, DOMAIN)
await hass.async_block_till_done()
hass.config_entries._entries.append(config_entry)
zha_device = zha_gateway.get_device(zigpy_device.ieee)
# allow traffic to flow through the gateway and device
await async_enable_traffic(hass, zha_gateway, [zha_device])
ieee_address = str(zha_device.ieee)
ha_device_registry = await async_get_registry(hass)
reg_device = ha_device_registry.async_get_device({("zha", ieee_address)}, set())
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
"device_id": reg_device.id,
"domain": "zha",
"platform": "device",
"type": SHORT_PRESS,
"subtype": SHORT_PRESS,
},
"action": {
"service": "test.automation",
"data": {"message": "service called"},
},
}
]
},
)
await hass.async_block_till_done()
on_off_channel = zha_device.cluster_channels[CHANNEL_ON_OFF]
on_off_channel.zha_send_event(on_off_channel.cluster, COMMAND_SINGLE, [])
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["message"] == "service called"
async def test_exception_no_triggers(hass, config_entry, zha_gateway, calls):
"""Test for exception on event triggers firing."""
from zigpy.zcl.clusters.general import OnOff, Basic
# create zigpy device
zigpy_device = await async_init_zigpy_device(
hass, [Basic.cluster_id], [OnOff.cluster_id], None, zha_gateway
)
await hass.config_entries.async_forward_entry_setup(config_entry, DOMAIN)
await hass.async_block_till_done()
hass.config_entries._entries.append(config_entry)
zha_device = zha_gateway.get_device(zigpy_device.ieee)
# allow traffic to flow through the gateway and device
await async_enable_traffic(hass, zha_gateway, [zha_device])
ieee_address = str(zha_device.ieee)
ha_device_registry = await async_get_registry(hass)
reg_device = ha_device_registry.async_get_device({("zha", ieee_address)}, set())
with patch("logging.Logger.error") as mock:
await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
"device_id": reg_device.id,
"domain": "zha",
"platform": "device",
"type": "junk",
"subtype": "junk",
},
"action": {
"service": "test.automation",
"data": {"message": "service called"},
},
}
]
},
)
await hass.async_block_till_done()
mock.assert_called_with("Error setting up trigger %s", "automation 0")
async def test_exception_bad_trigger(hass, config_entry, zha_gateway, calls):
"""Test for exception on event triggers firing."""
from zigpy.zcl.clusters.general import OnOff, Basic
# create zigpy device
zigpy_device = await async_init_zigpy_device(
hass, [Basic.cluster_id], [OnOff.cluster_id], None, zha_gateway
)
zigpy_device.device_automation_triggers = {
(SHAKEN, SHAKEN): {COMMAND: COMMAND_SHAKE},
(DOUBLE_PRESS, DOUBLE_PRESS): {COMMAND: COMMAND_DOUBLE},
(SHORT_PRESS, SHORT_PRESS): {COMMAND: COMMAND_SINGLE},
(LONG_PRESS, LONG_PRESS): {COMMAND: COMMAND_HOLD},
(LONG_RELEASE, LONG_RELEASE): {COMMAND: COMMAND_HOLD},
}
await hass.config_entries.async_forward_entry_setup(config_entry, DOMAIN)
await hass.async_block_till_done()
hass.config_entries._entries.append(config_entry)
zha_device = zha_gateway.get_device(zigpy_device.ieee)
# allow traffic to flow through the gateway and device
await async_enable_traffic(hass, zha_gateway, [zha_device])
ieee_address = str(zha_device.ieee)
ha_device_registry = await async_get_registry(hass)
reg_device = ha_device_registry.async_get_device({("zha", ieee_address)}, set())
with patch("logging.Logger.error") as mock:
await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
"device_id": reg_device.id,
"domain": "zha",
"platform": "device",
"type": "junk",
"subtype": "junk",
},
"action": {
"service": "test.automation",
"data": {"message": "service called"},
},
}
]
},
)
await hass.async_block_till_done()
mock.assert_called_with("Error setting up trigger %s", "automation 0")