Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add device triggers for Hue remotes (#33476) 

* Store device_registry entry id in HueEvent

so it can be retrieved with that key when using device triggers

* Add device_trigger for hue_event from hue remotes

* supporting Hue dimmer switch & Hue Tap
* State mapping and strings are copied from deCONZ

* refactor mock_bridge for hue tests

and also share `setup_bridge_for_sensors`
for test_sensor_base and test_device_trigger.

* Add tests for device triggers with hue remotes

* Remove some triggers
pull/33508/head
Eugenio Panadero 2020-04-01 20:42:22 +02:00 committed by GitHub
parent 400602a8b3
commit 71aaf2d809
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 443 additions and 120 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
homeassistant/components/hue/.translations/en.json
View File

@ -27,5 +27,22 @@
}
},
"title": "Philips Hue"
},
"device_automation": {
"trigger_subtype": {
"button_1": "First button",
"button_2": "Second button",
"button_3": "Third button",
"button_4": "Fourth button",
"dim_down": "Dim down",
"dim_up": "Dim up",
"turn_off": "Turn off",
"turn_on": "Turn on"
},
"trigger_type": {
"remote_button_long_release": "\"{subtype}\" button released after long press",
"remote_button_short_press": "\"{subtype}\" button pressed",
"remote_button_short_release": "\"{subtype}\" button released"
}
}
}

149
homeassistant/components/hue/device_trigger.py Normal file
View File

@ -0,0 +1,149 @@
"""Provides device automations for Philips Hue events."""
import logging
import voluptuous as vol
import homeassistant.components.automation.event as event
from homeassistant.components.device_automation import TRIGGER_BASE_SCHEMA
from homeassistant.components.device_automation.exceptions import (
InvalidDeviceAutomationConfig,
)
from homeassistant.const import (
CONF_DEVICE_ID,
CONF_DOMAIN,
CONF_EVENT,
CONF_PLATFORM,
CONF_TYPE,
)
from . import DOMAIN
from .hue_event import CONF_HUE_EVENT, CONF_UNIQUE_ID
_LOGGER = logging.getLogger(__file__)
CONF_SUBTYPE = "subtype"
CONF_SHORT_PRESS = "remote_button_short_press"
CONF_SHORT_RELEASE = "remote_button_short_release"
CONF_LONG_RELEASE = "remote_button_long_release"
CONF_TURN_ON = "turn_on"
CONF_TURN_OFF = "turn_off"
CONF_DIM_UP = "dim_up"
CONF_DIM_DOWN = "dim_down"
CONF_BUTTON_1 = "button_1"
CONF_BUTTON_2 = "button_2"
CONF_BUTTON_3 = "button_3"
CONF_BUTTON_4 = "button_4"
HUE_DIMMER_REMOTE_MODEL = "Hue dimmer switch" # RWL020/021
HUE_DIMMER_REMOTE = {
(CONF_SHORT_RELEASE, CONF_TURN_ON): {CONF_EVENT: 1002},
(CONF_LONG_RELEASE, CONF_TURN_ON): {CONF_EVENT: 1003},
(CONF_SHORT_RELEASE, CONF_DIM_UP): {CONF_EVENT: 2002},
(CONF_LONG_RELEASE, CONF_DIM_UP): {CONF_EVENT: 2003},
(CONF_SHORT_RELEASE, CONF_DIM_DOWN): {CONF_EVENT: 3002},
(CONF_LONG_RELEASE, CONF_DIM_DOWN): {CONF_EVENT: 3003},
(CONF_SHORT_RELEASE, CONF_TURN_OFF): {CONF_EVENT: 4002},
(CONF_LONG_RELEASE, CONF_TURN_OFF): {CONF_EVENT: 4003},
}
HUE_TAP_REMOTE_MODEL = "Hue tap switch" # ZGPSWITCH
HUE_TAP_REMOTE = {
(CONF_SHORT_PRESS, CONF_BUTTON_1): {CONF_EVENT: 34},
(CONF_SHORT_PRESS, CONF_BUTTON_2): {CONF_EVENT: 16},
(CONF_SHORT_PRESS, CONF_BUTTON_3): {CONF_EVENT: 17},
(CONF_SHORT_PRESS, CONF_BUTTON_4): {CONF_EVENT: 18},
}
REMOTES = {
HUE_DIMMER_REMOTE_MODEL: HUE_DIMMER_REMOTE,
HUE_TAP_REMOTE_MODEL: HUE_TAP_REMOTE,
}
TRIGGER_SCHEMA = TRIGGER_BASE_SCHEMA.extend(
{vol.Required(CONF_TYPE): str, vol.Required(CONF_SUBTYPE): str}
)
def _get_hue_event_from_device_id(hass, device_id):
"""Resolve hue event from device id."""
for bridge in hass.data.get(DOMAIN, {}).values():
for hue_event in bridge.sensor_manager.current_events.values():
if device_id == hue_event.device_registry_id:
return hue_event
return None
async def async_validate_trigger_config(hass, config):
"""Validate config."""
config = TRIGGER_SCHEMA(config)
device_registry = await hass.helpers.device_registry.async_get_registry()
device = device_registry.async_get(config[CONF_DEVICE_ID])
trigger = (config[CONF_TYPE], config[CONF_SUBTYPE])
if (
not device
or device.model not in REMOTES
or trigger not in REMOTES[device.model]
):
raise InvalidDeviceAutomationConfig
return config
async def async_attach_trigger(hass, config, action, automation_info):
"""Listen for state changes based on configuration."""
device_registry = await hass.helpers.device_registry.async_get_registry()
device = device_registry.async_get(config[CONF_DEVICE_ID])
hue_event = _get_hue_event_from_device_id(hass, device.id)
if hue_event is None:
raise InvalidDeviceAutomationConfig
trigger = (config[CONF_TYPE], config[CONF_SUBTYPE])
trigger = REMOTES[device.model][trigger]
event_config = {
event.CONF_PLATFORM: "event",
event.CONF_EVENT_TYPE: CONF_HUE_EVENT,
event.CONF_EVENT_DATA: {CONF_UNIQUE_ID: hue_event.unique_id, **trigger},
}
event_config = event.TRIGGER_SCHEMA(event_config)
return await event.async_attach_trigger(
hass, event_config, action, automation_info, platform_type="device"
)
async def async_get_triggers(hass, device_id):
"""List device triggers.
Make sure device is a supported remote model.
Retrieve the hue event object matching device entry.
Generate device trigger list.
"""
device_registry = await hass.helpers.device_registry.async_get_registry()
device = device_registry.async_get(device_id)
if device.model not in REMOTES:
return
triggers = []
for trigger, subtype in REMOTES[device.model].keys():
triggers.append(
{
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_PLATFORM: "device",
CONF_TYPE: trigger,
CONF_SUBTYPE: subtype,
}
)
return triggers

