Repository: XKNX/xknx
Pull request #1483 (2024-06-11 19:55:27)

Eager telegram data decoding for GroupValueWrite / GroupValueResponse Telegrams
<!--
  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.
-->
Decoded data and transcoder is added to `Telegram` to be used in Devices and Telegram-callbacks. DPT for each Group Address can be set by `xknx.group_address_dpt.set()`

- [x] let user add DPTs for group addresses
- [x] decode payloads of incoming Telegrams when found in group_address_dpt table
- [x] set decoded payloads for outgoing Telegrams
- [x] check if already decoded data in incoming Telegram matches DPT used in Device / RemoteValue to avoid double-decoding
- [x] add decoded data to Telegram string representation

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
- [ ] The changes are documented in the changelog (docs/changelog.md)


Patch:
diff --git a/docs/changelog.md b/docs/changelog.md
index 6aa92bea..db5dddfa 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -11,6 +11,7 @@ nav_order: 2
 ### General
 
 - Drop support for Python 3.9
+- Added eager telegram data decoding for GroupValueWrite / GroupValueResponse Telegrams. DPTs for group addresses can be set using `xknx.group_address_dpt.set()`. `Telegram` has a new attribute `decoded_data` which is set when a decoder was found.
 
 ### Devices
 
@@ -36,6 +37,7 @@ nav_order: 2
 - Use `slots` in addresses, Telegram, DPTBinary, DPTArray, TPCI, APCI, DPTComplexData
 - Convert Telegram and APCI to dataclasses. `Telegram` is not hashable anymore.
 - Use `asyncio.gather` where applicable (mostly device telegram process pipeline)
+- RemoteValue instances use pre-decoded data from Telegrams if available and `dpt_class` for is set - otherwise they decode the data themselves in `from_knx` like before.
 
 # 2.12.2 Fix thread leak 2024-03-05
 
diff --git a/xknx/core/__init__.py b/xknx/core/__init__.py
index 69a04803..3cfbdc67 100644
--- a/xknx/core/__init__.py
+++ b/xknx/core/__init__.py
@@ -3,6 +3,7 @@
 # flake8: noqa
 from .connection_manager import ConnectionManager
 from .connection_state import XknxConnectionState, XknxConnectionType
+from .group_address_dpt import GroupAddressDPT
 from .state_updater import StateUpdater
 from .task_registry import Task, TaskRegistry
 from .telegram_queue import TelegramQueue
diff --git a/xknx/core/group_address_dpt.py b/xknx/core/group_address_dpt.py
new file mode 100644
index 00000000..d3702337
--- /dev/null
+++ b/xknx/core/group_address_dpt.py
@@ -0,0 +1,78 @@
+"""Group address data point type table."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+import logging
+
+from xknx.dpt.dpt import DPTBase, _DPTMainSubDict
+from xknx.exceptions import ConversionError, CouldNotParseAddress, CouldNotParseTelegram
+from xknx.telegram import Telegram, TelegramDecodedData
+from xknx.telegram.address import (
+    DeviceAddressableType,
+    DeviceGroupAddress,
+    GroupAddress,
+    InternalGroupAddress,
+    parse_device_group_address,
+)
+from xknx.telegram.apci import GroupValueResponse, GroupValueWrite
+
+_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",)
+
+    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]] = {}
+
+    def set(
+        self,
+        ga_dpt: Mapping[DeviceAddressableType, int | str | _DPTMainSubDict],
+    ) -> None:
+        """Assign decoders to group addresses."""
+        unknown_dpts = set()
+        for addr, dpt in ga_dpt.items():
+            try:
+                address = parse_device_group_address(addr)
+            except CouldNotParseAddress as err:
+                _GA_DPT_LOGGER.warning("Invalid group address %s: %s", addr, err)
+                continue
+            if (transcoder := DPTBase.parse_transcoder(dpt)) is None:
+                unknown_dpts.add(repr(dpt))  # prevent unhashable types (dict)
+                continue
+            self._ga_dpts[address.raw] = transcoder  # type: ignore[type-abstract]
+        if unknown_dpts:
+            _GA_DPT_LOGGER.debug("No transcoder found for DPTs: %s", unknown_dpts)
+
+    def get(self, address: DeviceGroupAddress) -> type[DPTBase] | None:
+        """Return transcoder for group address."""
+        return self._ga_dpts.get(address.raw)
+
+    def clear(self) -> None:
+        """Clear all group addresses."""
+        self._ga_dpts = {}
+
+    def set_decoded_data(self, telegram: Telegram) -> None:
+        """Update telegram data with decoded value."""
+        if telegram.decoded_data is not None:
+            return
+        if not isinstance(telegram.payload, GroupValueWrite | GroupValueResponse):
+            return
+        assert isinstance(  # GroupValueWrite and GroupValueResponse can not have IndividualAddress
+            telegram.destination_address, GroupAddress | InternalGroupAddress
+        )
+        if (transcoder := self.get(telegram.destination_address)) is None:
+            return
+        try:
+            value = transcoder.from_knx(telegram.payload.value)
+        except (CouldNotParseTelegram, ConversionError) as err:
+            _GA_DPT_LOGGER.warning(
+                "DPT decoding error for %s of %s: %s", transcoder, self, err
+            )
+            return
+        telegram.decoded_data = TelegramDecodedData(transcoder, value)
diff --git a/xknx/core/telegram_queue.py b/xknx/core/telegram_queue.py
index 6a8044e4..9f4c392e 100644
--- a/xknx/core/telegram_queue.py
+++ b/xknx/core/telegram_queue.py
@@ -125,6 +125,8 @@ class TelegramQueue:
                 self.xknx.telegrams.task_done()
                 break
 
+            self.xknx.group_address_dpt.set_decoded_data(telegram)
+
             if telegram.direction == TelegramDirection.INCOMING:
                 try:
                     await self.process_telegram_incoming(telegram)
diff --git a/xknx/devices/light.py b/xknx/devices/light.py
index 96a9e373..a1d87e6b 100644
--- a/xknx/devices/light.py
+++ b/xknx/devices/light.py
@@ -196,6 +196,7 @@ class Light(Device):
             group_address_color_state,
             sync_state=sync_state,
             device_name=self.name,
+            feature_name="Color RGB",
             after_update_cb=self.after_update,
         )
 
@@ -205,6 +206,7 @@ class Light(Device):
             group_address_rgbw_state,
             sync_state=sync_state,
             device_name=self.name,
+            feature_name="Color RGBW",
             after_update_cb=self.after_update,
         )
 
@@ -237,6 +239,7 @@ class Light(Device):
             group_address_xyy_color_state,
             sync_state=sync_state,
             device_name=self.name,
+            feature_name="Color XYY",
             after_update_cb=self._xyy_color_from_rv,
         )
 
diff --git a/xknx/remote_value/remote_value.py b/xknx/remote_value/remote_value.py
index 76587f69..6ac307ad 100644
--- a/xknx/remote_value/remote_value.py
+++ b/xknx/remote_value/remote_value.py
@@ -9,12 +9,12 @@ Remote value can be :
 
 from __future__ import annotations
 
-from abc import ABC, abstractmethod
+from abc import ABC
 from collections.abc import Awaitable, Callable, Iterator
 import logging
 from typing import TYPE_CHECKING, Generic, TypeVar, Union
 
-from xknx.dpt import DPTArray, DPTBinary
+from xknx.dpt import DPTArray, DPTBase, DPTBinary
 from xknx.exceptions import ConversionError, CouldNotParseTelegram
 from xknx.telegram import GroupAddress, Telegram
 from xknx.telegram.address import (
@@ -40,6 +40,8 @@ ValueT = TypeVar("ValueT")
 class RemoteValue(ABC, Generic[ValueT]):
     """Class for managing remote knx value."""
 
+    dpt_class: type[DPTBase] | None = None
+
     def __init__(
         self,
         xknx: XKNX,
@@ -152,13 +154,21 @@ class RemoteValue(ABC, Generic[ValueT]):
 
         return group_address in remote_value_addresses()
 
-    @abstractmethod
     def from_knx(self, payload: DPTArray | DPTBinary) -> ValueT:
-        """Convert current payload to value - to be implemented in derived class."""
+        """Convert current payload to value - to be implemented in derived class when `dpt_class` can't be used."""
+        if self.dpt_class is None:
+            raise NotImplementedError(
+                "Either `dpt_class` must be set or `from_knx` must be implemented"
+            )
+        return self.dpt_class.from_knx(payload)  # type: ignore[no-any-return]
 
