versatile_thermostat/custom_components/versatile_thermostat/feature_power_manager.py

""" Implements the Power Feature Manager """

# pylint: disable=line-too-long

import logging
from typing import Any

from homeassistant.const import (
    STATE_ON,
    STATE_OFF,
    STATE_UNAVAILABLE,
    STATE_UNKNOWN,
)

from homeassistant.core import (
    HomeAssistant,
)

from .const import *  # pylint: disable=wildcard-import, unused-wildcard-import
from .commons import ConfigData

from .base_manager import BaseFeatureManager
from .vtherm_api import VersatileThermostatAPI

_LOGGER = logging.getLogger(__name__)


class FeaturePowerManager(BaseFeatureManager):
    """The implementation of the Power feature"""

    unrecorded_attributes = frozenset(
        {
            "power_sensor_entity_id",
            "max_power_sensor_entity_id",
            "is_power_configured",
            "device_power",
            "power_temp",
            "current_power",
            "current_max_power",
        }
    )

    def __init__(self, vtherm: Any, hass: HomeAssistant):
        """Init of a featureManager"""
        super().__init__(vtherm, hass)
        self._power_temp = None
        self._overpowering_state = STATE_UNAVAILABLE
        self._is_configured: bool = False
        self._device_power: float = 0
        self._use_power_feature: bool = False

    @overrides
    def post_init(self, entry_infos: ConfigData):
        """Reinit of the manager"""

        # Power management
        self._power_temp = entry_infos.get(CONF_PRESET_POWER)

        self._device_power = entry_infos.get(CONF_DEVICE_POWER) or 0
        self._use_power_feature = entry_infos.get(CONF_USE_POWER_FEATURE, False)
        self._is_configured = False

    @overrides
    async def start_listening(self):
        """Start listening the underlying entity. There is nothing to listen"""
        central_power_configuration = (
            VersatileThermostatAPI.get_vtherm_api().central_power_manager.is_configured
        )

        if self._use_power_feature and self._device_power and central_power_configuration:
            self._is_configured = True
            # Try to restore _overpowering_state from previous state
            old_state = await self._vtherm.async_get_last_state()
            self._overpowering_state = (
                old_state.attributes.get("overpowering_state", STATE_UNKNOWN)
                if old_state and old_state.attributes and old_state.attributes in (STATE_OFF, STATE_ON)
                else STATE_UNKNOWN
            )
        else:
            if self._use_power_feature:
                if not central_power_configuration:
                    _LOGGER.warning(
                        "%s - Power management is not fully configured. You have to configure the central configuration power",
                        self,
                    )
                else:
                    _LOGGER.warning(
                        "%s - Power management is not fully configured. You have to configure the power feature of the VTherm",
                        self,
                    )

    def add_custom_attributes(self, extra_state_attributes: dict[str, Any]):
        """Add some custom attributes"""
        vtherm_api = VersatileThermostatAPI.get_vtherm_api()
        extra_state_attributes.update(
            {
                "power_sensor_entity_id": vtherm_api.central_power_manager.power_sensor_entity_id,
                "max_power_sensor_entity_id": vtherm_api.central_power_manager.max_power_sensor_entity_id,
                "overpowering_state": self._overpowering_state,
                "is_power_configured": self._is_configured,
                "device_power": self._device_power,
                "power_temp": self._power_temp,
                "current_power": vtherm_api.central_power_manager.current_power,
                "current_max_power": vtherm_api.central_power_manager.current_max_power,
                "mean_cycle_power": self.mean_cycle_power,
            }
        )

    async def check_power_available(self) -> bool:
        """Check if the Vtherm can be started considering overpowering.
        Returns True if no overpowering conditions are found
        """

        vtherm_api = VersatileThermostatAPI.get_vtherm_api()
        if (
            not self._is_configured
            or not vtherm_api.central_power_manager.is_configured
        ):
            return True

        current_power = vtherm_api.central_power_manager.current_power
        current_max_power = vtherm_api.central_power_manager.current_max_power
        if (
            current_power is None
            or current_max_power is None
            or self._device_power is None
        ):
            _LOGGER.warning(
                "%s - power not valued. check_power_available not available", self
            )
            return True

        _LOGGER.debug(
            "%s - overpowering check: power=%.3f, max_power=%.3f heater power=%.3f",
            self,
            current_power,
            current_max_power,
            self._device_power,
        )

        # issue 407 - power_consumption_max is power we need to add. If already active we don't need to add more power
        if self._vtherm.is_device_active:
            power_consumption_max = 0
        else:
            if self._vtherm.is_over_climate:
                power_consumption_max = self._device_power
            else:
                power_consumption_max = max(
                    self._device_power / self._vtherm.nb_underlying_entities,
                    self._device_power * self._vtherm.proportional_algorithm.on_percent,
                )

        ret = (current_power + power_consumption_max) < current_max_power
        if not ret:
            _LOGGER.info(
                "%s - there is not enough power available power=%.3f, max_power=%.3f heater power=%.3f",
                self,
                current_power,
                current_max_power,
                self._device_power,
            )
        return ret

    async def set_overpowering(self, overpowering: bool, power_consumption_max=0):
        """Force the overpowering state for the VTherm"""

        vtherm_api = VersatileThermostatAPI.get_vtherm_api()
        current_power = vtherm_api.central_power_manager.current_power
        current_max_power = vtherm_api.central_power_manager.current_max_power

        if overpowering and not self.is_overpowering_detected:
            _LOGGER.warning(
                "%s - overpowering is detected. Heater preset will be set to 'power'",
                self,
            )

            self._overpowering_state = STATE_ON

            if self._vtherm.is_over_climate:
                self._vtherm.save_hvac_mode()

            self._vtherm.save_preset_mode()
            await self._vtherm.async_underlying_entity_turn_off()
            await self._vtherm.async_set_preset_mode_internal(PRESET_POWER, force=True)
            self._vtherm.send_event(
                EventType.POWER_EVENT,
                {
                    "type": "start",
                    "current_power": current_power,
                    "device_power": self._device_power,
                    "current_max_power": current_max_power,
                    "current_power_consumption": power_consumption_max,
                },
            )
        elif not overpowering and self.is_overpowering_detected:
            _LOGGER.warning(
                "%s - end of overpowering is detected. Heater preset will be restored to '%s'",
                self,
                self._vtherm._saved_preset_mode,  # pylint: disable=protected-access
            )
            self._overpowering_state = STATE_OFF

            # restore state
            if self._vtherm.is_over_climate:
                await self._vtherm.restore_hvac_mode()
            await self._vtherm.restore_preset_mode()
            # restart cycle
            await self._vtherm.async_control_heating(force=True)
            self._vtherm.send_event(
                EventType.POWER_EVENT,
                {
                    "type": "end",
                    "current_power": current_power,
                    "device_power": self._device_power,
                    "current_max_power": current_max_power,
                },
            )
        elif not overpowering and self._overpowering_state != STATE_OFF:
            # just set to not overpowering the state which was not set
            self._overpowering_state = STATE_OFF
        else:
            # Nothing to do (already in the right state)
            return
        self._vtherm.update_custom_attributes()

    @overrides
    @property
    def is_configured(self) -> bool:
        """Return True of the presence is configured"""
        return self._is_configured

    @property
    def overpowering_state(self) -> str | None:
        """Return the current overpowering state STATE_ON or STATE_OFF
        or STATE_UNAVAILABLE if not configured"""
        if not self._is_configured:
            return STATE_UNAVAILABLE
        return self._overpowering_state

    @property
    def is_overpowering_detected(self) -> str | None:
        """Return True if the Vtherm is in overpowering state"""
        return self._overpowering_state == STATE_ON

    @property
    def power_temperature(self) -> bool:
        """Return the power temperature"""
        return self._power_temp

    @property
    def device_power(self) -> bool:
        """Return the device power"""
        return self._device_power

    @property
    def mean_cycle_power(self) -> float | None:
        """Returns the mean power consumption during the cycle"""
        if not self._device_power or not self._vtherm.proportional_algorithm:
            return None

        return float(
            self._device_power * self._vtherm.proportional_algorithm.on_percent
        )

    def __str__(self):
        return f"PowerManager-{self.name}"