Repository: XKNX/xknx
Pull request #1027 (2022-09-08 04:39:39)

Convenience functions for KNX group communication
<!--
  You are awesome! Thanks for contributing to our project!
  Please, DO NOT DELETE ANY TEXT from this template!.
-->
## Description
<!--
Please include a summary of the change and which issue is fixed.
Please also include relevant motivation and context.
List any dependencies that are required for this change.
-->
convenience functions for KNX group communication

See #961 

## Type of change
<!--
Please tick the applicable options.
NOTE: Ticking multiple options most likely indicates
that your change is to big and it is suggested to split it into several smaller PRs.
-->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] This change requires a documentation update

## Checklist:

- [x] The documentation has been adjusted accordingly
- [x] Tests have been added that prove the fix is effective or that the feature works
- [x] The changes are documented in the changelog (docs/changelog.md)

codecov[bot]: # [Codecov](https://codecov.io/gh/XKNX/xknx/pull/1027?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#1027](https://codecov.io/gh/XKNX/xknx/pull/1027?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (9e01bc1) into [main](https://codecov.io/gh/XKNX/xknx/commit/9c47cd515af87b92b82b8c90af7c327e24aadd41?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (9c47cd5) will **decrease** coverage by `0.59%`.
> The diff coverage is `0.00%`.