-    @abstractmethod
     def to_knx(self, value: ValueT) -> DPTArray | DPTBinary:
-        """Convert value to payload - to be implemented in derived class."""
+        """Convert value to payload - to be implemented in derived class when `dpt_class` can't be used."""
+        if self.dpt_class is None:
+            raise NotImplementedError(
+                "Either `dpt_class` must be set or `to_knx` must be implemented"
+            )
+        return self.dpt_class.to_knx(value)
 
     async def process(self, telegram: Telegram, always_callback: bool = False) -> bool:
         """Process incoming or outgoing telegram."""
@@ -177,7 +187,14 @@ class RemoteValue(ABC, Generic[ValueT]):
             )
 
         try:
-            decoded_payload = self.from_knx(telegram.payload.value)
+            decoded_payload: ValueT
+            if (
+                telegram.decoded_data is not None
+                and telegram.decoded_data.transcoder is self.dpt_class
+            ):
+                decoded_payload = telegram.decoded_data.value  # type: ignore[assignment]
+            else:
+                decoded_payload = self.from_knx(telegram.payload.value)
         except (ConversionError, CouldNotParseTelegram) as err:
             logger.warning(
                 "Can not process %s for %s - %s: %s",
@@ -199,17 +216,16 @@ class RemoteValue(ABC, Generic[ValueT]):
         self, payload: DPTArray | DPTBinary, response: bool = False
     ) -> None:
         """Send payload as telegram to KNX bus."""
-        if self.group_address is not None:
-            telegram = Telegram(
-                destination_address=self.group_address,
-                payload=(
-                    GroupValueResponse(payload)
-                    if response
-                    else GroupValueWrite(payload)
-                ),
-                source_address=self.xknx.current_address,
-            )
-            await self.xknx.telegrams.put(telegram)
+        if self.group_address is None:
+            return
+        telegram = Telegram(
+            destination_address=self.group_address,
+            payload=(
+                GroupValueResponse(payload) if response else GroupValueWrite(payload)
+            ),
+            source_address=self.xknx.current_address,
+        )
+        await self.xknx.telegrams.put(telegram)
 
     async def set(self, value: ValueT, response: bool = False) -> None:
         """Set new value."""
@@ -229,16 +245,16 @@ class RemoteValue(ABC, Generic[ValueT]):
                 value,
             )
             return
-
         payload = self.to_knx(value)
         await self._send(payload, response)
         # self._value is set and after_update_cb() called when the outgoing telegram is processed.
 
     async def respond(self) -> None:
         """Send current payload as GroupValueResponse telegram to KNX bus."""
-        if self._value is not None:
-            payload = self.to_knx(self._value)
-            await self._send(payload, response=True)
+        if self._value is None:
+            return
+        payload = self.to_knx(self._value)
+        await self._send(payload, response=True)
 
     async def read_state(self, wait_for_result: bool = False) -> None:
         """Send GroupValueRead telegram for state address to KNX bus."""
