diff --git a/homeassistant/components/isy994/const.py b/homeassistant/components/isy994/const.py index 9a0acf73601857..45987907f3415b 100644 --- a/homeassistant/components/isy994/const.py +++ b/homeassistant/components/isy994/const.py @@ -695,3 +695,5 @@ HTTP_PORT = 80 SCHEME_HTTPS = "https" HTTPS_PORT = 443 + +EVENT_ISY994_CONTROL = "isy994_control" diff --git a/homeassistant/components/isy994/device_trigger.py b/homeassistant/components/isy994/device_trigger.py new file mode 100644 index 00000000000000..24265b74710348 --- /dev/null +++ b/homeassistant/components/isy994/device_trigger.py @@ -0,0 +1,191 @@ +"""Provides device triggers for ISY994 Insteon devices. + +Triggers are exposed for any entity whose underlying ISY node has a +``node_def_id`` in :data:`SUPPORTED_NODE_DEF_IDS`. This covers SwitchLinc +dimmers and relays, KeypadLinc loads (dimmer or relay), and the secondary +``KeypadButton_ADV`` child nodes that share the same on/off/fast/fade +command set. +""" + +from __future__ import annotations + +from collections.abc import Iterator +from typing import Final, cast + +from pyisy.constants import ( + CMD_FADE_DOWN, + CMD_FADE_STOP, + CMD_FADE_UP, + CMD_OFF, + CMD_OFF_FAST, + CMD_ON, + CMD_ON_FAST, +) +import voluptuous as vol + +from homeassistant.components.device_automation import ( + DEVICE_TRIGGER_BASE_SCHEMA, + InvalidDeviceAutomationConfig, +) +from homeassistant.components.homeassistant.triggers import event as event_trigger +from homeassistant.config_entries import ConfigEntryState +from homeassistant.const import ( + CONF_DEVICE_ID, + CONF_DOMAIN, + CONF_ENTITY_ID, + CONF_PLATFORM, + CONF_TYPE, +) +from homeassistant.core import CALLBACK_TYPE, HomeAssistant +from homeassistant.helpers import device_registry as dr, entity_registry as er +from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo +from homeassistant.helpers.typing import ConfigType + +from .const import DOMAIN, EVENT_ISY994_CONTROL +from .models import IsyData + +CONF_SUBTYPE: Final = "subtype" + +TRIGGER_TYPES: Final[dict[str, str]] = { + "on": CMD_ON, + "off": CMD_OFF, + "fast_on": CMD_ON_FAST, + "fast_off": CMD_OFF_FAST, + "fade_up": CMD_FADE_UP, + "fade_down": CMD_FADE_DOWN, + "fade_stop": CMD_FADE_STOP, +} + +SUPPORTED_NODE_DEF_IDS: Final = frozenset( + { + "BallastRelayLampSwitch_ADV", + "DimmerLampSwitch_ADV", + "DimmerSwitchOnly_ADV", + "KeypadButton_ADV", + "KeypadDimmer_ADV", + "KeypadRelay_ADV", + "RelayLampOnly_ADV", + "RelayLampSwitch_ADV", + "RelaySwitchOnlyPlusQuery_ADV", + } +) + +TRIGGER_SCHEMA = DEVICE_TRIGGER_BASE_SCHEMA.extend( + { + vol.Required(CONF_TYPE): vol.In(TRIGGER_TYPES), + vol.Required(CONF_SUBTYPE): str, + vol.Optional(CONF_ENTITY_ID): str, + } +) + + +def _resolve_isy_data(hass: HomeAssistant, device_id: str) -> IsyData | None: + """Return the IsyData backing the given Home Assistant device, if any.""" + device = dr.async_get(hass).async_get(device_id) + if device is None: + return None + for entry_id in device.config_entries: + entry = hass.config_entries.async_get_entry(entry_id) + if ( + entry is not None + and entry.domain == DOMAIN + and entry.state is ConfigEntryState.LOADED + ): + return cast(IsyData, entry.runtime_data) + return None + + +def _supported_button_entries( + hass: HomeAssistant, device_id: str, isy_data: IsyData +) -> Iterator[tuple[er.RegistryEntry, str]]: + """Yield (entity, node_address) for each entity backed by a supported node.""" + prefix = f"{isy_data.uuid}_" + for entry in er.async_entries_for_device( + er.async_get(hass), device_id, include_disabled_entities=False + ): + if entry.platform != DOMAIN or not entry.unique_id.startswith(prefix): + continue + address = entry.unique_id[len(prefix) :] + node = isy_data.root.nodes.get_by_id(address) + if node is None: + continue + if getattr(node, "node_def_id", None) in SUPPORTED_NODE_DEF_IDS: + yield entry, address + + +async def async_get_triggers( + hass: HomeAssistant, device_id: str +) -> list[dict[str, str]]: + """List device triggers for supported ISY Insteon load/button nodes.""" + isy_data = _resolve_isy_data(hass, device_id) + if isy_data is None: + return [] + + triggers: list[dict[str, str]] = [] + for entry, address in _supported_button_entries(hass, device_id, isy_data): + triggers.extend( + { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_DEVICE_ID: device_id, + CONF_ENTITY_ID: entry.entity_id, + CONF_TYPE: trigger_type, + CONF_SUBTYPE: address, + } + for trigger_type in TRIGGER_TYPES + ) + return triggers + + +async def async_get_trigger_capabilities( + hass: HomeAssistant, config: ConfigType +) -> dict[str, vol.Schema]: + """No additional fields — type and subtype fully describe the trigger.""" + return {} + + +async def async_attach_trigger( + hass: HomeAssistant, + config: ConfigType, + action: TriggerActionType, + trigger_info: TriggerInfo, +) -> CALLBACK_TYPE: + """Attach a trigger that filters isy994_control events.""" + device_id: str = config[CONF_DEVICE_ID] + isy_data = _resolve_isy_data(hass, device_id) + if isy_data is None: + raise InvalidDeviceAutomationConfig( + f"ISY device {device_id} not found or not loaded" + ) + + address = config[CONF_SUBTYPE] + node = isy_data.root.nodes.get_by_id(address) + if node is None or getattr(node, "node_def_id", None) not in SUPPORTED_NODE_DEF_IDS: + raise InvalidDeviceAutomationConfig( + f"ISY node {address} is not a supported device-trigger source" + ) + + target_unique_id = f"{isy_data.uuid}_{address}" + target_entity_id: str | None = None + for entry in er.async_entries_for_device(er.async_get(hass), device_id): + if entry.platform == DOMAIN and entry.unique_id == target_unique_id: + target_entity_id = entry.entity_id + break + if target_entity_id is None: + raise InvalidDeviceAutomationConfig( + f"No ISY entity found for device {device_id} subtype {address}" + ) + + event_config = event_trigger.TRIGGER_SCHEMA( + { + event_trigger.CONF_PLATFORM: "event", + event_trigger.CONF_EVENT_TYPE: EVENT_ISY994_CONTROL, + event_trigger.CONF_EVENT_DATA: { + CONF_ENTITY_ID: target_entity_id, + "control": TRIGGER_TYPES[config[CONF_TYPE]], + }, + } + ) + return await event_trigger.async_attach_trigger( + hass, event_config, action, trigger_info, platform_type="device" + ) diff --git a/homeassistant/components/isy994/entity.py b/homeassistant/components/isy994/entity.py index 139a17846d6220..86f8135bbe147b 100644 --- a/homeassistant/components/isy994/entity.py +++ b/homeassistant/components/isy994/entity.py @@ -25,7 +25,7 @@ from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.entity import Entity, EntityDescription -from .const import DOMAIN +from .const import DOMAIN, EVENT_ISY994_CONTROL class ISYEntity(Entity): @@ -81,7 +81,7 @@ def async_on_control(self, event: NodeProperty) -> None: # New state attributes may be available, update the state. self.async_write_ha_state() - self.hass.bus.async_fire("isy994_control", event_data) + self.hass.bus.async_fire(EVENT_ISY994_CONTROL, event_data) class ISYNodeEntity(ISYEntity): diff --git a/homeassistant/components/isy994/strings.json b/homeassistant/components/isy994/strings.json index f7c8aa228386b5..0d82f859477abc 100644 --- a/homeassistant/components/isy994/strings.json +++ b/homeassistant/components/isy994/strings.json @@ -33,6 +33,17 @@ } } }, + "device_automation": { + "trigger_type": { + "fade_down": "{entity_name} was switched Fade Down", + "fade_stop": "{entity_name} was switched Fade Stop", + "fade_up": "{entity_name} was switched Fade Up", + "fast_off": "{entity_name} was switched Fast Off", + "fast_on": "{entity_name} was switched Fast On", + "off": "{entity_name} was switched Off", + "on": "{entity_name} was switched On" + } + }, "options": { "step": { "init": { diff --git a/tests/components/isy994/test_device_trigger.py b/tests/components/isy994/test_device_trigger.py new file mode 100644 index 00000000000000..1925452afde622 --- /dev/null +++ b/tests/components/isy994/test_device_trigger.py @@ -0,0 +1,506 @@ +"""Tests for the ISY994 device triggers.""" + +from unittest.mock import MagicMock + +import pytest + +from homeassistant.components.automation import DOMAIN as AUTOMATION_DOMAIN +from homeassistant.components.device_automation import ( + DeviceAutomationType, + InvalidDeviceAutomationConfig, +) +from homeassistant.components.isy994.const import DOMAIN, EVENT_ISY994_CONTROL +from homeassistant.components.isy994.device_trigger import ( + CONF_SUBTYPE, + TRIGGER_TYPES, + async_attach_trigger, + async_get_trigger_capabilities, + async_get_triggers, +) +from homeassistant.config_entries import ConfigEntryState +from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF_TYPE +from homeassistant.core import HomeAssistant, ServiceCall +from homeassistant.helpers import device_registry as dr, entity_registry as er +from homeassistant.setup import async_setup_component + +from .conftest import MOCK_UUID + +from tests.common import MockConfigEntry, async_get_device_automations + +ROOT_ADDRESS = "11 22 33 1" +ROOT_UNIQUE_ID = f"{MOCK_UUID}_{ROOT_ADDRESS}" + + +def _make_node(address: str, name: str, node_def_id: str | None) -> MagicMock: + node = MagicMock() + node.address = address + node.name = name + node.node_def_id = node_def_id + return node + + +def _wire_runtime( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_isy: MagicMock, + nodes_by_addr: dict[str, MagicMock], +) -> dr.DeviceEntry: + """Set up runtime_data + a single device entry covering all given nodes.""" + mock_config_entry.add_to_hass(hass) + + mock_isy.nodes.get_by_id = MagicMock(side_effect=nodes_by_addr.get) + + runtime_data = MagicMock() + runtime_data.root = mock_isy + runtime_data.uuid = MOCK_UUID + runtime_data.devices = {ROOT_ADDRESS: MagicMock()} + mock_config_entry.runtime_data = runtime_data + mock_config_entry.mock_state(hass, ConfigEntryState.LOADED) + + return dr.async_get(hass).async_get_or_create( + config_entry_id=mock_config_entry.entry_id, + identifiers={(DOMAIN, ROOT_UNIQUE_ID)}, + name="Test Insteon Device", + ) + + +def _register_entity( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + device: dr.DeviceEntry, + *, + domain: str, + address: str, + object_id: str, +) -> str: + entry = er.async_get(hass).async_get_or_create( + domain=domain, + platform=DOMAIN, + unique_id=f"{MOCK_UUID}_{address}", + config_entry=mock_config_entry, + device_id=device.id, + suggested_object_id=object_id, + ) + return entry.entity_id + + +async def test_get_triggers_dimmer_switch( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_isy: MagicMock, +) -> None: + """A SwitchLinc dimmer exposes one trigger per type.""" + nodes = { + ROOT_ADDRESS: _make_node(ROOT_ADDRESS, "Office Dimmer", "DimmerSwitchOnly_ADV") + } + device = _wire_runtime(hass, mock_config_entry, mock_isy, nodes) + entity_id = _register_entity( + hass, + mock_config_entry, + device, + domain="light", + address=ROOT_ADDRESS, + object_id="office_dimmer", + ) + + triggers = [ + t + for t in await async_get_device_automations( + hass, DeviceAutomationType.TRIGGER, device.id + ) + if t.get(CONF_DOMAIN) == DOMAIN + ] + + assert {t[CONF_TYPE] for t in triggers} == set(TRIGGER_TYPES) + assert all(t[CONF_SUBTYPE] == ROOT_ADDRESS for t in triggers) + assert all(t[CONF_DEVICE_ID] == device.id for t in triggers) + assert all(t["entity_id"] == entity_id for t in triggers) + + +async def test_get_triggers_keypad_dimmer_with_buttons( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_isy: MagicMock, +) -> None: + """A KPL dimmer exposes triggers for the load and each secondary button.""" + button_b_addr = "11 22 33 2" + button_c_addr = "11 22 33 3" + nodes = { + ROOT_ADDRESS: _make_node(ROOT_ADDRESS, "KPL Load", "KeypadDimmer_ADV"), + button_b_addr: _make_node(button_b_addr, "KPL B", "KeypadButton_ADV"), + button_c_addr: _make_node(button_c_addr, "KPL C", "KeypadButton_ADV"), + } + device = _wire_runtime(hass, mock_config_entry, mock_isy, nodes) + load_entity = _register_entity( + hass, + mock_config_entry, + device, + domain="light", + address=ROOT_ADDRESS, + object_id="kpl_load", + ) + button_b_entity = _register_entity( + hass, + mock_config_entry, + device, + domain="sensor", + address=button_b_addr, + object_id="kpl_b", + ) + button_c_entity = _register_entity( + hass, + mock_config_entry, + device, + domain="sensor", + address=button_c_addr, + object_id="kpl_c", + ) + + triggers = [ + t + for t in await async_get_device_automations( + hass, DeviceAutomationType.TRIGGER, device.id + ) + if t.get(CONF_DOMAIN) == DOMAIN + ] + + by_subtype: dict[str, set[str]] = {} + entity_by_subtype: dict[str, set[str]] = {} + for t in triggers: + by_subtype.setdefault(t[CONF_SUBTYPE], set()).add(t[CONF_TYPE]) + entity_by_subtype.setdefault(t[CONF_SUBTYPE], set()).add(t["entity_id"]) + assert by_subtype == { + ROOT_ADDRESS: set(TRIGGER_TYPES), + button_b_addr: set(TRIGGER_TYPES), + button_c_addr: set(TRIGGER_TYPES), + } + assert entity_by_subtype == { + ROOT_ADDRESS: {load_entity}, + button_b_addr: {button_b_entity}, + button_c_addr: {button_c_entity}, + } + + +async def test_get_triggers_unsupported_node_def_returns_empty( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_isy: MagicMock, +) -> None: + """Nodes whose node_def_id is not in the supported set yield no triggers.""" + nodes = { + ROOT_ADDRESS: _make_node(ROOT_ADDRESS, "Motion Sensor", "MotionSensor_ADV"), + } + device = _wire_runtime(hass, mock_config_entry, mock_isy, nodes) + _register_entity( + hass, + mock_config_entry, + device, + domain="binary_sensor", + address=ROOT_ADDRESS, + object_id="motion", + ) + + triggers = [ + t + for t in await async_get_device_automations( + hass, DeviceAutomationType.TRIGGER, device.id + ) + if t.get(CONF_DOMAIN) == DOMAIN + ] + assert triggers == [] + + +@pytest.mark.parametrize( + ("trigger_type", "control_code", "should_fire"), + [ + ("fast_on", "DFON", True), + ("fast_on", "DON", False), + ("on", "DON", True), + ("fade_up", "FDUP", True), + ("fade_stop", "FDSTOP", True), + ("fade_stop", "FDUP", False), + ], +) +async def test_trigger_fires_on_matching_control( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_isy: MagicMock, + trigger_type: str, + control_code: str, + should_fire: bool, +) -> None: + """A configured trigger fires only on matching entity_id + control code.""" + nodes = { + ROOT_ADDRESS: _make_node( + ROOT_ADDRESS, "Bedroom Dimmer", "DimmerSwitchOnly_ADV" + ), + } + device = _wire_runtime(hass, mock_config_entry, mock_isy, nodes) + entity_id = _register_entity( + hass, + mock_config_entry, + device, + domain="light", + address=ROOT_ADDRESS, + object_id="bedroom_dimmer", + ) + + calls: list[ServiceCall] = [] + + async def _capture(call: ServiceCall) -> None: + calls.append(call) + + hass.services.async_register("test", "automation", _capture) + + assert await async_setup_component( + hass, + AUTOMATION_DOMAIN, + { + AUTOMATION_DOMAIN: [ + { + "alias": "t", + "trigger": { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_DEVICE_ID: device.id, + CONF_TYPE: trigger_type, + CONF_SUBTYPE: ROOT_ADDRESS, + }, + "action": {"service": "test.automation"}, + } + ] + }, + ) + await hass.async_block_till_done() + + hass.bus.async_fire( + EVENT_ISY994_CONTROL, + {"entity_id": entity_id, "control": control_code}, + ) + await hass.async_block_till_done() + + assert len(calls) == (1 if should_fire else 0) + + +async def test_trigger_isolated_per_button( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_isy: MagicMock, +) -> None: + """A trigger bound to KPL button B does not fire when button C is pressed.""" + button_b_addr = "11 22 33 2" + button_c_addr = "11 22 33 3" + nodes = { + ROOT_ADDRESS: _make_node(ROOT_ADDRESS, "KPL Load", "KeypadDimmer_ADV"), + button_b_addr: _make_node(button_b_addr, "KPL B", "KeypadButton_ADV"), + button_c_addr: _make_node(button_c_addr, "KPL C", "KeypadButton_ADV"), + } + device = _wire_runtime(hass, mock_config_entry, mock_isy, nodes) + _register_entity( + hass, + mock_config_entry, + device, + domain="light", + address=ROOT_ADDRESS, + object_id="kpl_load", + ) + button_b_entity = _register_entity( + hass, + mock_config_entry, + device, + domain="sensor", + address=button_b_addr, + object_id="kpl_b", + ) + button_c_entity = _register_entity( + hass, + mock_config_entry, + device, + domain="sensor", + address=button_c_addr, + object_id="kpl_c", + ) + + calls: list[ServiceCall] = [] + + async def _capture(call: ServiceCall) -> None: + calls.append(call) + + hass.services.async_register("test", "automation", _capture) + + assert await async_setup_component( + hass, + AUTOMATION_DOMAIN, + { + AUTOMATION_DOMAIN: [ + { + "alias": "t", + "trigger": { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_DEVICE_ID: device.id, + CONF_TYPE: "fast_on", + CONF_SUBTYPE: button_b_addr, + }, + "action": {"service": "test.automation"}, + } + ] + }, + ) + await hass.async_block_till_done() + + hass.bus.async_fire( + EVENT_ISY994_CONTROL, + {"entity_id": button_c_entity, "control": "DFON"}, + ) + await hass.async_block_till_done() + assert calls == [] + + hass.bus.async_fire( + EVENT_ISY994_CONTROL, + {"entity_id": button_b_entity, "control": "DFON"}, + ) + await hass.async_block_till_done() + assert len(calls) == 1 + + +async def test_get_triggers_unknown_device_returns_empty( + hass: HomeAssistant, +) -> None: + """An unknown device id yields no triggers.""" + assert await async_get_triggers(hass, "does-not-exist") == [] + + +async def test_get_triggers_returns_empty_when_entry_not_loaded( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_isy: MagicMock, +) -> None: + """A device whose ISY config entry is not LOADED yields no triggers.""" + nodes = { + ROOT_ADDRESS: _make_node(ROOT_ADDRESS, "Office Dimmer", "DimmerSwitchOnly_ADV") + } + device = _wire_runtime(hass, mock_config_entry, mock_isy, nodes) + mock_config_entry.mock_state(hass, ConfigEntryState.SETUP_ERROR) + + triggers = await async_get_device_automations( + hass, DeviceAutomationType.TRIGGER, device.id + ) + assert [t for t in triggers if t.get(CONF_DOMAIN) == DOMAIN] == [] + + +async def test_get_triggers_skips_non_isy_and_unresolvable_entities( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_isy: MagicMock, +) -> None: + """Entities from other platforms or with no backing node are skipped.""" + missing_addr = "11 22 33 9" + nodes = { + ROOT_ADDRESS: _make_node(ROOT_ADDRESS, "Office Dimmer", "DimmerSwitchOnly_ADV"), + } + device = _wire_runtime(hass, mock_config_entry, mock_isy, nodes) + _register_entity( + hass, + mock_config_entry, + device, + domain="light", + address=ROOT_ADDRESS, + object_id="office_dimmer", + ) + er.async_get(hass).async_get_or_create( + domain="sensor", + platform="other_integration", + unique_id="other-1", + config_entry=mock_config_entry, + device_id=device.id, + ) + _register_entity( + hass, + mock_config_entry, + device, + domain="sensor", + address=missing_addr, + object_id="ghost", + ) + + triggers = [ + t + for t in await async_get_device_automations( + hass, DeviceAutomationType.TRIGGER, device.id + ) + if t.get(CONF_DOMAIN) == DOMAIN + ] + assert {t[CONF_SUBTYPE] for t in triggers} == {ROOT_ADDRESS} + + +async def test_get_trigger_capabilities_is_empty(hass: HomeAssistant) -> None: + """Triggers expose no extra capability fields.""" + assert await async_get_trigger_capabilities(hass, {}) == {} + + +async def test_attach_trigger_unknown_device_raises(hass: HomeAssistant) -> None: + """Attaching a trigger for an unknown device raises.""" + with pytest.raises(InvalidDeviceAutomationConfig): + await async_attach_trigger( + hass, + { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_DEVICE_ID: "does-not-exist", + CONF_TYPE: "on", + CONF_SUBTYPE: ROOT_ADDRESS, + }, + MagicMock(), + MagicMock(), + ) + + +async def test_attach_trigger_unsupported_subtype_raises( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_isy: MagicMock, +) -> None: + """Subtypes whose node_def_id is not supported are rejected.""" + nodes = { + ROOT_ADDRESS: _make_node(ROOT_ADDRESS, "Motion", "MotionSensor_ADV"), + } + device = _wire_runtime(hass, mock_config_entry, mock_isy, nodes) + with pytest.raises(InvalidDeviceAutomationConfig): + await async_attach_trigger( + hass, + { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_DEVICE_ID: device.id, + CONF_TYPE: "on", + CONF_SUBTYPE: ROOT_ADDRESS, + }, + MagicMock(), + MagicMock(), + ) + + +async def test_attach_trigger_missing_entity_raises( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_isy: MagicMock, +) -> None: + """A supported node with no registered entity is rejected.""" + nodes = { + ROOT_ADDRESS: _make_node(ROOT_ADDRESS, "Office Dimmer", "DimmerSwitchOnly_ADV"), + } + device = _wire_runtime(hass, mock_config_entry, mock_isy, nodes) + with pytest.raises(InvalidDeviceAutomationConfig): + await async_attach_trigger( + hass, + { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_DEVICE_ID: device.id, + CONF_TYPE: "on", + CONF_SUBTYPE: ROOT_ADDRESS, + }, + MagicMock(), + MagicMock(), + )