Repository: XKNX/xknx
Pull request #1661 (2025-02-12 09:47:52)

Collect group addresses with decoding errors at eager decoder
<!--
  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.
-->


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)
- [ ] Code quality improvements to existing code or addition of tests

## Checklist

- [ ] The documentation has been adjusted accordingly
- [ ] Tests have been added that prove the fix is effective or that the feature works
- [x] The changes are documented in the changelog (docs/changelog.md)


Patch:
diff --git a/docs/changelog.md b/docs/changelog.md
index 36fdaf98..abe4c1fc 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -11,6 +11,10 @@ nav_order: 2
 - Add `DPTBase.dpt_number_str` and `DPTBase.dpt_name` classmethods for human readable DPT number (eg. "9.001") and class name (eg. "DPTTemperature (9.001)").
 - Add `DPTBase.get_dpt` classmethod to get a DPT class by its number, name or DPTBase type.
 
+### Features
+
+- Collect group addresses with decoding errors at eager decoder in `xknx.group_address_dpt.ga_decoding_error` set.
+
 # 3.5.0 Swing it 2025-01-28
 
 ### Devices
diff --git a/xknx/core/group_address_dpt.py b/xknx/core/group_address_dpt.py
index ff105c28..3d0ee546 100644
--- a/xknx/core/group_address_dpt.py
+++ b/xknx/core/group_address_dpt.py
@@ -24,12 +24,13 @@ _GA_DPT_LOGGER = logging.getLogger("xknx.ga_dpt")
 class GroupAddressDPT:
     """Class for mapping group addresses to data point types for eager decoding."""
 
-    __slots__ = ("_ga_dpts",)
+    __slots__ = ("_ga_dpts", "ga_decoding_error")
 
     def __init__(self) -> None:
         """Initialize GADataTypes class."""
         # using dict[int | str] instead of dict[DeviceGroupAddress] is faster.
         self._ga_dpts: dict[int | str, type[DPTBase]] = {}
+        self.ga_decoding_error: set[GroupAddress | InternalGroupAddress] = set()
 
     def set(
         self,
@@ -72,7 +73,12 @@ class GroupAddressDPT:
         try:
             value = transcoder.from_knx(telegram.payload.value)
         except (CouldNotParseTelegram, ConversionError) as err:
-            _GA_DPT_LOGGER.debug(
+            if telegram.destination_address in self.ga_decoding_error:
+                _logger_fn = _GA_DPT_LOGGER.debug
+            else:
+                _logger_fn = _GA_DPT_LOGGER.warning
+            self.ga_decoding_error.add(telegram.destination_address)
+            _logger_fn(
                 "DPT decoding error. Telegram from %s to %s with payload %s can't be decoded by %s: %s",
                 telegram.source_address,
                 telegram.destination_address,


Test patch:
diff --git a/test/core_tests/group_address_dpt_test.py b/test/core_tests/group_address_dpt_test.py
index 49008828..596e254e 100644
--- a/test/core_tests/group_address_dpt_test.py
+++ b/test/core_tests/group_address_dpt_test.py
@@ -45,9 +45,10 @@ async def test_group_address_dpt_in_telegram_queue(xknx_no_interface: XKNX) -> N
         assert test_telegram.decoded_data is None
 
 
+@patch("logging.Logger.warning")
 @patch("logging.Logger.debug")
 async def test_get_invalid_payload(
-    logger_debug_mock: Mock, xknx_no_interface: XKNX
+    logger_debug_mock: Mock, logger_warning_mock: Mock, xknx_no_interface: XKNX
 ) -> None:
     """Test processing when DPT doesn't fit payload."""
     xknx = xknx_no_interface
@@ -61,15 +62,31 @@ async def test_get_invalid_payload(
         payload=apci.GroupValueWrite(DPTArray((0x01, 0x02))),  # wrong payload size
     )
     async with xknx:
-        xknx.telegrams.put_nowait(telegram)
         logger_debug_mock.reset_mock()
+        logger_warning_mock.reset_mock()
+        assert len(xknx.group_address_dpt.ga_decoding_error) == 0
+
+        xknx.telegrams.put_nowait(telegram)
         await xknx.telegrams.join()
+        assert len(xknx.group_address_dpt.ga_decoding_error) == 1
         assert telegram_callback.call_count == 1
         test_telegram = telegram_callback.call_args[0][0]
         assert test_telegram.decoded_data is None
-        assert logger_debug_mock.call_count == 2
+        assert logger_debug_mock.call_count == 1  # telegram log
+        assert logger_warning_mock.call_count == 1  # first occurrence goes to warning
+        assert "DPT decoding error" in logger_warning_mock.call_args_list[0][0][0]
+        # second telegram with same GA should go to debug
+        logger_debug_mock.reset_mock()
+        logger_warning_mock.reset_mock()
+        xknx.telegrams.put_nowait(telegram)
+        await xknx.telegrams.join()
+        assert len(xknx.group_address_dpt.ga_decoding_error) == 1  # still 1 - same GA
+        assert logger_warning_mock.call_count == 0
+        assert logger_debug_mock.call_count == 2  # second occurrence goes to debug
         assert "DPT decoding error" in logger_debug_mock.call_args_list[0][0][0]
 
+        assert xknx.group_address_dpt.ga_decoding_error == {GroupAddress("1/2/3")}
+
 
 def test_set(xknx_no_interface: XKNX) -> None:
     """Test set."""