Refactor CEMILData.flags into a typed CEMIFlags dataclass

docs/changelog.md

@@ -8,6 +8,20 @@ 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: 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 (`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 `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
 
 - 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`.
@@ -18,8 +32,9 @@ 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.
+- Reject incoming CEMI L_Data frames with a non-standard Extended Frame Format field with `UnsupportedCEMIMessage`. LTE-HEE frames use zone addressing, so their address fields can not be parsed as `GroupAddress`; reserved encodings are rejected too. 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. The `CEMIFrameFormat` enum covers the field instead, resolving the whole `01xxb` range to `LTE_HEE`.
+- 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

@@ -3,14 +3,17 @@
 import pytest
 
 from xknx.cemi import (
+    CEMIAddressType,
     CEMIFlags,
     CEMIFrame,
+    CEMIFrameType,
     CEMILData,
     CEMIMessageCode,
     CEMIMPropReadRequest,
     CEMIMPropReadResponse,
     CEMIMPropWriteRequest,
     CEMIMPropWriteResponse,
+    CEMIPriority,
 )
 from xknx.cemi.const import CEMIErrorCode
 from xknx.dpt import DPTArray
@@ -56,8 +59,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 +79,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 +171,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 +245,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.address_type is CEMIAddressType.GROUP
+    assert frame.data.flags == CEMIFlags(priority=CEMIPriority.LOW)
     # test CEMIFrame.telegram property
     assert frame.data.telegram() == _telegram
 
@@ -249,8 +259,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.address_type is CEMIAddressType.GROUP
+    assert frame.data.flags == CEMIFlags(priority=CEMIPriority.SYSTEM)
     assert frame.data.tpci == TDataBroadcast()
     # test CEMIFrame.telegram property
     assert frame.data.telegram() == _telegram
@@ -264,9 +274,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 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
     assert frame.data.telegram() == _telegram
 
@@ -289,45 +299,41 @@ 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 (
+        CEMIFrameType.from_knx(raw[0] << 8) is CEMIFrameType.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 CEMIFrameType.from_knx(cemi_data.to_knx()[0] << 8) is CEMIFrameType.EXTENDED
 
-    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 CEMIFrameType.from_knx(cemi_data.to_knx()[0] << 8) is CEMIFrameType.STANDARD
 
 
 def test_npdu_length_exceeded() -> None:
@@ -336,23 +342,30 @@ def test_npdu_length_exceeded() -> None:
         _cemi_l_data_from_payload(GroupValueWrite(DPTArray(bytes(254))))
 
 
-def test_extended_frame_format_not_supported() -> None:
+@pytest.mark.parametrize(
+    "eff,err_msg",
+    [
+        # 01xxb - LTE-HEE zone addressed frames
+        (0b0100, r".*Extended Frame Format not supported: LTE_HEE.*"),
+        (0b0111, r".*Extended Frame Format not supported: LTE_HEE.*"),
+        # reserved
+        (0b0001, r".*Reserved Extended Frame Format: 0b0001.*"),
+        (0b1111, r".*Reserved Extended Frame Format: 0b1111.*"),
+    ],
+)
+def test_extended_frame_format_not_supported(eff: int, err_msg: str) -> None:
     """Test parsing of LTE-HEE and reserved Extended Frame Formats."""
     raw = get_data(
         0x29,
         0,
-        CEMIFlags.FRAME_TYPE_EXTENDED
-        | CEMIFlags.DESTINATION_GROUP_ADDRESS
-        | CEMIFlags.LTE_FRAME_FORMAT,
+        (CEMIAddressType.GROUP.to_knx() | eff),  # Ctrl2; Ctrl1 = extended frame
         1,
         1,
         1,
         0,
         [],
     )
-    with pytest.raises(
-        UnsupportedCEMIMessage, match=r".*Extended Frame Format not supported.*"
-    ):
+    with pytest.raises(UnsupportedCEMIMessage, match=err_msg):
         CEMIFrame.from_knx(raw)
 
 

test/cemi_tests/flags_test.py

@@ -0,0 +1,172 @@
+"""Tests for the cEMI L_Data control fields."""
+
+import pytest
+
+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(
+    "ctrl1,ctrl2,flags",
+    [
+        (
+            # Ctrl1 0xBC: standard frame, do not repeat, broadcast, low priority
+            # Ctrl2 0xE0: group address, hop count 6, standard frame format
+            0xBC,
+            0xE0,
+            CEMIFlags(priority=CEMIPriority.LOW, hop_count=6),
+        ),
+        (
+            # Ctrl1 0x3C: extended frame - kept, but not evaluated
+            0x3C,
+            0xE0,
+            CEMIFlags(
+                priority=CEMIPriority.LOW,
+                hop_count=6,
+                frame_type=CEMIFrameType.EXTENDED,
+            ),
+        ),
+        (
+            # Ctrl1 0xB0: system priority; Ctrl2 0x60: individual address
+            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
+            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(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(
+    "eff,frame_format",
+    [
+        (0b0000, CEMIFrameFormat.STANDARD),
+        # the whole 01xxb range is LTE-HEE; the 2 lsb are the LTE address type
+        (0b0100, CEMIFrameFormat.LTE_HEE),
+        (0b0101, CEMIFrameFormat.LTE_HEE),
+        (0b0110, CEMIFrameFormat.LTE_HEE),
+        (0b0111, CEMIFrameFormat.LTE_HEE),
+    ],
+)
+def test_frame_format_from_knx(eff: int, frame_format: CEMIFrameFormat) -> None:
+    """Test parsing of the Extended Frame Format field."""
+    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(control_field(0xBC, 0xE0 | eff))
+
+
+@pytest.mark.parametrize(
+    "priority,ctrl1",
+    [
+        (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."""
+    # the Frame Type bit is not set by `CEMIFlags.to_knx()`
+    assert CEMIFlags(priority=priority).to_knx() >> 8 == ctrl1
+
+
+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(control_field(0x3C, 0xE0))
+    assert flags.frame_type is CEMIFrameType.EXTENDED
+    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 control field."""
+    for ctrl1 in (0xBC, 0xB0, 0x9C, 0xBF, 0x3C):
+        for ctrl2 in (0xE0, 0x60, 0xF0, 0x00):
+            raw = control_field(ctrl1, ctrl2)
+            flags = CEMIFlags.from_knx(raw)
+            assert (
+                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."""
+    with pytest.raises(ConversionError, match=r".*Hop count out of range.*"):
+        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(control_field(0b1000_1011, 0b0111_0000)))
+        == "URGENT STANDARD hop_count=7 repeat_on_error system_broadcast "
+        "acknowledge_request confirm_error"
+    )