jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-19 14:57:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make evohome strictly typed (#106012)

Browse Source 
* initial commit

* return to conventional approach

* add type hint for wrapper

* use walrus operator
This commit is contained in:
David Bonnes 2023-12-21 12:22:42 +00:00 committed by GitHub
parent 2b65fb22d3
commit aa9f00099d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 130 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@ -126,6 +126,7 @@ homeassistant.components.energy.*
homeassistant.components.esphome.*
homeassistant.components.event.*
homeassistant.components.evil_genius_labs.*
homeassistant.components.evohome.*
homeassistant.components.faa_delays.*
homeassistant.components.fan.*
homeassistant.components.fastdotcom.*

117
homeassistant/components/evohome/__init__.py
View File

@ -4,15 +4,16 @@ Such systems include evohome, Round Thermostat, and others.
"""
from __future__ import annotations
from datetime import datetime as dt, timedelta
from collections.abc import Awaitable
from datetime import datetime, timedelta
from http import HTTPStatus
import logging
import re
from typing import Any
import evohomeasync
import evohomeasync as ev1
from evohomeasync.schema import SZ_ID, SZ_SESSION_ID, SZ_TEMP
import evohomeasync2
import evohomeasync2 as evo
from evohomeasync2.schema.const import (
SZ_ALLOWED_SYSTEM_MODES,
SZ_AUTO_WITH_RESET,
@ -112,15 +113,15 @@ SET_ZONE_OVERRIDE_SCHEMA = vol.Schema(
# system mode schemas are built dynamically, below
def _dt_local_to_aware(dt_naive: dt) -> dt:
dt_aware = dt_util.now() + (dt_naive - dt.now())
def _dt_local_to_aware(dt_naive: datetime) -> datetime:
dt_aware = dt_util.now() + (dt_naive - datetime.now())
if dt_aware.microsecond >= 500000:
dt_aware += timedelta(seconds=1)
return dt_aware.replace(microsecond=0)
def _dt_aware_to_naive(dt_aware: dt) -> dt:
dt_naive = dt.now() + (dt_aware - dt_util.now())
def _dt_aware_to_naive(dt_aware: datetime) -> datetime:
dt_naive = datetime.now() + (dt_aware - dt_util.now())
if dt_naive.microsecond >= 500000:
dt_naive += timedelta(seconds=1)
return dt_naive.replace(microsecond=0)
@ -157,12 +158,12 @@ def convert_dict(dictionary: dict[str, Any]) -> dict[str, Any]:
}
def _handle_exception(err) -> None:
def _handle_exception(err: evo.RequestFailed) -> None:
"""Return False if the exception can't be ignored."""
try:
raise err
except evohomeasync2.AuthenticationFailed:
except evo.AuthenticationFailed:
_LOGGER.error(
(
"Failed to authenticate with the vendor's server. Check your username"
@ -173,7 +174,7 @@ def _handle_exception(err) -> None:
err,
)
except evohomeasync2.RequestFailed:
except evo.RequestFailed:
if err.status is None:
_LOGGER.warning(
(
@ -206,7 +207,7 @@ def _handle_exception(err) -> None:
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Create a (EMEA/EU-based) Honeywell TCC system."""
async def load_auth_tokens(store) -> tuple[dict[str, str | dt], dict[str, str]]:
async def load_auth_tokens(store: Store) -> tuple[dict, dict | None]:
app_storage = await store.async_load()
tokens = dict(app_storage or {})
@ -227,16 +228,16 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
store = Store[dict[str, Any]](hass, STORAGE_VER, STORAGE_KEY)
tokens, user_data = await load_auth_tokens(store)
client_v2 = evohomeasync2.EvohomeClient(
client_v2 = evo.EvohomeClient(
config[DOMAIN][CONF_USERNAME],
config[DOMAIN][CONF_PASSWORD],
**tokens, # type: ignore[arg-type]
**tokens,
session=async_get_clientsession(hass),
)
try:
await client_v2.login()
except evohomeasync2.AuthenticationFailed as err:
except evo.AuthenticationFailed as err:
_handle_exception(err)
return False
finally:
@ -268,7 +269,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
_config[GWS][0][TCS] = loc_config[GWS][0][TCS]
_LOGGER.debug("Config = %s", _config)
client_v1 = evohomeasync.EvohomeClient(
client_v1 = ev1.EvohomeClient(
client_v2.username,
client_v2.password,
session_id=user_data.get(SZ_SESSION_ID) if user_data else None, # STORAGE_VER 1
@ -301,7 +302,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
@callback
def setup_service_functions(hass: HomeAssistant, broker):
def setup_service_functions(hass: HomeAssistant, broker: EvoBroker) -> None:
"""Set up the service handlers for the system/zone operating modes.
Not all Honeywell TCC-compatible systems support all operating modes. In addition,
@ -401,7 +402,7 @@ def setup_service_functions(hass: HomeAssistant, broker):
DOMAIN,
SVC_SET_SYSTEM_MODE,
set_system_mode,
schema=vol.Any(*system_mode_schemas),
schema=vol.Schema(vol.Any(*system_mode_schemas)),
)
# The zone modes are consistent across all systems and use the same schema
@ -425,8 +426,8 @@ class EvoBroker:
def __init__(
self,
hass: HomeAssistant,
client: evohomeasync2.EvohomeClient,
client_v1: evohomeasync.EvohomeClient | None,
client: evo.EvohomeClient,
client_v1: ev1.EvohomeClient | None,
store: Store[dict[str, Any]],
params: ConfigType,
) -> None:
@ -438,11 +439,11 @@ class EvoBroker:
self.params = params
loc_idx = params[CONF_LOCATION_IDX]
self._location: evo.Location = client.locations[loc_idx]
self.config = client.installation_info[loc_idx][GWS][0][TCS][0]
self.tcs = client.locations[loc_idx]._gateways[0]._control_systems[0]
self.tcs_utc_offset = timedelta(
minutes=client.locations[loc_idx].timeZone[UTC_OFFSET]
)
self.tcs: evo.ControlSystem = self._location._gateways[0]._control_systems[0]
self.tcs_utc_offset = timedelta(minutes=self._location.timeZone[UTC_OFFSET])
self.temps: dict[str, float | None] = {}
async def save_auth_tokens(self) -> None:
@ -461,38 +462,46 @@ class EvoBroker:
if self.client_v1:
app_storage[USER_DATA] = { # type: ignore[assignment]
"sessionId": self.client_v1.broker.session_id,
SZ_SESSION_ID: self.client_v1.broker.session_id,
} # this is the schema for STORAGE_VER == 1
else:
app_storage[USER_DATA] = {} # type: ignore[assignment]
await self._store.async_save(app_storage)
async def call_client_api(self, api_function, update_state=True) -> Any:
async def call_client_api(
self,
api_function: Awaitable[dict[str, Any] | None],
update_state: bool = True,
) -> dict[str, Any] | None:
"""Call a client API and update the broker state if required."""
try:
result = await api_function
except evohomeasync2.EvohomeError as err:
except evo.RequestFailed as err:
_handle_exception(err)
return
return None
if update_state: # wait a moment for system to quiesce before updating state
async_call_later(self.hass, 1, self._update_v2_api_state)
return result
async def _update_v1_api_temps(self, *args, **kwargs) -> None:
async def _update_v1_api_temps(self) -> None:
"""Get the latest high-precision temperatures of the default Location."""
assert self.client_v1 # mypy check
assert self.client_v1 is not None # mypy check
session_id = self.client_v1.broker.session_id # maybe receive a new session_id?
def get_session_id(client_v1: ev1.EvohomeClient) -> str | None:
user_data = client_v1.user_data if client_v1 else None
return user_data.get(SZ_SESSION_ID) if user_data else None # type: ignore[return-value]
session_id = get_session_id(self.client_v1)
self.temps = {} # these are now stale, will fall back to v2 temps
try:
temps = await self.client_v1.get_temperatures()
except evohomeasync.InvalidSchema as exc:
except ev1.InvalidSchema as err:
_LOGGER.warning(
(
"Unable to obtain high-precision temperatures. "
@ -500,11 +509,11 @@ class EvoBroker:
"so the high-precision feature will be disabled until next restart."
"Message is: %s"
),
exc,
err,
)
self.client_v1 = None
except evohomeasync.EvohomeError as exc:
except ev1.RequestFailed as err:
_LOGGER.warning(
(
"Unable to obtain the latest high-precision temperatures. "
@ -512,14 +521,11 @@ class EvoBroker:
"Proceeding without high-precision temperatures for now. "
"Message is: %s"
),
exc,
err,
)
else:
if (
str(self.client_v1.location_id)
!= self.client.locations[self.params[CONF_LOCATION_IDX]].locationId
):
if str(self.client_v1.location_id) != self._location.locationId:
_LOGGER.warning(
"The v2 API's configured location doesn't match "
"the v1 API's default location (there is more than one location), "
@ -535,15 +541,14 @@ class EvoBroker:
_LOGGER.debug("Temperatures = %s", self.temps)
async def _update_v2_api_state(self, *args, **kwargs) -> None:
async def _update_v2_api_state(self, *args: Any) -> None:
"""Get the latest modes, temperatures, setpoints of a Location."""
access_token = self.client.access_token # maybe receive a new token?
loc_idx = self.params[CONF_LOCATION_IDX]
try:
status = await self.client.locations[loc_idx].refresh_status()
except evohomeasync2.EvohomeError as err:
status = await self._location.refresh_status()
except evo.RequestFailed as err:
_handle_exception(err)
else:
async_dispatcher_send(self.hass, DOMAIN)
@ -553,7 +558,7 @@ class EvoBroker:
if access_token != self.client.access_token:
await self.save_auth_tokens()
async def async_update(self, *args, **kwargs) -> None:
async def async_update(self, *args: Any) -> None:
"""Get the latest state data of an entire Honeywell TCC Location.
This includes state data for a Controller and all its child devices, such as the
@ -575,9 +580,11 @@ class EvoDevice(Entity):
_attr_should_poll = False
_evo_id: str
def __init__(self, evo_broker, evo_device) -> None:
def __init__(
self,
evo_broker: EvoBroker,
evo_device: evo.ControlSystem | evo.HotWater | evo.Zone,
) -> None:
"""Initialize the evohome entity."""
self._evo_device = evo_device
self._evo_broker = evo_broker
@ -629,9 +636,14 @@ class EvoChild(EvoDevice):
This includes (up to 12) Heating Zones and (optionally) a DHW controller.
"""
def __init__(self, evo_broker, evo_device) -> None:
_evo_id: str # mypy hint
def __init__(
self, evo_broker: EvoBroker, evo_device: evo.HotWater | evo.Zone
) -> None:
"""Initialize a evohome Controller (hub)."""
super().__init__(evo_broker, evo_device)
self._schedule: dict[str, Any] = {}
self._setpoints: dict[str, Any] = {}
@ -639,6 +651,8 @@ class EvoChild(EvoDevice):
def current_temperature(self) -> float | None:
"""Return the current temperature of a Zone."""
assert isinstance(self._evo_device, evo.HotWater | evo.Zone) # mypy check
if self._evo_broker.temps.get(self._evo_id) is not None:
return self._evo_broker.temps[self._evo_id]
return self._evo_device.temperature
@ -650,7 +664,7 @@ class EvoChild(EvoDevice):
Only Zones & DHW controllers (but not the TCS) can have schedules.
"""
def _dt_evo_to_aware(dt_naive: dt, utc_offset: timedelta) -> dt:
def _dt_evo_to_aware(dt_naive: datetime, utc_offset: timedelta) -> datetime:
dt_aware = dt_naive.replace(tzinfo=dt_util.UTC) - utc_offset
return dt_util.as_local(dt_aware)
@ -686,7 +700,7 @@ class EvoChild(EvoDevice):
switchpoint_time_of_day = dt_util.parse_datetime(
f"{sp_date}T{switchpoint['TimeOfDay']}"
)
assert switchpoint_time_of_day # mypy check
assert switchpoint_time_of_day is not None # mypy check
dt_aware = _dt_evo_to_aware(
switchpoint_time_of_day, self._evo_broker.tcs_utc_offset
)
@ -708,7 +722,10 @@ class EvoChild(EvoDevice):
async def _update_schedule(self) -> None:
"""Get the latest schedule, if any."""
self._schedule = await self._evo_broker.call_client_api(
assert isinstance(self._evo_device, evo.HotWater | evo.Zone) # mypy check
self._schedule = await self._evo_broker.call_client_api( # type: ignore[assignment]
self._evo_device.get_schedule(), update_state=False
)

41
homeassistant/components/evohome/climate.py
View File

@ -1,10 +1,11 @@
"""Support for Climate devices of (EMEA/EU-based) Honeywell TCC systems."""
from __future__ import annotations
from datetime import datetime as dt
from datetime import datetime, timedelta
import logging
from typing import Any
from typing import TYPE_CHECKING, Any
import evohomeasync2 as evo
from evohomeasync2.schema.const import (
SZ_ACTIVE_FAULTS,
SZ_ALLOWED_SYSTEM_MODES,
@ -61,6 +62,10 @@ from .const import (
EVO_TEMPOVER,
)
if TYPE_CHECKING:
from . import EvoBroker
_LOGGER = logging.getLogger(__name__)
PRESET_RESET = "Reset" # reset all child zones to EVO_FOLLOW
@ -104,7 +109,7 @@ async def async_setup_platform(
if discovery_info is None:
return
broker = hass.data[DOMAIN]["broker"]
broker: EvoBroker = hass.data[DOMAIN]["broker"]
_LOGGER.debug(
"Found the Location/Controller (%s), id=%s, name=%s (location_idx=%s)",
@ -163,16 +168,19 @@ class EvoZone(EvoChild, EvoClimateEntity):
_attr_preset_modes = list(HA_PRESET_TO_EVO)
def __init__(self, evo_broker, evo_device) -> None:
_evo_device: evo.Zone # mypy hint
def __init__(self, evo_broker: EvoBroker, evo_device: evo.Zone) -> None:
"""Initialize a Honeywell TCC Zone."""
super().__init__(evo_broker, evo_device)
self._evo_id = evo_device.zoneId
if evo_device.modelType.startswith("VisionProWifi"):
# this system does not have a distinct ID for the zone
self._attr_unique_id = f"{evo_device.zoneId}z"
else:
self._attr_unique_id = evo_device.zoneId
self._evo_id = evo_device.zoneId
self._attr_name = evo_device.name
@ -197,7 +205,7 @@ class EvoZone(EvoChild, EvoClimateEntity):
temperature = max(min(data[ATTR_ZONE_TEMP], self.max_temp), self.min_temp)
if ATTR_DURATION_UNTIL in data:
duration = data[ATTR_DURATION_UNTIL]
duration: timedelta = data[ATTR_DURATION_UNTIL]
if duration.total_seconds() == 0:
await self._update_schedule()
until = dt_util.parse_datetime(self.setpoints.get("next_sp_from", ""))
@ -232,6 +240,8 @@ class EvoZone(EvoChild, EvoClimateEntity):
"""Return the current preset mode, e.g., home, away, temp."""
if self._evo_tcs.system_mode in (EVO_AWAY, EVO_HEATOFF):
return TCS_PRESET_TO_HA.get(self._evo_tcs.system_mode)
if self._evo_device.mode is None:
return None
return EVO_PRESET_TO_HA.get(self._evo_device.mode)
@property
@ -252,6 +262,9 @@ class EvoZone(EvoChild, EvoClimateEntity):
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set a new target temperature."""
assert self._evo_device.setpointStatus is not None # mypy check
temperature = kwargs["temperature"]
if (until := kwargs.get("until")) is None:
@ -300,14 +313,15 @@ class EvoZone(EvoChild, EvoClimateEntity):
await self._evo_broker.call_client_api(self._evo_device.reset_mode())
return
temperature = self._evo_device.target_heat_temperature
if evo_preset_mode == EVO_TEMPOVER:
await self._update_schedule()
until = dt_util.parse_datetime(self.setpoints.get("next_sp_from", ""))
else: # EVO_PERMOVER
until = None
temperature = self._evo_device.target_heat_temperature
assert temperature is not None # mypy check
until = dt_util.as_utc(until) if until else None
await self._evo_broker.call_client_api(
self._evo_device.set_temperature(temperature, until=until)
@ -334,12 +348,15 @@ class EvoController(EvoClimateEntity):
_attr_icon = "mdi:thermostat"
_attr_precision = PRECISION_TENTHS
def __init__(self, evo_broker, evo_device) -> None:
_evo_device: evo.ControlSystem # mypy hint
def __init__(self, evo_broker: EvoBroker, evo_device: evo.ControlSystem) -> None:
"""Initialize a Honeywell TCC Controller/Location."""
super().__init__(evo_broker, evo_device)
self._evo_id = evo_device.systemId
self._attr_unique_id = evo_device.systemId
self._evo_id = evo_device.systemId
self._attr_name = evo_device.location.name
modes = [m[SZ_SYSTEM_MODE] for m in evo_broker.config[SZ_ALLOWED_SYSTEM_MODES]]
@ -371,11 +388,11 @@ class EvoController(EvoClimateEntity):
await self._set_tcs_mode(mode, until=until)
async def _set_tcs_mode(self, mode: str, until: dt | None = None) -> None:
async def _set_tcs_mode(self, mode: str, until: datetime | None = None) -> None:
"""Set a Controller to any of its native EVO_* operating modes."""
until = dt_util.as_utc(until) if until else None
await self._evo_broker.call_client_api(
self._evo_tcs.set_mode(mode, until=until)
self._evo_tcs.set_mode(mode, until=until) # type: ignore[arg-type]
)
@property

31
homeassistant/components/evohome/water_heater.py
View File

@ -2,7 +2,9 @@
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any
import evohomeasync2 as evo
from evohomeasync2.schema.const import (
SZ_ACTIVE_FAULTS,
SZ_DHW_ID,
@ -31,6 +33,10 @@ import homeassistant.util.dt as dt_util
from . import EvoChild
from .const import DOMAIN, EVO_FOLLOW, EVO_PERMOVER
if TYPE_CHECKING:
from . import EvoBroker
_LOGGER = logging.getLogger(__name__)
STATE_AUTO = "auto"
@ -51,7 +57,9 @@ async def async_setup_platform(
if discovery_info is None:
return
broker = hass.data[DOMAIN]["broker"]
broker: EvoBroker = hass.data[DOMAIN]["broker"]
assert broker.tcs.hotwater is not None # mypy check
_LOGGER.debug(
"Adding: DhwController (%s), id=%s",
@ -72,12 +80,15 @@ class EvoDHW(EvoChild, WaterHeaterEntity):
_attr_operation_list = list(HA_STATE_TO_EVO)
_attr_temperature_unit = UnitOfTemperature.CELSIUS
def __init__(self, evo_broker, evo_device) -> None:
_evo_device: evo.HotWater # mypy hint
def __init__(self, evo_broker: EvoBroker, evo_device: evo.HotWater) -> None:
"""Initialize an evohome DHW controller."""
super().__init__(evo_broker, evo_device)
self._evo_id = evo_device.dhwId
self._attr_unique_id = evo_device.dhwId
self._evo_id = evo_device.dhwId
self._attr_precision = (
PRECISION_TENTHS if evo_broker.client_v1 else PRECISION_WHOLE
@ -87,15 +98,19 @@ class EvoDHW(EvoChild, WaterHeaterEntity):
)
@property
def current_operation(self) -> str:
def current_operation(self) -> str | None:
"""Return the current operating mode (Auto, On, or Off)."""
if self._evo_device.mode == EVO_FOLLOW:
return STATE_AUTO
return EVO_STATE_TO_HA[self._evo_device.state]
if (device_state := self._evo_device.state) is None:
return None
return EVO_STATE_TO_HA[device_state]
@property
def is_away_mode_on(self):
def is_away_mode_on(self) -> bool | None:
"""Return True if away mode is on."""
if self._evo_device.state is None:
return None
is_off = EVO_STATE_TO_HA[self._evo_device.state] == STATE_OFF
is_permanent = self._evo_device.mode == EVO_PERMOVER
return is_off and is_permanent
@ -129,11 +144,11 @@ class EvoDHW(EvoChild, WaterHeaterEntity):
"""Turn away mode off."""
await self._evo_broker.call_client_api(self._evo_device.reset_mode())
async def async_turn_on(self):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on."""
await self._evo_broker.call_client_api(self._evo_device.set_on())
async def async_turn_off(self):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off."""
await self._evo_broker.call_client_api(self._evo_device.set_off())

10
mypy.ini
View File

@ -1021,6 +1021,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.evohome.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.faa_delays.*]
check_untyped_defs = true
disallow_incomplete_defs = true