Add DPTBase.get_dpt DPT definition helper classmethod

test/dpt_tests/dpt_test.py

@@ -6,6 +6,8 @@
 
 from xknx.dpt import (
     DPT2ByteFloat,
+    DPT2ByteUnsigned,
+    DPTActiveEnergy,
     DPTArray,
     DPTBase,
     DPTBinary,
@@ -24,12 +26,12 @@
 class TestDPTBase:
     """Test class for transcoder base object."""
 
-    def test_dpt_abstract_subclasses_ignored(self):
+    def test_dpt_abstract_subclasses_ignored(self) -> None:
         """Test if abstract base classes are ignored by dpt_class_tree and __recursive_subclasses__."""
         for dpt in DPTBase.dpt_class_tree():
             assert dpt not in (DPTBase, DPTNumeric, DPTEnum, DPTComplex)
 
-    def test_dpt_concrete_subclasses_included(self):
+    def test_dpt_concrete_subclasses_included(self) -> None:
         """Test if concrete subclasses are included by dpt_class_tree."""
         for dpt in (
             DPT2ByteFloat,
@@ -42,27 +44,29 @@ def test_dpt_concrete_subclasses_included(self):
             assert dpt in DPTBase.dpt_class_tree()
 
     @pytest.mark.parametrize("dpt_class", [DPTString, DPT2ByteFloat])
-    def test_dpt_non_abstract_baseclass_included(self, dpt_class):
+    def test_dpt_non_abstract_baseclass_included(
+        self, dpt_class: type[DPTBase]
+    ) -> None:
         """Test if non-abstract base classes is included by dpt_class_tree."""
         assert dpt_class in dpt_class.dpt_class_tree()
 
-    def test_dpt_subclasses_definition_types(self):
+    def test_dpt_subclasses_definition_types(self) -> None:
         """Test value_type and dpt_*_number values for correct type in subclasses of DPTBase."""
         for dpt in DPTBase.dpt_class_tree():
             if dpt.value_type is not None:
-                assert isinstance(
-                    dpt.value_type, str
-                ), f"Wrong type for value_type in {dpt} : {type(dpt.value_type)} - str `None` expected"
+                assert isinstance(dpt.value_type, str), (
+                    f"Wrong type for value_type in {dpt} : {type(dpt.value_type)} - str `None` expected"
+                )
             if dpt.dpt_main_number is not None:
-                assert isinstance(
-                    dpt.dpt_main_number, int
-                ), f"Wrong type for dpt_main_number in {dpt} : {type(dpt.dpt_main_number)} - int or `None` expected"
+                assert isinstance(dpt.dpt_main_number, int), (
+                    f"Wrong type for dpt_main_number in {dpt} : {type(dpt.dpt_main_number)} - int or `None` expected"
+                )
             if dpt.dpt_sub_number is not None:
-                assert isinstance(
-                    dpt.dpt_sub_number, int
-                ), f"Wrong type for dpt_sub_number in {dpt} : {type(dpt.dpt_sub_number)} - int or `None` expected"
+                assert isinstance(dpt.dpt_sub_number, int), (
+                    f"Wrong type for dpt_sub_number in {dpt} : {type(dpt.dpt_sub_number)} - int or `None` expected"
+                )
 
-    def test_dpt_subclasses_no_duplicate_value_types(self):
+    def test_dpt_subclasses_no_duplicate_value_types(self) -> None:
         """Test for duplicate value_type values in subclasses of DPTBase."""
         value_types = [
             dpt.value_type
@@ -71,7 +75,7 @@ def test_dpt_subclasses_no_duplicate_value_types(self):
         ]
         assert len(value_types) == len(set(value_types))
 
-    def test_dpt_subclasses_no_duplicate_dpt_number(self):
+    def test_dpt_subclasses_no_duplicate_dpt_number(self) -> None:
         """Test for duplicate value_type values in subclasses of DPTBase."""
         dpt_tuples = [
             (dpt.dpt_main_number, dpt.dpt_sub_number)
@@ -89,13 +93,66 @@ def test_dpt_subclasses_no_duplicate_dpt_number(self):
             ["active_energy", "13.010", {"main": 13, "sub": 10}],
         ],
     )
-    def test_dpt_alternative_notations(self, equal_dpts: list[Any]):
+    def test_dpt_alternative_notations(self, equal_dpts: list[Any]) -> None:
         """Test the parser for accepting alternative notations for the same DPT class."""
         parsed = [DPTBase.parse_transcoder(dpt) for dpt in equal_dpts]
         assert issubclass(parsed[0], DPTBase)
         assert all(parsed[0] == dpt for dpt in parsed)
 
-    def test_parse_transcoder_from_subclass(self):
+    @pytest.mark.parametrize(
+        "equal_dpts",
+        [
+            # strings in dictionaries would fail type checking, but should work nevertheless
+            [
+                "2byte_unsigned",
+                7,
+                "DPT-7",
+                {"main": 7},
+                {"main": "7", "sub": None},
+                DPT2ByteUnsigned,
+            ],
+            [
+                "temperature",
+                "9.001",
+                {"main": 9, "sub": 1},
+                {"main": "9", "sub": "1"},
+                DPTTemperature,
+            ],
+            ["active_energy", "13.010", {"main": 13, "sub": 10}, DPTActiveEnergy],
+        ],
+    )
+    def test_get_dpt_alternative_notations(self, equal_dpts: list[Any]) -> None:
+        """Test the parser for accepting alternative notations for the same DPT class."""
+        parsed = [DPTBase.get_dpt(dpt) for dpt in equal_dpts]
+        assert issubclass(parsed[0], DPTBase)
+        assert all(parsed[0] == dpt for dpt in parsed)
+
+    INVALID_DPT_IDENTIFIERS = [
+        None,
+        0,
+        999999999,
+        9.001,  # float is not valid
+        "invalid_string",
+        {"sub": 1},
+        {"main": None, "sub": None},
+        {"main": "invalid"},
+        {"main": 9, "sub": "invalid"},
+        [9, 1],
+        (9,),
+    ]
+
+    @pytest.mark.parametrize("value", INVALID_DPT_IDENTIFIERS)
+    def test_parse_transcoder_invalid_data(self, value: Any) -> None:
+        """Test parsing invalid data."""
+        assert DPTBase.parse_transcoder(value) is None
+
+    @pytest.mark.parametrize("value", INVALID_DPT_IDENTIFIERS)
+    def test_get_dpt_invalid_data(self, value: Any) -> None:
+        """Test parsing invalid data."""
+        with pytest.raises(ValueError):
+            DPTBase.get_dpt(value)
+
+    def test_parse_transcoder_from_subclass(self) -> None:
         """Test parsing only subclasses of a DPT class."""
         assert DPTBase.parse_transcoder("string") == DPTString
         assert DPTNumeric.parse_transcoder("string") is None
@@ -109,37 +166,30 @@ def test_parse_transcoder_from_subclass(self):
         assert DPTNumeric.parse_transcoder("temperature") == DPTTemperature
         assert DPT2ByteFloat.parse_transcoder("temperature") == DPTTemperature
 
-    @pytest.mark.parametrize(
-        "value",
-        [
-            None,
-            0,
-            999999999,
-            9.001,  # float is not valid
-            "invalid_string",
-            {"sub": 1},
-            {"main": None, "sub": None},
-            {"main": "invalid"},
-            {"main": 9, "sub": "invalid"},
-            [9, 1],
-            (9,),
-        ],
-    )
-    def test_parse_transcoder_invalid_data(self, value: Any):
-        """Test parsing invalid data."""
-        assert DPTBase.parse_transcoder(value) is None
+    def test_get_dpt_from_subclass(self) -> None:
+        """Test parsing only subclasses of a DPT class."""
+        assert DPTBase.get_dpt("string") == DPTString
+        with pytest.raises(ValueError):
+            DPTNumeric.get_dpt("string")
+
+        assert DPTBase.get_dpt("percent") == DPTScaling
+        assert DPTNumeric.get_dpt("percent") == DPTScaling
+
+        assert DPTBase.get_dpt("temperature") == DPTTemperature
+        assert DPTNumeric.get_dpt("temperature") == DPTTemperature
+        assert DPT2ByteFloat.get_dpt("temperature") == DPTTemperature
 
 
 class TestDPTBaseSubclass:
     """Test subclass of transcoder base object."""
 
     @pytest.mark.parametrize("dpt_class", DPTBase.dpt_class_tree())
-    def test_required_values(self, dpt_class):
+    def test_required_values(self, dpt_class: type[DPTBase]) -> None:
         """Test required class variables are set for definitions."""
         assert dpt_class.payload_type in (DPTArray, DPTBinary)
         assert dpt_class.payload_length is not None
 
-    def test_validate_payload_array(self):
+    def test_validate_payload_array(self) -> None:
         """Test validate_payload method."""
 
         class DPTArrayTest(DPTBase):
@@ -157,7 +207,7 @@ class DPTArrayTest(DPTBase):
 
         assert DPTArrayTest.validate_payload(DPTArray((1, 1))) == (1, 1)
 
-    def test_validate_payload_binary(self):
+    def test_validate_payload_binary(self) -> None:
         """Test validate_payload method."""
 
         class DPTBinaryTest(DPTBase):
@@ -178,7 +228,7 @@ class TestDPTNumeric:
     """Test class for numeric transcoder base object."""
 
     @pytest.mark.parametrize("dpt_class", DPTNumeric.dpt_class_tree())
-    def test_values(self, dpt_class):
+    def test_values(self, dpt_class: type[DPTNumeric]) -> None:
         """Test boundary values are set for numeric definitions (because mypy doesn't)."""
 
         assert isinstance(dpt_class.value_min, int | float)

xknx/core/group_address_dpt.py

@@ -5,7 +5,7 @@
 from collections.abc import Mapping
 import logging
 
-from xknx.dpt.dpt import DPTBase, _DPTMainSubDict
+from xknx.dpt.dpt import DPTBase
 from xknx.exceptions import ConversionError, CouldNotParseAddress, CouldNotParseTelegram
 from xknx.telegram import Telegram, TelegramDecodedData
 from xknx.telegram.address import (
@@ -16,6 +16,7 @@
     parse_device_group_address,
 )
 from xknx.telegram.apci import GroupValueResponse, GroupValueWrite
+from xknx.typing import DPTParsable
 
 _GA_DPT_LOGGER = logging.getLogger("xknx.ga_dpt")
 
@@ -32,7 +33,7 @@ def __init__(self) -> None:
 
     def set(
         self,
-        ga_dpt: Mapping[DeviceAddressableType, int | str | _DPTMainSubDict],
+        ga_dpt: Mapping[DeviceAddressableType, DPTParsable],
     ) -> None:
         """Assign decoders to group addresses."""
         unknown_dpts = set()

xknx/devices/expose_sensor.py

@@ -18,12 +18,14 @@
 from typing import TYPE_CHECKING, Any
 
 from xknx.core import Task
+from xknx.dpt import DPTBase
 from xknx.remote_value import (
     GroupAddressesType,
     RemoteValue,
     RemoteValueSensor,
     RemoteValueSwitch,
 )
+from xknx.typing import DPTParsable
 
 from .device import Device, DeviceCallbackType
 
@@ -41,7 +43,7 @@ def __init__(
         name: str,
         group_address: GroupAddressesType = None,
         respond_to_read: bool = True,
-        value_type: int | str | None = None,
+        value_type: DPTParsable | type[DPTBase] | None = None,
         cooldown: float = 0,
         device_updated_cb: DeviceCallbackType[ExposeSensor] | None = None,
     ):

xknx/devices/notification.py

@@ -5,7 +5,9 @@
 from collections.abc import Iterator
 from typing import TYPE_CHECKING
 
+from xknx.dpt import DPTBase
 from xknx.remote_value import GroupAddressesType, RemoteValueString
+from xknx.typing import DPTParsable
 
 from .device import Device, DeviceCallbackType
 
@@ -25,7 +27,7 @@ def __init__(
         group_address_state: GroupAddressesType = None,
         respond_to_read: bool = False,
         sync_state: bool | int | float | str = True,
-        value_type: int | str | None = None,
+        value_type: DPTParsable | type[DPTBase] | None = None,
         device_updated_cb: DeviceCallbackType[Notification] | None = None,
     ):
         """Initialize notification class."""

xknx/devices/numeric_value.py

@@ -13,7 +13,9 @@
 from collections.abc import Iterator
 from typing import TYPE_CHECKING
 
+from xknx.dpt import DPTBase
 from xknx.remote_value import GroupAddressesType, RemoteValueNumeric
+from xknx.typing import DPTParsable
 
 from .device import Device, DeviceCallbackType
 
@@ -33,7 +35,7 @@ def __init__(
         group_address_state: GroupAddressesType = None,
         respond_to_read: bool = False,
         sync_state: bool | int | float | str = True,
-        value_type: int | str | None = None,
+        value_type: DPTParsable | type[DPTBase] | None = None,
         always_callback: bool = False,
         device_updated_cb: DeviceCallbackType[NumericValue] | None = None,
     ):

xknx/devices/sensor.py

@@ -12,11 +12,13 @@
 from collections.abc import Iterator
 from typing import TYPE_CHECKING, Any
 
+from xknx.dpt import DPTBase
 from xknx.remote_value import (
     GroupAddressesType,
     RemoteValue,
     RemoteValueSensor,
 )
+from xknx.typing import DPTParsable
 
 from .device import Device, DeviceCallbackType
 
@@ -35,7 +37,7 @@ def __init__(
         group_address_state: GroupAddressesType = None,
         sync_state: bool | int | float | str = True,
         always_callback: bool = False,
-        value_type: int | str | None = None,
+        value_type: DPTParsable | type[DPTBase] | None = None,
         device_updated_cb: DeviceCallbackType[Sensor] | None = None,
     ):
         """Initialize Sensor class."""

xknx/dpt/dpt.py

@@ -8,21 +8,14 @@
 from enum import Enum
 from inspect import isabstract
 import struct
-from typing import Any, Generic, TypedDict, TypeVar, cast, final
+from typing import Any, Generic, TypeVar, cast, final
 
 from xknx.exceptions import ConversionError, CouldNotParseTelegram
-from xknx.typing import Self
+from xknx.typing import DPTParsable, Self
 
 from .payload import DPTArray, DPTBinary
 
 
-class _DPTMainSubDict(TypedDict):
-    """DPT type dictionary in accordance to xknxproject DPTType data."""
-
-    main: int
-    sub: int | None
-
-
 class DPTBase(ABC):
     """
     Base class for KNX data point type transcoder.
@@ -169,9 +162,7 @@ def transcoder_by_value_type(cls: type[Self], value_type: str) -> type[Self] | N
         return None
 
     @classmethod
-    def parse_transcoder(
-        cls: type[Self], value_type: int | str | _DPTMainSubDict
-    ) -> type[Self] | None:
+    def parse_transcoder(cls: type[Self], value_type: DPTParsable) -> type[Self] | None:
         """
         Return Class reference of DPTBase subclass from value_type or DPT number.
 
@@ -209,6 +200,23 @@ def parse_transcoder(
                 return None
             return cls.transcoder_by_dpt(dpt_main=main, dpt_sub=_sub)
 
+    @classmethod
+    def get_dpt(cls: type[Self], value_type: DPTParsable | type[DPTBase]) -> type[Self]:
+        """
+        Return DPT class from value.
+
+        Raises ValueError if value_type can't be parsed to DPT class.
+        """
+        if isinstance(value_type, type):
+            if issubclass(value_type, cls) and not isabstract(value_type):
+                return value_type
+        else:
+            if transcoder := cls.parse_transcoder(value_type):
+                return transcoder
+        raise ValueError(
+            f"Invalid value type for base class {cls.__name__}: {value_type}"
+        )
+
 
 class DPTNumeric(DPTBase):
     """Base class for KNX data point types decoding numeric values."""