Merge pull request #639 from farmio/rgbw-type

Unwrap Optional RemoteValue.value

test/devices_tests/climate_test.py

@@ -1067,6 +1067,21 @@ def test_sync_heat_cool(self):
             telegram1, Telegram(GroupAddress("1/2/15"), payload=GroupValueRead())
         )
 
+    def test_sync_mode_from_climate(self):
+        """Test sync function / propagating to mode."""
+        xknx = XKNX()
+        climate_mode = ClimateMode(
+            xknx, "TestClimateMode", group_address_operation_mode_state="1/2/4"
+        )
+        climate = Climate(xknx, "TestClimate", mode=climate_mode)
+
+        self.loop.run_until_complete(climate.sync())
+        self.assertEqual(xknx.telegrams.qsize(), 1)
+        telegram1 = xknx.telegrams.get_nowait()
+        self.assertEqual(
+            telegram1, Telegram(GroupAddress("1/2/4"), payload=GroupValueRead())
+        )
+
     #
     # TEST PROCESS
     #

xknx/devices/binary_sensor.py

@@ -119,7 +119,8 @@ def from_config(
 
     async def _state_from_remote_value(self) -> None:
         """Update the internal state from RemoteValue (Callback)."""
-        await self._set_internal_state(self.remote_value.value)
+        if self.remote_value.value is not None:
+            await self._set_internal_state(self.remote_value.value)
 
     async def _set_internal_state(self, state: bool) -> None:
         """Set the internal state of the device. If state was changed after_update hooks and connected Actions are executed."""

xknx/devices/climate.py

@@ -6,7 +6,7 @@
 """
 from enum import Enum
 import logging
-from typing import TYPE_CHECKING, Any, Dict, Iterator, Optional, Union, cast
+from typing import TYPE_CHECKING, Any, Dict, Iterator, Optional, Union
 
 from xknx.remote_value import (
     GroupAddressesType,
@@ -242,15 +242,14 @@ async def turn_off(self) -> None:
     @property
     def initialized_for_setpoint_shift_calculations(self) -> bool:
         """Test if object is initialized for setpoint shift calculations."""
-        if not self._setpoint_shift.initialized:
-            return False
-        if self._setpoint_shift.value is None:
-            return False
-        if not self.target_temperature.initialized:
-            return False
-        if self.target_temperature.value is None:
-            return False
-        return True
+        if (
+            self._setpoint_shift.initialized
+            and self._setpoint_shift.value is not None
+            and self.target_temperature.initialized
+            and self.target_temperature.value is not None
+        ):
+            return True
+        return False
 
     async def set_target_temperature(self, target_temperature: float) -> None:
         """Send new target temperature or setpoint_shift to KNX bus."""
@@ -273,8 +272,12 @@ def base_temperature(self) -> Optional[float]:
         As this value is usually not available via KNX, we have to derive this from the current
         target temperature and the current set point shift.
         """
-        if self.initialized_for_setpoint_shift_calculations:
-            return cast(float, self.target_temperature.value - self.setpoint_shift)
+        # implies self.initialized_for_setpoint_shift_calculations in a mypy compatible way:
+        if (
+            self.target_temperature.value is not None
+            and self._setpoint_shift.value is not None
+        ):
+            return self.target_temperature.value - self._setpoint_shift.value  # type: ignore
         return None
 
     @property

xknx/devices/climate_mode.py

@@ -361,7 +361,8 @@ async def process_group_write(self, telegram: "Telegram") -> None:
         if self.supports_controller_mode:
             for rv in self._iter_controller_remote_values():
                 if await rv.process(telegram):
-                    await self._set_internal_controller_mode(rv.value)
+                    if rv.value is not None:
+                        await self._set_internal_controller_mode(rv.value)
                     return
 
     def __str__(self) -> str:

xknx/devices/cover.py

@@ -266,18 +266,22 @@ async def set_position(self, position: int) -> None:
 
     async def _target_position_from_rv(self) -> None:
         """Update the target postion from RemoteValue (Callback)."""
-        self.travelcalculator.start_travel(self.position_target.value)
-        await self.after_update()
+        new_target = self.position_target.value
+        if new_target is not None:
+            self.travelcalculator.start_travel(new_target)
+            await self.after_update()
 
     async def _current_position_from_rv(self) -> None:
         """Update the current postion from RemoteValue (Callback)."""
         position_before_update = self.travelcalculator.current_position()
-        if self.is_traveling():
-            self.travelcalculator.update_position(self.position_current.value)
-        else:
-            self.travelcalculator.set_position(self.position_current.value)
-        if position_before_update != self.travelcalculator.current_position():
-            await self.after_update()
+        new_position = self.position_current.value
+        if new_position is not None:
+            if self.is_traveling():
+                self.travelcalculator.update_position(new_position)
+            else:
+                self.travelcalculator.set_position(new_position)
+            if position_before_update != self.travelcalculator.current_position():
+                await self.after_update()
 
     async def set_angle(self, angle: int) -> None:
         """Move cover to designated angle."""

xknx/devices/light.py

@@ -21,6 +21,7 @@
     Iterator,
     Optional,
     Tuple,
+    cast,
 )
 
 from xknx.remote_value import (
@@ -549,7 +550,7 @@ def current_color(self) -> Tuple[Optional[Tuple[int, int, int]], Optional[int]]:
         )
         if None in colors:
             return None, self.white.brightness.value
-        return colors, self.white.brightness.value
+        return cast(Tuple[int, int, int], colors), self.white.brightness.value
 
     async def set_color(
         self, color: Tuple[int, int, int], white: Optional[int] = None

xknx/remote_value/remote_value_color_rgbw.py

@@ -36,7 +36,7 @@ def __init__(
             feature_name=feature_name,
             after_update_cb=after_update_cb,
         )
-        self.previous_value: Tuple[int, ...] = (0, 0, 0, 0)
+        self.previous_value: Tuple[int, int, int, int] = (0, 0, 0, 0)
 
     def payload_valid(
         self, payload: Optional[Union[DPTArray, DPTBinary]]
@@ -101,17 +101,17 @@ def to_knx(self, value: Sequence[int]) -> DPTArray:
             return DPTArray(list(rgbw) + [0x00] + list(value[4:]))
         return DPTArray(value)
 
-    def from_knx(self, payload: DPTArray) -> Tuple[int, ...]:
+    def from_knx(self, payload: DPTArray) -> Tuple[int, int, int, int]:
         """
         Convert current payload to value. Always 4 byte (RGBW).
 
         If one element is invalid, use the previous value. All previous element
         values are initialized to 0.
         """
-        _result = []
+        _result = list(self.previous_value)
         for i in range(0, len(payload.value) - 2):
-            valid = (payload.value[5] & (0x08 >> i)) != 0  # R,G,B,W value valid?
-            _result.append(payload.value[i] if valid else self.previous_value[i])
-        result = tuple(_result)
+            if payload.value[5] & (0x08 >> i):  # R,G,B,W value valid?
+                _result[i] = payload.value[i]
+        result = (_result[0], _result[1], _result[2], _result[3])
         self.previous_value = result
         return result