Repository: XKNX/xknx
Pull request #946 (2022-05-06 20:40:28)

Support different TPCI
<!--
  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 different TPCI in a CEMIFrame and Telegram

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/946?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#946](https://codecov.io/gh/XKNX/xknx/pull/946?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (4bf3d14) into [main](https://codecov.io/gh/XKNX/xknx/commit/b65d98e930fb6e2de3e2e74aaf16c797099427a1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (b65d98e) will **increase** coverage by `0.00%`.
> The diff coverage is `97.56%`.

> :exclamation: Current head 4bf3d14 differs from pull request most recent head e022c97. Consider uploading reports for the commit e022c97 to get more accurate results

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

```diff
@@           Coverage Diff            @@
##             main     #946    +/-   ##
========================================
  Coverage   95.96%   95.97%            
========================================
  Files         130      131     +1     
  Lines        7785     7893   +108     
========================================
+ Hits         7471     7575   +104     
- Misses        314      318     +4     
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/946?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/telegram/tpci.py](https://codecov.io/gh/XKNX/xknx/pull/946/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC90ZWxlZ3JhbS90cGNpLnB5) | `96.73% <96.73%> (ø)` | |
| [xknx/knxip/cemi\_frame.py](https://codecov.io/gh/XKNX/xknx/pull/946/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9rbnhpcC9jZW1pX2ZyYW1lLnB5) | `98.92% <97.29%> (-1.08%)` | :arrow_down: |
| [xknx/telegram/apci.py](https://codecov.io/gh/XKNX/xknx/pull/946/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC90ZWxlZ3JhbS9hcGNpLnB5) | `99.29% <100.00%> (-0.01%)` | :arrow_down: |


Patch:
diff --git a/changelog.md b/changelog.md
index 422f7a53..ee864abf 100644
--- a/changelog.md
+++ b/changelog.md
@@ -2,6 +2,11 @@
 
 ## Unreleased changes
 
+### Protocol
+
+- Parse and encode different TPCI in a CEMIFrame or Telegram
+- Set priority "System" flag for point-to-point CEMI frames initialized by a Telegram
+
 ## 0.21.2 IP Secure Bug fixes
 
 ### Bugfixes
diff --git a/xknx/knxip/cemi_frame.py b/xknx/knxip/cemi_frame.py
index 2cf91a41..49755785 100644
--- a/xknx/knxip/cemi_frame.py
+++ b/xknx/knxip/cemi_frame.py
@@ -16,6 +16,7 @@ from __future__ import annotations
 from xknx.exceptions import ConversionError, CouldNotParseKNXIP, UnsupportedCEMIMessage
 from xknx.telegram import GroupAddress, IndividualAddress, Telegram
 from xknx.telegram.apci import APCI
+from xknx.telegram.tpci import TPCI, TDataGroup
 
 from .knxip_enum import CEMIFlags, CEMIMessageCode
 
@@ -29,7 +30,7 @@ class CEMIFrame:
         flags: int = 0,
         src_addr: IndividualAddress = IndividualAddress(None),
         dst_addr: GroupAddress | IndividualAddress = GroupAddress(None),
-        mpdu_len: int = 0,
+        tpci: TPCI = TDataGroup(),
         payload: APCI | None = None,
     ):
         """Initialize CEMIFrame object."""
@@ -37,7 +38,7 @@ class CEMIFrame:
         self.flags = flags
         self.src_addr = src_addr
         self.dst_addr = dst_addr
-        self.mpdu_len = mpdu_len
+        self.tpci = tpci
         self.payload = payload
 
     @staticmethod
@@ -48,7 +49,7 @@ class CEMIFrame:
     ) -> CEMIFrame:
         """Return CEMIFrame from a Telegram."""
         cemi = CEMIFrame(code=code, src_addr=src_addr)
-        # dst_addr, payload and cmd are set by telegram.setter - mpdu_len not needed for outgoing telegram
+        # dst_addr, payload and cmd are set by telegram.setter
         cemi.telegram = telegram
         return cemi
 
@@ -60,6 +61,7 @@ class CEMIFrame:
             destination_address=self.dst_addr,
             payload=self.payload,
             source_address=self.src_addr,
+            tpci=self.tpci,
         )
 
     @telegram.setter
@@ -71,20 +73,26 @@ class CEMIFrame:
             CEMIFlags.FRAME_TYPE_STANDARD
             | CEMIFlags.DO_NOT_REPEAT
             | CEMIFlags.BROADCAST
-            | CEMIFlags.PRIORITY_LOW
-            | CEMIFlags.NO_ACK_REQUESTED
+            | (
+                CEMIFlags.ACK_REQUESTED
+                if telegram.tpci.ack_request
+                else CEMIFlags.NO_ACK_REQUESTED
+            )
             | CEMIFlags.CONFIRM_NO_ERROR
             | CEMIFlags.HOP_COUNT_1ST
         )
 
         if isinstance(telegram.destination_address, GroupAddress):
-            self.flags |= CEMIFlags.DESTINATION_GROUP_ADDRESS
+            self.flags |= CEMIFlags.DESTINATION_GROUP_ADDRESS | CEMIFlags.PRIORITY_LOW
         elif isinstance(telegram.destination_address, IndividualAddress):
-            self.flags |= CEMIFlags.DESTINATION_INDIVIDUAL_ADDRESS
+            self.flags |= (
+                CEMIFlags.DESTINATION_INDIVIDUAL_ADDRESS | CEMIFlags.PRIORITY_SYSTEM
+            )
         else:
             raise TypeError()
 
         self.dst_addr = telegram.destination_address
+        self.tpci = telegram.tpci
         self.payload = telegram.payload
 
     def set_hops(self, hops: int) -> None:
@@ -96,9 +104,11 @@ class CEMIFrame:
 
     def calculated_length(self) -> int:
         """Get length of KNX/IP body."""
-        if not isinstance(self.payload, APCI):
-            raise TypeError()
-        return 10 + self.payload.calculated_length()
+        if not self.tpci.control and self.payload is not None:
+            return 10 + self.payload.calculated_length()
+        if self.tpci.control and self.payload is None:
+            return 10
+        raise TypeError("Data TPDU must have a payload; control TPDU must not.")
 
     def from_knx(self, raw: bytes) -> int:
         """Parse/deserialize from KNX/IP raw data."""
@@ -122,9 +132,7 @@ class CEMIFrame:
 
     def from_knx_data_link_layer(self, cemi: bytes) -> int:
         """Parse L_DATA_IND, CEMIMessageCode.L_DATA_REQ, CEMIMessageCode.L_DATA_CON."""
-        if len(cemi) < 11:
-            # eg. ETS Line-Scan issues L_DATA_IND with length 10 or checking for individual addresses
-            # at start of ETS programming procedure
+        if len(cemi) < 10:
             raise UnsupportedCEMIMessage(
                 f"CEMI too small. Length: {len(cemi)}; CEMI: {cemi.hex()}"
             )
@@ -138,43 +146,70 @@ class CEMIFrame:
 
         self.src_addr = IndividualAddress((cemi[4 + addil], cemi[5 + addil]))
 
-        if self.flags & CEMIFlags.DESTINATION_GROUP_ADDRESS:
-            self.dst_addr = GroupAddress((cemi[6 + addil], cemi[7 + addil]))
-        else:
-            self.dst_addr = IndividualAddress((cemi[6 + addil], cemi[7 + addil]))
-
-        self.mpdu_len = cemi[8 + addil]
+        dst_is_group_address = bool(self.flags & CEMIFlags.DESTINATION_GROUP_ADDRESS)
+        dst_raw_address = (cemi[6 + addil], cemi[7 + addil])
+        self.dst_addr = (
+            GroupAddress(dst_raw_address)
+            if dst_is_group_address
+            else IndividualAddress(dst_raw_address)
+        )
 
-        # TPCI (transport layer control information)   -> First 14 bit
-        # APCI (application layer control information) -> Last  10 bit
+        npdu_len = cemi[8 + addil]
 
         apdu = cemi[9 + addil :]
-        if len(apdu) != (self.mpdu_len + 1):
+        if len(apdu) != (npdu_len + 1):  # TCPI octet not included in NPDU length
             raise CouldNotParseKNXIP(
-                f"APDU LEN should be {self.mpdu_len} but is {len(apdu)} in CEMI: {cemi.hex()}"
+                f"APDU LEN should be {npdu_len} but is {len(apdu) - 1} in CEMI: {cemi.hex()}"
             )
 
-        tpci_apci = (apdu[0] << 8) + apdu[1]
-
+        # TPCI (transport layer control information)
+        # - with control bit set -> 8 bit; no APDU
+        # - no control bit set (data) -> First 6 bit
+        # APCI (application layer control information) -> Last  10 bit of TPCI/APCI
         try:
-            self.payload = APCI.resolve_apci(tpci_apci & 0x03FF)
-        except ConversionError:
-            raise UnsupportedCEMIMessage(
-                f"APCI not supported: {(tpci_apci & 0x03FF):#012b}"
+            self.tpci = TPCI.resolve(
+                raw_tpci=cemi[9 + addil], dst_is_group_address=dst_is_group_address
             )
+        except ConversionError as err:
+            raise UnsupportedCEMIMessage(
+                f"TPCI not supported: {cemi[9 + addil]:#10b}"
+            ) from err
+
+        if self.tpci.control:
+            if npdu_len:
+                raise UnsupportedCEMIMessage(
+                    f"Invalid length for control TPDU {self.tpci}: {npdu_len}"
+                )
+            return 10 + addil
+
+        _apci = apdu[0] * 256 + apdu[1]
+        try:
+            self.payload = APCI.resolve_apci(_apci)
+        except ConversionError as err:
+            raise UnsupportedCEMIMessage(f"APCI not supported: {_apci:#012b}") from err
 
         self.payload.from_knx(apdu)
 
-        return 10 + addil + self.mpdu_len
+        return 10 + addil + npdu_len
 
     def to_knx(self) -> bytes:
         """Serialize to KNX/IP raw data."""
-        if not isinstance(self.payload, APCI):
-            raise TypeError()
-        if not isinstance(self.src_addr, (GroupAddress, IndividualAddress)):
-            raise ConversionError("src_addr not set")
+        if self.tpci.control:
+            tpdu = bytes([self.tpci.to_knx()])
+            npdu_len = 0
+        else:
+            if not isinstance(self.payload, APCI):
+                raise ConversionError(
+                    f"Invalid payload set for data TPDU: {self.payload.__class__}"
+                )
+            tpdu = self.payload.to_knx()
+            tpdu[0] |= self.tpci.to_knx()
+            npdu_len = self.payload.calculated_length()
+
+        if not isinstance(self.src_addr, IndividualAddress):
+            raise ConversionError("src_addr invalid")
         if not isinstance(self.dst_addr, (GroupAddress, IndividualAddress)):
-            raise ConversionError("dst_addr not set")
+            raise ConversionError("dst_addr invalid")
 
         return (
             bytes(
@@ -188,20 +223,21 @@ class CEMIFrame:
                 (
                     *self.src_addr.to_knx(),
                     *self.dst_addr.to_knx(),
-                    self.payload.calculated_length(),
+                    npdu_len,
                 )
             )
-            + self.payload.to_knx()
+            + tpdu
         )
 
     def __repr__(self) -> str:
         """Return object as readable string."""
         return (
             "<CEMIFrame "
-            f'SourceAddress="{self.src_addr.__repr__()}" '
-            f'DestinationAddress="{self.dst_addr.__repr__()}" '
-            f'Flags="{self.flags:16b}" '
             f'code="{self.code.name}" '
+            f'src_addr="{self.src_addr.__repr__()}" '
+            f'dst_addr="{self.dst_addr.__repr__()}" '
+            f'flags="{self.flags:16b}" '
+            f'tpci="{self.tpci}" '
             f'payload="{self.payload}" />'
         )
 
diff --git a/xknx/knxip/knxip_enum.py b/xknx/knxip/knxip_enum.py
index 75789cda..fa6aa39b 100644
--- a/xknx/knxip/knxip_enum.py
+++ b/xknx/knxip/knxip_enum.py
@@ -93,7 +93,7 @@ class CEMIFlags:
     BROADCAST = 0x1000
 
     # Bit 1/3+2
-    PRIORITY_SYSTE = 0x0000
+    PRIORITY_SYSTEM = 0x0000
     PRIORITY_NORMAL = 0x0400
     PRIORITY_URGENT = 0x0800
     PRIORITY_LOW = 0x0C00
diff --git a/xknx/telegram/apci.py b/xknx/telegram/apci.py
index fc79b639..0a945594 100644
--- a/xknx/telegram/apci.py
+++ b/xknx/telegram/apci.py
@@ -22,18 +22,16 @@ def encode_cmd_and_payload(
     cmd: APCIService | APCIUserService | APCIExtendedService,
     encoded_payload: int = 0,
     appended_payload: bytes | None = None,
-) -> bytes:
+) -> bytearray:
     """Encode cmd and payload."""
-    if appended_payload is None:
-        appended_payload = bytes()
-
     data = bytearray(
         [
-            (cmd.value >> 8) & 0xFF,
+            (cmd.value >> 8) & 0b11,
             (cmd.value & 0xFF) | (encoded_payload & DPTBinary.APCI_BITMASK),
         ]
     )
-    data.extend(appended_payload)
+    if appended_payload:
+        data.extend(appended_payload)
     return data
 
 
@@ -118,7 +116,7 @@ class APCI(ABC):
         """Parse/deserialize from KNX/IP raw data - to be implemented in derived class."""
 
     @abstractmethod
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data - to be implemented in derived class."""
 
     def __eq__(self, other: object) -> bool:
@@ -225,7 +223,7 @@ class GroupValueRead(APCI):
         # Nothing to parse, but must be implemented explicitly.
         return
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         return encode_cmd_and_payload(self.CODE)
 
@@ -262,7 +260,7 @@ class GroupValueWrite(APCI):
         else:
             self.value = DPTArray(raw[2:])
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         if isinstance(self.value, DPTBinary):
             return encode_cmd_and_payload(self.CODE, encoded_payload=self.value.value)
@@ -305,7 +303,7 @@ class GroupValueResponse(APCI):
         else:
             self.value = DPTArray(raw[2:])
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         if isinstance(self.value, DPTBinary):
             return encode_cmd_and_payload(self.CODE, encoded_payload=self.value.value)
@@ -349,7 +347,7 @@ class IndividualAddressWrite(APCI):
 
         self.address = IndividualAddress((address_high, address_low))
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         return encode_cmd_and_payload(
             self.CODE, appended_payload=bytes(self.address.to_knx())
@@ -375,7 +373,7 @@ class IndividualAddressRead(APCI):
         # Nothing to parse, but must be implemented explicitly.
         return
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         return encode_cmd_and_payload(self.CODE)
 
@@ -404,7 +402,7 @@ class IndividualAddressResponse(APCI):
         # Nothing to parse, but must be implemented explicitly.
         return
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         return encode_cmd_and_payload(self.CODE)
 
@@ -437,7 +435,7 @@ class ADCRead(APCI):
 
         self.channel = channel & DPTBinary.APCI_BITMASK
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         payload = struct.pack("!BB", self.channel, self.count)
 
@@ -475,7 +473,7 @@ class ADCResponse(APCI):
 
         self.channel = channel & DPTBinary.APCI_BITMASK
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         payload = struct.pack("!BBH", self.channel, self.count, self.value)
 
@@ -512,7 +510,7 @@ class MemoryRead(APCI):
 
         self.count = count & DPTBinary.APCI_BITMASK
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         if self.address < 0 or self.address >= 2**16:
             raise ConversionError("Address out of range.")
@@ -563,7 +561,7 @@ class MemoryWrite(APCI):
 
         self.count = count & DPTBinary.APCI_BITMASK
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         if self.address < 0 or self.address >= 2**16:
             raise ConversionError("Address out of range.")
@@ -615,7 +613,7 @@ class MemoryResponse(APCI):
 
         self.count = count & DPTBinary.APCI_BITMASK
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         if self.address < 0 or self.address >= 2**16:
             raise ConversionError("Address out of range.")
@@ -655,7 +653,7 @@ class DeviceDescriptorRead(APCI):
         """Parse/deserialize from KNX/IP raw data."""
         self.descriptor = raw[1] & 0x3F
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         if self.descriptor < 0 or self.descriptor >= 2**6:
             raise ConversionError("Descriptor out of range.")
@@ -691,7 +689,7 @@ class DeviceDescriptorResponse(APCI):
 
         self.descriptor = self.descriptor & 0x3F
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         if self.descriptor < 0 or self.descriptor >= 2**6:
             raise ConversionError("Descriptor out of range.")
@@ -726,7 +724,7 @@ class Restart(APCI):
         # Nothing to parse, but must be implemented explicitly.
         return
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         return encode_cmd_and_payload(self.CODE)
 
@@ -760,7 +758,7 @@ class UserMemoryRead(APCI):
         self.count = byte0 & 0x0F
         self.address = (((byte0 & 0xF0) >> 4) << 16) + address
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         if self.address < 0 or self.address >= 2**20:
             raise ConversionError("Address out of range.")
@@ -813,7 +811,7 @@ class UserMemoryWrite(APCI):
         self.count = byte0 & 0x0F
         self.address = (((byte0 & 0xF0) >> 4) << 16) + address
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         if self.address < 0 or self.address >= 2**20:
             raise ConversionError("Address out of range.")
@@ -867,7 +865,7 @@ class UserMemoryResponse(APCI):
         self.count = byte0 & 0x0F
         self.address = (((byte0 & 0xF0) >> 4) << 16) + address
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         if self.address < 0 or self.address >= 2**20:
             raise ConversionError("Address out of range.")
@@ -902,7 +900,7 @@ class UserManufacturerInfoRead(APCI):
         # Nothing to parse, but must be implemented explicitly.
         return
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
 
         return encode_cmd_and_payload(self.CODE)
@@ -933,7 +931,7 @@ class UserManufacturerInfoResponse(APCI):
         """Parse/deserialize from KNX/IP raw data."""
         self.manufacturer_id, self.data = struct.unpack("!B2s", raw[2:])
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         payload = struct.pack("!B2s", self.manufacturer_id, self.data)
 
@@ -972,7 +970,7 @@ class FunctionPropertyCommand(APCI):
             f"!BB{size}s", raw[2:]
         )
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         size = len(self.data)
         payload = struct.pack(
@@ -1014,7 +1012,7 @@ class FunctionPropertyStateRead(APCI):
             f"!BB{size}s", raw[2:]
         )
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         size = len(self.data)
         payload = struct.pack(
@@ -1064,7 +1062,7 @@ class FunctionPropertyStateResponse(APCI):
             self.data,
         ) = struct.unpack(f"!BBB{size}s", raw[2:])
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         size = len(self.data)
         payload = struct.pack(
@@ -1099,7 +1097,7 @@ class AuthorizeRequest(APCI):
         """Parse/deserialize from KNX/IP raw data."""
         _, self.key = struct.unpack("!BI", raw[2:])
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         payload = struct.pack("!BI", 0, self.key)
 
@@ -1127,7 +1125,7 @@ class AuthorizeResponse(APCI):
         """Parse/deserialize from KNX/IP raw data."""
         (self.level,) = struct.unpack("!B", raw[2:])
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         payload = struct.pack("!B", self.level)
 
@@ -1175,7 +1173,7 @@ class PropertyValueRead(APCI):
 
         self.count = count >> 4
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         if self.count < 0 or self.count > 2**4:
             raise ConversionError("Count out of range.")
@@ -1234,7 +1232,7 @@ class PropertyValueWrite(APCI):
         """Get length of APCI payload."""
         return 5 + len(self.data)
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         if self.count < 0 or self.count > 2**4:
             raise ConversionError("Count out of range.")
@@ -1328,7 +1326,7 @@ class PropertyValueResponse(APCI):
 
         self.count = self.count >> 4
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         if self.count < 0 or self.count > 2**4:
             raise ConversionError("Count out of range.")
@@ -1382,7 +1380,7 @@ class PropertyDescriptionRead(APCI):
             "!BBB", raw[2:]
         )
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         payload = struct.pack(
             "!BBB", self.object_index, self.property_id, self.property_index
@@ -1434,7 +1432,7 @@ class PropertyDescriptionResponse(APCI):
 
         self.max_count = max_count & 0x0FFF
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         if self.max_count < 0 or self.max_count >= 2**12:
             raise ConversionError("Max count out of range.")
@@ -1476,7 +1474,7 @@ class IndividualAddressSerialRead(APCI):
         """Parse/deserialize from KNX/IP raw data."""
         (self.serial,) = struct.unpack("!6s", raw[2:])
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         if len(self.serial) != 6:
             raise ConversionError("Serial must be 6 bytes.")
@@ -1519,7 +1517,7 @@ class IndividualAddressSerialResponse(APCI):
 
         self.address = IndividualAddress((address_high, address_low))
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         if len(self.serial) != 6:
             raise ConversionError("Serial must be 6 bytes.")
@@ -1563,7 +1561,7 @@ class IndividualAddressSerialWrite(APCI):
 
         self.address = IndividualAddress((address_high, address_low))
 
-    def to_knx(self) -> bytes:
+    def to_knx(self) -> bytearray:
         """Serialize to KNX/IP raw data."""
         if len(self.serial) != 6:
             raise ConversionError("Serial must be 6 bytes.")
diff --git a/xknx/telegram/telegram.py b/xknx/telegram/telegram.py
index a65a9b16..5ee4ab4a 100644
--- a/xknx/telegram/telegram.py
+++ b/xknx/telegram/telegram.py
@@ -20,6 +20,7 @@ from enum import Enum
 
 from .address import GroupAddress, IndividualAddress, InternalGroupAddress
 from .apci import APCI
+from .tpci import TPCI, TDataGroup
 
 
 class TelegramDirection(Enum):
@@ -40,6 +41,7 @@ class Telegram:
         direction: TelegramDirection = TelegramDirection.OUTGOING,
         payload: APCI | None = None,
         source_address: IndividualAddress = IndividualAddress(0),
+        tpci: TPCI = TDataGroup(),
     ) -> None:
         """Initialize Telegram class."""
         self.destination_address = destination_address
@@ -47,6 +49,7 @@ class Telegram:
         self.payload = payload
         self.source_address = source_address
         self.timestamp = datetime.now()
+        self.tpci = tpci
 
     def __str__(self) -> str:
         """Return object as readable string."""
@@ -59,6 +62,18 @@ class Telegram:
             "/>"
         )
 
+    def __repr__(self) -> str:
+        """Return object as string representation."""
+        return (
+            "Telegram("
+            f"destination_address={self.destination_address}, "
+            f"direction={self.direction}, "
+            f"payload={self.payload}, "
+            f"source_address={self.source_address}, "
+            f"tpci={self.tpci}"
+            ")"
+        )
+
     def __eq__(self, other: object) -> bool:
         """Equal operator."""
         for key, value in self.__dict__.items():
@@ -76,4 +91,4 @@ class Telegram:
     def __hash__(self) -> int:
         """Hash function."""
         # used to turn lists of Telegram into sets in unittests
-        return hash(str(self))
+        return hash(repr(self))
diff --git a/xknx/telegram/tpci.py b/xknx/telegram/tpci.py
new file mode 100644
index 00000000..1b22b8e1
--- /dev/null
+++ b/xknx/telegram/tpci.py
@@ -0,0 +1,195 @@
+"""
+Module for serialization and deserialization of TPCI payloads.
+
+TPCI stands for Transport Layer Protocol Control Information.
+
+An APCI payload contains a service and payload. For example, a GroupValueWrite
+is a service that takes a DPT as a value.
+"""
+from __future__ import annotations
+
+from abc import ABC
+from typing import ClassVar
+
+from xknx.exceptions import ConversionError
+
+CONTROL_BIT_MASK = 0x80
+NUMBERED_BIT_MASK = 0x40
+
+
+class TPCI(ABC):
+    """
+    Base class for TCPI services.
+
+    This base class is only the interface for the derived classes.
+    """
+
+    control: ClassVar[bool]
+    numbered: ClassVar[bool]
+    sequence_number: int
+    control_flags: ClassVar[int | None] = None
+
+    ack_request: ClassVar[bool] = False
+
+    def to_knx(self) -> int:
+        """Serialize to KNX/IP raw data."""
+        return (
+            self.control << 7
+            | self.numbered << 6
+            | (self.sequence_number & 0xF) << 2
+            | (self.control_flags or 0)
+        )
+
+    def __eq__(self, other: object) -> bool:
+        """Equal operator."""
+        if self.__class__ != other.__class__:
+            return False
+        return self.__dict__ == other.__dict__
+
+    def __repr__(self) -> str:
+        """Return object as readable string."""
+        _sequence_number = (
+            f"sequence_number={self.sequence_number}" if self.numbered else ""
+        )
+        return f"{self.__class__.__name__}({_sequence_number})"
+
+    @staticmethod
+    def resolve(raw_tpci: int, dst_is_group_address: bool) -> TPCI:
+        """
+        Return TPCI instance from TPCI command.
+
+        See KNX Specifications 03_03_04 Transport Layer v01.02.02 AS §2 TPDU
+        """
+        control = raw_tpci & CONTROL_BIT_MASK
+        numbered = raw_tpci & NUMBERED_BIT_MASK
+        sequence_number = (raw_tpci >> 2) & 0xF
+
+        if dst_is_group_address:
+            if control or numbered:
+                raise ConversionError("Invalid TPCI flags in group addressed frame.")
+            if not sequence_number:
+                return TDataGroup()
+            if sequence_number == 1:
+                # TDataTagGroup uses sequence number field as flag
+                return TDataTagGroup()
+
+        if not numbered and sequence_number:
+            raise ConversionError("Sequence number not allowed for unnumbered TPCI")
+
+        if not control:
+            # data - last 2 bits are part of APCI
+            if numbered:
+                return TDataConnected(sequence_number=sequence_number)
+            return TDataIndividual()
+        # unnumbered control
+        control_flags = raw_tpci & 0b11
+        if not numbered:
+            if control_flags == 0:
+                return TConnect()
+            if control_flags == 1:
+                return TDisconnect()
+        # numbered control
+        if control_flags == 0b10:
+            return TAck(sequence_number=sequence_number)
+        if control_flags == 0b11:
+            return TNak(sequence_number=sequence_number)
+
+        raise ConversionError(f"Unknown TPCI {raw_tpci:#10b}.")
+
+
+class TDataGroup(TPCI):
+    """
+    T_Data_Group class.
+
+    Includes T_Data_Broadcast and T_Data_Group depending on destination address.
+    """
+
+    control = False
+    numbered = False
+    sequence_number = 0
+
+    def to_knx(self) -> int:
+        """Serialize to KNX/IP raw data."""
+        return 0
+
+
+class TDataTagGroup(TPCI):
+    """T_Data_Tag_Group class."""
+
+    control = False
+    numbered = False
+    sequence_number = 0b0001
+
+
+class TDataConnected(TPCI):
+    """T_Data_Connected class."""
+
+    control = False
+    numbered = False
+    sequence_number = 0
+
+    ack_request = True
+
+    def __init__(self, sequence_number: int):
+        """Initialize TDataConnected."""
+        self.sequence_number = sequence_number
+
+
+class TDataIndividual(TPCI):
+    """T_Data_Individual class."""
+
+    control = False
+    numbered = False
+    sequence_number = 0
+
+    def to_knx(self) -> int:
+        """Serialize to KNX/IP raw data."""
+        return 0
+
+
+class TConnect(TPCI):
+    """T_Connect class."""
+
+    control = True
+    numbered = False
+    sequence_number = 0
+    control_flags = 0b00
+
+    ack_request = True
+
+
+class TDisconnect(TPCI):
+    """T_Disconnect class."""
+
+    control = True
+    numbered = False
+    sequence_number = 0
+    control_flags = 0b01
+
+    ack_request = True
+
+
+class TAck(TPCI):
+    """T_Ack class."""
+
+    control = True
+    numbered = True
+    sequence_number = 0
+    control_flags = 0b10
+
+    def __init__(self, sequence_number: int):
+        """Initialize TAck."""
+        self.sequence_number = sequence_number
+
+
+class TNak(TPCI):
+    """T_Nak class."""
+
+    control = True
+    numbered = True
+    sequence_number = 0
+    control_flags = 0b11
+
+    def __init__(self, sequence_number: int):
+        """Initialize TNak."""
+        self.sequence_number = sequence_number


Test patch:
diff --git a/test/io_tests/tunnel_test.py b/test/io_tests/tunnel_test.py
index e6a085b7..9a354e5d 100644
--- a/test/io_tests/tunnel_test.py
+++ b/test/io_tests/tunnel_test.py
@@ -46,12 +46,21 @@ class TestUDPTunnel:
             route_back=False,
         )
 
+    @pytest.mark.parametrize(
+        "raw",
+        [
+            # LDataInd GroupValueWrite from 1.1.22 to to 5/1/22 with DPT9 payload 0C 3F
+            # communication_channel_id: 0x02   sequence_counter: 0x21
+            bytes.fromhex("0610 0420 0017 04 02 21 00 2900bcd011162916030080 0c 3f"),
+            # LDataInd T_Connect from 1.0.250 to 1.0.255 (xknx tunnel endpoint) - ETS Line-Scan
+            # <UnsupportedCEMIMessage description="CEMI too small. Length: 10; CEMI: 2900b06010fa10ff0080" />
+            # communication_channel_id: 0x02   sequence_counter: 0x81
+            bytes.fromhex("0610 0420 0014 04 02 81 00 2900b06010fa10ff0080"),
+        ],
+    )
     @patch("xknx.io.UDPTunnel._send_tunnelling_ack")
-    def test_tunnel_request_received(self, send_ack_mock):
+    def test_tunnel_request_received(self, send_ack_mock, raw):
         """Test Tunnel for calling send_ack on normal frames."""
-        # LDataInd GroupValueWrite from 1.1.22 to to 5/1/22 with DPT9 payload 0C 3F
-        # communication_channel_id: 0x02   sequence_counter: 0x21
-        raw = bytes.fromhex("0610 0420 0017 04 02 21 00 2900bcd011162916030080 0c 3f")
         _cemi = CEMIFrame()
         _cemi.from_knx(raw[10:])
         telegram = _cemi.telegram
@@ -59,15 +68,14 @@ class TestUDPTunnel:
 
         self.tunnel.transport.data_received_callback(raw, ("192.168.1.2", 3671))
         self.tg_received_mock.assert_called_once_with(telegram)
-        send_ack_mock.assert_called_once_with(0x02, 0x21)
+        send_ack_mock.assert_called_once_with(raw[7], raw[8])
 
     @patch("xknx.io.UDPTunnel._send_tunnelling_ack")
     def test_tunnel_request_received_cemi_too_small(self, send_ack_mock):
         """Test Tunnel sending ACK for unsupported frames."""
-        # LDataInd T_Connect from 1.0.250 to 1.0.255 (xknx tunnel endpoint) - ETS Line-Scan
-        # <UnsupportedCEMIMessage description="CEMI too small. Length: 10; CEMI: 2900b06010fa10ff0080" />
+        # <UnsupportedCEMIMessage description="CEMI too small. Length: 9; CEMI: 2900b06010fa10ff00" />
         # communication_channel_id: 0x02   sequence_counter: 0x81
-        raw = bytes.fromhex("0610 0420 0014 04 02 81 00 2900b06010fa10ff0080")
+        raw = bytes.fromhex("0610 0420 0013 04 02 81 00 2900b06010fa10ff00")
 
         self.tunnel.transport.data_received_callback(raw, ("192.168.1.2", 3671))
         self.tg_received_mock.assert_not_called()
diff --git a/test/knxip_tests/cemi_frame_test.py b/test/knxip_tests/cemi_frame_test.py
index 2408353b..46a7073f 100644
--- a/test/knxip_tests/cemi_frame_test.py
+++ b/test/knxip_tests/cemi_frame_test.py
@@ -1,15 +1,15 @@
 """Tests for the CEMIFrame object."""
-from pytest import raises
+import pytest
 
-from xknx.dpt import DPTBinary
 from xknx.exceptions import ConversionError, CouldNotParseKNXIP, UnsupportedCEMIMessage
 from xknx.knxip.cemi_frame import CEMIFrame
 from xknx.knxip.knxip_enum import CEMIFlags, CEMIMessageCode
 from xknx.telegram import GroupAddress, IndividualAddress, Telegram
 from xknx.telegram.apci import GroupValueRead
+from xknx.telegram.tpci import TAck, TConnect, TDataGroup
 
 
-def get_data(code, adil, flags, src, dst, mpdu_len, tpci_apci, payload):
+def get_data(code, adil, flags, src, dst, npdu_len, tpci_apci, payload):
     """Encode to cemi data raw bytes."""
     return bytes(
         [
@@ -21,7 +21,7 @@ def get_data(code, adil, flags, src, dst, mpdu_len, tpci_apci, payload):
             src & 255,  # src
             (dst >> 8) & 255,  # dst
             dst & 255,  # dst
-            mpdu_len,  # mpdu_len
+            npdu_len,  # npdu_len
             (tpci_apci >> 8) & 255,  # tpci_apci
             tpci_apci & 255,  # tpci_apci
             *payload,  # payload
@@ -31,28 +31,64 @@ def get_data(code, adil, flags, src, dst, mpdu_len, tpci_apci, payload):
 
 def test_valid_command():
     """Test for valid frame parsing."""
+    raw = get_data(0x29, 0, 0x0080, 0, 0, 1, 0, [])
     frame = CEMIFrame()
-    packet_len = frame.from_knx(get_data(0x29, 0, 0, 0, 0, 1, 0, []))
+    packet_len = frame.from_knx(raw)
     assert frame.code == CEMIMessageCode.L_DATA_IND
-    assert frame.flags == 0
-    assert frame.mpdu_len == 1
-    assert frame.payload == GroupValueRead()
+    assert frame.flags == 0x0080
     assert frame.src_addr == IndividualAddress(0)
-    assert frame.dst_addr == IndividualAddress(0)
+    assert frame.dst_addr == GroupAddress(0)
+    assert frame.payload == GroupValueRead()
+    assert frame.tpci == TDataGroup()
     assert packet_len == 11
+    assert frame.calculated_length() == 11
+    assert frame.to_knx() == raw
 
 
-def test_invalid_tpci_apci():
+def test_valid_tpci_control():
+    """Test for valid tpci control."""
+    raw = bytes((0x29, 0, 0, 0, 0, 0, 0, 0, 0, 0x80))
+    frame = CEMIFrame()
+    packet_len = frame.from_knx(raw)
+    assert frame.code == CEMIMessageCode.L_DATA_IND
+    assert frame.flags == 0
+    assert frame.payload is None
+    assert frame.src_addr == IndividualAddress(0)
+    assert frame.dst_addr == IndividualAddress(0)
+    assert frame.tpci == TConnect()
+    assert packet_len == 10
+    assert frame.calculated_length() == 10
+    assert frame.to_knx() == raw
+
+
+@pytest.mark.parametrize(
+    "raw,err_msg",
+    [
+        (
+            get_data(0x29, 0, 0, 0, 0, 1, 0xFFC0, []),
+            r".*Invalid length for control TPDU.*",
+        ),
+        (
+            get_data(0x29, 0, 0, 0, 0, 1, 0x08C0, []),
+            r".*TPCI not supported.*",
+        ),
+        (
+            get_data(0x29, 0, 0, 0, 0, 1, 0x03C0, []),
+            r".*APCI not supported*",
+        ),
+    ],
+)
+def test_invalid_tpci_apci(raw, err_msg):
     """Test for invalid APCIService."""
     frame = CEMIFrame()
-    with raises(UnsupportedCEMIMessage, match=r".*APCI not supported: .*"):
-        frame.from_knx_data_link_layer(get_data(0x29, 0, 0, 0, 0, 1, 0xFFC0, []))
+    with pytest.raises(UnsupportedCEMIMessage, match=err_msg):
+        frame.from_knx_data_link_layer(raw)
 
 
 def test_invalid_apdu_len():
     """Test for invalid apdu len."""
     frame = CEMIFrame()
-    with raises(CouldNotParseKNXIP, match=r".*APDU LEN should be .*"):
+    with pytest.raises(CouldNotParseKNXIP, match=r".*APDU LEN should be .*"):
         frame.from_knx(get_data(0x29, 0, 0, 0, 0, 2, 0, []))
 
 
@@ -61,12 +97,11 @@ def test_invalid_src_addr():
     frame = CEMIFrame()
     frame.code = CEMIMessageCode.L_DATA_IND
     frame.flags = 0
-    frame.mpdu_len = 1
     frame.payload = GroupValueRead()
-    frame.src_addr = None
+    frame.src_addr = GroupAddress(0)
     frame.dst_addr = IndividualAddress(0)
 
-    with raises(ConversionError, match=r"src_addr not set"):
+    with pytest.raises(ConversionError, match=r"src_addr invalid"):
         frame.to_knx()
 
 
@@ -75,12 +110,11 @@ def test_invalid_dst_addr():
     frame = CEMIFrame()
     frame.code = CEMIMessageCode.L_DATA_IND
     frame.flags = 0
-    frame.mpdu_len = 1
     frame.payload = GroupValueRead()
     frame.src_addr = IndividualAddress(0)
     frame.dst_addr = None
 
-    with raises(ConversionError, match=r"dst_addr not set"):
+    with pytest.raises(ConversionError, match=r"dst_addr invalid"):
         frame.to_knx()
 
 
@@ -89,29 +123,32 @@ def test_invalid_payload():
     frame = CEMIFrame()
     frame.code = CEMIMessageCode.L_DATA_IND
     frame.flags = 0
-    frame.mpdu_len = 1
-    frame.payload = DPTBinary(1)
+    frame.payload = None
     frame.src_addr = IndividualAddress(0)
     frame.dst_addr = IndividualAddress(0)
 
-    with raises(TypeError):
+    with pytest.raises(TypeError):
         frame.calculated_length()
 
-    with raises(TypeError):
+    with pytest.raises(ConversionError):
         frame.to_knx()
 
 
 def test_from_knx_with_not_handleable_cemi():
     """Test for having unhandlebale cemi set."""
     frame = CEMIFrame()
-    with raises(UnsupportedCEMIMessage, match=r".*CEMIMessageCode not implemented:.*"):
+    with pytest.raises(
+        UnsupportedCEMIMessage, match=r".*CEMIMessageCode not implemented:.*"
+    ):
         frame.from_knx(get_data(0x30, 0, 0, 0, 0, 2, 0, []))
 
 
 def test_from_knx_with_not_implemented_cemi():
     """Test for having not implemented CEMI set."""
     frame = CEMIFrame()
-    with raises(UnsupportedCEMIMessage, match=r".*Could not handle CEMIMessageCode:.*"):
+    with pytest.raises(
+        UnsupportedCEMIMessage, match=r".*Could not handle CEMIMessageCode:.*"
+    ):
         frame.from_knx(
             get_data(CEMIMessageCode.L_BUSMON_IND.value, 0, 0, 0, 0, 2, 0, [])
         )
@@ -120,7 +157,7 @@ def test_from_knx_with_not_implemented_cemi():
 def test_invalid_invalid_len():
     """Test for invalid cemi len."""
     frame = CEMIFrame()
-    with raises(UnsupportedCEMIMessage, match=r".*CEMI too small.*"):
+    with pytest.raises(UnsupportedCEMIMessage, match=r".*CEMI too small.*"):
         frame.from_knx_data_link_layer(get_data(0x29, 0, 0, 0, 0, 2, 0, [])[:5])
 
 
@@ -143,25 +180,45 @@ def test_from_knx_individual_address():
 def test_telegram_group_address():
     """Test telegram conversion flags with a group address."""
     frame = CEMIFrame()
-    frame.telegram = Telegram(destination_address=GroupAddress(0))
-
-    assert (
-        frame.flags & CEMIFlags.DESTINATION_GROUP_ADDRESS
-    ) == CEMIFlags.DESTINATION_GROUP_ADDRESS
+    _telegram = Telegram(destination_address=GroupAddress(0))
+    # test CEMIFrame.telegram setter
+    frame.telegram = _telegram
+    assert frame.flags & 0x0080 == CEMIFlags.DESTINATION_GROUP_ADDRESS
+    assert frame.flags & 0x0C00 == CEMIFlags.PRIORITY_LOW
+    # test CEMIFrame.telegram property
+    assert frame.telegram == _telegram
 
 
 def test_telegram_individual_address():
     """Test telegram conversion flags with a individual address."""
     frame = CEMIFrame()
-    frame.telegram = Telegram(destination_address=IndividualAddress(0))
+    _telegram = Telegram(destination_address=IndividualAddress(0), tpci=TConnect())
+    # test CEMIFrame.telegram setter
+    frame.telegram = _telegram
+    assert frame.flags & 0x0080 == CEMIFlags.DESTINATION_INDIVIDUAL_ADDRESS
+    assert frame.flags & 0x0C00 == CEMIFlags.PRIORITY_SYSTEM
+    assert frame.flags & 0x0200 == CEMIFlags.ACK_REQUESTED
+    # test CEMIFrame.telegram property
+    assert frame.telegram == _telegram
+
 
-    assert (
-        frame.flags & CEMIFlags.DESTINATION_INDIVIDUAL_ADDRESS
-    ) == CEMIFlags.DESTINATION_INDIVIDUAL_ADDRESS
+def test_telegram_individual_address_ack():
+    """Test telegram conversion flags with a individual address."""
+    frame = CEMIFrame()
+    _telegram = Telegram(
+        destination_address=IndividualAddress(0), tpci=TAck(sequence_number=2)
+    )
+    # test CEMIFrame.telegram setter
+    frame.telegram = _telegram
+    assert frame.flags & 0x0080 == CEMIFlags.DESTINATION_INDIVIDUAL_ADDRESS
+    assert frame.flags & 0x0C00 == CEMIFlags.PRIORITY_SYSTEM
+    assert frame.flags & 0x0200 == CEMIFlags.NO_ACK_REQUESTED
+    # test CEMIFrame.telegram property
+    assert frame.telegram == _telegram
 
 
 def test_telegram_unsupported_address():
     """Test telegram conversion flags with an unsupported address."""
     frame = CEMIFrame()
-    with raises(TypeError):
+    with pytest.raises(TypeError):
         frame.telegram = Telegram(destination_address=object())
diff --git a/test/str_test.py b/test/str_test.py
index 8bca0a84..7f63a7d7 100644
--- a/test/str_test.py
+++ b/test/str_test.py
@@ -627,8 +627,8 @@ class TestStringRepresentations:
         assert (
             str(tunnelling_request)
             == '<TunnellingRequest communication_channel_id="23" sequence_counter="42" '
-            'cemi="<CEMIFrame SourceAddress="IndividualAddress("0.0.0")" DestinationAddress="GroupAddress("0/0/0")" '
-            'Flags="               0" code="L_DATA_REQ" payload="None" />" />'
+            'cemi="<CEMIFrame code="L_DATA_REQ" src_addr="IndividualAddress("0.0.0")" dst_addr="GroupAddress("0/0/0")" '
+            'flags="               0" tpci="TDataGroup()" payload="None" />" />'
         )
 
     def test_tunnelling_ack(self):
@@ -651,8 +651,8 @@ class TestStringRepresentations:
         )
         assert (
             str(cemi_frame)
-            == '<CEMIFrame SourceAddress="GroupAddress("1/2/3")" DestinationAddress="GroupAddress("1/2/5")" '
-            'Flags="1011110011100000" code="L_DATA_IND" payload="<GroupValueWrite value="<DPTBinary value="7" />" />" />'
+            == '<CEMIFrame code="L_DATA_IND" src_addr="GroupAddress("1/2/3")" dst_addr="GroupAddress("1/2/5")" '
+            'flags="1011110011100000" tpci="TDataGroup()" payload="<GroupValueWrite value="<DPTBinary value="7" />" />" />'
         )
 
     def test_knxip_frame(self):
@@ -691,6 +691,6 @@ class TestStringRepresentations:
         routing_indication = RoutingIndication()
         assert (
             str(routing_indication)
-            == '<RoutingIndication cemi="<CEMIFrame SourceAddress="IndividualAddress("0.0.0")" '
-            'DestinationAddress="GroupAddress("0/0/0")" Flags="               0" code="L_DATA_IND" payload="None" />" />'
+            == '<RoutingIndication cemi="<CEMIFrame code="L_DATA_IND" src_addr="IndividualAddress("0.0.0")" '
+            'dst_addr="GroupAddress("0/0/0")" flags="               0" tpci="TDataGroup()" payload="None" />" />'
         )
diff --git a/test/telegram_tests/tpci_test.py b/test/telegram_tests/tpci_test.py
new file mode 100644
index 00000000..a9111151
--- /dev/null
+++ b/test/telegram_tests/tpci_test.py
@@ -0,0 +1,60 @@
+"""Unit test for TPCI objects."""
+import pytest
+
+from xknx.exceptions import ConversionError
+from xknx.telegram.tpci import (
+    TPCI,
+    TAck,
+    TConnect,
+    TDataConnected,
+    TDataGroup,
+    TDataIndividual,
+    TDataTagGroup,
+    TDisconnect,
+    TNak,
+)
+
+
+@pytest.mark.parametrize(
+    "tpci_int,dst_is_group_address,expected",
+    [
+        (0b00000000, True, TDataGroup()),
+        (0b00000100, True, TDataTagGroup()),
+        (0b00000000, False, TDataIndividual()),
+        (0b01011100, False, TDataConnected(sequence_number=0b0111)),
+        (0b10000000, False, TConnect()),
+        (0b10000001, False, TDisconnect()),
+        (0b11101010, False, TAck(sequence_number=10)),
+        (0b11010011, False, TNak(sequence_number=4)),
+    ],
+)
+def test_resolve_tpci(tpci_int, dst_is_group_address, expected):
+    """Test resolving TPCI classes."""
+    assert (
+        TPCI.resolve(raw_tpci=tpci_int, dst_is_group_address=dst_is_group_address)
+        == expected
+    )
+
+
+@pytest.mark.parametrize(
+    "tpci_int,dst_is_group_address",
+    [
+        # sequence_number for non-numbered
+        (0b00100000, True),
+        (0b00001000, False),
+        (0b10000100, True),
+        (0b10000100, False),
+        # numbered for group addressed
+        (0b01000000, True),
+    ],
+)
+def test_invalid_tpci(tpci_int, dst_is_group_address):
+    """Test resolving TPCI classes."""
+    with pytest.raises(ConversionError):
+        TPCI.resolve(raw_tpci=tpci_int, dst_is_group_address=dst_is_group_address)
+
+
+def test_equality():
+    """Test equality."""
+    assert TDataGroup() == TDataGroup()
+    assert TDataGroup() != TDataIndividual()