gilles/versatile_thermostat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

All tests not ok

Browse Source
This commit is contained in:
Jean-Marc Collin
2025-01-03 17:43:27 +00:00
parent 03fbc5362a
commit 24fcb7a161
14 changed files with 441 additions and 392 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
custom_components/versatile_thermostat/base_manager.py
View File

@@ -38,6 +38,10 @@ class BaseFeatureManager:
self._active_listener = []
async def refresh_state(self):
"""Refresh the state and return True if a change have been made"""
return False
def add_listener(self, func: CALLBACK_TYPE) -> None:
"""Add a listener to the list of active listener"""
self._active_listener.append(func)

18
custom_components/versatile_thermostat/base_thermostat.py
View File

@@ -50,11 +50,10 @@ from homeassistant.const import (
ATTR_TEMPERATURE,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
STATE_ON,
)
from .const import * # pylint: disable=wildcard-import, unused-wildcard-import
from .commons import ConfigData, T
from .commons import ConfigData, T, deprecated
from .config_schema import * # pylint: disable=wildcard-import, unused-wildcard-import
@@ -1611,10 +1610,11 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
# Check overpowering condition
# Not necessary for switch because each switch is checking at startup
overpowering = await self._power_manager.check_overpowering()
if overpowering == STATE_ON:
_LOGGER.debug("%s - End of cycle (overpowering)", self)
return True
# overpowering is now centralized
# overpowering = await self._power_manager.check_overpowering()
# if overpowering == STATE_ON:
# _LOGGER.debug("%s - End of cycle (overpowering)", self)
# return True
safety: bool = await self._safety_manager.refresh_state()
if safety and self.is_over_climate:
@@ -1957,14 +1957,14 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
return self._presets.get(preset, None) is not None
# For testing purpose
@DeprecationWarning
# @deprecated
def _set_now(self, now: datetime):
"""Set the now timestamp. This is only for tests purpose
This method should be replaced by the vthermAPI equivalent"""
VersatileThermostatAPI.get_vtherm_api(self._hass).set_now(now)
VersatileThermostatAPI.get_vtherm_api(self._hass)._set_now(now)
# @deprecated
@property
@DeprecationWarning
def now(self) -> datetime:
"""Get now. The local datetime or the overloaded _set_now date
This method should be replaced by the vthermAPI equivalent"""

41
custom_components/versatile_thermostat/central_feature_power_manager.py
View File