diff --git a/xknx/remote_value/remote_value_1count.py b/xknx/remote_value/remote_value_1count.py
index 8e7abdac..9bd665bd 100644
--- a/xknx/remote_value/remote_value_1count.py
+++ b/xknx/remote_value/remote_value_1count.py
@@ -6,7 +6,7 @@ DPT 6.010.
 
 from __future__ import annotations
 
-from xknx.dpt import DPTArray, DPTBinary, DPTValue1Count
+from xknx.dpt import DPTValue1Count
 
 from .remote_value import RemoteValue
 
@@ -14,10 +14,4 @@ from .remote_value import RemoteValue
 class RemoteValue1Count(RemoteValue[int]):
     """Abstraction for remote value of KNX 6.010 (DPT_Value_1_Count)."""
 
-    def to_knx(self, value: int) -> DPTArray:
-        """Convert value to payload."""
-        return DPTValue1Count.to_knx(value)
-
-    def from_knx(self, payload: DPTArray | DPTBinary) -> int:
-        """Convert current payload to value."""
-        return DPTValue1Count.from_knx(payload)
+    dpt_class = DPTValue1Count
diff --git a/xknx/remote_value/remote_value_color_rgb.py b/xknx/remote_value/remote_value_color_rgb.py
index 645a2ac5..4f572cf8 100644
--- a/xknx/remote_value/remote_value_color_rgb.py
+++ b/xknx/remote_value/remote_value_color_rgb.py
@@ -6,45 +6,12 @@ DPT 232.600.
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
-
-from xknx.dpt import DPTArray, DPTBinary
 from xknx.dpt.dpt_color import DPTColorRGB, RGBColor
 
-from .remote_value import AsyncCallbackType, GroupAddressesType, RemoteValue
-
-if TYPE_CHECKING:
-    from xknx.xknx import XKNX
+from .remote_value import RemoteValue
 
 
 class RemoteValueColorRGB(RemoteValue[RGBColor]):
     """Abstraction for remote value of KNX DPT 232.600 (DPT_Color_RGB)."""
 
-    def __init__(
-        self,
-        xknx: XKNX,
-        group_address: GroupAddressesType = None,
-        group_address_state: GroupAddressesType = None,
-        sync_state: bool | int | float | str = True,
-        device_name: str | None = None,
-        feature_name: str = "Color RGB",
-        after_update_cb: AsyncCallbackType | None = None,
-    ):
-        """Initialize remote value of KNX DPT 232.600 (DPT_Color_RGB)."""
-        super().__init__(
-            xknx,
-            group_address,
-            group_address_state,
-            sync_state=sync_state,
-            device_name=device_name,
-            feature_name=feature_name,
-            after_update_cb=after_update_cb,
-        )
-
-    def to_knx(self, value: RGBColor) -> DPTArray:
-        """Convert value to payload."""
-        return DPTColorRGB.to_knx(value)
-
-    def from_knx(self, payload: DPTArray | DPTBinary) -> RGBColor:
-        """Convert current payload to value."""
-        return DPTColorRGB.from_knx(payload)
+    dpt_class = DPTColorRGB
diff --git a/xknx/remote_value/remote_value_color_xyy.py b/xknx/remote_value/remote_value_color_xyy.py
index 16d20879..0a92e670 100644
--- a/xknx/remote_value/remote_value_color_xyy.py
+++ b/xknx/remote_value/remote_value_color_xyy.py
@@ -6,45 +6,12 @@ DPT 242.600.
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
-
-from xknx.dpt import DPTArray, DPTBinary
 from xknx.dpt.dpt_color import DPTColorXYY, XYYColor
 
-from .remote_value import AsyncCallbackType, GroupAddressesType, RemoteValue
-
-if TYPE_CHECKING:
-    from xknx.xknx import XKNX
+from .remote_value import RemoteValue
 
 
 class RemoteValueColorXYY(RemoteValue[XYYColor]):
     """Abstraction for remote value of KNX DPT 242.600 (DPT_Colour_xyY)."""
 
-    def __init__(
-        self,
-        xknx: XKNX,
-        group_address: GroupAddressesType = None,
-        group_address_state: GroupAddressesType = None,
-        sync_state: bool | int | float | str = True,
-        device_name: str | None = None,
-        feature_name: str = "Color xyY",
-        after_update_cb: AsyncCallbackType | None = None,
-    ):
-        """Initialize remote value of KNX DPT 242.600 (DPT_Colour_xyY)."""
-        super().__init__(
-            xknx,
-            group_address,
-            group_address_state,
-            sync_state=sync_state,
-            device_name=device_name,
-            feature_name=feature_name,
-            after_update_cb=after_update_cb,
-        )
-
-    def to_knx(self, value: XYYColor) -> DPTArray:
-        """Convert value to payload."""
-        return DPTColorXYY.to_knx(value)
-
-    def from_knx(self, payload: DPTArray | DPTBinary) -> XYYColor:
-        """Convert current payload to value."""
-        return DPTColorXYY.from_knx(payload)
+    dpt_class = DPTColorXYY
diff --git a/xknx/remote_value/remote_value_control.py b/xknx/remote_value/remote_value_control.py
index 2ace7a74..a5058a56 100644
--- a/xknx/remote_value/remote_value_control.py
+++ b/xknx/remote_value/remote_value_control.py
@@ -9,7 +9,7 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING, Any
 
-from xknx.dpt import DPTArray, DPTBinary, DPTControlStepCode
+from xknx.dpt import DPTControlStepCode
 from xknx.exceptions import ConversionError
 
 from .remote_value import AsyncCallbackType, GroupAddressesType, RemoteValue
@@ -51,14 +51,6 @@ class RemoteValueControl(RemoteValue[Any]):
             after_update_cb=after_update_cb,
         )
 
-    def to_knx(self, value: Any) -> DPTBinary:
-        """Convert value to payload."""
-        return self.dpt_class.to_knx(value)
-
-    def from_knx(self, payload: DPTArray | DPTBinary) -> Any:
-        """Convert current payload to value."""
-        return self.dpt_class.from_knx(payload)
-
     @property
     def unit_of_measurement(self) -> str | None:
         """Return the unit of measurement."""
diff --git a/xknx/remote_value/remote_value_datetime.py b/xknx/remote_value/remote_value_datetime.py
index 1f971da3..51d77ea3 100644
--- a/xknx/remote_value/remote_value_datetime.py
+++ b/xknx/remote_value/remote_value_datetime.py
@@ -10,7 +10,7 @@ from enum import Enum
 import time
 from typing import TYPE_CHECKING
 
-from xknx.dpt import DPTArray, DPTBinary, DPTDate, DPTDateTime, DPTTime
+from xknx.dpt import DPTDate, DPTDateTime, DPTTime
 from xknx.exceptions import ConversionError
 
 from .remote_value import AsyncCallbackType, GroupAddressesType, RemoteValue
@@ -62,11 +62,3 @@ class RemoteValueDateTime(RemoteValue[time.struct_time]):
             feature_name=feature_name,
             after_update_cb=after_update_cb,
         )
-
-    def to_knx(self, value: time.struct_time) -> DPTArray:
-        """Convert value to payload."""
-        return self.dpt_class.to_knx(value)
-
-    def from_knx(self, payload: DPTArray | DPTBinary) -> time.struct_time:
-        """Convert current payload to value."""
-        return self.dpt_class.from_knx(payload)
diff --git a/xknx/remote_value/remote_value_dpt_value_1_ucount.py b/xknx/remote_value/remote_value_dpt_value_1_ucount.py
index 795bd375..c3235d5d 100644
--- a/xknx/remote_value/remote_value_dpt_value_1_ucount.py
+++ b/xknx/remote_value/remote_value_dpt_value_1_ucount.py
@@ -6,7 +6,7 @@ DPT 5.010.
 
 from __future__ import annotations
 
-from xknx.dpt import DPTArray, DPTBinary, DPTValue1Ucount
+from xknx.dpt import DPTValue1Ucount
 
 from .remote_value import RemoteValue
 
@@ -14,10 +14,4 @@ from .remote_value import RemoteValue
 class RemoteValueDptValue1Ucount(RemoteValue[int]):
     """Abstraction for remote value of KNX DPT 5.010."""
 
-    def to_knx(self, value: int) -> DPTArray:
-        """Convert value to payload."""
-        return DPTValue1Ucount.to_knx(value)
-
-    def from_knx(self, payload: DPTArray | DPTBinary) -> int:
-        """Convert current payload to value."""
-        return DPTValue1Ucount.from_knx(payload)
+    dpt_class = DPTValue1Ucount
diff --git a/xknx/remote_value/remote_value_scene_number.py b/xknx/remote_value/remote_value_scene_number.py
index 37b55fdd..b4ed1c26 100644
--- a/xknx/remote_value/remote_value_scene_number.py
+++ b/xknx/remote_value/remote_value_scene_number.py
@@ -6,7 +6,7 @@ DPT 17.001.
 
 from __future__ import annotations
 
-from xknx.dpt import DPTArray, DPTBinary, DPTSceneNumber
+from xknx.dpt import DPTSceneNumber
 
 from .remote_value import RemoteValue
 
@@ -14,10 +14,4 @@ from .remote_value import RemoteValue
 class RemoteValueSceneNumber(RemoteValue[int]):
     """Abstraction for remote value of KNX DPT 17.001 (DPT_Scene_Number)."""
 
-    def to_knx(self, value: int) -> DPTArray:
-        """Convert value to payload."""
-        return DPTSceneNumber.to_knx(value)
-
-    def from_knx(self, payload: DPTArray | DPTBinary) -> int:
-        """Convert current payload to value."""
-        return DPTSceneNumber.from_knx(payload)
+    dpt_class = DPTSceneNumber
diff --git a/xknx/remote_value/remote_value_sensor.py b/xknx/remote_value/remote_value_sensor.py
index 417a5131..e1ad1931 100644
--- a/xknx/remote_value/remote_value_sensor.py
+++ b/xknx/remote_value/remote_value_sensor.py
@@ -7,10 +7,9 @@ for serialization and deserialization of the KNX value.
 
 from __future__ import annotations
 
-from abc import abstractmethod
 from typing import TYPE_CHECKING, TypeVar
 
-from xknx.dpt import DPTArray, DPTBase, DPTBinary, DPTNumeric, DPTString
+from xknx.dpt import DPTBase, DPTNumeric, DPTString
 from xknx.exceptions import ConversionError
 
 from .remote_value import AsyncCallbackType, GroupAddressesType, RemoteValue
@@ -26,6 +25,7 @@ class _RemoteValueGeneric(RemoteValue[ValueT]):
 
     dpt_base_class: type[DPTBase]
     _default_dpt_class: type[DPTBase] | None = None
+    dpt_class: type[DPTBase]
 
     def __init__(
         self,
@@ -61,14 +61,6 @@ class _RemoteValueGeneric(RemoteValue[ValueT]):
             after_update_cb=after_update_cb,
         )
 
