refactor: let each control field part serialize itself

`CEMIFlags.to_knx()` took a `frame_type` argument that shadowed its own
`frame_type` field, which reads as if the field were the default. Give
`CEMIFrameType` and the new `CEMIAddressType` their own `to_knx()` / `from_knx()`
placing their single bit, and let the frame compose:

    self.flags.to_knx() | frame_type.to_knx() | self.address_type.to_knx()

`CEMIFlags.to_knx()` now only sets the bits it owns and takes no arguments, so
nothing shadows anything. Control fields are handled as one 16 bit value
(Ctrl1 << 8 | Ctrl2) to make the composition a plain OR; `to_knx()` returns an
int and `from_knx()` takes one, as `TPCI.to_knx()` already does for a bit level
field. The Frame Type and Address Type bit masks are gone from the module level
constants - each enum places its own bit - which was the only place where the
8/16 bit mixing was actually confusing.

`CEMILData.dst_is_group_address` is replaced by `CEMILData.address_type`.
`block_0()` and the `SecureData` methods take `address_type: CEMIAddressType`
accordingly; the B0 flags octet is literally the Address Type bit plus the
Extended Frame Format, so `address_type.to_knx()` is that octet and the manual
constant is gone. Byte vectors are unchanged.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: farmio

docs/changelog.md

@@ -17,10 +17,10 @@ nav_order: 2
   - `confirm_error: bool` - only meaningful in an L_Data.con frame
   - `hop_count: int` - replaces the `CEMILData.hops` property, which was removed
   - `frame_type: CEMIFrameType` and `frame_format: CEMIFrameFormat` - informational; they hold what was received and are ignored when serializing
-  The Frame Type of an outgoing frame is derived from its NPDU length and the Address Type from the type of the destination address (also available as `CEMILData.dst_is_group_address`), so neither can disagree with the frame that is put on the wire. The raw bit masks moved to `xknx.cemi.flags` as module level constants.
+  The Frame Type of an outgoing frame is derived from its NPDU length and the Address Type from the type of the destination address (`CEMILData.address_type`), so neither can disagree with the frame that is put on the wire. `CEMIFrameType` and the new `CEMIAddressType` place their own bit via `to_knx()` / `from_knx()`; `CEMILData` composes the control field as `flags.to_knx() | frame_type.to_knx() | address_type.to_knx()`. Ctrl1 and Ctrl2 are handled as one 16 bit int, so `CEMIFlags.to_knx()` returns an `int` and `CEMIFlags.from_knx()` takes one. The remaining bit masks moved to `xknx.cemi.flags` as module level constants.
 - `CEMIFlags` has a compact `__str__` used in `CEMILData.__repr__` - eg. `LOW STANDARD hop_count=6`, listing only the boolean flags that are set. The full dataclass `__repr__` shows every field.
 - `CEMILData(flags=...)` is optional now and defaults to `CEMIFlags()` - low priority, hop count 6, no acknowledge request.
-- `xknx.secure.data_secure_asdu.block_0()`, `SecureData.init_from_plain_apdu()` and `SecureData.get_plain_apdu()` take `dst_is_group_address: bool` instead of `frame_flags: int`. Only the Address Type bit of Ctrl2 ever reached the CCM input; passing it explicitly removes the mask.
+- `xknx.secure.data_secure_asdu.block_0()`, `SecureData.init_from_plain_apdu()` and `SecureData.get_plain_apdu()` take `address_type: CEMIAddressType` instead of `frame_flags: int`. Only the Address Type bit of Ctrl2 ever reached the CCM input, so the B0 flags octet is `address_type.to_knx()` and `B0_AT_FIELD_FLAGS_MASK` is gone.
 
 ### Bugfixes
 

test/cemi_tests/cemi_frame_test.py

@@ -3,8 +3,10 @@
 import pytest
 
 from xknx.cemi import (
+    CEMIAddressType,
     CEMIFlags,
     CEMIFrame,
+    CEMIFrameType,
     CEMILData,
     CEMIMessageCode,
     CEMIMPropReadRequest,
@@ -14,7 +16,6 @@
     CEMIPriority,
 )
 from xknx.cemi.const import CEMIErrorCode
-from xknx.cemi.flags import DESTINATION_GROUP_ADDRESS, FRAME_TYPE_STANDARD
 from xknx.dpt import DPTArray
 from xknx.exceptions import ConversionError, CouldNotParseCEMI, UnsupportedCEMIMessage
 from xknx.profile.const import ResourceKNXNETIPPropertyId, ResourceObjectType
@@ -244,7 +245,7 @@ def test_telegram_group_address() -> None:
         data=CEMILData.init_from_telegram(_telegram),
     )
     assert isinstance(frame.data, CEMILData)
-    assert frame.data.dst_is_group_address
+    assert frame.data.address_type is CEMIAddressType.GROUP
     assert frame.data.flags == CEMIFlags(priority=CEMIPriority.LOW)
     # test CEMIFrame.telegram property
     assert frame.data.telegram() == _telegram
@@ -258,7 +259,7 @@ def test_telegram_broadcast() -> None:
         data=CEMILData.init_from_telegram(_telegram),
     )
     assert isinstance(frame.data, CEMILData)
-    assert frame.data.dst_is_group_address
+    assert frame.data.address_type is CEMIAddressType.GROUP
     assert frame.data.flags == CEMIFlags(priority=CEMIPriority.SYSTEM)
     assert frame.data.tpci == TDataBroadcast()
     # test CEMIFrame.telegram property
@@ -273,7 +274,7 @@ def test_telegram_individual_address() -> None:
         data=CEMILData.init_from_telegram(_telegram),
     )
     assert isinstance(frame.data, CEMILData)
-    assert not frame.data.dst_is_group_address
+    assert frame.data.address_type is CEMIAddressType.INDIVIDUAL
     assert frame.data.flags == CEMIFlags(priority=CEMIPriority.SYSTEM)
     assert not frame.data.flags.acknowledge_request
     # test CEMIFrame.telegram property
@@ -315,7 +316,9 @@ def test_frame_type_from_npdu_length(
         GroupValueWrite(DPTArray(bytes(apdu_payload_length)))
     )
     assert raw[6] == expected_npdu_len
-    assert bool(raw[0] & FRAME_TYPE_STANDARD) is expected_standard_frame
+    assert (
+        CEMIFrameType.from_knx(raw[0] << 8) is CEMIFrameType.STANDARD
+    ) is expected_standard_frame
 
 
 def test_frame_type_follows_payload() -> None:
@@ -326,11 +329,11 @@ def test_frame_type_follows_payload() -> None:
         tpci=TDataGroup(),
         payload=GroupValueWrite(DPTArray(bytes(15))),
     )
-    assert not cemi_data.to_knx()[0] & FRAME_TYPE_STANDARD
+    assert CEMIFrameType.from_knx(cemi_data.to_knx()[0] << 8) is CEMIFrameType.EXTENDED
 
     # replacing the payload - as Data Secure does - changes the Frame Type
     cemi_data.payload = GroupValueWrite(DPTArray(bytes(1)))
-    assert cemi_data.to_knx()[0] & FRAME_TYPE_STANDARD
+    assert CEMIFrameType.from_knx(cemi_data.to_knx()[0] << 8) is CEMIFrameType.STANDARD
 
 
 def test_npdu_length_exceeded() -> None:
@@ -355,7 +358,7 @@ def test_extended_frame_format_not_supported(eff: int, err_msg: str) -> None:
     raw = get_data(
         0x29,
         0,
-        (DESTINATION_GROUP_ADDRESS | eff),  # Ctrl2; Ctrl1 = extended frame
+        (CEMIAddressType.GROUP.to_knx() | eff),  # Ctrl2; Ctrl1 = extended frame
         1,
         1,
         1,

test/cemi_tests/flags_test.py

@@ -2,22 +2,35 @@
 
 import pytest
 
-from xknx.cemi import CEMIFlags, CEMIFrameFormat, CEMIFrameType, CEMIPriority
+from xknx.cemi import (
+    CEMIAddressType,
+    CEMIFlags,
+    CEMIFrameFormat,
+    CEMIFrameType,
+    CEMIPriority,
+)
 from xknx.exceptions import ConversionError
 
 
+def control_field(ctrl1: int, ctrl2: int) -> int:
+    """Return Ctrl1 and Ctrl2 as one 16 bit control field."""
+    return ctrl1 << 8 | ctrl2
+
+
 @pytest.mark.parametrize(
-    "raw,flags",
+    "ctrl1,ctrl2,flags",
     [
         (
             # Ctrl1 0xBC: standard frame, do not repeat, broadcast, low priority
             # Ctrl2 0xE0: group address, hop count 6, standard frame format
-            bytes((0xBC, 0xE0)),
+            0xBC,
+            0xE0,
             CEMIFlags(priority=CEMIPriority.LOW, hop_count=6),
         ),
         (
             # Ctrl1 0x3C: extended frame - kept, but not evaluated
-            bytes((0x3C, 0xE0)),
+            0x3C,
+            0xE0,
             CEMIFlags(
                 priority=CEMIPriority.LOW,
                 hop_count=6,
@@ -26,13 +39,15 @@
         ),
         (
             # Ctrl1 0xB0: system priority; Ctrl2 0x60: individual address
-            bytes((0xB0, 0x60)),
+            0xB0,
+            0x60,
             CEMIFlags(priority=CEMIPriority.SYSTEM, hop_count=6),
         ),
         (
             # every optional flag set: repeat on error, system broadcast,
             # acknowledge requested, confirm error, urgent priority, hop count 7
-            bytes((0b1000_1011, 0b0111_0000)),
+            0b1000_1011,
+            0b0111_0000,
             CEMIFlags(
                 priority=CEMIPriority.URGENT,
                 repeat_on_error=True,
@@ -44,9 +59,9 @@
         ),
     ],
 )
-def test_from_knx(raw: bytes, flags: CEMIFlags) -> None:
-    """Test parsing of Ctrl1 and Ctrl2."""
-    assert CEMIFlags.from_knx(raw) == flags
+def test_from_knx(ctrl1: int, ctrl2: int, flags: CEMIFlags) -> None:
+    """Test parsing of the control field."""
+    assert CEMIFlags.from_knx(control_field(ctrl1, ctrl2)) == flags
 
 
 @pytest.mark.parametrize(
@@ -62,84 +77,96 @@ def test_from_knx(raw: bytes, flags: CEMIFlags) -> None:
 )
 def test_frame_format_from_knx(eff: int, frame_format: CEMIFrameFormat) -> None:
     """Test parsing of the Extended Frame Format field."""
-    assert CEMIFlags.from_knx(bytes((0xBC, 0xE0 | eff))).frame_format is frame_format
+    flags = CEMIFlags.from_knx(control_field(0xBC, 0xE0 | eff))
+    assert flags.frame_format is frame_format
 
 
 @pytest.mark.parametrize("eff", [0b0001, 0b0010, 0b0011, 0b1000, 0b1100, 0b1111])
 def test_reserved_frame_format(eff: int) -> None:
     """Test parsing of a reserved Extended Frame Format."""
     with pytest.raises(ConversionError, match=r".*Reserved Extended Frame Format.*"):
-        CEMIFlags.from_knx(bytes((0xBC, 0xE0 | eff)))
+        CEMIFlags.from_knx(control_field(0xBC, 0xE0 | eff))
 
 
 @pytest.mark.parametrize(
     "priority,ctrl1",
     [
-        (CEMIPriority.SYSTEM, 0b1011_0000),
-        (CEMIPriority.NORMAL, 0b1011_0100),
-        (CEMIPriority.URGENT, 0b1011_1000),
-        (CEMIPriority.LOW, 0b1011_1100),
+        (CEMIPriority.SYSTEM, 0b0011_0000),
+        (CEMIPriority.NORMAL, 0b0011_0100),
+        (CEMIPriority.URGENT, 0b0011_1000),
+        (CEMIPriority.LOW, 0b0011_1100),
     ],
 )
 def test_priority_to_knx(priority: CEMIPriority, ctrl1: int) -> None:
     """Test priority encoding - 3/2/2 §2.2.2 Figure 28."""
-    raw = CEMIFlags(priority=priority).to_knx(
-        frame_type=CEMIFrameType.STANDARD, dst_is_group_address=False
-    )
-    assert raw[0] == ctrl1
+    # the Frame Type bit is not set by `CEMIFlags.to_knx()`
+    assert CEMIFlags(priority=priority).to_knx() >> 8 == ctrl1
 
 
-def test_to_knx_derived_fields() -> None:
-    """Test Frame Type and Address Type are supplied by the frame, not by `flags`."""
-    flags = CEMIFlags()
-    assert flags.to_knx(
-        frame_type=CEMIFrameType.STANDARD, dst_is_group_address=True
-    ) == bytes((0xBC, 0xE0))
-    assert flags.to_knx(
-        frame_type=CEMIFrameType.EXTENDED, dst_is_group_address=True
-    ) == bytes((0x3C, 0xE0))
-    assert flags.to_knx(
-        frame_type=CEMIFrameType.STANDARD, dst_is_group_address=False
-    ) == bytes((0xBC, 0x60))
+def test_to_knx_leaves_derived_bits_clear() -> None:
+    """Test Frame Type and Address Type are not set by `CEMIFlags.to_knx()`."""
+    raw = CEMIFlags().to_knx()
+    assert raw == control_field(0b0011_1100, 0b0110_0000)
+    assert CEMIFrameType.from_knx(raw) is CEMIFrameType.EXTENDED  # ie. bit not set
+    assert CEMIAddressType.from_knx(raw) is CEMIAddressType.INDIVIDUAL
+
+
+@pytest.mark.parametrize(
+    "frame_type,address_type,expected",
+    [
+        (CEMIFrameType.STANDARD, CEMIAddressType.GROUP, (0xBC, 0xE0)),
+        (CEMIFrameType.EXTENDED, CEMIAddressType.GROUP, (0x3C, 0xE0)),
+        (CEMIFrameType.STANDARD, CEMIAddressType.INDIVIDUAL, (0xBC, 0x60)),
+        (CEMIFrameType.EXTENDED, CEMIAddressType.INDIVIDUAL, (0x3C, 0x60)),
+    ],
+)
+def test_composition(
+    frame_type: CEMIFrameType,
+    address_type: CEMIAddressType,
+    expected: tuple[int, int],
+) -> None:
+    """Test the frame ORs the derived bits into the flags."""
+    raw = CEMIFlags().to_knx() | frame_type.to_knx() | address_type.to_knx()
+    assert raw == control_field(*expected)
 
 
 def test_to_knx_ignores_received_frame_type() -> None:
     """Test the Frame Type of a received frame is not used when serializing."""
-    flags = CEMIFlags.from_knx(bytes((0x3C, 0xE0)))
+    flags = CEMIFlags.from_knx(control_field(0x3C, 0xE0))
     assert flags.frame_type is CEMIFrameType.EXTENDED
-    assert flags.to_knx(
-        frame_type=CEMIFrameType.STANDARD, dst_is_group_address=True
-    ) == bytes((0xBC, 0xE0))
+    raw = (
+        flags.to_knx()
+        | CEMIFrameType.STANDARD.to_knx()
+        | CEMIAddressType.GROUP.to_knx()
+    )
+    assert raw == control_field(0xBC, 0xE0)
 
 
 def test_round_trip() -> None:
-    """Test parsing and serializing yields the same octets."""
+    """Test parsing and serializing yields the same control field."""
     for ctrl1 in (0xBC, 0xB0, 0x9C, 0xBF, 0x3C):
         for ctrl2 in (0xE0, 0x60, 0xF0, 0x00):
-            raw = bytes((ctrl1, ctrl2))
+            raw = control_field(ctrl1, ctrl2)
             flags = CEMIFlags.from_knx(raw)
             assert (
-                flags.to_knx(
-                    frame_type=flags.frame_type,
-                    dst_is_group_address=bool(ctrl2 & 0x80),
-                )
-                == raw
-            )
+                flags.to_knx()
+                | flags.frame_type.to_knx()
+                | CEMIAddressType.from_knx(raw).to_knx()
+            ) == raw
 
 
 @pytest.mark.parametrize("hop_count", [-1, 8, 255])
 def test_invalid_hop_count(hop_count: int) -> None:
     """Test hop count out of range."""
-    flags = CEMIFlags(hop_count=hop_count)
     with pytest.raises(ConversionError, match=r".*Hop count out of range.*"):
-        flags.to_knx(frame_type=CEMIFrameType.STANDARD, dst_is_group_address=True)
+        CEMIFlags(hop_count=hop_count).to_knx()
 
 
 def test_str() -> None:
     """Test the compact string representation only lists flags that are set."""
     assert str(CEMIFlags()) == "LOW STANDARD hop_count=6"
     assert (
-        str(CEMIFlags.from_knx(bytes((0b1000_1011, 0b0111_0000))))
+        str(CEMIFlags.from_knx(control_field(0b1000_1011, 0b0111_0000)))
         == "URGENT STANDARD hop_count=7 repeat_on_error system_broadcast "
         "acknowledge_request confirm_error"
     )

test/secure_tests/data_secure_test.py

@@ -7,8 +7,7 @@
 import pytest
 
 from xknx import XKNX
-from xknx.cemi import CEMIFrame, CEMILData, CEMIMessageCode
-from xknx.cemi.flags import FRAME_TYPE_STANDARD
+from xknx.cemi import CEMIFrame, CEMIFrameType, CEMILData, CEMIMessageCode
 from xknx.dpt import DPTArray
 from xknx.exceptions import DataSecureError
 from xknx.secure.data_secure import is_data_secure
@@ -483,7 +482,7 @@ def test_data_secure_authentication_only(self) -> None:
         # 4 octet plain APDU + 12 octets Data Secure overhead doesn't fit in a
         # standard frame - it is serialized as L_Data_Extended frame
         assert outgoing_raw[6] == 16
-        assert not outgoing_raw[0] & FRAME_TYPE_STANDARD
+        assert CEMIFrameType.from_knx(outgoing_raw[0] << 8) is CEMIFrameType.EXTENDED
 
         # create new cemi to avoid mixed bytearray / byte parts
         incoming_cemi = CEMIFrame.from_knx(b"\x11\x00" + outgoing_raw)

xknx/cemi/__init__.py

@@ -12,4 +12,10 @@
 )
 from .cemi_handler import CEMIHandler
 from .const import CEMIErrorCode, CEMIMessageCode
-from .flags import CEMIFlags, CEMIFrameFormat, CEMIFrameType, CEMIPriority
+from .flags import (
+    CEMIAddressType,
+    CEMIFlags,
+    CEMIFrameFormat,
+    CEMIFrameType,
+    CEMIPriority,
+)