gilles/versatile_thermostat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor Presence Feature

Browse Source
This commit is contained in:
Jean-Marc Collin
2024-12-23 12:04:22 +00:00
parent 6e5e304b71
commit b41d0f34dc
13 changed files with 618 additions and 263 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
custom_components/versatile_thermostat/base_entity.py Normal file
View File

@@ -0,0 +1,118 @@
""" A base class for all VTherm entities"""
import logging
from datetime import timedelta
from homeassistant.core import HomeAssistant, callback, Event
from homeassistant.components.climate import ClimateEntity, DOMAIN as CLIMATE_DOMAIN
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.device_registry import DeviceInfo, DeviceEntryType
from homeassistant.helpers.event import async_track_state_change_event, async_call_later
from .const import DOMAIN, DEVICE_MANUFACTURER
from .base_thermostat import BaseThermostat
_LOGGER = logging.getLogger(__name__)
class VersatileThermostatBaseEntity(Entity):
"""A base class for all entities"""
_my_climate: BaseThermostat
hass: HomeAssistant
_config_id: str
_device_name: str
def __init__(self, hass: HomeAssistant, config_id, device_name) -> None:
"""The CTOR"""
self.hass = hass
self._config_id = config_id
self._device_name = device_name
self._my_climate = None
self._cancel_call = None
self._attr_has_entity_name = True
@property
def should_poll(self) -> bool:
"""Do not poll for those entities"""
return False
@property
def my_climate(self) -> BaseThermostat | None:
"""Returns my climate if found"""
if not self._my_climate:
self._my_climate = self.find_my_versatile_thermostat()
if self._my_climate:
# Only the first time
self.my_climate_is_initialized()
return self._my_climate
@property
def device_info(self) -> DeviceInfo:
"""Return the device info."""
return DeviceInfo(
entry_type=DeviceEntryType.SERVICE,
identifiers={(DOMAIN, self._config_id)},
name=self._device_name,
manufacturer=DEVICE_MANUFACTURER,
model=DOMAIN,
)
def find_my_versatile_thermostat(self) -> BaseThermostat:
"""Find the underlying climate entity"""
try:
component: EntityComponent[ClimateEntity] = self.hass.data[CLIMATE_DOMAIN]
for entity in component.entities:
# _LOGGER.debug("Device_info is %s", entity.device_info)
if entity.device_info == self.device_info:
_LOGGER.debug("Found %s!", entity)
return entity
except KeyError:
pass
return None
@callback
async def async_added_to_hass(self):
"""Listen to my climate state change"""
# Check delay condition
async def try_find_climate(_):
_LOGGER.debug(
"%s - Calling VersatileThermostatBaseEntity.async_added_to_hass", self
)
mcl = self.my_climate
if mcl:
if self._cancel_call:
self._cancel_call()
self._cancel_call = None
self.async_on_remove(
async_track_state_change_event(
self.hass,
[mcl.entity_id],
self.async_my_climate_changed,
)
)
else:
_LOGGER.debug("%s - no entity to listen. Try later", self)
self._cancel_call = async_call_later(
self.hass, timedelta(seconds=1), try_find_climate
)
await try_find_climate(None)
@callback
def my_climate_is_initialized(self):
"""Called when the associated climate is initialized"""
return
@callback
async def async_my_climate_changed(
self, event: Event
): # pylint: disable=unused-argument
"""Called when my climate have change
This method aims to be overriden to take the status change
"""
return

58
custom_components/versatile_thermostat/base_manager.py Normal file
View File

@@ -0,0 +1,58 @@
""" Implements a base Feature Manager for Versatile Thermostat """
import logging
from typing import Any
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
from .const import * # pylint: disable=wildcard-import, unused-wildcard-import
from .commons import ConfigData
from .config_schema import * # pylint: disable=wildcard-import, unused-wildcard-import
_LOGGER = logging.getLogger(__name__)
class BaseFeatureManager:
"""A base class for all feature"""
def __init__(self, vtherm: Any, hass: HomeAssistant):
"""Init of a featureManager"""
self._vtherm = vtherm
self._name = vtherm.name
self._active_listener: list[CALLBACK_TYPE] = []
self._hass = hass
def post_init(self, entry_infos: ConfigData):
"""Initialize the attributes of the FeatureManager"""
raise NotImplementedError()
def start_listening(self):
"""Start listening the underlying entity"""
raise NotImplementedError()
def stop_listening(self) -> bool:
"""stop listening to the sensor"""
while self._active_listener:
self._active_listener.pop()()
self._active_listener = []
def add_listener(self, func: CALLBACK_TYPE) -> None:
"""Add a listener to the list of active listener"""
self._active_listener.append(func)
@property
def is_configured(self) -> bool:
"""True if the FeatureManager is fully configured"""
raise NotImplementedError()
@property
def name(self) -> str:
"""The name"""
return self._name
@property
def hass(self) -> HomeAssistant:
"""The HA instance"""
return self._hass

