"""Device triggers for Reticulum. Exposes the integration's bus events as device triggers so they can be picked from the automation UI ("Do something when a Reticulum device …"). """ from __future__ import annotations import voluptuous as vol from homeassistant.components.device_automation import DEVICE_TRIGGER_BASE_SCHEMA from homeassistant.components.homeassistant.triggers import event as event_trigger from homeassistant.const import ( CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF_TYPE, ) from homeassistant.core import CALLBACK_TYPE, HomeAssistant from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo from homeassistant.helpers.typing import ConfigType from .const import ( DOMAIN, EVENT_ANNOUNCE_RECEIVED, EVENT_MESSAGE_DELIVERED, EVENT_MESSAGE_FAILED, EVENT_MESSAGE_RECEIVED, ) TRIGGER_MESSAGE_RECEIVED = "message_received" TRIGGER_MESSAGE_DELIVERED = "message_delivered" TRIGGER_MESSAGE_FAILED = "message_failed" TRIGGER_ANNOUNCE_RECEIVED = "announce_received" _EVENT_FOR_TYPE: dict[str, str] = { TRIGGER_MESSAGE_RECEIVED: EVENT_MESSAGE_RECEIVED, TRIGGER_MESSAGE_DELIVERED: EVENT_MESSAGE_DELIVERED, TRIGGER_MESSAGE_FAILED: EVENT_MESSAGE_FAILED, TRIGGER_ANNOUNCE_RECEIVED: EVENT_ANNOUNCE_RECEIVED, } TRIGGER_TYPES = set(_EVENT_FOR_TYPE) TRIGGER_SCHEMA = DEVICE_TRIGGER_BASE_SCHEMA.extend( { vol.Required(CONF_TYPE): vol.In(TRIGGER_TYPES), } ) async def async_get_triggers( hass: HomeAssistant, device_id: str ) -> list[dict[str, str]]: """List device triggers for a Reticulum device.""" return [ { CONF_PLATFORM: "device", CONF_DOMAIN: DOMAIN, CONF_DEVICE_ID: device_id, CONF_TYPE: trigger_type, } for trigger_type in TRIGGER_TYPES ] async def async_attach_trigger( hass: HomeAssistant, config: ConfigType, action: TriggerActionType, trigger_info: TriggerInfo, ) -> CALLBACK_TYPE: """Attach a device trigger by wrapping the corresponding bus event.""" event_config = event_trigger.TRIGGER_SCHEMA( { event_trigger.CONF_PLATFORM: "event", event_trigger.CONF_EVENT_TYPE: _EVENT_FOR_TYPE[config[CONF_TYPE]], } ) return await event_trigger.async_attach_trigger( hass, event_config, action, trigger_info, platform_type="device" )