# pylint: disable=line-too-long, abstract-method """TPI algorithm handler for ThermostatProp.""" import logging from vtherm_api.log_collector import get_vtherm_logger from typing import Any, TYPE_CHECKING from datetime import datetime from homeassistant.exceptions import ServiceValidationError from .prop_algo_tpi import TpiAlgorithm from .auto_tpi_manager import AutoTpiManager from .const import * # pylint: disable=wildcard-import, unused-wildcard-import from .vtherm_hvac_mode import VThermHvacMode_OFF, VThermHvacMode_HEAT, VThermHvacMode_COOL from .vtherm_central_api import VersatileThermostatAPI from .commons import write_event_log from .cycle_scheduler import calculate_cycle_times if TYPE_CHECKING: from .thermostat_prop import ThermostatProp _LOGGER = get_vtherm_logger(__name__) class TPIHandler: """Handler for TPI-specific logic. This class encapsulates all TPI-specific behavior and is used via composition by ThermostatProp. It updates thermostat attributes directly for backward compatibility with child classes. """ def __init__(self, thermostat: "ThermostatProp"): """Initialize handler with parent thermostat reference.""" self._thermostat = thermostat self._auto_tpi_manager: AutoTpiManager | None = None # Save default values for Auto TPI reset self._default_coef_int: float = 0 self._default_coef_ext: float = 0 @property def tpi_coef_int(self) -> float: """Return TPI internal coefficient from thermostat.""" return self._thermostat.tpi_coef_int @property def tpi_coef_ext(self) -> float: """Return TPI external coefficient from thermostat.""" return self._thermostat.tpi_coef_ext @property def tpi_threshold_low(self) -> float: """Return TPI low threshold from thermostat.""" return self._thermostat.tpi_threshold_low @property def tpi_threshold_high(self) -> float: """Return TPI high threshold from thermostat.""" return self._thermostat.tpi_threshold_high @property def minimal_activation_delay(self) -> int: """Return minimal activation delay from thermostat.""" return self._thermostat.minimal_activation_delay @property def minimal_deactivation_delay(self) -> int: """Return minimal deactivation delay from thermostat.""" return self._thermostat.minimal_deactivation_delay @property def proportional_function(self) -> str | None: """Return proportional function from thermostat.""" return self._thermostat.proportional_function @property def auto_tpi_manager(self) -> AutoTpiManager | None: """Return the Auto TPI manager.""" return self._auto_tpi_manager def init_algorithm(self): """Initialize PropAlgorithm and AutoTpiManager. Updates thermostat attributes directly for backward compatibility. """ t = self._thermostat entry = t.entry_infos # Read and set TPI-specific config on thermostat using public setters t.tpi_coef_int = entry.get(CONF_TPI_COEF_INT) t.tpi_coef_ext = entry.get(CONF_TPI_COEF_EXT) t.tpi_threshold_low = entry.get(CONF_TPI_THRESHOLD_LOW, 0.0) t.tpi_threshold_high = entry.get(CONF_TPI_THRESHOLD_HIGH, 0.0) t.minimal_activation_delay = entry.get(CONF_MINIMAL_ACTIVATION_DELAY, 0) t.minimal_deactivation_delay = entry.get(CONF_MINIMAL_DEACTIVATION_DELAY, 0) # Save default values for Auto TPI reset self._default_coef_int = t.tpi_coef_int self._default_coef_ext = t.tpi_coef_ext # Validation: thresholds - if one is 0 then both are 0 if t.tpi_threshold_low == 0.0 or t.tpi_threshold_high == 0.0: t.tpi_threshold_low = 0.0 t.tpi_threshold_high = 0.0 # Validation: delays if (t.minimal_activation_delay + t.minimal_deactivation_delay) / 60 > t.cycle_min: _LOGGER.warning( "%s - The sum of minimal_activation_delay (%s sec) and " "minimal_deactivation_delay (%s sec) is greater than cycle_min (%s). " "This can create some unexpected behavior. Please review your configuration", t, t.minimal_activation_delay, t.minimal_deactivation_delay, t.cycle_min, ) # Validation: external temp sensor for TPI if ( t.proportional_function == PROPORTIONAL_FUNCTION_TPI and t.ext_temp_sensor_entity_id is None ): _LOGGER.warning( "Using TPI function but not external temperature sensor is set. " "Removing the delta temp ext factor. " "Thermostat will not be fully operational." ) t.tpi_coef_ext = 0 # Create TpiAlgorithm on thermostat t.prop_algorithm = TpiAlgorithm( t.tpi_coef_int, t.tpi_coef_ext, t.name, max_on_percent=t.max_on_percent, tpi_threshold_low=t.tpi_threshold_low, tpi_threshold_high=t.tpi_threshold_high, ) # Initialize Auto TPI Manager from config heater_heating_time = entry.get(CONF_AUTO_TPI_HEATER_HEATING_TIME, 5) heater_cooling_time = entry.get(CONF_AUTO_TPI_HEATER_COOLING_TIME, 5) calculation_method = entry.get(CONF_AUTO_TPI_CALCULATION_METHOD, AUTO_TPI_METHOD_EMA) ema_alpha = entry.get(CONF_AUTO_TPI_EMA_ALPHA, 0.2) avg_initial_weight = entry.get(CONF_AUTO_TPI_AVG_INITIAL_WEIGHT, 1) heating_rate = entry.get(CONF_AUTO_TPI_HEATING_POWER, 1.0) cooling_rate = entry.get(CONF_AUTO_TPI_COOLING_POWER, 1.0) aggressiveness = entry.get(CONF_AUTO_TPI_AGGRESSIVENESS, 1.0) continuous_kext = entry.get(CONF_AUTO_TPI_CONTINUOUS_KEXT, False) continuous_kext_alpha = entry.get(CONF_AUTO_TPI_CONTINUOUS_KEXT_ALPHA, 0.04) _LOGGER.info("%s - DEBUG: TPI coefficients from entry_infos: int=%.3f, ext=%.3f", t, t.tpi_coef_int, t.tpi_coef_ext) self._auto_tpi_manager = AutoTpiManager( t.hass, t.config_entry, t.unique_id, t.name, t.cycle_min, t.tpi_threshold_low, t.tpi_threshold_high, t.minimal_deactivation_delay, coef_int=t.tpi_coef_int, coef_ext=t.tpi_coef_ext, heater_heating_time=heater_heating_time, heater_cooling_time=heater_cooling_time, calculation_method=calculation_method, ema_alpha=ema_alpha, avg_initial_weight=avg_initial_weight, heating_rate=heating_rate, cooling_rate=cooling_rate, aggressiveness=aggressiveness, continuous_kext=continuous_kext, continuous_kext_alpha=continuous_kext_alpha, ) self._auto_tpi_manager.set_is_vtherm_stopping_callback(lambda: t.is_removed) _LOGGER.info("%s - DEBUG: AutoTpiManager initialized with defaults: int=%.3f, ext=%.3f", t, self._auto_tpi_manager._default_coef_int, self._auto_tpi_manager._default_coef_ext) async def async_added_to_hass(self): """Load Auto TPI data.""" t = self._thermostat if self._auto_tpi_manager: # Set entity_id for pre-bootstrap calibration sensor lookup self._auto_tpi_manager._entity_id = t.entity_id _LOGGER.info("%s - DEBUG: Before load_data - int=%.3f, ext=%.3f", t, t.tpi_coef_int, t.tpi_coef_ext) await self._auto_tpi_manager.async_load_data() # If we have learned parameters, apply them learned_params = self._auto_tpi_manager.get_calculated_params() if learned_params: _LOGGER.info("%s - DEBUG: Learned params found: %s, learning_active=%s", t, learned_params, self._auto_tpi_manager.learning_active) if self._auto_tpi_manager.learning_active: t.tpi_coef_int = learned_params.get(CONF_TPI_COEF_INT, t.tpi_coef_int) t.tpi_coef_ext = learned_params.get(CONF_TPI_COEF_EXT, t.tpi_coef_ext) _LOGGER.info("%s - Restored Auto TPI parameters: %s", t, learned_params) else: _LOGGER.info("%s - Auto TPI parameters found but not applied because learning is disabled", t) _LOGGER.info("%s - DEBUG: After load_data - int=%.3f, ext=%.3f", t, t.tpi_coef_int, t.tpi_coef_ext) if self._auto_tpi_manager.learning_active: # Security: if the feature is disabled in config, we must stop learning if not t.entry_infos.get(CONF_AUTO_TPI_MODE, False): _LOGGER.info("%s - Auto TPI learning was active but feature is disabled in config. Stopping learning.", t) await self._auto_tpi_manager.stop_learning() else: _LOGGER.info("%s - Auto TPI learning is active (restored from storage)", t) async def async_startup(self): """Startup actions.""" # No more active cycle loop to start, it's now handled passively in control_heating pass def remove(self): """Cleanup on removal.""" t = self._thermostat if self._auto_tpi_manager: t.hass.async_create_task(self._auto_tpi_manager.async_save_data()) def on_scheduler_ready(self, scheduler) -> None: """Register AutoTPI learning callbacks on the cycle scheduler.""" if self._auto_tpi_manager: scheduler.register_cycle_start_callback(self._auto_tpi_manager.on_cycle_started) scheduler.register_cycle_end_callback(self._auto_tpi_manager.on_cycle_completed) def should_publish_intermediate(self) -> bool: """TPI always publishes the current control iteration.""" return True def _is_central_boiler_off(self) -> bool: """Check if the central boiler is configured but currently off.""" t = self._thermostat if not t.is_used_by_central_boiler: return False api = VersatileThermostatAPI.get_vtherm_api() if api and api.central_boiler_manager: return not api.central_boiler_manager.is_on return False async def _get_tpi_data(self) -> dict[str, Any]: """Calculate and return TPI cycle parameters.""" t = self._thermostat # Feed current temperatures to AutoTpiManager BEFORE getting params if self._auto_tpi_manager: await self._auto_tpi_manager.update( room_temp=t.current_temperature, ext_temp=t.current_outdoor_temperature, target_temp=t.target_temperature, hvac_mode=str(t.vtherm_hvac_mode), is_overpowering_detected=t.power_manager.is_overpowering_detected, is_central_boiler_off=self._is_central_boiler_off(), is_heating_failure=t.heating_failure_detection_manager.is_failure_detected, ) # Sync coefficients from AutoTpiManager before calculating if self._auto_tpi_manager and self._auto_tpi_manager.learning_active: new_params = await self._auto_tpi_manager.calculate() if new_params: new_coef_int = new_params.get(CONF_TPI_COEF_INT) new_coef_ext = new_params.get(CONF_TPI_COEF_EXT) if new_coef_int != t.prop_algorithm.tpi_coef_int or \ new_coef_ext != t.prop_algorithm.tpi_coef_ext: t.prop_algorithm.update_parameters(tpi_coef_int=new_coef_int, tpi_coef_ext=new_coef_ext) _LOGGER.debug("%s - Synced TPI coeffs before cycle: int=%.3f, ext=%.3f", t, new_coef_int, new_coef_ext) # Force recalculation with potentially updated coefficients t.recalculate() # Calculate time (in seconds) if t.prop_algorithm: on_time_sec, off_time_sec, forced_by_timing = calculate_cycle_times( t.on_percent, t.cycle_min, t.minimal_activation_delay, t.minimal_deactivation_delay, ) realized_percent = on_time_sec / (t.cycle_min * 60) # Notify prop_algorithm if forced by timing (existing behavior) if forced_by_timing: if t.prop_algorithm and hasattr(t.prop_algorithm, "update_realized_power"): t.prop_algorithm.update_realized_power(realized_percent) else: on_time_sec = 0 off_time_sec = 0 return { "on_time_sec": on_time_sec, "off_time_sec": off_time_sec, "on_percent": t.safe_on_percent, "hvac_mode": str(t.vtherm_hvac_mode), } async def control_heating(self, timestamp=None, force=False): """TPI-specific control heating logic.""" del timestamp t = self._thermostat # Feed the Auto TPI manager if self._auto_tpi_manager: # 1. Update manager's transient state await self._auto_tpi_manager.update( room_temp=t.current_temperature, ext_temp=t.current_outdoor_temperature, target_temp=t.target_temperature, hvac_mode=str(t.vtherm_hvac_mode), is_overpowering_detected=t.power_manager.is_overpowering_detected, is_central_boiler_off=self._is_central_boiler_off(), is_heating_failure=t.heating_failure_detection_manager.is_failure_detected, ) # 2. Synchronize parameters if learning is active new_params = await self._auto_tpi_manager.calculate() if self._auto_tpi_manager.learning_active and new_params: new_coef_int = new_params.get(CONF_TPI_COEF_INT) new_coef_ext = new_params.get(CONF_TPI_COEF_EXT) if new_coef_int is not None and new_coef_ext is not None: if t.prop_algorithm: # Update effective algo parameters t.prop_algorithm.update_parameters(tpi_coef_int=new_coef_int, tpi_coef_ext=new_coef_ext) # Keep thermostat attributes in sync t.tpi_coef_int = new_coef_int t.tpi_coef_ext = new_coef_ext _LOGGER.debug("%s - Synced PropAlgorithm with current Auto TPI coeffs: int=%.3f, ext=%.3f", t, new_coef_int, new_coef_ext) # Stop here if we are off if t.vtherm_hvac_mode == VThermHvacMode_OFF: _LOGGER.debug("%s - End of cycle (HVAC_MODE_OFF)", t) t._on_time_sec = 0 t._off_time_sec = int(t.cycle_min * 60) if t.is_device_active: await t.async_underlying_entity_turn_off() elif t.cycle_scheduler and t.cycle_scheduler.is_cycle_running: # The master scheduler may still hold a pending heat cycle while # the physical device is already in its OFF phase. await t.cycle_scheduler.cancel_cycle() else: on_percent = 0 if t.prop_algorithm: on_percent = t.on_percent if on_percent is None: # Temperature sensor was not yet available at the last # recalculate() call (e.g. HA restart before sensor comes # back online). Preserve the current switch state instead of # turning it off with on_percent=0 (bug 1884). _LOGGER.info( "%s - on_percent is None (temperature unavailable). " "Skipping cycle to preserve current switch state.", t, ) return on_time_sec, off_time_sec, forced_by_timing = calculate_cycle_times( on_percent, t.cycle_min, t.minimal_activation_delay, t.minimal_deactivation_delay, ) realized_percent = on_time_sec / (t.cycle_min * 60) if forced_by_timing: if t.prop_algorithm and hasattr(t.prop_algorithm, "update_realized_power"): t.prop_algorithm.update_realized_power(realized_percent) await t.cycle_scheduler.start_cycle( t.vtherm_hvac_mode, on_percent, force, ) async def on_state_changed(self, changed: bool): """Handle state changes.""" del changed # Cycle management is now passive, no need to start/stop loop pass def update_attributes(self): """Add TPI-specific attributes to thermostat.""" t = self._thermostat t._attr_extra_state_attributes["specific_states"].update({ "auto_tpi_state": "on" if self._auto_tpi_manager and self._auto_tpi_manager.learning_active else "off", "auto_tpi_continuous_kext": "on" if self._auto_tpi_manager and self._auto_tpi_manager._continuous_kext else "off", "auto_tpi_learning": ( self._auto_tpi_manager.get_filtered_state() if self._auto_tpi_manager and (self._auto_tpi_manager.learning_active or self._auto_tpi_manager._continuous_kext) else {} ), }) t._attr_extra_state_attributes["configuration"].update({ "minimal_activation_delay_sec": t.minimal_activation_delay, "minimal_deactivation_delay_sec": t.minimal_deactivation_delay, }) async def _async_update_tpi_config_entry(self): """Update the config entry with current TPI parameters.""" t = self._thermostat entry = t.hass.config_entries.async_get_entry(t.unique_id) if entry: new_data = entry.data.copy() new_data[CONF_TPI_COEF_INT] = t.tpi_coef_int new_data[CONF_TPI_COEF_EXT] = t.tpi_coef_ext new_data[CONF_TPI_THRESHOLD_LOW] = t.tpi_threshold_low new_data[CONF_TPI_THRESHOLD_HIGH] = t.tpi_threshold_high new_data[CONF_MINIMAL_ACTIVATION_DELAY] = t.minimal_activation_delay new_data[CONF_MINIMAL_DEACTIVATION_DELAY] = t.minimal_deactivation_delay result = t.hass.config_entries.async_update_entry(entry, data=new_data) _LOGGER.debug("%s - Config entry updated with new TPI params: %s", t, result) 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, ): """Service handler for set_tpi_parameters.""" t = self._thermostat if t.lock_manager.check_is_locked("service_set_tpi_parameters"): return write_event_log( _LOGGER, t, f"Calling SERVICE_SET_TPI_PARAMETERS, tpi_coef_int: {tpi_coef_int}, " f"tpi_coef_ext: {tpi_coef_ext}" f"minimal_activation_delay: {minimal_activation_delay}, " f"minimal_deactivation_delay: {minimal_deactivation_delay}, " f"tpi_threshold_low: {tpi_threshold_low}, " f"tpi_threshold_high: {tpi_threshold_high}", ) if t.prop_algorithm is None: raise ServiceValidationError(f"{t} - No TPI algorithm configured for this thermostat.") entry = t.hass.config_entries.async_get_entry(t.unique_id) if not entry: raise ServiceValidationError(f"{t} - No config entry has been found for this thermostat.") if entry.data.get(CONF_USE_TPI_CENTRAL_CONFIG, False): raise ServiceValidationError(f"{t} - Impossible to set TPI parameters when using central TPI configuration.") # Update the algorithm with coefficients and thresholds t.prop_algorithm.update_parameters( tpi_coef_int, tpi_coef_ext, tpi_threshold_low, tpi_threshold_high, ) # Update thermostat attributes directly (handler updates them via setters) t.tpi_coef_int = t.prop_algorithm.tpi_coef_int t.tpi_coef_ext = t.prop_algorithm.tpi_coef_ext t.tpi_threshold_low = t.prop_algorithm.tpi_threshold_low t.tpi_threshold_high = t.prop_algorithm.tpi_threshold_high # Update delays directly on thermostat (not in algo anymore) if minimal_activation_delay is not None: t.minimal_activation_delay = minimal_activation_delay t.cycle_scheduler.min_activation_delay = t.minimal_activation_delay if minimal_deactivation_delay is not None: t.minimal_deactivation_delay = minimal_deactivation_delay t.cycle_scheduler.min_deactivation_delay = t.minimal_deactivation_delay await self._async_update_tpi_config_entry() if t.is_removed: _LOGGER.debug("%s - Entity is removed, stop service_set_tpi_parameters", t) return t.recalculate() await t.async_control_heating(force=True) 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, ): """Service handler for set_auto_tpi_mode.""" t = self._thermostat if t.proportional_function != PROPORTIONAL_FUNCTION_TPI: raise ServiceValidationError(f"{t} - This service is only available for TPI algorithm.") if not t.entry_infos.get(CONF_AUTO_TPI_MODE, False): raise ServiceValidationError(f"{t} - Auto TPI is not enabled in configuration.") write_event_log( _LOGGER, t, f"Calling SERVICE_SET_AUTO_TPI_MODE, auto_tpi_mode: {auto_tpi_mode}, " f"reinitialise: {reinitialise}, " f"allow_kint_boost: {allow_kint_boost_on_stagnation}, " f"allow_kext_overshoot: {allow_kext_compensation_on_overshoot}", ) await self.async_set_auto_tpi_mode( auto_tpi_mode, reinitialise, allow_kint_boost_on_stagnation, allow_kext_compensation_on_overshoot, ) 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, ): """Service handler for auto_tpi_calibrate_capacity.""" t = self._thermostat if t.proportional_function != PROPORTIONAL_FUNCTION_TPI: raise ServiceValidationError(f"{t} - This service is only available for TPI algorithm.") if not t.entry_infos.get(CONF_AUTO_TPI_MODE, False): raise ServiceValidationError(f"{t} - Auto TPI is not enabled in configuration.") write_event_log(_LOGGER, t, f"Calling SERVICE_AUTO_TPI_CALIBRATE_CAPACITY, save_to_config: {save_to_config}, start_date: {start_date}, end_date: {end_date}, min_power_threshold: {min_power_threshold}") if not self._auto_tpi_manager: raise ServiceValidationError(f"{t} - Auto TPI Manager not initialized, cannot calibrate capacity.") result = await self._auto_tpi_manager.service_calibrate_capacity( thermostat_entity_id=t.entity_id, ext_temp_entity_id=t.ext_temp_sensor_entity_id, save_to_config=save_to_config, start_date=start_date, end_date=end_date, min_power_threshold=min_power_threshold / 100.0, ) if result and result.get("success") and result.get("max_capacity"): t.recalculate() t.update_custom_attributes() t.async_write_ha_state() return result async def async_set_auto_tpi_mode( self, auto_tpi_mode: bool, reinitialise: bool = True, allow_kint_boost: bool = False, allow_kext_overshoot: bool = False, ): """Set the auto TPI mode.""" t = self._thermostat _LOGGER.debug( "%s - async_set_auto_tpi_mode called with auto_tpi_mode=%s, reinitialise=%s, kint_boost=%s, kext_overshoot=%s", t, auto_tpi_mode, reinitialise, allow_kint_boost, allow_kext_overshoot, ) if not self._auto_tpi_manager: _LOGGER.warning("%s - Auto TPI Manager not initialized", t) return # Safety check: Prevent enabling learning if the feature is disabled in config if auto_tpi_mode and not t.entry_infos.get(CONF_AUTO_TPI_MODE, False): _LOGGER.warning("%s - Cannot start Auto TPI Learning: feature is disabled in configuration", t) await self._auto_tpi_manager.stop_learning() return if auto_tpi_mode: # Use the original configured default values await self._auto_tpi_manager.start_learning( coef_int=self._auto_tpi_manager._default_coef_int, coef_ext=self._auto_tpi_manager._default_coef_ext, reset_data=reinitialise, allow_kint_boost=allow_kint_boost, allow_kext_overshoot=allow_kext_overshoot, ) # Sync PropAlgorithm with the configured coefficients if t.prop_algorithm: t.prop_algorithm.update_parameters(tpi_coef_int=t.tpi_coef_int, tpi_coef_ext=t.tpi_coef_ext) _LOGGER.info("%s - PropAlgorithm synced with config: Kint=%.3f, Kext=%.3f", t, t.tpi_coef_int, t.tpi_coef_ext) # If we enable auto_tpi, we must disable central config for TPI # Note: _entry_infos is a dict, we can update it directly if t._entry_infos: t._entry_infos[CONF_USE_TPI_CENTRAL_CONFIG] = False # Persist the change to the config entry entry = t.hass.config_entries.async_get_entry(t.unique_id) if entry and entry.data.get(CONF_USE_TPI_CENTRAL_CONFIG, True): new_data = entry.data.copy() new_data[CONF_USE_TPI_CENTRAL_CONFIG] = False t.hass.config_entries.async_update_entry(entry, data=new_data) if t.is_removed: _LOGGER.debug("%s - Entity is removed, stop async_set_auto_tpi_mode", t) return # Starting learning is sufficient, cycle processing is passive pass else: await self._auto_tpi_manager.stop_learning() # Apply configured coefficients to PropAlgorithm if t.prop_algorithm: t.prop_algorithm.update_parameters(tpi_coef_int=t.tpi_coef_int, tpi_coef_ext=t.tpi_coef_ext) _LOGGER.info( "%s - PropAlgorithm reset to config values: Kint=%.3f, Kext=%.3f", t, t.tpi_coef_int, t.tpi_coef_ext ) # Fire event to notify listeners t.hass.bus.async_fire( AUTO_TPI_EVENT, { "entity_id": t.entity_id, "auto_tpi_mode": self._auto_tpi_manager.learning_active, }, ) # Force update of state attributes t.update_custom_attributes() t.async_write_ha_state()