From bcff4fdc58da57a43c8643d609c48c36414c3783 Mon Sep 17 00:00:00 2001 From: SukramJ Date: Sat, 25 Jan 2025 21:35:15 +0100 Subject: [PATCH] Add calculated data points: FrostPoint (#2014) --- changelog.md | 4 + hahomematic/const.py | 2 +- hahomematic/model/calculated/__init__.py | 5 +- hahomematic/model/calculated/climate.py | 144 ++++++++++------------- hahomematic/model/calculated/support.py | 10 +- tests/test_central.py | 14 +-- tests/test_central_pydevccu.py | 8 +- 7 files changed, 93 insertions(+), 94 deletions(-) diff --git a/changelog.md b/changelog.md index 1aa3895a..7f6d0b18 100644 --- a/changelog.md +++ b/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 diff --git a/hahomematic/const.py b/hahomematic/const.py index e656d6b0..d9e8ef57 100644 --- a/hahomematic/const.py +++ b/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" diff --git a/hahomematic/model/calculated/__init__.py b/hahomematic/model/calculated/__init__.py index 9459e0ca..b3eb0ffc 100644 --- a/hahomematic/model/calculated/__init__.py +++ b/hahomematic/model/calculated/__init__.py @@ -7,7 +7,7 @@ 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 @@ -15,6 +15,7 @@ "ApparentTemperature", "CalculatedDataPoint", "DewPoint", + "FrostPoint", "OperatingVoltageLevel", "VaporConcentration", "create_calculated_data_points", @@ -22,7 +23,7 @@ _LOGGER: Final = logging.getLogger(__name__) -_CALCULATORS: Final = (ApparentTemperature, DewPoint, OperatingVoltageLevel, VaporConcentration) +_CALCULATORS: Final = (ApparentTemperature, DewPoint, FrostPoint, OperatingVoltageLevel, VaporConcentration) @inspector() diff --git a/hahomematic/model/calculated/climate.py b/hahomematic/model/calculated/climate.py index 8024f37b..7e0dc229 100644 --- a/hahomematic/model/calculated/climate.py +++ b/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,10 +21,9 @@ _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: @@ -31,17 +31,54 @@ def __init__(self, channel: hmd.Channel) -> None: 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.""" diff --git a/hahomematic/model/calculated/support.py b/hahomematic/model/calculated/support.py index 4e358b21..da3a78eb 100644 --- a/hahomematic/model/calculated/support.py +++ b/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) diff --git a/tests/test_central.py b/tests/test_central.py index cd29c8e2..b492d1d3 100644 --- a/tests/test_central.py +++ b/tests/test_central.py @@ -502,7 +502,7 @@ 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.""" @@ -510,7 +510,7 @@ def _device_changed(self, *args: Any, **kwargs: Any) -> None: 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() diff --git a/tests/test_central_pydevccu.py b/tests/test_central_pydevccu.py index 50f1a11e..910229bd 100644 --- a/tests/test_central_pydevccu.py +++ b/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"]