Repository: XKNX/xknx
Pull request #1048 (2022-10-08 20:23:53)

Support InternalGroupAddress in tools
<!--
  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.
-->
- Support InternalGroupAddress in xknx.tools package.

Fixes # (issue)

## 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:

- [ ] 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/1048?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#1048](https://codecov.io/gh/XKNX/xknx/pull/1048?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (dc1727b) into [main](https://codecov.io/gh/XKNX/xknx/commit/bbafc8acb22aaf86b435f5606ce1e34f80bd3211?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (bbafc8a) will **not change** coverage.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/XKNX/xknx/pull/1048/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/1048?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)

```diff
@@           Coverage Diff           @@
##             main    #1048   +/-   ##
=======================================
  Coverage   96.36%   96.36%           
=======================================
  Files         138      138           
  Lines        8579     8579           
=======================================
  Hits         8267     8267           
  Misses        312      312           
```


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


Patch:
diff --git a/docs/changelog.md b/docs/changelog.md
index 90009ebe..aa116cd3 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -14,6 +14,7 @@ nav_order: 2
   Thanks to Weinzierl for providing us a router for testing purposes!
 - Parse `latency` from a .knxkeys keyring files `Backbone` tag.
 - Use `multicast_group` from a .knxkeys keyring files `MulticastAddress` tag (Routing).
+- Support InternalGroupAddress in xknx.tools package.
 
 ### Protocol
 
diff --git a/xknx/tools/group_communication.py b/xknx/tools/group_communication.py
index 4725d34a..41e54538 100644
--- a/xknx/tools/group_communication.py
+++ b/xknx/tools/group_communication.py
@@ -6,8 +6,8 @@ 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 import Telegram
+from xknx.telegram.address import DeviceAddressableType, parse_device_group_address
 from xknx.telegram.apci import GroupValueRead, GroupValueResponse, GroupValueWrite
 
 if TYPE_CHECKING:
@@ -17,11 +17,11 @@ logger = logging.getLogger("xknx.tools")
 
 async def group_value_read(
     xknx: XKNX,
-    group_address: GroupAddressableType,
+    group_address: DeviceAddressableType,
 ) -> None:
     """Send a GroupValueRead telegram."""
     telegram = Telegram(
-        destination_address=GroupAddress(group_address),
+        destination_address=parse_device_group_address(group_address),
         payload=GroupValueRead(),
     )
 
@@ -31,14 +31,14 @@ async def group_value_read(
 
 async def group_value_response(
     xknx: XKNX,
-    group_address: GroupAddressableType,
+    group_address: DeviceAddressableType,
     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),
+        destination_address=parse_device_group_address(group_address),
         payload=GroupValueResponse(payload),
     )
     logger.debug(
@@ -52,14 +52,14 @@ async def group_value_response(
 
 async def group_value_write(
     xknx: XKNX,
-    group_address: GroupAddressableType,
+    group_address: DeviceAddressableType,
     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),
+        destination_address=parse_device_group_address(group_address),
         payload=GroupValueWrite(payload),
     )
     logger.debug(
@@ -73,12 +73,12 @@ async def group_value_write(
 
 async def read_group_value(
     xknx: XKNX,
-    group_address: GroupAddressableType,
+    group_address: DeviceAddressableType,
     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))
+    value_reader = ValueReader(xknx, parse_device_group_address(group_address))
     response = await value_reader.read()
     if response is not None:
         assert isinstance(response.payload, (GroupValueWrite, GroupValueResponse))


Test patch:
diff --git a/test/tools_tests/tools_test.py b/test/tools_tests/tools_test.py
index 12681394..d092bdab 100644
--- a/test/tools_tests/tools_test.py
+++ b/test/tools_tests/tools_test.py
@@ -1,9 +1,10 @@
 """Test xknx tools package."""
-from unittest.mock import patch
+from unittest.mock import AsyncMock, patch
 
 import pytest
 
 from xknx import XKNX
+from xknx.devices import NumericValue
 from xknx.dpt import DPTArray, DPTBinary, DPTTemperature
 from xknx.exceptions import ConversionError
 from xknx.telegram import GroupAddress, Telegram, TelegramDirection, apci
@@ -111,3 +112,31 @@ async def test_read_group_value(value_reader_read_mock, value, value_type, expec
     # GroupValueRead telegram is not in queue because ValueReader.read is mocked.
     # This is tested in ValueReader tests.
     assert response_value == value
+
+
+async def test_tools_with_internal_addresses():
+    """Test tools using internal addresses."""
+    xknx = XKNX()
+    xknx.knxip_interface = AsyncMock()
+    await xknx.start()
+
+    internal_address = "i-test"
+    test_type = "1byte_unsigned"
+    number = NumericValue(
+        xknx,
+        "Test",
+        group_address=internal_address,
+        value_type=test_type,
+        respond_to_read=True,
+    )
+
+    assert number.resolve_state() is None
+    await group_value_write(xknx, internal_address, 1, value_type=test_type)
+    await xknx.telegrams.join()
+    assert number.resolve_state() == 1
+
+    response_value = await read_group_value(
+        xknx, internal_address, value_type=test_type
+    )
+    assert response_value == 1
+    await xknx.stop()