[![Impacted file tree graph](https://codecov.io/gh/XKNX/xknx/pull/1027/graphs/tree.svg?width=650&height=150&src=pr&token=irWbIygS84&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)](https://codecov.io/gh/XKNX/xknx/pull/1027?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)

```diff
@@            Coverage Diff             @@
##             main    #1027      +/-   ##
==========================================
- Coverage   96.42%   95.82%   -0.60%     
==========================================
  Files         133      134       +1     
  Lines        8162     8213      +51     
==========================================
  Hits         7870     7870              
- Misses        292      343      +51     
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/1027?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/tools/\_\_init\_\_.py](https://codecov.io/gh/XKNX/xknx/pull/1027/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC90b29scy9fX2luaXRfXy5weQ==) | `0.00% <0.00%> (ø)` | |


Patch:
diff --git a/docs/changelog.md b/docs/changelog.md
index ae3ec666..f33b8928 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -6,6 +6,12 @@ nav_order: 2
 
 # Changelog
 
+## Unreleased changes
+
+### Added
+
+- Convenience functions for KNX group communication (`xknx.tools`)
+
 ## 1.0.2 Route-back reconnect 2022-08-31
 
 ### Bugfixes
diff --git a/docs/index.md b/docs/index.md
index ae9cb809..abe8876f 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -39,19 +39,16 @@ pip install xknx
 ```python
 import asyncio
 from xknx import XKNX
-from xknx.devices import Light
+from xknx.tools import group_value_write
 
 async def main():
-    xknx = XKNX()
-    await xknx.start()
-    light = Light(xknx,
-                  name='HelloWorldLight',
-                  group_address_switch='1/0/9')
-    await light.set_on()
-    await asyncio.sleep(2)
-    await light.set_off()
-    await asyncio.sleep(2)
-    await xknx.stop()
+    async with XKNX() as xknx:
+        # send a binary Telegram
+        await group_value_write(xknx, "1/2/3", True)
+        # send a generic 1-byte Telegram
+        await group_value_write(xknx, "1/2/4", [0x80])
+        # send a Telegram with an encoded value
+        await group_value_write(xknx, "1/2/4", 50, value_type="percent")
 
 asyncio.run(main())
 ```
diff --git a/examples/example_callback.py b/examples/example_callback.py
index 2e4fabd3..5b3d173f 100644
--- a/examples/example_callback.py
+++ b/examples/example_callback.py
@@ -1,4 +1,4 @@
-"""Example for reading the state from the KNX bus."""
+"""Example for running a callback when a devices state changed."""
 import asyncio
 
 from xknx import XKNX
@@ -12,7 +12,7 @@ async def light_callback(light: Light) -> None:
 
 
 async def main() -> None:
-    """Connect to KNX/IP bus and read the state of a Light device."""
+    """Connect to KNX/IP bus, turn on Light device and wait."""
     xknx = XKNX(
         connection_config=ConnectionConfig(
             gateway_ip="10.1.0.41",
diff --git a/examples/example_manual_connection.py b/examples/example_manual_connection.py
index 377ccca2..09d885ae 100644
--- a/examples/example_manual_connection.py
+++ b/examples/example_manual_connection.py
@@ -1,18 +1,18 @@
-"""Example for requesting a temperature."""
+"""Example for connecting to a specific KNX interface."""
 import asyncio
 import logging
 import time
 
 from xknx import XKNX
-from xknx.devices import Sensor
 from xknx.io import ConnectionConfig, ConnectionType
+from xknx.tools import read_group_value
 
 logging.basicConfig(level=logging.INFO)
 logging.getLogger("xknx.log").level = logging.DEBUG
 logging.getLogger("xknx.knx").level = logging.DEBUG
 
 
-async def main():
+async def main() -> None:
     """Connect to specific tunnelling server, time a request for a group address value."""
     connection_config = ConnectionConfig(
         connection_type=ConnectionType.TUNNELING,
@@ -22,19 +22,10 @@ async def main():
     )
     xknx = XKNX(connection_config=connection_config)
 
-    sensor = Sensor(
-        xknx,
-        name="Sensor",
-        group_address_state="1/2/3",
-        value_type="temperature",
-    )
-
     async with xknx:
         start_time = time.time()
-        await sensor.sync(wait_for_result=True)
-        print(
-            f"Value: {sensor.resolve_state()} - took {(time.time() - start_time):0.3f} seconds"
-        )
+        result = await read_group_value(xknx, "5/1/20", value_type="temperature")
+        print(f"Value: {result} - took {(time.time() - start_time):0.3f} seconds")
 
 
 asyncio.run(main())
diff --git a/examples/example_secure.py b/examples/example_secure.py
index 0e8207f7..27211b65 100644
--- a/examples/example_secure.py
+++ b/examples/example_secure.py
@@ -1,13 +1,11 @@
-"""Example for KNX IP Secure."""
+"""Example for connecting to a KNX IP Secure Tunnel."""
 
 import asyncio
 import logging
 
 from xknx import XKNX
-from xknx.dpt import DPTBinary
 from xknx.io import ConnectionConfig, ConnectionType, SecureConfig
-from xknx.telegram import GroupAddress, Telegram, TelegramDirection
-from xknx.telegram.apci import GroupValueWrite
+from xknx.tools import group_value_write
 
 logging.basicConfig(level=logging.INFO)
 logging.getLogger("xknx.log").level = logging.DEBUG
@@ -15,7 +13,7 @@ logging.getLogger("xknx.knx").level = logging.DEBUG
 
 
 async def main() -> None:
-    """Test connection with secure."""
+    """Test connection with IP secure tunnelling."""
 
     connection_config = ConnectionConfig(
         connection_type=ConnectionType.TUNNELING_TCP_SECURE,
@@ -27,17 +25,11 @@ async def main() -> None:
         ),
     )
     xknx = XKNX(connection_config=connection_config)
-    await xknx.start()
 
-    await xknx.knxip_interface.send_telegram(
-        Telegram(
-            GroupAddress("1/0/32"),
-            TelegramDirection.OUTGOING,
-            GroupValueWrite(DPTBinary(1)),
-        )
-    )
-    await asyncio.sleep(5)
+    await xknx.start()
     print("Tunnel connected")
+    await group_value_write(xknx, "1/0/32", True)
+    await asyncio.sleep(5)
     await xknx.stop()
 
 
diff --git a/examples/example_send_telegrams.py b/examples/example_send_telegrams.py
new file mode 100644
index 00000000..76509703
--- /dev/null
+++ b/examples/example_send_telegrams.py
@@ -0,0 +1,22 @@
+"""Example for sending generic telegrams to the KNX bus."""
+import asyncio
+import logging
+
+from xknx import XKNX
+from xknx.tools import group_value_response, group_value_write
+
+logging.basicConfig(level=logging.INFO)
+logging.getLogger("xknx.log").level = logging.DEBUG
+logging.getLogger("xknx.tools").level = logging.DEBUG
+
+
+async def main() -> None:
+    """Connect to KNX bus and send telegrams."""
+    async with XKNX() as xknx:
+        # send a DPT 9.001 temperature value
+        await group_value_write(xknx, "5/1/20", 21.7, value_type="temperature")
+        # send a response DPT 1 binary value
+        await group_value_response(xknx, "5/1/20", True)
+
+
+asyncio.run(main())
diff --git a/examples/example_value_reader.py b/examples/example_value_reader.py
index 246d3012..912c77cf 100644
--- a/examples/example_value_reader.py
+++ b/examples/example_value_reader.py
@@ -4,22 +4,21 @@ import asyncio
 from xknx import XKNX
 from xknx.core import ValueReader
 from xknx.telegram import GroupAddress
+from xknx.tools import read_group_value
 
 
-async def main():
+async def main() -> None:
     """Connect and read value from KNX bus."""
     xknx = XKNX()
     await xknx.start()
 
-    value_reader = ValueReader(xknx, GroupAddress("2/0/8"))
-    telegram = await value_reader.read()
-    if telegram is not None:
-        print(telegram)
+    # get the value only (can be decoded when passing `value_type`)
+    result = await read_group_value(xknx, "5/1/20")
+    print(f"Value: {result}")
 
-    value_reader = ValueReader(xknx, GroupAddress("2/1/8"))
-    telegram = await value_reader.read()
-    if telegram is not None:
-        print(telegram)
+    # get the whole telegram
+    telegram = await ValueReader(xknx, GroupAddress("5/1/20")).read()
+    print(f"Telegram: {telegram}")
 
     await xknx.stop()
 
diff --git a/examples/readme.md b/examples/readme.md
index e65370e5..8786b4df 100644
--- a/examples/readme.md
+++ b/examples/readme.md
@@ -4,12 +4,14 @@
 
 |Example|Description|
 |-|-|
-|[Config](./example_config.py)|Example for config file parser|
+|[Callback](./example_callback.py)|Example for running a callback when a devices state changed|
+|[Connection](./example_manual_connection.py)|Example for specifying connection parameters|
 |[Daemon](./example_daemon.py)|Example for daemon mode within XKNX|
 |[Disconnect](./example_disconnect.py)|Example on how to disconnect/reset all available tunneling channels|
 |[Gateway scanner](./example_gatewayscanner.py)|Example for GatewayScanner|
-|[Telegram monitor](./example_telegram_monitor.py)|Example for the telegram monitor callback|
-|[Tunnel](./example_tunnel.py)|Example on how to connecto to a KNX/IP tunneling device|
+|[Secure Tunnel](./example_secure.py)|Example for connecting to a KNX IP Secure Tunnel|
+|[Send telegrams](./example_send_telegrams.py)|Example for sending generic telegrams|
+|[Telegram monitor](./example_telegram_monitor.py)|An example telegram monitor|
 |[Value reader](./example_value_reader.py)|Example on how to read a value from KNX bus|
 |[MQTT powermeter](./example_powermeter_mqtt.py)|Example of a daemon listening for values from my main power-meter and resend them on a MQTT bus|
 
@@ -25,12 +27,12 @@
 |[Switch light](./example_light_switch.py)|Example for switching a light on and off|
 |[Sensor](./example_sensor.py)|Example for Sensor device|
 |[Switch](./example_switch.py)|Example for Switch device|
-|[Scene](./example_scene.py)|Example for switching a light on and off|
-|[Weather](./example_weather.py)|Example for Weather device and reading sensor data|
+|[Scene](./example_scene.py)|Example for calling a scene|
+|[Weather](./example_weather.py)|Example for reading a Weather devices data|
 
 ## Low-level
 
 |Example|Description|
 |-|-|
-|[Info](./example_info.py)|Example on how to read device information such as serial and programming mode (depends on actor if supported)|
-|[Restart](./example_restart.py)|Example on how to restart an actor|
\ No newline at end of file
+|[Restart](./example_restart.py)|Example on how to restart an KNX device|
+|[Tunnel](./example_tunnel.py)|Example on how to connect to a KNX/IP tunneling device|
diff --git a/xknx/tools/__init__.py b/xknx/tools/__init__.py
new file mode 100644
index 00000000..20514b87
--- /dev/null
+++ b/xknx/tools/__init__.py
@@ -0,0 +1,15 @@
+"""Module for communication tools."""
+# flake8: noqa
+from .group_communication import (
+    group_value_read,
+    group_value_response,
+    group_value_write,
+    read_group_value,
+)
+
+__all__ = [
+    "group_value_read",
+    "group_value_response",
+    "group_value_write",
+    "read_group_value",
+]
diff --git a/xknx/tools/group_communication.py b/xknx/tools/group_communication.py
new file mode 100644
index 00000000..4725d34a
--- /dev/null
+++ b/xknx/tools/group_communication.py
@@ -0,0 +1,117 @@
+"""Package for convenience functions for KNX group communication."""
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+from xknx.core.value_reader import ValueReader
+from xknx.dpt import DPTArray, DPTBase, DPTBinary
+from xknx.telegram import GroupAddress, Telegram
+from xknx.telegram.address import GroupAddressableType
+from xknx.telegram.apci import GroupValueRead, GroupValueResponse, GroupValueWrite
+
+if TYPE_CHECKING:
+    from xknx.xknx import XKNX
+logger = logging.getLogger("xknx.tools")
+
+
+async def group_value_read(
+    xknx: XKNX,
+    group_address: GroupAddressableType,
+) -> None:
+    """Send a GroupValueRead telegram."""
+    telegram = Telegram(
+        destination_address=GroupAddress(group_address),
+        payload=GroupValueRead(),
+    )
+
+    logger.debug("Sending GroupValueRead telegram to %s", group_address)
+    await xknx.telegrams.put(telegram)
+
+
+async def group_value_response(
+    xknx: XKNX,
+    group_address: GroupAddressableType,
+    value: Any,
+    value_type: int | str | type[DPTBase] | None = None,
+) -> None:
+    """Send a GroupValueResponse telegram."""
+    payload = _parse_payload(value, value_type)
+    telegram = Telegram(
+        destination_address=GroupAddress(group_address),
+        payload=GroupValueResponse(payload),
+    )
+    logger.debug(
+        "Sending GroupValueResponse telegram with value '%s' of type '%s' to %s.",
+        value,
+        value_type or "raw",
+        group_address,
+    )
+    await xknx.telegrams.put(telegram)
+
+
+async def group_value_write(
+    xknx: XKNX,
+    group_address: GroupAddressableType,
+    value: Any,
+    value_type: int | str | type[DPTBase] | None = None,
+) -> None:
+    """Send a GroupValueWrite telegram."""
+    payload = _parse_payload(value, value_type)
+    telegram = Telegram(
+        destination_address=GroupAddress(group_address),
+        payload=GroupValueWrite(payload),
+    )
+    logger.debug(
+        "Sending GroupValueWrite telegram with value '%s' of type '%s' to %s.",
+        value,
+        value_type or "raw",
+        group_address,
+    )
+    await xknx.telegrams.put(telegram)
+
+
+async def read_group_value(
+    xknx: XKNX,
+    group_address: GroupAddressableType,
+    value_type: int | str | type[DPTBase] | None = None,
+) -> DPTArray | DPTBinary | Any | None:
+    """Read a value from a KNX group address."""
+    transcoder = _parse_dpt(value_type)
+    value_reader = ValueReader(xknx, GroupAddress(group_address))
+    response = await value_reader.read()
+    if response is not None:
+        assert isinstance(response.payload, (GroupValueWrite, GroupValueResponse))
+        assert response.payload.value is not None
+        if transcoder is not None:
+            return transcoder.from_knx(response.payload.value.value)  # type: ignore[arg-type]
+        return response.payload.value.value
+    return None
+
+
+def _parse_dpt(value_type: int | str | type[DPTBase] | None) -> type[DPTBase] | None:
+    if value_type is None:
+        return None
+    if isinstance(value_type, (int, str)):
+        if transcoder := DPTBase.parse_transcoder(value_type):
+            return transcoder
+    else:
+        try:
+            if issubclass(value_type, DPTBase):
+                return value_type
+        except TypeError:
+            pass
+    raise ValueError(f"Invalid value_type: {value_type}")
+
+
+def _parse_payload(
+    value: Any,
+    value_type: int | str | type[DPTBase] | None = None,
+) -> DPTBinary | DPTArray:
+    if isinstance(value, (DPTArray, DPTBinary)):
+        return value
+    if transcoder := _parse_dpt(value_type):
+        return DPTArray(transcoder.to_knx(value))
+    if isinstance(value, int):
+        return DPTBinary(value)
+    return DPTArray(value)


Test patch:
diff --git a/requirements/testing.txt b/requirements/testing.txt
index aa547dfc..778a8411 100644
--- a/requirements/testing.txt
+++ b/requirements/testing.txt
@@ -4,10 +4,10 @@ isort==5.10.1
 flake8==5.0.4
 flake8-isort==4.2.0
 pydocstyle==6.1.1
-pylint==2.15.0
+pylint==2.15.2
 pytest==7.1.3
 pytest-asyncio==0.19.0
 pytest-cov==3.0.0
 setuptools==65.3.0
-tox==3.25.1
+tox==3.26.0
 mypy==0.971
diff --git a/test/tools_tests/tools_test.py b/test/tools_tests/tools_test.py
new file mode 100644
index 00000000..12681394
--- /dev/null
+++ b/test/tools_tests/tools_test.py
@@ -0,0 +1,113 @@
+"""Test xknx tools package."""
+from unittest.mock import patch
+
+import pytest
+
+from xknx import XKNX
+from xknx.dpt import DPTArray, DPTBinary, DPTTemperature
+from xknx.exceptions import ConversionError
+from xknx.telegram import GroupAddress, Telegram, TelegramDirection, apci
+from xknx.tools import (
+    group_value_read,
+    group_value_response,
+    group_value_write,
+    read_group_value,
+)
+
+
+async def test_group_value_read():
+    """Test group_value_read."""
+    xknx = XKNX()
+    group_address = "1/2/3"
+
+    read = Telegram(
+        destination_address=GroupAddress(group_address),
+        payload=apci.GroupValueRead(),
+    )
+    await group_value_read(xknx, "1/2/3")
+    assert xknx.telegrams.qsize() == 1
+    assert xknx.telegrams.get_nowait() == read
+
+
+async def test_group_value_response():
+    """Test group_value_response."""
+    xknx = XKNX()
+    group_address = "1/2/3"
+
+    response = Telegram(
+        destination_address=GroupAddress(group_address),
+        payload=apci.GroupValueResponse(DPTBinary(1)),
+    )
+    await group_value_response(xknx, "1/2/3", True)
+    assert xknx.telegrams.qsize() == 1
+    assert xknx.telegrams.get_nowait() == response
+
+
+@pytest.mark.parametrize(
+    "value,value_type,expected",
+    [
+        (50, "percent", DPTArray((0x80,))),
+        (True, None, DPTBinary(1)),
+        (False, None, DPTBinary(0)),
+        (20.48, DPTTemperature, DPTArray((0x0C, 0x00))),
+        (-100, 6, DPTArray((0x9C,))),
+        ((0x0C, 0x00), None, DPTArray((0x0C, 0x00))),
+        (DPTBinary(1), None, DPTBinary(1)),
+    ],
+)
+async def test_group_value_write(value, value_type, expected):
+    """Test group_value_write."""
+    xknx = XKNX()
+    group_address = "1/2/3"
+
+    write = Telegram(
+        destination_address=GroupAddress(group_address),
+        payload=apci.GroupValueWrite(expected),
+    )
+    await group_value_write(xknx, "1/2/3", value, value_type=value_type)
+    assert xknx.telegrams.qsize() == 1
+    assert xknx.telegrams.get_nowait() == write
+
+
+@pytest.mark.parametrize(
+    "value,value_type,error_type",
+    [
+        (50, "unknown", ValueError),
+        (50, 9.001, ValueError),  # float is invalid
+        (101, "percent", ConversionError),  # too big
+    ],
+)
+async def test_group_value_write_invalid(value, value_type, error_type):
+    """Test group_value_write."""
+    xknx = XKNX()
+    with pytest.raises(error_type):
+        await group_value_write(xknx, "1/2/3", value, value_type=value_type)
+
+
+@pytest.mark.parametrize(
+    "value,value_type,expected",
+    [
+        (50, "percent", DPTArray((0x80,))),
+        ((0x80,), None, DPTArray((0x80,))),
+        (True, None, DPTBinary(1)),
+    ],
+)
+@patch("xknx.core.value_reader.ValueReader.read")
+async def test_read_group_value(value_reader_read_mock, value, value_type, expected):
+    """Test read_group_value."""
+    xknx = XKNX()
+    test_group_address = "1/2/3"
+
+    response_telegram = Telegram(
+        destination_address=GroupAddress(test_group_address),
+        direction=TelegramDirection.INCOMING,
+        payload=apci.GroupValueResponse(expected),
+    )
+    value_reader_read_mock.return_value = response_telegram
+
+    response_value = await read_group_value(
+        xknx, test_group_address, value_type=value_type
+    )
+    # GroupValueRead telegram is not in queue because ValueReader.read is mocked.
+    # This is tested in ValueReader tests.
+    assert response_value == value