4
homeassistant/components/hue/hue_event.py
View File

@ -28,6 +28,7 @@ class HueEvent(GenericHueDevice):
def __init__(self, sensor, name, bridge, primary_sensor=None):
"""Register callback that will be used for signals."""
super().__init__(sensor, name, bridge, primary_sensor)
self.device_registry_id = None
self.event_id = slugify(self.sensor.name)
# Use the 'lastupdated' string to detect new remote presses
@ -79,9 +80,10 @@ class HueEvent(GenericHueDevice):
entry = device_registry.async_get_or_create(
config_entry_id=self.bridge.config_entry.entry_id, **self.device_info
)
self.device_registry_id = entry.id
_LOGGER.debug(
"Event registry with entry_id: %s and device_id: %s",
entry.id,
self.device_registry_id,
self.device_id,
)

19
homeassistant/components/hue/strings.json
View File

@ -27,5 +27,22 @@
"already_in_progress": "Config flow for bridge is already in progress.",
"not_hue_bridge": "Not a Hue bridge"
}
},
"device_automation": {
"trigger_subtype": {
"button_1": "First button",
"button_2": "Second button",
"button_3": "Third button",
"button_4": "Fourth button",
"dim_down": "Dim down",
"dim_up": "Dim up",
"turn_off": "Turn off",
"turn_on": "Turn on"
},
"trigger_type": {
"remote_button_long_release": "\"{subtype}\" button released after long press",
"remote_button_short_press": "\"{subtype}\" button pressed",
"remote_button_short_release": "\"{subtype}\" button released"
}
}
}
}

