Small cleanups to HomeKit thermostats (#101962)

This commit is contained in:
J. Nick Koston 2023-10-13 14:24:23 -10:00 committed by GitHub
parent f8f39a29de
commit 0e499e07d2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 102 additions and 58 deletions

View File

@ -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)):

View File

@ -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",