@@ -4,10 +4,6 @@ import logging
from typing import Any
from functools import cmp_to_key
from homeassistant.const import (
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant, Event, callback
from homeassistant.helpers.event import (
async_track_state_change_event,
@@ -106,17 +102,17 @@ class CentralFeaturePowerManager(BaseFeatureManager):
"""Handle power changes."""
_LOGGER.debug("Thermostat %s - Receive new Power event", self)
_LOGGER.debug(event)
self.refresh_state()
await self.refresh_state()
@callback
async def _max_power_sensor_changed(self, event: Event[EventStateChangedData]):
"""Handle power max changes."""
_LOGGER.debug("Thermostat %s - Receive new Power Max event", self.name)
_LOGGER.debug(event)
self.refresh_state()
await self.refresh_state()
@overrides
def refresh_state(self) -> bool:
async def refresh_state(self) -> bool:
"""Tries to get the last state from sensor
Returns True if a change has been made"""
ret = False
@@ -156,7 +152,7 @@ class CentralFeaturePowerManager(BaseFeatureManager):
else 999
)
if dtimestamp >= MIN_DTEMP_SECS:
self.calculate_shedding()
await self.calculate_shedding()
self._last_shedding_date = now
return ret
@@ -207,16 +203,19 @@ class CentralFeaturePowerManager(BaseFeatureManager):
if not vtherm.power_manager.is_overpowering_detected:
# To force all others vtherms to be in overpowering
force_overpowering = True
await vtherm.power_manager.set_overpowering(True)
await vtherm.power_manager.set_overpowering(
True, power_consumption_max
)
else:
total_affected_power += power_consumption_max
if vtherm.power_manager.is_overpowering_detected:
_LOGGER.debug(
"%s - vtherm %s should not be in overpowering state",
self,
vtherm.name,
)
await vtherm.power_manager.set_overpowering(False)
# Always set to false
# if vtherm.power_manager.is_overpowering_detected:
_LOGGER.debug(
"%s - vtherm %s should not be in overpowering state",
self,
vtherm.name,
)
await vtherm.power_manager.set_overpowering(False)
_LOGGER.debug(
"%s - after vtherm %s total_affected_power=%s, available_power=%s",
@@ -290,5 +289,15 @@ class CentralFeaturePowerManager(BaseFeatureManager):
"""Return the power temperature"""
return self._power_temp
@property
def power_sensor_entity_id(self) -> float | None:
"""Return the power sensor entity id"""
return self._power_sensor_entity_id
@property
def max_power_sensor_entity_id(self) -> float | None:
"""Return the max power sensor entity id"""
return self._max_power_sensor_entity_id
def __str__(self):
return "CentralPowerManager"

18
custom_components/versatile_thermostat/commons.py
View File

@@ -3,6 +3,7 @@
# pylint: disable=line-too-long
import logging
import warnings
from types import MappingProxyType
from typing import Any, TypeVar
@@ -132,3 +133,20 @@ def check_and_extract_service_configuration(service_config) -> dict:
"check_and_extract_service_configuration(%s) gives '%s'", service_config, ret
)
return ret
def deprecated(message):
"""A decorator to indicate that the method/attribut is deprecated"""
def decorator(func):
def wrapper(*args, **kwargs):
warnings.warn(
f"{func.__name__} is deprecated: {message}",
DeprecationWarning,
stacklevel=2,
)
return func(*args, **kwargs)
return wrapper
return decorator

6
custom_components/versatile_thermostat/config_schema.py
View File

@@ -339,6 +339,12 @@ STEP_CENTRAL_POWER_DATA_SCHEMA = vol.Schema( # pylint: disable=invalid-name
}
)
STEP_NON_CENTRAL_POWER_DATA_SCHEMA = vol.Schema( # pylint: disable=invalid-name
{
vol.Optional(CONF_PRESET_POWER, default="13"): vol.Coerce(float),
}
)
STEP_POWER_DATA_SCHEMA = vol.Schema( # pylint: disable=invalid-name
{
vol.Required(CONF_USE_POWER_CENTRAL_CONFIG, default=True): cv.boolean,

262
custom_components/versatile_thermostat/feature_power_manager.py
View File

@@ -12,22 +12,15 @@ from homeassistant.const import (
STATE_UNKNOWN,
)
from homeassistant.core import (
HomeAssistant,
callback,
Event,
)
from homeassistant.helpers.event import (
async_track_state_change_event,
EventStateChangedData,
)
from homeassistant.components.climate import HVACMode
from .const import * # pylint: disable=wildcard-import, unused-wildcard-import
from .commons import ConfigData
from .base_manager import BaseFeatureManager
from .vtherm_api import VersatileThermostatAPI
_LOGGER = logging.getLogger(__name__)
@@ -50,10 +43,6 @@ class FeaturePowerManager(BaseFeatureManager):
def __init__(self, vtherm: Any, hass: HomeAssistant):
"""Init of a featureManager"""
super().__init__(vtherm, hass)
self._power_sensor_entity_id = None
self._max_power_sensor_entity_id = None
self._current_power = None
self._current_max_power = None
self._power_temp = None
self._overpowering_state = STATE_UNAVAILABLE
self._is_configured: bool = False
@@ -64,20 +53,11 @@ class FeaturePowerManager(BaseFeatureManager):
"""Reinit of the manager"""
# Power management
self._power_sensor_entity_id = entry_infos.get(CONF_POWER_SENSOR)
self._max_power_sensor_entity_id = entry_infos.get(CONF_MAX_POWER_SENSOR)
self._power_temp = entry_infos.get(CONF_PRESET_POWER)
self._device_power = entry_infos.get(CONF_DEVICE_POWER) or 0
self._is_configured = False
self._current_power = None
self._current_max_power = None
if (
entry_infos.get(CONF_USE_POWER_FEATURE, False)
and self._max_power_sensor_entity_id
and self._power_sensor_entity_id
and self._device_power
):
if entry_infos.get(CONF_USE_POWER_FEATURE, False) and self._device_power:
self._is_configured = True
self._overpowering_state = STATE_UNKNOWN
else:
@@ -85,159 +65,54 @@ class FeaturePowerManager(BaseFeatureManager):
@overrides
def start_listening(self):
"""Start listening the underlying entity"""
if self._is_configured:
self.stop_listening()
else:
return
self.add_listener(
async_track_state_change_event(
self.hass,
[self._power_sensor_entity_id],
self._async_power_sensor_changed,
)
)
self.add_listener(
async_track_state_change_event(
self.hass,
[self._max_power_sensor_entity_id],
self._async_max_power_sensor_changed,
)
)
@overrides
async def refresh_state(self) -> bool:
"""Tries to get the last state from sensor
Returns True if a change has been made"""
ret = False
if self._is_configured:
# try to acquire current power and power max
current_power_state = self.hass.states.get(self._power_sensor_entity_id)
if current_power_state and current_power_state.state not in (
STATE_UNAVAILABLE,
STATE_UNKNOWN,
):
self._current_power = float(current_power_state.state)
_LOGGER.debug(
"%s - Current power have been retrieved: %.3f",
self,
self._current_power,
)
ret = True
# Try to acquire power max
current_power_max_state = self.hass.states.get(
self._max_power_sensor_entity_id
)
if current_power_max_state and current_power_max_state.state not in (
STATE_UNAVAILABLE,
STATE_UNKNOWN,
):
self._current_max_power = float(current_power_max_state.state)
_LOGGER.debug(
"%s - Current power max have been retrieved: %.3f",
self,
self._current_max_power,
)
ret = True
return ret
@callback
async def _async_power_sensor_changed(self, event: Event[EventStateChangedData]):
"""Handle power changes."""
_LOGGER.debug("Thermostat %s - Receive new Power event", self)
_LOGGER.debug(event)
new_state = event.data.get("new_state")
old_state = event.data.get("old_state")
if (
new_state is None
or new_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN)
or (old_state is not None and new_state.state == old_state.state)
):
return
try:
current_power = float(new_state.state)
if math.isnan(current_power) or math.isinf(current_power):
raise ValueError(f"Sensor has illegal state {new_state.state}")
self._current_power = current_power
if self._vtherm.preset_mode == PRESET_POWER:
await self._vtherm.async_control_heating()
except ValueError as ex:
_LOGGER.error("Unable to update current_power from sensor: %s", ex)
@callback
async def _async_max_power_sensor_changed(
self, event: Event[EventStateChangedData]
):
"""Handle power max changes."""
_LOGGER.debug("Thermostat %s - Receive new Power Max event", self.name)
_LOGGER.debug(event)
new_state = event.data.get("new_state")
old_state = event.data.get("old_state")
if (
new_state is None
or new_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN)
or (old_state is not None and new_state.state == old_state.state)
):
return
try:
current_power_max = float(new_state.state)
if math.isnan(current_power_max) or math.isinf(current_power_max):
raise ValueError(f"Sensor has illegal state {new_state.state}")
self._current_max_power = current_power_max
if self._vtherm.preset_mode == PRESET_POWER:
await self._vtherm.async_control_heating()
except ValueError as ex:
_LOGGER.error("Unable to update current_power from sensor: %s", ex)
"""Start listening the underlying entity. There is nothing to listen"""
def add_custom_attributes(self, extra_state_attributes: dict[str, Any]):
"""Add some custom attributes"""
vtherm_api = VersatileThermostatAPI.get_vtherm_api()
extra_state_attributes.update(
{
"power_sensor_entity_id": self._power_sensor_entity_id,
"max_power_sensor_entity_id": self._max_power_sensor_entity_id,
"power_sensor_entity_id": vtherm_api.central_power_manager.power_sensor_entity_id,
"max_power_sensor_entity_id": vtherm_api.central_power_manager.max_power_sensor_entity_id,
"overpowering_state": self._overpowering_state,
"is_power_configured": self._is_configured,
"device_power": self._device_power,
"power_temp": self._power_temp,
"current_power": self._current_power,
"current_max_power": self._current_max_power,
"current_power": vtherm_api.central_power_manager.current_power,
"current_max_power": vtherm_api.central_power_manager.current_max_power,
"mean_cycle_power": self.mean_cycle_power,
}
)
async def check_overpowering(self) -> bool:
"""Check the overpowering condition
Turn the preset_mode of the heater to 'power' if power conditions are exceeded
Returns True if overpowering is 'on'
async def check_power_available(self) -> bool:
"""Check if the Vtherm can be started considering overpowering.
Returns True if no overpowering conditions are found
"""
if not self._is_configured:
return False
vtherm_api = VersatileThermostatAPI.get_vtherm_api()
if (
self._current_power is None
not self._is_configured
or not vtherm_api.central_power_manager.is_configured
):
return True
current_power = vtherm_api.central_power_manager.current_power
current_max_power = vtherm_api.central_power_manager.current_max_power
if (
current_power is None
or current_max_power is None
or self._device_power is None
or self._current_max_power is None
):
_LOGGER.warning(
"%s - power not valued. check_overpowering not available", self
"%s - power not valued. check_power_available not available", self
)
return False
return True
_LOGGER.debug(
"%s - overpowering check: power=%.3f, max_power=%.3f heater power=%.3f",
self,
self._current_power,
self._current_max_power,
current_power,
current_max_power,
self._device_power,
)
@@ -253,18 +128,35 @@ class FeaturePowerManager(BaseFeatureManager):
self._device_power * self._vtherm.proportional_algorithm.on_percent,
)
ret = (self._current_power + power_consumption_max) >= self._current_max_power
if (
self._overpowering_state == STATE_OFF
and ret
and self._vtherm.hvac_mode != HVACMode.OFF
):
ret = (current_power + power_consumption_max) < current_max_power
if not ret:
_LOGGER.info(
"%s - there is not enough power available power=%.3f, max_power=%.3f heater power=%.3f",
self,
current_power,
current_max_power,
self._device_power,
)
return ret
async def set_overpowering(self, overpowering: bool, power_consumption_max=0):
"""Force the overpowering state for the VTherm"""
vtherm_api = VersatileThermostatAPI.get_vtherm_api()
current_power = vtherm_api.central_power_manager.current_power
current_max_power = vtherm_api.central_power_manager.current_max_power
if overpowering and not self.is_overpowering_detected:
_LOGGER.warning(
"%s - overpowering is detected. Heater preset will be set to 'power'",
self,
)
self._overpowering_state = STATE_ON
if self._vtherm.is_over_climate:
self._vtherm.save_hvac_mode()
self._vtherm.save_preset_mode()
await self._vtherm.async_underlying_entity_turn_off()
await self._vtherm.async_set_preset_mode_internal(PRESET_POWER)
@@ -272,24 +164,20 @@ class FeaturePowerManager(BaseFeatureManager):
EventType.POWER_EVENT,
{
"type": "start",
"current_power": self._current_power,
"current_power": current_power,
"device_power": self._device_power,
"current_max_power": self._current_max_power,
"current_max_power": current_max_power,
"current_power_consumption": power_consumption_max,
},
)
# Check if we need to remove the POWER preset
if (
self._overpowering_state == STATE_ON
and not ret
and self._vtherm.preset_mode == PRESET_POWER
):
elif not overpowering and self.is_overpowering_detected:
_LOGGER.warning(
"%s - end of overpowering is detected. Heater preset will be restored to '%s'",
self,
self._vtherm._saved_preset_mode, # pylint: disable=protected-access
)
self._overpowering_state = STATE_OFF
if self._vtherm.is_over_climate:
await self._vtherm.restore_hvac_mode(False)
await self._vtherm.restore_preset_mode()
@@ -297,22 +185,18 @@ class FeaturePowerManager(BaseFeatureManager):
EventType.POWER_EVENT,
{
"type": "end",
"current_power": self._current_power,
"current_power": current_power,
"device_power": self._device_power,
"current_max_power": self._current_max_power,
"current_max_power": current_max_power,
},
)
new_overpowering_state = STATE_ON if ret else STATE_OFF
if self._overpowering_state != new_overpowering_state:
self._overpowering_state = new_overpowering_state
self._vtherm.update_custom_attributes()
return self._overpowering_state == STATE_ON
async def set_overpowering(self, overpowering: bool):
"""Force the overpowering state for the VTherm"""
raise NotImplementedError
elif not overpowering and self._overpowering_state != STATE_OFF:
# just set to not overpowering the state which was not set
self._overpowering_state = STATE_OFF
else:
# Nothing to do (already in the right state)
return
self._vtherm.update_custom_attributes()
@overrides
@property
@@ -333,16 +217,6 @@ class FeaturePowerManager(BaseFeatureManager):
"""Return True if the Vtherm is in overpowering state"""
return self._overpowering_state == STATE_ON
@property
def max_power_sensor_entity_id(self) -> bool:
"""Return the power max entity id"""
return self._max_power_sensor_entity_id
@property
def power_sensor_entity_id(self) -> bool:
"""Return the power entity id"""
return self._power_sensor_entity_id
@property
def power_temperature(self) -> bool:
"""Return the power temperature"""
@@ -353,16 +227,6 @@ class FeaturePowerManager(BaseFeatureManager):
"""Return the device power"""
return self._device_power
@property
def current_power(self) -> bool:
"""Return the current power from sensor"""
return self._current_power
@property
def current_max_power(self) -> bool:
"""Return the current power from sensor"""
return self._current_max_power
@property
def mean_cycle_power(self) -> float | None:
"""Returns the mean power consumption during the cycle"""

7
custom_components/versatile_thermostat/underlyings.py
View File

@@ -409,9 +409,10 @@ class UnderlyingSwitch(UnderlyingEntity):
await self.turn_off()
return
if await self._thermostat.power_manager.check_overpowering():
_LOGGER.debug("%s - End of cycle (3)", self)
return
# if await self._thermostat.power_manager.check_overpowering():
# _LOGGER.debug("%s - End of cycle (3)", self)
# return
# safety mode could have change the on_time percent
await self._thermostat.safety_manager.refresh_state()
time = self._on_time_sec