Refactored evohome, weather and presence clients and added caching for evohome schedules

Author: lexbrugman

src/evohome_helper/evohome.py

@@ -1,22 +1,14 @@
-import asyncio
 import logging
 import settings
 
 from datetime import datetime, timedelta
-from evohomeasync2 import EvohomeClientOld as EvohomeClient, ControlSystem, Location, Zone
+from evohomeasync2 import ControlSystem, Location, Zone
 from evohomeasync2.schemas import SystemMode, ZoneMode
+from evohome_helper import evohome_client
 from evohome_helper import weather
-from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential
 from typing import Generator
 
 logger = logging.getLogger(__name__)
-_evohome_client = None
-_retry = retry(
-    retry=retry_if_exception_type(Exception),
-    wait=wait_exponential(),
-    stop=stop_after_attempt(6),
-    reraise=True,
-)
 
 _AWAY_MODE_MAP = {
     "auto": SystemMode.AUTO,
@@ -30,62 +22,10 @@
 _DAY_NAMES = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
 
 
-class LocationNotFound(Exception):
-    def __init__(self, location_name: str):
-        super().__init__(f"the location '{location_name}' does not exist in the evohome account")
-
-
-@_retry
-async def _client() -> EvohomeClient:
-    global _evohome_client
-
-    if _evohome_client is None:
-        new_client = EvohomeClient(
-            settings.EVOHOME_USERNAME,
-            settings.EVOHOME_PASSWORD,
-        )
-        await new_client.update(dont_update_status=True)
-        _evohome_client = new_client
-
-    return _evohome_client
-
-
 def get_current_time(location: Location) -> datetime:
     return location.now().replace(microsecond=0)
 
 
-def get_control_systems(location: Location) -> Generator[ControlSystem, None, None]:
-    for gateway in location.gateways:
-        for control_system in gateway.systems:
-            yield control_system
-
-
-@_retry
-async def _update_location(location: Location) -> None:
-    await location.update()
-
-
-@_retry
-async def _fetch_schedules(system: ControlSystem) -> None:
-    await system.get_schedules()
-
-
-async def get_location(location_name: str = None) -> Location:
-    if not location_name:
-        location_name = settings.EVOHOME_LOCATION_NAME
-
-    client = await _client()
-    for location in client.locations:
-        if location.name == location_name:
-            await _update_location(location)
-            await asyncio.gather(
-                *[_fetch_schedules(system) for system in get_control_systems(location)],
-            )
-            return location
-
-    raise LocationNotFound(location_name)
-
-
 def _switchpoint_to_datetime(day_of_week: str, time_of_day: str, now: datetime) -> datetime:
     target_weekday = _DAY_NAMES.index(day_of_week)
     days_ago = (now.weekday() - target_weekday) % 7
@@ -154,7 +94,7 @@ def is_in_schedule_grace_period(location: Location) -> bool:
 
 
 def get_zones(location: Location) -> Generator[Zone, None, None]:
-    for control_system in get_control_systems(location):
+    for control_system in evohome_client.get_control_systems(location):
         for zone in control_system.zones:
             if zone.active_faults:
                 continue
@@ -197,7 +137,7 @@ async def _is_normal_heating_needed(location: Location) -> bool:
 
     # can we fetch a valid temperature?
     outside_current_temp = await weather.get_current_temperature()
-    if outside_current_temp <= -99:
+    if outside_current_temp is None:
         return True
 
     logger.debug(
@@ -222,13 +162,8 @@ def _get_highest_set_point_temp(location: Location) -> float | None:
     return max(valid_setpoints, default=None)
 
 
-@_retry
-async def _set_control_system_mode(control_system: ControlSystem, new_mode: SystemMode) -> None:
-    await control_system.set_mode(new_mode)
-
-
 async def _set_mode(new_mode: SystemMode, location: Location) -> None:
-    for control_system in get_control_systems(location):
+    for control_system in evohome_client.get_control_systems(location):
         current_mode = control_system.mode
         if new_mode == current_mode:
             continue
@@ -238,7 +173,7 @@ async def _set_mode(new_mode: SystemMode, location: Location) -> None:
             continue
 
         logger.debug("changing thermostat (%s) mode to '%s'", control_system.id, new_mode)
-        await _set_control_system_mode(control_system, new_mode)
+        await evohome_client.set_system_mode(control_system, new_mode)
 
 
 async def set_normal(location: Location) -> None:

src/evohome_helper/evohome_client.py

@@ -0,0 +1,151 @@
+import aiohttp
+import asyncio
+import json
+import logging
+import settings
+
+from datetime import datetime, timedelta
+from evohomeasync2 import EvohomeClient, ControlSystem, Location
+from evohomeasync2.auth import AbstractTokenManager
+from evohomeasync2.exceptions import ApiRequestFailedError, AuthenticationFailedError, BadUserCredentialsError
+from evohomeasync2.schemas import SystemMode
+from tenacity import retry, retry_if_exception_type, retry_if_not_exception_type, stop_after_attempt, wait_exponential
+from typing import Generator
+
+logger = logging.getLogger(__name__)
+_evohome_client: EvohomeClient | None = None
+_websession: aiohttp.ClientSession | None = None
+
+# only retry errors that can resolve on their own; invalid credentials never will
+_TRANSIENT_ERRORS = (
+    aiohttp.ClientError,
+    TimeoutError,
+    ApiRequestFailedError,
+    AuthenticationFailedError,
+)
+_retry = retry(
+    retry=retry_if_exception_type(_TRANSIENT_ERRORS) & retry_if_not_exception_type(BadUserCredentialsError),
+    wait=wait_exponential(),
+    stop=stop_after_attempt(6),
+    reraise=True,
+)
+
+# schedules rarely change; refetching them every cycle wastes the vendor's tight API rate limit
+_SCHEDULE_REFRESH_INTERVAL = timedelta(hours=1)
+_schedule_refresh_times: dict[str, datetime] = {}
+
+
+class LocationNotFound(Exception):
+    def __init__(self, location_name: str):
+        super().__init__(f"the location '{location_name}' does not exist in the evohome account")
+
+
+class _TokenManager(AbstractTokenManager):
+    """Caches auth tokens on disk so restarts reuse them instead of
+    re-authenticating against the heavily rate-limited vendor API."""
+
+    def __init__(self, username: str, password: str, websession: aiohttp.ClientSession, token_cache_path: str):
+        super().__init__(username, password, websession)
+        self._token_cache_path = token_cache_path
+
+    async def load_access_token(self) -> None:
+        try:
+            with open(self._token_cache_path) as f:
+                self._import_access_token(json.load(f))
+        except FileNotFoundError:
+            pass
+        except (KeyError, ValueError):
+            logger.warning("ignoring the invalid token cache at '%s'", self._token_cache_path)
+
+    async def save_access_token(self) -> None:
+        with open(self._token_cache_path, "w") as f:
+            json.dump(self._export_access_token(), f)
+
+
+async def close() -> None:
+    global _evohome_client, _websession
+
+    _schedule_refresh_times.clear()
+    _evohome_client = None
+
+    if _websession is not None:
+        await _websession.close()
+        _websession = None
+
+
+# discard the cached client so the next call recreates it from scratch
+reset_client = close
+
+
+@_retry
+async def _client() -> EvohomeClient:
+    global _evohome_client, _websession
+
+    if _evohome_client is None:
+        websession = aiohttp.ClientSession()
+        try:
+            token_manager = _TokenManager(
+                settings.EVOHOME_USERNAME,
+                settings.EVOHOME_PASSWORD,
+                websession,
+                settings.EVOHOME_TOKEN_CACHE_PATH,
+            )
+            await token_manager.load_access_token()
+
+            new_client = EvohomeClient(token_manager)
+            await new_client.update(dont_update_status=True)
+        except BaseException:
+            await websession.close()
+            raise
+
+        _websession = websession
+        _evohome_client = new_client
+
+    return _evohome_client
+
+
+def get_control_systems(location: Location) -> Generator[ControlSystem, None, None]:
+    for gateway in location.gateways:
+        for control_system in gateway.systems:
+            yield control_system
+
+
+async def get_location(location_name: str | None = None) -> Location:
+    if not location_name:
+        location_name = settings.EVOHOME_LOCATION_NAME
+
+    client = await _client()
+    for location in client.locations:
+        if location.name == location_name:
+            await _update_location(location)
+            await asyncio.gather(
+                *[
+                    _fetch_schedules(system)
+                    for system in get_control_systems(location)
+                    if _schedules_need_refresh(system)
+                ],
+            )
+            return location
+
+    raise LocationNotFound(location_name)
+
+
+@_retry
+async def set_system_mode(control_system: ControlSystem, new_mode: SystemMode) -> None:
+    await control_system.set_mode(new_mode)
+
+
+@_retry
+async def _update_location(location: Location) -> None:
+    await location.update()
+
+
+@_retry
+async def _fetch_schedules(system: ControlSystem) -> None:
+    await system.get_schedules()
+    _schedule_refresh_times[system.id] = datetime.now()
+
+
+def _schedules_need_refresh(system: ControlSystem) -> bool:
+    last_refresh = _schedule_refresh_times.get(system.id)
+    return last_refresh is None or datetime.now() - last_refresh >= _SCHEDULE_REFRESH_INTERVAL

src/evohome_helper/homeassistant.py

@@ -0,0 +1,42 @@
+import aiohttp
+import logging
+import settings
+
+logger = logging.getLogger(__name__)
+_session: aiohttp.ClientSession | None = None
+
+
+async def get_entity_state(entity_id: str) -> dict | None:
+    url = f"{settings.HOMEASSISTANT_URL}/api/states/{entity_id}"
+
+    try:
+        async with _get_session().get(url, headers=_headers()) as response:
+            response.raise_for_status()
+            return await response.json()
+    except Exception:
+        logger.exception("failed getting the state of entity '%s'", entity_id)
+        return None
+
+
+async def close() -> None:
+    global _session
+
+    if _session is not None:
+        await _session.close()
+        _session = None
+
+
+def _get_session() -> aiohttp.ClientSession:
+    global _session
+
+    if _session is None or _session.closed:
+        _session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=5))
+
+    return _session
+
+
+def _headers() -> dict:
+    return {
+        "Authorization": f"Bearer {settings.HOMEASSISTANT_TOKEN}",
+        "content-type": "application/json",
+    }

