francop/HomeAssistantVS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
6644b9e4a83d0407c3ad01fd4b1ead8780f12f14
HomeAssistantVS/custom_components/versatile_thermostat/base_thermostat.py
T

2408 lines
96 KiB
Python
Raw Blame History

# pylint: disable=line-too-long
# pylint: disable=too-many-lines
# pylint: disable=invalid-name
""" Implements the VersatileThermostat climate component """
import math
from typing import Optional, Any, Generic
from datetime import datetime
from collections.abc import Callable
from vtherm_api.log_collector import get_vtherm_logger
from homeassistant.util import dt as dt_util
from homeassistant.exceptions import ServiceValidationError
from homeassistant.core import (
HomeAssistant,
callback,
Event,
State,
)
from homeassistant.exceptions import HomeAssistantError
from homeassistant.components.climate import ClimateEntity
from homeassistant.helpers.restore_state import (
RestoreEntity,
async_get as restore_async_get,
)
from homeassistant.helpers.entity import Entity
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.event import (
async_track_state_change_event,
)
from homeassistant.components.climate.const import (
ATTR_PRESET_MODE,
# ATTR_FAN_MODE,
HVACMode,
HVACAction,
ClimateEntityFeature,
)
from homeassistant.const import (
ATTR_TEMPERATURE,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from .const import * # pylint: disable=wildcard-import, unused-wildcard-import
from .commons import write_event_log
from .commons_type import ConfigData
from .config_schema import * # pylint: disable=wildcard-import, unused-wildcard-import
from .vtherm_central_api import VersatileThermostatAPI
from .underlyings import UnderlyingEntity, T
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
from .feature_safety_manager import FeatureSafetyManager
from .feature_auto_start_stop_manager import FeatureAutoStartStopManager
from .feature_lock_manager import FeatureLockManager
from .feature_timed_preset_manager import FeatureTimedPresetManager
from .feature_heating_failure_detection_manager import FeatureHeatingFailureDetectionManager
from .feature_repair_incorrect_state_manager import FeatureRepairIncorrectStateManager
from .state_manager import StateManager
from .vtherm_state import VThermState
from .vtherm_preset import VThermPreset, HIDDEN_PRESETS, PRESET_AC_SUFFIX
from .vtherm_hvac_mode import VThermHvacMode, VThermHvacMode_OFF, to_legacy_ha_hvac_mode
_LOGGER = get_vtherm_logger(__name__)
class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
"""Representation of a base class for all Versatile Thermostat device."""
# breaking change with 2024.12 climate workaround
_attr_swing_horizontal_modes = []
_attr_swing_horizontal_mode = ""
_entity_component_unrecorded_attributes = (
ClimateEntity._entity_component_unrecorded_attributes.union(frozenset({"configuration", "preset_temperatures", "specific_states"}))
.union(FeaturePresenceManager.unrecorded_attributes)
.union(FeaturePowerManager.unrecorded_attributes)
.union(FeatureMotionManager.unrecorded_attributes)
.union(FeatureWindowManager.unrecorded_attributes)
.union(FeatureSafetyManager.unrecorded_attributes)
.union(FeatureLockManager.unrecorded_attributes)
.union(FeatureTimedPresetManager.unrecorded_attributes)
.union(FeatureHeatingFailureDetectionManager.unrecorded_attributes)
.union(FeatureRepairIncorrectStateManager.unrecorded_attributes)
)
##
## Startup functions
##
def __init__(
self,
hass: HomeAssistant,
unique_id: str,
name: str,
entry_infos: ConfigData,
):
"""Initialize the thermostat."""
super().__init__()
# To remove some silly warning event if code is fixed
self._enable_turn_on_off_backwards_compatibility = False
self._is_removed = False
self._is_startup_done = False
self._hass = hass
self._entry_infos = None
self._attr_extra_state_attributes = {}
self._unique_id = unique_id
self._name = name
self._is_ready: bool = False
self._cycle_scheduler = None
# Callbacks for TPI cycle events
self._on_cycle_start_callbacks: list[Callable] = []
self._state_manager = StateManager()
self._fan_mode = None
self._humidity = None
self._swing_mode = None
self._swing_horizontal_mode = None
self._ac_mode = None
self._cur_temp = None
self._temp_sensor_entity_id = None
self._last_seen_temp_sensor_entity_id = None
self._ext_temp_sensor_entity_id = None
self._last_ext_temperature_measure = None
self._last_temperature_measure = None
self._cur_ext_temp = None
self._should_relaunch_control_heating = None
self._attr_translation_key = "versatile_thermostat"
self._total_energy = None
_LOGGER.debug("%s - _init_ resetting energy to None", self)
# Because energy of climate is calculated in the thermostat we have to keep
# that here and not in underlying entity
self._underlying_climate_start_hvac_action_date = None
self._underlying_climate_delta_t = 0
self._underlying_climate_mean_power_cycle = 0.0
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 it is used in `over_climate` when a
# state changes from underlying to avoid loops
self._last_change_time_from_vtherm = None
self._underlyings: list[T] = []
self._ema_temp = None
self._ema_algo = None
self._prop_algorithm = None
self._proportional_function = None
self._tpi_coef_int = None
self._tpi_coef_ext = None
self._minimal_activation_delay = None
self._minimal_deactivation_delay = None
self._tpi_threshold_low = None
self._tpi_threshold_high = None
self._attr_fan_mode = None
self._is_central_mode = None
self._last_central_mode = None
self._is_used_by_central_boiler = False
self._support_flags = None
# Preset will be initialized from Number entities
self._presets: dict[str, Any] = {} # presets
self._presets_away: dict[str, Any] = {} # presets_away
self._attr_preset_modes: list[str] = []
self._vtherm_preset_modes: list[VThermPreset] = []
self._use_central_config_temperature = False
self._hvac_off_reason: str | None = None
self._hvac_mode_reason: str | None = None
self._hvac_list: list[VThermHvacMode] = []
self._str_hvac_list: list[str] = []
self._temperature_reason: str | None = None
# Instantiate all features manager
self._managers: list[BaseFeatureManager] = []
# Names of feature managers provided by external plugins that have
# already been instantiated for this thermostat. Used to avoid
# registering the same external manager twice (post_init + startup retry).
self._external_manager_names: set[str] = set()
# Instances of feature managers provided by external plugins. They are
# also present in ``self._managers`` (for lifecycle) but are tracked
# separately so they can be refreshed on every control cycle (internal
# managers have their own dedicated per-cycle mechanisms).
self._external_managers: list[BaseFeatureManager] = []
self._presence_manager: FeaturePresenceManager = FeaturePresenceManager(
self, hass
)
self._power_manager: FeaturePowerManager = FeaturePowerManager(self, hass)
self._motion_manager: FeatureMotionManager = FeatureMotionManager(self, hass)
self._window_manager: FeatureWindowManager = FeatureWindowManager(self, hass)
self._safety_manager: FeatureSafetyManager = FeatureSafetyManager(self, hass)
# Auto start/stop is only for over_climate
self._auto_start_stop_manager: FeatureAutoStartStopManager | None = None
self._lock_manager: FeatureLockManager = FeatureLockManager(self, hass)
self._timed_preset_manager: FeatureTimedPresetManager = FeatureTimedPresetManager(self, hass)
self._heating_failure_detection_manager: FeatureHeatingFailureDetectionManager = FeatureHeatingFailureDetectionManager(self, hass)
self._repair_incorrect_state_manager: FeatureRepairIncorrectStateManager = FeatureRepairIncorrectStateManager(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.register_manager(self._safety_manager)
self.register_manager(self._lock_manager)
self.register_manager(self._timed_preset_manager)
self.register_manager(self._heating_failure_detection_manager)
self.register_manager(self._repair_incorrect_state_manager)
self._cancel_recalculate_later: Callable[[], None] | None = None
self.post_init(entry_infos)
def register_manager(self, manager: BaseFeatureManager):
"""Register a manager"""
self._managers.append(manager)
def _load_external_feature_managers(self):
"""Instantiate feature managers provided by external plugins.
Query the VTherm API registry and create one manager instance per
eligible thermostat (the factory decides eligibility through its
``supports`` method). Managers whose plugin registers after this
thermostat has been built are picked up later during ``async_startup``,
mirroring the external proportional algorithm retry behavior.
"""
api = VersatileThermostatAPI.get_vtherm_api(self.hass)
if api is None or not hasattr(api, "get_feature_manager_factories"):
return
for factory in api.get_feature_manager_factories():
name = factory.name
if name in self._external_manager_names:
continue
try:
if not factory.supports(self):
continue
manager = factory.create(self)
manager.post_init(self._entry_infos)
except Exception as exc: # pylint: disable=broad-except
_LOGGER.error(
"%s - Error while creating external feature manager '%s': %s",
self,
name,
exc,
)
continue
self.register_manager(manager)
self._external_manager_names.add(name)
self._external_managers.append(manager)
_LOGGER.info("%s - Registered external feature manager '%s'", self, name)
def clean_central_config_doublon(
self, config_entry: ConfigData, central_config: ConfigEntry | None
) -> dict[str, Any]:
"""Removes all values from config with are concerned by central_config"""
def clean_one(cfg, schema: vol.Schema):
"""Clean one schema"""
for marker in schema.schema:
# Extract the actual key from Voluptuous Marker objects
key = marker.schema if hasattr(marker, 'schema') else marker
if key in cfg:
del cfg[key]
cfg = config_entry.copy()
if central_config and central_config.data:
# Removes config if central is used
if cfg.get(CONF_USE_MAIN_CENTRAL_CONFIG) is True:
clean_one(cfg, STEP_CENTRAL_MAIN_DATA_SCHEMA)
if cfg.get(CONF_USE_TPI_CENTRAL_CONFIG) is True:
clean_one(cfg, STEP_CENTRAL_TPI_DATA_SCHEMA)
if cfg.get(CONF_USE_WINDOW_CENTRAL_CONFIG) is True:
clean_one(cfg, STEP_CENTRAL_WINDOW_DATA_SCHEMA)
if cfg.get(CONF_USE_MOTION_CENTRAL_CONFIG) is True:
clean_one(cfg, STEP_CENTRAL_MOTION_DATA_SCHEMA)
if cfg.get(CONF_USE_POWER_CENTRAL_CONFIG) is True:
clean_one(cfg, STEP_CENTRAL_POWER_DATA_SCHEMA)
if cfg.get(CONF_USE_PRESENCE_CENTRAL_CONFIG) is True:
clean_one(cfg, STEP_CENTRAL_PRESENCE_DATA_SCHEMA)
if cfg.get(CONF_USE_ADVANCED_CENTRAL_CONFIG) is True:
clean_one(cfg, STEP_CENTRAL_ADVANCED_DATA_SCHEMA)
if cfg.get(CONF_USE_LOCK_CENTRAL_CONFIG) is True:
clean_one(cfg, STEP_CENTRAL_LOCK_DATA_SCHEMA)
# take all central config
entry_infos = central_config.data.copy()
# and merge with cleaned config_entry
entry_infos.update(cfg)
else:
entry_infos = cfg
return entry_infos
def post_init(self, config_entry: ConfigData):
"""Finish the initialization of the thermostat"""
_LOGGER.info(
"%s - Updating VersatileThermostat with infos %s",
self,
config_entry,
)
api: VersatileThermostatAPI = VersatileThermostatAPI.get_vtherm_api(self._hass)
central_config = api.find_central_configuration()
entry_infos = self.clean_central_config_doublon(config_entry, central_config)
_LOGGER.info("%s - The merged configuration is %s", self, entry_infos)
self._entry_infos = entry_infos
# Post init all managers
for manager in self._managers:
manager.post_init(entry_infos)
# Instantiate feature managers provided by external plugins that are
# already registered at this point. Late-registered plugins are handled
# by a retry in async_startup.
self._load_external_feature_managers()
self._use_central_config_temperature = entry_infos.get(
CONF_USE_PRESETS_CENTRAL_CONFIG
) or (
entry_infos.get(CONF_USE_PRESENCE_CENTRAL_CONFIG)
and entry_infos.get(CONF_USE_PRESENCE_FEATURE)
)
self._ac_mode = entry_infos.get(CONF_AC_MODE) is True
self._attr_max_temp = float(entry_infos.get(CONF_TEMP_MAX, 0.0))
self._attr_min_temp = float(entry_infos.get(CONF_TEMP_MIN, 0.0))
if (step := entry_infos.get(CONF_STEP_TEMPERATURE)) is not None:
self._attr_target_temperature_step = step
self._attr_preset_modes = []
self._vtherm_preset_modes = []
self._cycle_min = max(1, entry_infos.get(CONF_CYCLE_MIN, 1))
# Initialize underlying entities (will be done in subclasses)
self._underlyings = []
self._temp_sensor_entity_id = entry_infos.get(CONF_TEMP_SENSOR)
self._last_seen_temp_sensor_entity_id = entry_infos.get(
CONF_LAST_SEEN_TEMP_SENSOR
)
self._ext_temp_sensor_entity_id = entry_infos.get(CONF_EXTERNAL_TEMP_SENSOR)
self.set_hvac_list()
self._unit = self._hass.config.units.temperature_unit
# Will be restored if possible
self._state_manager = StateManager()
self._support_flags = SUPPORT_FLAGS
# Preset will be initialized from Number entities
self._presets: dict[str, Any] = {} # presets
self._presets_away: dict[str, Any] = {} # presets_away
self._humidity = None
self._fan_mode = None
self._swing_mode = None
self._swing_horizontal_mode = None
self._cur_temp = None
self._cur_ext_temp = None
self._last_temperature_measure = self.now
self._last_ext_temperature_measure = self.now
self._total_energy = None
_LOGGER.debug("%s - post_init_ resetting energy to None", self)
# Read the parameter from configuration.yaml if it exists
short_ema_params = DEFAULT_SHORT_EMA_PARAMS
if api and api.short_ema_params:
short_ema_params = api.short_ema_params
self._ema_algo = ExponentialMovingAverage(
self.name,
short_ema_params.get("halflife_sec"),
# Needed for time calculation
get_tz(self._hass),
# two digits after the coma for temperature slope calculation
short_ema_params.get("precision"),
short_ema_params.get("max_alpha"),
)
self._is_central_mode = not (
entry_infos.get(CONF_USE_CENTRAL_MODE) is False
) # Default value (None) is True
self._is_used_by_central_boiler = (
entry_infos.get(CONF_USED_BY_CENTRAL_BOILER) is True
)
self._max_on_percent = api.max_on_percent
_LOGGER.debug(
"%s - Creation of a new VersatileThermostat entity: unique_id=%s",
self,
self.unique_id,
)
async def async_added_to_hass(self):
"""Run when entity about to be added."""
_LOGGER.debug("%s - Calling async_added_to_hass", self)
await super().async_added_to_hass()
self.async_on_remove(
async_track_state_change_event(
self.hass,
[self._temp_sensor_entity_id],
self._async_temperature_changed,
)
)
if self._last_seen_temp_sensor_entity_id:
self.async_on_remove(
async_track_state_change_event(
self.hass,
[self._last_seen_temp_sensor_entity_id],
self._async_last_seen_temperature_changed,
)
)
if self._ext_temp_sensor_entity_id:
self.async_on_remove(
async_track_state_change_event(
self.hass,
[self._ext_temp_sensor_entity_id],
self._async_ext_temperature_changed,
)
)
self.async_on_remove(self.remove_thermostat)
# issue 428. Link to others entities will start at link
# await self.async_startup()
async def async_will_remove_from_hass(self):
"""Try to force backup of entity"""
self._is_removed = True
_LOGGER.debug(
"%s - force write before remove. Energy is %s", self, self.total_energy
)
# Force dump in background
await restore_async_get(self.hass).async_dump_states()
def remove_thermostat(self):
"""Called when the thermostat will be removed"""
_LOGGER.info("%s - Removing thermostat", self)
self.stop_recalculate_later()
# Cancel scheduler timers so leftover async_call_later handles cannot fire
# after the entity has been unloaded (e.g. on cycle duration config change).
if self._cycle_scheduler:
self._cycle_scheduler.shutdown()
# stop listening for all managers
for manager in self._managers:
manager.stop_listening()
for under in self._underlyings:
under.remove_entity()
def register_cycle_callback(
self,
on_start: Callable | None = None,
):
"""Register callbacks for TPI cycle events.
Args:
on_start: Callback called at the start of each TPI cycle
Signature: async def callback(on_time_sec, off_time_sec, on_percent, hvac_mode)
"""
if on_start:
self._on_cycle_start_callbacks.append(on_start)
_LOGGER.debug("%s - Registered cycle start callback: %s", self, on_start)
# Register on the CycleScheduler if available
if self._cycle_scheduler:
self._cycle_scheduler.register_cycle_start_callback(on_start)
def stop_recalculate_later(self):
"""Stop any scheduled call later tasks if any."""
if self._cancel_recalculate_later:
self._cancel_recalculate_later() # pylint: disable=not-callable
self._cancel_recalculate_later = None
async def async_startup(self, central_configuration):
"""Triggered on startup, used to get old state and set internal states
accordingly. This is triggered by VTherm API"""
write_event_log(_LOGGER, self, "Start up VTherm")
_LOGGER.debug("%s - Calling async_startup_internal", self)
# need_write_state = False
# Retry loading external feature managers: a plugin may have registered
# its factory after this thermostat was built (load order between the
# core and the plugin is not guaranteed). Newly created managers are
# then started in the loop below.
self._load_external_feature_managers()
# start listening for all managers
for manager in self._managers:
await manager.start_listening()
await self.get_my_previous_state()
# Start underlyings and register cycle callbacks on the scheduler
for under in self._underlyings:
under.startup()
if self._cycle_scheduler:
for cb in self._on_cycle_start_callbacks:
self._cycle_scheduler.register_cycle_start_callback(cb)
# init presets. Should be after underlyings init because for over_climate it uses the hvac_modes
await self.init_presets(central_configuration)
temperature_state = self.hass.states.get(self._temp_sensor_entity_id)
if temperature_state and temperature_state.state not in (
STATE_UNAVAILABLE,
STATE_UNKNOWN,
):
_LOGGER.debug(
"%s - temperature sensor have been retrieved: %.1f",
self,
float(temperature_state.state),
)
await self._async_update_temp(temperature_state)
if self._ext_temp_sensor_entity_id:
ext_temperature_state = self.hass.states.get(
self._ext_temp_sensor_entity_id
)
if ext_temperature_state and ext_temperature_state.state not in (
STATE_UNAVAILABLE,
STATE_UNKNOWN,
):
_LOGGER.debug(
"%s - external temperature sensor have been retrieved: %.1f",
self,
float(ext_temperature_state.state),
)
await self._async_update_ext_temp(ext_temperature_state)
else:
_LOGGER.debug(
"%s - external temperature sensor have NOT been retrieved "
"cause unknown or unavailable",
self,
)
else:
_LOGGER.debug(
"%s - external temperature sensor have NOT been retrieved "
"cause no external sensor",
self,
)
self._is_startup_done = True
if self.is_ready:
await self.init_underlyings_completed()
async def init_underlyings_completed(self, under_entity_id: Optional[str] = None):
"""All underlyings have been initialized. Then we can finish our initialization"""
_LOGGER.debug("%s - Calling init_underlyings_completed", self)
if not self.is_ready:
return
# We:
# - refresh all managers states,
# - calculate the current state of the VTherm (it depends on the managers states and the requested state)
# - check if the initial conditions are met
# - force the first cycle if changes has been detected
# in over_climate, we have to remove the FROST preset for a COOL only device
if self._ac_mode and VThermPreset.FROST in self._vtherm_preset_modes and HVACMode.HEAT not in self.hvac_modes:
self._vtherm_preset_modes.remove(VThermPreset.FROST)
if VThermPreset.FROST in self._attr_preset_modes:
self._attr_preset_modes.remove(VThermPreset.FROST)
_LOGGER.debug(
"%s - removed FROST preset for COOL only device, new presets: %s",
self,
self._attr_preset_modes,
)
# refresh states for all managers
for manager in self._managers:
await manager.refresh_state()
# need_write_state = True
await self.update_states(force=True)
self.recalculate()
# check initial state should be done after the current state has been calculated and so after the manager has been updated
# issue 1654 - initial state check should be done after the underlyings has come to life
# await self._check_initial_state()
self.reset_last_change_time_from_vtherm()
def restore_specific_previous_state(self, old_state: State):
"""Should be overridden in each specific thermostat
if a specific previous state or attribute should be
restored
"""
async def get_my_previous_state(self):
"""Try to get my previous state"""
# Check If we have an old state
old_state = await self.async_get_last_state()
_LOGGER.debug(
"%s - Calling get_my_previous_state old_state is %s", self, old_state
)
if old_state is not None:
# Restore current_state
if current_state_attr := old_state.attributes.get(ATTR_CURRENT_STATE, None):
current_state = VThermState.from_dict(current_state_attr)
self._state_manager.current_state.set_state(
hvac_mode=current_state.hvac_mode,
target_temperature=current_state.target_temperature,
preset=current_state.preset,
)
else:
# Try to init current_state with old temperature, preset and mode
self._state_manager.current_state.set_state(
target_temperature=old_state.attributes.get(ATTR_TEMPERATURE, None),
preset=VThermPreset(old_state.attributes.get(ATTR_PRESET_MODE, None) or VThermPreset.NONE),
hvac_mode=old_state.state if isinstance(old_state.state, VThermHvacMode) else from_ha_hvac_mode(old_state.state),
)
# If we have no initial temperature set with min (or max depending on ac_mode)
if self._state_manager.current_state.target_temperature is None:
self._state_manager.current_state.set_target_temperature(self.max_temp if self._ac_mode else self.min_temp)
_LOGGER.warning(
"%s - Undefined target temperature, falling back to %s",
self,
self._state_manager.current_state.target_temperature,
)
# Restore requested_state or default with current_state
if requested_state_attr := old_state.attributes.get(ATTR_REQUESTED_STATE, None):
requested_state = VThermState.from_dict(requested_state_attr)
self._state_manager.requested_state.set_state(
hvac_mode=requested_state.hvac_mode,
target_temperature=requested_state.target_temperature,
preset=requested_state.preset,
)
else:
# Try to init requested_state with old temperature, preset and mode
self._state_manager.requested_state.set_state(
target_temperature=old_state.attributes.get(ATTR_TEMPERATURE, None),
preset=VThermPreset(old_state.attributes.get(ATTR_PRESET_MODE, None) or VThermPreset.NONE),
hvac_mode=old_state.state if isinstance(old_state.state, VThermHvacMode) else from_ha_hvac_mode(old_state.state),
)
self._hvac_off_reason = old_state.attributes.get(HVAC_OFF_REASON_NAME, None)
# Try to get total_energy from specific_states (new format) or root level (old format)
specific_states = old_state.attributes.get("specific_states", {})
self._hvac_mode_reason = specific_states.get(HVAC_MODE_REASON_NAME, self._hvac_off_reason)
old_total_energy = specific_states.get(ATTR_TOTAL_ENERGY)
if old_total_energy is None:
# Fallback to root level for backward compatibility
old_total_energy = old_state.attributes.get(ATTR_TOTAL_ENERGY)
self._total_energy = old_total_energy if old_total_energy is not None else 0
_LOGGER.debug(
"%s - get_my_previous_state restored energy is %s",
self,
self._total_energy,
)
self.restore_specific_previous_state(old_state)
# Restore all managers states from previous state
for manager in self._managers:
manager.restore_state(old_state)
else:
# No previous state, try and restore defaults
if self._state_manager.current_state.target_temperature is None:
self._state_manager.current_state.set_target_temperature(self.max_temp if self._ac_mode else self.min_temp)
_LOGGER.warning("No previously saved temperature, setting to %s", self._state_manager.current_state.target_temperature)
self._total_energy = 0
_LOGGER.debug(
"%s - get_my_previous_state no previous state energy is %s",
self,
self._total_energy,
)
if not self.is_on and self.hvac_off_reason is None:
self.set_hvac_off_reason(HVAC_OFF_REASON_MANUAL)
_LOGGER.info(
"%s - restored current state is %s, requested state is %s",
self,
self._state_manager.current_state,
self._state_manager.requested_state,
)
def __str__(self) -> str:
return self.name
def set_hvac_list(self):
"""Set the hvac list depending on ac_mode"""
self._hvac_list = self.build_hvac_list()
self._str_hvac_list = [to_ha_hvac_mode(mode) for mode in self._hvac_list if to_ha_hvac_mode(mode) is not None]
@callback
def async_registry_entry_updated(self):
"""update the entity if the config entry have been updated
Note: this don't work either
"""
write_event_log(_LOGGER, self, "The config entry have been updated")
@callback
async def entry_update_listener(self, _, config_entry: ConfigEntry) -> None: # hass: HomeAssistant,
"""Called when the entry have changed in ConfigFlow"""
_LOGGER.info("%s - Change entry with the values: %s", self, config_entry.data)
async def init_presets(self, central_config):
"""Init all presets of the VTherm"""
# If preset central config is used and central config is set,
# take the presets from central config
vtherm_api: VersatileThermostatAPI = VersatileThermostatAPI.get_vtherm_api()
presets: dict[str, Any] = {}
presets_away: dict[str, Any] = {}
def calculate_presets(items, use_central_conf_key):
presets: dict[str, Any] = {}
config_id = self._unique_id
if central_config and self._entry_infos.get(use_central_conf_key, False) is True:
config_id = central_config.entry_id
for key, preset_name in items:
_LOGGER.debug("%s - looking for key=%s, preset_name=%s", self, key, preset_name)
# removes preset_name frost if heat is not in hvac_modes. vtherm_hvac_modes is initialized when the underlyings are initialized.
# So it may be not be ready yet here. In that case, the FROST is added anyway. So it is possible that FROST preset in a COOL only device
if len(self.vtherm_hvac_modes) == 0 and key == VThermPreset.FROST and VThermHvacMode_HEAT not in self.vtherm_hvac_modes:
_LOGGER.debug("%s - removing preset_name %s which reserved for HEAT devices", self, preset_name)
continue
value = vtherm_api.get_temperature_number_value(config_id=config_id, preset_name=preset_name)
if value is not None:
presets[key] = value
else:
_LOGGER.debug("%s - preset_name %s not found in VTherm API", self, preset_name)
presets[key] = self._attr_max_temp if self._ac_mode else self._attr_min_temp
return presets
# Calculate all presets
presets = calculate_presets(
CONF_PRESETS_WITH_AC.items() if self._ac_mode else CONF_PRESETS.items(),
CONF_USE_PRESETS_CENTRAL_CONFIG,
)
# refacto
# if self._entry_infos.get(CONF_USE_PRESENCE_FEATURE) is True:
if self._presence_manager.is_configured:
presets_away = calculate_presets(
(CONF_PRESETS_AWAY_WITH_AC.items() if self._ac_mode else CONF_PRESETS_AWAY.items()),
CONF_USE_PRESETS_CENTRAL_CONFIG,
)
# aggregate all available presets now
self._presets: dict[str, Any] = presets
self._presets_away: dict[str, Any] = presets_away
# Calculate all possible presets
self._attr_preset_modes = [VThermPreset.NONE]
if len(self._presets):
self._support_flags = SUPPORT_FLAGS | ClimateEntityFeature.PRESET_MODE
for key, _ in CONF_PRESETS.items():
preset_value = self.find_preset_temp(key)
if preset_value is not None and preset_value > 0:
self._attr_preset_modes.append(key)
_LOGGER.debug("%s - After adding presets, preset_modes to %s", self, self._attr_preset_modes)
else:
_LOGGER.debug("%s - No preset_modes", self)
if self._motion_manager.is_configured:
self._attr_preset_modes.append(VThermPreset.ACTIVITY)
# transform _attr_preset_modes into _vtherm_preset_modes
self._vtherm_preset_modes = [VThermPreset(mode) for mode in self._attr_preset_modes]
# Re-applicate the last preset if any to take change into account
if self._state_manager.current_state.preset and self._state_manager.current_state.preset != VThermPreset.NONE:
await self.async_set_preset_mode_internal(self._state_manager.current_state.preset)
##
## Properties overriden from ClimateEntity (or not)
##
@property
def is_initialized(self) -> bool:
"""Check if all underlyings are initialized
This is useful only for over_climate in which we
should have found the underlying climate to be operational"""
for under in self._underlyings:
if not under.is_initialized:
return False
return True
@property
def is_ready(self) -> bool:
"""Check if all underlyings are ready (initialized and startup is complete)"""
if not self._is_ready:
self._is_ready = self._is_startup_done and self.is_initialized
return self._is_ready
@property
def vtherm_hvac_modes(self) -> list[VThermHvacMode]:
"""List of available VTherm operation modes."""
return self._hvac_list
@property
def hvac_modes(self) -> list[str]:
"""List of available VTherm operation modes."""
return self._str_hvac_list
def build_hvac_list(self) -> list[VThermHvacMode]:
"""Build the hvac list depending on ac_mode"""
if self._ac_mode:
return [VThermHvacMode_HEAT, VThermHvacMode_COOL, VThermHvacMode_OFF]
else:
return [VThermHvacMode_HEAT, VThermHvacMode_OFF]
@property
def is_over_climate(self) -> bool:
"""True if the Thermostat is over_climate"""
return False
@property
def is_over_switch(self) -> bool:
"""True if the Thermostat is over_switch"""
return False
@property
def has_prop(self) -> bool:
"""True if the Thermostat uses a proportional algorithm (TPI, SmartPI)"""
return False
@property
def safe_on_percent(self) -> float:
"""Return the on_percent safe value"""
return 0
@property
def is_over_valve(self) -> bool:
"""True if the Thermostat is over_valve"""
return False
@property
def device_info(self) -> DeviceInfo:
"""Return the device info."""
return DeviceInfo(
entry_type=None,
identifiers={(DOMAIN, self._unique_id)},
name=self._name,
manufacturer=DEVICE_MANUFACTURER,
model=DOMAIN,
)
@property
def unique_id(self) -> str:
return self._unique_id
@property
def should_poll(self) -> bool:
return False
@property
def name(self) -> str:
return self._name
@property
def ac_mode(self) -> bool | None:
"""Get the ac_mode of the Themostat"""
return self._ac_mode
@property
def temperature_unit(self) -> str:
"""Return the unit of measurement."""
return self._unit
@property
def ema_temperature(self) -> float | None:
"""Return the EMA temperature."""
return self._ema_temp
@property
def hvac_mode(self) -> HVACMode | None:
"""Return current operation."""
return to_legacy_ha_hvac_mode(self._state_manager.current_state.hvac_mode)
@property
def vtherm_hvac_mode(self) -> VThermHvacMode | None:
"""Return current operation."""
return self._state_manager.current_state.hvac_mode
@property
def is_recalculate_scheduled(self) -> bool:
"""Return true if a recalculation is scheduled."""
return self._cancel_recalculate_later is not None
@property
def is_used_by_central_boiler(self) -> HVACAction | None:
"""Return true is the VTherm is configured to be used by
central boiler"""
return self._is_used_by_central_boiler
@property
def target_temperature(self) -> float | None:
"""Return the temperature we try to reach."""
return self._state_manager.current_state.target_temperature
@property
def regulated_target_temperature(self) -> float | None:
"""Return the regulated target temperature used to drive the underlying.
The base implementation returns the plain target temperature. Over
climate thermostats override this to expose their regulated value.
"""
return self.target_temperature
@property
def underlying_fan_modes(self) -> list[str] | None:
"""Return the fan modes exposed by the underlying climate(s).
The base implementation returns None because most thermostats do not
expose an underlying climate with fan control.
"""
return None
async def async_set_underlying_fan_mode(self, fan_mode: str) -> None:
"""Send a fan mode to the underlying climate(s).
The base implementation is a no-op for thermostats without fan control.
"""
@property
def supported_features(self) -> ClimateEntityFeature:
"""Return the list of supported features."""
return self._support_flags
@property
def should_device_be_active(self) -> bool:
"""Returns true if one underlying is active"""
for under in self._underlyings:
if under.should_device_be_active:
return True
return False
@property
def is_device_active(self) -> bool:
"""Returns true if one underlying is active"""
for under in self._underlyings:
if under.is_device_active:
return True
return False
@property
def device_actives(self) -> int:
"""Calculate the active devices"""
ret = []
for under in self._underlyings:
if under.is_device_active:
ret.append(under.entity_id)
return ret
@property
def nb_device_actives(self) -> int:
"""Calculate the number of active devices"""
return len(self.device_actives)
@property
def current_temperature(self) -> float | None:
"""Return the sensor temperature."""
return self._cur_temp
@property
def current_outdoor_temperature(self) -> float | None:
"""Return the outdoor sensor temperature."""
return self._cur_ext_temp
@property
def total_energy(self) -> float | None:
"""Returns the total energy calculated for this thermostat"""
if self._total_energy is not None:
return round(self._total_energy, 2)
else:
return None
@property
def overpowering_state(self) -> bool | None:
"""Get the overpowering_state"""
return self._power_manager.overpowering_state
@property
def power_manager(self) -> FeaturePowerManager | None:
"""Get the power manager"""
return self._power_manager
@property
def is_overpowering_detected(self) -> bool:
"""Return True when power shedding is currently active."""
return (
self._power_manager.is_overpowering_detected
if self._power_manager is not None
else False
)
@property
def presence_manager(self) -> FeaturePresenceManager | None:
"""Get the presence manager"""
return self._presence_manager
@property
def motion_manager(self) -> FeatureMotionManager | None:
"""Get the motion manager"""
return self._motion_manager
@property
def window_manager(self) -> FeatureWindowManager | None:
"""Get the window manager"""
return self._window_manager
@property
def safety_manager(self) -> FeatureSafetyManager | None:
"""Get the safety manager"""
return self._safety_manager
@property
def auto_start_stop_manager(self) -> FeatureAutoStartStopManager | None:
"""Get the auto start/stop manager (only implemented in over_climate)"""
return self._auto_start_stop_manager
@property
def lock_manager(self) -> FeatureLockManager | None:
"""Get the lock manager"""
return self._lock_manager
@property
def timed_preset_manager(self) -> FeatureTimedPresetManager:
"""Get the timed preset manager"""
return self._timed_preset_manager
@property
def heating_failure_detection_manager(self) -> FeatureHeatingFailureDetectionManager:
"""Get the heating failure detection manager"""
return self._heating_failure_detection_manager
@property
def repair_incorrect_state_manager(self) -> FeatureRepairIncorrectStateManager:
"""Get the repair incorrect state manager"""
return self._repair_incorrect_state_manager
@property
def current_state(self) -> VThermState | None:
"""Get the current state"""
return self._state_manager.current_state
@property
def requested_state(self) -> VThermState | None:
"""Get the requested state"""
return self._state_manager.requested_state
@property
def window_state(self) -> str | None:
"""Get the window_state"""
return self._window_manager.window_state
@property
def window_auto_state(self) -> str | None:
"""Get the window_auto_state"""
return self._window_manager.window_auto_state
@property
def is_window_bypass(self) -> bool | None:
"""Get the Window Bypass"""
return self._window_manager.is_window_bypass
@property
def safety_state(self) -> str | None:
"""Get the safety_state"""
return self._safety_manager.safety_state
@property
def motion_state(self) -> str | None:
"""Get the motion_state"""
return self._motion_manager.motion_state
@property
def presence_state(self) -> str | None:
"""Get the presence_state"""
return self._presence_manager.presence_state
# =========================================================================
# PUBLIC PROPERTIES FOR HANDLER ACCESS
# These properties expose internal state to handlers without breaking encapsulation
# =========================================================================
@property
def cycle_min(self) -> int:
"""Return the cycle duration in minutes."""
return self._cycle_min
@property
def entry_infos(self) -> ConfigData:
"""Return the configuration entry data."""
return self._entry_infos
@property
def hass(self) -> HomeAssistant:
"""Return the Home Assistant instance.
This overrides the Entity.hass property to return our stored _hass
reference during initialization (before async_added_to_hass is called).
"""
return self._hass
@hass.setter
def hass(self, value: HomeAssistant):
"""Set the Home Assistant instance.
This is called by the entity platform when adding the entity.
"""
self._hass = value
@property
def is_removed(self) -> bool:
"""Return True if the thermostat has been removed."""
return self._is_removed
@property
def ext_temp_sensor_entity_id(self) -> str | None:
"""Return the external temperature sensor entity ID."""
return self._ext_temp_sensor_entity_id
@property
def max_on_percent(self) -> float | None:
"""Return the maximum on percentage."""
return self._max_on_percent
@property
def tpi_coef_int(self) -> float | None:
"""Return the TPI internal coefficient."""
return self._tpi_coef_int
@tpi_coef_int.setter
def tpi_coef_int(self, value: float):
"""Set the TPI internal coefficient."""
self._tpi_coef_int = value
@property
def tpi_coef_ext(self) -> float | None:
"""Return the TPI external coefficient."""
return self._tpi_coef_ext
@tpi_coef_ext.setter
def tpi_coef_ext(self, value: float):
"""Set the TPI external coefficient."""
self._tpi_coef_ext = value
@property
def tpi_threshold_low(self) -> float:
"""Return the TPI low threshold."""
return self._tpi_threshold_low or 0.0
@tpi_threshold_low.setter
def tpi_threshold_low(self, value: float):
"""Set the TPI low threshold."""
self._tpi_threshold_low = value
@property
def tpi_threshold_high(self) -> float:
"""Return the TPI high threshold."""
return self._tpi_threshold_high or 0.0
@tpi_threshold_high.setter
def tpi_threshold_high(self, value: float):
"""Set the TPI high threshold."""
self._tpi_threshold_high = value
@property
def minimal_activation_delay(self) -> int:
"""Return the minimal activation delay in seconds."""
return self._minimal_activation_delay or 0
@minimal_activation_delay.setter
def minimal_activation_delay(self, value: int):
"""Set the minimal activation delay in seconds."""
self._minimal_activation_delay = value
@property
def minimal_deactivation_delay(self) -> int:
"""Return the minimal deactivation delay in seconds."""
return self._minimal_deactivation_delay or 0
@minimal_deactivation_delay.setter
def minimal_deactivation_delay(self, value: int):
"""Set the minimal deactivation delay in seconds."""
self._minimal_deactivation_delay = value
@property
def proportional_function(self) -> str | None:
"""Return the proportional function type (TPI, SmartPI)."""
return self._proportional_function
@property
def prop_algorithm(self):
"""Return the proportional algorithm instance."""
return self._prop_algorithm
@prop_algorithm.setter
def prop_algorithm(self, value):
"""Set the proportional algorithm instance."""
self._prop_algorithm = value
@property
def cycle_scheduler(self):
"""Return the CycleScheduler instance, if any."""
return self._cycle_scheduler
@property
def underlyings(self) -> list:
"""Return the list of underlying entities."""
return self._underlyings
@property
def proportional_algorithm(self):
"""Get the eventual ProportionalAlgorithm"""
return None
@property
def last_temperature_measure(self) -> datetime | None:
"""Get the last temperature datetime"""
return self._last_temperature_measure
@property
def last_ext_temperature_measure(self) -> datetime | None:
"""Get the last external temperature datetime"""
return self._last_ext_temperature_measure
@property
def preset_mode(self) -> str | None:
"""Return the current preset mode comfort, eco, boost,...,"""
return str(self._state_manager.current_state.preset)
@property
def preset_modes(self) -> list[str] | None:
"""Return a list of available preset modes.
Requires ClimateEntityFeature.PRESET_MODE.
"""
return self._attr_preset_modes
@property
def vtherm_preset_mode(self) -> VThermPreset | None:
"""Return the current preset mode comfort, eco, boost,...,"""
return self._state_manager.current_state.preset
@property
def vtherm_preset_modes(self) -> list[str] | None:
"""Return a list of available preset modes.
Requires ClimateEntityFeature.PRESET_MODE.
"""
return self._vtherm_preset_modes
@property
def last_temperature_slope(self) -> float | None:
"""Return the last temperature slope curve if any"""
return self._window_manager.last_slope
@property
def nb_underlying_entities(self) -> int:
"""Returns the number of underlying entities"""
return len(self._underlyings)
@property
def underlying_entities(self) -> list | None:
"""Returns the underlying entities"""
return self._underlyings
@property
def activable_underlying_entities(self) -> list | None:
"""Returns the activable underlying entities for controlling
the central boiler"""
return self.underlying_entities
@property
def all_underlying_entities(self) -> list | None:
"""Returns all underlying entities for controling the central boiler"""
return self.activable_underlying_entities
def find_underlying_by_entity_id(self, entity_id: str) -> Entity | None:
"""Get the underlying entity by a entity_id"""
for under in self._underlyings:
if under.entity_id == entity_id:
return under
return None
@property
def is_on(self) -> bool:
"""True if the VTherm is on (! HVAC_OFF)"""
return self.vtherm_hvac_mode and self.vtherm_hvac_mode != VThermHvacMode_OFF
@property
def is_controlled_by_central_mode(self) -> bool:
"""Returns True if this VTherm can be controlled by the central_mode"""
return self._is_central_mode
@property
def last_central_mode(self) -> str | None:
"""Returns the last central_mode taken into account.
Is None if the VTherm is not controlled by central_mode"""
return self._last_central_mode
@property
def use_central_config_temperature(self):
"""True if this VTHerm uses the central configuration temperature"""
return self._use_central_config_temperature
@property
def hvac_off_reason(self) -> str | None:
"""Returns the reason of the last switch to HVAC_OFF
This is useful for features that turns off the VTherm like
window detection or auto-start-stop"""
return self._hvac_off_reason
@property
def hvac_mode_reason(self) -> str | None:
"""Returns the reason why the current hvac_mode is forced.
Unlike hvac_off_reason, this is set whatever the forced hvac_mode is
(off, fan_only, dry, ...) so the UI can explain why the VTherm is not
in the requested mode."""
return self._hvac_mode_reason
@property
def temperature_reason(self) -> str | None:
"""Returns the reason of the target temperature
This is useful for features that changes the VTherm like window detection or power management"""
return self._temperature_reason
@property
def is_sleeping(self) -> bool:
"""True if the thermostat is in sleep mode. Only for over_climate with valve regulation"""
return False
@property
def have_valve_regulation(self) -> bool:
"""True if the Thermostat is regulated by valve"""
return False
@property
def power_percent(self) -> float | None:
"""Get the current on_percent as a percentage value. valid only for Vtherm with a TPI algo
Get the current on_percent value"""
prop_algo = getattr(self, '_prop_algorithm', None)
if prop_algo and prop_algo.on_percent is not None:
return round(prop_algo.on_percent * 100, 0)
else:
return None
@property
def on_percent(self) -> float | None:
"""Get the current on_percent value. valid only for Vtherm with a TPI algo"""
prop_algo = getattr(self, '_prop_algorithm', None)
if prop_algo and prop_algo.on_percent is not None:
return prop_algo.on_percent
else:
return None
@property
def vtherm_type(self) -> str | None:
"""Return the type of thermostat"""
return None
@property
def config_entry(self) -> ConfigEntry | None:
"""Return the config entry associated with the thermostat."""
return self._hass.config_entries.async_get_entry(self.unique_id)
def underlying_entity_id(self, index=0) -> str | None:
"""The climate_entity_id. Added for retrocompatibility reason"""
if index < self.nb_underlying_entities:
return self.underlying_entity(index).entity_id
else:
return None
def underlying_entity(self, index=0) -> UnderlyingEntity | None:
"""Get the underlying entity at specified index"""
if index < self.nb_underlying_entities:
return self._underlyings[index]
else:
return None
def turn_aux_heat_on(self) -> None:
"""Turn auxiliary heater on."""
raise NotImplementedError()
@overrides
async def async_turn_aux_heat_on(self) -> None:
"""Turn auxiliary heater on."""
raise NotImplementedError()
@overrides
def turn_aux_heat_off(self) -> None:
"""Turn auxiliary heater off."""
raise NotImplementedError()
@overrides
async def async_turn_aux_heat_off(self) -> None:
"""Turn auxiliary heater off."""
raise NotImplementedError()
##
## Entry events (from linked devices or users)
##
@staticmethod
def check_lock(func):
"""Decorator to check if the thermostat is locked."""
async def wrapper(self, *args, **kwargs):
if self.lock_manager.check_is_locked(func.__name__):
return
return await func(self, *args, **kwargs)
return wrapper
@overrides
@check_lock
async def async_set_hvac_mode(self, hvac_mode: VThermHvacMode): # , need_control_heating=True):
"""Set new target hvac mode. Uses the HA VThermHvacMode enum to respect the original type."""
write_event_log(_LOGGER, self, f"Set hvac mode: {hvac_mode}")
if hvac_mode is None:
return
# Change the requested state
self._state_manager.requested_state.set_hvac_mode(hvac_mode)
await self.update_states(force=False)
@overrides
@check_lock
async def async_set_preset_mode(self, preset_mode: str):
"""Set new preset mode."""
# We accept a new preset when:
# 1. last_central_mode is not set,
# 2. or last_central_mode is AUTO,
# 3. or last_central_mode is CENTRAL_MODE_FROST_PROTECTION and preset_mode is
# VThermPreset.FROST (to be abel to re-set the preset_mode)
write_event_log(_LOGGER, self, f"Set preset mode: {preset_mode}")
vtherm_preset_mode = VThermPreset(preset_mode)
accept = self._last_central_mode in [
None,
CENTRAL_MODE_AUTO,
CENTRAL_MODE_COOL_ONLY,
CENTRAL_MODE_HEAT_ONLY,
CENTRAL_MODE_STOPPED,
] or (self._last_central_mode == CENTRAL_MODE_FROST_PROTECTION and preset_mode == VThermPreset.FROST)
if not accept:
_LOGGER.info(
"%s - Impossible to change the preset to %s because central mode is %s",
self,
preset_mode,
self._last_central_mode,
)
return
await self.async_set_preset_mode_internal(vtherm_preset_mode) # overwrite_saved_preset=overwrite_saved_preset)
await self.update_states(force=True)
async def async_set_preset_mode_internal(self, preset_mode: VThermPreset): # , overwrite_saved_preset=True):
"""Set new preset mode."""
_LOGGER.info("%s - Set requested preset_mode: %s", self, preset_mode)
if preset_mode not in (self._vtherm_preset_modes or []) and preset_mode not in HIDDEN_PRESETS:
raise ValueError(f"Got unsupported preset_mode {preset_mode}. Must be one of {self._vtherm_preset_modes}")
self._state_manager.requested_state.set_preset(preset_mode)
async def async_set_fan_mode(self, fan_mode: str):
"""Set new target fan mode."""
_LOGGER.info("%s - Set fan mode: %s", self, fan_mode)
return
async def async_set_humidity(self, humidity: int):
"""Set new target humidity."""
write_event_log(_LOGGER, self, f"Set humidity: {humidity}")
return
@check_lock
async def async_set_temperature(self, **kwargs):
"""Set new requested target temperature and turn off any active presets."""
temperature = kwargs.get(ATTR_TEMPERATURE)
write_event_log(_LOGGER, self, f"Set target temp: {temperature}")
if temperature is None:
return
self._state_manager.requested_state.set_target_temperature(temperature)
self._state_manager.requested_state.set_preset(VThermPreset.NONE)
await self.update_states(force=True)
async def set_preset_temperature(
self,
preset: str,
temperature: float | None = None,
temperature_away: float | None = None,
):
"""Called by a to change the temperature of a preset
data:
preset: boost
temperature: 17.8
temperature_away: 15
target:
entity_id: climate.thermostat_2
"""
write_event_log(_LOGGER, self, f"Calling SERVICE_SET_PRESET_TEMPERATURE, preset: {preset}, temperature: {temperature}, temperature_away: {temperature_away}")
if preset in self._presets:
if temperature is not None:
self._presets[preset] = temperature
if self._presence_manager.is_configured and temperature_away is not None:
self._presets_away[self.get_preset_away_name(preset)] = temperature_away
else:
_LOGGER.warning("%s - No preset %s configured for this thermostat. ", self, preset)
if preset in self._presets:
if temperature is not None:
self._presets[preset] = temperature
if self._presence_manager.is_configured and temperature_away is not None:
self._presets_away[self.get_preset_away_name(preset)] = temperature_away
else:
_LOGGER.warning(
"%s - No preset %s configured for this thermostat. Ignoring set_preset_temperature call",
self,
preset,
)
# If the changed preset is active, change the current temperature
# Issue #119 - reload new preset temperature also in ac mode
is_active_preset = preset.startswith(self.preset_mode)
# Also check if activity is the current preset and the changed preset
# is the current motion sub-preset (e.g. comfort or boost)
if not is_active_preset and self.preset_mode == VThermPreset.ACTIVITY and self._motion_manager.is_configured:
current_motion_preset = self._motion_manager.get_current_motion_preset()
is_active_preset = preset.startswith(current_motion_preset)
if is_active_preset:
self.requested_state.force_changed()
await self.update_states(force=True)
##
## Calculation and utility functions
##
async def update_states(self, force=False):
"""Update the states of the thermostat considering the requested state and the current state"""
if not self.is_ready:
_LOGGER.debug("%s - update_states is called but the entity is not initialized yet. Skip the update", self)
return False
changed = False
if self._state_manager.requested_state.is_changed:
if changed := await self._state_manager.calculate_current_state(self):
_LOGGER.info("%s - current state changed to %s", self, self._state_manager.current_state)
sub_need_control_heating = False
# Apply preset
if self._state_manager.current_state.is_preset_changed:
_LOGGER.info("%s - Applying new preset: %s", self, self.vtherm_preset_mode)
self.send_event(EventType.PRESET_EVENT, {"preset": self.preset_mode})
if self.preset_mode not in HIDDEN_PRESETS:
self._attr_preset_mode = self.preset_mode
# Reset auto_start_stop switch delay to allow immediate restart when preset changes
if self.auto_start_stop_manager:
self.auto_start_stop_manager.reset_switch_delay()
# Apply temperature
if self._state_manager.current_state.is_target_temperature_changed:
_LOGGER.info("%s - Applying new target temperature: %s", self, self.target_temperature)
self._attr_target_temperature = self.target_temperature
# Reset auto_start_stop switch delay to allow immediate restart when target temp changes
if self.auto_start_stop_manager:
self.auto_start_stop_manager.reset_switch_delay()
# Apply hvac_mode
if self._state_manager.current_state.is_hvac_mode_changed:
_LOGGER.info("%s - Applying new hvac mode: %s", self, self.vtherm_hvac_mode)
# Delegate to all underlying
for under in self._underlyings:
sub_need_control_heating = await under.set_hvac_mode(self.vtherm_hvac_mode) or sub_need_control_heating
self._attr_hvac_mode = to_legacy_ha_hvac_mode(self.vtherm_hvac_mode)
self.send_event(EventType.HVAC_MODE_EVENT, {"hvac_mode": str(self.vtherm_hvac_mode)})
# Remove eventual overpowering if we want to turn-off
if self.hvac_mode in [VThermHvacMode_OFF, VThermHvacMode_SLEEP] and self.power_manager.is_overpowering_detected:
await self.power_manager.set_overpowering(False)
if changed:
self.recalculate(force=force)
self.reset_last_change_time_from_vtherm()
await self.async_control_heating(force=force or sub_need_control_heating)
self.calculate_hvac_action()
self.update_custom_attributes()
self.async_write_ha_state()
else:
_LOGGER.debug("%s - current state did not change. Still %s", self, self._state_manager.current_state)
self._state_manager.requested_state.reset_changed()
self._state_manager.current_state.reset_changed()
return changed
async def async_control_heating(self, timestamp=None, force=False) -> bool:
"""The main function used to run the calculation at each cycle
Returns True if the cycle was successfully calculated, False otherwise."""
if timestamp:
# Timestamp is set only on periodical calls
write_event_log(_LOGGER, self, "Periodical control cycle started")
_LOGGER.debug(
"%s - Checking new cycle. hvac_mode=%s, safety_state=%s, preset_mode=%s, force=%s",
self,
self.vtherm_hvac_mode,
self._safety_manager.safety_state,
self.vtherm_preset_mode,
force,
)
if not self.is_ready:
_LOGGER.info("%s - async_control_heating is called but the entity is not ready yet (not initialized or startup not done). Skip the cycle", self)
return False
# check auto_window conditions
await self._window_manager.manage_window_auto(in_cycle=True)
if timestamp and await self._safety_manager.refresh_and_update_if_changed():
return False
if self._safety_manager.is_safety_detected and self.is_over_climate:
_LOGGER.debug("%s - End of cycle (safety and over climate)", self)
return True
previous_hvac_action = self.hvac_action
# Call specific control heating
await self._control_heating_specific(timestamp, force)
# Refresh external feature managers (provided by plugins) on every cycle.
# This is done after the specific control heating so that a manager can
# react to the regulated target temperature (e.g. drive the underlying
# fan mode), and before the publication block below so that any custom
# attributes updated here are written to HA within the same cycle.
# Only external managers are refreshed here: internal managers have their
# own dedicated per-cycle mechanisms. A per-manager guard prevents a
# faulty plugin from breaking the control loop.
for manager in self._external_managers:
try:
await manager.refresh_state()
except Exception as exc: # pylint: disable=broad-except
_LOGGER.error(
"%s - Error while refreshing external feature manager '%s': %s",
self,
manager.name,
exc,
)
# Check for heating/cooling failures (only for TPI VTherms)
await self._heating_failure_detection_manager.refresh_state()
# Check and repair state discrepancies
await self._repair_incorrect_state_manager.check_and_repair()
self.calculate_hvac_action()
should_publish = True
algo_handler = getattr(self, "_algo_handler", None)
if algo_handler and hasattr(algo_handler, "should_publish_intermediate"):
should_publish = algo_handler.should_publish_intermediate()
if not should_publish and previous_hvac_action != self.hvac_action:
should_publish = True
if should_publish:
self.update_custom_attributes()
self.async_write_ha_state()
# For each manager display the manager state in debug and send an event with the manager state
current_state = self._state_manager.current_state.to_dict()
current_state["room_temperature"] = self.current_temperature
current_state["outdoor_temperature"] = self.current_outdoor_temperature
current_state["hvac_action"] = str(self.hvac_action)
current_state["power_percent"] = self.power_percent
manager_states = {}
for manager in self._managers:
try:
manager_states[manager.__class__.__name__] = manager.is_detected
except RuntimeError as e:
_LOGGER.error("%s - Error while getting is_detected state of manager %s: %s", self, manager.__class__.__name__, e)
_LOGGER.debug("%s - End of cycle. current_state: %s, managers_states: %s", self, current_state, manager_states)
return True
async def _control_heating_specific(self, timestamp=None, force=False):
"""To be overridden by subclasses."""
pass
def reset_last_change_time_from_vtherm(self, old_preset_mode: VThermPreset | None = None): # pylint: disable=unused-argument
"""Reset to now the last change time"""
self._last_change_time_from_vtherm = self.now
_LOGGER.debug(
"%s - last_change_time is now %s", self, self._last_change_time_from_vtherm
)
def reset_last_temperature_time(self, old_preset_mode: VThermPreset | str | None = None):
"""Reset to now the last temperature time if conditions are satisfied"""
if self._state_manager.current_state.preset not in HIDDEN_PRESETS and old_preset_mode not in HIDDEN_PRESETS:
self._last_temperature_measure = self._last_ext_temperature_measure = (
self.now
)
def find_preset_temp(self, preset_mode: VThermPreset):
"""Find the right temperature of a preset considering
the presence if configured"""
if preset_mode is None or preset_mode == VThermPreset.NONE:
return self._attr_max_temp if self._ac_mode and self.vtherm_hvac_mode == VThermHvacMode_COOL else self._attr_min_temp
if preset_mode == VThermPreset.SAFETY:
return self._state_manager.current_state.target_temperature
# In safety just keep the current target temperature,
# the thermostat should be off
if preset_mode == VThermPreset.POWER:
return self._power_manager.power_temperature
if preset_mode == VThermPreset.ACTIVITY:
motion_preset = self._motion_manager.get_current_motion_preset()
if self._ac_mode and self.vtherm_hvac_mode == VThermHvacMode_COOL:
motion_preset = motion_preset + PRESET_AC_SUFFIX
if motion_preset in self._presets:
if self._presence_manager.is_absence_detected:
return self._presets_away[motion_preset + PRESET_AWAY_SUFFIX]
else:
return self._presets[motion_preset]
else:
return None
else:
# Select _ac presets if in COOL Mode (or over_switch with _ac_mode) or OFF but requested is cool
preset_name = preset_mode
if (
(self._ac_mode and self.vtherm_hvac_mode == VThermHvacMode_COOL)
or (self.vtherm_hvac_mode == VThermHvacMode_OFF and self.requested_state.hvac_mode == VThermHvacMode_COOL)
# issue #1958 - when window_action=fan_only temporarily switches the mode to FAN_ONLY,
# the user's intent (requested_state) is still COOL, so we must use AC presets.
or (self.vtherm_hvac_mode == VThermHvacMode_FAN_ONLY and self.requested_state.hvac_mode == VThermHvacMode_COOL)
# when auto-start/stop temporarily switches the mode to DRY as its stop mode,
# the user's intent (requested_state) is still COOL, so we must use AC presets.
or (self.vtherm_hvac_mode == VThermHvacMode_DRY and self.requested_state.hvac_mode == VThermHvacMode_COOL)
# (self.is_over_switch and self._ac_mode)
# or self.vtherm_hvac_mode == VThermHvacMode_COOL
# or (self.vtherm_hvac_mode == VThermHvacMode_OFF and self.requested_state.hvac_mode == VThermHvacMode_COOL)
):
# if self._ac_mode and self.vtherm_hvac_mode == VThermHvacMode_COOL:
preset_name += PRESET_AC_SUFFIX
_LOGGER.info("%s - find preset temp: %s", self, preset_mode)
temp_val = self._presets.get(preset_name, 0)
# if not self._presence_on or self._presence_state in [
# None,
# STATE_ON,
# STATE_HOME,
# ]:
if self._presence_manager.is_absence_detected:
# We should return the preset_away temp val but if
# preset temp is 0, that means the user don't want to use
# the preset so we return 0, even if there is a value is preset_away
return self._presets_away.get(self.get_preset_away_name(preset_name), 0) if temp_val > 0 else temp_val
else:
return temp_val
def get_preset_away_name(self, preset_mode: str) -> str:
"""Get the preset name in away mode (when presence is off)"""
return preset_mode + PRESET_AWAY_SUFFIX
def get_state_date_or_now(self, state: State) -> datetime:
"""Extract the last_updated state from State or return now if not available"""
return state.last_updated.astimezone(self._current_tz) if isinstance(state.last_updated, datetime) else self.now
def get_last_updated_date_or_now(self, state: State) -> datetime:
"""Extract the last_updated state from State or return now if not available"""
return (
state.last_updated.astimezone(self._current_tz)
if isinstance(state.last_updated, datetime)
else self.now
)
async def async_underlying_entity_turn_off(self):
"""Turn heater toggleable device off. Used by Window, overpowering,
control_heating to turn all off"""
if self._cycle_scheduler:
await self._cycle_scheduler.cancel_cycle()
for under in self._underlyings:
await under.turn_off()
def set_hvac_off_reason(self, hvac_off_reason: str | None):
"""Set the reason of hvac_off"""
self._hvac_off_reason = hvac_off_reason
def set_hvac_mode_reason(self, hvac_mode_reason: str | None):
"""Set the reason why the current hvac_mode is forced"""
self._hvac_mode_reason = hvac_mode_reason
def set_temperature_reason(self, temperature_reason: str | None):
"""Set the reason of temperature"""
self._temperature_reason = temperature_reason
async def check_central_mode(self, new_central_mode: str | None, old_central_mode: str | None):
"""Take into account a central mode change"""
if not self.is_controlled_by_central_mode:
self._last_central_mode = None
return
_LOGGER.info(
"%s - Central mode have change from %s to %s",
self,
old_central_mode,
new_central_mode,
)
self._last_central_mode = new_central_mode
self.requested_state.force_changed()
await self.update_states()
return
def recalculate(self, force=False):
"""A utility function to force the calculation of a the algo and
update the custom attributes and write the state.
Should be overridden by super class
"""
raise NotImplementedError()
def incremente_energy(self):
"""increment the energy counter if device is active
Should be overridden by super class
"""
raise NotImplementedError()
def calculate_hvac_action(self, _: list = None) -> HVACAction | None:
"""Calculate the HVAC action based on the current state and underlying devices"""
if self.vtherm_hvac_mode == VThermHvacMode_OFF:
action = HVACAction.OFF
elif not self.is_device_active:
action = HVACAction.IDLE
elif self.vtherm_hvac_mode == VThermHvacMode_COOL:
action = HVACAction.COOLING
else:
action = HVACAction.HEATING
self._attr_hvac_action = action
def update_custom_attributes(self):
"""Update the custom extra attributes for the entity"""
messages: list[str] = []
if self.safety_manager.is_safety_detected:
messages.append(MSG_SAFETY_DETECTED)
if self.power_manager.is_overpowering_detected:
messages.append(MSG_OVERPOWERING_DETECTED)
if self.hvac_off_reason:
messages.append(self.hvac_off_reason)
if self.temperature_reason:
messages.append(self.temperature_reason)
not_initialized_entities = []
if not self.is_initialized:
messages.append(MSG_NOT_INITIALIZED)
# Find all underlying entities that are not initialized
for under in self._underlyings:
if not under.is_initialized:
not_initialized_entities.extend(under.state_manager.get_uninitialized_entities())
self._attr_extra_state_attributes: dict[str, Any] = {
"hvac_action": self.hvac_action,
"hvac_mode": self.hvac_mode,
"preset_mode": self.preset_mode,
"ema_temp": self._ema_temp,
"is_ready": self.is_ready,
"specific_states": {
"is_initialized": self.is_initialized,
"is_on": self.is_on,
"last_central_mode": self.last_central_mode,
"last_update_datetime": self.now.isoformat(),
"ext_current_temperature": self._cur_ext_temp,
"last_temperature_datetime": self._last_temperature_measure.astimezone(self._current_tz).isoformat(),
"last_ext_temperature_datetime": self._last_ext_temperature_measure.astimezone(self._current_tz).isoformat(),
"should_device_be_active": self.should_device_be_active,
"is_device_active": self.is_device_active,
"device_actives": self.device_actives,
"nb_device_actives": self.nb_device_actives,
"ema_temp": self._ema_temp,
"temperature_slope": round(self.last_temperature_slope or 0, 3),
"hvac_off_reason": self.hvac_off_reason,
"hvac_mode_reason": self.hvac_mode_reason,
ATTR_TOTAL_ENERGY: self.total_energy,
"last_change_time_from_vtherm": (
self._last_change_time_from_vtherm.astimezone(self._current_tz).isoformat() if self._last_change_time_from_vtherm is not None else None
),
"is_sleeping": self.is_sleeping,
"is_locked": self.lock_manager.is_locked,
"is_recalculate_scheduled": self.is_recalculate_scheduled,
"not_initialized_entities": not_initialized_entities,
"messages": messages,
},
"configuration": {
"ac_mode": self._ac_mode,
"type": self.vtherm_type,
"proportional_function": self.proportional_function,
"is_controlled_by_central_mode": self.is_controlled_by_central_mode,
"target_temperature_step": self.target_temperature_step,
"timezone": str(self._current_tz),
"temperature_unit": self.temperature_unit,
"is_used_by_central_boiler": self.is_used_by_central_boiler,
"max_on_percent": self._max_on_percent,
"have_valve_regulation": self.have_valve_regulation,
"cycle_min": self._cycle_min,
},
"preset_temperatures": {
"frost_temp": self._presets.get(VThermPreset.FROST, 0),
"eco_temp": self._presets.get(VThermPreset.ECO, 0),
"boost_temp": self._presets.get(VThermPreset.BOOST, 0),
"comfort_temp": self._presets.get(VThermPreset.COMFORT, 0),
"frost_away_temp": self._presets_away.get(self.get_preset_away_name(VThermPreset.FROST), 0),
"eco_away_temp": self._presets_away.get(self.get_preset_away_name(VThermPreset.ECO), 0),
"boost_away_temp": self._presets_away.get(self.get_preset_away_name(VThermPreset.BOOST), 0),
"comfort_away_temp": self._presets_away.get(self.get_preset_away_name(VThermPreset.COMFORT), 0),
},
}
self._state_manager.add_custom_attributes(self._attr_extra_state_attributes)
for manager in self._managers:
# add_custom_attributes is optional for external feature managers
# (it is not part of the InterfaceFeatureManager contract), so call
# it defensively.
publish_attributes = getattr(manager, "add_custom_attributes", None)
if callable(publish_attributes):
publish_attributes(self._attr_extra_state_attributes)
def send_event(self, event_type: EventType, data: dict):
"""Send an event"""
send_vtherm_event(self._hass, event_type=event_type, entity=self, data=data)
async def async_turn_off(self) -> None:
await self.async_set_hvac_mode(VThermHvacMode_OFF)
async def async_turn_on(self) -> None:
if self._ac_mode:
await self.async_set_hvac_mode(VThermHvacMode_COOL)
else:
await self.async_set_hvac_mode(VThermHvacMode_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
##
## Device changes events
##
@callback
async def _async_temperature_changed(self, event: Event) -> callable:
"""Handle temperature of the temperature sensor changes.
Return the function to dearm (clear) the window auto check"""
new_state: State = event.data.get("new_state")
write_event_log(_LOGGER, self, f"Temperature changed to state {new_state.state if new_state else None}")
if new_state is None or new_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
return
dearm_window_auto = await self._async_update_temp(new_state)
self.recalculate()
# Potentially it generates a safety event
if await self._safety_manager.refresh_and_update_if_changed() or (self.auto_start_stop_manager and await self.auto_start_stop_manager.refresh_and_update_if_changed()):
_LOGGER.info("%s - Change in safety alert or auto_start_stopis detected. Force update states", self)
return dearm_window_auto
await self.async_control_heating(force=False)
return dearm_window_auto
@callback
async def _async_last_seen_temperature_changed(self, event: Event):
"""Handle last seen temperature sensor changes."""
new_state: State = event.data.get("new_state")
write_event_log(_LOGGER, self, f"Last seen temperature changed to state {new_state.state if new_state else None}")
if new_state is None or new_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
return
# try to extract the datetime (from state)
try:
old_safety: bool = self._safety_manager.is_safety_detected
safety: bool = old_safety
# Convert ISO 8601 string to datetime object
self._last_temperature_measure = self.get_last_updated_date_or_now(new_state)
# issue 690 - don't reset the last change time on lastSeen
# self.reset_last_change_time_from_vtherm()
_LOGGER.debug(
"%s - new last_temperature_measure is now: %s",
self,
self._last_temperature_measure,
)
# try to restart if we were in safety mode
if self._safety_manager.is_safety_detected:
safety = await self._safety_manager.refresh_state()
if safety != old_safety:
_LOGGER.debug("%s - Change in safety alert is detected. Force update states", self)
self.requested_state.force_changed()
await self.update_states(force=True)
except ValueError as err:
# La conversion a échoué, la chaîne n'est pas au format ISO 8601
_LOGGER.warning(
"%s - impossible to convert last seen datetime %s. Error is: %s",
self,
new_state.state,
err,
)
async def _async_ext_temperature_changed(self, event: Event):
"""Handle external temperature of the sensor changes."""
new_state: State = event.data.get("new_state")
write_event_log(_LOGGER, self, f"Outdoor temperature changed to state {new_state.state if new_state else None}")
if new_state is None or new_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
return
await self._async_update_ext_temp(new_state)
self.recalculate()
# Potentially it generates a safety event
if await self._safety_manager.refresh_and_update_if_changed() or (self.auto_start_stop_manager and await self.auto_start_stop_manager.refresh_and_update_if_changed()):
_LOGGER.info("%s - Change in safety alert or auto_start_stopis detected. Force update states", self)
return
await self.async_control_heating(force=False)
@callback
async def _async_update_temp(self, state: State):
"""Update thermostat with latest state from sensor."""
try:
cur_temp = float(state.state)
if math.isnan(cur_temp) or math.isinf(cur_temp):
raise ValueError(f"Sensor has illegal state {state.state}")
self._cur_temp = cur_temp
self._last_temperature_measure = self.get_state_date_or_now(state)
# calculate the smooth_temperature with EMA calculation
self._ema_temp = self._ema_algo.calculate_ema(self._cur_temp, self._last_temperature_measure)
_LOGGER.debug(
"%s - After setting _last_temperature_measure %s, state.last_changed.replace=%s",
self,
self._last_temperature_measure,
state.last_changed.astimezone(self._current_tz),
)
# try to restart if we were in safety mode
# if self._safety_manager.is_safety_detected:
# await self._safety_manager.refresh_state()
# check window_auto
return await self._window_manager.manage_window_auto()
except ValueError as ex:
_LOGGER.error("Unable to update temperature from sensor: %s", ex)
@callback
async def _async_update_ext_temp(self, state: State):
"""Update thermostat with latest state from sensor."""
try:
cur_ext_temp = float(state.state)
if math.isnan(cur_ext_temp) or math.isinf(cur_ext_temp):
raise ValueError(f"Sensor has illegal state {state.state}")
self._cur_ext_temp = cur_ext_temp
self._last_ext_temperature_measure = self.get_state_date_or_now(state)
_LOGGER.debug(
"%s - After setting _last_ext_temperature_measure %s, state.last_changed.replace=%s",
self,
self._last_ext_temperature_measure,
state.last_changed.astimezone(self._current_tz),
)
# try to restart if we were in safety mode
# if self._safety_manager.is_safety_detected:
# await self._safety_manager.refresh_state()
except ValueError as ex:
_LOGGER.error("Unable to update external temperature from sensor: %s", ex)
##
## Services (actions)
##
async def service_set_presence(self, presence: str):
"""Called by a service call:
service: versatile_thermostat.set_presence
data:
presence: "off"
target:
entity_id: climate.thermostat_1
"""
if self.lock_manager.check_is_locked("service_set_presence"):
return
write_event_log(_LOGGER, self, f"Calling SERVICE_SET_PRESENCE, presence: {presence}")
await self._presence_manager.update_presence(presence)
await self.async_control_heating(force=True)
async def service_set_safety(
self,
delay_min: int,
min_on_percent: float,
default_on_percent: float,
):
"""Called by a service call:
service: versatile_thermostat.set_safety
data:
delay_min: 15
min_on_percent: 0.5
default_on_percent: 0.2
target:
entity_id: climate.thermostat_2
"""
if self.lock_manager.check_is_locked("service_set_safety"):
return
write_event_log(
_LOGGER, self, f"Calling SERVICE_SET_SAFETY, delay_min: {delay_min}, " f"min_on_percent: {min_on_percent * 100} %, default_on_percent: {default_on_percent * 100} %"
)
if delay_min:
self._safety_manager.set_safety_delay_min(delay_min)
if min_on_percent:
self._safety_manager.set_safety_min_on_percent(min_on_percent)
if default_on_percent:
self._safety_manager.set_safety_default_on_percent(default_on_percent)
await self.async_control_heating()
self.update_custom_attributes()
self.async_write_ha_state()
async def service_set_window_bypass_state(self, window_bypass: bool):
"""Called by a service call:
service: versatile_thermostat.set_window_bypass
data:
window_bypass: True
target:
entity_id: climate.thermostat_1
"""
if self.lock_manager.check_is_locked("service_set_window_bypass_state"):
return
write_event_log(_LOGGER, self, f"Calling SERVICE_SET_WINDOW_BYPASS, window_bypass: {window_bypass}")
if await self._window_manager.set_window_bypass(window_bypass):
self.requested_state.force_changed()
await self.update_states(force=True)
async def service_set_hvac_mode_sleep(self):
"""Set the hvac_mode to SLEEP mode (valid only for over_climate with valve regulation):
service: versatile_thermostat.set_hvac_mode_sleep
target:
entity_id: climate.thermostat_1
"""
if self.lock_manager.check_is_locked("service_set_hvac_mode_sleep"):
return
write_event_log(_LOGGER, self, "Calling SERVICE_SET_HVAC_MODE_SLEEP")
raise NotImplementedError("service_set_hva_mode_sleep not implemented for this kind of thermostat. Only for over_climate with valve regulation is supported")
async def service_lock(self, code: str | None = None):
"""Handle the lock service call."""
if not self.lock_manager.has_lock_settings_enabled:
_LOGGER.error("%s - Cannot lock thermostat: no lock settings enabled (neither users nor automations)", self)
raise HomeAssistantError("Cannot lock thermostat: no lock settings enabled. At least one of 'lock users' or 'lock automations' must be enabled.")
if self.lock_manager.change_lock_state(True, code):
self.update_custom_attributes()
self.async_write_ha_state()
async def service_unlock(self, code: str | None = None):
"""Handle the unlock service call."""
if self.lock_manager.change_lock_state(False, code):
self.update_custom_attributes()
self.async_write_ha_state()
async def service_set_tpi_parameters(
self,
tpi_coef_int: float | None = None,
tpi_coef_ext: float | None = None,
minimal_activation_delay: int | None = None,
minimal_deactivation_delay: int | None = None,
tpi_threshold_low: float | None = None,
tpi_threshold_high: float | None = None,
):
"""Stub method for TPI parameter service on non-TPI thermostats.
This service is only available for switch/valve type thermostats that use TPI algorithm.
For over_climate thermostats, this service is not supported.
Raises:
ServiceValidationError: Always raised to indicate the service is not available
"""
raise ServiceValidationError(
f"{self} - The set_tpi_parameters service is only available for switch/valve type thermostats. "
"This thermostat does not use TPI algorithm."
)
async def service_set_auto_tpi_mode(
self,
auto_tpi_mode: bool,
reinitialise: bool = True,
allow_kint_boost_on_stagnation: bool = False,
allow_kext_compensation_on_overshoot: bool = False,
):
"""Stub method for Auto TPI mode service on non-TPI thermostats.
This service is only available for switch/valve type thermostats that use TPI algorithm.
For over_climate thermostats, this service is not supported.
Raises:
ServiceValidationError: Always raised to indicate the service is not available
"""
raise ServiceValidationError(
f"{self} - The set_auto_tpi_mode service is only available for switch/valve type thermostats. "
"This thermostat does not use TPI algorithm."
)
async def service_auto_tpi_calibrate_capacity(
self,
save_to_config: bool,
min_power_threshold: int,
start_date: datetime | None = None,
end_date: datetime | None = None,
):
"""Stub method for Auto TPI capacity calibration service on non-TPI thermostats.
This service is only available for switch/valve type thermostats that use TPI algorithm.
For over_climate thermostats, this service is not supported.
Raises:
ServiceValidationError: Always raised to indicate the service is not available
"""
raise ServiceValidationError(
f"{self} - The auto_tpi_calibrate_capacity service is only available for switch/valve type thermostats. "
"This thermostat does not use TPI algorithm."
)
async def service_set_timed_preset(self, preset: str, duration_minutes: float):
"""Called by a service call:
service: versatile_thermostat.set_timed_preset
data:
preset: "boost"
duration_minutes: 30
target:
entity_id: climate.thermostat_1
Force a preset for a given duration. After the duration expires,
the original preset (from requested_state) will be restored.
"""
if self.lock_manager.check_is_locked("service_set_timed_preset"):
return
write_event_log(
_LOGGER,
self,
f"Calling SERVICE_SET_TIMED_PRESET, preset: {preset}, duration: {duration_minutes} min",
)
# Validate that the preset is in the list of accepted presets
preset_enum = VThermPreset(preset) if preset else None
if preset_enum not in self._vtherm_preset_modes:
raise ServiceValidationError(
f"{self} - The preset '{preset}' is not available for this thermostat. " f"Available presets are: {[str(p) for p in self._vtherm_preset_modes]}"
)
# Validate duration
if duration_minutes <= 0:
raise ServiceValidationError(f"{self} - The duration must be a positive number, got {duration_minutes}")
# Set the timed preset
success = await self._timed_preset_manager.set_timed_preset(preset_enum, duration_minutes)
if not success:
raise ServiceValidationError(f"{self} - Failed to set timed preset '{preset}' for {duration_minutes} minutes")
async def service_cancel_timed_preset(self):
"""Called by a service call:
service: versatile_thermostat.cancel_timed_preset
target:
entity_id: climate.thermostat_1
Cancel any active timed preset and restore the original preset.
"""
if self.lock_manager.check_is_locked("service_cancel_timed_preset"):
return
write_event_log(_LOGGER, self, "Calling SERVICE_CANCEL_TIMED_PRESET")
await self._timed_preset_manager.cancel_timed_preset()
async def service_recalibrate_valves(self, delay_seconds: int):
"""Stub method for recalibrate_valves service on unsupported thermostat types.
Raises:
ServiceValidationError: Always raised to indicate the service is not available
"""
raise ServiceValidationError(
f"{self} - The recalibrate_valves service is only available for ThermostatClimateValve thermostats."
)
async def service_download_logs(
self,
log_level: str = "DEBUG",
period_start: datetime | None = None,
period_end: datetime | None = None,
):
"""Called by the download_logs entity service."""
from vtherm_api.log_collector import async_export_logs # pylint: disable=import-outside-toplevel
handler = self._hass.data.get(DOMAIN, {}).get("log_handler")
if handler is None:
_LOGGER.warning("%s - Log collector is not initialized", self._name)
return
await async_export_logs(
hass=self._hass,
handler=handler,
thermostat_name=self._name,
entity_id=self.entity_id,
log_level=log_level,
period_start=period_start,
period_end=period_end,
config_entry=self._entry_infos,
)
##
## For testing purpose
##
# @deprecated
def _set_now(self, now: datetime):
"""Set the now timestamp. This is only for tests purpose
This method should be replaced by the vthermAPI equivalent"""
VersatileThermostatAPI.get_vtherm_api(self._hass)._set_now(now) # pylint: disable=protected-access
# @deprecated
@property
def now(self) -> datetime:
"""Get now. The local datetime or the overloaded _set_now date
This method should be replaced by the vthermAPI equivalent"""
return VersatileThermostatAPI.get_vtherm_api(self._hass).now
Reference in New Issue View Git Blame Copy Permalink