Allow sending GroupValueResponse telegrams with knx.send service (#62639)
* Add knx.respond service * Combine knx.send and knx.respond services * Rename knx attribute and fix tests * Use parametrization in tests Co-authored-by: Marvin Wichmann <marvin.wichmann@unic.com>
This commit is contained in:
Mirko Lenz
GitHub
parent
7185e1140d
commit
ab4effc7e2
4 changed files with 119 additions and 41 deletions
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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."
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue