typecheck RemoteValues

home-assistant-plugin/custom_components/xknx/sensor.py

@@ -30,7 +30,7 @@ def state(self):
         return self._device.resolve_state()
 
     @property
-    def unit_of_measurement(self):
+    def unit_of_measurement(self) -> str:
         """Return the unit this state is expressed in."""
         return self._device.unit_of_measurement()
 

setup.cfg

@@ -51,7 +51,7 @@ warn_unreachable = true
 implicit_reexport = true
 
 # partly typechecked modules (extra block for better overview)
-[mypy-xknx.remote_value.remote_value,xknx.remote_value.remote_value_climate_mode]
+[mypy-xknx.remote_value.*]
 strict = true
 ignore_errors = false
 warn_unreachable = true

test/devices_tests/light_test.py

@@ -665,7 +665,7 @@ def test_set_individual_color(self):
                 )
             )
         )
-        self.assertEqual(light.current_color, ([23, 24, 25], None))
+        self.assertEqual(light.current_color, ((23, 24, 25), None))
 
     def test_set_individual_color_not_possible(self):
         """Test setting the color of a non light without color."""
@@ -707,7 +707,7 @@ def test_set_color_rgbw(self):
             ),
         )
         self.loop.run_until_complete(xknx.devices.process(telegram))
-        self.assertEqual(light.current_color, ([23, 24, 25], 26))
+        self.assertEqual(light.current_color, ((23, 24, 25), 26))
 
     def test_set_color_rgbw_not_possible(self):
         """Test setting RGBW value of a non light without color."""
@@ -808,7 +808,7 @@ def test_set_individual_color_rgbw(self):
                 )
             )
         )
-        self.assertEqual(light.current_color, ([23, 24, 25], 26))
+        self.assertEqual(light.current_color, ((23, 24, 25), 26))
 
     def test_set_individual_color_rgbw_not_possible(self):
         """Test setting RGBW value of a non light without color."""
@@ -1111,7 +1111,7 @@ def test_process_individual_color(self):
 
         for telegram in telegrams:
             self.loop.run_until_complete(light.process(telegram))
-        self.assertEqual(light.current_color, ([42, 43, 44], None))
+        self.assertEqual(light.current_color, ((42, 43, 44), None))
 
     def test_process_color_rgbw(self):
         """Test process / reading telegrams from telegram queue. Test if RGBW is processed."""
@@ -1129,7 +1129,7 @@ def test_process_color_rgbw(self):
             payload=GroupValueWrite(DPTArray((23, 24, 25, 26, 0, 15))),
         )
         self.loop.run_until_complete(light.process(telegram))
-        self.assertEqual(light.current_color, ([23, 24, 25], 26))
+        self.assertEqual(light.current_color, ((23, 24, 25), 26))
 
     def test_process_individual_color_rgbw(self):
         """Test process / reading telegrams from telegram queue. Test if RGBW is processed."""

test/remote_value_tests/remote_value_color_rgbw_test.py

@@ -42,23 +42,23 @@ def test_from_knx(self):
         remote_value = RemoteValueColorRGBW(xknx)
         self.assertEqual(
             remote_value.from_knx(DPTArray((0x64, 0x65, 0x66, 0x7F, 0x00, 0x00))),
-            [0, 0, 0, 0],
+            (0, 0, 0, 0),
         )
         self.assertEqual(
             remote_value.from_knx(DPTArray((0x64, 0x65, 0x66, 0x7F, 0x00, 0x0F))),
-            [100, 101, 102, 127],
+            (100, 101, 102, 127),
         )
         self.assertEqual(
             remote_value.from_knx(DPTArray((0x64, 0x65, 0x66, 0x7F, 0x00, 0x00))),
-            [100, 101, 102, 127],
+            (100, 101, 102, 127),
         )
         self.assertEqual(
             remote_value.from_knx(DPTArray((0xFF, 0x65, 0x66, 0xFF, 0x00, 0x09))),
-            [255, 101, 102, 255],
+            (255, 101, 102, 255),
         )
         self.assertEqual(
             remote_value.from_knx(DPTArray((0x64, 0x65, 0x66, 0x7F, 0x00, 0x01))),
-            [255, 101, 102, 127],
+            (255, 101, 102, 127),
         )
 
     def test_to_knx_error(self):
@@ -118,7 +118,7 @@ def test_process(self):
             payload=GroupValueWrite(DPTArray((0x64, 0x65, 0x66, 0x67, 0x00, 0x0F))),
         )
         self.loop.run_until_complete(remote_value.process(telegram))
-        self.assertEqual(remote_value.value, [100, 101, 102, 103])
+        self.assertEqual(remote_value.value, (100, 101, 102, 103))
 
     def test_to_process_error(self):
         """Test process errornous telegram."""

xknx/devices/binary_sensor.py

@@ -32,7 +32,7 @@ def __init__(
         xknx: "XKNX",
         name: str,
         group_address_state: "GroupAddressableType" = None,
-        invert: Optional[bool] = False,
+        invert: bool = False,
         sync_state: bool = True,
         ignore_internal_state: bool = False,
         device_class: Optional[str] = None,

xknx/devices/light.py

@@ -534,7 +534,7 @@ async def set_brightness(self, brightness: int) -> None:
         await self.brightness.set(brightness)
 
     @property
-    def current_color(self) -> Tuple[Optional[List[int]], Optional[int]]:
+    def current_color(self) -> Tuple[Optional[Tuple[int, int, int]], Optional[int]]:
         """
         Return current color of light.
 
@@ -548,11 +548,11 @@ def current_color(self) -> Tuple[Optional[List[int]], Optional[int]]:
         if self.color.initialized:
             return self.color.value, None
         # individual RGB addresses - white will return None when it is not initialized
-        colors = [
+        colors = (
             self.red.brightness.value,
             self.green.brightness.value,
             self.blue.brightness.value,
-        ]
+        )
         if None in colors:
             return None, self.white.brightness.value
         return colors, self.white.brightness.value

xknx/devices/switch.py

@@ -31,7 +31,7 @@ def __init__(
         name: str,
         group_address: Optional["GroupAddressableType"] = None,
         group_address_state: Optional["GroupAddressableType"] = None,
-        invert: Optional[bool] = False,
+        invert: bool = False,
         reset_after: Optional[float] = None,
         device_updated_cb: Optional[DeviceCallbackType] = None,
     ):

xknx/dpt/dpt.py

@@ -48,6 +48,8 @@ class DPTBase(ABC):
     dpt_main_number: Optional[int] = None
     dpt_sub_number: Optional[int] = None
     value_type: Optional[str] = None
+    unit: Optional[str] = None
+    ha_device_class: Optional[str] = None
 
     @classmethod
     @abstractmethod
@@ -110,6 +112,7 @@ def transcoder_by_value_type(value_type: str) -> Optional[Type["DPTBase"]]:
                     return dpt
         return None
 
+    # TODO: convert to classmethod to allow parsing only subclasses (eg. for Numeric, Control etc.)
     @staticmethod
     def parse_transcoder(
         value_type: Union[int, float, str]

xknx/remote_value/remote_value.py

@@ -29,10 +29,10 @@
     from xknx.telegram.address import GroupAddressableType
     from xknx.xknx import XKNX
 
-    AsyncCallback = Callable[[], Awaitable[None]]
-
 logger = logging.getLogger("xknx.log")
 
+AsyncCallbackType = Callable[[], Awaitable[None]]
+
 
 class RemoteValue(ABC, Generic[DPTPayloadType]):
     """Class for managing remote knx value."""
@@ -46,7 +46,7 @@ def __init__(
         sync_state: bool = True,
         device_name: Optional[str] = None,
         feature_name: Optional[str] = None,
-        after_update_cb: Optional["AsyncCallback"] = None,
+        after_update_cb: Optional[AsyncCallbackType] = None,
         passive_group_addresses: Optional[List["GroupAddressableType"]] = None,
     ):
         """Initialize RemoteValue class."""
@@ -66,7 +66,7 @@ def __init__(
 
         self.device_name: str = "Unknown" if device_name is None else device_name
         self.feature_name: str = "Unknown" if feature_name is None else feature_name
-        self.after_update_cb: Optional["AsyncCallback"] = after_update_cb
+        self.after_update_cb: Optional[AsyncCallbackType] = after_update_cb
         self.payload: Optional[DPTPayloadType] = None
 
         if sync_state and self.group_address_state:

xknx/remote_value/remote_value_1count.py

@@ -3,7 +3,9 @@
 
 DPT 6.010.
 """
-from xknx.dpt import DPTArray, DPTValue1Count
+from typing import Optional, Union
+
+from xknx.dpt import DPTArray, DPTBinary, DPTValue1Count
 
 from .remote_value import RemoteValue
 
@@ -13,18 +15,20 @@ class RemoteValue1Count(RemoteValue[DPTArray]):
 
     # pylint: disable=no-self-use
 
-    def payload_valid(self, payload):
+    def payload_valid(
+        self, payload: Optional[Union[DPTArray, DPTBinary]]
+    ) -> Optional[DPTArray]:
         """Test if telegram payload may be parsed."""
         return (
             payload
             if isinstance(payload, DPTArray) and len(payload.value) == 1
             else None
         )
 
-    def to_knx(self, value):
+    def to_knx(self, value: int) -> DPTArray:
         """Convert value to payload."""
         return DPTArray(DPTValue1Count.to_knx(value))
 
-    def from_knx(self, payload):
+    def from_knx(self, payload: DPTArray) -> int:
         """Convert current payload to value."""
         return DPTValue1Count.from_knx(payload.value)

xknx/remote_value/remote_value_climate_mode.py

@@ -5,16 +5,7 @@
 """
 from abc import abstractmethod
 from enum import Enum
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    Awaitable,
-    Callable,
-    Generic,
-    List,
-    Optional,
-    Union,
-)
+from typing import TYPE_CHECKING, Any, Generic, List, Optional, Union
 
 from xknx.dpt import (
     DPTArray,
@@ -27,14 +18,12 @@
 from xknx.dpt.dpt_hvac_mode import HVACControllerMode, HVACModeType, HVACOperationMode
 from xknx.exceptions import ConversionError, CouldNotParseTelegram
 
-from .remote_value import RemoteValue
+from .remote_value import AsyncCallbackType, RemoteValue
 
 if TYPE_CHECKING:
     from xknx.telegram.address import GroupAddressableType
     from xknx.xknx import XKNX
 
-    AsyncCallback = Callable[[], Awaitable[None]]
-
 
 class RemoteValueClimateModeBase(
     RemoteValue[DPTPayloadType], Generic[DPTPayloadType, HVACModeType]
@@ -66,7 +55,7 @@ def __init__(
         device_name: Optional[str] = None,
         feature_name: str = "Climate Mode",
         climate_mode_type: Optional[ClimateModeType] = None,
-        after_update_cb: Optional["AsyncCallback"] = None,
+        after_update_cb: Optional[AsyncCallbackType] = None,
         passive_group_addresses: Optional[List["GroupAddressableType"]] = None,
     ):
         """Initialize remote value of KNX climate mode."""
@@ -131,7 +120,7 @@ def __init__(
         sync_state: bool = True,
         device_name: Optional[str] = None,
         feature_name: str = "Controller Mode",
-        after_update_cb: Optional["AsyncCallback"] = None,
+        after_update_cb: Optional[AsyncCallbackType] = None,
         passive_group_addresses: Optional[List["GroupAddressableType"]] = None,
     ):
         """Initialize remote value of KNX climate mode."""
@@ -186,7 +175,7 @@ def __init__(
         sync_state: bool = True,
         device_name: Optional[str] = None,
         feature_name: str = "Climate Mode Binary",
-        after_update_cb: Optional["AsyncCallback"] = None,
+        after_update_cb: Optional[AsyncCallbackType] = None,
         operation_mode: Optional[HVACOperationMode] = None,
     ):
         """Initialize remote value of KNX DPT 1 representing a climate operation mode."""
@@ -272,7 +261,7 @@ def __init__(
         sync_state: bool = True,
         device_name: Optional[str] = None,
         feature_name: str = "Controller Mode Heat/Cool",
-        after_update_cb: Optional["AsyncCallback"] = None,
+        after_update_cb: Optional[AsyncCallbackType] = None,
         controller_mode: Optional[HVACControllerMode] = None,
     ):
         """Initialize remote value of KNX DPT 1 representing a climate controller mode."""

xknx/remote_value/remote_value_color_rgb.py

@@ -3,12 +3,16 @@
 
 DPT 232.600.
 """
-from typing import List
+from typing import TYPE_CHECKING, List, Optional, Sequence, Tuple, Union
 
-from xknx.dpt import DPTArray
+from xknx.dpt import DPTArray, DPTBinary
 from xknx.exceptions import ConversionError
 
-from .remote_value import RemoteValue
+from .remote_value import AsyncCallbackType, RemoteValue
+
+if TYPE_CHECKING:
+    from xknx.telegram.address import GroupAddressableType
+    from xknx.xknx import XKNX
 
 
 class RemoteValueColorRGB(RemoteValue[DPTArray]):
@@ -18,13 +22,13 @@ class RemoteValueColorRGB(RemoteValue[DPTArray]):
 
     def __init__(
         self,
-        xknx,
-        group_address=None,
-        group_address_state=None,
-        device_name=None,
-        feature_name="Color RGB",
-        after_update_cb=None,
-        passive_group_addresses: List[str] = None,
+        xknx: "XKNX",
+        group_address: Optional["GroupAddressableType"] = None,
+        group_address_state: Optional["GroupAddressableType"] = None,
+        device_name: Optional[str] = None,
+        feature_name: str = "Color RGB",
+        after_update_cb: Optional[AsyncCallbackType] = None,
+        passive_group_addresses: Optional[List["GroupAddressableType"]] = None,
     ):
         """Initialize remote value of KNX DPT 232.600 (DPT_Color_RGB)."""
         # pylint: disable=too-many-arguments
@@ -38,15 +42,17 @@ def __init__(
             passive_group_addresses=passive_group_addresses,
         )
 
-    def payload_valid(self, payload):
+    def payload_valid(
+        self, payload: Optional[Union[DPTArray, DPTBinary]]
+    ) -> Optional[DPTArray]:
         """Test if telegram payload may be parsed."""
         return (
             payload
             if isinstance(payload, DPTArray) and len(payload.value) == 3
             else None
         )
 
-    def to_knx(self, value):
+    def to_knx(self, value: Sequence[int]) -> DPTArray:
         """Convert value to payload."""
         if not isinstance(value, (list, tuple)):
             raise ConversionError(
@@ -71,6 +77,6 @@ def to_knx(self, value):
 
         return DPTArray(list(value))
 
-    def from_knx(self, payload):
+    def from_knx(self, payload: DPTArray) -> Tuple[int, int, int]:
         """Convert current payload to value."""
         return payload.value[0], payload.value[1], payload.value[2]