diff --git a/homeassistant/components/knx/device_trigger.py b/homeassistant/components/knx/device_trigger.py index 36e7b7c89258..b0eaba0c60c7 100644 --- a/homeassistant/components/knx/device_trigger.py +++ b/homeassistant/components/knx/device_trigger.py @@ -2,19 +2,18 @@ from typing import Any, Final -import voluptuous as vol +import probatio from homeassistant.components.device_automation import ( DEVICE_TRIGGER_BASE_SCHEMA, InvalidDeviceAutomationConfig, ) from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF_TYPE -from homeassistant.core import CALLBACK_TYPE, HomeAssistant +from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback from homeassistant.helpers import selector from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo from homeassistant.helpers.typing import ConfigType -from . import trigger from .const import DOMAIN, KNX_MODULE_KEY from .trigger import ( CONF_KNX_DESTINATION, @@ -23,19 +22,19 @@ from .trigger import ( CONF_KNX_GROUP_VALUE_WRITE, CONF_KNX_INCOMING, CONF_KNX_OUTGOING, - PLATFORM_TYPE_TRIGGER_TELEGRAM, TELEGRAM_TRIGGER_SCHEMA, - TRIGGER_SCHEMA as TRIGGER_TRIGGER_SCHEMA, + async_subscribe_telegrams, ) TRIGGER_TELEGRAM: Final = "telegram" TRIGGER_SCHEMA: Final = DEVICE_TRIGGER_BASE_SCHEMA.extend( { - vol.Required(CONF_TYPE): TRIGGER_TELEGRAM, + probatio.Required(CONF_TYPE): TRIGGER_TELEGRAM, **TELEGRAM_TRIGGER_SCHEMA, } ) +_TELEGRAM_OPTIONS_SCHEMA: Final = probatio.Schema(TELEGRAM_TRIGGER_SCHEMA) async def async_get_triggers( @@ -62,7 +61,7 @@ async def async_get_triggers( async def async_get_trigger_capabilities( hass: HomeAssistant, config: ConfigType -) -> dict[str, vol.Schema]: +) -> dict[str, probatio.Schema]: """List trigger capabilities.""" project = hass.data[KNX_MODULE_KEY].project options = [ @@ -70,9 +69,9 @@ async def async_get_trigger_capabilities( for ga in project.group_addresses.values() ] return { - "extra_fields": vol.Schema( + "extra_fields": probatio.Schema( { - vol.Optional(CONF_KNX_DESTINATION): selector.SelectSelector( + probatio.Optional(CONF_KNX_DESTINATION): selector.SelectSelector( selector.SelectSelectorConfig( mode=selector.SelectSelectorMode.DROPDOWN, multiple=True, @@ -80,19 +79,19 @@ async def async_get_trigger_capabilities( options=options, ), ), - vol.Optional( + probatio.Optional( CONF_KNX_GROUP_VALUE_WRITE, default=True ): selector.BooleanSelector(), - vol.Optional( + probatio.Optional( CONF_KNX_GROUP_VALUE_RESPONSE, default=True ): selector.BooleanSelector(), - vol.Optional( + probatio.Optional( CONF_KNX_GROUP_VALUE_READ, default=True ): selector.BooleanSelector(), - vol.Optional( + probatio.Optional( CONF_KNX_INCOMING, default=True ): selector.BooleanSelector(), - vol.Optional( + probatio.Optional( CONF_KNX_OUTGOING, default=True ): selector.BooleanSelector(), } @@ -107,20 +106,26 @@ async def async_attach_trigger( trigger_info: TriggerInfo, ) -> CALLBACK_TYPE: """Attach a trigger.""" - # Remove device trigger specific fields and add trigger platform identifier - trigger_config = { + # Remove device trigger specific fields + telegram_options = { key: config[key] for key in (config.keys() & TELEGRAM_TRIGGER_SCHEMA.keys()) - } | {CONF_PLATFORM: PLATFORM_TYPE_TRIGGER_TELEGRAM} + } try: - trigger_config = TRIGGER_TRIGGER_SCHEMA(trigger_config) - except vol.Invalid as err: + telegram_options = _TELEGRAM_OPTIONS_SCHEMA(telegram_options) + except probatio.Invalid as err: raise InvalidDeviceAutomationConfig( translation_domain=DOMAIN, translation_key="device_trigger_invalid_config", translation_placeholders={"error": str(err)}, ) from err - return await trigger.async_attach_trigger( - hass, config=trigger_config, action=action, trigger_info=trigger_info - ) + job = HassJob(action, f"KNX device trigger {trigger_info}") + trigger_data = trigger_info["trigger_data"] + + @callback + def async_telegram_received(telegram_data: dict[str, Any]) -> None: + """Run the action for a matching telegram.""" + hass.async_run_hass_job(job, {"trigger": {**trigger_data, **telegram_data}}) + + return async_subscribe_telegrams(hass, telegram_options, async_telegram_received) diff --git a/homeassistant/components/knx/icons.json b/homeassistant/components/knx/icons.json index 09a6c5b89772..e3b42be8970d 100644 --- a/homeassistant/components/knx/icons.json +++ b/homeassistant/components/knx/icons.json @@ -54,5 +54,10 @@ "send": { "service": "mdi:email-arrow-right" } + }, + "triggers": { + "telegram": { + "trigger": "mdi:email-alert-outline" + } } } diff --git a/homeassistant/components/knx/strings.json b/homeassistant/components/knx/strings.json index 0636c30732fd..be9609ee3e86 100644 --- a/homeassistant/components/knx/strings.json +++ b/homeassistant/components/knx/strings.json @@ -1215,11 +1215,11 @@ }, "extra_fields_descriptions": { "destination": "The trigger will listen to telegrams sent or received on these group addresses. If no address is selected, the trigger will fire for every group address.", - "group_value_read": "Listen on GroupValueRead telegrams.", - "group_value_response": "Listen on GroupValueResponse telegrams.", - "group_value_write": "Listen on GroupValueWrite telegrams.", - "incoming": "Listen on incoming telegrams.", - "outgoing": "Listen on outgoing telegrams." + "group_value_read": "Listen on GroupValueRead telegrams. These request the current value of a group address.", + "group_value_response": "Listen on GroupValueResponse telegrams. These answer a GroupValueRead request.", + "group_value_write": "Listen on GroupValueWrite telegrams. These write a new value to a group address.", + "incoming": "Listen on telegrams received from the KNX bus.", + "outgoing": "Listen on telegrams sent to the KNX bus by Home Assistant." }, "trigger_type": { "telegram": "Telegram" @@ -1481,5 +1481,41 @@ }, "name": "Send to KNX bus" } + }, + "triggers": { + "telegram": { + "description": "Triggers when a KNX telegram is received from or sent to the bus.", + "fields": { + "destination": { + "description": "[%key:component::knx::device_automation::extra_fields_descriptions::destination%]", + "name": "[%key:component::knx::device_automation::extra_fields::destination%]" + }, + "group_value_read": { + "description": "[%key:component::knx::device_automation::extra_fields_descriptions::group_value_read%]", + "name": "[%key:component::knx::device_automation::extra_fields::group_value_read%]" + }, + "group_value_response": { + "description": "[%key:component::knx::device_automation::extra_fields_descriptions::group_value_response%]", + "name": "[%key:component::knx::device_automation::extra_fields::group_value_response%]" + }, + "group_value_write": { + "description": "[%key:component::knx::device_automation::extra_fields_descriptions::group_value_write%]", + "name": "[%key:component::knx::device_automation::extra_fields::group_value_write%]" + }, + "incoming": { + "description": "[%key:component::knx::device_automation::extra_fields_descriptions::incoming%]", + "name": "[%key:component::knx::device_automation::extra_fields::incoming%]" + }, + "outgoing": { + "description": "[%key:component::knx::device_automation::extra_fields_descriptions::outgoing%]", + "name": "[%key:component::knx::device_automation::extra_fields::outgoing%]" + }, + "type": { + "description": "KNX datapoint type used to decode the telegram payload, for example \"9.001\" or \"temperature\". If omitted, the datapoint type from the ETS project is used, when available.", + "name": "Datapoint type" + } + }, + "name": "KNX telegram received or sent" + } } } diff --git a/homeassistant/components/knx/trigger.py b/homeassistant/components/knx/trigger.py index 29745a7bd73b..fe48023b97cc 100644 --- a/homeassistant/components/knx/trigger.py +++ b/homeassistant/components/knx/trigger.py @@ -1,29 +1,34 @@ """Provide KNX automation triggers.""" -from typing import Final +from collections.abc import Callable +from typing import Any, Final, cast, override -import voluptuous as vol +import probatio from xknx.dpt import DPTBase from xknx.telegram import Telegram, TelegramDirection from xknx.telegram.address import DeviceGroupAddress, parse_device_group_address from xknx.telegram.apci import GroupValueRead, GroupValueResponse, GroupValueWrite -from homeassistant.const import CONF_PLATFORM, CONF_TYPE -from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback +from homeassistant.const import CONF_OPTIONS, CONF_TYPE +from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback from homeassistant.helpers import config_validation as cv +from homeassistant.helpers.automation import move_top_level_schema_fields_to_options from homeassistant.helpers.dispatcher import async_dispatcher_connect -from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo -from homeassistant.helpers.typing import ConfigType, VolDictType +from homeassistant.helpers.trigger import ( + Trigger, + TriggerActionRunner, + TriggerConfig, + TriggerNotTriggeredReporter, +) +from homeassistant.helpers.typing import ConfigType -from .const import DOMAIN, SIGNAL_KNX_TELEGRAM +from .const import SIGNAL_KNX_TELEGRAM from .schema import ga_validator from .telegrams import TelegramDict, decode_telegram_payload from .validation import dpt_base_type_validator TRIGGER_TELEGRAM: Final = "telegram" -PLATFORM_TYPE_TRIGGER_TELEGRAM: Final = f"{DOMAIN}.{TRIGGER_TELEGRAM}" - CONF_KNX_DESTINATION: Final = "destination" CONF_KNX_GROUP_VALUE_WRITE: Final = "group_value_write" CONF_KNX_GROUP_VALUE_READ: Final = "group_value_read" @@ -32,61 +37,66 @@ CONF_KNX_INCOMING: Final = "incoming" CONF_KNX_OUTGOING: Final = "outgoing" -TELEGRAM_TRIGGER_SCHEMA: VolDictType = { - vol.Optional(CONF_KNX_DESTINATION): vol.All(cv.ensure_list, [ga_validator]), - vol.Optional(CONF_KNX_GROUP_VALUE_WRITE, default=True): cv.boolean, - vol.Optional(CONF_KNX_GROUP_VALUE_RESPONSE, default=True): cv.boolean, - vol.Optional(CONF_KNX_GROUP_VALUE_READ, default=True): cv.boolean, - vol.Optional(CONF_KNX_INCOMING, default=True): cv.boolean, - vol.Optional(CONF_KNX_OUTGOING, default=True): cv.boolean, +TELEGRAM_TRIGGER_SCHEMA: dict[probatio.Marker, Any] = { + probatio.Required(CONF_KNX_DESTINATION, default=list): probatio.All( + probatio.EnsureList(), [ga_validator] + ), + probatio.Optional(CONF_KNX_GROUP_VALUE_WRITE, default=True): cv.boolean, + probatio.Optional(CONF_KNX_GROUP_VALUE_RESPONSE, default=True): cv.boolean, + probatio.Optional(CONF_KNX_GROUP_VALUE_READ, default=True): cv.boolean, + probatio.Optional(CONF_KNX_INCOMING, default=True): cv.boolean, + probatio.Optional(CONF_KNX_OUTGOING, default=True): cv.boolean, } -# TRIGGER_SCHEMA is exclusive to triggers, the above are used in device triggers too -TRIGGER_SCHEMA = cv.TRIGGER_BASE_SCHEMA.extend( - { - vol.Required(CONF_PLATFORM): PLATFORM_TYPE_TRIGGER_TELEGRAM, - vol.Optional(CONF_TYPE, default=None): vol.Any(dpt_base_type_validator, None), - **TELEGRAM_TRIGGER_SCHEMA, - } +# the DPT type is exclusive to the telegram trigger, the above are used +# in device triggers too +_OPTIONS_SCHEMA_DICT: dict[probatio.Marker, Any] = { + probatio.Optional(CONF_TYPE, default=None): probatio.Maybe(dpt_base_type_validator), + **TELEGRAM_TRIGGER_SCHEMA, +} +_TELEGRAM_TRIGGER_SCHEMA = probatio.Schema( + {probatio.Required(CONF_OPTIONS, default=dict): _OPTIONS_SCHEMA_DICT} ) -async def async_attach_trigger( +@callback +def async_subscribe_telegrams( hass: HomeAssistant, - config: ConfigType, - action: TriggerActionType, - trigger_info: TriggerInfo, + options: ConfigType, + telegram_callback: Callable[[dict[str, Any]], None], ) -> CALLBACK_TYPE: - """Listen for telegrams based on configuration.""" - _addresses: list[str] = config.get(CONF_KNX_DESTINATION, []) + """Call `telegram_callback` for telegrams matching the filter options. + + Shared by the telegram trigger and the interface device trigger. The payload + passed to the callback is the telegram dict, with the values re-decoded when + a DPT type is configured. + """ + # an empty destination list matches every group address dst_addresses: list[DeviceGroupAddress] = [ - parse_device_group_address(address) for address in _addresses + parse_device_group_address(address) for address in options[CONF_KNX_DESTINATION] ] - _transcoder = config.get(CONF_TYPE) + _transcoder = options.get(CONF_TYPE) trigger_transcoder = DPTBase.parse_transcoder(_transcoder) if _transcoder else None - job = HassJob(action, f"KNX trigger {trigger_info}") - trigger_data = trigger_info["trigger_data"] - @callback - def async_call_trigger_action( + def async_telegram_received( telegram: Telegram, telegram_dict: TelegramDict ) -> None: - """Filter Telegram and call trigger action.""" + """Filter Telegram and call the callback.""" payload_apci = type(telegram.payload) if payload_apci is GroupValueWrite: - if config[CONF_KNX_GROUP_VALUE_WRITE] is False: + if options[CONF_KNX_GROUP_VALUE_WRITE] is False: return elif payload_apci is GroupValueResponse: - if config[CONF_KNX_GROUP_VALUE_RESPONSE] is False: + if options[CONF_KNX_GROUP_VALUE_RESPONSE] is False: return elif payload_apci is GroupValueRead: - if config[CONF_KNX_GROUP_VALUE_READ] is False: + if options[CONF_KNX_GROUP_VALUE_READ] is False: return if telegram.direction is TelegramDirection.INCOMING: - if config[CONF_KNX_INCOMING] is False: + if options[CONF_KNX_INCOMING] is False: return - elif config[CONF_KNX_OUTGOING] is False: + elif options[CONF_KNX_OUTGOING] is False: return if dst_addresses and telegram.destination_address not in dst_addresses: @@ -102,14 +112,74 @@ async def async_attach_trigger( transcoder=trigger_transcoder, ) # overwrite decoded payload values in telegram_dict - telegram_trigger_data = {**trigger_data, **telegram_dict, **decoded_payload} - else: - telegram_trigger_data = {**trigger_data, **telegram_dict} + telegram_callback({**telegram_dict, **decoded_payload}) + return - hass.async_run_hass_job(job, {"trigger": telegram_trigger_data}) + telegram_callback(dict(telegram_dict)) return async_dispatcher_connect( hass, signal=SIGNAL_KNX_TELEGRAM, - target=async_call_trigger_action, + target=async_telegram_received, ) + + +class TelegramTrigger(Trigger): + """Trigger for KNX telegrams.""" + + _options: dict[str, Any] + + @override + @classmethod + async def async_validate_complete_config( + cls, hass: HomeAssistant, complete_config: ConfigType + ) -> ConfigType: + """Validate complete config, migrating the legacy top-level fields.""" + complete_config = move_top_level_schema_fields_to_options( + complete_config, _OPTIONS_SCHEMA_DICT + ) + return await super().async_validate_complete_config(hass, complete_config) + + @override + @classmethod + async def async_validate_config( + cls, hass: HomeAssistant, config: ConfigType + ) -> ConfigType: + """Validate config.""" + return cast(ConfigType, _TELEGRAM_TRIGGER_SCHEMA(config)) + + def __init__(self, hass: HomeAssistant, config: TriggerConfig) -> None: + """Initialize the trigger.""" + super().__init__(hass, config) + assert config.options is not None + self._options = config.options + + @override + async def async_attach_runner( + self, + run_action: TriggerActionRunner, + did_not_trigger: TriggerNotTriggeredReporter | None = None, + ) -> CALLBACK_TYPE: + """Attach the trigger to an action runner.""" + + @callback + def async_telegram_received(telegram_data: dict[str, Any]) -> None: + """Run the action for a matching telegram.""" + run_action( + telegram_data, + f"KNX telegram to {telegram_data['destination']}", + ) + + return async_subscribe_telegrams( + self._hass, self._options, async_telegram_received + ) + + +TRIGGERS: dict[str, type[Trigger]] = { + TRIGGER_TELEGRAM: TelegramTrigger, +} + + +async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]: + """Return the triggers for KNX.""" + return TRIGGERS diff --git a/homeassistant/components/knx/triggers.yaml b/homeassistant/components/knx/triggers.yaml new file mode 100644 index 000000000000..fe106907014a --- /dev/null +++ b/homeassistant/components/knx/triggers.yaml @@ -0,0 +1,38 @@ +telegram: + fields: + destination: + required: true + default: [] + example: "1/2/3" + selector: + text: + multiple: true + group_value_write: + required: true + default: true + selector: + boolean: + group_value_response: + required: true + default: true + selector: + boolean: + group_value_read: + required: true + default: true + selector: + boolean: + incoming: + required: true + default: true + selector: + boolean: + outgoing: + required: true + default: true + selector: + boolean: + type: + example: "9.001" + selector: + text: diff --git a/script/hassfest/triggers.py b/script/hassfest/triggers.py index 55573aaccecf..54de60f45d59 100644 --- a/script/hassfest/triggers.py +++ b/script/hassfest/triggers.py @@ -148,7 +148,6 @@ NON_MIGRATED_INTEGRATIONS = { "device_automation", "geo_location", "homeassistant", - "knx", "lg_netcast", "litejet", "persistent_notification", diff --git a/tests/components/knx/test_trigger.py b/tests/components/knx/test_trigger.py index 1824336a5d96..1f57f40ab95a 100644 --- a/tests/components/knx/test_trigger.py +++ b/tests/components/knx/test_trigger.py @@ -1,6 +1,8 @@ """Tests for KNX integration specific triggers.""" +from collections.abc import Callable import logging +from typing import Any import pytest @@ -8,15 +10,56 @@ from homeassistant.components import automation from homeassistant.components.knx import DOMAIN from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_OFF from homeassistant.core import HomeAssistant, ServiceCall +from homeassistant.helpers.trigger import async_get_all_descriptions from homeassistant.setup import async_setup_component from .conftest import KNXTestKit +TriggerStyle = Callable[[dict[str, Any]], dict[str, Any]] +# The telegram trigger accepts its options both at the top level - the config +# format from before the trigger was migrated to a trigger platform - and +# nested in `options`, which is what the automation editor writes. +TRIGGER_STYLES = [ + pytest.param(lambda options: options, id="top_level_options"), + pytest.param(lambda options: {"options": options}, id="nested_options"), +] + + +async def test_telegram_trigger_description( + hass: HomeAssistant, + knx: KNXTestKit, +) -> None: + """Test the telegram trigger is offered to the automation editor.""" + await knx.setup_integration() + + descriptions = await async_get_all_descriptions(hass) + assert descriptions["knx.telegram"] is not None + assert set(descriptions["knx.telegram"]["fields"]) == { + "destination", + "group_value_write", + "group_value_response", + "group_value_read", + "incoming", + "outgoing", + "type", + } + + +@pytest.mark.parametrize("trigger_style", TRIGGER_STYLES) +@pytest.mark.parametrize( + "catch_all_options", + [ + pytest.param({}, id="destination_omitted"), + pytest.param({"destination": []}, id="destination_empty"), + ], +) async def test_telegram_trigger( hass: HomeAssistant, service_calls: list[ServiceCall], knx: KNXTestKit, + catch_all_options: dict[str, Any], + trigger_style: TriggerStyle, ) -> None: """Test telegram triggers firing.""" await knx.setup_integration() @@ -32,6 +75,7 @@ async def test_telegram_trigger( { "trigger": { "platform": "knx.telegram", + **trigger_style(catch_all_options), }, "action": { "service": "test.automation", @@ -46,12 +90,17 @@ async def test_telegram_trigger( "trigger": { "platform": "knx.telegram", "id": "test-id", - "destination": ["1/2/3", 2564], # 2564 -> "1/2/4" in raw format - "group_value_write": True, - "group_value_response": False, - "group_value_read": False, - "incoming": True, - "outgoing": True, + **trigger_style( + { + # 2564 -> "1/2/4" in raw format + "destination": ["1/2/3", 2564], + "group_value_write": True, + "group_value_response": False, + "group_value_read": False, + "incoming": True, + "outgoing": True, + } + ), }, "action": { "service": "test.automation", @@ -89,11 +138,12 @@ async def test_telegram_trigger( assert test_call.data["id"] == 0 +@pytest.mark.parametrize("trigger_style", TRIGGER_STYLES) @pytest.mark.parametrize( ("payload", "type_option", "expected_value", "expected_unit"), [ ((0x4C,), {"type": "percent"}, 30, "%"), - ((0x03,), {}, None, None), # "dpt" omitted defaults to None + ((0x03,), {}, None, None), # "type" omitted defaults to None ((0x0C, 0x1A), {"type": "temperature"}, 21.00, "°C"), ], ) @@ -102,9 +152,10 @@ async def test_telegram_trigger_dpt_option( service_calls: list[ServiceCall], knx: KNXTestKit, payload: tuple[int, ...], - type_option: dict[str, bool], + type_option: dict[str, str], expected_value: int | None, expected_unit: str | None, + trigger_style: TriggerStyle, ) -> None: """Test telegram trigger type option.""" await knx.setup_integration() @@ -117,7 +168,7 @@ async def test_telegram_trigger_dpt_option( { "trigger": { "platform": "knx.telegram", - **type_option, + **trigger_style(type_option), }, "action": { "service": "test.automation", @@ -147,6 +198,7 @@ async def test_telegram_trigger_dpt_option( assert test_call.data["trigger"]["unit"] is None +@pytest.mark.parametrize("trigger_style", TRIGGER_STYLES) @pytest.mark.parametrize( "group_value_options", [ @@ -190,6 +242,7 @@ async def test_telegram_trigger_options( knx: KNXTestKit, group_value_options: dict[str, bool], direction_options: dict[str, bool], + trigger_style: TriggerStyle, ) -> None: """Test telegram trigger options.""" await knx.setup_integration() @@ -202,8 +255,7 @@ async def test_telegram_trigger_options( { "trigger": { "platform": "knx.telegram", - **group_value_options, - **direction_options, + **trigger_style({**group_value_options, **direction_options}), }, "action": { "service": "test.automation",