Store tmp value for sysvar data points (#1864)

* Store tmp value for sysvar data points

* Update data_point.py

* Update changelog.md

* Update test_binary_sensor.py

changelog.md

@@ -2,12 +2,13 @@
 
 - Fix returned version of client
 - Improve store tmp value
+- Store temporary value for sysvar data points
 
 # Version 2024.11.4 (2024-11-19)
 
 - Add sysvar/program refresh to scheduler
 - Run periodic tasks with an individual interval
-- Store tmp value for polling client data points
+- Store temporary value for polling client data points
 
 # Version 2024.11.3 (2024-11-18)
 

hahomematic/const.py

@@ -485,6 +485,7 @@ class ParameterType(StrEnum):
 # interface_id, channel_address, paramset_key,parameter
 DP_KEY = tuple[str, str, ParamsetKey, str]
 DP_KEY_VALUE = tuple[DP_KEY, Any]
+SYSVAR_TYPE = bool | float | int | str | None
 
 HMIP_FIRMWARE_UPDATE_IN_PROGRESS_STATES: Final[tuple[DeviceFirmwareState, ...]] = (
     DeviceFirmwareState.DO_UPDATE_PENDING,
@@ -603,7 +604,7 @@ class ProgramData(HubData):
 class SystemVariableData(HubData):
     """Dataclass for system variables."""
 
-    value: bool | float | int | str | None
+    value: SYSVAR_TYPE
     data_type: SysvarType | None = None
     extended_sysvar: bool = False
     max_value: float | int | None = None

hahomematic/model/hub/data_point.py

@@ -8,7 +8,7 @@
 from slugify import slugify
 
 from hahomematic import central as hmcu
-from hahomematic.const import SYSVAR_ADDRESS, HubData, SystemVariableData
+from hahomematic.const import SYSVAR_ADDRESS, SYSVAR_TYPE, HubData, SystemVariableData
 from hahomematic.model.data_point import CallbackDataPoint
 from hahomematic.model.decorators import (
     config_property,
@@ -83,8 +83,12 @@ def __init__(
         self._max: Final = data.max_value
         self._min: Final = data.min_value
         self._unit: Final = data.unit
-        self._value = data.value
-        self._previous_value: bool | float | int | str | None = None
+
+        self._current_value: SYSVAR_TYPE = data.value
+        self._previous_value: SYSVAR_TYPE = None
+        self._temporary_value: SYSVAR_TYPE = None
+
+        self._state_uncertain: bool = True
         self._service_methods = get_service_calls(obj=self)
 
     @state_property
@@ -93,10 +97,22 @@ def available(self) -> bool:
         return self.central.available
 
     @property
-    def previous_value(self) -> Any | None:
+    def previous_value(self) -> SYSVAR_TYPE:
         """Return the previous value."""
         return self._previous_value
 
+    @property
+    def state_uncertain(self) -> bool:
+        """Return, if the state is uncertain."""
+        return self._state_uncertain
+
+    @property
+    def _value(self) -> Any | None:
+        """Return the value."""
+        if self._temporary_refreshed_at > self._refreshed_at:
+            return self._temporary_value
+        return self._current_value
+
     @state_property
     def value(self) -> Any | None:
         """Return the value."""
@@ -139,32 +155,50 @@ def get_name(self, data: HubData) -> str:
 
     def write_value(self, value: Any) -> None:
         """Set variable value on CCU/Homegear."""
-        old_value = self._value
-        if self.data_type:
-            value = parse_sys_var(data_type=self.data_type, raw_value=value)
-        elif isinstance(old_value, bool):
-            value = bool(value)
-        elif isinstance(old_value, int):
-            value = int(value)
-        elif isinstance(old_value, str):
-            value = str(value)
-        elif isinstance(old_value, float):
-            value = float(value)
-
-        if old_value == value:
+        old_value = self._current_value
+        new_value = self._convert_value(old_value=old_value, new_value=value)
+        if old_value == new_value:
             self._set_refreshed_at()
         else:
             self._set_modified_at()
             self._previous_value = old_value
-            self._value = value
+            self._current_value = new_value
+            self._state_uncertain = False
+        self.fire_data_point_updated_callback()
 
+    def _write_temporary_value(self, value: Any) -> None:
+        """Update the temporary value of the data_point."""
+        temp_value = self._convert_value(old_value=self._current_value, new_value=value)
+        if self._value == temp_value:
+            self._set_temporary_refreshed_at()
+        else:
+            self._set_temporary_modified_at()
+            self._temporary_value = temp_value
+            self._state_uncertain = True
         self.fire_data_point_updated_callback()
 
+    def _convert_value(self, old_value: Any, new_value: Any) -> Any:
+        """Convert to value to SYSVAR_TYPE."""
+        if new_value is None:
+            return None
+        value = new_value
+        if self.data_type:
+            value = parse_sys_var(data_type=self.data_type, raw_value=new_value)
+        elif isinstance(old_value, bool):
+            value = bool(new_value)
+        elif isinstance(old_value, int):
+            value = int(new_value)
+        elif isinstance(old_value, str):
+            value = str(new_value)
+        elif isinstance(old_value, float):
+            value = float(new_value)
+        return value
+
     @service()
     async def send_variable(self, value: Any) -> None:
         """Set variable value on CCU/Homegear."""
         if client := self.central.primary_client:
             await client.set_system_variable(
                 name=self.ccu_var_name, value=parse_sys_var(self.data_type, value)
             )
-        self.write_value(value=value)
+        self._write_temporary_value(value=value)

hahomematic/model/support.py

@@ -20,6 +20,7 @@
     SYSVAR_ADDRESS,
     SYSVAR_SET_PATH_ROOT,
     SYSVAR_STATE_PATH_ROOT,
+    SYSVAR_TYPE,
     VIRTDEV_SET_PATH_ROOT,
     VIRTDEV_STATE_PATH_ROOT,
     VIRTUAL_REMOTE_ADDRESSES,
@@ -550,7 +551,7 @@ def check_channel_is_the_only_primary_channel(
 
 
 def get_value_from_value_list(
-    value: bool | float | int | str | None, value_list: tuple[str, ...] | list[str] | None
+    value: SYSVAR_TYPE, value_list: tuple[str, ...] | list[str] | None
 ) -> str | None:
     """Check if value is in value list."""
     if (
@@ -564,7 +565,7 @@ def get_value_from_value_list(
 
 
 def get_index_of_value_from_value_list(
-    value: bool | float | int | str | None, value_list: tuple[str, ...] | list[str] | None
+    value: SYSVAR_TYPE, value_list: tuple[str, ...] | list[str] | None
 ) -> int | None:
     """Check if value is in value list."""
     if (

tests/test_binary_sensor.py

@@ -89,6 +89,6 @@ async def test_hmsysvarbinarysensor(
     assert binary_sensor.value is False
     assert binary_sensor.is_extended is False
     assert binary_sensor.data_type == "LOGIC"
-    with pytest.raises(TypeError):
-        binary_sensor.write_value(None)
     assert binary_sensor.value is False
+    binary_sensor.write_value(True)
+    assert binary_sensor.value is True