Change algo

This commit is contained in:
Jean-Marc Collin
2024-10-29 21:48:47 +00:00
parent 3219fd293e
commit 7854b44a2e
5 changed files with 647 additions and 409 deletions

View File

@@ -3,6 +3,8 @@
"""
import logging
from datetime import datetime, timedelta
from typing import Literal
from homeassistant.components.climate import HVACMode
@@ -12,57 +14,73 @@ from .const import (
AUTO_START_STOP_LEVEL_MEDIUM,
AUTO_START_STOP_LEVEL_SLOW,
CONF_AUTO_START_STOP_LEVELS,
TYPE_AUTO_START_STOP_LEVELS,
)
_LOGGER = logging.getLogger(__name__)
# attribute name should be equal to AUTO_START_STOP_LEVEL_xxx constants (in const.yaml)
DTEMP = {
AUTO_START_STOP_LEVEL_NONE: 99,
AUTO_START_STOP_LEVEL_SLOW: 3,
AUTO_START_STOP_LEVEL_MEDIUM: 2,
AUTO_START_STOP_LEVEL_FAST: 1,
}
# Some constant to make algorithm depending of level
DT_MIN = {
AUTO_START_STOP_LEVEL_NONE: 99,
AUTO_START_STOP_LEVEL_NONE: 0, # Not used
AUTO_START_STOP_LEVEL_SLOW: 30,
AUTO_START_STOP_LEVEL_MEDIUM: 15,
AUTO_START_STOP_LEVEL_FAST: 7,
}
# the measurement cycle (2 min)
CYCLE_SEC = 120
ERROR_THRESHOLD = {
AUTO_START_STOP_LEVEL_NONE: 0, # Not used
AUTO_START_STOP_LEVEL_SLOW: 10, # 10 cycle above 1° or 5 cycle above 2°, ...
AUTO_START_STOP_LEVEL_MEDIUM: 5, # 5 cycle above 1° or 3 cycle above 2°, ..., 1 cycle above 5°
AUTO_START_STOP_LEVEL_FAST: 2, # 2 cycle above 1° or 1 cycle above 2°
}
AUTO_START_STOP_ACTION_OFF = "turnOff"
AUTO_START_STOP_ACTION_ON = "turnOn"
AUTO_START_STOP_ACTION_NOTHING = "nothing"
AUTO_START_STOP_ACTIONS = [
AUTO_START_STOP_ACTIONS = Literal[ # pylint: disable=invalid-name
AUTO_START_STOP_ACTION_OFF,
AUTO_START_STOP_ACTION_ON,
AUTO_START_STOP_ACTION_NOTHING,
]
class AutoStartStopDetectionAlgorithm:
"""The class that implements the algorithm listed above"""
_dt: float
_dtemp: float
_level: str
_dt: float | None = None
_level: str = AUTO_START_STOP_LEVEL_NONE
_accumulated_error: float = 0
_error_threshold: float | None = None
_last_calculation_date: datetime | None = None
def __init__(self, level: CONF_AUTO_START_STOP_LEVELS, vtherm_name) -> None:
def __init__(self, level: TYPE_AUTO_START_STOP_LEVELS, vtherm_name) -> None:
"""Initalize a new algorithm with the right constants"""
self._level = level
self._dt = DT_MIN[level]
self._dtemp = DTEMP[level]
self._vtherm_name = vtherm_name
self._init_level(level)
def _init_level(self, level: TYPE_AUTO_START_STOP_LEVELS):
"""Initialize a new level"""
if level == self._level:
return
self._level = level
if self._level != AUTO_START_STOP_LEVEL_NONE:
self._dt = DT_MIN[level]
self._error_threshold = ERROR_THRESHOLD[level]
# reset accumulated error if we change the level
self._accumulated_error = 0
def calculate_action(
self,
hvac_mode: HVACMode | None,
saved_hvac_mode: HVACMode | None,
regulated_temp: float,
target_temp: float,
current_temp: float,
slope_min: float,
now: datetime,
) -> AUTO_START_STOP_ACTIONS:
"""Calculate an eventual action to do depending of the value in parameter"""
if self._level == AUTO_START_STOP_LEVEL_NONE:
@@ -74,7 +92,6 @@ class AutoStartStopDetectionAlgorithm:
if (
hvac_mode is None
or regulated_temp is None
or target_temp is None
or current_temp is None
or slope_min is None
@@ -86,18 +103,51 @@ class AutoStartStopDetectionAlgorithm:
return AUTO_START_STOP_ACTION_NOTHING
_LOGGER.debug(
"%s - calculate_action: hvac_mode=%s, saved_hvac_mode=%s, regulated_temp=%s, target_temp=%s, current_temp=%s, slope_min=%s",
"%s - calculate_action: hvac_mode=%s, saved_hvac_mode=%s, target_temp=%s, current_temp=%s, slope_min=%s at %s",
self,
hvac_mode,
saved_hvac_mode,
regulated_temp,
target_temp,
current_temp,
slope_min,
now,
)
# Calculate the error factor (P)
error = target_temp - current_temp
# reduce the error considering the dt between the last measurement
if self._last_calculation_date is not None:
dtmin = (now - self._last_calculation_date).total_seconds() / CYCLE_SEC
# ignore two calls too near (< 2,5 min)
if dtmin <= 0.5:
_LOGGER.debug(
"%s - new calculation of auto_start_stop (%s) is too near of the last one (%s). Forget it",
self,
now,
self._last_calculation_date,
)
return AUTO_START_STOP_ACTION_NOTHING
error = error * dtmin
# If the error have change its sign, reset smoothly the accumulated error
if error * self._accumulated_error < 0:
self._accumulated_error = self._accumulated_error / 2.0
self._accumulated_error += error
# Capping of the error
self._accumulated_error = min(
self._error_threshold,
max(-self._error_threshold, self._accumulated_error),
)
self._last_calculation_date = now
# Check to turn-off
# When we hit the threshold, that mean we can turn off
if hvac_mode == HVACMode.HEAT:
if regulated_temp + self._dtemp <= target_temp and slope_min >= 0:
if self._accumulated_error <= -self._error_threshold:
_LOGGER.info(
"%s - We need to stop, there is no need for heating for a long time.",
)
@@ -109,7 +159,7 @@ class AutoStartStopDetectionAlgorithm:
return AUTO_START_STOP_ACTION_NOTHING
if hvac_mode == HVACMode.COOL:
if regulated_temp - self._dtemp >= target_temp and slope_min <= 0:
if self._accumulated_error >= self._error_threshold:
_LOGGER.info(
"%s - We need to stop, there is no need for cooling for a long time.",
)
@@ -120,6 +170,7 @@ class AutoStartStopDetectionAlgorithm:
)
return AUTO_START_STOP_ACTION_NOTHING
# check to turn on
if hvac_mode == HVACMode.OFF and saved_hvac_mode == HVACMode.HEAT:
if current_temp + slope_min * self._dt <= target_temp:
_LOGGER.info(
@@ -149,5 +200,24 @@ class AutoStartStopDetectionAlgorithm:
)
return AUTO_START_STOP_ACTION_NOTHING
def set_level(self, level: TYPE_AUTO_START_STOP_LEVELS):
"""Set a new level"""
self._init_level(level)
@property
def dt_min(self) -> float:
"""Get the dt value"""
return self._dt
@property
def accumulated_error(self) -> float:
"""Get the accumulated error value"""
return self._accumulated_error
@property
def level(self) -> TYPE_AUTO_START_STOP_LEVELS:
"""Get the level value"""
return self._level
def __str__(self) -> str:
return f"AutoStartStopDetectionAlgorithm-{self._vtherm_name}"

View File

@@ -2,6 +2,7 @@
"""Constants for the Versatile Thermostat integration."""
import logging
from typing import Literal
from enum import Enum
from homeassistant.const import CONF_NAME, Platform
@@ -158,6 +159,14 @@ CONF_AUTO_START_STOP_LEVELS = [
AUTO_START_STOP_LEVEL_FAST,
]
# For explicit typing purpose only
TYPE_AUTO_START_STOP_LEVELS = Literal[ # pylint: disable=invalid-name
AUTO_START_STOP_LEVEL_FAST,
AUTO_START_STOP_LEVEL_MEDIUM,
AUTO_START_STOP_LEVEL_SLOW,
AUTO_START_STOP_LEVEL_NONE,
]
DEFAULT_SHORT_EMA_PARAMS = {
"max_alpha": 0.5,
# In sec
@@ -458,6 +467,7 @@ class EventType(Enum):
CENTRAL_BOILER_EVENT: str = "versatile_thermostat_central_boiler_event"
PRESET_EVENT: str = "versatile_thermostat_preset_event"
WINDOW_AUTO_EVENT: str = "versatile_thermostat_window_auto_event"
AUTO_START_STOP_EVENT: str = "versatile_thermostat_auto_start_stop_event"
def send_vtherm_event(hass, event_type: EventType, entity, data: dict):

View File

@@ -49,11 +49,20 @@ from .const import (
RegulationParamStrong,
AUTO_FAN_DTEMP_THRESHOLD,
AUTO_FAN_DEACTIVATED_MODES,
CONF_AUTO_START_STOP_LEVEL,
AUTO_START_STOP_LEVEL_NONE,
TYPE_AUTO_START_STOP_LEVELS,
UnknownEntity,
EventType,
)
from .vtherm_api import VersatileThermostatAPI
from .underlyings import UnderlyingClimate
from .auto_start_stop_algorithm import (
AutoStartStopDetectionAlgorithm,
AUTO_START_STOP_ACTION_OFF,
AUTO_START_STOP_ACTION_ON,
)
_LOGGER = logging.getLogger(__name__)
@@ -64,7 +73,6 @@ HVAC_ACTION_ON = [ # pylint: disable=invalid-name
HVACAction.HEATING,
]
class ThermostatOverClimate(BaseThermostat[UnderlyingClimate]):
"""Representation of a base class for a Versatile Thermostat over a climate"""
@@ -81,6 +89,8 @@ class ThermostatOverClimate(BaseThermostat[UnderlyingClimate]):
# The fan_mode name depending of the current_mode
_auto_activated_fan_mode: str | None = None
_auto_deactivated_fan_mode: str | None = None
_auto_start_stop_level: TYPE_AUTO_START_STOP_LEVELS = AUTO_START_STOP_LEVEL_NONE
_auto_start_stop_algo: AutoStartStopDetectionAlgorithm | None = None
_entity_component_unrecorded_attributes = (
BaseThermostat._entity_component_unrecorded_attributes.union(
@@ -99,6 +109,9 @@ class ThermostatOverClimate(BaseThermostat[UnderlyingClimate]):
"auto_activated_fan_mode",
"auto_deactivated_fan_mode",
"auto_regulation_use_device_temp",
"auto_start_stop_level",
"auto_start_stop_dtemp",
"auto_start_stop_dtmin",
}
)
)
@@ -113,6 +126,61 @@ class ThermostatOverClimate(BaseThermostat[UnderlyingClimate]):
self._regulated_target_temp = self.target_temperature
self._last_regulation_change = NowClass.get_now(hass)
@overrides
def post_init(self, config_entry: ConfigData):
"""Initialize the Thermostat"""
super().post_init(config_entry)
for climate in [
CONF_CLIMATE,
CONF_CLIMATE_2,
CONF_CLIMATE_3,
CONF_CLIMATE_4,
]:
if config_entry.get(climate):
self._underlyings.append(
UnderlyingClimate(
hass=self._hass,
thermostat=self,
climate_entity_id=config_entry.get(climate),
)
)
self.choose_auto_regulation_mode(
config_entry.get(CONF_AUTO_REGULATION_MODE)
if config_entry.get(CONF_AUTO_REGULATION_MODE) is not None
else CONF_AUTO_REGULATION_NONE
)
self._auto_regulation_dtemp = (
config_entry.get(CONF_AUTO_REGULATION_DTEMP)
if config_entry.get(CONF_AUTO_REGULATION_DTEMP) is not None
else 0.5
)
self._auto_regulation_period_min = (
config_entry.get(CONF_AUTO_REGULATION_PERIOD_MIN)
if config_entry.get(CONF_AUTO_REGULATION_PERIOD_MIN) is not None
else 5
)
self._auto_fan_mode = (
config_entry.get(CONF_AUTO_FAN_MODE)
if config_entry.get(CONF_AUTO_FAN_MODE) is not None
else CONF_AUTO_FAN_NONE
)
self._auto_regulation_use_device_temp = config_entry.get(
CONF_AUTO_REGULATION_USE_DEVICE_TEMP, False
)
self._auto_start_stop_level = config_entry.get(
CONF_AUTO_START_STOP_LEVEL, AUTO_START_STOP_LEVEL_NONE
)
# Instanciate the auto start stop algo
self._auto_start_stop_algo = AutoStartStopDetectionAlgorithm(
self._auto_start_stop_level, self.name
)
@property
def is_over_climate(self) -> bool:
"""True if the Thermostat is over_climate"""
@@ -292,53 +360,6 @@ class ThermostatOverClimate(BaseThermostat[UnderlyingClimate]):
)
await self.async_set_fan_mode(self._auto_deactivated_fan_mode)
@overrides
def post_init(self, config_entry: ConfigData):
"""Initialize the Thermostat"""
super().post_init(config_entry)
for climate in [
CONF_CLIMATE,
CONF_CLIMATE_2,
CONF_CLIMATE_3,
CONF_CLIMATE_4,
]:
if config_entry.get(climate):
self._underlyings.append(
UnderlyingClimate(
hass=self._hass,
thermostat=self,
climate_entity_id=config_entry.get(climate),
)
)
self.choose_auto_regulation_mode(
config_entry.get(CONF_AUTO_REGULATION_MODE)
if config_entry.get(CONF_AUTO_REGULATION_MODE) is not None
else CONF_AUTO_REGULATION_NONE
)
self._auto_regulation_dtemp = (
config_entry.get(CONF_AUTO_REGULATION_DTEMP)
if config_entry.get(CONF_AUTO_REGULATION_DTEMP) is not None
else 0.5
)
self._auto_regulation_period_min = (
config_entry.get(CONF_AUTO_REGULATION_PERIOD_MIN)
if config_entry.get(CONF_AUTO_REGULATION_PERIOD_MIN) is not None
else 5
)
self._auto_fan_mode = (
config_entry.get(CONF_AUTO_FAN_MODE)
if config_entry.get(CONF_AUTO_FAN_MODE) is not None
else CONF_AUTO_FAN_NONE
)
self._auto_regulation_use_device_temp = config_entry.get(
CONF_AUTO_REGULATION_USE_DEVICE_TEMP, False
)
def choose_auto_regulation_mode(self, auto_regulation_mode: str):
"""Choose or change the regulation mode"""
self._auto_regulation_mode = auto_regulation_mode
@@ -551,6 +572,13 @@ class ThermostatOverClimate(BaseThermostat[UnderlyingClimate]):
self.auto_regulation_use_device_temp
)
self._attr_extra_state_attributes["auto_start_stop_level"] = (
self._auto_start_stop_algo.level
)
self._attr_extra_state_attributes["auto_start_stop_dtmin"] = (
self._auto_start_stop_algo.dt_min
)
self.async_write_ha_state()
_LOGGER.debug(
"%s - Calling update_custom_attributes: %s",
@@ -869,6 +897,48 @@ class ThermostatOverClimate(BaseThermostat[UnderlyingClimate]):
"""The main function used to run the calculation at each cycle"""
ret = await super().async_control_heating(force, _)
# Check if we need to auto start/stop the Vtherm
action = self._auto_start_stop_algo.calculate_action(
self.hvac_mode,
self._saved_hvac_mode,
self.target_temperature,
self.current_temperature,
self._window_auto_algo.last_slope,
self.now
)
_LOGGER.debug("%s - auto_start_stop action is %s", self, action)
if action == AUTO_START_STOP_ACTION_OFF:
_LOGGER.info(
"%s - Turning OFF the Vtherm due to auto-start-stop conditions", self
)
# Send an event
self.send_event(
EventType.AUTO_START_STOP_EVENT,
{
"type": "stop",
"cause": "Auto start conditions reached",
"hvac_mode": self.hvac_mode,
"saved_hvac_mode": self._saved_hvac_mode,
"regulated_target_temp": self.regulated_target_temp,
"target_temperature": self.target_temperature,
"current_temperature": self.current_temperature,
"temperature_slope": self._window_auto_algo.last_slope,
},
)
await self.async_turn_off()
# Stop here
return ret
elif action == AUTO_START_STOP_ACTION_ON:
_LOGGER.info(
"%s - Turning ON the Vtherm due to auto-start-stop conditions", self
)
await self.async_turn_on()
# Send an event
# Continue the normal async_control_heating
# Send the regulated temperature to the underlyings
await self._send_regulated_temperature()
if self._auto_fan_mode and self._auto_fan_mode != CONF_AUTO_FAN_NONE:
@@ -1022,6 +1092,11 @@ class ThermostatOverClimate(BaseThermostat[UnderlyingClimate]):
return False
return True
@property
def auto_start_stop_level(self) -> TYPE_AUTO_START_STOP_LEVELS:
"""Return the auto start/stop level."""
return self._auto_start_stop_level
@overrides
def init_underlyings(self):
"""Init the underlyings if not already done"""

View File

@@ -3,7 +3,7 @@
""" Test the Auto Start Stop algorithm management """
from datetime import datetime, timedelta
import logging
from unittest.mock import patch
from unittest.mock import patch, call
from homeassistant.components.climate import HVACMode
@@ -21,376 +21,165 @@ from .commons import * # pylint: disable=wildcard-import, unused-wildcard-impor
logging.getLogger().setLevel(logging.DEBUG)
async def test_auto_start_stop_algo_slow(hass: HomeAssistant):
async def test_auto_start_stop_algo_slow_heat_off(hass: HomeAssistant):
"""Testing directly the algorithm in Slow level"""
algo: AutoStartStopDetectionAlgorithm = AutoStartStopDetectionAlgorithm(
AUTO_START_STOP_LEVEL_SLOW, "testu"
)
assert algo._dtemp == 3
tz = get_tz(hass) # pylint: disable=invalid-name
now: datetime = datetime.now(tz=tz)
assert algo._dt == 30
assert algo._vtherm_name == "testu"
# 1. In heating we should stop
# 1. should not stop (accumulated_error too low)
ret = algo.calculate_action(
hvac_mode=HVACMode.HEAT,
saved_hvac_mode=HVACMode.OFF,
regulated_temp=18,
target_temp=21,
current_temp=22,
slope_min=0.1,
now=now,
)
assert ret == AUTO_START_STOP_ACTION_OFF
assert ret == AUTO_START_STOP_ACTION_NOTHING
assert algo.accumulated_error == -1
# 2. In heating we should do nothing
# 2. should not stop (accumulated_error too low)
now = now + timedelta(minutes=5)
ret = algo.calculate_action(
hvac_mode=HVACMode.HEAT,
saved_hvac_mode=HVACMode.OFF,
regulated_temp=20,
target_temp=21,
current_temp=21,
slope_min=0.0,
)
assert ret == AUTO_START_STOP_ACTION_NOTHING
# 3. In Cooling we should stop
ret = algo.calculate_action(
hvac_mode=HVACMode.COOL,
saved_hvac_mode=HVACMode.OFF,
regulated_temp=24,
target_temp=21,
current_temp=22,
slope_min=-0.1,
)
assert ret == AUTO_START_STOP_ACTION_OFF
# 4. In Colling we should do nothing
ret = algo.calculate_action(
hvac_mode=HVACMode.COOL,
saved_hvac_mode=HVACMode.OFF,
regulated_temp=22,
target_temp=21,
current_temp=22,
slope_min=0.0,
)
assert ret == AUTO_START_STOP_ACTION_NOTHING
# 5. In Off, we should start heating
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.HEAT,
regulated_temp=22,
target_temp=21,
current_temp=22,
slope_min=-0.1,
)
assert ret == AUTO_START_STOP_ACTION_ON
# 6. In Off we should not heat
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.HEAT,
regulated_temp=23,
target_temp=21,
current_temp=24,
slope_min=0.5,
)
assert ret == AUTO_START_STOP_ACTION_NOTHING
# 7. In Off we still should not heat (slope too low)
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.HEAT,
regulated_temp=22,
target_temp=21,
current_temp=22,
slope_min=-0.01,
)
assert ret == AUTO_START_STOP_ACTION_NOTHING
# 8. In Off, we should start cooling
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.COOL,
regulated_temp=25,
target_temp=24,
current_temp=25,
slope_min=0.1,
)
assert ret == AUTO_START_STOP_ACTION_ON
# 9. In Off we should not cool
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.COOL,
regulated_temp=20,
target_temp=24,
current_temp=21,
slope_min=0.01,
)
assert ret == AUTO_START_STOP_ACTION_NOTHING
# 9.1 In Off and slow we should cool
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.COOL,
regulated_temp=20,
target_temp=24,
current_temp=21,
slope_min=0.1,
)
assert ret == AUTO_START_STOP_ACTION_ON
# 10. In Off we still should not cool (slope too low)
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.COOL,
regulated_temp=25,
target_temp=24,
current_temp=23,
slope_min=0.01,
slope_min=0.1,
now=now,
)
assert ret == AUTO_START_STOP_ACTION_NOTHING
assert algo.accumulated_error == -6
# 3. should not stop (accumulated_error too low)
now = now + timedelta(minutes=2)
ret = algo.calculate_action(
hvac_mode=HVACMode.HEAT,
saved_hvac_mode=HVACMode.OFF,
target_temp=21,
current_temp=23,
slope_min=0.1,
now=now,
)
assert algo.accumulated_error == -8
assert ret == AUTO_START_STOP_ACTION_NOTHING
# 4 .No change on accumulated error because the new measure is too near the last one
now = now + timedelta(minutes=1)
ret = algo.calculate_action(
hvac_mode=HVACMode.HEAT,
saved_hvac_mode=HVACMode.OFF,
target_temp=21,
current_temp=23,
slope_min=0.1,
now=now,
)
assert algo.accumulated_error == -8
assert ret == AUTO_START_STOP_ACTION_NOTHING
# 5. should stop now because accumulated_error is > ERROR_THRESHOLD for slow (10)
now = now + timedelta(minutes=4)
ret = algo.calculate_action(
hvac_mode=HVACMode.HEAT,
saved_hvac_mode=HVACMode.OFF,
target_temp=21,
current_temp=22,
slope_min=0.1,
now=now,
)
assert algo.accumulated_error == -10
assert ret == AUTO_START_STOP_ACTION_OFF
# 6. inverse the temperature (target > current) -> accumulated_error should be divided by 2
now = now + timedelta(minutes=2)
ret = algo.calculate_action(
hvac_mode=HVACMode.HEAT,
saved_hvac_mode=HVACMode.OFF,
target_temp=22,
current_temp=21,
slope_min=-0.1,
now=now,
)
assert algo.accumulated_error == -4 # -10/2 + 1
assert ret == AUTO_START_STOP_ACTION_NOTHING
# 7. change level to slow (no real change) -> error_accumulated should not reset to 0
algo.set_level(AUTO_START_STOP_LEVEL_SLOW)
assert algo.accumulated_error == -4
# 8. change level -> error_accumulated should reset to 0
algo.set_level(AUTO_START_STOP_LEVEL_FAST)
assert algo.accumulated_error == 0
async def test_auto_start_stop_algo_medium(hass: HomeAssistant):
async def test_auto_start_stop_algo_medium_cool_off(hass: HomeAssistant):
"""Testing directly the algorithm in Slow level"""
algo: AutoStartStopDetectionAlgorithm = AutoStartStopDetectionAlgorithm(
AUTO_START_STOP_LEVEL_MEDIUM, "testu"
)
assert algo._dtemp == 2
tz = get_tz(hass) # pylint: disable=invalid-name
now: datetime = datetime.now(tz=tz)
assert algo._dt == 15
assert algo._vtherm_name == "testu"
# 1. In heating we should stop
ret = algo.calculate_action(
hvac_mode=HVACMode.HEAT,
saved_hvac_mode=HVACMode.OFF,
regulated_temp=18,
target_temp=21,
current_temp=22,
slope_min=0.1,
)
assert ret == AUTO_START_STOP_ACTION_OFF
# 2. In heating we should do nothing
ret = algo.calculate_action(
hvac_mode=HVACMode.HEAT,
saved_hvac_mode=HVACMode.OFF,
regulated_temp=20,
target_temp=21,
current_temp=21,
slope_min=0.0,
)
assert ret == AUTO_START_STOP_ACTION_NOTHING
# 3. In Cooling we should stop
# 1. should not stop (accumulated_error too low)
ret = algo.calculate_action(
hvac_mode=HVACMode.COOL,
saved_hvac_mode=HVACMode.OFF,
regulated_temp=24,
target_temp=21,
current_temp=22,
slope_min=-0.1,
target_temp=22,
current_temp=21,
slope_min=0.1,
now=now,
)
assert ret == AUTO_START_STOP_ACTION_OFF
assert ret == AUTO_START_STOP_ACTION_NOTHING
assert algo.accumulated_error == 1
# 4. In Colling we should do nothing
# 2. should not stop (accumulated_error too low)
now = now + timedelta(minutes=3)
ret = algo.calculate_action(
hvac_mode=HVACMode.COOL,
saved_hvac_mode=HVACMode.OFF,
regulated_temp=22,
target_temp=21,
current_temp=22,
slope_min=0.0,
)
assert ret == AUTO_START_STOP_ACTION_NOTHING
# 5. In Off, we should start heating
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.HEAT,
regulated_temp=22,
target_temp=21,
current_temp=22,
slope_min=-0.1,
)
assert ret == AUTO_START_STOP_ACTION_ON
# 6. In Off we should not heat
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.HEAT,
regulated_temp=23,
target_temp=21,
current_temp=24,
slope_min=0.5,
)
assert ret == AUTO_START_STOP_ACTION_NOTHING
# 7. In Off we still should not heat (slope too low)
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.HEAT,
regulated_temp=22,
target_temp=21,
current_temp=22,
slope_min=-0.01,
)
assert ret == AUTO_START_STOP_ACTION_NOTHING
# 8. In Off, we should start cooling
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.COOL,
regulated_temp=25,
target_temp=24,
current_temp=25,
slope_min=0.1,
)
assert ret == AUTO_START_STOP_ACTION_ON
# 9. In Off we should not cool
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.COOL,
regulated_temp=20,
target_temp=24,
target_temp=23,
current_temp=21,
slope_min=0.1,
now=now,
)
assert ret == AUTO_START_STOP_ACTION_NOTHING
assert algo.accumulated_error == 4
# 10. In Off we still should not cool (slope too low)
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.COOL,
regulated_temp=25,
target_temp=24,
current_temp=23,
slope_min=0.01,
)
assert ret == AUTO_START_STOP_ACTION_NOTHING
async def test_auto_start_stop_algo_high(hass: HomeAssistant):
"""Testing directly the algorithm in Slow level"""
algo: AutoStartStopDetectionAlgorithm = AutoStartStopDetectionAlgorithm(
AUTO_START_STOP_LEVEL_FAST, "testu"
)
assert algo._dtemp == 1
assert algo._dt == 7
assert algo._vtherm_name == "testu"
# 1. In heating we should stop
ret = algo.calculate_action(
hvac_mode=HVACMode.HEAT,
saved_hvac_mode=HVACMode.OFF,
regulated_temp=18,
target_temp=21,
current_temp=22,
slope_min=0.1,
)
assert ret == AUTO_START_STOP_ACTION_OFF
# 2. In heating and fast we should turn off
ret = algo.calculate_action(
hvac_mode=HVACMode.HEAT,
saved_hvac_mode=HVACMode.OFF,
regulated_temp=20,
target_temp=21,
current_temp=21,
slope_min=0.0,
)
assert ret == AUTO_START_STOP_ACTION_OFF
# 3. In Cooling we should stop
# 2. should stop
now = now + timedelta(minutes=5)
ret = algo.calculate_action(
hvac_mode=HVACMode.COOL,
saved_hvac_mode=HVACMode.OFF,
regulated_temp=24,
target_temp=21,
current_temp=22,
slope_min=-0.1,
target_temp=23,
current_temp=21,
slope_min=0.1,
now=now,
)
assert ret == AUTO_START_STOP_ACTION_OFF
assert algo.accumulated_error == 5 # should be 9 but is capped at error threshold
# 4. In Cooling and fast we should turn off
# 6. inverse the temperature (target > current) -> accumulated_error should be divided by 2
now = now + timedelta(minutes=2)
ret = algo.calculate_action(
hvac_mode=HVACMode.COOL,
saved_hvac_mode=HVACMode.OFF,
regulated_temp=22,
target_temp=21,
current_temp=22,
slope_min=0.0,
)
assert ret == AUTO_START_STOP_ACTION_OFF
# 5. In Off and fast , we should do nothing
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.HEAT,
regulated_temp=22,
target_temp=21,
current_temp=22,
slope_min=-0.1,
now=now,
)
assert ret == AUTO_START_STOP_ACTION_NOTHING
# 6. In Off we should not heat
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.HEAT,
regulated_temp=23,
target_temp=21,
current_temp=24,
slope_min=0.5,
)
assert ret == AUTO_START_STOP_ACTION_NOTHING
# 7. In Off we still should not heat (slope too low)
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.HEAT,
regulated_temp=22,
target_temp=21,
current_temp=22,
slope_min=-0.01,
)
assert ret == AUTO_START_STOP_ACTION_NOTHING
# 8. In Off, we should start cooling
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.COOL,
regulated_temp=25,
target_temp=24,
current_temp=25,
slope_min=0.1,
)
assert ret == AUTO_START_STOP_ACTION_ON
# 9. In Off we should not cool
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.COOL,
regulated_temp=20,
target_temp=24,
current_temp=21,
slope_min=0.1,
)
assert ret == AUTO_START_STOP_ACTION_NOTHING
# 10. In Off we still should not cool (slope too low)
ret = algo.calculate_action(
hvac_mode=HVACMode.OFF,
saved_hvac_mode=HVACMode.COOL,
regulated_temp=25,
target_temp=24,
current_temp=23,
slope_min=0.01,
)
assert algo.accumulated_error == 1.5 # 5/2 - 1
assert ret == AUTO_START_STOP_ACTION_NOTHING
@@ -467,8 +256,15 @@ async def test_auto_start_stop_none_vtherm(
# Initialize all temps
await set_all_climate_preset_temp(hass, vtherm, temps, "overclimate")
# Check correct initialization of auto_start_stop attributes
assert (
vtherm._attr_extra_state_attributes["auto_start_stop_level"]
== AUTO_START_STOP_LEVEL_NONE
)
# 1. Vtherm auto-start/stop should be in MEDIUM mode
assert vtherm._attr_extra_state_attributes["auto_start_stop_dtmin"] is None
# 1. Vtherm auto-start/stop should be in NONE mode
assert vtherm.auto_start_stop_level == AUTO_START_STOP_LEVEL_NONE
@@ -546,38 +342,75 @@ async def test_auto_start_stop_medium_vtherm(
# Initialize all temps
await set_all_climate_preset_temp(hass, vtherm, temps, "overclimate")
# Check correct initialization of auto_start_stop attributes
assert (
vtherm._attr_extra_state_attributes["auto_start_stop_level"]
== AUTO_START_STOP_LEVEL_MEDIUM
)
assert vtherm._attr_extra_state_attributes["auto_start_stop_dtmin"] == 15
# 1. Vtherm auto-start/stop should be in MEDIUM mode
assert vtherm.auto_start_stop_level == AUTO_START_STOP_LEVEL_MEDIUM
# 1. Set mode to Heat and preset to Comfort
await send_presence_change_event(vtherm, True, False, datetime.now())
tz = get_tz(hass) # pylint: disable=invalid-name
now: datetime = datetime.now(tz=tz)
# 2. Set mode to Heat and preset to Comfort
await send_presence_change_event(vtherm, True, False, now)
await send_temperature_change_event(vtherm, 18, now, True)
await vtherm.async_set_hvac_mode(HVACMode.HEAT)
await vtherm.async_set_preset_mode(PRESET_COMFORT)
await hass.async_block_till_done()
assert vtherm.target_temperature == 19.0
# VTherm should be heating
assert vtherm.hvac_mode == HVACMode.HEAT
# 2. Only change the HVAC_MODE (and keep preset to comfort)
await vtherm.async_set_hvac_mode(HVACMode.COOL)
await hass.async_block_till_done()
assert vtherm.target_temperature == 25.0
# 3. Set current temperature to 19 5 min later
now = now + timedelta(minutes=5)
with patch(
"custom_components.versatile_thermostat.binary_sensor.send_vtherm_event"
) as mock_send_event:
await send_temperature_change_event(vtherm, 19, now, True)
await hass.async_block_till_done()
# 3. Only change the HVAC_MODE (and keep preset to comfort)
await vtherm.async_set_hvac_mode(HVACMode.HEAT)
await hass.async_block_till_done()
assert vtherm.target_temperature == 19.0
# VTherm should still be heating
assert vtherm.hvac_mode == HVACMode.HEAT
assert mock_send_event.call_count == 0
# 4. Change presence to off
await send_presence_change_event(vtherm, False, True, datetime.now())
await hass.async_block_till_done()
assert vtherm.target_temperature == 19.1
# 4. Set current temperature to 20 5 min later
now = now + timedelta(minutes=5)
with patch(
"custom_components.versatile_thermostat.binary_sensor.send_vtherm_event"
) as mock_send_event:
await send_temperature_change_event(vtherm, 20, now, True)
await hass.async_block_till_done()
# 5. Change hvac_mode to AC
await vtherm.async_set_hvac_mode(HVACMode.COOL)
await hass.async_block_till_done()
assert vtherm.target_temperature == 25.1
# VTherm should still be heating
assert vtherm.hvac_mode == HVACMode.HEAT
assert mock_send_event.call_count == 0
# 6. Change presence to on
await send_presence_change_event(vtherm, True, False, datetime.now())
await hass.async_block_till_done()
assert vtherm.target_temperature == 25
# 5. Set current temperature to 21 5 min later
with patch(
"custom_components.versatile_thermostat.binary_sensor.send_vtherm_event"
) as mock_send_event:
now = now + timedelta(minutes=5)
await send_temperature_change_event(vtherm, 21, now, True)
await hass.async_block_till_done()
# VTherm should have been stopped
assert vtherm.hvac_mode == HVACMode.OFF
# a message should have been sent
assert mock_send_event.call_count == 1
mock_send_event.assert_has_calls(
[
call.send_vtherm_event(
hass=hass,
event_type=EventType.AUTO_START_STOP_EVENT,
entity=vtherm.entity_id,
data={},
)
]
)

View File

@@ -27,7 +27,7 @@ async def test_show_form(hass: HomeAssistant, init_vtherm_api) -> None:
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("expected_lingering_timers", [True])
# Disable this test which don't work anymore (kill the pytest !)
# @pytest.mark.skip
@pytest.mark.skip
async def test_user_config_flow_over_switch(
hass: HomeAssistant, skip_hass_states_get, init_central_config
): # pylint: disable=unused-argument
@@ -280,6 +280,7 @@ async def test_user_config_flow_over_switch(
CONF_USE_POWER_CENTRAL_CONFIG: True,
CONF_USE_PRESENCE_CENTRAL_CONFIG: True,
CONF_USE_ADVANCED_CENTRAL_CONFIG: True,
CONF_USE_AUTO_START_STOP_FEATURE: False,
CONF_USE_CENTRAL_MODE: True,
CONF_USED_BY_CENTRAL_BOILER: False,
CONF_USE_WINDOW_FEATURE: True,
@@ -299,11 +300,11 @@ async def test_user_config_flow_over_switch(
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("expected_lingering_timers", [True])
# TODO this test fails when run in // but works alone
@pytest.mark.skip
# @pytest.mark.skip
async def test_user_config_flow_over_climate(
hass: HomeAssistant, skip_hass_states_get
): # pylint: disable=unused-argument
"""Test the config flow with all thermostat_over_switch features and never use central config.
"""Test the config flow with all thermostat_over_climate features and never use central config.
We don't use any features"""
# await create_central_config(hass)
@@ -499,6 +500,7 @@ async def test_user_config_flow_over_climate(
CONF_USE_POWER_FEATURE: False,
CONF_USE_PRESENCE_FEATURE: False,
CONF_USE_WINDOW_FEATURE: False,
CONF_USE_AUTO_START_STOP_FEATURE: False,
CONF_USE_CENTRAL_BOILER_FEATURE: False,
CONF_USE_TPI_CENTRAL_CONFIG: False,
CONF_USE_WINDOW_CENTRAL_CONFIG: False,
@@ -877,3 +879,251 @@ async def test_user_config_flow_over_4_switches(
assert result["result"].version == 1
assert result["result"].title == "TheOver4SwitchMockName"
assert isinstance(result["result"], ConfigEntry)
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("expected_lingering_timers", [True])
# TODO this test fails when run in // but works alone
# @pytest.mark.skip
async def test_user_config_flow_over_climate_auto_start_stop(
hass: HomeAssistant, skip_hass_states_get
): # pylint: disable=unused-argument
"""Test the config flow with auto_start_stop thermostat_over_climate features."""
# await create_central_config(hass)
# 1. start a config flow in over_climate
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == SOURCE_USER
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_THERMOSTAT_TYPE: CONF_THERMOSTAT_CLIMATE,
},
)
assert result["type"] == FlowResultType.MENU
assert result["step_id"] == "menu"
assert result["menu_options"] == [
"main",
"features",
"type",
"presets",
"advanced",
"configuration_not_complete",
]
assert result.get("errors") is None
# 2. Add auto-start-stop feature
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"next_step_id": "features"}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "features"
assert result.get("errors") == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USE_MOTION_FEATURE: False,
CONF_USE_POWER_FEATURE: False,
CONF_USE_PRESENCE_FEATURE: False,
CONF_USE_WINDOW_FEATURE: False,
CONF_USE_AUTO_START_STOP_FEATURE: True,
},
)
assert result["type"] == FlowResultType.MENU
assert result["step_id"] == "menu"
assert result.get("errors") is None
assert result["menu_options"] == [
"main",
"features",
"type",
"presets",
"auto_start_stop",
"advanced",
"configuration_not_complete",
# "finalize", finalize is not present waiting for advanced configuration
]
# 3. Configure auto-start-stop attributes
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"next_step_id": "auto_start_stop"}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "auto_start_stop"
assert result.get("errors") == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_AUTO_START_STOP_LEVEL: AUTO_START_STOP_LEVEL_MEDIUM,
},
)
assert result["type"] == FlowResultType.MENU
assert result["step_id"] == "menu"
assert result.get("errors") is None
# 4. Configure main attributes
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"next_step_id": "main"}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "main"
assert result.get("errors") == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_NAME: "TheOverClimateMockName",
CONF_TEMP_SENSOR: "sensor.mock_temp_sensor",
CONF_CYCLE_MIN: 5,
CONF_DEVICE_POWER: 1,
CONF_USE_MAIN_CENTRAL_CONFIG: False,
CONF_USE_CENTRAL_MODE: True,
# Keep default values which are False
},
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "main"
assert result.get("errors") == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_EXTERNAL_TEMP_SENSOR: "sensor.mock_ext_temp_sensor",
CONF_TEMP_MIN: 15,
CONF_TEMP_MAX: 30,
CONF_STEP_TEMPERATURE: 0.1,
# Keep default values which are False
},
)
assert result["type"] == FlowResultType.MENU
assert result["step_id"] == "menu"
assert result.get("errors") is None
# 5. Configure type attributes
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"next_step_id": "type"}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "type"
assert result.get("errors") == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_CLIMATE: "climate.mock_climate",
CONF_AC_MODE: False,
CONF_AUTO_REGULATION_MODE: CONF_AUTO_REGULATION_STRONG,
CONF_AUTO_REGULATION_DTEMP: 0.5,
CONF_AUTO_REGULATION_PERIOD_MIN: 2,
CONF_AUTO_FAN_MODE: CONF_AUTO_FAN_HIGH,
CONF_AUTO_REGULATION_USE_DEVICE_TEMP: False,
},
)
assert result["type"] == FlowResultType.MENU
assert result["step_id"] == "menu"
assert result["menu_options"] == [
"main",
"features",
"type",
"presets",
"auto_start_stop",
"advanced",
"configuration_not_complete",
# "finalize", # because we need Advanced default parameters
]
assert result.get("errors") is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"next_step_id": "presets"}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "presets"
assert result.get("errors") == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_USE_PRESETS_CENTRAL_CONFIG: False}
)
assert result["type"] == FlowResultType.MENU
assert result["step_id"] == "menu"
assert result.get("errors") is None
# 6. configure advanced attributes
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"next_step_id": "advanced"}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "advanced"
assert result.get("errors") == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_USE_ADVANCED_CENTRAL_CONFIG: False},
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "advanced"
assert result.get("errors") == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_MINIMAL_ACTIVATION_DELAY: 10,
CONF_SECURITY_DELAY_MIN: 5,
CONF_SECURITY_MIN_ON_PERCENT: 0.4,
CONF_SECURITY_DEFAULT_ON_PERCENT: 0.3,
},
)
assert result["type"] == FlowResultType.MENU
assert result["step_id"] == "menu"
assert result.get("errors") is None
assert result["menu_options"] == [
"main",
"features",
"type",
"presets",
"auto_start_stop",
"advanced",
"finalize", # Now finalize is present
]
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"next_step_id": "finalize"}
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result.get("errors") is None
assert result[
"data"
] == MOCK_TH_OVER_CLIMATE_USER_CONFIG | MOCK_TH_OVER_CLIMATE_MAIN_CONFIG | MOCK_TH_OVER_CLIMATE_CENTRAL_MAIN_CONFIG | MOCK_TH_OVER_CLIMATE_TYPE_CONFIG | {
CONF_MINIMAL_ACTIVATION_DELAY: 10,
CONF_SECURITY_DELAY_MIN: 5,
CONF_SECURITY_MIN_ON_PERCENT: 0.4,
CONF_SECURITY_DEFAULT_ON_PERCENT: 0.3,
} | MOCK_DEFAULT_FEATURE_CONFIG | {
CONF_USE_MAIN_CENTRAL_CONFIG: False,
CONF_USE_TPI_CENTRAL_CONFIG: False,
CONF_USE_PRESETS_CENTRAL_CONFIG: False,
CONF_USE_MOTION_FEATURE: False,
CONF_USE_POWER_FEATURE: False,
CONF_USE_PRESENCE_FEATURE: False,
CONF_USE_WINDOW_FEATURE: False,
CONF_USE_AUTO_START_STOP_FEATURE: False,
CONF_USE_CENTRAL_BOILER_FEATURE: False,
CONF_USE_TPI_CENTRAL_CONFIG: False,
CONF_USE_WINDOW_CENTRAL_CONFIG: False,
CONF_USE_MOTION_CENTRAL_CONFIG: False,
CONF_USE_POWER_CENTRAL_CONFIG: False,
CONF_USE_PRESENCE_CENTRAL_CONFIG: False,
CONF_USE_ADVANCED_CENTRAL_CONFIG: False,
CONF_USED_BY_CENTRAL_BOILER: False,
CONF_USE_AUTO_START_STOP_FEATURE: True,
CONF_AUTO_START_STOP_LEVEL: AUTO_START_STOP_LEVEL_MEDIUM,
}
assert result["result"]
assert result["result"].domain == DOMAIN
assert result["result"].version == 1
assert result["result"].title == "TheOverClimateMockName"
assert isinstance(result["result"], ConfigEntry)