src/evohome_helper/presence.py

@@ -1,7 +1,8 @@
-import aiohttp
 import logging
 import settings
 
+from evohome_helper import homeassistant
+
 logger = logging.getLogger(__name__)
 last_known_presence_state = {}
 
@@ -26,30 +27,16 @@ async def is_in_away_grace_period() -> bool:
 
 
 async def _get_data(entity_id: str) -> dict:
-    url = f"{settings.HOMEASSISTANT_URL}/api/states/{entity_id}"
-
-    try:
-        async with aiohttp.ClientSession() as session:
-            async with session.get(url, headers=_headers(), timeout=aiohttp.ClientTimeout(total=5)) as response:
-                response.raise_for_status()
-                response_data = await response.json()
-                attributes = response_data.get("attributes", {})
-
-                last_known_presence_state[entity_id] = {
-                    "is_someone_home": response_data.get("state") == "home",
-                    "seconds_since_last_seen": attributes.get("seconds_since_last_seen"),
-                }
-    except Exception:
-        logger.exception("failed getting presence information")
+    entity_state = await homeassistant.get_entity_state(entity_id)
+    if entity_state is not None:
+        attributes = entity_state.get("attributes", {})
+
+        last_known_presence_state[entity_id] = {
+            "is_someone_home": entity_state.get("state") == "home",
+            "seconds_since_last_seen": attributes.get("seconds_since_last_seen"),
+        }
 
     return last_known_presence_state.setdefault(entity_id, {
         "is_someone_home": False,
         "seconds_since_last_seen": 0,
     })
-
-
-def _headers() -> dict:
-    return {
-        "Authorization": f"Bearer {settings.HOMEASSISTANT_TOKEN}",
-        "content-type": "application/json",
-    }