Add calculated data points: FrostPoint (#2014)

changelog.md

@@ -1,3 +1,7 @@
+# Version 2025.1.15 (2025-01-25)
+
+- Add calculated data points: FrostPoint
+
 # Version 2025.1.14 (2025-01-25)
 
 - Add calculated data points: ApparentTemperature, DewPoint, VaporConcentration

hahomematic/const.py

@@ -9,7 +9,7 @@
 import re
 from typing import Any, Final, NamedTuple, Required, TypedDict
 
-VERSION: Final = "2025.1.14"
+VERSION: Final = "2025.1.15"
 
 # default
 DEFAULT_CUSTOM_ID: Final = "custom_id"

hahomematic/model/calculated/__init__.py

@@ -7,22 +7,23 @@
 
 from hahomematic.decorators import inspector
 from hahomematic.model import device as hmd
-from hahomematic.model.calculated.climate import ApparentTemperature, DewPoint, VaporConcentration
+from hahomematic.model.calculated.climate import ApparentTemperature, DewPoint, FrostPoint, VaporConcentration
 from hahomematic.model.calculated.data_point import CalculatedDataPoint
 from hahomematic.model.calculated.operating_voltage_level import OperatingVoltageLevel
 
 __all__ = [
     "ApparentTemperature",
     "CalculatedDataPoint",
     "DewPoint",
+    "FrostPoint",
     "OperatingVoltageLevel",
     "VaporConcentration",
     "create_calculated_data_points",
 ]
 
 _LOGGER: Final = logging.getLogger(__name__)
 
-_CALCULATORS: Final = (ApparentTemperature, DewPoint, OperatingVoltageLevel, VaporConcentration)
+_CALCULATORS: Final = (ApparentTemperature, DewPoint, FrostPoint, OperatingVoltageLevel, VaporConcentration)
 
 
 @inspector()

hahomematic/model/calculated/climate.py

@@ -11,6 +11,7 @@
 from hahomematic.model.calculated.support import (
     calculate_apparent_temperature,
     calculate_dew_point,
+    calculate_frost_point,
     calculate_vapor_concentration,
 )
 from hahomematic.model.decorators import state_property
@@ -20,28 +21,64 @@
 _LOGGER: Final = logging.getLogger(__name__)
 
 
-class ApparentTemperature[SensorT: float | None](CalculatedDataPoint[SensorT]):
-    """Implementation of a calculated sensor for apparent temperature."""
+class BaseClimateSensor[SensorT: float | None](CalculatedDataPoint[SensorT]):
+    """Implementation of a calculated climate sensor."""
 
-    _calculated_parameter = "APPARENT_TEMPERATURE"
     _category = DataPointCategory.SENSOR
 
     def __init__(self, channel: hmd.Channel) -> None:
         """Initialize the data point."""
 
         super().__init__(channel=channel)
         self._type = ParameterType.FLOAT
-        self._unit = "°C"
 
     def _init_data_point_fields(self) -> None:
         """Init the data point fields."""
         super()._init_data_point_fields()
-        self._dp_temperature: DpSensor = self._add_data_point(
-            parameter=Parameter.ACTUAL_TEMPERATURE, paramset_key=ParamsetKey.VALUES, data_point_type=DpSensor
+        self._dp_temperature: DpSensor = (
+            self._add_data_point(
+                parameter=Parameter.TEMPERATURE, paramset_key=ParamsetKey.VALUES, data_point_type=DpSensor
+            )
+            if self._channel.get_generic_data_point(parameter=Parameter.TEMPERATURE, paramset_key=ParamsetKey.VALUES)
+            else self._add_data_point(
+                parameter=Parameter.ACTUAL_TEMPERATURE, paramset_key=ParamsetKey.VALUES, data_point_type=DpSensor
+            )
         )
         self._dp_humidity: DpSensor = self._add_data_point(
             parameter=Parameter.HUMIDITY, paramset_key=ParamsetKey.VALUES, data_point_type=DpSensor
         )
+
+    @staticmethod
+    def is_relevant_for_model(channel: hmd.Channel) -> bool:
+        """Return if this calculated data point is relevant for the model."""
+        return (
+            element_matches_key(search_elements=_RELEVANT_MODELS, compare_with=channel.device.model)
+            and (
+                channel.get_generic_data_point(parameter=Parameter.TEMPERATURE, paramset_key=ParamsetKey.VALUES)
+                is not None
+                or channel.get_generic_data_point(
+                    parameter=Parameter.ACTUAL_TEMPERATURE, paramset_key=ParamsetKey.VALUES
+                )
+                is not None
+            )
+            and channel.get_generic_data_point(parameter=Parameter.HUMIDITY, paramset_key=ParamsetKey.VALUES)
+            is not None
+        )
+
+
+class ApparentTemperature(BaseClimateSensor):
+    """Implementation of a calculated sensor for apparent temperature."""
+
+    _calculated_parameter = "APPARENT_TEMPERATURE"
+
+    def __init__(self, channel: hmd.Channel) -> None:
+        """Initialize the data point."""
+        super().__init__(channel=channel)
+        self._unit = "°C"
+
+    def _init_data_point_fields(self) -> None:
+        """Init the data point fields."""
+        super()._init_data_point_fields()
         self._dp_wind_speed: DpSensor = self._add_data_point(
             parameter=Parameter.WIND_SPEED, paramset_key=ParamsetKey.VALUES, data_point_type=DpSensor
         )
@@ -75,109 +112,58 @@ def value(self) -> float | None:
         return None
 
 
-class DewPoint[SensorT: float | None](CalculatedDataPoint[SensorT]):
+class DewPoint(BaseClimateSensor):
     """Implementation of a calculated sensor for dew point."""
 
-    _calculated_parameter = "DEWPOINT"
-    _category = DataPointCategory.SENSOR
+    _calculated_parameter = "DEW_POINT"
 
     def __init__(self, channel: hmd.Channel) -> None:
         """Initialize the data point."""
-
         super().__init__(channel=channel)
-        self._type = ParameterType.FLOAT
         self._unit = "°C"
 
-    def _init_data_point_fields(self) -> None:
-        """Init the data point fields."""
-        super()._init_data_point_fields()
-        self._dp_temperature: DpSensor = (
-            self._add_data_point(
-                parameter=Parameter.TEMPERATURE, paramset_key=ParamsetKey.VALUES, data_point_type=DpSensor
-            )
-            if self._channel.get_generic_data_point(parameter=Parameter.TEMPERATURE, paramset_key=ParamsetKey.VALUES)
-            else self._add_data_point(
-                parameter=Parameter.ACTUAL_TEMPERATURE, paramset_key=ParamsetKey.VALUES, data_point_type=DpSensor
+    @state_property
+    def value(self) -> float | None:
+        """Return the value."""
+        if self._dp_temperature.value is not None and self._dp_humidity.value is not None:
+            return calculate_dew_point(
+                temperature=self._dp_temperature.value,
+                humidity=self._dp_humidity.value,
             )
-        )
-        self._dp_humidity: DpSensor = self._add_data_point(
-            parameter=Parameter.HUMIDITY, paramset_key=ParamsetKey.VALUES, data_point_type=DpSensor
-        )
+        return None
 
-    @staticmethod
-    def is_relevant_for_model(channel: hmd.Channel) -> bool:
-        """Return if this calculated data point is relevant for the model."""
-        return (
-            element_matches_key(search_elements=_RELEVANT_MODELS, compare_with=channel.device.model)
-            and (
-                channel.get_generic_data_point(parameter=Parameter.TEMPERATURE, paramset_key=ParamsetKey.VALUES)
-                is not None
-                or channel.get_generic_data_point(
-                    parameter=Parameter.ACTUAL_TEMPERATURE, paramset_key=ParamsetKey.VALUES
-                )
-                is not None
-            )
-            and channel.get_generic_data_point(parameter=Parameter.HUMIDITY, paramset_key=ParamsetKey.VALUES)
-            is not None
-        )
+
+class FrostPoint(BaseClimateSensor):
+    """Implementation of a calculated sensor for frost point."""
+
+    _calculated_parameter = "FROST_POINT"
+
+    def __init__(self, channel: hmd.Channel) -> None:
+        """Initialize the data point."""
+        super().__init__(channel=channel)
+        self._unit = "°C"
 
     @state_property
     def value(self) -> float | None:
         """Return the value."""
         if self._dp_temperature.value is not None and self._dp_humidity.value is not None:
-            return calculate_dew_point(
+            return calculate_frost_point(
                 temperature=self._dp_temperature.value,
                 humidity=self._dp_humidity.value,
             )
         return None
 
 
-class VaporConcentration[SensorT: float | None](CalculatedDataPoint[SensorT]):
+class VaporConcentration(BaseClimateSensor):
     """Implementation of a calculated sensor for vapor concentration."""
 
     _calculated_parameter = "VAPOR_CONCENTRATION"
-    _category = DataPointCategory.SENSOR
 
     def __init__(self, channel: hmd.Channel) -> None:
         """Initialize the data point."""
-
         super().__init__(channel=channel)
-        self._type = ParameterType.FLOAT
         self._unit = "g/m³"
 
-    def _init_data_point_fields(self) -> None:
-        """Init the data point fields."""
-        super()._init_data_point_fields()
-        self._dp_temperature: DpSensor = (
-            self._add_data_point(
-                parameter=Parameter.TEMPERATURE, paramset_key=ParamsetKey.VALUES, data_point_type=DpSensor
-            )
-            if self._channel.get_generic_data_point(parameter=Parameter.TEMPERATURE, paramset_key=ParamsetKey.VALUES)
-            else self._add_data_point(
-                parameter=Parameter.ACTUAL_TEMPERATURE, paramset_key=ParamsetKey.VALUES, data_point_type=DpSensor
-            )
-        )
-        self._dp_humidity: DpSensor = self._add_data_point(
-            parameter=Parameter.HUMIDITY, paramset_key=ParamsetKey.VALUES, data_point_type=DpSensor
-        )
-
-    @staticmethod
-    def is_relevant_for_model(channel: hmd.Channel) -> bool:
-        """Return if this calculated data point is relevant for the model."""
-        return (
-            element_matches_key(search_elements=_RELEVANT_MODELS, compare_with=channel.device.model)
-            and (
-                channel.get_generic_data_point(parameter=Parameter.TEMPERATURE, paramset_key=ParamsetKey.VALUES)
-                is not None
-                or channel.get_generic_data_point(
-                    parameter=Parameter.ACTUAL_TEMPERATURE, paramset_key=ParamsetKey.VALUES
-                )
-                is not None
-            )
-            and channel.get_generic_data_point(parameter=Parameter.HUMIDITY, paramset_key=ParamsetKey.VALUES)
-            is not None
-        )
-
     @state_property
     def value(self) -> float | None:
         """Return the value."""

hahomematic/model/calculated/support.py

@@ -98,7 +98,7 @@ def calculate_apparent_temperature(temperature: float, humidity: float, wind_spe
 
 
 def calculate_dew_point(temperature: float, humidity: float) -> float:
-    """Calculate the dew point based on NOAA."""
+    """Calculate the dew point."""
     a0 = 373.15 / (273.15 + temperature)
     s = -7.90298 * (a0 - 1)
     s += 5.02808 * math.log(a0, 10)
@@ -108,3 +108,11 @@ def calculate_dew_point(temperature: float, humidity: float) -> float:
     vp = pow(10, s - 3) * humidity
     td = math.log(vp / 0.61078)
     return round((241.88 * td) / (17.558 - td), 1)
+
+
+def calculate_frost_point(temperature: float, humidity: float) -> float:
+    """Calculate the frost point."""
+    dewpoint = calculate_dew_point(temperature=temperature, humidity=humidity)
+    t = temperature + 273.15
+    td = dewpoint + 273.15
+    return round((td + (2671.02 / ((2954.61 / t) + 2.193665 * math.log(t) - 13.3448)) - t) - 273.15, 1)

tests/test_central.py

@@ -502,15 +502,15 @@ async def test_data_points_by_category(
     central, _, _ = central_client_factory
     ebp_sensor = central.get_data_points(category=DataPointCategory.SENSOR)
     assert ebp_sensor
-    assert len(ebp_sensor) == 15
+    assert len(ebp_sensor) == 16
 
     def _device_changed(self, *args: Any, **kwargs: Any) -> None:
         """Handle device state changes."""
 
     ebp_sensor[0].register_data_point_updated_callback(cb=_device_changed, custom_id="some_id")
     ebp_sensor2 = central.get_data_points(category=DataPointCategory.SENSOR, registered=False)
     assert ebp_sensor2
-    assert len(ebp_sensor2) == 14
+    assert len(ebp_sensor2) == 15
 
 
 @pytest.mark.asyncio
@@ -576,13 +576,13 @@ async def test_add_device(
     """Test add_device."""
     central, _, _ = central_client_factory
     assert len(central._devices) == 1
-    assert len(central.get_data_points(exclude_no_create=False)) == 31
+    assert len(central.get_data_points(exclude_no_create=False)) == 32
     assert len(central.device_descriptions._raw_device_descriptions.get(const.INTERFACE_ID)) == 9
     assert len(central.paramset_descriptions._raw_paramset_descriptions.get(const.INTERFACE_ID)) == 9
     dev_desc = helper.load_device_description(central=central, filename="HmIP-BSM.json")
     await central.add_new_devices(interface_id=const.INTERFACE_ID, device_descriptions=dev_desc)
     assert len(central._devices) == 2
-    assert len(central.get_data_points(exclude_no_create=False)) == 62
+    assert len(central.get_data_points(exclude_no_create=False)) == 63
     assert len(central.device_descriptions._raw_device_descriptions.get(const.INTERFACE_ID)) == 20
     assert len(central.paramset_descriptions._raw_paramset_descriptions.get(const.INTERFACE_ID)) == 20
     await central.add_new_devices(interface_id="NOT_ANINTERFACE_ID", device_descriptions=dev_desc)
@@ -609,13 +609,13 @@ async def test_delete_device(
     """Test device delete_device."""
     central, _, _ = central_client_factory
     assert len(central._devices) == 2
-    assert len(central.get_data_points(exclude_no_create=False)) == 62
+    assert len(central.get_data_points(exclude_no_create=False)) == 63
     assert len(central.device_descriptions._raw_device_descriptions.get(const.INTERFACE_ID)) == 20
     assert len(central.paramset_descriptions._raw_paramset_descriptions.get(const.INTERFACE_ID)) == 20
 
     await central.delete_devices(interface_id=const.INTERFACE_ID, addresses=["VCU2128127"])
     assert len(central._devices) == 1
-    assert len(central.get_data_points(exclude_no_create=False)) == 31
+    assert len(central.get_data_points(exclude_no_create=False)) == 32
     assert len(central.device_descriptions._raw_device_descriptions.get(const.INTERFACE_ID)) == 9
     assert len(central.paramset_descriptions._raw_paramset_descriptions.get(const.INTERFACE_ID)) == 9
 
@@ -828,7 +828,7 @@ async def test_central_direct(factory: helper.Factory) -> None:
         assert central.available is False
         assert central.system_information.serial == "0815_4711"
         assert len(central._devices) == 2
-        assert len(central.get_data_points(exclude_no_create=False)) == 62
+        assert len(central.get_data_points(exclude_no_create=False)) == 63
     finally:
         await central.stop()
 

tests/test_central_pydevccu.py

@@ -30,7 +30,7 @@ async def test_central_mini(central_unit_mini) -> None:
     assert central_unit_mini.get_client(const.INTERFACE_ID).model == "PyDevCCU"
     assert central_unit_mini.primary_client.model == "PyDevCCU"
     assert len(central_unit_mini._devices) == 2
-    assert len(central_unit_mini.get_data_points(exclude_no_create=False)) == 68
+    assert len(central_unit_mini.get_data_points(exclude_no_create=False)) == 69
 
     usage_types: dict[DataPointUsage, int] = {}
     for data_point in central_unit_mini.get_data_points(exclude_no_create=False):
@@ -42,7 +42,7 @@ async def test_central_mini(central_unit_mini) -> None:
 
     assert usage_types[DataPointUsage.NO_CREATE] == 45
     assert usage_types[DataPointUsage.CDP_PRIMARY] == 4
-    assert usage_types[DataPointUsage.DATA_POINT] == 14
+    assert usage_types[DataPointUsage.DATA_POINT] == 15
     assert usage_types[DataPointUsage.CDP_VISIBLE] == 5
 
 
@@ -133,13 +133,13 @@ async def test_central_full(central_unit_full) -> None:
 
     assert usage_types[DataPointUsage.NO_CREATE] == 4150
     assert usage_types[DataPointUsage.CDP_PRIMARY] == 261
-    assert usage_types[DataPointUsage.DATA_POINT] == 3769
+    assert usage_types[DataPointUsage.DATA_POINT] == 3783
     assert usage_types[DataPointUsage.CDP_VISIBLE] == 133
     assert usage_types[DataPointUsage.CDP_SECONDARY] == 154
 
     assert len(ce_channels) == 124
     assert len(data_point_types) == 6
-    assert len(parameters) == 225
+    assert len(parameters) == 227
 
     assert len(central_unit_full._devices) == 386
     virtual_remotes = ["VCU4264293", "VCU0000057", "VCU0000001"]