starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-24 23:41:48 -05:00
Code Issues Packages Projects Releases Wiki Activity

Migrate KNX telegram trigger to the trigger platform (#181146)

Browse Source 
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Matthias Alphart
2026-09-05 16:05:09 +02:00
committed by GitHub
co-authored by Claude Opus 5
parent 932838840b
commit 653218326b
7 changed files with 290 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/knx/device_trigger.py
+27 -22
View File
@@ -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)
homeassistant/components/knx/icons.json
+5
View File
@@ -54,5 +54,10 @@
"send": {
"service": "mdi:email-arrow-right"
}
},
"triggers": {
"telegram": {
"trigger": "mdi:email-alert-outline"
}
}
}
homeassistant/components/knx/strings.json
+41 -5
View File
@@ -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"
}
}
}
homeassistant/components/knx/trigger.py
+116 -46
View File
@@ -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
homeassistant/components/knx/triggers.yaml
+38
View File
@@ -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:
script/hassfest/triggers.py
-1
View File
@@ -148,7 +148,6 @@ NON_MIGRATED_INTEGRATIONS = {
"device_automation",
"geo_location",
"homeassistant",
"knx",
"lg_netcast",
"litejet",
"persistent_notification",
tests/components/knx/test_trigger.py
+63 -11
View File
@@ -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",