From 0e499e07d2030c5681394ad9a12b5f5813b7f911 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 13 Oct 2023 14:24:23 -1000 Subject: [PATCH] Small cleanups to HomeKit thermostats (#101962) --- .../components/homekit/type_thermostats.py | 100 ++++++++++-------- .../homekit/test_type_thermostats.py | 60 ++++++++--- 2 files changed, 102 insertions(+), 58 deletions(-) diff --git a/homeassistant/components/homekit/type_thermostats.py b/homeassistant/components/homekit/type_thermostats.py index 85ad713012b..1fc8b3f2430 100644 --- a/homeassistant/components/homekit/type_thermostats.py +++ b/homeassistant/components/homekit/type_thermostats.py @@ -1,5 +1,6 @@ """Class to hold all thermostat accessories.""" import logging +from typing import Any from pyhap.const import CATEGORY_THERMOSTAT @@ -56,6 +57,7 @@ from homeassistant.const import ( UnitOfTemperature, ) from homeassistant.core import State, callback +from homeassistant.util.enum import try_parse_enum from homeassistant.util.percentage import ( ordered_list_item_to_percentage, percentage_to_ordered_list_item, @@ -163,17 +165,27 @@ HC_HASS_TO_HOMEKIT_FAN_STATE = { HEAT_COOL_DEADBAND = 5 +def _hk_hvac_mode_from_state(state: State) -> int | None: + """Return the equivalent HomeKit HVAC mode for a given state.""" + if not (hvac_mode := try_parse_enum(HVACMode, state.state)): + _LOGGER.error( + "%s: Received invalid HVAC mode: %s", state.entity_id, state.state + ) + return None + return HC_HASS_TO_HOMEKIT.get(hvac_mode) + + @TYPES.register("Thermostat") class Thermostat(HomeAccessory): """Generate a Thermostat accessory for a climate.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a Thermostat accessory object.""" super().__init__(*args, category=CATEGORY_THERMOSTAT) self._unit = self.hass.config.units.temperature_unit - self.hc_homekit_to_hass = None - self.hc_hass_to_homekit = None - hc_min_temp, hc_max_temp = self.get_temperature_range() + state = self.hass.states.get(self.entity_id) + assert state + hc_min_temp, hc_max_temp = self.get_temperature_range(state) self._reload_on_change_attrs.extend( ( ATTR_MIN_HUMIDITY, @@ -185,9 +197,9 @@ class Thermostat(HomeAccessory): ) # Add additional characteristics if auto mode is supported - self.chars = [] - self.fan_chars = [] - state: State = self.hass.states.get(self.entity_id) + self.chars: list[str] = [] + self.fan_chars: list[str] = [] + attributes = state.attributes min_humidity = attributes.get(ATTR_MIN_HUMIDITY, DEFAULT_MIN_HUMIDITY) features = attributes.get(ATTR_SUPPORTED_FEATURES, 0) @@ -285,8 +297,8 @@ class Thermostat(HomeAccessory): CHAR_CURRENT_HUMIDITY, value=50 ) - fan_modes = {} - self.ordered_fan_speeds = [] + fan_modes: dict[str, str] = {} + self.ordered_fan_speeds: list[str] = [] if features & ClimateEntityFeature.FAN_MODE: fan_modes = { @@ -358,13 +370,13 @@ class Thermostat(HomeAccessory): serv_thermostat.setter_callback = self._set_chars - def _set_fan_swing_mode(self, swing_on) -> None: + def _set_fan_swing_mode(self, swing_on: int) -> None: _LOGGER.debug("%s: Set swing mode to %s", self.entity_id, swing_on) mode = self.swing_on_mode if swing_on else SWING_OFF params = {ATTR_ENTITY_ID: self.entity_id, ATTR_SWING_MODE: mode} self.async_call_service(DOMAIN_CLIMATE, SERVICE_SET_SWING_MODE, params) - def _set_fan_speed(self, speed) -> None: + def _set_fan_speed(self, speed: int) -> None: _LOGGER.debug("%s: Set fan speed to %s", self.entity_id, speed) mode = percentage_to_ordered_list_item(self.ordered_fan_speeds, speed - 1) params = {ATTR_ENTITY_ID: self.entity_id, ATTR_FAN_MODE: mode} @@ -375,7 +387,7 @@ class Thermostat(HomeAccessory): return percentage_to_ordered_list_item(self.ordered_fan_speeds, 50) return self.fan_modes[FAN_ON] - def _set_fan_active(self, active) -> None: + def _set_fan_active(self, active: int) -> None: _LOGGER.debug("%s: Set fan active to %s", self.entity_id, active) if FAN_OFF not in self.fan_modes: _LOGGER.debug( @@ -388,28 +400,27 @@ class Thermostat(HomeAccessory): params = {ATTR_ENTITY_ID: self.entity_id, ATTR_FAN_MODE: mode} self.async_call_service(DOMAIN_CLIMATE, SERVICE_SET_FAN_MODE, params) - def _set_fan_auto(self, auto) -> None: + def _set_fan_auto(self, auto: int) -> None: _LOGGER.debug("%s: Set fan auto to %s", self.entity_id, auto) mode = self.fan_modes[FAN_AUTO] if auto else self._get_on_mode() params = {ATTR_ENTITY_ID: self.entity_id, ATTR_FAN_MODE: mode} self.async_call_service(DOMAIN_CLIMATE, SERVICE_SET_FAN_MODE, params) - def _temperature_to_homekit(self, temp): + def _temperature_to_homekit(self, temp: float | int) -> float: return temperature_to_homekit(temp, self._unit) - def _temperature_to_states(self, temp): + def _temperature_to_states(self, temp: float | int) -> float: return temperature_to_states(temp, self._unit) - def _set_chars(self, char_values): + def _set_chars(self, char_values: dict[str, Any]) -> None: _LOGGER.debug("Thermostat _set_chars: %s", char_values) events = [] - params = {ATTR_ENTITY_ID: self.entity_id} + params: dict[str, Any] = {ATTR_ENTITY_ID: self.entity_id} service = None state = self.hass.states.get(self.entity_id) + assert state features = state.attributes.get(ATTR_SUPPORTED_FEATURES, 0) - - hvac_mode = state.state - homekit_hvac_mode = HC_HASS_TO_HOMEKIT[hvac_mode] + homekit_hvac_mode = _hk_hvac_mode_from_state(state) # Homekit will reset the mode when VIEWING the temp # Ignore it if its the same mode if ( @@ -493,10 +504,12 @@ class Thermostat(HomeAccessory): CHAR_HEATING_THRESHOLD_TEMPERATURE in char_values or CHAR_COOLING_THRESHOLD_TEMPERATURE in char_values ): + assert self.char_cooling_thresh_temp + assert self.char_heating_thresh_temp service = SERVICE_SET_TEMPERATURE_THERMOSTAT high = self.char_cooling_thresh_temp.value low = self.char_heating_thresh_temp.value - min_temp, max_temp = self.get_temperature_range() + min_temp, max_temp = self.get_temperature_range(state) if CHAR_COOLING_THRESHOLD_TEMPERATURE in char_values: events.append( f"{CHAR_COOLING_THRESHOLD_TEMPERATURE} to" @@ -539,7 +552,7 @@ class Thermostat(HomeAccessory): if CHAR_TARGET_HUMIDITY in char_values: self.set_target_humidity(char_values[CHAR_TARGET_HUMIDITY]) - def _configure_hvac_modes(self, state): + def _configure_hvac_modes(self, state: State) -> None: """Configure target mode characteristics.""" # This cannot be none OR an empty list hc_modes = state.attributes.get(ATTR_HVAC_MODES) or DEFAULT_HVAC_MODES @@ -567,16 +580,16 @@ class Thermostat(HomeAccessory): } self.hc_hass_to_homekit = {k: v for v, k in self.hc_homekit_to_hass.items()} - def get_temperature_range(self): + def get_temperature_range(self, state: State) -> tuple[float, float]: """Return min and max temperature range.""" return _get_temperature_range_from_state( - self.hass.states.get(self.entity_id), + state, self._unit, DEFAULT_MIN_TEMP, DEFAULT_MAX_TEMP, ) - def set_target_humidity(self, value): + def set_target_humidity(self, value: float) -> None: """Set target humidity to value if call came from HomeKit.""" _LOGGER.debug("%s: Set target humidity to %d", self.entity_id, value) params = {ATTR_ENTITY_ID: self.entity_id, ATTR_HUMIDITY: value} @@ -585,15 +598,13 @@ class Thermostat(HomeAccessory): ) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update state without rechecking the device features.""" attributes = new_state.attributes features = attributes.get(ATTR_SUPPORTED_FEATURES, 0) # Update target operation mode FIRST - hvac_mode = new_state.state - if hvac_mode and hvac_mode in HC_HASS_TO_HOMEKIT: - homekit_hvac_mode = HC_HASS_TO_HOMEKIT[hvac_mode] + if (homekit_hvac_mode := _hk_hvac_mode_from_state(new_state)) is not None: if homekit_hvac_mode in self.hc_homekit_to_hass: self.char_target_heat_cool.set_value(homekit_hvac_mode) else: @@ -602,7 +613,7 @@ class Thermostat(HomeAccessory): "Cannot map hvac target mode: %s to homekit as only %s modes" " are supported" ), - hvac_mode, + new_state.state, self.hc_homekit_to_hass, ) @@ -618,12 +629,14 @@ class Thermostat(HomeAccessory): # Update current humidity if CHAR_CURRENT_HUMIDITY in self.chars: + assert self.char_current_humidity current_humdity = attributes.get(ATTR_CURRENT_HUMIDITY) if isinstance(current_humdity, (int, float)): self.char_current_humidity.set_value(current_humdity) # Update target humidity if CHAR_TARGET_HUMIDITY in self.chars: + assert self.char_target_humidity target_humdity = attributes.get(ATTR_HUMIDITY) if isinstance(target_humdity, (int, float)): self.char_target_humidity.set_value(target_humdity) @@ -671,7 +684,7 @@ class Thermostat(HomeAccessory): self._async_update_fan_state(new_state) @callback - def _async_update_fan_state(self, new_state): + def _async_update_fan_state(self, new_state: State) -> None: """Update state without rechecking the device features.""" attributes = new_state.attributes @@ -710,7 +723,7 @@ class Thermostat(HomeAccessory): class WaterHeater(HomeAccessory): """Generate a WaterHeater accessory for a water_heater.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a WaterHeater accessory object.""" super().__init__(*args, category=CATEGORY_THERMOSTAT) self._reload_on_change_attrs.extend( @@ -720,7 +733,9 @@ class WaterHeater(HomeAccessory): ) ) self._unit = self.hass.config.units.temperature_unit - min_temp, max_temp = self.get_temperature_range() + state = self.hass.states.get(self.entity_id) + assert state + min_temp, max_temp = self.get_temperature_range(state) serv_thermostat = self.add_preload_service(SERV_THERMOSTAT) @@ -751,25 +766,24 @@ class WaterHeater(HomeAccessory): CHAR_TEMP_DISPLAY_UNITS, value=0 ) - state = self.hass.states.get(self.entity_id) self.async_update_state(state) - def get_temperature_range(self): + def get_temperature_range(self, state: State) -> tuple[float, float]: """Return min and max temperature range.""" return _get_temperature_range_from_state( - self.hass.states.get(self.entity_id), + state, self._unit, DEFAULT_MIN_TEMP_WATER_HEATER, DEFAULT_MAX_TEMP_WATER_HEATER, ) - def set_heat_cool(self, value): + def set_heat_cool(self, value: int) -> None: """Change operation mode to value if call came from HomeKit.""" _LOGGER.debug("%s: Set heat-cool to %d", self.entity_id, value) if HC_HOMEKIT_TO_HASS[value] != HVACMode.HEAT: self.char_target_heat_cool.set_value(1) # Heat - def set_target_temperature(self, value): + def set_target_temperature(self, value: float) -> None: """Set target temperature to value if call came from HomeKit.""" _LOGGER.debug("%s: Set target temperature to %.1f°C", self.entity_id, value) temperature = temperature_to_states(value, self._unit) @@ -782,7 +796,7 @@ class WaterHeater(HomeAccessory): ) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update water_heater state after state change.""" # Update current and target temperature target_temperature = _get_target_temperature(new_state, self._unit) @@ -803,7 +817,9 @@ class WaterHeater(HomeAccessory): self.char_target_heat_cool.set_value(1) # Heat -def _get_temperature_range_from_state(state, unit, default_min, default_max): +def _get_temperature_range_from_state( + state: State, unit: str, default_min: float, default_max: float +) -> tuple[float, float]: """Calculate the temperature range from a state.""" if min_temp := state.attributes.get(ATTR_MIN_TEMP): min_temp = round(temperature_to_homekit(min_temp, unit) * 2) / 2 @@ -825,7 +841,7 @@ def _get_temperature_range_from_state(state, unit, default_min, default_max): return min_temp, max_temp -def _get_target_temperature(state, unit): +def _get_target_temperature(state: State, unit: str) -> float | None: """Calculate the target temperature from a state.""" target_temp = state.attributes.get(ATTR_TEMPERATURE) if isinstance(target_temp, (int, float)): @@ -833,7 +849,7 @@ def _get_target_temperature(state, unit): return None -def _get_current_temperature(state, unit): +def _get_current_temperature(state: State, unit: str) -> float | None: """Calculate the current temperature from a state.""" target_temp = state.attributes.get(ATTR_CURRENT_TEMPERATURE) if isinstance(target_temp, (int, float)): diff --git a/tests/components/homekit/test_type_thermostats.py b/tests/components/homekit/test_type_thermostats.py index a4ce765d795..1c3fb0914f3 100644 --- a/tests/components/homekit/test_type_thermostats.py +++ b/tests/components/homekit/test_type_thermostats.py @@ -106,7 +106,9 @@ async def test_thermostat(hass: HomeAssistant, hk_driver, events) -> None: assert acc.aid == 1 assert acc.category == 9 # Thermostat - assert acc.get_temperature_range() == (7.0, 35.0) + state = hass.states.get(entity_id) + assert state + assert acc.get_temperature_range(state) == (7.0, 35.0) assert acc.char_current_heat_cool.value == 0 assert acc.char_target_heat_cool.value == 0 assert acc.char_current_temp.value == 21.0 @@ -841,7 +843,9 @@ async def test_thermostat_fahrenheit(hass: HomeAssistant, hk_driver, events) -> }, ) await hass.async_block_till_done() - assert acc.get_temperature_range() == (7.0, 35.0) + state = hass.states.get(entity_id) + assert state + assert acc.get_temperature_range(state) == (7.0, 35.0) assert acc.char_heating_thresh_temp.value == 20.1 assert acc.char_cooling_thresh_temp.value == 24.0 assert acc.char_current_temp.value == 23.0 @@ -929,14 +933,18 @@ async def test_thermostat_get_temperature_range(hass: HomeAssistant, hk_driver) entity_id, HVACMode.OFF, {ATTR_MIN_TEMP: 20, ATTR_MAX_TEMP: 25} ) await hass.async_block_till_done() - assert acc.get_temperature_range() == (20, 25) + state = hass.states.get(entity_id) + assert state + assert acc.get_temperature_range(state) == (20, 25) acc._unit = UnitOfTemperature.FAHRENHEIT hass.states.async_set( entity_id, HVACMode.OFF, {ATTR_MIN_TEMP: 60, ATTR_MAX_TEMP: 70} ) await hass.async_block_till_done() - assert acc.get_temperature_range() == (15.5, 21.0) + state = hass.states.get(entity_id) + assert state + assert acc.get_temperature_range(state) == (15.5, 21.0) async def test_thermostat_temperature_step_whole( @@ -982,9 +990,14 @@ async def test_thermostat_restore(hass: HomeAssistant, hk_driver, events) -> Non hass.bus.async_fire(EVENT_HOMEASSISTANT_START, {}) await hass.async_block_till_done() - acc = Thermostat(hass, hk_driver, "Climate", "climate.simple", 2, None) + entity_id = "climate.simple" + hass.states.async_set(entity_id, HVACMode.OFF) + + acc = Thermostat(hass, hk_driver, "Climate", entity_id, 2, None) assert acc.category == 9 - assert acc.get_temperature_range() == (7, 35) + state = hass.states.get(entity_id) + assert state + assert acc.get_temperature_range(state) == (7, 35) assert set(acc.char_target_heat_cool.properties["ValidValues"].keys()) == { "cool", "heat", @@ -992,9 +1005,13 @@ async def test_thermostat_restore(hass: HomeAssistant, hk_driver, events) -> Non "off", } - acc = Thermostat(hass, hk_driver, "Climate", "climate.all_info_set", 3, None) + entity_id = "climate.all_info_set" + state = hass.states.get(entity_id) + assert state + + acc = Thermostat(hass, hk_driver, "Climate", entity_id, 3, None) assert acc.category == 9 - assert acc.get_temperature_range() == (60.0, 70.0) + assert acc.get_temperature_range(state) == (60.0, 70.0) assert set(acc.char_target_heat_cool.properties["ValidValues"].keys()) == { "heat_cool", "off", @@ -1762,15 +1779,19 @@ async def test_water_heater_get_temperature_range( hass.states.async_set( entity_id, HVACMode.HEAT, {ATTR_MIN_TEMP: 20, ATTR_MAX_TEMP: 25} ) + state = hass.states.get(entity_id) + assert state await hass.async_block_till_done() - assert acc.get_temperature_range() == (20, 25) + assert acc.get_temperature_range(state) == (20, 25) acc._unit = UnitOfTemperature.FAHRENHEIT hass.states.async_set( entity_id, HVACMode.OFF, {ATTR_MIN_TEMP: 60, ATTR_MAX_TEMP: 70} ) + state = hass.states.get(entity_id) + assert state await hass.async_block_till_done() - assert acc.get_temperature_range() == (15.5, 21.0) + assert acc.get_temperature_range(state) == (15.5, 21.0) async def test_water_heater_restore(hass: HomeAssistant, hk_driver, events) -> None: @@ -1795,20 +1816,27 @@ async def test_water_heater_restore(hass: HomeAssistant, hk_driver, events) -> N hass.bus.async_fire(EVENT_HOMEASSISTANT_START, {}) await hass.async_block_till_done() - acc = Thermostat(hass, hk_driver, "WaterHeater", "water_heater.simple", 2, None) + entity_id = "water_heater.simple" + hass.states.async_set(entity_id, "off") + state = hass.states.get(entity_id) + assert state + + acc = Thermostat(hass, hk_driver, "WaterHeater", entity_id, 2, None) assert acc.category == 9 - assert acc.get_temperature_range() == (7, 35) + assert acc.get_temperature_range(state) == (7, 35) assert set(acc.char_current_heat_cool.properties["ValidValues"].keys()) == { "Cool", "Heat", "Off", } - acc = WaterHeater( - hass, hk_driver, "WaterHeater", "water_heater.all_info_set", 3, None - ) + entity_id = "water_heater.all_info_set" + state = hass.states.get(entity_id) + assert state + + acc = WaterHeater(hass, hk_driver, "WaterHeater", entity_id, 3, None) assert acc.category == 9 - assert acc.get_temperature_range() == (60.0, 70.0) + assert acc.get_temperature_range(state) == (60.0, 70.0) assert set(acc.char_current_heat_cool.properties["ValidValues"].keys()) == { "Cool", "Heat",