Update src/data/data_providers.py: circuit breaker, weather fix, TB health fix

src/data/data_providers.py

@@ -50,6 +50,46 @@ import numpy as np
 import pandas as pd
 
 
+# ═══════════════════════════════════════════════════════════════════════
+# Circuit breaker — fail-open to cached data when external services down
+# ═══════════════════════════════════════════════════════════════════════
+
+class CircuitBreaker:
+    """Simple circuit breaker: after `threshold` consecutive failures within
+    `window_sec`, the circuit opens and calls are short-circuited for
+    `cooldown_sec` before retrying.
+    """
+
+    def __init__(self, threshold: int = 3, cooldown_sec: float = 300, window_sec: float = 60):
+        self.threshold = threshold
+        self.cooldown_sec = cooldown_sec
+        self.window_sec = window_sec
+        self._failures: list[float] = []
+        self._opened_at: float | None = None
+
+    @property
+    def is_open(self) -> bool:
+        if self._opened_at is None:
+            return False
+        if time.monotonic() - self._opened_at > self.cooldown_sec:
+            # Cooldown expired — allow retry (half-open)
+            self._opened_at = None
+            self._failures.clear()
+            return False
+        return True
+
+    def record_success(self) -> None:
+        self._failures.clear()
+        self._opened_at = None
+
+    def record_failure(self) -> None:
+        now = time.monotonic()
+        self._failures = [t for t in self._failures if now - t < self.window_sec]
+        self._failures.append(now)
+        if len(self._failures) >= self.threshold:
+            self._opened_at = now
+
+
 # ═══════════════════════════════════════════════════════════════════════
 # TTL Cache helper
 # ═══════════════════════════════════════════════════════════════════════
@@ -104,9 +144,9 @@ class TTLCache:
         return None
 
     def set(self, key: str, value: Any) -> None:
-        # Write to Redis if available
+        # Write to Redis if available (skip DataFrames — too large for JSON serialisation)
         redis = self._get_redis()
-        if redis:
+        if redis and not isinstance(value, pd.DataFrame):
             redis.set_json(self._rkey(key), value, ttl=int(self.ttl))
         # Always write in-memory too (local fast path)
         self._store[key] = _CacheEntry(value=value, expires_at=time.monotonic() + self.ttl)
@@ -204,7 +244,10 @@ class WeatherService(BaseService):
     def _load_df(self) -> pd.DataFrame:
         cached = self._df_cache.get("ims")
         if cached is not None:
-            return cached
+            # Redis may deserialise as dict/list — only accept DataFrames
+            if isinstance(cached, pd.DataFrame):
+                return cached
+            # Discard stale non-DataFrame from Redis and reload from CSV
         df = self._client().load_cached()
         if not df.empty:
             self._df_cache.set("ims", df)
@@ -292,6 +335,7 @@ class VineSensorService(BaseService):
     def __init__(self, tb_client: Any = None, snapshot_ttl: float = 300):
         self._tb = tb_client            # lazy
         self._snap_cache = TTLCache(ttl_seconds=snapshot_ttl, redis_prefix="vine:")
+        self._breaker = CircuitBreaker(threshold=3, cooldown_sec=300)
 
     def _client(self):
         if self._tb is None:
@@ -316,12 +360,16 @@ class VineSensorService(BaseService):
         cached = self._snap_cache.get(cache_key)
         if cached is not None:
             return cached
+        if self._breaker.is_open:
+            return {"error": "ThingsBoard circuit breaker open — retrying in 5 min", "cached": True}
         try:
             snapshot = self._client().get_vine_snapshot(light=light, mode=mode)
             result = snapshot.to_dict()
             self._snap_cache.set(cache_key, result)
+            self._breaker.record_success()
             return result
         except Exception as exc:
+            self._breaker.record_failure()
             return {
                 "error": f"ThingsBoard unavailable: {exc}",
                 "hint": "Check THINGSBOARD_USERNAME/PASSWORD in .env",
@@ -823,6 +871,7 @@ class EnergyService(BaseService):
 
     def __init__(self, tb_client: Any = None):
         self._tb = tb_client
+        self._breaker = CircuitBreaker(threshold=3, cooldown_sec=300)
 
     def _client(self):
         if self._tb is None:
@@ -836,14 +885,18 @@ class EnergyService(BaseService):
 
     def get_current(self) -> Dict[str, Any]:
         """Latest power reading from the Plant asset."""
+        if self._breaker.is_open:
+            return {"error": "ThingsBoard circuit breaker open — retrying in 5 min"}
         try:
             vals = self._client().get_asset_latest("Plant", ["power", "production"])
             power_w = vals.get("power")
+            self._breaker.record_success()
             return {
                 "power_kw": round(power_w / 1000, 1) if power_w else None,
                 "source": "ThingsBoard Plant asset",
             }
         except Exception as exc:
+            self._breaker.record_failure()
             return {"error": f"Energy current failed: {exc}"}
 
     def get_daily_production(self, target_date: Optional[str] = None) -> Dict[str, Any]: