Refactor CEMILData.flags into a typed CEMIFlags dataclass

docs/changelog.md

@@ -8,6 +8,18 @@ nav_order: 2
 
 # Unreleased changes
 
+### Breaking changes
+
+- `CEMILData.flags` is now a `CEMIFlags` object instead of a 16 bit `int`. `CEMIFlags` was previously a collection of bit constants; it is now a slotted dataclass holding the control field values that are independent state:
+  - `priority: CEMIPriority` - new `SYSTEM` / `NORMAL` / `URGENT` / `LOW` enum
+  - `repeat_on_error: bool`, `system_broadcast: bool` - named for the positive meaning; both are inverted on the wire
+  - `acknowledge_request: bool`
+  - `confirm_error: bool` - only meaningful in an L_Data.con frame
+  - `hop_count: int` - replaces the `CEMILData.hops` property, which was removed
+  Frame Type, Address Type and Extended Frame Format are no longer stored: they follow from the NPDU length, the destination address type and the supported frame formats, and are derived when serializing. `flags` therefore can no longer disagree with the frame that is put on the wire. The raw bit masks moved to `xknx.cemi.flags` as module level constants.
+- `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.
+
 ### Bugfixes
 
 - Send frames with an APDU longer than 15 octets as L_Data_Extended frames. The Frame Type flag of Ctrl1 is now derived from the NPDU length when serializing a `CEMILData` instead of always being set to standard frame. This fixes sending telegrams over KNX Data Secure with a payload of 2 octets or more (eg. DPT 9.x or DPT 232.600) - Data Secure adds 12 octets to the plain APDU, which no longer fits in a standard frame. Sending an APDU longer than 254 octets now raises `ConversionError` instead of `OverflowError`.
@@ -19,7 +31,8 @@ nav_order: 2
 ### Protocol
 
 - Reject incoming CEMI L_Data frames with a non-zero Extended Frame Format field with `UnsupportedCEMIMessage`. LTE-HEE frames use zone addressing, so their address fields can not be parsed as `GroupAddress`. The Frame Type flag is still not validated against the NPDU length when parsing - a receiver shall be tolerant towards the used frame format.
-- `CEMIFlags.EXTENDED_FRAME_FORMAT` was removed; its value `0x0001` was reserved, not an "extended frame format" indicator - `0x0000` is used for standard frames as well as for long extended frames. `CEMIFlags.LTE_FRAME_FORMAT` and `CEMIFlags.EXTENDED_FRAME_FORMAT_MASK` were added instead.
+- `CEMIFlags.EXTENDED_FRAME_FORMAT` was removed; its value `0x0001` was reserved, not an "extended frame format" indicator - `0x0000` is used for standard frames as well as for long extended frames. `LTE_FRAME_FORMAT` and `EXTENDED_FRAME_FORMAT_MASK` were added instead.
+- The hop count is validated when serializing; a value outside `0..7` raises `ConversionError` instead of silently corrupting Ctrl2.
 - Add explicit length checks to every remaining APCI `from_knx` (and the top-level `APCI.from_knx` dispatcher) as defense-in-depth on top of the broad `except (IndexError, struct.error, ValueError)` added in 3.17.0: each service now raises `ConversionError` with a specific "Invalid length for A_X in CEMI" message for a truncated, malformed or overlong frame instead of relying solely on the generic dispatcher-level catch.
 
 # 3.17.0 APCIs and DPTs 2026-07-25

test/cemi_tests/cemi_frame_test.py

@@ -11,8 +11,14 @@
     CEMIMPropReadResponse,
     CEMIMPropWriteRequest,
     CEMIMPropWriteResponse,
+    CEMIPriority,
 )
 from xknx.cemi.const import CEMIErrorCode
+from xknx.cemi.flags import (
+    DESTINATION_GROUP_ADDRESS,
+    FRAME_TYPE_STANDARD,
+    LTE_FRAME_FORMAT,
+)
 from xknx.dpt import DPTArray
 from xknx.exceptions import ConversionError, CouldNotParseCEMI, UnsupportedCEMIMessage
 from xknx.profile.const import ResourceKNXNETIPPropertyId, ResourceObjectType
@@ -56,8 +62,12 @@ def test_valid_command() -> None:
     frame = CEMIFrame.from_knx(raw)
     assert frame.code == CEMIMessageCode.L_DATA_IND
     assert isinstance(frame.data, CEMILData)
-    assert frame.data.flags == 0x8080
-    assert frame.data.hops == 0
+    assert frame.data.flags == CEMIFlags(
+        priority=CEMIPriority.SYSTEM,
+        repeat_on_error=True,
+        system_broadcast=True,
+        hop_count=0,
+    )
     assert frame.data.src_addr == IndividualAddress(1)
     assert frame.data.dst_addr == GroupAddress(1)
     assert frame.data.payload == GroupValueRead()
@@ -72,8 +82,12 @@ def test_valid_tpci_control() -> None:
     frame = CEMIFrame.from_knx(raw)
     assert frame.code == CEMIMessageCode.L_DATA_IND
     assert isinstance(frame.data, CEMILData)
-    assert frame.data.flags == 0x8000
-    assert frame.data.hops == 0
+    assert frame.data.flags == CEMIFlags(
+        priority=CEMIPriority.SYSTEM,
+        repeat_on_error=True,
+        system_broadcast=True,
+        hop_count=0,
+    )
     assert frame.data.payload is None
     assert frame.data.src_addr == IndividualAddress(0)
     assert frame.data.dst_addr == IndividualAddress(0)
@@ -160,7 +174,6 @@ def test_invalid_payload() -> None:
     frame = CEMIFrame(
         code=CEMIMessageCode.L_DATA_IND,
         data=CEMILData(
-            flags=0,
             src_addr=IndividualAddress(0),
             dst_addr=IndividualAddress(0),
             tpci=TDataGroup(),
@@ -235,8 +248,8 @@ def test_telegram_group_address() -> None:
         data=CEMILData.init_from_telegram(_telegram),
     )
     assert isinstance(frame.data, CEMILData)
-    assert frame.data.flags & 0x0080 == CEMIFlags.DESTINATION_GROUP_ADDRESS
-    assert frame.data.flags & 0x0C00 == CEMIFlags.PRIORITY_LOW
+    assert frame.data.dst_is_group_address
+    assert frame.data.flags == CEMIFlags(priority=CEMIPriority.LOW)
     # test CEMIFrame.telegram property
     assert frame.data.telegram() == _telegram
 
@@ -249,8 +262,8 @@ def test_telegram_broadcast() -> None:
         data=CEMILData.init_from_telegram(_telegram),
     )
     assert isinstance(frame.data, CEMILData)
-    assert frame.data.flags & 0x0080 == CEMIFlags.DESTINATION_GROUP_ADDRESS
-    assert frame.data.flags & 0x0C00 == CEMIFlags.PRIORITY_SYSTEM
+    assert frame.data.dst_is_group_address
+    assert frame.data.flags == CEMIFlags(priority=CEMIPriority.SYSTEM)
     assert frame.data.tpci == TDataBroadcast()
     # test CEMIFrame.telegram property
     assert frame.data.telegram() == _telegram
@@ -264,9 +277,9 @@ def test_telegram_individual_address() -> None:
         data=CEMILData.init_from_telegram(_telegram),
     )
     assert isinstance(frame.data, CEMILData)
-    assert frame.data.flags & 0x0080 == CEMIFlags.DESTINATION_INDIVIDUAL_ADDRESS
-    assert frame.data.flags & 0x0C00 == CEMIFlags.PRIORITY_SYSTEM
-    assert frame.data.flags & 0x0200 == CEMIFlags.NO_ACK_REQUESTED
+    assert not frame.data.dst_is_group_address
+    assert frame.data.flags == CEMIFlags(priority=CEMIPriority.SYSTEM)
+    assert not frame.data.flags.acknowledge_request
     # test CEMIFrame.telegram property
     assert frame.data.telegram() == _telegram
 
@@ -289,45 +302,39 @@ def _cemi_l_data_from_payload(payload: GroupValueWrite) -> bytes:
 
 
 @pytest.mark.parametrize(
-    "apdu_payload_length,expected_npdu_len,expected_frame_type",
+    "apdu_payload_length,expected_npdu_len,expected_standard_frame",
     [
-        (1, 2, CEMIFlags.FRAME_TYPE_STANDARD),
+        (1, 2, True),
         # 15 octets after the TPCI octet is the maximum of a standard frame
-        (14, 15, CEMIFlags.FRAME_TYPE_STANDARD),
-        (15, 16, CEMIFlags.FRAME_TYPE_EXTENDED),
-        (253, 254, CEMIFlags.FRAME_TYPE_EXTENDED),
+        (14, 15, True),
+        (15, 16, False),
+        (253, 254, False),
     ],
 )
 def test_frame_type_from_npdu_length(
-    apdu_payload_length: int, expected_npdu_len: int, expected_frame_type: int
+    apdu_payload_length: int, expected_npdu_len: int, expected_standard_frame: bool
 ) -> None:
     """Test Frame Type flag is derived from the NPDU length."""
     raw = _cemi_l_data_from_payload(
         GroupValueWrite(DPTArray(bytes(apdu_payload_length)))
     )
     assert raw[6] == expected_npdu_len
-    assert (raw[0] << 8) & CEMIFlags.FRAME_TYPE_STANDARD == expected_frame_type
+    assert bool(raw[0] & FRAME_TYPE_STANDARD) is expected_standard_frame
 
 
-def test_frame_type_overrides_flags() -> None:
-    """Test Frame Type flag of `flags` is overridden by the payload length."""
-    long_payload = GroupValueWrite(DPTArray(bytes(15)))
-    short_payload = GroupValueWrite(DPTArray(bytes(1)))
+def test_frame_type_follows_payload() -> None:
+    """Test the Frame Type follows the current payload; it is not held by `flags`."""
     cemi_data = CEMILData(
-        # standard frame flag set although the payload requires an extended frame
-        flags=CEMIFlags.FRAME_TYPE_STANDARD | CEMIFlags.DESTINATION_GROUP_ADDRESS,
         src_addr=IndividualAddress(1),
         dst_addr=GroupAddress(1),
         tpci=TDataGroup(),
-        payload=long_payload,
+        payload=GroupValueWrite(DPTArray(bytes(15))),
     )
-    assert not cemi_data.to_knx()[0] & 0x80
-    # `flags` is not modified by serialization
-    assert cemi_data.flags & CEMIFlags.FRAME_TYPE_STANDARD
+    assert not cemi_data.to_knx()[0] & FRAME_TYPE_STANDARD
 
-    cemi_data.flags = CEMIFlags.DESTINATION_GROUP_ADDRESS  # extended frame flag
-    cemi_data.payload = short_payload
-    assert cemi_data.to_knx()[0] & 0x80
+    # 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
 
 
 def test_npdu_length_exceeded() -> None:
@@ -341,9 +348,7 @@ def test_extended_frame_format_not_supported() -> None:
     raw = get_data(
         0x29,
         0,
-        CEMIFlags.FRAME_TYPE_EXTENDED
-        | CEMIFlags.DESTINATION_GROUP_ADDRESS
-        | CEMIFlags.LTE_FRAME_FORMAT,
+        (DESTINATION_GROUP_ADDRESS | LTE_FRAME_FORMAT),  # Ctrl2; Ctrl1 = extended
         1,
         1,
         1,

test/cemi_tests/flags_test.py

@@ -0,0 +1,99 @@
+"""Tests for the cEMI L_Data control fields."""
+
+import pytest
+
+from xknx.cemi import CEMIFlags, CEMIPriority
+from xknx.exceptions import ConversionError
+
+
+@pytest.mark.parametrize(
+    "raw,flags",
+    [
+        (
+            # Ctrl1 0xBC: standard frame, do not repeat, broadcast, low priority
+            # Ctrl2 0xE0: group address, hop count 6, standard frame format
+            bytes((0xBC, 0xE0)),
+            CEMIFlags(priority=CEMIPriority.LOW, hop_count=6),
+        ),
+        (
+            # Ctrl1 0x3C: extended frame - not evaluated when parsing
+            bytes((0x3C, 0xE0)),
+            CEMIFlags(priority=CEMIPriority.LOW, hop_count=6),
+        ),
+        (
+            # Ctrl1 0xB0: system priority; Ctrl2 0x60: individual address
+            bytes((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)),
+            CEMIFlags(
+                priority=CEMIPriority.URGENT,
+                repeat_on_error=True,
+                system_broadcast=True,
+                acknowledge_request=True,
+                confirm_error=True,
+                hop_count=7,
+            ),
+        ),
+    ],
+)
+def test_from_knx(raw: bytes, flags: CEMIFlags) -> None:
+    """Test parsing of Ctrl1 and Ctrl2."""
+    assert CEMIFlags.from_knx(raw) == flags
+
+
+@pytest.mark.parametrize(
+    "priority,ctrl1",
+    [
+        (CEMIPriority.SYSTEM, 0b1011_0000),
+        (CEMIPriority.NORMAL, 0b1011_0100),
+        (CEMIPriority.URGENT, 0b1011_1000),
+        (CEMIPriority.LOW, 0b1011_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_standard=True, dst_is_group_address=False
+    )
+    assert raw[0] == 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_standard=True, dst_is_group_address=True) == bytes(
+        (0xBC, 0xE0)
+    )
+    assert flags.to_knx(frame_type_standard=False, dst_is_group_address=True) == bytes(
+        (0x3C, 0xE0)
+    )
+    assert flags.to_knx(frame_type_standard=True, dst_is_group_address=False) == bytes(
+        (0xBC, 0x60)
+    )
+
+
+def test_round_trip() -> None:
+    """Test parsing and serializing yields the same octets."""
+    for ctrl1 in (0xBC, 0xB0, 0x9C, 0xBF):
+        for ctrl2 in (0xE0, 0x60, 0xF0, 0x00):
+            raw = bytes((ctrl1, ctrl2))
+            flags = CEMIFlags.from_knx(raw)
+            assert (
+                flags.to_knx(
+                    frame_type_standard=bool(ctrl1 & 0x80),
+                    dst_is_group_address=bool(ctrl2 & 0x80),
+                )
+                == 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_standard=True, dst_is_group_address=True)

test/knxip_tests/routing_indication_test.py

@@ -44,7 +44,7 @@ def test_telegram_set(self) -> None:
             ),
         )
         assert isinstance(cemi.data, CEMILData)
-        cemi.data.hops = 5
+        cemi.data.flags.hop_count = 5
         routing_indication = RoutingIndication(raw_cemi=cemi.to_knx())
         knxipframe = KNXIPFrame.init_from_body(routing_indication)
 

test/secure_tests/data_secure_test.py

@@ -7,7 +7,8 @@
 import pytest
 
 from xknx import XKNX
-from xknx.cemi import CEMIFlags, CEMIFrame, CEMILData, CEMIMessageCode
+from xknx.cemi import CEMIFrame, CEMILData, CEMIMessageCode
+from xknx.cemi.flags import FRAME_TYPE_STANDARD
 from xknx.dpt import DPTArray
 from xknx.exceptions import DataSecureError
 from xknx.secure.data_secure import is_data_secure
@@ -482,7 +483,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] & CEMIFlags.FRAME_TYPE_STANDARD >> 8
+        assert not outgoing_raw[0] & FRAME_TYPE_STANDARD
 
         # create new cemi to avoid mixed bytearray / byte parts
         incoming_cemi = CEMIFrame.from_knx(b"\x11\x00" + outgoing_raw)

test/str_test.py

@@ -735,7 +735,9 @@ def test_cemi_ldata_frame(self) -> None:
         assert (
             str(cemi_frame)
             == '<CEMIFrame code="L_DATA_IND" info="CEMIInfo("")" data="CEMILData(src_addr="IndividualAddress("1.2.3")" '
-            'dst_addr="GroupAddress("1/2/5")" flags="1011110011100000" tpci="TDataGroup()" '
+            'dst_addr="GroupAddress("1/2/5")" flags="CEMIFlags(priority=LOW hop_count=6 '
+            "repeat_on_error=False system_broadcast=False acknowledge_request=False "
+            'confirm_error=False)" tpci="TDataGroup()" '
             'payload="<GroupValueWrite value="<DPTBinary value="7" />" />")" />'
         )
 

xknx/cemi/__init__.py

@@ -11,4 +11,5 @@
     CEMIMPropWriteResponse,
 )
 from .cemi_handler import CEMIHandler
-from .const import CEMIErrorCode, CEMIFlags, CEMIMessageCode
+from .const import CEMIErrorCode, CEMIMessageCode
+from .flags import CEMIFlags, CEMIPriority