-    def to_knx(self, value: ValueT) -> DPTArray:
-        """Convert value to payload."""
-        return self.dpt_class.to_knx(value)  # type: ignore[return-value]
-
-    @abstractmethod
-    def from_knx(self, payload: DPTArray | DPTBinary) -> ValueT:
-        """Convert current payload to value."""
-
     @property
     def unit_of_measurement(self) -> str | None:
         """Return the unit of measurement."""
@@ -86,10 +78,6 @@ class RemoteValueSensor(_RemoteValueGeneric[int | float | str]):
     dpt_base_class = DPTBase
     dpt_class: type[DPTBase]
 
-    def from_knx(self, payload: DPTArray | DPTBinary) -> int | float | str:
-        """Convert current payload to value."""
-        return self.dpt_class.from_knx(payload)  # type: ignore[no-any-return]
-
 
 class RemoteValueNumeric(_RemoteValueGeneric[int | float]):
     """Abstraction for numeric DPT types."""
@@ -97,10 +85,6 @@ class RemoteValueNumeric(_RemoteValueGeneric[int | float]):
     dpt_base_class = DPTNumeric
     dpt_class: type[DPTNumeric]
 
-    def from_knx(self, payload: DPTArray | DPTBinary) -> int | float:
-        """Convert current payload to value."""
-        return self.dpt_class.from_knx(payload)
-
 
 class RemoteValueString(_RemoteValueGeneric[str]):
     """Abstraction for string DPT types."""
@@ -108,7 +92,3 @@ class RemoteValueString(_RemoteValueGeneric[str]):
     dpt_base_class = DPTString
     dpt_class: type[DPTString]
     _default_dpt_class = DPTString
-
-    def from_knx(self, payload: DPTArray | DPTBinary) -> str:
-        """Convert current payload to value."""
-        return self.dpt_class.from_knx(payload)
diff --git a/xknx/remote_value/remote_value_setpoint_shift.py b/xknx/remote_value/remote_value_setpoint_shift.py
index 063d7810..02a1f9f7 100644
--- a/xknx/remote_value/remote_value_setpoint_shift.py
+++ b/xknx/remote_value/remote_value_setpoint_shift.py
@@ -50,29 +50,29 @@ class RemoteValueSetpointShift(RemoteValue[float]):
             after_update_cb=after_update_cb,
         )
 
