Repository: XKNX/xknx
Pull request #853 (2022-01-23 08:03:03)

add DescriptionRequest and DescriptionResponse frame parser
<!--
  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.
-->
add DescriptionRequest and DescriptionResponse frame parser

These are basically unicast SearchRequest / SearchResponse. They can be used to query information about a specific KNX device - eg. for HA diagnosis.

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:

- [x] I have performed a self-review of my own code
- [ ] The documentation has been adjusted accordingly
- [x] The changes generate no new warnings
- [x] Tests have been added that prove the fix is effective or that the feature works
- [x] The changes are documented in the changelog
- [ ] The Homeassistant plugin has been adjusted in case of new config options


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

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

```diff
@@            Coverage Diff             @@
##             main     #853      +/-   ##
==========================================
- Coverage   95.68%   95.67%   -0.01%     
==========================================
  Files         112      114       +2     
  Lines        6904     6963      +59     
==========================================
+ Hits         6606     6662      +56     
- Misses        298      301       +3     
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/853?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/io/transport/ip\_transport.py](https://codecov.io/gh/XKNX/xknx/pull/853/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby90cmFuc3BvcnQvaXBfdHJhbnNwb3J0LnB5) | `89.36% <ø> (ø)` | |
| [xknx/knxip/search\_request.py](https://codecov.io/gh/XKNX/xknx/pull/853/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9rbnhpcC9zZWFyY2hfcmVxdWVzdC5weQ==) | `100.00% <ø> (ø)` | |
| [xknx/knxip/description\_response.py](https://codecov.io/gh/XKNX/xknx/pull/853/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9rbnhpcC9kZXNjcmlwdGlvbl9yZXNwb25zZS5weQ==) | `93.93% <93.93%> (ø)` | |
| [xknx/knxip/description\_request.py](https://codecov.io/gh/XKNX/xknx/pull/853/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9rbnhpcC9kZXNjcmlwdGlvbl9yZXF1ZXN0LnB5) | `94.44% <94.44%> (ø)` | |
| [xknx/knxip/\_\_init\_\_.py](https://codecov.io/gh/XKNX/xknx/pull/853/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9rbnhpcC9fX2luaXRfXy5weQ==) | `100.00% <100.00%> (ø)` | |
| [xknx/knxip/knxip.py](https://codecov.io/gh/XKNX/xknx/pull/853/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9rbnhpcC9rbnhpcC5weQ==) | `100.00% <100.00%> (ø)` | |


Patch:
diff --git a/changelog.md b/changelog.md
index fd3246d3..fbd9fc47 100644
--- a/changelog.md
+++ b/changelog.md
@@ -6,6 +6,10 @@
 
 - Tunneling: Fix route_back connections sending to invalid address
 
+### Protocol
+
+- add DescriptionRequest and DescriptionResponse Frame parser
+
 ## 0.19.0 Tunneling connection protocol 2022-01-18
 
 ### Devices
diff --git a/xknx/io/transport/ip_transport.py b/xknx/io/transport/ip_transport.py
index 1612a8d2..f4437162 100644
--- a/xknx/io/transport/ip_transport.py
+++ b/xknx/io/transport/ip_transport.py
@@ -68,9 +68,8 @@ class KNXIPTransport(ABC):
                 handled = True
         if not handled:
             knx_logger.debug(
-                "Unhandled %s: %s from: %s",
+                "Unhandled: %s from: %s",
                 knxipframe.header.service_type_ident,
-                knxipframe,
                 source,
             )
 
diff --git a/xknx/knxip/__init__.py b/xknx/knxip/__init__.py
index dc0e3ee9..5f3866d8 100644
--- a/xknx/knxip/__init__.py
+++ b/xknx/knxip/__init__.py
@@ -6,6 +6,8 @@ from .connect_request import ConnectRequest
 from .connect_response import ConnectResponse
 from .connectionstate_request import ConnectionStateRequest
 from .connectionstate_response import ConnectionStateResponse
+from .description_request import DescriptionRequest
+from .description_response import DescriptionResponse
 from .dib import DIB, DIBDeviceInformation, DIBGeneric, DIBSuppSVCFamilies
 from .disconnect_request import DisconnectRequest
 from .disconnect_response import DisconnectResponse
@@ -37,6 +39,8 @@ __all__ = [
     "ConnectResponse",
     "ConnectionStateRequest",
     "ConnectionStateResponse",
+    "DescriptionRequest",
+    "DescriptionResponse",
     "DIB",
     "DIBDeviceInformation",
     "DIBGeneric",
diff --git a/xknx/knxip/description_request.py b/xknx/knxip/description_request.py
new file mode 100644
index 00000000..b9fa9e9f
--- /dev/null
+++ b/xknx/knxip/description_request.py
@@ -0,0 +1,43 @@
+"""
+Module for Serialization and Deserialization of KNX Description Requests.
+
+The DESCRIPTION_REQUEST frame shall be sent by the KNXnet/IP Client to the control endpoint
+of the KNXnet/IP Server to obtain a self-description of the KNXnet/IP Server device.
+"""
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from .body import KNXIPBody
+from .hpai import HPAI
+from .knxip_enum import KNXIPServiceType
+
+if TYPE_CHECKING:
+    from xknx.xknx import XKNX
+
+
+class DescriptionRequest(KNXIPBody):
+    """Representation of a KNX Descriptoin Request."""
+
+    SERVICE_TYPE = KNXIPServiceType.DESCRIPTION_REQUEST
+
+    def __init__(self, xknx: XKNX, control_endpoint: HPAI = HPAI()):
+        """Initialize SearchRequest object."""
+        super().__init__(xknx)
+        self.control_endpoint = control_endpoint
+
+    def calculated_length(self) -> int:
+        """Get length of KNX/IP body."""
+        return HPAI.LENGTH
+
+    def from_knx(self, raw: bytes) -> int:
+        """Parse/deserialize from KNX/IP raw data."""
+        return self.control_endpoint.from_knx(raw)
+
+    def to_knx(self) -> list[int]:
+        """Serialize to KNX/IP raw data."""
+        return self.control_endpoint.to_knx()
+
+    def __str__(self) -> str:
+        """Return object as readable string."""
+        return f'<DescriptionRequest control_endpoint="{self.control_endpoint}" />'
diff --git a/xknx/knxip/description_response.py b/xknx/knxip/description_response.py
new file mode 100644
index 00000000..75763e5d
--- /dev/null
+++ b/xknx/knxip/description_response.py
@@ -0,0 +1,63 @@
+"""
+Module for Serialization and Deserialization of KNX Description Response.
+
+The DESCRIPTION_RESPONSE frame shall be sent by the KNXnet/IP Server as an answer to
+a received DESCRIPTION_REQUEST frame. It shall be addressed to the KNXnet/IP Clients
+control endpoint using the HPAI included in the received DESCRIPTION_REQUEST frame.
+The size of the KNXnet/IP body varies depending on the number of DIB structures sent
+by the KNXnet/IP Server in response to the KNXnet/IP Clients DESCRIPTION_REQUEST.
+"""
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from .body import KNXIPBody
+from .dib import DIB, DIBDeviceInformation
+from .knxip_enum import KNXIPServiceType
+
+if TYPE_CHECKING:
+    from xknx.xknx import XKNX
+
+
+class DescriptionResponse(KNXIPBody):
+    """Representation of a KNX Description Response."""
+
+    SERVICE_TYPE = KNXIPServiceType.DESCRIPTION_RESPONSE
+
+    def __init__(self, xknx: XKNX):
+        """Initialize SearchResponse object."""
+        super().__init__(xknx)
+        self.dibs: list[DIB] = []
+
+    def calculated_length(self) -> int:
+        """Get length of KNX/IP body."""
+        return sum([dib.calculated_length() for dib in self.dibs])
+
+    def from_knx(self, raw: bytes) -> int:
+        """Parse/deserialize from KNX/IP raw data."""
+        pos = 0
+        while raw[pos:]:
+            dib = DIB.determine_dib(raw[pos:])
+            pos += dib.from_knx(raw[pos:])
+            self.dibs.append(dib)
+        return pos
+
+    @property
+    def device_name(self) -> str:
+        """Return name of device."""
+        for dib in self.dibs:
+            if isinstance(dib, DIBDeviceInformation):
+                return dib.name
+        return "UNKNOWN"
+
+    def to_knx(self) -> list[int]:
+        """Serialize to KNX/IP raw data."""
+        data = []
+        for dib in self.dibs:
+            data.extend(dib.to_knx())
+        return data
+
+    def __str__(self) -> str:
+        """Return object as readable string."""
+        _dibs_str = ",\n".join(dib.__str__() for dib in self.dibs)
+        return "<DescriptionResponse " f'dibs="[\n{_dibs_str}\n]" />'
diff --git a/xknx/knxip/knxip.py b/xknx/knxip/knxip.py
index 023eda9c..a1715532 100644
--- a/xknx/knxip/knxip.py
+++ b/xknx/knxip/knxip.py
@@ -15,6 +15,8 @@ from .connect_request import ConnectRequest
 from .connect_response import ConnectResponse
 from .connectionstate_request import ConnectionStateRequest
 from .connectionstate_response import ConnectionStateResponse
+from .description_request import DescriptionRequest
+from .description_response import DescriptionResponse
 from .disconnect_request import DisconnectRequest
 from .disconnect_response import DisconnectResponse
 from .header import KNXIPHeader
@@ -53,6 +55,10 @@ class KNXIPFrame:
             body = TunnellingRequest(self.xknx)
         elif service_type_ident == KNXIPServiceType.TUNNELLING_ACK:
             body = TunnellingAck(self.xknx)
+        elif service_type_ident == KNXIPServiceType.DESCRIPTION_REQUEST:
+            body = DescriptionRequest(self.xknx)
+        elif service_type_ident == KNXIPServiceType.DESCRIPTION_RESPONSE:
+            body = DescriptionResponse(self.xknx)
         elif service_type_ident == KNXIPServiceType.SEARCH_REQUEST:
             body = SearchRequest(self.xknx)
         elif service_type_ident == KNXIPServiceType.SEARCH_RESPONSE:
diff --git a/xknx/knxip/search_request.py b/xknx/knxip/search_request.py
index 8c4d8fdf..87249349 100644
--- a/xknx/knxip/search_request.py
+++ b/xknx/knxip/search_request.py
@@ -16,7 +16,7 @@ if TYPE_CHECKING:
 
 
 class SearchRequest(KNXIPBody):
-    """Representation of a KNX Connect Request."""
+    """Representation of a KNX Search Request."""
 
     SERVICE_TYPE = KNXIPServiceType.SEARCH_REQUEST
 


Test patch:
diff --git a/test/knxip_tests/description_request_test.py b/test/knxip_tests/description_request_test.py
new file mode 100644
index 00000000..bfdd806f
--- /dev/null
+++ b/test/knxip_tests/description_request_test.py
@@ -0,0 +1,40 @@
+"""Unit test for KNX/IP DescriptionRequest objects."""
+from xknx import XKNX
+from xknx.knxip import HPAI, DescriptionRequest, KNXIPFrame
+
+
+class TestKNXIPDescriptionRequest:
+    """Test class for KNX/IP DescriptionRequest objects."""
+
+    def test_description_request(self):
+        """Test parsing and streaming DescriptionRequest KNX/IP packet."""
+        raw = (
+            0x06,
+            0x10,
+            0x02,
+            0x03,
+            0x00,
+            0x0E,
+            0x08,
+            0x01,
+            0x7F,
+            0x00,
+            0x00,
+            0x02,
+            0x0E,
+            0x57,
+        )
+        xknx = XKNX()
+        knxipframe = KNXIPFrame(xknx)
+        knxipframe.from_knx(raw)
+
+        assert isinstance(knxipframe.body, DescriptionRequest)
+        assert knxipframe.body.control_endpoint == HPAI(ip_addr="127.0.0.2", port=3671)
+
+        knxipframe2 = KNXIPFrame.init_from_body(
+            DescriptionRequest(
+                xknx, control_endpoint=HPAI(ip_addr="127.0.0.2", port=3671)
+            )
+        )
+
+        assert knxipframe2.to_knx() == list(raw)
diff --git a/test/knxip_tests/description_response_test.py b/test/knxip_tests/description_response_test.py
new file mode 100644
index 00000000..8ef1636b
--- /dev/null
+++ b/test/knxip_tests/description_response_test.py
@@ -0,0 +1,55 @@
+"""Unit test for KNX/IP DescriptionResponse objects."""
+from xknx import XKNX
+from xknx.knxip import (
+    DescriptionResponse,
+    DIBDeviceInformation,
+    DIBServiceFamily,
+    DIBSuppSVCFamilies,
+    KNXIPFrame,
+)
+
+
+class TestKNXIPDescriptionResponse:
+    """Test class for KNX/IP DescriptionResponse objects."""
+
+    def test_description_response(self):
+        """Test parsing and streaming DescriptionResponse KNX/IP packet."""
+        raw = bytes.fromhex(
+            "061002040048360102001000000000082d40834de000170c000ab3274a324b4e582f49502d526f7574657200000000000000000000000000000000000c0202020302040205020701",
+        )
+        xknx = XKNX()
+        knxipframe = KNXIPFrame(xknx)
+        assert knxipframe.from_knx(raw) == 72
+        assert knxipframe.to_knx() == list(raw)
+
+        assert isinstance(knxipframe.body, DescriptionResponse)
+        assert len(knxipframe.body.dibs) == 2
+        # Specific testing of parsing and serializing of
+        # DIBDeviceInformation and DIBSuppSVCFamilies is
+        # done within knxip_dib_test.py
+        assert isinstance(knxipframe.body.dibs[0], DIBDeviceInformation)
+        assert isinstance(knxipframe.body.dibs[1], DIBSuppSVCFamilies)
+        assert knxipframe.body.device_name == "KNX/IP-Router"
+        assert knxipframe.body.dibs[1].supports(DIBServiceFamily.CORE, version=2)
+        assert knxipframe.body.dibs[1].supports(
+            DIBServiceFamily.DEVICE_MANAGEMENT, version=2
+        )
+        assert knxipframe.body.dibs[1].supports(DIBServiceFamily.ROUTING, version=2)
+        assert knxipframe.body.dibs[1].supports(DIBServiceFamily.TUNNELING, version=2)
+        assert knxipframe.body.dibs[1].supports(
+            DIBServiceFamily.REMOTE_CONFIGURATION_DIAGNOSIS
+        )
+        assert not knxipframe.body.dibs[1].supports(DIBServiceFamily.OBJECT_SERVER)
+
+        description_response = DescriptionResponse(xknx)
+        description_response.dibs.append(knxipframe.body.dibs[0])
+        description_response.dibs.append(knxipframe.body.dibs[1])
+        knxipframe2 = KNXIPFrame.init_from_body(description_response)
+
+        assert knxipframe2.to_knx() == list(raw)
+
+    def test_unknown_device_name(self):
+        """Test device_name if no DIBDeviceInformation is present."""
+        xknx = XKNX()
+        description_response = DescriptionResponse(xknx)
+        assert description_response.device_name == "UNKNOWN"