diff --git a/homeassistant/components/knx/__init__.py b/homeassistant/components/knx/__init__.py index 61d49243430..d19b505d7ff 100644 --- a/homeassistant/components/knx/__init__.py +++ b/homeassistant/components/knx/__init__.py @@ -82,6 +82,7 @@ CONF_KNX_EVENT_FILTER: Final = "event_filter" SERVICE_KNX_SEND: Final = "send" SERVICE_KNX_ATTR_PAYLOAD: Final = "payload" SERVICE_KNX_ATTR_TYPE: Final = "type" +SERVICE_KNX_ATTR_RESPONSE: Final = "response" SERVICE_KNX_ATTR_REMOVE: Final = "remove" SERVICE_KNX_EVENT_REGISTER: Final = "event_register" SERVICE_KNX_EXPOSURE_REGISTER: Final = "exposure_register" @@ -142,6 +143,7 @@ SERVICE_KNX_SEND_SCHEMA = vol.Any( ), vol.Required(SERVICE_KNX_ATTR_PAYLOAD): cv.match_all, vol.Required(SERVICE_KNX_ATTR_TYPE): sensor_type_validator, + vol.Optional(SERVICE_KNX_ATTR_RESPONSE, default=False): cv.boolean, } ), vol.Schema( @@ -154,6 +156,7 @@ SERVICE_KNX_SEND_SCHEMA = vol.Any( vol.Required(SERVICE_KNX_ATTR_PAYLOAD): vol.Any( cv.positive_int, [cv.positive_int] ), + vol.Optional(SERVICE_KNX_ATTR_RESPONSE, default=False): cv.boolean, } ), ) @@ -551,6 +554,7 @@ class KNXModule: attr_address = call.data[KNX_ADDRESS] attr_payload = call.data[SERVICE_KNX_ATTR_PAYLOAD] attr_type = call.data.get(SERVICE_KNX_ATTR_TYPE) + attr_response = call.data[SERVICE_KNX_ATTR_RESPONSE] payload: DPTBinary | DPTArray if attr_type is not None: @@ -566,7 +570,9 @@ class KNXModule: for address in attr_address: telegram = Telegram( destination_address=parse_device_group_address(address), - payload=GroupValueWrite(payload), + payload=GroupValueResponse(payload) + if attr_response + else GroupValueWrite(payload), ) await self.xknx.telegrams.put(telegram) diff --git a/homeassistant/components/knx/services.yaml b/homeassistant/components/knx/services.yaml index 11519be48f3..45ef6dcbd12 100644 --- a/homeassistant/components/knx/services.yaml +++ b/homeassistant/components/knx/services.yaml @@ -23,6 +23,12 @@ send: example: "temperature" selector: text: + response: + name: "Send as Response" + description: "If set to `True`, the telegram will be sent as a `GroupValueResponse` instead of a `GroupValueWrite`." + default: false + selector: + boolean: read: name: "Read from KNX bus" description: "Send GroupValueRead requests to the KNX bus. Response can be used from `knx_event` and will be processed in KNX entities." diff --git a/tests/components/knx/conftest.py b/tests/components/knx/conftest.py index a692fa97814..e089963bfa1 100644 --- a/tests/components/knx/conftest.py +++ b/tests/components/knx/conftest.py @@ -109,7 +109,7 @@ class KNXTestKit: # APCI Service tests #################### - async def _assert_telegram( + async def assert_telegram( self, group_address: str, payload: int | tuple[int, ...] | None, @@ -141,19 +141,19 @@ class KNXTestKit: async def assert_read(self, group_address: str) -> None: """Assert outgoing GroupValueRead telegram. One by one in timely order.""" - await self._assert_telegram(group_address, None, GroupValueRead) + await self.assert_telegram(group_address, None, GroupValueRead) async def assert_response( self, group_address: str, payload: int | tuple[int, ...] ) -> None: """Assert outgoing GroupValueResponse telegram. One by one in timely order.""" - await self._assert_telegram(group_address, payload, GroupValueResponse) + await self.assert_telegram(group_address, payload, GroupValueResponse) async def assert_write( self, group_address: str, payload: int | tuple[int, ...] ) -> None: """Assert outgoing GroupValueWrite telegram. One by one in timely order.""" - await self._assert_telegram(group_address, payload, GroupValueWrite) + await self.assert_telegram(group_address, payload, GroupValueWrite) #################### # Incoming telegrams diff --git a/tests/components/knx/test_services.py b/tests/components/knx/test_services.py index c61dc542586..039dd5986ca 100644 --- a/tests/components/knx/test_services.py +++ b/tests/components/knx/test_services.py @@ -1,4 +1,7 @@ """Test KNX services.""" +import pytest +from xknx.telegram.apci import GroupValueResponse, GroupValueWrite + from homeassistant.const import STATE_OFF, STATE_ON from homeassistant.core import HomeAssistant @@ -7,51 +10,114 @@ from .conftest import KNXTestKit from tests.common import async_capture_events -async def test_send(hass: HomeAssistant, knx: KNXTestKit): +@pytest.mark.parametrize( + "service_payload,expected_telegrams,expected_apci", + [ + # send DPT 1 telegram + ( + {"address": "1/2/3", "payload": True, "response": True}, + [("1/2/3", True)], + GroupValueResponse, + ), + ( + {"address": "1/2/3", "payload": True, "response": False}, + [("1/2/3", True)], + GroupValueWrite, + ), + # send DPT 5 telegram + ( + {"address": "1/2/3", "payload": [99], "response": True}, + [("1/2/3", (99,))], + GroupValueResponse, + ), + ( + {"address": "1/2/3", "payload": [99], "response": False}, + [("1/2/3", (99,))], + GroupValueWrite, + ), + # send DPT 5 percent telegram + ( + {"address": "1/2/3", "payload": 99, "type": "percent", "response": True}, + [("1/2/3", (0xFC,))], + GroupValueResponse, + ), + ( + {"address": "1/2/3", "payload": 99, "type": "percent", "response": False}, + [("1/2/3", (0xFC,))], + GroupValueWrite, + ), + # send temperature DPT 9 telegram + ( + { + "address": "1/2/3", + "payload": 21.0, + "type": "temperature", + "response": True, + }, + [("1/2/3", (0x0C, 0x1A))], + GroupValueResponse, + ), + ( + { + "address": "1/2/3", + "payload": 21.0, + "type": "temperature", + "response": False, + }, + [("1/2/3", (0x0C, 0x1A))], + GroupValueWrite, + ), + # send multiple telegrams + ( + { + "address": ["1/2/3", "2/2/2", "3/3/3"], + "payload": 99, + "type": "percent", + "response": True, + }, + [ + ("1/2/3", (0xFC,)), + ("2/2/2", (0xFC,)), + ("3/3/3", (0xFC,)), + ], + GroupValueResponse, + ), + ( + { + "address": ["1/2/3", "2/2/2", "3/3/3"], + "payload": 99, + "type": "percent", + "response": False, + }, + [ + ("1/2/3", (0xFC,)), + ("2/2/2", (0xFC,)), + ("3/3/3", (0xFC,)), + ], + GroupValueWrite, + ), + ], +) +async def test_send( + hass: HomeAssistant, + knx: KNXTestKit, + service_payload, + expected_telegrams, + expected_apci, +): """Test `knx.send` service.""" - test_address = "1/2/3" await knx.setup_integration({}) - # send DPT 1 telegram - await hass.services.async_call( - "knx", "send", {"address": test_address, "payload": True}, blocking=True - ) - await knx.assert_write(test_address, True) - - # send raw DPT 5 telegram - await hass.services.async_call( - "knx", "send", {"address": test_address, "payload": [99]}, blocking=True - ) - await knx.assert_write(test_address, (99,)) - - # send "percent" DPT 5 telegram await hass.services.async_call( "knx", "send", - {"address": test_address, "payload": 99, "type": "percent"}, + service_payload, blocking=True, ) - await knx.assert_write(test_address, (0xFC,)) - # send "temperature" DPT 9 telegram - await hass.services.async_call( - "knx", - "send", - {"address": test_address, "payload": 21.0, "type": "temperature"}, - blocking=True, - ) - await knx.assert_write(test_address, (0x0C, 0x1A)) - - # send multiple telegrams - await hass.services.async_call( - "knx", - "send", - {"address": [test_address, "2/2/2", "3/3/3"], "payload": 99, "type": "percent"}, - blocking=True, - ) - await knx.assert_write(test_address, (0xFC,)) - await knx.assert_write("2/2/2", (0xFC,)) - await knx.assert_write("3/3/3", (0xFC,)) + for expected_response in expected_telegrams: + group_address, payload = expected_response + await knx.assert_telegram(group_address, payload, expected_apci) async def test_read(hass: HomeAssistant, knx: KNXTestKit):