Tests ok. But tests are not complete

This commit is contained in:
Jean-Marc Collin
2024-12-26 20:00:51 +00:00
parent a11eaef9f7
commit 18248aee9e
12 changed files with 869 additions and 431 deletions

View File

@@ -69,7 +69,7 @@ motion_state: 'off'
overpowering_state: false
presence_state: 'on'
window_auto_state: false
window_bypass_state: false
is_window_bypass: false
security_delay_min: 2
security_min_on_percent: 0.5
security_default_on_percent: 0.1

View File

@@ -6,7 +6,6 @@ import math
import logging
from typing import Any, Generic
from datetime import timedelta, datetime
from homeassistant.core import (
HomeAssistant,
callback,
@@ -26,11 +25,8 @@ from homeassistant.helpers.device_registry import DeviceInfo, DeviceEntryType
from homeassistant.helpers.event import (
async_track_state_change_event,
async_call_later,
)
from homeassistant.exceptions import ConditionError
from homeassistant.helpers import condition
from homeassistant.components.climate import (
ATTR_PRESET_MODE,
@@ -55,7 +51,6 @@ from homeassistant.const import (
ATTR_TEMPERATURE,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
STATE_OFF,
STATE_ON,
)
@@ -68,13 +63,13 @@ from .vtherm_api import VersatileThermostatAPI
from .underlyings import UnderlyingEntity
from .prop_algorithm import PropAlgorithm
from .open_window_algorithm import WindowOpenDetectionAlgorithm
from .ema import ExponentialMovingAverage
from .base_manager import BaseFeatureManager
from .feature_presence_manager import FeaturePresenceManager
from .feature_power_manager import FeaturePowerManager
from .feature_motion_manager import FeatureMotionManager
from .feature_window_manager import FeatureWindowManager
_LOGGER = logging.getLogger(__name__)
@@ -115,13 +110,6 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
"minimal_activation_delay_sec",
"last_update_datetime",
"timezone",
"window_sensor_entity_id",
"window_delay_sec",
"window_auto_enabled",
"window_auto_open_threshold",
"window_auto_close_threshold",
"window_auto_max_duration",
"window_action",
"temperature_unit",
"is_device_active",
"device_actives",
@@ -168,10 +156,8 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
self._fan_mode = None
self._humidity = None
self._swing_mode = None
self._window_state = None
self._saved_hvac_mode = None
self._window_call_cancel = None
self._cur_temp = None
self._ac_mode = None
@@ -199,17 +185,6 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
self._underlying_climate_start_hvac_action_date = None
self._underlying_climate_delta_t = 0
self._window_sensor_entity_id = None
self._window_delay_sec = None
self._window_auto_open_threshold = 0
self._window_auto_close_threshold = 0
self._window_auto_max_duration = 0
self._window_auto_state = False
self._window_auto_on = False
self._window_auto_algo = None
self._window_bypass_state = False
self._window_action = None
self._current_tz = dt_util.get_time_zone(self._hass.config.time_zone)
# Last change time is the datetime of the last change sent by VTherm to the device
@@ -246,10 +221,12 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
)
self._power_manager: FeaturePowerManager = FeaturePowerManager(self, hass)
self._motion_manager: FeatureMotionManager = FeatureMotionManager(self, hass)
self._window_manager: FeatureWindowManager = FeatureWindowManager(self, hass)
self.register_manager(self._presence_manager)
self.register_manager(self._power_manager)
self.register_manager(self._motion_manager)
self.register_manager(self._window_manager)
self.post_init(entry_infos)
@@ -338,10 +315,6 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
self._attr_preset_modes: list[str] | None
if self._window_call_cancel is not None:
self._window_call_cancel()
self._window_call_cancel = None
self._cycle_min = entry_infos.get(CONF_CYCLE_MIN)
# Initialize underlying entities (will be done in subclasses)
@@ -353,29 +326,6 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
CONF_LAST_SEEN_TEMP_SENSOR
)
self._ext_temp_sensor_entity_id = entry_infos.get(CONF_EXTERNAL_TEMP_SENSOR)
self._window_sensor_entity_id = entry_infos.get(CONF_WINDOW_SENSOR)
self._window_delay_sec = entry_infos.get(CONF_WINDOW_DELAY)
self._window_auto_open_threshold = entry_infos.get(
CONF_WINDOW_AUTO_OPEN_THRESHOLD
)
self._window_auto_close_threshold = entry_infos.get(
CONF_WINDOW_AUTO_CLOSE_THRESHOLD
)
self._window_auto_max_duration = entry_infos.get(CONF_WINDOW_AUTO_MAX_DURATION)
self._window_auto_on = (
self._window_sensor_entity_id is None
and self._window_auto_open_threshold is not None
and self._window_auto_open_threshold > 0.0
and self._window_auto_close_threshold is not None
and self._window_auto_max_duration is not None
and self._window_auto_max_duration > 0
)
self._window_auto_state = False
self._window_auto_algo = WindowOpenDetectionAlgorithm(
alert_threshold=self._window_auto_open_threshold,
end_alert_threshold=self._window_auto_close_threshold,
)
self._tpi_coef_int = entry_infos.get(CONF_TPI_COEF_INT)
self._tpi_coef_ext = entry_infos.get(CONF_TPI_COEF_EXT)
@@ -441,9 +391,6 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
if self._prop_algorithm is not None:
del self._prop_algorithm
# Memory synthesis state
self._window_state = None
self._total_energy = None
_LOGGER.debug("%s - post_init_ resetting energy to None", self)
@@ -470,10 +417,6 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
entry_infos.get(CONF_USED_BY_CENTRAL_BOILER) is True
)
self._window_action = (
entry_infos.get(CONF_WINDOW_ACTION) or CONF_WINDOW_TURN_OFF
)
self._max_on_percent = api.max_on_percent
_LOGGER.debug(
@@ -514,15 +457,6 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
)
)
if self._window_sensor_entity_id:
self.async_on_remove(
async_track_state_change_event(
self.hass,
[self._window_sensor_entity_id],
self._async_windows_changed,
)
)
# start listening for all managers
for manager in self._managers:
manager.start_listening()
@@ -604,21 +538,6 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
self,
)
# try to acquire window entity state
if self._window_sensor_entity_id:
window_state = self.hass.states.get(self._window_sensor_entity_id)
if window_state and window_state.state not in (
STATE_UNAVAILABLE,
STATE_UNKNOWN,
):
self._window_state = window_state.state == STATE_ON
_LOGGER.debug(
"%s - Window state have been retrieved: %s",
self,
self._window_state,
)
need_write_state = True
# refresh states for all managers
for manager in self._managers:
if await manager.refresh_state():
@@ -692,7 +611,7 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
]:
self._hvac_mode = old_state.state
# restpre also saved info so that window detection will work
# restore also saved info so that window detection will work
self._saved_hvac_mode = old_state.attributes.get("saved_hvac_mode", None)
self._saved_preset_mode = old_state.attributes.get(
"saved_preset_mode", None
@@ -730,7 +649,7 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
if not self.is_on and self.hvac_off_reason is None:
self.set_hvac_off_reason(HVAC_OFF_REASON_MANUAL)
self._saved_target_temp = self._target_temp
self.save_target_temp()
self.send_event(EventType.PRESET_EVENT, {"preset": self._attr_preset_mode})
self.send_event(EventType.HVAC_MODE_EVENT, {"hvac_mode": self._hvac_mode})
@@ -938,22 +857,17 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
@property
def window_state(self) -> str | None:
"""Get the window_state"""
return STATE_ON if self._window_state else STATE_OFF
return self._window_manager.window_state
@property
def window_auto_state(self) -> str | None:
"""Get the window_auto_state"""
return STATE_ON if self._window_auto_state else STATE_OFF
return self._window_manager.window_auto_state
@property
def window_bypass_state(self) -> bool | None:
def is_window_bypass(self) -> bool | None:
"""Get the Window Bypass"""
return self._window_bypass_state
@property
def window_action(self) -> bool | None:
"""Get the Window Action"""
return self._window_action
return self._window_manager.is_window_bypass
@property
def security_state(self) -> bool | None:
@@ -1000,15 +914,7 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
@property
def last_temperature_slope(self) -> float | None:
"""Return the last temperature slope curve if any"""
if not self._window_auto_algo:
return None
else:
return self._window_auto_algo.last_slope
@property
def is_window_auto_enabled(self) -> bool:
"""True if the Window auto feature is enabled"""
return self._window_auto_on
return self._window_manager.last_slope
@property
def nb_underlying_entities(self) -> int:
@@ -1213,16 +1119,16 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
if preset_mode == PRESET_NONE:
self._attr_preset_mode = PRESET_NONE
if self._saved_target_temp:
await self.change_target_temperature(self._saved_target_temp)
await self.restore_target_temp()
elif preset_mode == PRESET_ACTIVITY:
self._attr_preset_mode = PRESET_ACTIVITY
await self._motion_manager.update_motion(None, False)
await self._motion_manager.update_motion_state(None, False)
else:
if self._attr_preset_mode == PRESET_NONE:
self._saved_target_temp = self._target_temp
self.save_target_temp()
self._attr_preset_mode = preset_mode
# Switch the temperature if window is not 'on'
if self.window_state != STATE_ON:
if not self._window_manager.is_window_detected:
await self.change_target_temperature(self.find_preset_temp(preset_mode))
else:
# Window is on, so we just save the new expected temp
@@ -1339,7 +1245,7 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
return
self._attr_preset_mode = PRESET_NONE
if self.window_state != STATE_ON:
if not self._window_manager.is_window_detected:
await self.change_target_temperature(temperature)
self.recalculate()
self.reset_last_change_time_from_vtherm()
@@ -1376,8 +1282,9 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
_LOGGER.info("%s - Change entry with the values: %s", self, config_entry.data)
@callback
async def _async_temperature_changed(self, event: Event):
"""Handle temperature of the temperature sensor changes."""
async def _async_temperature_changed(self, event: Event) -> callable:
"""Handle temperature of the temperature sensor changes.
Return the fonction to desarm (clear) the window auto check"""
new_state: State = event.data.get("new_state")
_LOGGER.debug(
"%s - Temperature changed. Event.new_state is %s",
@@ -1387,6 +1294,7 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
if new_state is None or new_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
return
# TODO ce code avec du dearm est curieux. A voir après refacto
dearm_window_auto = await self._async_update_temp(new_state)
self.recalculate()
await self.async_control_heating(force=False)
@@ -1446,70 +1354,6 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
self.recalculate()
await self.async_control_heating(force=False)
@callback
async def _async_windows_changed(self, event):
"""Handle window changes."""
new_state = event.data.get("new_state")
old_state = event.data.get("old_state")
_LOGGER.info(
"%s - Window changed. Event.new_state is %s, _hvac_mode=%s, _saved_hvac_mode=%s",
self,
new_state,
self._hvac_mode,
self._saved_hvac_mode,
)
# Check delay condition
async def try_window_condition(_):
try:
long_enough = condition.state(
self.hass,
self._window_sensor_entity_id,
new_state.state,
timedelta(seconds=self._window_delay_sec),
)
except ConditionError:
long_enough = False
if not long_enough:
_LOGGER.debug(
"Window delay condition is not satisfied. Ignore window event"
)
self._window_state = old_state.state == STATE_ON
return
_LOGGER.debug("%s - Window delay condition is satisfied", self)
# if not self._saved_hvac_mode:
# self._saved_hvac_mode = self._hvac_mode
if self._window_state == (new_state.state == STATE_ON):
_LOGGER.debug("%s - no change in window state. Forget the event")
return
self._window_state = new_state.state == STATE_ON
_LOGGER.debug("%s - Window ByPass is : %s", self, self._window_bypass_state)
if self._window_bypass_state:
_LOGGER.info(
"%s - Window ByPass is activated. Ignore window event", self
)
else:
await self.change_window_detection_state(self._window_state)
self.update_custom_attributes()
if new_state is None or old_state is None or new_state.state == old_state.state:
return try_window_condition
if self._window_call_cancel:
self._window_call_cancel()
self._window_call_cancel = None
self._window_call_cancel = async_call_later(
self.hass, timedelta(seconds=self._window_delay_sec), try_window_condition
)
# For testing purpose we need to access the inner function
return try_window_condition
@callback
async def _check_initial_state(self):
"""Prevent the device from keep running if HVAC_MODE_OFF."""
@@ -1518,17 +1362,9 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
await under.check_initial_state(self._hvac_mode)
# Prevent from starting a VTherm if window is open
if (
self.is_window_auto_enabled
and self._window_sensor_entity_id is not None
and self._hass.states.is_state(self._window_sensor_entity_id, STATE_ON)
and self.is_on
and self.window_action == CONF_WINDOW_TURN_OFF
):
if self.is_on:
_LOGGER.info("%s - the window is open. Prevent starting the VTherm")
self._window_auto_state = True
self.save_hvac_mode()
await self.async_set_hvac_mode(HVACMode.OFF)
await self._window_manager.refresh_state()
# Starts the initial control loop (don't wait for an update of temperature)
await self.async_control_heating(force=True)
@@ -1561,7 +1397,7 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
await self.check_safety()
# check window_auto
return await self._async_manage_window_auto()
return await self._window_manager.manage_window_auto()
except ValueError as ex:
_LOGGER.error("Unable to update temperature from sensor: %s", ex)
@@ -1595,111 +1431,6 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
for under in self._underlyings:
await under.turn_off()
async def _async_manage_window_auto(self, in_cycle=False):
"""The management of the window auto feature"""
async def dearm_window_auto(_):
"""Callback that will be called after end of WINDOW_AUTO_MAX_DURATION"""
_LOGGER.info("Unset window auto because MAX_DURATION is exceeded")
await deactivate_window_auto(auto=True)
async def deactivate_window_auto(auto=False):
"""Deactivation of the Window auto state"""
_LOGGER.warning(
"%s - End auto detection of open window slope=%.3f", self, slope
)
# Send an event
cause = "max duration expiration" if auto else "end of slope alert"
self.send_event(
EventType.WINDOW_AUTO_EVENT,
{"type": "end", "cause": cause, "curve_slope": slope},
)
# Set attributes
self._window_auto_state = False
await self.change_window_detection_state(self._window_auto_state)
# await self.restore_hvac_mode(True)
if self._window_call_cancel:
self._window_call_cancel()
self._window_call_cancel = None
if not self._window_auto_algo:
return
if in_cycle:
slope = self._window_auto_algo.check_age_last_measurement(
temperature=self._ema_temp,
datetime_now=self.now,
)
else:
slope = self._window_auto_algo.add_temp_measurement(
temperature=self._ema_temp,
datetime_measure=self._last_temperature_measure,
)
_LOGGER.debug(
"%s - Window auto is on, check the alert. last slope is %.3f",
self,
slope if slope is not None else 0.0,
)
if self.window_bypass_state or not self.is_window_auto_enabled:
_LOGGER.debug(
"%s - Window auto event is ignored because bypass is ON or window auto detection is disabled",
self,
)
return
if (
self._window_auto_algo.is_window_open_detected()
and self._window_auto_state is False
and self.hvac_mode != HVACMode.OFF
):
if (
self.proportional_algorithm
and self.proportional_algorithm.on_percent <= 0.0
):
_LOGGER.info(
"%s - Start auto detection of open window slope=%.3f but no heating detected (on_percent<=0). Forget the event",
self,
slope,
)
return dearm_window_auto
_LOGGER.warning(
"%s - Start auto detection of open window slope=%.3f", self, slope
)
# Send an event
self.send_event(
EventType.WINDOW_AUTO_EVENT,
{"type": "start", "cause": "slope alert", "curve_slope": slope},
)
# Set attributes
self._window_auto_state = True
await self.change_window_detection_state(self._window_auto_state)
# self.save_hvac_mode()
# await self.async_set_hvac_mode(HVACMode.OFF)
# Arm the end trigger
if self._window_call_cancel:
self._window_call_cancel()
self._window_call_cancel = None
self._window_call_cancel = async_call_later(
self.hass,
timedelta(minutes=self._window_auto_max_duration),
dearm_window_auto,
)
elif (
self._window_auto_algo.is_window_close_detected()
and self._window_auto_state is True
):
await deactivate_window_auto(False)
# For testing purpose we need to return the inner function
return dearm_window_auto
def save_preset_mode(self):
"""Save the current preset mode to be restored later
We never save a hidden preset mode
@@ -1744,6 +1475,14 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
self._hvac_mode,
)
def save_target_temp(self):
"""Save the target temperature"""
self._saved_target_temp = self._target_temp
async def restore_target_temp(self):
"""Restore the saved target temp"""
await self.change_target_temperature(self._saved_target_temp)
async def check_central_mode(
self, new_central_mode: str | None, old_central_mode: str | None
):
@@ -1768,16 +1507,17 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
self.save_preset_mode()
self.save_hvac_mode()
is_window_detected = self._window_manager.is_window_detected
if new_central_mode == CENTRAL_MODE_AUTO:
if self.window_state is not STATE_ON and not first_init:
if not is_window_detected and not first_init:
await self.restore_hvac_mode()
await self.restore_preset_mode()
elif self.window_state is STATE_ON and self.hvac_mode == HVACMode.OFF:
elif is_window_detected and self.hvac_mode == HVACMode.OFF:
# do not restore but mark the reason of off with window detection
self.set_hvac_off_reason(HVAC_OFF_REASON_WINDOW_DETECTION)
return
if old_central_mode == CENTRAL_MODE_AUTO and self.window_state is not STATE_ON:
if old_central_mode == CENTRAL_MODE_AUTO and not is_window_detected:
save_all()
if new_central_mode == CENTRAL_MODE_STOPPED:
@@ -1990,78 +1730,6 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
should have found the underlying climate to be operational"""
return True
async def change_window_detection_state(self, new_state):
"""Change the window detection state.
new_state is on if an open window have been detected or off else
"""
if new_state is False:
_LOGGER.info(
"%s - Window is closed. Restoring hvac_mode '%s' if stopped by window detection or temperature %s",
self,
self._saved_hvac_mode,
self._saved_target_temp,
)
if self._window_action in [CONF_WINDOW_FROST_TEMP, CONF_WINDOW_ECO_TEMP]:
await self.change_target_temperature(self._saved_target_temp)
# default to TURN_OFF
elif self._window_action in [CONF_WINDOW_TURN_OFF]:
if (
self.last_central_mode != CENTRAL_MODE_STOPPED
and self.hvac_off_reason == HVAC_OFF_REASON_WINDOW_DETECTION
):
self.set_hvac_off_reason(None)
await self.restore_hvac_mode(True)
elif self._window_action in [CONF_WINDOW_FAN_ONLY]:
if self.last_central_mode != CENTRAL_MODE_STOPPED:
self.set_hvac_off_reason(None)
await self.restore_hvac_mode(True)
else:
_LOGGER.error(
"%s - undefined window_action %s. Please open a bug in the github of this project with this log",
self,
self._window_action,
)
else:
_LOGGER.info(
"%s - Window is open. Apply window action %s", self, self._window_action
)
if self._window_action == CONF_WINDOW_TURN_OFF and not self.is_on:
_LOGGER.debug(
"%s is already off. Forget turning off VTherm due to window detection"
)
return
if self.last_central_mode in [CENTRAL_MODE_AUTO, None]:
if self._window_action in [CONF_WINDOW_TURN_OFF, CONF_WINDOW_FAN_ONLY]:
self.save_hvac_mode()
elif self._window_action in [
CONF_WINDOW_FROST_TEMP,
CONF_WINDOW_ECO_TEMP,
]:
self._saved_target_temp = self._target_temp
if (
self._window_action == CONF_WINDOW_FAN_ONLY
and HVACMode.FAN_ONLY in self.hvac_modes
):
await self.async_set_hvac_mode(HVACMode.FAN_ONLY)
elif (
self._window_action == CONF_WINDOW_FROST_TEMP
and self._presets.get(PRESET_FROST_PROTECTION) is not None
):
await self.change_target_temperature(
self.find_preset_temp(PRESET_FROST_PROTECTION)
)
elif (
self._window_action == CONF_WINDOW_ECO_TEMP
and self._presets.get(PRESET_ECO) is not None
):
await self.change_target_temperature(self.find_preset_temp(PRESET_ECO))
else: # default is to turn_off
self.set_hvac_off_reason(HVAC_OFF_REASON_WINDOW_DETECTION)
await self.async_set_hvac_mode(HVACMode.OFF)
async def async_control_heating(self, force=False, _=None) -> bool:
"""The main function used to run the calculation at each cycle"""
@@ -2074,7 +1742,7 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
)
# check auto_window conditions
await self._async_manage_window_auto(in_cycle=True)
await self._window_manager.manage_window_auto(in_cycle=True)
# In over_climate mode, if the underlying climate is not initialized, try to initialize it
if not self.is_initialized:
@@ -2160,16 +1828,6 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
"saved_preset_mode": self._saved_preset_mode,
"saved_target_temp": self._saved_target_temp,
"saved_hvac_mode": self._saved_hvac_mode,
"window_state": self.window_state,
"window_auto_state": self.window_auto_state,
"window_bypass_state": self._window_bypass_state,
"window_sensor_entity_id": self._window_sensor_entity_id,
"window_delay_sec": self._window_delay_sec,
"window_auto_enabled": self.is_window_auto_enabled,
"window_auto_open_threshold": self._window_auto_open_threshold,
"window_auto_close_threshold": self._window_auto_close_threshold,
"window_auto_max_duration": self._window_auto_max_duration,
"window_action": self.window_action,
"security_delay_min": self._security_delay_min,
"security_min_on_percent": self._security_min_on_percent,
"security_default_on_percent": self._security_default_on_percent,
@@ -2215,6 +1873,21 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
"""True if the Thermostat is regulated by valve"""
return False
@property
def saved_target_temp(self) -> float:
"""Returns the saved_target_temp"""
return self._saved_target_temp
@property
def saved_hvac_mode(self) -> float:
"""Returns the saved_hvac_mode"""
return self._saved_hvac_mode
@property
def saved_preset_mode(self) -> float:
"""Returns the saved_preset_mode"""
return self._saved_preset_mode
@callback
def async_registry_entry_updated(self):
"""update the entity if the config entry have been updated
@@ -2324,22 +1997,8 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
self,
window_bypass,
)
self._window_bypass_state = window_bypass
if not self._window_bypass_state and self._window_state:
_LOGGER.info(
"%s - Last window state was open & ByPass is now off. Set hvac_mode to '%s'",
self,
HVACMode.OFF,
)
self.save_hvac_mode()
await self.async_set_hvac_mode(HVACMode.OFF)
if self._window_bypass_state and self._window_state:
_LOGGER.info(
"%s - Last window state was open & ByPass is now on. Set hvac_mode to last available mode",
self,
)
await self.restore_hvac_mode(True)
self.update_custom_attributes()
if await self._window_manager.set_window_bypass(window_bypass):
self.update_custom_attributes()
def send_event(self, event_type: EventType, data: dict):
"""Send an event"""
@@ -2428,3 +2087,7 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
await self.async_set_hvac_mode(HVACMode.COOL)
else:
await self.async_set_hvac_mode(HVACMode.HEAT)
def is_preset_configured(self, preset) -> bool:
"""Returns True if the preset in argument is configured"""
return self._presets.get(preset, None) is not None

View File

@@ -317,8 +317,8 @@ class WindowByPassBinarySensor(VersatileThermostatBaseEntity, BinarySensorEntity
"""Called when my climate have change"""
# _LOGGER.debug("%s - climate state change", self._attr_unique_id)
old_state = self._attr_is_on
if self.my_climate.window_bypass_state in [True, False]:
self._attr_is_on = self.my_climate.window_bypass_state
if self.my_climate.is_window_bypass in [True, False]:
self._attr_is_on = self.my_climate.is_window_bypass
if old_state != self._attr_is_on:
self.async_write_ha_state()
return

View File

@@ -39,7 +39,7 @@ _LOGGER = logging.getLogger(__name__)
class FeatureMotionManager(BaseFeatureManager):
"""The implementation of the Presence feature"""
"""The implementation of the Motion feature"""
unrecorded_attributes = frozenset(
{
@@ -67,9 +67,7 @@ class FeatureMotionManager(BaseFeatureManager):
@overrides
def post_init(self, entry_infos: ConfigData):
"""Reinit of the manager"""
if self._motion_call_cancel is not None:
self._motion_call_cancel() # pylint: disable="not-callable"
self._motion_call_cancel = None
self.dearm_motion_timer()
self._motion_sensor_entity_id = entry_infos.get(CONF_MOTION_SENSOR, None)
self._motion_delay_sec = entry_infos.get(CONF_MOTION_DELAY, 0)
@@ -130,7 +128,7 @@ class FeatureMotionManager(BaseFeatureManager):
self._motion_state,
)
# recalculate the right target_temp in activity mode
ret = await self.update_motion(motion_state.state, False)
ret = await self.update_motion_state(motion_state.state, False)
return ret
@@ -193,9 +191,9 @@ class FeatureMotionManager(BaseFeatureManager):
if long_enough:
_LOGGER.debug("%s - Motion delay condition is satisfied", self)
await self.update_motion(new_state.state)
await self.update_motion_state(new_state.state)
else:
await self.update_motion(
await self.update_motion_state(
STATE_ON if new_state.state == STATE_OFF else STATE_OFF
)
@@ -242,10 +240,10 @@ class FeatureMotionManager(BaseFeatureManager):
_LOGGER.debug("%s - Event ignored cause i'm already on", self)
return None
async def update_motion(
async def update_motion_state(
self, new_state: str = None, recalculate: bool = True
) -> bool:
"""Update the value of the presence sensor and update the VTherm state accordingly
"""Update the value of the motion sensor and update the VTherm state accordingly
Return true if a change has been made"""
_LOGGER.info("%s - Updating motion state. New state is %s", self, new_state)
@@ -261,7 +259,7 @@ class FeatureMotionManager(BaseFeatureManager):
new_preset,
)
# We do not change the preset which is kept to ACTIVITY but only the target_temperature
# We take the presence into account
# We take the motion into account
new_temp = self._vtherm.find_preset_temp(new_preset)
old_temp = self._vtherm.target_temperature
if new_temp != old_temp:
@@ -298,12 +296,12 @@ class FeatureMotionManager(BaseFeatureManager):
@overrides
@property
def is_configured(self) -> bool:
"""Return True of the presence is configured"""
"""Return True of the motion is configured"""
return self._is_configured
@property
def motion_state(self) -> str | None:
"""Return the current presence state STATE_ON or STATE_OFF
"""Return the current motion state STATE_ON or STATE_OFF
or STATE_UNAVAILABLE if not configured"""
if not self._is_configured:
return STATE_UNAVAILABLE
@@ -311,14 +309,14 @@ class FeatureMotionManager(BaseFeatureManager):
@property
def is_motion_detected(self) -> bool:
"""Return true if the presence is configured and presence sensor is OFF"""
"""Return true if the motion is configured and motion sensor is OFF"""
return self._is_configured and self._motion_state in [
STATE_ON,
]
@property
def motion_sensor_entity_id(self) -> bool:
"""Return true if the presence is configured and presence sensor is OFF"""
"""Return true if the motion is configured and motion sensor is OFF"""
return self._motion_sensor_entity_id
@property

View File

@@ -0,0 +1,553 @@
""" Implements the Window Feature Manager """
# pylint: disable=line-too-long
import logging
from typing import Any
from datetime import timedelta
from homeassistant.const import (
STATE_ON,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import (
HomeAssistant,
callback,
Event,
)
from homeassistant.helpers.event import (
async_track_state_change_event,
EventStateChangedData,
async_call_later,
)
from homeassistant.components.climate import HVACMode
from homeassistant.exceptions import ConditionError
from homeassistant.helpers import condition
from .const import * # pylint: disable=wildcard-import, unused-wildcard-import
from .commons import ConfigData
from .base_manager import BaseFeatureManager
from .open_window_algorithm import WindowOpenDetectionAlgorithm
_LOGGER = logging.getLogger(__name__)
class FeatureWindowManager(BaseFeatureManager):
"""The implementation of the Window feature"""
unrecorded_attributes = frozenset(
{
"window_sensor_entity_id",
"is_window_configured",
"is_window_bypass",
"window_delay_sec",
"window_auto_configured",
"window_auto_open_threshold",
"window_auto_close_threshold",
"window_auto_max_duration",
"window_action",
}
)
def __init__(self, vtherm: Any, hass: HomeAssistant):
"""Init of a featureManager"""
super().__init__(vtherm, hass)
self._window_sensor_entity_id: str = None
self._window_state: str = STATE_UNAVAILABLE
self._window_auto_open_threshold: float = 0
self._window_auto_close_threshold: float = 0
self._window_auto_max_duration: int = 0
self._window_auto_state: bool = False
self._window_auto_algo: WindowOpenDetectionAlgorithm = None
self._is_window_bypass: bool = False
self._window_action: str = None
self._window_delay_sec: int | None = 0
self._is_configured: bool = False
self._is_window_auto_configured: bool = False
self._window_call_cancel: callable = None
@overrides
def post_init(self, entry_infos: ConfigData):
"""Reinit of the manager"""
self.dearm_window_timer()
self._window_auto_state = STATE_UNAVAILABLE
self._window_state = STATE_UNAVAILABLE
self._window_sensor_entity_id = entry_infos.get(CONF_WINDOW_SENSOR)
self._window_delay_sec = entry_infos.get(CONF_WINDOW_DELAY)
self._window_action = (
entry_infos.get(CONF_WINDOW_ACTION) or CONF_WINDOW_TURN_OFF
)
self._window_auto_open_threshold = entry_infos.get(
CONF_WINDOW_AUTO_OPEN_THRESHOLD
)
self._window_auto_close_threshold = entry_infos.get(
CONF_WINDOW_AUTO_CLOSE_THRESHOLD
)
self._window_auto_max_duration = entry_infos.get(CONF_WINDOW_AUTO_MAX_DURATION)
use_window_feature = entry_infos.get(CONF_USE_WINDOW_FEATURE, False)
if ( # pylint: disable=too-many-boolean-expressions
use_window_feature
and self._window_sensor_entity_id is None
and self._window_auto_open_threshold is not None
and self._window_auto_open_threshold > 0.0
and self._window_auto_close_threshold is not None
and self._window_auto_max_duration is not None
and self._window_auto_max_duration > 0
and self._window_action is not None
):
self._is_window_auto_configured = True
self._window_auto_state = STATE_UNKNOWN
self._window_auto_algo = WindowOpenDetectionAlgorithm(
alert_threshold=self._window_auto_open_threshold,
end_alert_threshold=self._window_auto_close_threshold,
)
if self._is_window_auto_configured or (
use_window_feature
and self._window_sensor_entity_id is not None
and self._window_delay_sec is not None
and self._window_action is not None
):
self._is_configured = True
self._window_state = STATE_UNKNOWN
@overrides
def start_listening(self):
"""Start listening the underlying entity"""
if self._is_configured:
self.stop_listening()
if self._window_sensor_entity_id:
self.add_listener(
async_track_state_change_event(
self.hass,
[self._window_sensor_entity_id],
self._window_sensor_changed,
)
)
@overrides
def stop_listening(self):
"""Stop listening and remove the eventual timer still running"""
self.dearm_window_timer()
super().stop_listening()
def dearm_window_timer(self):
"""Dearm the eventual motion time running"""
if self._window_call_cancel:
self._window_call_cancel()
self._window_call_cancel = None
@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:
window_state = self.hass.states.get(self._window_sensor_entity_id)
if window_state and window_state.state not in (
STATE_UNAVAILABLE,
STATE_UNKNOWN,
):
_LOGGER.debug(
"%s - Window state have been retrieved: %s",
self,
self._window_state,
)
# recalculate the right target_temp in activity mode
ret = await self.update_window_state(window_state.state)
return ret
@callback
async def _window_sensor_changed(self, event: Event[EventStateChangedData]):
"""Handle window sensor changes."""
new_state = event.data.get("new_state")
old_state = event.data.get("old_state")
_LOGGER.info(
"%s - Window changed. Event.new_state is %s, _hvac_mode=%s, _saved_hvac_mode=%s",
self,
new_state,
self._vtherm.hvac_mode,
self._vtherm.saved_hvac_mode,
)
# Check delay condition
async def try_window_condition(_):
try:
long_enough = condition.state(
self._hass,
self._window_sensor_entity_id,
new_state.state,
timedelta(seconds=self._window_delay_sec),
)
except ConditionError:
long_enough = False
if not long_enough:
_LOGGER.debug(
"Window delay condition is not satisfied. Ignore window event"
)
self._window_state = old_state.state == STATE_ON
return
_LOGGER.debug("%s - Window delay condition is satisfied", self)
if self._window_state == (new_state.state == STATE_ON):
_LOGGER.debug("%s - no change in window state. Forget the event")
return
self._window_state = new_state.state == STATE_ON
_LOGGER.debug("%s - Window ByPass is : %s", self, self._is_window_bypass)
if self._is_window_bypass:
_LOGGER.info(
"%s - Window ByPass is activated. Ignore window event", self
)
else:
await self.update_window_state(self._window_state)
self._vtherm.update_custom_attributes()
if new_state is None or old_state is None or new_state.state == old_state.state:
return try_window_condition
if self._window_call_cancel:
self._window_call_cancel()
self._window_call_cancel = None
self._window_call_cancel = async_call_later(
self.hass, timedelta(seconds=self._window_delay_sec), try_window_condition
)
# For testing purpose we need to access the inner function
return try_window_condition
async def update_window_state(self, new_state: str = None) -> bool:
"""Change the window detection state.
new_state is on if an open window have been detected or off else
return True if the state have changed
"""
if self._window_state == new_state:
return False
if new_state != STATE_ON:
_LOGGER.info(
"%s - Window is closed. Restoring hvac_mode '%s' if stopped by window detection or temperature %s",
self,
self._vtherm.saved_hvac_mode,
self._vtherm.saved_target_temp,
)
self._window_state = new_state
if self._window_action in [
CONF_WINDOW_FROST_TEMP,
CONF_WINDOW_ECO_TEMP,
]:
await self._vtherm.restore_target_temp()
# default to TURN_OFF
elif self._window_action in [CONF_WINDOW_TURN_OFF]:
if (
self._vtherm.last_central_mode != CENTRAL_MODE_STOPPED
and self._vtherm.hvac_off_reason == HVAC_OFF_REASON_WINDOW_DETECTION
):
self._vtherm.set_hvac_off_reason(None)
await self._vtherm.restore_hvac_mode(True)
elif self._window_action in [CONF_WINDOW_FAN_ONLY]:
if self._vtherm.last_central_mode != CENTRAL_MODE_STOPPED:
self._vtherm.set_hvac_off_reason(None)
await self._vtherm.restore_hvac_mode(True)
else:
_LOGGER.error(
"%s - undefined window_action %s. Please open a bug in the github of this project with this log",
self,
self._window_action,
)
else:
_LOGGER.info(
"%s - Window is open. Apply window action %s", self, self._window_action
)
if self._window_action == CONF_WINDOW_TURN_OFF and not self._vtherm.is_on:
_LOGGER.debug(
"%s is already off. Forget turning off VTherm due to window detection"
)
return
self._window_state = new_state
if self._vtherm.last_central_mode in [CENTRAL_MODE_AUTO, None]:
if self._window_action in [
CONF_WINDOW_TURN_OFF,
CONF_WINDOW_FAN_ONLY,
]:
self._vtherm.save_hvac_mode()
elif self._window_action in [
CONF_WINDOW_FROST_TEMP,
CONF_WINDOW_ECO_TEMP,
]:
self._vtherm.save_target_temp()
if (
self._window_action == CONF_WINDOW_FAN_ONLY
and HVACMode.FAN_ONLY in self._vtherm.hvac_modes
):
await self._vtherm.async_set_hvac_mode(HVACMode.FAN_ONLY)
elif (
self._window_action == CONF_WINDOW_FROST_TEMP
and self._vtherm.is_preset_configured(PRESET_FROST_PROTECTION)
is not None
):
await self._vtherm.change_target_temperature(
self._vtherm.find_preset_temp(PRESET_FROST_PROTECTION)
)
elif (
self._window_action == CONF_WINDOW_ECO_TEMP
and self._vtherm.is_preset_configured(PRESET_ECO) is not None
):
await self._vtherm.change_target_temperature(
self._vtherm.find_preset_temp(PRESET_ECO)
)
else: # default is to turn_off
self._vtherm.set_hvac_off_reason(HVAC_OFF_REASON_WINDOW_DETECTION)
await self._vtherm.async_set_hvac_mode(HVACMode.OFF)
return True
async def manage_window_auto(self, in_cycle=False) -> callable:
"""The management of the window auto feature
Returns the dearm function used to deactivate the window auto"""
async def dearm_window_auto(_):
"""Callback that will be called after end of WINDOW_AUTO_MAX_DURATION"""
_LOGGER.info("Unset window auto because MAX_DURATION is exceeded")
await deactivate_window_auto(auto=True)
async def deactivate_window_auto(auto=False):
"""Deactivation of the Window auto state"""
_LOGGER.warning(
"%s - End auto detection of open window slope=%.3f", self, slope
)
# Send an event
cause = "max duration expiration" if auto else "end of slope alert"
self._vtherm.send_event(
EventType.WINDOW_AUTO_EVENT,
{"type": "end", "cause": cause, "curve_slope": slope},
)
# Set attributes
self._window_auto_state = False
await self.update_window_state(self._window_auto_state)
# await self.restore_hvac_mode(True)
if self._window_call_cancel:
self._window_call_cancel()
self._window_call_cancel = None
if not self._window_auto_algo:
return
if in_cycle:
slope = self._window_auto_algo.check_age_last_measurement(
temperature=self._vtherm.ema_temp,
datetime_now=self._vtherm.now,
)
else:
slope = self._window_auto_algo.add_temp_measurement(
temperature=self._vtherm.ema_temp,
datetime_measure=self._vtherm.last_temperature_measure,
)
_LOGGER.debug(
"%s - Window auto is on, check the alert. last slope is %.3f",
self,
slope if slope is not None else 0.0,
)
if self.is_window_bypass or not self._is_window_auto_configured:
_LOGGER.debug(
"%s - Window auto event is ignored because bypass is ON or window auto detection is disabled",
self,
)
return
if (
self._window_auto_algo.is_window_open_detected()
and self._window_auto_state is False
and self._vtherm.hvac_mode != HVACMode.OFF
):
if (
self._vtherm.proportional_algorithm
and self._vtherm.proportional_algorithm.on_percent <= 0.0
):
_LOGGER.info(
"%s - Start auto detection of open window slope=%.3f but no heating detected (on_percent<=0). Forget the event",
self,
slope,
)
return dearm_window_auto
_LOGGER.warning(
"%s - Start auto detection of open window slope=%.3f", self, slope
)
# Send an event
self._vtherm.send_event(
EventType.WINDOW_AUTO_EVENT,
{"type": "start", "cause": "slope alert", "curve_slope": slope},
)
# Set attributes
self._window_auto_state = True
await self.update_window_state(self._window_auto_state)
# self.save_hvac_mode()
# await self.async_set_hvac_mode(HVACMode.OFF)
# Arm the end trigger
if self._window_call_cancel:
self._window_call_cancel()
self._window_call_cancel = None
self._window_call_cancel = async_call_later(
self.hass,
timedelta(minutes=self._window_auto_max_duration),
dearm_window_auto,
)
elif (
self._window_auto_algo.is_window_close_detected()
and self._window_auto_state is True
):
await deactivate_window_auto(False)
# For testing purpose we need to return the inner function
return dearm_window_auto
def add_custom_attributes(self, extra_state_attributes: dict[str, Any]):
"""Add some custom attributes"""
extra_state_attributes.update(
{
"window_state": self.window_state,
"window_auto_state": self.window_auto_state,
"window_action": self.window_action,
"is_window_bypass": self._is_window_bypass,
"window_sensor_entity_id": self._window_sensor_entity_id,
"window_delay_sec": self._window_delay_sec,
"is_window_configured": self._is_configured,
"is_window_auto_configured": self._is_window_auto_configured,
"window_auto_open_threshold": self._window_auto_open_threshold,
"window_auto_close_threshold": self._window_auto_close_threshold,
"window_auto_max_duration": self._window_auto_max_duration,
}
)
async def set_window_bypass(self, window_bypass: bool) -> bool:
"""Set the window bypass flag
Return True if state have been changed"""
self._is_window_bypass = window_bypass
if not self._is_window_bypass and self._window_state:
_LOGGER.info(
"%s - Last window state was open & ByPass is now off. Set hvac_mode to '%s'",
self,
HVACMode.OFF,
)
self._vtherm.save_hvac_mode()
await self._vtherm.async_set_hvac_mode(HVACMode.OFF)
return True
if self._is_window_bypass and self._window_state:
_LOGGER.info(
"%s - Last window state was open & ByPass is now on. Set hvac_mode to last available mode",
self,
)
await self._vtherm.restore_hvac_mode(True)
return True
return False
@overrides
@property
def is_configured(self) -> bool:
"""Return True of the window feature is configured"""
return self._is_configured
@property
def is_window_auto_configured(self) -> bool:
"""Return True of the window automatic detection is configured"""
return self._is_window_auto_configured
@property
def window_state(self) -> str | None:
"""Return the current window state STATE_ON or STATE_OFF
or STATE_UNAVAILABLE if not configured"""
if not self._is_configured:
return STATE_UNAVAILABLE
return self._window_state
@property
def window_auto_state(self) -> str | None:
"""Return the current window auto state STATE_ON or STATE_OFF
or STATE_UNAVAILABLE if not configured"""
if not self._is_configured:
return STATE_UNAVAILABLE
return self._window_auto_state
@property
def is_window_bypass(self) -> str | None:
"""Return True if the window bypass is activated"""
if not self._is_configured:
return False
return self._is_window_bypass
@property
def is_window_detected(self) -> bool:
"""Return true if the presence is configured and presence sensor is OFF"""
return self._is_configured and (
self._window_state == STATE_ON or self._window_auto_state == STATE_ON
)
@property
def window_sensor_entity_id(self) -> bool:
"""Return true if the presence is configured and presence sensor is OFF"""
return self._window_sensor_entity_id
@property
def window_delay_sec(self) -> bool:
"""Return the motion delay"""
return self._window_delay_sec
@property
def window_action(self) -> bool:
"""Return the window action"""
return self._window_action
@property
def window_auto_open_threshold(self) -> bool:
"""Return the window_auto_open_threshold"""
return self._window_auto_open_threshold
@property
def window_auto_close_threshold(self) -> bool:
"""Return the window_auto_close_threshold"""
return self._window_auto_close_threshold
@property
def window_auto_max_duration(self) -> bool:
"""Return the window_auto_max_duration"""
return self._window_auto_max_duration
@property
def last_slope(self) -> float:
"""Return the last slope (in °C/hour)"""
if not self.is_configured:
return None
return self._window_auto_algo.last_slope
def __str__(self):
return f"WindowManager-{self.name}"

View File

@@ -1126,7 +1126,7 @@ Les attributs personnalisés sont les suivants :
| ``saved_preset_mode`` | Le dernier preset utilisé avant le basculement automatique du preset |
| ``saved_target_temp`` | La dernière température utilisée avant la commutation automatique |
| ``window_state`` | Le dernier état connu du capteur de fenêtre. Aucun si la fenêtre n'est pas configurée |
| ``window_bypass_state`` | True si le bypass de la détection d'ouverture et activé |
| ``is_window_bypass`` | True si le bypass de la détection d'ouverture et activé |
| ``motion_state`` | Le dernier état connu du capteur de mouvement. Aucun si le mouvement n'est pas configuré |
| ``overpowering_state`` | Le dernier état connu du capteur surpuissant. Aucun si la gestion de l'alimentation n'est pas configurée |
| ``presence_state`` | Le dernier état connu du capteur de présence. Aucun si la gestion de présence n'est pas configurée |

View File

@@ -248,7 +248,7 @@ The custom attributes are as follows:
| ``saved_preset_mode`` | The last preset used before automatic preset switching |
| ``saved_target_temp`` | The last temperature used before automatic switching |
| ``window_state`` | The last known state of the window sensor. None if the window is not configured |
| ``window_bypass_state`` | True if the window open detection bypass is enabled |
| ``is_window_bypass`` | True if the window open detection bypass is enabled |
| ``motion_state`` | The last known state of the motion sensor. None if motion detection is not configured |
| ``overpowering_state`` | The last known state of the overpower sensor. None if power management is not configured |
| ``presence_state`` | The last known state of the presence sensor. None if presence detection is not configured |

View File

@@ -1126,7 +1126,7 @@ Les attributs personnalisés sont les suivants :
| ``saved_preset_mode`` | Le dernier preset utilisé avant le basculement automatique du preset |
| ``saved_target_temp`` | La dernière température utilisée avant la commutation automatique |
| ``window_state`` | Le dernier état connu du capteur de fenêtre. Aucun si la fenêtre n'est pas configurée |
| ``window_bypass_state`` | True si le bypass de la détection d'ouverture et activé |
| ``is_window_bypass`` | True si le bypass de la détection d'ouverture et activé |
| ``motion_state`` | Le dernier état connu du capteur de mouvement. Aucun si le mouvement n'est pas configuré |
| ``overpowering_state`` | Le dernier état connu du capteur surpuissant. Aucun si la gestion de l'alimentation n'est pas configurée |
| ``presence_state`` | Le dernier état connu du capteur de présence. Aucun si la gestion de présence n'est pas configurée |

View File

@@ -247,7 +247,7 @@ Les attributs personnalisés sont les suivants :
| ``saved_preset_mode`` | Le dernier preset utilisé avant le basculement automatique du preset |
| ``saved_target_temp`` | La dernière température utilisée avant la commutation automatique |
| ``window_state`` | Le dernier état connu du capteur de fenêtre. Aucun si la fenêtre n'est pas configurée |
| ``window_bypass_state`` | True si le bypass de la détection d'ouverture et activé |
| ``is_window_bypass`` | True si le bypass de la détection d'ouverture et activé |
| ``motion_state`` | Le dernier état connu du capteur de mouvement. Aucun si le mouvement n'est pas configuré |
| ``overpowering_state`` | Le dernier état connu du capteur surpuissant. Aucun si la gestion de l'alimentation n'est pas configurée |
| ``presence_state`` | Le dernier état connu du capteur de présence. Aucun si la gestion de présence n'est pas configurée |

View File

@@ -167,7 +167,7 @@ async def test_minimal_over_switch_wo_central_config(
assert entity.max_temp == 18
assert entity.target_temperature_step == 0.3
assert entity.preset_modes == ["none", "frost", "eco", "comfort", "boost"]
assert entity.is_window_auto_enabled is False
assert entity.is_window_auto_configured is False
assert entity.nb_underlying_entities == 1
assert entity.underlying_entity_id(0) == "switch.mock_switch"
assert entity.proportional_algorithm is not None
@@ -279,7 +279,7 @@ async def test_full_over_switch_wo_central_config(
assert entity._security_default_on_percent == 0.1
assert entity.is_inversed is False
assert entity.is_window_auto_enabled is False # we have an entity_id
assert entity.is_window_auto_configured is False # we have an entity_id
assert entity._window_sensor_entity_id == "binary_sensor.mock_window_sensor"
assert entity._window_delay_sec == 30
assert entity._window_auto_close_threshold == 0.1
@@ -402,7 +402,7 @@ async def test_full_over_switch_with_central_config(
assert entity.is_inversed is False
# We have an entity so window auto is not enabled
assert entity.is_window_auto_enabled is False
assert entity.is_window_auto_configured is False
assert entity._window_sensor_entity_id == "binary_sensor.mock_window_sensor"
assert entity._window_delay_sec == 15
assert entity._window_auto_close_threshold == 1

View File

@@ -1,4 +1,4 @@
# pylint: disable=wildcard-import, unused-wildcard-import, protected-access, unused-argument, line-too-long, unused-variable
# pylint: disable=wildcard-import, unused-wildcard-import, protected-access, unused-argument, line-too-long, unused-variable, too-many-lines
""" Test the Window management """
from datetime import datetime, timedelta

View File

@@ -2,18 +2,242 @@
""" Test the Window management """
import asyncio
import logging
from unittest.mock import patch, call, PropertyMock
from unittest.mock import patch, call, PropertyMock, AsyncMock, MagicMock
from datetime import datetime, timedelta
from custom_components.versatile_thermostat.base_thermostat import BaseThermostat
from custom_components.versatile_thermostat.thermostat_climate import (
ThermostatOverClimate,
)
from custom_components.versatile_thermostat.feature_window_manager import (
FeatureWindowManager,
)
from .commons import * # pylint: disable=wildcard-import, unused-wildcard-import
logging.getLogger().setLevel(logging.DEBUG)
async def test_window_feature_manager_create(
hass: HomeAssistant,
):
"""Test the FeatureMotionManager class direclty"""
fake_vtherm = MagicMock(spec=BaseThermostat)
type(fake_vtherm).name = PropertyMock(return_value="the name")
# 1. creation
window_manager = FeatureWindowManager(fake_vtherm, hass)
assert window_manager is not None
assert window_manager.is_configured is False
assert window_manager.is_window_auto_configured is False
assert window_manager.is_window_detected is False
assert window_manager.window_state == STATE_UNAVAILABLE
assert window_manager.name == "the name"
assert len(window_manager._active_listener) == 0
custom_attributes = {}
window_manager.add_custom_attributes(custom_attributes)
assert custom_attributes["window_sensor_entity_id"] is None
assert custom_attributes["window_state"] == STATE_UNAVAILABLE
assert custom_attributes["window_auto_state"] == STATE_UNAVAILABLE
assert custom_attributes["is_window_configured"] is False
assert custom_attributes["is_window_auto_configured"] is False
assert custom_attributes["window_delay_sec"] == 0
assert custom_attributes["window_auto_open_threshold"] == 0
assert custom_attributes["window_auto_close_threshold"] == 0
assert custom_attributes["window_auto_max_duration"] == 0
assert custom_attributes["window_action"] is None
@pytest.mark.parametrize(
"use_window_feature, window_sensor_entity_id, window_delay_sec, window_auto_open_threshold, window_auto_close_threshold, window_auto_max_duration, window_action, is_configured, is_auto_configured, window_state, window_auto_state",
[
# fmt: off
( True, "sensor.the_window_sensor", 10, None, None, None, CONF_WINDOW_TURN_OFF, True, False, STATE_UNKNOWN, STATE_UNAVAILABLE ),
( False, "sensor.the_window_sensor", 10, None, None, None, CONF_WINDOW_TURN_OFF, False, False, STATE_UNAVAILABLE, STATE_UNAVAILABLE ),
( True, "sensor.the_window_sensor", 10, None, None, None, CONF_WINDOW_TURN_OFF, True, False, STATE_UNKNOWN, STATE_UNAVAILABLE ),
# delay is missing
( True, "sensor.the_window_sensor", None, None, None, None, CONF_WINDOW_TURN_OFF, False, False, STATE_UNAVAILABLE, STATE_UNAVAILABLE ),
# action is missing -> defaults to TURN_OFF
( True, "sensor.the_window_sensor", 10, None, None, None, None, True, False, STATE_UNKNOWN, STATE_UNAVAILABLE ),
# With Window auto config complete
( True, None, None, 1, 2, 3, CONF_WINDOW_TURN_OFF, True, True, STATE_UNKNOWN, STATE_UNKNOWN ),
# With Window auto config not complete -> missing open threshold but defaults to 0
( True, None, None, None, 2, 3, CONF_WINDOW_TURN_OFF, False, False, STATE_UNAVAILABLE, STATE_UNAVAILABLE ),
# With Window auto config not complete -> missing close threshold
( True, None, None, 1, None, 3, CONF_WINDOW_TURN_OFF, False, False, STATE_UNAVAILABLE, STATE_UNAVAILABLE ),
# With Window auto config not complete -> missing max duration threshold but defaults to 0
( True, None, None, 1, 2, None, CONF_WINDOW_TURN_OFF, False, False, STATE_UNAVAILABLE, STATE_UNAVAILABLE ),
# fmt: on
],
)
async def test_window_feature_manager_post_init(
hass: HomeAssistant,
use_window_feature,
window_sensor_entity_id,
window_delay_sec,
window_auto_open_threshold,
window_auto_close_threshold,
window_auto_max_duration,
window_action,
is_configured,
is_auto_configured,
window_state,
window_auto_state,
):
"""Test the FeatureMotionManager class direclty"""
fake_vtherm = MagicMock(spec=BaseThermostat)
type(fake_vtherm).name = PropertyMock(return_value="the name")
# 1. creation
window_manager = FeatureWindowManager(fake_vtherm, hass)
assert window_manager is not None
# 2. post_init
window_manager.post_init(
{
CONF_USE_WINDOW_FEATURE: use_window_feature,
CONF_WINDOW_SENSOR: window_sensor_entity_id,
CONF_WINDOW_DELAY: window_delay_sec,
CONF_WINDOW_AUTO_OPEN_THRESHOLD: window_auto_open_threshold,
CONF_WINDOW_AUTO_CLOSE_THRESHOLD: window_auto_close_threshold,
CONF_WINDOW_AUTO_MAX_DURATION: window_auto_max_duration,
CONF_WINDOW_ACTION: window_action,
}
)
assert window_manager.is_configured is is_configured
assert window_manager.is_window_auto_configured == is_auto_configured
assert window_manager.window_sensor_entity_id == window_sensor_entity_id
assert window_manager.window_state == window_state
assert window_manager.window_auto_state == window_auto_state
assert window_manager.window_delay_sec == window_delay_sec
assert window_manager.window_auto_open_threshold == window_auto_open_threshold
assert window_manager.window_auto_close_threshold == window_auto_close_threshold
custom_attributes = {}
window_manager.add_custom_attributes(custom_attributes)
assert custom_attributes["window_sensor_entity_id"] == window_sensor_entity_id
assert custom_attributes["window_state"] == window_state
assert custom_attributes["window_auto_state"] == window_auto_state
assert custom_attributes["is_window_bypass"] is False
assert custom_attributes["is_window_configured"] is is_configured
assert custom_attributes["is_window_auto_configured"] is is_auto_configured
assert custom_attributes["is_window_bypass"] is False
assert custom_attributes["window_delay_sec"] is window_delay_sec
assert custom_attributes["window_auto_open_threshold"] is window_auto_open_threshold
assert (
custom_attributes["window_auto_close_threshold"] is window_auto_close_threshold
)
assert custom_attributes["window_auto_max_duration"] is window_auto_max_duration
@pytest.mark.parametrize(
"current_state, new_state, nb_call, window_state, is_window_detected, changed",
[
(STATE_OFF, STATE_ON, 1, STATE_ON, True, True),
(STATE_OFF, STATE_OFF, 0, STATE_OFF, False, False),
],
)
async def test_window_feature_manager_refresh_sensor(
hass: HomeAssistant,
current_state,
new_state, # new state of motion event
nb_call,
window_state,
is_window_detected,
changed,
):
"""Test the FeatureMotionManager class direclty"""
fake_vtherm = MagicMock(spec=BaseThermostat)
type(fake_vtherm).name = PropertyMock(return_value="the name")
type(fake_vtherm).preset_mode = PropertyMock(return_value=PRESET_COMFORT)
# 1. creation
window_manager = FeatureWindowManager(fake_vtherm, hass)
# 2. post_init
window_manager.post_init(
{
CONF_WINDOW_SENSOR: "sensor.the_window_sensor",
CONF_USE_WINDOW_FEATURE: True,
CONF_WINDOW_DELAY: 10,
}
)
# 3. start listening
window_manager.start_listening()
assert window_manager.is_configured is True
assert window_manager.window_state == STATE_UNKNOWN
assert window_manager.window_auto_state == STATE_UNAVAILABLE
assert len(window_manager._active_listener) == 1
# 4. test refresh with the parametrized
# fmt:off
with patch("homeassistant.core.StateMachine.get", return_value=State("sensor.the_motion_sensor", new_state)) as mock_get_state:
# fmt:on
# Configurer les méthodes mockées
fake_vtherm.async_set_hvac_mode = AsyncMock()
fake_vtherm.set_hvac_off_reason = MagicMock()
# force old state for the test
window_manager._window_state = current_state
ret = await window_manager.refresh_state()
assert ret == changed
assert window_manager.is_configured is True
# in the refresh there is no delay
assert window_manager.window_state == new_state
assert mock_get_state.call_count == 1
assert mock_get_state.call_count == 1
assert fake_vtherm.async_set_hvac_mode.call_count == nb_call
assert fake_vtherm.set_hvac_off_reason.call_count == nb_call
if nb_call == 1:
fake_vtherm.async_set_hvac_mode.assert_has_calls(
[
call.async_set_hvac_mode(HVACMode.OFF),
]
)
fake_vtherm.set_hvac_off_reason.assert_has_calls(
[
call.set_hvac_off_reason(HVAC_OFF_REASON_WINDOW_DETECTION),
]
)
fake_vtherm.reset_mock()
# 5. Check custom_attributes
custom_attributes = {}
window_manager.add_custom_attributes(custom_attributes)
assert custom_attributes["window_sensor_entity_id"] == "sensor.the_window_sensor"
assert custom_attributes["window_state"] == new_state
assert custom_attributes["window_auto_state"] == STATE_UNAVAILABLE
assert custom_attributes["is_window_bypass"] is False
assert custom_attributes["is_window_configured"] is True
assert custom_attributes["is_window_auto_configured"] is False
assert custom_attributes["is_window_bypass"] is False
assert custom_attributes["window_delay_sec"] is 10
assert custom_attributes["window_auto_open_threshold"] is None
assert (
custom_attributes["window_auto_close_threshold"] is None
)
assert custom_attributes["window_auto_max_duration"] is None
window_manager.stop_listening()
await hass.async_block_till_done()
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("expected_lingering_timers", [True])
async def test_window_management_time_not_enough(
@@ -308,7 +532,7 @@ async def test_window_auto_fast(hass: HomeAssistant, skip_hass_states_is_state):
assert entity.target_temperature == 21
assert entity.window_state is STATE_OFF
assert entity.is_window_auto_enabled is True
assert entity.is_window_auto_configured is True
# Initialize the slope algo with 2 measurements
event_timestamp = now + timedelta(minutes=1)
@@ -505,7 +729,7 @@ async def test_window_auto_fast_and_sensor(
assert entity.target_temperature == 21
assert entity.window_state is STATE_OFF
assert entity.is_window_auto_enabled is False
assert entity.is_window_auto_configured is False
# Initialize the slope algo with 2 measurements
event_timestamp = now + timedelta(minutes=1)
@@ -621,7 +845,7 @@ async def test_window_auto_auto_stop(hass: HomeAssistant, skip_hass_states_is_st
assert entity.target_temperature == 21
assert entity.window_state is STATE_OFF
assert entity.is_window_auto_enabled is True
assert entity.is_window_auto_configured is True
# 1. Initialize the slope algo with 2 measurements
event_timestamp = now + timedelta(minutes=1)
@@ -895,7 +1119,7 @@ async def test_window_bypass(hass: HomeAssistant, skip_hass_states_is_state):
assert entity.target_temperature == 19
assert entity.window_state is STATE_OFF
assert entity.is_window_auto_enabled is False
assert entity.is_window_auto_configured is False
# change temperature to force turning on the heater
with patch(
@@ -918,9 +1142,9 @@ async def test_window_bypass(hass: HomeAssistant, skip_hass_states_is_state):
# Set Window ByPass to true
await entity.service_set_window_bypass_state(True)
assert entity.window_bypass_state is True
assert entity.is_window_bypass is True
# entity._window_bypass_state = True
# entity._is_window_bypass = True
# Open the window, condition of time is satisfied, check the thermostat and heater turns off
with patch(
@@ -1038,7 +1262,7 @@ async def test_window_auto_bypass(hass: HomeAssistant, skip_hass_states_is_state
assert entity.target_temperature == 21
assert entity.window_state is STATE_OFF
assert entity.is_window_auto_enabled
assert entity.is_window_auto_configured
# Initialize the slope algo with 2 measurements
event_timestamp = now + timedelta(minutes=1)
@@ -1072,7 +1296,7 @@ async def test_window_auto_bypass(hass: HomeAssistant, skip_hass_states_is_state
# send one degre down in one minute with window bypass on
await entity.service_set_window_bypass_state(True)
assert entity.window_bypass_state is True
assert entity.is_window_bypass is True
with patch(
"custom_components.versatile_thermostat.base_thermostat.BaseThermostat.send_event"
@@ -1595,7 +1819,7 @@ async def test_window_action_eco_temp(hass: HomeAssistant, skip_hass_states_is_s
assert entity.target_temperature == 21
assert entity.window_state is STATE_OFF
assert entity.is_window_auto_enabled is True
assert entity.is_window_auto_configured is True
# 1. Initialize the slope algo with 2 measurements
event_timestamp = now + timedelta(minutes=1)
@@ -1792,7 +2016,7 @@ async def test_window_action_frost_temp(hass: HomeAssistant, skip_hass_states_is
assert entity.target_temperature == 21
assert entity.window_state is STATE_OFF
assert entity.is_window_auto_enabled is True
assert entity.is_window_auto_configured is True
# 1. Initialize the slope algo with 2 measurements
event_timestamp = now + timedelta(minutes=1)
@@ -2151,7 +2375,7 @@ async def test_window_action_frost_temp_preset_change(
assert vtherm.target_temperature == 21
assert vtherm.window_state is STATE_OFF
assert vtherm.is_window_auto_enabled is False
assert vtherm.is_window_auto_configured is False
# 1. Turn on the window sensor
now = now + timedelta(minutes=1)
@@ -2261,7 +2485,7 @@ async def test_window_action_frost_temp_temp_change(
assert vtherm.target_temperature == 21
assert vtherm.window_state is STATE_OFF
assert vtherm.is_window_auto_enabled is False
assert vtherm.is_window_auto_configured is False
# 1. Turn on the window sensor
now = now + timedelta(minutes=1)