86
tests/components/hue/conftest.py
View File

@ -1,11 +1,95 @@
"""Test helpers for Hue."""
from unittest.mock import patch
from collections import deque
from unittest.mock import Mock, patch
from aiohue.groups import Groups
from aiohue.lights import Lights
from aiohue.sensors import Sensors
import pytest
from homeassistant import config_entries
from homeassistant.components import hue
from homeassistant.components.hue import sensor_base as hue_sensor_base
@pytest.fixture(autouse=True)
def no_request_delay():
"""Make the request refresh delay 0 for instant tests."""
with patch("homeassistant.components.hue.light.REQUEST_REFRESH_DELAY", 0):
yield
def create_mock_bridge(hass):
"""Create a mock Hue bridge."""
bridge = Mock(
hass=hass,
available=True,
authorized=True,
allow_unreachable=False,
allow_groups=False,
api=Mock(),
reset_jobs=[],
spec=hue.HueBridge,
)
bridge.sensor_manager = hue_sensor_base.SensorManager(bridge)
bridge.mock_requests = []
# We're using a deque so we can schedule multiple responses
# and also means that `popleft()` will blow up if we get more updates
# than expected.
bridge.mock_light_responses = deque()
bridge.mock_group_responses = deque()
bridge.mock_sensor_responses = deque()
async def mock_request(method, path, **kwargs):
kwargs["method"] = method
kwargs["path"] = path
bridge.mock_requests.append(kwargs)
if path == "lights":
return bridge.mock_light_responses.popleft()
if path == "groups":
return bridge.mock_group_responses.popleft()
if path == "sensors":
return bridge.mock_sensor_responses.popleft()
return None
async def async_request_call(task):
await task()
bridge.async_request_call = async_request_call
bridge.api.config.apiversion = "9.9.9"
bridge.api.lights = Lights({}, mock_request)
bridge.api.groups = Groups({}, mock_request)
bridge.api.sensors = Sensors({}, mock_request)
return bridge
@pytest.fixture
def mock_bridge(hass):
"""Mock a Hue bridge."""
return create_mock_bridge(hass)
async def setup_bridge_for_sensors(hass, mock_bridge, hostname=None):
"""Load the Hue platform with the provided bridge for sensor-related platforms."""
if hostname is None:
hostname = "mock-host"
hass.config.components.add(hue.DOMAIN)
config_entry = config_entries.ConfigEntry(
1,
hue.DOMAIN,
"Mock Title",
{"host": hostname},
"test",
config_entries.CONN_CLASS_LOCAL_POLL,
system_options={},
)
mock_bridge.config_entry = config_entry
hass.data[hue.DOMAIN] = {config_entry.entry_id: mock_bridge}
await hass.config_entries.async_forward_entry_setup(config_entry, "binary_sensor")
await hass.config_entries.async_forward_entry_setup(config_entry, "sensor")
# simulate a full setup by manually adding the bridge config entry
hass.config_entries._entries.append(config_entry)
# and make sure it completes before going further
await hass.async_block_till_done()

169
tests/components/hue/test_device_trigger.py Normal file
View File

