core/homeassistant/components/hue/device_trigger.py

"""Provides device automations for Philips Hue events."""
from __future__ import annotations

from typing import TYPE_CHECKING

from homeassistant.components.device_automation.exceptions import (
    InvalidDeviceAutomationConfig,
)
from homeassistant.const import CONF_DEVICE_ID
from homeassistant.core import CALLBACK_TYPE
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.typing import ConfigType

from .const import DOMAIN
from .v1.device_trigger import (
    async_attach_trigger as async_attach_trigger_v1,
    async_get_triggers as async_get_triggers_v1,
    async_validate_trigger_config as async_validate_trigger_config_v1,
)
from .v2.device_trigger import (
    async_attach_trigger as async_attach_trigger_v2,
    async_get_triggers as async_get_triggers_v2,
    async_validate_trigger_config as async_validate_trigger_config_v2,
)

if TYPE_CHECKING:
    from homeassistant.components.automation import (
        AutomationActionType,
        AutomationTriggerInfo,
    )
    from homeassistant.core import HomeAssistant

    from .bridge import HueBridge


async def async_validate_trigger_config(hass: "HomeAssistant", config: ConfigType):
    """Validate config."""
    if DOMAIN not in hass.data:
        # happens at startup
        return config
    device_id = config[CONF_DEVICE_ID]
    # lookup device in HASS DeviceRegistry
    dev_reg: dr.DeviceRegistry = dr.async_get(hass)
    if (device_entry := dev_reg.async_get(device_id)) is None:
        raise InvalidDeviceAutomationConfig(f"Device ID {device_id} is not valid")

    for conf_entry_id in device_entry.config_entries:
        if conf_entry_id not in hass.data[DOMAIN]:
            continue
        bridge: HueBridge = hass.data[DOMAIN][conf_entry_id]
        if bridge.api_version == 1:
            return await async_validate_trigger_config_v1(bridge, device_entry, config)
        return await async_validate_trigger_config_v2(bridge, device_entry, config)


async def async_attach_trigger(
    hass: "HomeAssistant",
    config: ConfigType,
    action: "AutomationActionType",
    automation_info: "AutomationTriggerInfo",
) -> CALLBACK_TYPE:
    """Listen for state changes based on configuration."""
    device_id = config[CONF_DEVICE_ID]
    # lookup device in HASS DeviceRegistry
    dev_reg: dr.DeviceRegistry = dr.async_get(hass)
    if (device_entry := dev_reg.async_get(device_id)) is None:
        raise InvalidDeviceAutomationConfig(f"Device ID {device_id} is not valid")

    for conf_entry_id in device_entry.config_entries:
        if conf_entry_id not in hass.data[DOMAIN]:
            continue
        bridge: HueBridge = hass.data[DOMAIN][conf_entry_id]
        if bridge.api_version == 1:
            return await async_attach_trigger_v1(
                bridge, device_entry, config, action, automation_info
            )
        return await async_attach_trigger_v2(
            bridge, device_entry, config, action, automation_info
        )
    raise InvalidDeviceAutomationConfig(
        f"Device ID {device_id} is not found on any Hue bridge"
    )


async def async_get_triggers(hass: "HomeAssistant", device_id: str):
    """Get device triggers for given (hass) device id."""
    if DOMAIN not in hass.data:
        return []
    # lookup device in HASS DeviceRegistry
    dev_reg: dr.DeviceRegistry = dr.async_get(hass)
    if (device_entry := dev_reg.async_get(device_id)) is None:
        raise ValueError(f"Device ID {device_id} is not valid")

    # Iterate all config entries for this device
    # and work out the bridge version
    for conf_entry_id in device_entry.config_entries:
        if conf_entry_id not in hass.data[DOMAIN]:
            continue
        bridge: HueBridge = hass.data[DOMAIN][conf_entry_id]

        if bridge.api_version == 1:
            return await async_get_triggers_v1(bridge, device_entry)
        return await async_get_triggers_v2(bridge, device_entry)