183
custom_components/versatile_thermostat/base_thermostat.py
View File

@@ -4,11 +4,9 @@
""" Implements the VersatileThermostat climate component """
import math
import logging
from typing import Any, Generic
from datetime import timedelta, datetime
from types import MappingProxyType
from typing import Any, TypeVar, Generic
from homeassistant.core import (
HomeAssistant,
callback,
@@ -60,11 +58,10 @@ from homeassistant.const import (
STATE_UNKNOWN,
STATE_OFF,
STATE_ON,
STATE_HOME,
STATE_NOT_HOME,
)
from .const import * # pylint: disable=wildcard-import, unused-wildcard-import
from .commons import ConfigData, T
from .config_schema import * # pylint: disable=wildcard-import, unused-wildcard-import
@@ -75,9 +72,9 @@ from .prop_algorithm import PropAlgorithm
from .open_window_algorithm import WindowOpenDetectionAlgorithm
from .ema import ExponentialMovingAverage
from .presence_manager import FeaturePresenceManager
_LOGGER = logging.getLogger(__name__)
ConfigData = MappingProxyType[str, Any]
T = TypeVar("T", bound=UnderlyingEntity)
class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
"""Representation of a base class for all Versatile Thermostat device."""
@@ -187,7 +184,6 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
self._last_ext_temperature_measure = None
self._last_temperature_measure = None
self._cur_ext_temp = None
self._presence_state = None
self._overpowering_state = None
self._should_relaunch_control_heating = None
@@ -247,6 +243,10 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
self._hvac_off_reason: HVAC_OFF_REASONS | None = None
self._presence_manager: FeaturePresenceManager = FeaturePresenceManager(
self, hass
)
self.post_init(entry_infos)
def clean_central_config_doublon(
@@ -311,6 +311,8 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
self._entry_infos = entry_infos
self._presence_manager.post_init(entry_infos)
self._use_central_config_temperature = entry_infos.get(
CONF_USE_PRESETS_CENTRAL_CONFIG
) or (
@@ -386,14 +388,8 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
self._tpi_coef_int = entry_infos.get(CONF_TPI_COEF_INT)
self._tpi_coef_ext = entry_infos.get(CONF_TPI_COEF_EXT)
self._presence_sensor_entity_id = entry_infos.get(CONF_PRESENCE_SENSOR)
self._power_temp = entry_infos.get(CONF_PRESET_POWER)
self._presence_on = (
entry_infos.get(CONF_USE_PRESENCE_FEATURE, False)
and self._presence_sensor_entity_id is not None
)
if self._ac_mode:
# Added by https://github.com/jmcollin78/versatile_thermostat/pull/144
# Some over_switch can do both heating and cooling
@@ -473,7 +469,6 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
self._motion_state = None
self._window_state = None
self._overpowering_state = None
self._presence_state = None
self._total_energy = None
_LOGGER.debug("%s - post_init_ resetting energy to None", self)
@@ -580,14 +575,7 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
)
)
if self._presence_on:
self.async_on_remove(
async_track_state_change_event(
self.hass,
[self._presence_sensor_entity_id],
self._async_presence_changed,
)
)
self._presence_manager.start_listening()
self.async_on_remove(self.remove_thermostat)
@@ -606,6 +594,8 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
"""Called when the thermostat will be removed"""
_LOGGER.info("%s - Removing thermostat", self)
self._presence_manager.stop_listening()
for under in self._underlyings:
under.remove_entity()
@@ -725,20 +715,8 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
await self._async_update_motion_temp()
need_write_state = True
if self._presence_on:
# try to acquire presence entity state
presence_state = self.hass.states.get(self._presence_sensor_entity_id)
if presence_state and presence_state.state not in (
STATE_UNAVAILABLE,
STATE_UNKNOWN,
):
await self._async_update_presence(presence_state.state)
_LOGGER.debug(
"%s - Presence have been retrieved: %s",
self,
presence_state.state,
)
need_write_state = True
if await self._presence_manager.refresh_state():
need_write_state = True
if need_write_state:
self.async_write_ha_state()
@@ -776,16 +754,16 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
# If we have a previously saved temperature
if old_state.attributes.get(ATTR_TEMPERATURE) is None:
if self._ac_mode:
await self._async_internal_set_temperature(self.max_temp)
await self.change_target_temperature(self.max_temp)
else:
await self._async_internal_set_temperature(self.min_temp)
await self.change_target_temperature(self.min_temp)
_LOGGER.warning(
"%s - Undefined target temperature, falling back to %s",
self,
self._target_temp,
)
else:
await self._async_internal_set_temperature(
await self.change_target_temperature(
float(old_state.attributes[ATTR_TEMPERATURE])
)
@@ -827,9 +805,9 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
# No previous state, try and restore defaults
if self._target_temp is None:
if self._ac_mode:
await self._async_internal_set_temperature(self.max_temp)
await self.change_target_temperature(self.max_temp)
else:
await self._async_internal_set_temperature(self.min_temp)
await self.change_target_temperature(self.min_temp)
_LOGGER.warning(
"No previously saved temperature, setting to %s", self._target_temp
)
@@ -1075,14 +1053,14 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
return self._security_state
@property
def motion_state(self) -> bool | None:
def motion_state(self) -> str | None:
"""Get the motion_state"""
return self._motion_state
@property
def presence_state(self) -> bool | None:
def presence_state(self) -> str | None:
"""Get the presence_state"""
return self._presence_state
return self._presence_manager.presence_state
@property
def proportional_algorithm(self) -> PropAlgorithm | None:
@@ -1330,7 +1308,7 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
if preset_mode == PRESET_NONE:
self._attr_preset_mode = PRESET_NONE
if self._saved_target_temp:
await self._async_internal_set_temperature(self._saved_target_temp)
await self.change_target_temperature(self._saved_target_temp)
elif preset_mode == PRESET_ACTIVITY:
self._attr_preset_mode = PRESET_ACTIVITY
await self._async_update_motion_temp()
@@ -1340,9 +1318,7 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
self._attr_preset_mode = preset_mode
# Switch the temperature if window is not 'on'
if self.window_state != STATE_ON:
await self._async_internal_set_temperature(
self.find_preset_temp(preset_mode)
)
await self.change_target_temperature(self.find_preset_temp(preset_mode))
else:
# Window is on, so we just save the new expected temp
# so that closing the window will restore it
@@ -1409,7 +1385,7 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
)
if motion_preset in self._presets:
if self._presence_on and self.presence_state in [STATE_OFF, None]:
if self._presence_manager.is_absence_detected:
return self._presets_away[motion_preset + PRESET_AWAY_SUFFIX]
else:
return self._presets[motion_preset]
@@ -1423,13 +1399,12 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
_LOGGER.info("%s - find preset temp: %s", self, preset_mode)
temp_val = self._presets.get(preset_mode, 0)
if not self._presence_on or self._presence_state in [
None,
STATE_ON,
STATE_HOME,
]:
return temp_val
else:
# 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
@@ -1438,6 +1413,8 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
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)"""
@@ -1467,14 +1444,14 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
self._attr_preset_mode = PRESET_NONE
if self.window_state != STATE_ON:
await self._async_internal_set_temperature(temperature)
await self.change_target_temperature(temperature)
self.recalculate()
self.reset_last_change_time_from_vtherm()
await self.async_control_heating(force=True)
else:
self._saved_target_temp = temperature
async def _async_internal_set_temperature(self, temperature: float):
async def change_target_temperature(self, temperature: float):
"""Set the target temperature and the target temperature of underlying climate if any"""
if temperature:
self._target_temp = temperature
@@ -1710,7 +1687,7 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
# We do not change the preset which is kept to ACTIVITY but only the target_temperature
# We take the presence into account
await self._async_internal_set_temperature(
await self.change_target_temperature(
self.find_preset_temp(new_preset)
)
self.recalculate()
@@ -1900,59 +1877,6 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
except ValueError as ex:
_LOGGER.error("Unable to update current_power from sensor: %s", ex)
@callback
async def _async_presence_changed(self, event: Event[EventStateChangedData]):
"""Handle presence changes."""
new_state = event.data.get("new_state")
_LOGGER.info(
"%s - Presence changed. Event.new_state is %s, _attr_preset_mode=%s, activity=%s",
self,
new_state,
self._attr_preset_mode,
PRESET_ACTIVITY,
)
if new_state is None:
return
await self._async_update_presence(new_state.state)
await self.async_control_heating(force=True)
async def _async_update_presence(self, new_state: str):
_LOGGER.info("%s - Updating presence. New state is %s", self, new_state)
self._presence_state = (
STATE_ON if new_state in (STATE_ON, STATE_HOME) else STATE_OFF
)
if self._attr_preset_mode in HIDDEN_PRESETS or self._presence_on is False:
_LOGGER.info(
"%s - Ignoring presence change cause in Power or Security preset or presence not configured",
self,
)
return
if new_state is None or new_state not in (
STATE_OFF,
STATE_ON,
STATE_HOME,
STATE_NOT_HOME,
):
return
if self._attr_preset_mode not in [
PRESET_BOOST,
PRESET_COMFORT,
PRESET_ECO,
PRESET_ACTIVITY,
]:
return
new_temp = self.find_preset_temp(self.preset_mode)
if new_temp is not None:
_LOGGER.debug(
"%s - presence change in temperature mode new_temp will be: %.2f",
self,
new_temp,
)
await self._async_internal_set_temperature(new_temp)
self.recalculate()
async def _async_update_motion_temp(self):
"""Update the temperature considering the ACTIVITY preset and current motion state"""
_LOGGER.debug(
@@ -1980,9 +1904,7 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
# We do not change the preset which is kept to ACTIVITY but only the target_temperature
# We take the presence into account
await self._async_internal_set_temperature(
self.find_preset_temp(new_preset)
)
await self.change_target_temperature(self.find_preset_temp(new_preset))
_LOGGER.debug(
"%s - regarding motion, target_temp have been set to %.2f",
@@ -2496,7 +2418,7 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
self._saved_target_temp,
)
if self._window_action in [CONF_WINDOW_FROST_TEMP, CONF_WINDOW_ECO_TEMP]:
await self._async_internal_set_temperature(self._saved_target_temp)
await self.change_target_temperature(self._saved_target_temp)
# default to TURN_OFF
elif self._window_action in [CONF_WINDOW_TURN_OFF]:
@@ -2544,16 +2466,14 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
self._window_action == CONF_WINDOW_FROST_TEMP
and self._presets.get(PRESET_FROST_PROTECTION) is not None
):
await self._async_internal_set_temperature(
await self.change_target_temperature(
self.find_preset_temp(PRESET_FROST_PROTECTION)
)
elif (
self._window_action == CONF_WINDOW_ECO_TEMP
and self._presets.get(PRESET_ECO) is not None
):
await self._async_internal_set_temperature(
self.find_preset_temp(PRESET_ECO)
)
await self.change_target_temperature(self.find_preset_temp(PRESET_ECO))
else: # default is to turn_off
self.set_hvac_off_reason(HVAC_OFF_REASON_WINDOW_DETECTION)
await self.async_set_hvac_mode(HVACMode.OFF)
@@ -2664,8 +2584,6 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
"power_sensor_entity_id": self._power_sensor_entity_id,
"max_power_sensor_entity_id": self._max_power_sensor_entity_id,
"overpowering_state": self.overpowering_state,
"presence_sensor_entity_id": self._presence_sensor_entity_id,
"presence_state": self._presence_state,
"window_state": self.window_state,
"window_auto_state": self.window_auto_state,
"window_bypass_state": self._window_bypass_state,
@@ -2711,20 +2629,11 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
),
}
_LOGGER.debug(
"%s - update_custom_attributes saved energy is %s",
self,
self.total_energy,
)
self._presence_manager.add_custom_attributes(self._attr_extra_state_attributes)
@overrides
def async_write_ha_state(self):
"""overrides to have log"""
_LOGGER.debug(
"%s - async_write_ha_state written state energy is %s",
self,
self._total_energy,
)
return super().async_write_ha_state()
@property
@@ -2748,7 +2657,7 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
entity_id: climate.thermostat_1
"""
_LOGGER.info("%s - Calling service_set_presence, presence: %s", self, presence)
await self._async_update_presence(presence)
await self._presence_manager.update_presence(presence)
await self.async_control_heating(force=True)
async def service_set_preset_temperature(
@@ -2776,7 +2685,7 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
if preset in self._presets:
if temperature is not None:
self._presets[preset] = temperature
if self._presence_on and temperature_away is not None:
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(
@@ -2899,7 +2808,9 @@ class BaseThermostat(ClimateEntity, RestoreEntity, Generic[T]):
CONF_USE_PRESETS_CENTRAL_CONFIG,
)
if self._entry_infos.get(CONF_USE_PRESENCE_FEATURE) is True:
# 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()

6
custom_components/versatile_thermostat/binary_sensor.py
View File

@@ -25,10 +25,8 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .vtherm_api import VersatileThermostatAPI
from .commons import (
VersatileThermostatBaseEntity,
check_and_extract_service_configuration,
)
from .commons import check_and_extract_service_configuration
from .base_entity import VersatileThermostatBaseEntity
from .const import (
DOMAIN,
DEVICE_MANUFACTURER,

116
custom_components/versatile_thermostat/commons.py
View File

@@ -3,17 +3,14 @@
# pylint: disable=line-too-long
import logging
from datetime import timedelta
from homeassistant.core import HomeAssistant, callback, Event
from homeassistant.components.climate import ClimateEntity, DOMAIN as CLIMATE_DOMAIN
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.device_registry import DeviceInfo, DeviceEntryType
from homeassistant.helpers.event import async_track_state_change_event, async_call_later
from types import MappingProxyType
from typing import Any, TypeVar
from .const import ServiceConfigurationError
from .underlyings import UnderlyingEntity
from .base_thermostat import BaseThermostat
from .const import DOMAIN, DEVICE_MANUFACTURER, ServiceConfigurationError
ConfigData = MappingProxyType[str, Any]
T = TypeVar("T", bound=UnderlyingEntity)
_LOGGER = logging.getLogger(__name__)
@@ -135,104 +132,3 @@ def check_and_extract_service_configuration(service_config) -> dict:
"check_and_extract_service_configuration(%s) gives '%s'", service_config, ret
)
return ret
class VersatileThermostatBaseEntity(Entity):
"""A base class for all entities"""
_my_climate: BaseThermostat
hass: HomeAssistant
_config_id: str
_device_name: str
def __init__(self, hass: HomeAssistant, config_id, device_name) -> None:
"""The CTOR"""
self.hass = hass
self._config_id = config_id
self._device_name = device_name
self._my_climate = None
self._cancel_call = None
self._attr_has_entity_name = True
@property
def should_poll(self) -> bool:
"""Do not poll for those entities"""
return False
@property
def my_climate(self) -> BaseThermostat | None:
"""Returns my climate if found"""
if not self._my_climate:
self._my_climate = self.find_my_versatile_thermostat()
if self._my_climate:
# Only the first time
self.my_climate_is_initialized()
return self._my_climate
@property
def device_info(self) -> DeviceInfo:
"""Return the device info."""
return DeviceInfo(
entry_type=DeviceEntryType.SERVICE,
identifiers={(DOMAIN, self._config_id)},
name=self._device_name,
manufacturer=DEVICE_MANUFACTURER,
model=DOMAIN,
)
def find_my_versatile_thermostat(self) -> BaseThermostat:
"""Find the underlying climate entity"""
try:
component: EntityComponent[ClimateEntity] = self.hass.data[CLIMATE_DOMAIN]
for entity in component.entities:
# _LOGGER.debug("Device_info is %s", entity.device_info)
if entity.device_info == self.device_info:
_LOGGER.debug("Found %s!", entity)
return entity
except KeyError:
pass
return None
@callback
async def async_added_to_hass(self):
"""Listen to my climate state change"""
# Check delay condition
async def try_find_climate(_):
_LOGGER.debug(
"%s - Calling VersatileThermostatBaseEntity.async_added_to_hass", self
)
mcl = self.my_climate
if mcl:
if self._cancel_call:
self._cancel_call()
self._cancel_call = None
self.async_on_remove(
async_track_state_change_event(
self.hass,
[mcl.entity_id],
self.async_my_climate_changed,
)
)
else:
_LOGGER.debug("%s - no entity to listen. Try later", self)
self._cancel_call = async_call_later(
self.hass, timedelta(seconds=1), try_find_climate
)
await try_find_climate(None)
@callback
def my_climate_is_initialized(self):
"""Called when the associated climate is initialized"""
return
@callback
async def async_my_climate_changed(
self, event: Event
): # pylint: disable=unused-argument
"""Called when my climate have change
This method aims to be overriden to take the status change
"""
return

1
custom_components/versatile_thermostat/const.py
View File

@@ -4,6 +4,7 @@
import logging
import math
from typing import Literal
from datetime import datetime
from enum import Enum

2
custom_components/versatile_thermostat/number.py
View File

@@ -28,7 +28,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util import slugify
from .vtherm_api import VersatileThermostatAPI
from .commons import VersatileThermostatBaseEntity
from .base_entity import VersatileThermostatBaseEntity
from .const import (
DOMAIN,

189
custom_components/versatile_thermostat/presence_manager.py Normal file
View File

@@ -0,0 +1,189 @@
""" Implements the Presence Feature Manager """
# pylint: disable=line-too-long
import logging
from typing import Any
from homeassistant.const import (
STATE_ON,
STATE_OFF,
STATE_HOME,
STATE_NOT_HOME,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import (
HomeAssistant,
callback,
Event,
)
from homeassistant.helpers.event import (
async_track_state_change_event,
EventStateChangedData,
)
from homeassistant.components.climate import (
PRESET_ACTIVITY,
PRESET_BOOST,
PRESET_COMFORT,
PRESET_ECO,
)
from .const import * # pylint: disable=wildcard-import, unused-wildcard-import
from .commons import ConfigData
from .base_manager import BaseFeatureManager
_LOGGER = logging.getLogger(__name__)
class FeaturePresenceManager(BaseFeatureManager):
"""A base class for all feature"""
def __init__(self, vtherm: Any, hass: HomeAssistant):
"""Init of a featureManager"""
super().__init__(vtherm, hass)
self._presence_state: str = STATE_UNAVAILABLE
self._presence_sensor_entity_id: str = None
self._is_configured: bool = False
@overrides
def post_init(self, entry_infos: ConfigData):
"""Reinit of the manager"""
self._presence_sensor_entity_id = entry_infos.get(CONF_PRESENCE_SENSOR)
self._is_configured = (
entry_infos.get(CONF_USE_PRESENCE_FEATURE, False)
and self._presence_sensor_entity_id is not None
)
self._presence_state = STATE_UNKNOWN
@overrides
def start_listening(self):
"""Start listening the underlying entity"""
if self._is_configured:
self.stop_listening()
self.add_listener(
async_track_state_change_event(
self.hass,
[self._presence_sensor_entity_id],
self._presence_sensor_changed,
)
)
@overrides
async def refresh_state(self) -> bool:
"""Tries to get the last state from sensor
Returns True if a change has been made"""
ret = False
if self._is_configured:
# try to acquire presence entity state
presence_state = self.hass.states.get(self._presence_sensor_entity_id)
if presence_state and presence_state.state not in (
STATE_UNAVAILABLE,
STATE_UNKNOWN,
):
ret = await self.update_presence(presence_state.state)
_LOGGER.debug(
"%s - Presence have been retrieved: %s",
self,
presence_state.state,
)
return ret
@callback
async def _presence_sensor_changed(self, event: Event[EventStateChangedData]):
"""Handle presence changes."""
new_state = event.data.get("new_state")
_LOGGER.info(
"%s - Presence changed. Event.new_state is %s, _attr_preset_mode=%s, activity=%s",
self,
new_state,
self._vtherm.preset_mode,
PRESET_ACTIVITY,
)
if new_state is None:
return
if await self.update_presence(new_state.state):
await self._vtherm.async_control_heating(force=True)
async def update_presence(self, new_state: str) -> bool:
"""Update the value of the presence sensor and update the VTherm state accordingly
Return true if a change has been made"""
_LOGGER.info("%s - Updating presence. New state is %s", self, new_state)
old_presence_state = self._presence_state
self._presence_state = (
STATE_ON if new_state in (STATE_ON, STATE_HOME) else STATE_OFF
)
if self._vtherm.preset_mode in HIDDEN_PRESETS or self._is_configured is False:
_LOGGER.info(
"%s - Ignoring presence change cause in Power or Security preset or presence not configured",
self,
)
return old_presence_state != self._presence_state
if new_state is None or new_state not in (
STATE_OFF,
STATE_ON,
STATE_HOME,
STATE_NOT_HOME,
):
self._presence_state = STATE_UNKNOWN
return old_presence_state != self._presence_state
if self._vtherm.preset_mode not in [
PRESET_BOOST,
PRESET_COMFORT,
PRESET_ECO,
PRESET_ACTIVITY,
]:
return old_presence_state != self._presence_state
new_temp = self._vtherm.find_preset_temp(self._vtherm.preset_mode)
if new_temp is not None:
_LOGGER.debug(
"%s - presence change in temperature mode new_temp will be: %.2f",
self,
new_temp,
)
await self._vtherm.change_target_temperature(new_temp)
self._vtherm.recalculate()
return True
return old_presence_state != self._presence_state
def add_custom_attributes(self, extra_state_attributes: dict[str, Any]):
"""Add some custom attributes"""
extra_state_attributes.update(
{
"presence_sensor_entity_id": self._presence_sensor_entity_id,
"presence_state": self._presence_state,
"presence_configured": self._is_configured,
}
)
@overrides
@property
def is_configured(self) -> bool:
"""Return True of the presence is configured"""
return self._is_configured
@property
def presence_state(self) -> str | None:
"""Return the current presence state STATE_ON or STATE_OFF
or STATE_UNAVAILABLE if not configured"""
return self._presence_state
@property
def is_absence_detected(self) -> bool:
"""Return true if the presence is configured and presence sensor is OFF"""
return self._is_configured and self._presence_state in [
STATE_NOT_HOME,
STATE_OFF,
]
def __str__(self):
return f"PresenceManager-{self.name}"

2
custom_components/versatile_thermostat/sensor.py
View File

@@ -35,7 +35,7 @@ from homeassistant.components.climate import (
from .base_thermostat import BaseThermostat
from .vtherm_api import VersatileThermostatAPI
from .commons import VersatileThermostatBaseEntity
from .base_entity import VersatileThermostatBaseEntity
from .const import (
DOMAIN,
DEVICE_MANUFACTURER,

2
custom_components/versatile_thermostat/switch.py
View File

@@ -12,7 +12,7 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .commons import VersatileThermostatBaseEntity
from .base_entity import VersatileThermostatBaseEntity
from .const import * # pylint: disable=unused-wildcard-import,wildcard-import

6
custom_components/versatile_thermostat/thermostat_climate.py
View File

@@ -178,9 +178,9 @@ class ThermostatOverClimate(BaseThermostat[UnderlyingClimate]):
return self.calculate_hvac_action(self._underlyings)
@overrides
async def _async_internal_set_temperature(self, temperature: float):
async def change_target_temperature(self, temperature: float):
"""Set the target temperature and the target temperature of underlying climate if any"""
await super()._async_internal_set_temperature(temperature)
await super().change_target_temperature(temperature)
self._regulation_algo.set_target_temp(self.target_temperature)
# Is necessary cause control_heating method will not force the update.
@@ -593,7 +593,7 @@ class ThermostatOverClimate(BaseThermostat[UnderlyingClimate]):
_LOGGER.info(
"%s - Force resent target temp cause we turn on some over climate"
)
await self._async_internal_set_temperature(self._target_temp)
await self.change_target_temperature(self._target_temp)
@overrides
def incremente_energy(self):

23
tests/commons.py
View File

@@ -1,4 +1,4 @@
# pylint: disable=wildcard-import, unused-wildcard-import, protected-access, unused-argument, line-too-long, abstract-method, too-many-lines, redefined-builtin
# pylint: disable=wildcard-import, unused-wildcard-import, unused-import, protected-access, unused-argument, line-too-long, abstract-method, too-many-lines, redefined-builtin
""" Some common resources """
import asyncio
@@ -8,7 +8,16 @@ from unittest.mock import patch, MagicMock # pylint: disable=unused-import
import pytest # pylint: disable=unused-import
from homeassistant.core import HomeAssistant, Event, EVENT_STATE_CHANGED, State
from homeassistant.const import UnitOfTemperature, STATE_ON, STATE_OFF, ATTR_TEMPERATURE
from homeassistant.const import (
UnitOfTemperature,
STATE_ON,
STATE_OFF,
ATTR_TEMPERATURE,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
STATE_HOME,
STATE_NOT_HOME,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.helpers.entity import Entity
@@ -837,7 +846,7 @@ async def send_motion_change_event(
async def send_presence_change_event(
entity: BaseThermostat, new_state: bool, old_state: bool, date, sleep=True
vtherm: BaseThermostat, new_state: bool, old_state: bool, date, sleep=True
):
"""Sending a new presence event simulating a change on the window state"""
_LOGGER.info(
@@ -845,26 +854,26 @@ async def send_presence_change_event(
new_state,
old_state,
date,
entity,
vtherm,
)
presence_event = Event(
EVENT_STATE_CHANGED,
{
"new_state": State(
entity_id=entity.entity_id,
entity_id=vtherm.entity_id,
state=STATE_ON if new_state else STATE_OFF,
last_changed=date,
last_updated=date,
),
"old_state": State(
entity_id=entity.entity_id,
entity_id=vtherm.entity_id,
state=STATE_ON if old_state else STATE_OFF,
last_changed=date,
last_updated=date,
),
},
)
ret = await entity._async_presence_changed(presence_event)
ret = await vtherm._presence_manager._presence_sensor_changed(presence_event)
if sleep:
await asyncio.sleep(0.1)
return ret

175
tests/test_presence.py Normal file
View File

@@ -0,0 +1,175 @@
# pylint: disable=wildcard-import, unused-wildcard-import, protected-access, unused-argument, line-too-long
""" Test the Security featrure """
import logging
from unittest.mock import patch, call, AsyncMock, MagicMock, PropertyMock
# from datetime import timedelta, datetime
from custom_components.versatile_thermostat.base_thermostat import BaseThermostat
from custom_components.versatile_thermostat.presence_manager import (
FeaturePresenceManager,
)
from .commons import *
logging.getLogger().setLevel(logging.DEBUG)
@pytest.mark.parametrize(
"temp, absence, state, nb_call, presence_state, changed",
[
(19, False, STATE_ON, 1, STATE_ON, True),
(17, True, STATE_OFF, 1, STATE_OFF, True),
(19, False, STATE_HOME, 1, STATE_ON, True),
(17, True, STATE_NOT_HOME, 1, STATE_OFF, True),
(17, False, STATE_UNAVAILABLE, 0, STATE_UNKNOWN, False),
(17, False, STATE_UNKNOWN, 0, STATE_UNKNOWN, False),
(17, False, "wrong state", 0, STATE_UNKNOWN, False),
],
)
async def test_presence_feature_manager(
hass: HomeAssistant, temp, absence, state, nb_call, presence_state, changed
):
"""Test the FeaturePresenceManager class direclty"""
fake_vtherm = MagicMock(spec=BaseThermostat)
type(fake_vtherm).name = PropertyMock(return_value="the name")
type(fake_vtherm).preset_mode = PropertyMock(return_value=PRESET_COMFORT)
# 1. creation
presence_manager = FeaturePresenceManager(fake_vtherm, hass)
assert presence_manager is not None
assert presence_manager.is_configured is False
assert presence_manager.is_absence_detected is False
assert presence_manager.presence_state == STATE_UNAVAILABLE
assert presence_manager.name == "the name"
assert len(presence_manager._active_listener) == 0
custom_attributes = {}
presence_manager.add_custom_attributes(custom_attributes)
assert custom_attributes["presence_sensor_entity_id"] is None
assert custom_attributes["presence_state"] == STATE_UNAVAILABLE
assert custom_attributes["presence_configured"] is False
# 2. post_init
presence_manager.post_init(
{
CONF_PRESENCE_SENSOR: "sensor.the_presence_sensor",
CONF_USE_PRESENCE_FEATURE: True,
}
)
assert presence_manager.is_configured is True
assert presence_manager.presence_state == STATE_UNKNOWN
assert presence_manager.is_absence_detected is False
custom_attributes = {}
presence_manager.add_custom_attributes(custom_attributes)
assert (
custom_attributes["presence_sensor_entity_id"] == "sensor.the_presence_sensor"
)
assert custom_attributes["presence_state"] == STATE_UNKNOWN
assert custom_attributes["presence_configured"] is True
# 3. start listening
presence_manager.start_listening()
assert presence_manager.is_configured is True
assert presence_manager.presence_state == STATE_UNKNOWN
assert presence_manager.is_absence_detected is False
assert len(presence_manager._active_listener) == 1
# 4. test refresh with the parametrized
# fmt:off
with patch("homeassistant.core.StateMachine.get", return_value=State("sensor.the_presence_sensor", state)) as mock_get_state:
# fmt:on
# Configurer les méthodes mockées
fake_vtherm.find_preset_temp.return_value = temp
fake_vtherm.change_target_temperature = AsyncMock()
fake_vtherm.async_control_heating = AsyncMock()
ret = await presence_manager.refresh_state()
assert ret == changed
assert presence_manager.is_configured is True
assert presence_manager.presence_state == presence_state
assert presence_manager.is_absence_detected is absence
assert mock_get_state.call_count == 1
assert fake_vtherm.find_preset_temp.call_count == nb_call
if nb_call == 1:
fake_vtherm.find_preset_temp.assert_has_calls(
[
call.find_preset_temp(PRESET_COMFORT),
]
)
assert fake_vtherm.change_target_temperature.call_count == nb_call
fake_vtherm.change_target_temperature.assert_has_calls(
[
call.find_preset_temp(temp),
]
)
assert fake_vtherm.async_control_heating.call_count == 0
fake_vtherm.reset_mock()
# 5. Check custom_attributes
custom_attributes = {}
presence_manager.add_custom_attributes(custom_attributes)
assert custom_attributes["presence_sensor_entity_id"] == "sensor.the_presence_sensor"
assert custom_attributes["presence_state"] == presence_state
assert custom_attributes["presence_configured"] is True
# 6. test _presence_sensor_changed with the parametrized
fake_vtherm.find_preset_temp.return_value = temp
fake_vtherm.change_target_temperature = AsyncMock()
fake_vtherm.async_control_heating = AsyncMock()
await presence_manager._presence_sensor_changed(
event=Event(
event_type=EVENT_STATE_CHANGED,
data={
"entity_id": "sensor.the_presence_sensor",
"new_state": State("sensor.the_presence_sensor", state),
"old_state": State("sensor.the_presence_sensor", STATE_UNAVAILABLE),
}))
assert ret == changed
assert presence_manager.is_configured is True
assert presence_manager.presence_state == presence_state
assert presence_manager.is_absence_detected is absence
assert fake_vtherm.find_preset_temp.call_count == nb_call
if nb_call == 1:
fake_vtherm.find_preset_temp.assert_has_calls(
[
call.find_preset_temp(PRESET_COMFORT),
]
)
assert fake_vtherm.change_target_temperature.call_count == nb_call
fake_vtherm.change_target_temperature.assert_has_calls(
[
call.find_preset_temp(temp),
]
)
assert fake_vtherm.async_control_heating.call_count == 1
fake_vtherm.async_control_heating.assert_has_calls([
call.async_control_heating(force=True)
])
fake_vtherm.reset_mock()
# 7. Check custom_attributes
custom_attributes = {}
presence_manager.add_custom_attributes(custom_attributes)
assert custom_attributes["presence_sensor_entity_id"] == "sensor.the_presence_sensor"
assert custom_attributes["presence_state"] == presence_state
assert custom_attributes["presence_configured"] is True