-        self.dpt_class: type[DPTValue1Count | DPTTemperature] | None = (
+        self._internal_dpt_class: type[DPTValue1Count | DPTTemperature] | None = (
             setpoint_shift_mode.value if setpoint_shift_mode is not None else None
         )
         self.setpoint_shift_step = setpoint_shift_step
 
     def to_knx(self, value: float) -> DPTArray:
         """Convert value to payload."""
-        if self.dpt_class is None:
+        if self._internal_dpt_class is None:
             raise ConversionError(
                 f"Setpoint shift DPT not initialized for {self.device_name}"
             )
-        if self.dpt_class == DPTValue1Count:
+        if self._internal_dpt_class == DPTValue1Count:
             converted_value = int(value / self.setpoint_shift_step)
             return DPTValue1Count.to_knx(converted_value)
         return DPTTemperature.to_knx(value)
 
     def from_knx(self, payload: DPTArray | DPTBinary) -> float:
         """Convert current payload to value."""
-        if self.dpt_class is None:
-            self.dpt_class = self._determine_dpt_class(payload)
+        if self._internal_dpt_class is None:
+            self._internal_dpt_class = self._determine_dpt_class(payload)
 
-        payload_value = self.dpt_class.from_knx(payload)
-        if self.dpt_class == DPTValue1Count:
+        payload_value = self._internal_dpt_class.from_knx(payload)
+        if self._internal_dpt_class == DPTValue1Count:
             return payload_value * self.setpoint_shift_step
         return payload_value
 
diff --git a/xknx/remote_value/remote_value_temp.py b/xknx/remote_value/remote_value_temp.py
index f1436adf..7e5b2e16 100644
--- a/xknx/remote_value/remote_value_temp.py
+++ b/xknx/remote_value/remote_value_temp.py
@@ -6,7 +6,7 @@ DPT 9.001.
 
 from __future__ import annotations
 
-from xknx.dpt import DPTArray, DPTBinary, DPTTemperature
+from xknx.dpt import DPTTemperature
 
 from .remote_value import RemoteValue
 
@@ -14,10 +14,4 @@ from .remote_value import RemoteValue
 class RemoteValueTemp(RemoteValue[float]):
     """Abstraction for remote value of KNX 9.001 (DPT_Value_Temp)."""
 
-    def to_knx(self, value: float) -> DPTArray:
-        """Convert value to payload."""
-        return DPTTemperature.to_knx(value)
-
-    def from_knx(self, payload: DPTArray | DPTBinary) -> float:
-        """Convert current payload to value."""
-        return DPTTemperature.from_knx(payload)
+    dpt_class = DPTTemperature
diff --git a/xknx/telegram/__init__.py b/xknx/telegram/__init__.py
index 68fb9c04..eb65fa4d 100644
--- a/xknx/telegram/__init__.py
+++ b/xknx/telegram/__init__.py
@@ -9,7 +9,7 @@ Module for handling KNX primitives.
 # flake8: noqa
 from .address import GroupAddress, GroupAddressType, IndividualAddress
 from .address_filter import AddressFilter
-from .telegram import Telegram, TelegramDirection
+from .telegram import Telegram, TelegramDirection, TelegramDecodedData
 
 __all__ = [
     "AddressFilter",
@@ -18,4 +18,5 @@ __all__ = [
     "IndividualAddress",
     "Telegram",
     "TelegramDirection",
+    "TelegramDecodedData",
 ]
diff --git a/xknx/telegram/telegram.py b/xknx/telegram/telegram.py
index b3ef27e4..e09631cc 100644
--- a/xknx/telegram/telegram.py
+++ b/xknx/telegram/telegram.py
@@ -20,6 +20,8 @@ from __future__ import annotations
 from dataclasses import dataclass, field
 from enum import Enum
 
+from xknx.dpt import DPTBase, DPTComplexData
+
 from .address import GroupAddress, IndividualAddress, InternalGroupAddress
 from .apci import APCI
 from .tpci import TPCI, TDataBroadcast, TDataGroup, TDataIndividual
@@ -32,6 +34,21 @@ class TelegramDirection(Enum):
     OUTGOING = "Outgoing"
 
 
+@dataclass(slots=True)
+class TelegramDecodedData:
+    """Context for a telegram."""
+
+    transcoder: type[DPTBase]
+    value: bool | int | float | str | DPTComplexData
+
+    def __str__(self) -> str:
+        """Return object as readable string."""
+        return (
+            f"{self.value}{' ' + self.transcoder.unit if self.transcoder.unit is not None else ''}"
+            f" ({self.transcoder.__name__})"
+        )
+
+
 @dataclass(slots=True)
 class Telegram:
     """Class for KNX telegrams."""
@@ -43,6 +60,7 @@ class Telegram:
         default_factory=lambda: IndividualAddress(0)
     )
     tpci: TPCI = None  # type: ignore[assignment]  # set in __post_init__
+    decoded_data: TelegramDecodedData | None = None
 
     def __post_init__(self) -> None:
         """Initialize Telegram class."""
@@ -57,13 +75,27 @@ class Telegram:
             else:  # InternalGroupAddress
                 self.tpci = TDataGroup()
 
+    def __eq__(self, other: object) -> bool:
+        """Equal operator. Omit decoded_data for comparison."""
+        return (
+            isinstance(other, Telegram)
+            and self.destination_address == other.destination_address
+            and self.direction == other.direction
+            and self.payload == other.payload
+            and self.source_address == other.source_address
+            and self.tpci == other.tpci
+        )
+
     def __str__(self) -> str:
         """Return object as readable string."""
         data = f'payload="{self.payload}"' if self.payload else f'tpci="{self.tpci}"'
+        decoded_data = (
+            f' data="{self.decoded_data}"' if self.decoded_data is not None else ""
+        )
         return (
             "<Telegram "
             f'direction="{self.direction.value}" '
             f'source_address="{self.source_address}" '
             f'destination_address="{self.destination_address}" '
-            f"{data} />"
+            f"{data}{decoded_data} />"
         )
diff --git a/xknx/xknx.py b/xknx/xknx.py
index b72acf1a..3993a9e7 100644
--- a/xknx/xknx.py
+++ b/xknx/xknx.py
@@ -14,6 +14,7 @@ from types import TracebackType
 from xknx.cemi import CEMIHandler
 from xknx.core import (
     ConnectionManager,
+    GroupAddressDPT,
     TaskRegistry,
     TelegramQueue,
     XknxConnectionState,
@@ -63,6 +64,7 @@ class XKNX:
         self.cemi_handler = CEMIHandler(self)
         self.state_updater = StateUpdater(self, default_tracker_option=state_updater)
         self.task_registry = TaskRegistry(self)
+        self.group_address_dpt = GroupAddressDPT()
 
         self.current_address = IndividualAddress(0)
         self.daemon_mode = daemon_mode


Test patch:
diff --git a/test/core_tests/group_address_dpt_test.py b/test/core_tests/group_address_dpt_test.py
new file mode 100644
index 00000000..36945df1
--- /dev/null
+++ b/test/core_tests/group_address_dpt_test.py
@@ -0,0 +1,127 @@
+"""Test for group address dpt."""
+
+from unittest.mock import patch
+
+from xknx.dpt import DPTArray, DPTHumidity, DPTScaling, DPTTemperature
+from xknx.telegram import GroupAddress, Telegram, TelegramDirection, apci
+
+
+async def test_group_address_dpt_in_telegram_queue(xknx_no_interface):
+    """Test group address dpt."""
+    telegrams = 0
+    test_telegram: Telegram = None
+
+    async def _telegram_callback(telegram: Telegram):
+        nonlocal telegrams
+        nonlocal test_telegram
+        telegrams += 1
+        test_telegram = telegram
+
+    xknx = xknx_no_interface
+    telegram = Telegram(
+        destination_address=GroupAddress("1/2/3"),
+        direction=TelegramDirection.INCOMING,
+        payload=apci.GroupValueWrite(DPTArray((0x7F,))),
+    )
+    read_telegram = Telegram(
+        destination_address=GroupAddress("1/2/3"),
+        direction=TelegramDirection.INCOMING,
+        payload=apci.GroupValueRead(),
+    )
+    xknx.telegram_queue.register_telegram_received_cb(_telegram_callback)
+    async with xknx:
+        await xknx.telegrams.put(telegram)
+        await xknx.telegrams.join()
+        assert telegrams == 1
+        assert test_telegram.decoded_data is None
+
+        xknx.group_address_dpt.set({GroupAddress("1/2/3"): {"main": 5, "sub": 1}})
+        await xknx.telegrams.put(telegram)
+        await xknx.telegrams.join()
+        assert telegrams == 2
+        assert test_telegram.decoded_data is not None
+        assert test_telegram.decoded_data.transcoder is DPTScaling
+        assert test_telegram.decoded_data.value == 50
+        # do nothing for read-telegrams
+        await xknx.telegrams.put(read_telegram)
+        await xknx.telegrams.join()
+        assert telegrams == 3
+        assert test_telegram.decoded_data is None
+
+
+@patch("logging.Logger.warning")
+async def test_get_invalid_payload(logger_warning_mock, xknx_no_interface):
+    """Test processing when DPT doesn't fit payload."""
+
+    test_telegram: Telegram = None
+
+    async def _telegram_callback(telegram: Telegram):
+        nonlocal test_telegram
+        test_telegram = telegram
+
+    xknx = xknx_no_interface
+    xknx.group_address_dpt.set({GroupAddress("1/2/3"): {"main": 5}})
+    xknx.telegram_queue.register_telegram_received_cb(_telegram_callback)
+
+    telegram = Telegram(
+        destination_address=GroupAddress("1/2/3"),
+        direction=TelegramDirection.INCOMING,
+        payload=apci.GroupValueWrite(DPTArray((0x01, 0x02))),  # wrong payload size
+    )
+    async with xknx:
+        await xknx.telegrams.put(telegram)
+        await xknx.telegrams.join()
+        assert test_telegram.decoded_data is None
+        assert logger_warning_mock.call_count == 1
+        assert "DPT decoding error" in logger_warning_mock.call_args[0][0]
+
+
+def test_set(xknx_no_interface):
+    """Test set."""
+    xknx = xknx_no_interface
+    xknx.group_address_dpt.set(
+        {
+            GroupAddress("1/2/3"): {"main": 5, "sub": 1},
+            1: "temperature",
+            "i-internal": "9.007",
+        }
+    )
+    assert xknx.group_address_dpt._ga_dpts == {
+        2563: DPTScaling,
+        1: DPTTemperature,
+        "i-internal": DPTHumidity,
+    }
+    assert xknx.group_address_dpt.get(GroupAddress("0/0/1")) is DPTTemperature
+
+
+@patch("logging.Logger.warning")
+@patch("logging.Logger.debug")
+def test_set_invalid(logger_debug_mock, logger_warning_mock, xknx_no_interface):
+    """Test set invalid dpts."""
+    xknx = xknx_no_interface
+    xknx.group_address_dpt.set(
+        {
+            GroupAddress("0/0/1"): {"main": None},
+            2: "invalid",
+            0: "temperature",  # 0 is not a valid GA
+        }
+    )
+    assert logger_warning_mock.call_count == 1
+    assert "Invalid group address" in logger_warning_mock.call_args[0][0]
+    assert logger_debug_mock.call_count == 1
+    assert "No transcoder found for DPTs" in logger_debug_mock.call_args[0][0]
+    assert not xknx.group_address_dpt._ga_dpts
+
+
+def test_clear(xknx_no_interface):
+    """Test clear."""
+    xknx = xknx_no_interface
+    xknx.group_address_dpt.set(
+        {
+            GroupAddress("1/2/3"): {"main": 5, "sub": 1},
+            1: "temperature",
+        }
+    )
+    assert len(xknx.group_address_dpt._ga_dpts) == 2
+    xknx.group_address_dpt.clear()
+    assert len(xknx.group_address_dpt._ga_dpts) == 0
diff --git a/test/core_tests/telegram_queue_test.py b/test/core_tests/telegram_queue_test.py
index b4826478..20f12c85 100644
--- a/test/core_tests/telegram_queue_test.py
+++ b/test/core_tests/telegram_queue_test.py
@@ -6,7 +6,7 @@ from unittest.mock import AsyncMock, MagicMock, call, patch
 import pytest
 
 from xknx import XKNX
-from xknx.dpt import DPTBinary
+from xknx.dpt import DPTArray, DPTBinary
 from xknx.exceptions import CommunicationError, CouldNotParseTelegram
 from xknx.telegram import AddressFilter, Telegram, TelegramDirection
 from xknx.telegram.address import GroupAddress, InternalGroupAddress
@@ -211,6 +211,34 @@ class TestTelegramQueue:
         async_telegram_received_callback.assert_called_once_with(telegram)
         devices_process.assert_called_once_with(telegram)
 
+    async def test_callback_decoded_telegram_data(self):
+        """Test telegram_received_callback having decoded telegram data."""
+
+        xknx = XKNX()
+        xknx.group_address_dpt.set({"1/2/3": {"main": 5, "sub": 1}})
+        async_telegram_received_cb = AsyncMock()
+        xknx.telegram_queue.register_telegram_received_cb(async_telegram_received_cb)
+
+        telegram = Telegram(
+            destination_address=GroupAddress("1/2/3"),
+            direction=TelegramDirection.INCOMING,
+            payload=GroupValueWrite(
+                DPTArray(
+                    0x7F,
+                )
+            ),
+        )
+        await xknx.telegram_queue.start()
+        xknx.telegrams.put_nowait(telegram)
+        await xknx.telegrams.join()
+        await xknx.telegram_queue.stop()
+
+        assert async_telegram_received_cb.call_count == 1
+        received = async_telegram_received_cb.call_args_list[0][0][0]
+        assert received == telegram
+        assert received.decoded_data is not None
+        assert received.decoded_data.value == 50
+
     async def test_outgoing(self):
         """Test outgoing telegrams in telegram queue."""
         xknx = XKNX()
diff --git a/test/remote_value_tests/remote_value_setepoint_shift_test.py b/test/remote_value_tests/remote_value_setepoint_shift_test.py
index c6b5f44f..d89e60ce 100644
--- a/test/remote_value_tests/remote_value_setepoint_shift_test.py
+++ b/test/remote_value_tests/remote_value_setepoint_shift_test.py
@@ -27,18 +27,18 @@ class TestRemoteValueSetpointShift:
             remote_value.from_knx(DPTArray((0, 1, 2)))
 
         # DPT 6 - payload_length 1
-        assert remote_value.dpt_class is None
+        assert remote_value._internal_dpt_class is None
         # default setpoint_shift_step = 0.1
         assert remote_value.from_knx(dpt_6_payload) == 0.1
-        assert remote_value.dpt_class == SetpointShiftMode.DPT6010.value
+        assert remote_value._internal_dpt_class == SetpointShiftMode.DPT6010.value
         with pytest.raises(CouldNotParseTelegram):
             # DPT 9 is invalid now
             remote_value.from_knx(dpt_9_payload)
 
-        remote_value.dpt_class = None
+        remote_value._internal_dpt_class = None
         # DPT 9 - payload_length 2
         assert remote_value.from_knx(dpt_9_payload) == 1
-        assert remote_value.dpt_class == SetpointShiftMode.DPT9002.value
+        assert remote_value._internal_dpt_class == SetpointShiftMode.DPT9002.value
         with pytest.raises(CouldNotParseTelegram):
             # DPT 6 is invalid now
             remote_value.from_knx(dpt_6_payload)
@@ -55,7 +55,7 @@ class TestRemoteValueSetpointShift:
         dpt_6_payload = DPTValue1Count.to_knx(1)
         dpt_9_payload = DPTTemperature.to_knx(1)
 
-        assert remote_value_6.dpt_class == DPTValue1Count
+        assert remote_value_6._internal_dpt_class == DPTValue1Count
         with pytest.raises(CouldNotParseTelegram):
             remote_value_6.from_knx(None)
         with pytest.raises(CouldNotParseTelegram):
@@ -66,7 +66,7 @@ class TestRemoteValueSetpointShift:
             remote_value_6.from_knx(DPTBinary(1))
         assert remote_value_6.from_knx(dpt_6_payload) == 0.1
 
-        assert remote_value_9.dpt_class == DPTTemperature
+        assert remote_value_9._internal_dpt_class == DPTTemperature
         with pytest.raises(CouldNotParseTelegram):
             remote_value_9.from_knx(None)
         with pytest.raises(CouldNotParseTelegram):
@@ -82,7 +82,7 @@ class TestRemoteValueSetpointShift:
         xknx = XKNX()
         remote_value = RemoteValueSetpointShift(xknx=xknx)
 
-        assert remote_value.dpt_class is None
+        assert remote_value._internal_dpt_class is None
         with pytest.raises(ConversionError):
             remote_value.to_knx(1)
 
diff --git a/test/remote_value_tests/remote_value_test.py b/test/remote_value_tests/remote_value_test.py
index cc0f1fa2..04713966 100644
--- a/test/remote_value_tests/remote_value_test.py
+++ b/test/remote_value_tests/remote_value_test.py
@@ -8,7 +8,7 @@ from xknx import XKNX
 from xknx.dpt import DPT2ByteFloat, DPTArray, DPTBinary
 from xknx.exceptions import ConversionError, CouldNotParseTelegram
 from xknx.remote_value import RemoteValue, RemoteValueSwitch
-from xknx.telegram import GroupAddress, Telegram
+from xknx.telegram import GroupAddress, Telegram, TelegramDecodedData
 from xknx.telegram.apci import GroupValueWrite
 
 
@@ -239,6 +239,8 @@ class TestRemoteValue:
         """Test if passive group addresses are processed."""
         xknx = XKNX()
         remote_value = RemoteValue(xknx, group_address=["1/2/3", "1/1/1"])
+        remote_value.dpt_class = DPT2ByteFloat
+
         assert remote_value.writable
         assert not remote_value.readable
         # RemoteValue is initialized with only passive group address
@@ -252,6 +254,41 @@ class TestRemoteValue:
         assert await remote_value.process(telegram)
         assert remote_value.telegram.payload.value == test_payload
 
+    async def test_to_from_knx(self):
+        """Test to_knx and from_knx raises when not set properly."""
+        xknx = XKNX()
+        remote_value = RemoteValue(xknx, group_address="1/1/1")
+        with pytest.raises(NotImplementedError):
+            remote_value.value = 3.3  # without to_knx method
+
+        test_payload = DPTArray((0x01, 0x02))
+        telegram = Telegram(
+            destination_address=GroupAddress("1/1/1"),
+            payload=GroupValueWrite(test_payload),
+        )
+        with pytest.raises(NotImplementedError):
+            await remote_value.process(telegram)
+
+        # doesn't raise with `dpt_class` set
+        remote_value.dpt_class = DPT2ByteFloat
+        remote_value.value = 3.3
+        await remote_value.process(telegram)
+
+    async def test_pre_decoded_telegram(self):
+        """Test if pre-decoded Telegram is processed."""
+        xknx = XKNX()
+        remote_value = RemoteValue(xknx, group_address="1/1/1")
+        remote_value.dpt_class = DPT2ByteFloat
+
+        test_payload = "invalid for testing"
+        telegram = Telegram(
+            destination_address=GroupAddress("1/1/1"),
+            payload=GroupValueWrite(test_payload),
+            decoded_data=TelegramDecodedData(transcoder=DPT2ByteFloat, value=3.3),
+        )
+        assert await remote_value.process(telegram)
+        assert remote_value.value == 3.3
+
     def test_eq(self):
         """Test __eq__ operator."""
         xknx = XKNX()
diff --git a/test/str_test.py b/test/str_test.py
index d54df72d..b8e70b35 100644
--- a/test/str_test.py
+++ b/test/str_test.py
@@ -89,6 +89,7 @@ class TestStringRepresentations:
             == '<RemoteValue device_name="MyDevice" feature_name="Unknown" <1/2/3, 1/2/4, [], None /> />'
         )
 
+        remote_value.to_knx = lambda value: value  # to bypass NotImplementedError
         remote_value.value = 34
         assert (
             str(remote_value)