Repository: XKNX/xknx
Pull request #949 (2022-05-13 06:04:24)

Fix TPCI definitions
<!--
  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.
-->
- Fix TPCI definitions
- set default TPCI for Telegram based on destination address (eg. for equaltiy checks)

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.
-->

- [x] Bug fix (non-breaking change which fixes an issue)
- [ ] 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
- [ ] Tests have been added that prove the fix is effective or that the feature works
- [ ] The changes are documented in the changelog
- [ ] The Homeassistant plugin has been adjusted in case of new config options



Patch:
diff --git a/xknx/telegram/telegram.py b/xknx/telegram/telegram.py
index 5ee4ab4a..15bc5d71 100644
--- a/xknx/telegram/telegram.py
+++ b/xknx/telegram/telegram.py
@@ -20,7 +20,7 @@ from enum import Enum
 
 from .address import GroupAddress, IndividualAddress, InternalGroupAddress
 from .apci import APCI
-from .tpci import TPCI, TDataGroup
+from .tpci import TPCI, TDataGroup, TDataIndividual
 
 
 class TelegramDirection(Enum):
@@ -41,7 +41,7 @@ class Telegram:
         direction: TelegramDirection = TelegramDirection.OUTGOING,
         payload: APCI | None = None,
         source_address: IndividualAddress = IndividualAddress(0),
-        tpci: TPCI = TDataGroup(),
+        tpci: TPCI | None = None,
     ) -> None:
         """Initialize Telegram class."""
         self.destination_address = destination_address
@@ -49,7 +49,11 @@ class Telegram:
         self.payload = payload
         self.source_address = source_address
         self.timestamp = datetime.now()
-        self.tpci = tpci
+        self.tpci = tpci or (
+            TDataIndividual()
+            if isinstance(destination_address, IndividualAddress)
+            else TDataGroup()
+        )
 
     def __str__(self) -> str:
         """Return object as readable string."""
diff --git a/xknx/telegram/tpci.py b/xknx/telegram/tpci.py
index 1b22b8e1..b526546e 100644
--- a/xknx/telegram/tpci.py
+++ b/xknx/telegram/tpci.py
@@ -26,7 +26,7 @@ class TPCI(ABC):
 
     control: ClassVar[bool]
     numbered: ClassVar[bool]
-    sequence_number: int
+    sequence_number: int = 0
     control_flags: ClassVar[int | None] = None
 
     ack_request: ClassVar[bool] = False
@@ -106,7 +106,6 @@ class TDataGroup(TPCI):
 
     control = False
     numbered = False
-    sequence_number = 0
 
     def to_knx(self) -> int:
         """Serialize to KNX/IP raw data."""
@@ -121,12 +120,22 @@ class TDataTagGroup(TPCI):
     sequence_number = 0b0001
 
 
+class TDataIndividual(TPCI):
+    """T_Data_Individual class."""
+
+    control = False
+    numbered = False
+
+    def to_knx(self) -> int:
+        """Serialize to KNX/IP raw data."""
+        return 0
+
+
 class TDataConnected(TPCI):
     """T_Data_Connected class."""
 
     control = False
-    numbered = False
-    sequence_number = 0
+    numbered = True
 
     ack_request = True
 
@@ -135,24 +144,11 @@ class TDataConnected(TPCI):
         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
@@ -163,7 +159,6 @@ class TDisconnect(TPCI):
 
     control = True
     numbered = False
-    sequence_number = 0
     control_flags = 0b01
 
     ack_request = True
@@ -174,7 +169,6 @@ class TAck(TPCI):
 
     control = True
     numbered = True
-    sequence_number = 0
     control_flags = 0b10
 
     def __init__(self, sequence_number: int):
@@ -187,7 +181,6 @@ class TNak(TPCI):
 
     control = True
     numbered = True
-    sequence_number = 0
     control_flags = 0b11
 
     def __init__(self, sequence_number: int):


Test patch:
diff --git a/requirements/testing.txt b/requirements/testing.txt
index 6e29b5fb..d71ff0fb 100644
--- a/requirements/testing.txt
+++ b/requirements/testing.txt
@@ -8,6 +8,6 @@ pylint==2.13.8
 pytest==7.1.2
 pytest-asyncio==0.18.3
 pytest-cov==3.0.0
-setuptools==62.2.0
+setuptools==62.1.0
 tox==3.25.0
 mypy==0.950
diff --git a/test/telegram_tests/telegram_test.py b/test/telegram_tests/telegram_test.py
index ef6b3276..b02eee54 100644
--- a/test/telegram_tests/telegram_test.py
+++ b/test/telegram_tests/telegram_test.py
@@ -1,7 +1,8 @@
 """Unit test for Telegram objects."""
 from xknx.dpt import DPTBinary
-from xknx.telegram import GroupAddress, Telegram, TelegramDirection
+from xknx.telegram import GroupAddress, IndividualAddress, Telegram, TelegramDirection
 from xknx.telegram.apci import GroupValueRead, GroupValueWrite
+from xknx.telegram.tpci import TConnect, TDisconnect
 
 
 class TestTelegram:
@@ -29,3 +30,6 @@ class TestTelegram:
             TelegramDirection.INCOMING,
             payload=GroupValueRead(),
         )
+        assert Telegram(IndividualAddress(1), tpci=TConnect()) != Telegram(
+            IndividualAddress(1), tpci=TDisconnect()
+        )
diff --git a/test/telegram_tests/tpci_test.py b/test/telegram_tests/tpci_test.py
index a9111151..725f344b 100644
--- a/test/telegram_tests/tpci_test.py
+++ b/test/telegram_tests/tpci_test.py
@@ -16,7 +16,7 @@ from xknx.telegram.tpci import (
 
 
 @pytest.mark.parametrize(
-    "tpci_int,dst_is_group_address,expected",
+    "tpci_int,dst_is_group_address,tpci_expected",
     [
         (0b00000000, True, TDataGroup()),
         (0b00000100, True, TDataTagGroup()),
@@ -28,12 +28,13 @@ from xknx.telegram.tpci import (
         (0b11010011, False, TNak(sequence_number=4)),
     ],
 )
-def test_resolve_tpci(tpci_int, dst_is_group_address, expected):
-    """Test resolving TPCI classes."""
+def test_tpci_resolve_encode(tpci_int, dst_is_group_address, tpci_expected):
+    """Test resolving and encoding TPCI classes."""
     assert (
         TPCI.resolve(raw_tpci=tpci_int, dst_is_group_address=dst_is_group_address)
-        == expected
+        == tpci_expected
     )
+    assert tpci_expected.to_knx() == tpci_int
 
 
 @pytest.mark.parametrize(