introduce time_cached for low-level APIs

Author: jkt628

franklinwh/client.py

@@ -5,9 +5,9 @@
 """
 
 from __future__ import annotations
-from collections.abc import Callable
 
 import asyncio
+from collections.abc import Callable
 from dataclasses import dataclass
 from enum import Enum
 import hashlib
@@ -19,6 +19,7 @@
 import httpx
 
 from .api import DEFAULT_URL_BASE
+from .timed_cache import time_cached
 
 
 class AccessoryType(Enum):
@@ -599,19 +600,22 @@ def set_value(keys, value):
         return json.loads(data)
 
     # Sends a 203 which is a high level status
+    @time_cached()
     async def _status(self):
         payload = self._build_payload(203, {"opt": 1, "refreshData": 1})
         data = (await self._mqtt_send(payload))["result"]["dataArea"]
         return json.loads(data)
 
     # Sends a 311 which appears to be a more specific switch command
+    @time_cached()
     async def _switch_status(self):
         payload = self._build_payload(311, {"opt": 0, "order": self.gateway})
         data = (await self._mqtt_send(payload))["result"]["dataArea"]
         return json.loads(data)
 
     # Sends a 353 which grabs real-time smart-circuit load information
     # https://github.com/richo/homeassistant-franklinwh/issues/27#issuecomment-2714422732
+    @time_cached()
     async def _switch_usage(self):
         payload = self._build_payload(353, {"opt": 0, "order": self.gateway})
         data = (await self._mqtt_send(payload))["result"]["dataArea"]
@@ -739,6 +743,7 @@ async def set_grid_status(self, status: GridStatus, soc: int = 5):
         }
         await self._post(url, json.dumps(payload))
 
+    @time_cached()
     async def get_composite_info(self):
         """Get composite information about the FranklinWH gateway."""
         url = self.url_base + "hes-gateway/terminal/getDeviceCompositeInfo"

franklinwh/timed_cache.py

@@ -0,0 +1,30 @@
+"""Cache a function result for a specified time."""
+
+import asyncio
+from datetime import datetime, timedelta
+from functools import wraps
+
+__cache = {}
+__lock = asyncio.Lock()
+
+
+def time_cached(ttl: timedelta = timedelta(seconds=2)):
+    """Decorator to cache function results for a specified time-to-live (TTL)."""
+
+    def wrapper(func):
+        @wraps(func)
+        async def wrapped(*args, **kwargs):
+            async with __lock:
+                now = datetime.now()
+                for key, value in __cache.copy().items():
+                    if now > value[0]:
+                        del __cache[key]
+                key = (func.__name__, args, frozenset(kwargs.items()))
+                if key not in __cache:
+                    __cache[key] = (now + ttl, await func(*args, **kwargs))
+                return __cache[key][1]
+
+        setattr(wrapped, "clear", __cache.clear)
+        return wrapped
+
+    return wrapper