@ -0,0 +1,169 @@
"""The tests for Philips Hue device triggers."""
import pytest
from homeassistant.components import hue
import homeassistant.components.automation as automation
from homeassistant.components.hue import device_trigger
from homeassistant.setup import async_setup_component
from .conftest import setup_bridge_for_sensors as setup_bridge
from .test_sensor_base import HUE_DIMMER_REMOTE_1, HUE_TAP_REMOTE_1
from tests.common import (
assert_lists_same,
async_get_device_automations,
async_mock_service,
mock_device_registry,
)
REMOTES_RESPONSE = {"7": HUE_TAP_REMOTE_1, "8": HUE_DIMMER_REMOTE_1}
@pytest.fixture
def device_reg(hass):
"""Return an empty, loaded, registry."""
return mock_device_registry(hass)
@pytest.fixture
def calls(hass):
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_get_triggers(hass, mock_bridge, device_reg):
"""Test we get the expected triggers from a hue remote."""
mock_bridge.mock_sensor_responses.append(REMOTES_RESPONSE)
await setup_bridge(hass, mock_bridge)
assert len(mock_bridge.mock_requests) == 1
# 2 remotes, just 1 battery sensor
assert len(hass.states.async_all()) == 1
# Get triggers for specific tap switch
hue_tap_device = device_reg.async_get_device(
{(hue.DOMAIN, "00:00:00:00:00:44:23:08")}, connections={}
)
triggers = await async_get_device_automations(hass, "trigger", hue_tap_device.id)
expected_triggers = [
{
"platform": "device",
"domain": hue.DOMAIN,
"device_id": hue_tap_device.id,
"type": t_type,
"subtype": t_subtype,
}
for t_type, t_subtype in device_trigger.HUE_TAP_REMOTE.keys()
]
assert_lists_same(triggers, expected_triggers)
# Get triggers for specific dimmer switch
hue_dimmer_device = device_reg.async_get_device(
{(hue.DOMAIN, "00:17:88:01:10:3e:3a:dc")}, connections={}
)
triggers = await async_get_device_automations(hass, "trigger", hue_dimmer_device.id)
trigger_batt = {
"platform": "device",
"domain": "sensor",
"device_id": hue_dimmer_device.id,
"type": "battery_level",
"entity_id": "sensor.hue_dimmer_switch_1_battery_level",
}
expected_triggers = [
trigger_batt,
*[
{
"platform": "device",
"domain": hue.DOMAIN,
"device_id": hue_dimmer_device.id,
"type": t_type,
"subtype": t_subtype,
}
for t_type, t_subtype in device_trigger.HUE_DIMMER_REMOTE.keys()
],
]
assert_lists_same(triggers, expected_triggers)
async def test_if_fires_on_state_change(hass, mock_bridge, device_reg, calls):
"""Test for button press trigger firing."""
mock_bridge.mock_sensor_responses.append(REMOTES_RESPONSE)
await setup_bridge(hass, mock_bridge)
assert len(mock_bridge.mock_requests) == 1
assert len(hass.states.async_all()) == 1
# Set an automation with a specific tap switch trigger
hue_tap_device = device_reg.async_get_device(
{(hue.DOMAIN, "00:00:00:00:00:44:23:08")}, connections={}
)
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
"platform": "device",
"domain": hue.DOMAIN,
"device_id": hue_tap_device.id,
"type": "remote_button_short_press",
"subtype": "button_4",
},
"action": {
"service": "test.automation",
"data_template": {
"some": "B4 - {{ trigger.event.data.event }}"
},
},
},
{
"trigger": {
"platform": "device",
"domain": hue.DOMAIN,
"device_id": "mock-device-id",
"type": "remote_button_short_press",
"subtype": "button_1",
},
"action": {
"service": "test.automation",
"data_template": {
"some": "B1 - {{ trigger.event.data.event }}"
},
},
},
]
},
)
# Fake that the remote is being pressed.
new_sensor_response = dict(REMOTES_RESPONSE)
new_sensor_response["7"]["state"] = {
"buttonevent": 18,
"lastupdated": "2019-12-28T22:58:02",
}
mock_bridge.mock_sensor_responses.append(new_sensor_response)
# Force updates to run again
await mock_bridge.sensor_manager.coordinator.async_refresh()
await hass.async_block_till_done()
assert len(mock_bridge.mock_requests) == 2
assert len(calls) == 1
assert calls[0].data["some"] == "B4 - 18"
# Fake another button press.
new_sensor_response = dict(REMOTES_RESPONSE)
new_sensor_response["7"]["state"] = {
"buttonevent": 34,
"lastupdated": "2019-12-28T22:58:05",
}
mock_bridge.mock_sensor_responses.append(new_sensor_response)
# Force updates to run again
await mock_bridge.sensor_manager.coordinator.async_refresh()
await hass.async_block_till_done()
assert len(mock_bridge.mock_requests) == 3
assert len(calls) == 1

46
tests/components/hue/test_light.py
View File

