Add websock command to query device for triggers (#24044)

* Add websock command to query device for triggers

* Lint

* Refactor

* Add support for domain automations

* Make device automation an automation platform

* lint

* Support device_id in light trigger

* Review comments

* Add tests

* Add tests

* lint
This commit is contained in:
Erik Montnemery 2019-06-11 00:36:11 +02:00 committed by Paulus Schoutsen
parent 168f20bdf4
commit 935240f8c3
9 changed files with 388 additions and 1 deletions

View File

@ -61,6 +61,7 @@ homeassistant/components/daikin/* @fredrike @rofrantz
homeassistant/components/darksky/* @fabaff
homeassistant/components/deconz/* @kane610
homeassistant/components/demo/* @home-assistant/core
homeassistant/components/device_automation/* @home-assistant/core
homeassistant/components/digital_ocean/* @fabaff
homeassistant/components/discogs/* @thibmaek
homeassistant/components/doorbird/* @oblogic7

View File

@ -50,7 +50,7 @@ _LOGGER = logging.getLogger(__name__)
def _platform_validator(config):
"""Validate it is a valid platform."""
"""Validate it is a valid platform."""
try:
platform = importlib.import_module('.{}'.format(config[CONF_PLATFORM]),
__name__)

View File

@ -0,0 +1,18 @@
"""Offer device oriented automation."""
import voluptuous as vol
from homeassistant.const import CONF_DOMAIN, CONF_PLATFORM
from homeassistant.loader import async_get_integration
TRIGGER_SCHEMA = vol.Schema({
vol.Required(CONF_PLATFORM): 'device',
vol.Required(CONF_DOMAIN): str,
}, extra=vol.ALLOW_EXTRA)
async def async_trigger(hass, config, action, automation_info):
"""Listen for trigger."""
integration = await async_get_integration(hass, config[CONF_DOMAIN])
platform = integration.get_platform('device_automation')
return await platform.async_trigger(hass, config, action, automation_info)

View File

@ -0,0 +1,80 @@
"""Helpers for device automations."""
import asyncio
import logging
import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.core import split_entity_id
from homeassistant.helpers.entity_registry import async_entries_for_device
from homeassistant.loader import async_get_integration, IntegrationNotFound
DOMAIN = 'device_automation'
_LOGGER = logging.getLogger(__name__)
async def async_setup(hass, config):
"""Set up device automation."""
hass.components.websocket_api.async_register_command(
websocket_device_automation_list_triggers)
return True
async def _async_get_device_automation_triggers(hass, domain, device_id):
"""List device triggers."""
integration = None
try:
integration = await async_get_integration(hass, domain)
except IntegrationNotFound:
_LOGGER.warning('Integration %s not found', domain)
return None
try:
platform = integration.get_platform('device_automation')
except ImportError:
# The domain does not have device automations
return None
if hasattr(platform, 'async_get_triggers'):
return await platform.async_get_triggers(hass, device_id)
async def async_get_device_automation_triggers(hass, device_id):
"""List device triggers."""
device_registry, entity_registry = await asyncio.gather(
hass.helpers.device_registry.async_get_registry(),
hass.helpers.entity_registry.async_get_registry())
domains = set()
triggers = []
device = device_registry.async_get(device_id)
for entry_id in device.config_entries:
config_entry = hass.config_entries.async_get_entry(entry_id)
domains.add(config_entry.domain)
entities = async_entries_for_device(entity_registry, device_id)
for entity in entities:
domains.add(split_entity_id(entity.entity_id)[0])
device_triggers = await asyncio.gather(*[
_async_get_device_automation_triggers(hass, domain, device_id)
for domain in domains
])
for device_trigger in device_triggers:
if device_trigger is not None:
triggers.extend(device_trigger)
return triggers
@websocket_api.async_response
@websocket_api.websocket_command({
vol.Required('type'): 'device_automation/list_triggers',
vol.Required('device_id'): str,
})
async def websocket_device_automation_list_triggers(hass, connection, msg):
"""Handle request for device triggers."""
device_id = msg['device_id']
triggers = await async_get_device_automation_triggers(hass, device_id)
connection.send_result(msg['id'], {'triggers': triggers})

View File

@ -0,0 +1,12 @@
{
"domain": "device_automation",
"name": "Device automation",
"documentation": "https://www.home-assistant.io/components/device_automation",
"requirements": [],
"dependencies": [
"webhook"
],
"codeowners": [
"@home-assistant/core"
]
}

View File

@ -0,0 +1,80 @@
"""Provides device automations for lights."""
import voluptuous as vol
import homeassistant.components.automation.state as state
from homeassistant.core import split_entity_id
from homeassistant.const import (
CONF_DEVICE_ID, CONF_DOMAIN, CONF_ENTITY_ID, CONF_PLATFORM, CONF_TYPE)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_registry import async_entries_for_device
from . import DOMAIN
CONF_TURN_OFF = 'turn_off'
CONF_TURN_ON = 'turn_on'
ENTITY_TRIGGERS = [
{
# Trigger when light is turned on
CONF_PLATFORM: 'device',
CONF_DOMAIN: DOMAIN,
CONF_TYPE: CONF_TURN_OFF,
},
{
# Trigger when light is turned off
CONF_PLATFORM: 'device',
CONF_DOMAIN: DOMAIN,
CONF_TYPE: CONF_TURN_ON,
},
]
TRIGGER_SCHEMA = vol.All(vol.Schema({
vol.Required(CONF_PLATFORM): 'device',
vol.Optional(CONF_DEVICE_ID): str,
vol.Required(CONF_DOMAIN): DOMAIN,
vol.Required(CONF_ENTITY_ID): cv.entity_id,
vol.Required(CONF_TYPE): str,
}))
def _is_domain(entity, domain):
return split_entity_id(entity.entity_id)[0] == domain
async def async_attach_trigger(hass, config, action, automation_info):
"""Listen for state changes based on configuration."""
trigger_type = config.get(CONF_TYPE)
if trigger_type == CONF_TURN_ON:
from_state = 'off'
to_state = 'on'
else:
from_state = 'on'
to_state = 'off'
state_config = {
state.CONF_ENTITY_ID: config[CONF_ENTITY_ID],
state.CONF_FROM: from_state,
state.CONF_TO: to_state
}
return await state.async_trigger(hass, state_config, action,
automation_info)
async def async_trigger(hass, config, action, automation_info):
"""Temporary so existing automation framework can be used for testing."""
return await async_attach_trigger(hass, config, action, automation_info)
async def async_get_triggers(hass, device_id):
"""List device triggers."""
triggers = []
entity_registry = await hass.helpers.entity_registry.async_get_registry()
entities = async_entries_for_device(entity_registry, device_id)
domain_entities = [x for x in entities if _is_domain(x, DOMAIN)]
for entity in domain_entities:
for trigger in ENTITY_TRIGGERS:
trigger = dict(trigger)
trigger.update(device_id=device_id, entity_id=entity.entity_id)
triggers.append(trigger)
return triggers

View File

@ -59,6 +59,7 @@ CONF_CUSTOMIZE_GLOB = 'customize_glob'
CONF_DELAY_TIME = 'delay_time'
CONF_DEVICE = 'device'
CONF_DEVICE_CLASS = 'device_class'
CONF_DEVICE_ID = 'device_id'
CONF_DEVICES = 'devices'
CONF_DISARM_AFTER_TRIGGER = 'disarm_after_trigger'
CONF_DISCOVERY = 'discovery'

View File

@ -0,0 +1,67 @@
"""The test for light device automation."""
import pytest
from homeassistant.setup import async_setup_component
from homeassistant.components.websocket_api.const import TYPE_RESULT
from homeassistant.helpers import device_registry
from tests.common import (
MockConfigEntry, mock_device_registry, mock_registry)
@pytest.fixture
def device_reg(hass):
"""Return an empty, loaded, registry."""
return mock_device_registry(hass)
@pytest.fixture
def entity_reg(hass):
"""Return an empty, loaded, registry."""
return mock_registry(hass)
def _same_triggers(a, b):
if len(a) != len(b):
return False
for d in a:
if d not in b:
return False
return True
async def test_websocket_get_triggers(
hass, hass_ws_client, device_reg, entity_reg):
"""Test we get the expected triggers from a light through websocket."""
await async_setup_component(hass, 'device_automation', {})
config_entry = MockConfigEntry(domain='test', data={})
config_entry.add_to_hass(hass)
device_entry = device_reg.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={
(device_registry.CONNECTION_NETWORK_MAC, '12:34:56:AB:CD:EF')
})
entity_reg.async_get_or_create(
'light', 'test', '5678', device_id=device_entry.id)
expected_triggers = [
{'platform': 'device', 'domain': 'light', 'type': 'turn_off',
'device_id': device_entry.id, 'entity_id': 'light.test_5678'},
{'platform': 'device', 'domain': 'light', 'type': 'turn_on',
'device_id': device_entry.id, 'entity_id': 'light.test_5678'},
]
client = await hass_ws_client(hass)
await client.send_json({
'id': 1,
'type': 'device_automation/list_triggers',
'device_id': device_entry.id
})
msg = await client.receive_json()
assert msg['id'] == 1
assert msg['type'] == TYPE_RESULT
assert msg['success']
triggers = msg['result']['triggers']
assert _same_triggers(triggers, expected_triggers)

View File

@ -0,0 +1,128 @@
"""The test for light device automation."""
import pytest
from homeassistant.components import light
from homeassistant.const import (
STATE_ON, STATE_OFF, CONF_PLATFORM)
from homeassistant.setup import async_setup_component
import homeassistant.components.automation as automation
from homeassistant.components.device_automation import (
async_get_device_automation_triggers)
from homeassistant.helpers import device_registry
from tests.common import (
MockConfigEntry, async_mock_service, mock_device_registry, mock_registry)
@pytest.fixture
def device_reg(hass):
"""Return an empty, loaded, registry."""
return mock_device_registry(hass)
@pytest.fixture
def entity_reg(hass):
"""Return an empty, loaded, registry."""
return mock_registry(hass)
@pytest.fixture
def calls(hass):
"""Track calls to a mock serivce."""
return async_mock_service(hass, 'test', 'automation')
def _same_triggers(a, b):
if len(a) != len(b):
return False
for d in a:
if d not in b:
return False
return True
async def test_get_triggers(hass, device_reg, entity_reg):
"""Test we get the expected triggers from a light."""
config_entry = MockConfigEntry(domain='test', data={})
config_entry.add_to_hass(hass)
device_entry = device_reg.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={
(device_registry.CONNECTION_NETWORK_MAC, '12:34:56:AB:CD:EF')
})
entity_reg.async_get_or_create(
'light', 'test', '5678', device_id=device_entry.id)
expected_triggers = [
{'platform': 'device', 'domain': 'light', 'type': 'turn_off',
'device_id': device_entry.id, 'entity_id': 'light.test_5678'},
{'platform': 'device', 'domain': 'light', 'type': 'turn_on',
'device_id': device_entry.id, 'entity_id': 'light.test_5678'},
]
triggers = await async_get_device_automation_triggers(hass,
device_entry.id)
assert _same_triggers(triggers, expected_triggers)
async def test_if_fires_on_state_change(hass, calls):
"""Test for turn_on and turn_off triggers firing."""
platform = getattr(hass.components, 'test.light')
platform.init()
assert await async_setup_component(hass, light.DOMAIN,
{light.DOMAIN: {CONF_PLATFORM: 'test'}})
dev1, dev2, dev3 = platform.DEVICES
assert await async_setup_component(hass, automation.DOMAIN, {
automation.DOMAIN: [{
'trigger': {
'platform': 'device',
'domain': light.DOMAIN,
'entity_id': dev1.entity_id,
'type': 'turn_on'
},
'action': {
'service': 'test.automation',
'data_template': {
'some':
'turn_on {{ trigger.%s }}' % '}} - {{ trigger.'.join((
'platform', 'entity_id',
'from_state.state', 'to_state.state',
'for'))
},
}},
{'trigger': {
'platform': 'device',
'domain': light.DOMAIN,
'entity_id': dev1.entity_id,
'type': 'turn_off'
},
'action': {
'service': 'test.automation',
'data_template': {
'some':
'turn_off {{ trigger.%s }}' % '}} - {{ trigger.'.join((
'platform', 'entity_id',
'from_state.state', 'to_state.state',
'for'))
},
}},
]
})
await hass.async_block_till_done()
assert hass.states.get(dev1.entity_id).state == STATE_ON
assert len(calls) == 0
hass.states.async_set(dev1.entity_id, STATE_OFF)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data['some'] == \
'turn_off state - {} - on - off - None'.format(dev1.entity_id)
hass.states.async_set(dev1.entity_id, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[1].data['some'] == \
'turn_on state - {} - off - on - None'.format(dev1.entity_id)