@ -1,13 +1,9 @@
"""Philips Hue lights platform tests."""
import asyncio
from collections import deque
import logging
from unittest.mock import Mock
import aiohue
from aiohue.groups import Groups
from aiohue.lights import Lights
import pytest
from homeassistant import config_entries
from homeassistant.components import hue
@ -175,48 +171,6 @@ LIGHT_GAMUT = color.GamutType(
LIGHT_GAMUT_TYPE = "A"
@pytest.fixture
def mock_bridge(hass):
"""Mock a Hue bridge."""
bridge = Mock(
hass=hass,
available=True,
authorized=True,
allow_unreachable=False,
allow_groups=False,
api=Mock(),
reset_jobs=[],
spec=hue.HueBridge,
)
bridge.mock_requests = []
# We're using a deque so we can schedule multiple responses
# and also means that `popleft()` will blow up if we get more updates
# than expected.
bridge.mock_light_responses = deque()
bridge.mock_group_responses = deque()
async def mock_request(method, path, **kwargs):
kwargs["method"] = method
kwargs["path"] = path
bridge.mock_requests.append(kwargs)
if path == "lights":
return bridge.mock_light_responses.popleft()
if path == "groups":
return bridge.mock_group_responses.popleft()
return None
async def async_request_call(task):
await task()
bridge.async_request_call = async_request_call
bridge.api.config.apiversion = "9.9.9"
bridge.api.lights = Lights({}, mock_request)
bridge.api.groups = Groups({}, mock_request)
return bridge
async def setup_bridge(hass, mock_bridge):
"""Load the Hue light platform with the provided bridge."""
hass.config.components.add(hue.DOMAIN)

73
tests/components/hue/test_sensor_base.py
View File

@ -1,18 +1,14 @@
"""Philips Hue sensors platform tests."""
import asyncio
from collections import deque
import logging
from unittest.mock import Mock
import aiohue
from aiohue.sensors import Sensors
import pytest
from homeassistant import config_entries
from homeassistant.components import hue
from homeassistant.components.hue import sensor_base as hue_sensor_base
from homeassistant.components.hue.hue_event import CONF_HUE_EVENT
from .conftest import create_mock_bridge, setup_bridge_for_sensors as setup_bridge
_LOGGER = logging.getLogger(__name__)
PRESENCE_SENSOR_1_PRESENT = {
@ -281,71 +277,6 @@ SENSOR_RESPONSE = {
}
def create_mock_bridge(hass):
"""Create a mock Hue bridge."""
bridge = Mock(
hass=hass,
available=True,
authorized=True,
allow_unreachable=False,
allow_groups=False,
api=Mock(),
reset_jobs=[],
spec=hue.HueBridge,
)
bridge.sensor_manager = hue_sensor_base.SensorManager(bridge)
bridge.mock_requests = []
# We're using a deque so we can schedule multiple responses
# and also means that `popleft()` will blow up if we get more updates
# than expected.
bridge.mock_sensor_responses = deque()
async def mock_request(method, path, **kwargs):
kwargs["method"] = method
kwargs["path"] = path
bridge.mock_requests.append(kwargs)
if path == "sensors":
return bridge.mock_sensor_responses.popleft()
return None
async def async_request_call(task):
await task()
bridge.async_request_call = async_request_call
bridge.api.config.apiversion = "9.9.9"
bridge.api.sensors = Sensors({}, mock_request)
return bridge
@pytest.fixture
def mock_bridge(hass):
"""Mock a Hue bridge."""
return create_mock_bridge(hass)
async def setup_bridge(hass, mock_bridge, hostname=None):
"""Load the Hue platform with the provided bridge."""
if hostname is None:
hostname = "mock-host"
hass.config.components.add(hue.DOMAIN)
config_entry = config_entries.ConfigEntry(
1,
hue.DOMAIN,
"Mock Title",
{"host": hostname},
"test",
config_entries.CONN_CLASS_LOCAL_POLL,
system_options={},
)
mock_bridge.config_entry = config_entry
hass.data[hue.DOMAIN] = {config_entry.entry_id: mock_bridge}
await hass.config_entries.async_forward_entry_setup(config_entry, "binary_sensor")
await hass.config_entries.async_forward_entry_setup(config_entry, "sensor")
# and make sure it completes before going further
await hass.async_block_till_done()
async def test_no_sensors(hass, mock_bridge):
"""Test the update_items function when no sensors are found."""
mock_bridge.allow_groups = True