SSE live updates, route cleanup, advisor briefing

- Add EventBus + SSE endpoint for push-based data updates
- Route all sensor/control/biology data through DataHub services
- Remove all direct ThingsBoard calls from route handlers
- Add live status briefing to chatbot advisor

backend/api/events.py

@@ -0,0 +1,52 @@
+"""
+In-process event bus for SSE-based live updates.
+
+Uses asyncio primitives only — no Redis, no external dependencies.
+Works within a single uvicorn process (which is what HF Spaces runs).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from dataclasses import dataclass, field
+from typing import Dict
+
+log = logging.getLogger("solarwine.events")
+
+
+@dataclass
+class EventBus:
+    """Lightweight publish/subscribe within one async process.
+
+    Publishers call ``notify(topic)`` to bump a version counter.
+    SSE consumers call ``wait()`` to block until any topic changes.
+
+    Uses an asyncio.Condition to avoid the set/clear race on asyncio.Event.
+    """
+
+    _versions: Dict[str, int] = field(default_factory=dict)
+    _condition: asyncio.Condition = field(default_factory=asyncio.Condition)
+
+    async def notify(self, topic: str) -> None:
+        async with self._condition:
+            self._versions[topic] = self._versions.get(topic, 0) + 1
+            self._condition.notify_all()
+        log.debug("EventBus: %s → v%d", topic, self._versions[topic])
+
+    async def wait(self, timeout: float = 30) -> Dict[str, int]:
+        """Block until a notification arrives or *timeout* seconds elapse."""
+        try:
+            async with self._condition:
+                await asyncio.wait_for(self._condition.wait(), timeout)
+        except asyncio.TimeoutError:
+            pass
+        return dict(self._versions)
+
+    @property
+    def versions(self) -> Dict[str, int]:
+        return dict(self._versions)
+
+
+# Module-level singleton — imported by routes and data_providers
+event_bus = EventBus()

backend/api/main.py

@@ -19,7 +19,7 @@ from slowapi import Limiter, _rate_limit_exceeded_handler
 from slowapi.errors import RateLimitExceeded
 from slowapi.util import get_remote_address
 
-from backend.api.routes import health, weather, sensors, energy, photosynthesis, control, chatbot, biology, login
+from backend.api.routes import health, weather, sensors, energy, photosynthesis, control, chatbot, biology, login, events
 
 # ---------------------------------------------------------------------------
 # Structured logging
@@ -82,6 +82,7 @@ async def _ims_refresh_loop(interval_sec: int = 6 * 3600):
     picks up fresh data.
     """
     import asyncio
+    from backend.api.events import event_bus
     await asyncio.sleep(10)  # let the server finish starting
     while True:
         try:
@@ -106,11 +107,45 @@ async def _ims_refresh_loop(interval_sec: int = 6 * 3600):
                 hub.weather._df_cache._store.clear()
             except Exception:
                 pass
+            await event_bus.notify("weather")
         except Exception as exc:
             log.error("IMS refresh failed: %s", exc)
         await asyncio.sleep(interval_sec)
 
 
+async def _sensor_refresh_loop(interval_sec: int = 120):
+    """Background loop: poll ThingsBoard sensors every `interval_sec` seconds.
+
+    Refreshes the VineSensorService and EnergyService caches, then
+    notifies SSE clients so they refetch immediately.
+    """
+    import asyncio
+    from backend.api.events import event_bus
+    await asyncio.sleep(15)  # let startup finish
+    while True:
+        try:
+            from backend.api.deps import get_datahub
+            hub = get_datahub()
+            # Refresh sensor snapshot — invalidate all cached variants
+            for key in ("snap", "snap_light", "dashboard"):
+                hub.vine_sensors._snap_cache.invalidate(key)
+            snapshot = hub.vine_sensors.get_snapshot()
+            if snapshot and "error" not in snapshot:
+                await event_bus.notify("sensors")
+                # Photosynthesis depends on sensor data
+                await event_bus.notify("photosynthesis")
+                log.debug("Sensor snapshot refreshed")
+            # Refresh energy current
+            hub.energy._current_cache.invalidate("current")
+            current = hub.energy.get_current()
+            if current and "error" not in current:
+                await event_bus.notify("energy")
+                log.debug("Energy current refreshed")
+        except Exception as exc:
+            log.error("Sensor refresh failed: %s", exc)
+        await asyncio.sleep(interval_sec)
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     global _start_time
@@ -124,12 +159,14 @@ async def lifespan(app: FastAPI):
     if not os.environ.get("ADMIN_PASSWORD"):
         log.warning("ADMIN_PASSWORD not set — login endpoint disabled")
 
-    # Start background IMS data refresh (every 6 hours)
+    # Start background refresh loops
     import asyncio
     ims_task = asyncio.create_task(_ims_refresh_loop())
+    sensor_task = asyncio.create_task(_sensor_refresh_loop())
 
     yield
     ims_task.cancel()
+    sensor_task.cancel()
     log.info("SolarWine API shutting down (uptime=%.0fs)", get_uptime())
 
 
@@ -202,3 +239,4 @@ app.include_router(control.router, prefix="/api/control", tags=["control"])
 app.include_router(chatbot.router, prefix="/api/chatbot", tags=["chatbot"])
 app.include_router(biology.router, prefix="/api/biology", tags=["biology"])
 app.include_router(login.router, prefix="/api/auth", tags=["auth"])
+app.include_router(events.router, prefix="/api/events", tags=["events"])

backend/api/routes/biology.py

@@ -48,90 +48,5 @@ async def biology_chill_units(
     season_start: str = Query("2025-11-01", description="Season start (YYYY-MM-DD)"),
     hub: DataHub = Depends(get_datahub),
 ):
-    """Accumulated chill units from ThingsBoard Air1 on-site temperature (Utah model).
-
-    Model (Richardson et al. 1974):
-      T <= 7°C   → +1.0 CU/hour
-      7 < T <= 10 → +0.5
-      10 < T <= 18 → 0.0
-      T > 18      → -1.0
-    Daily totals clipped at 0 (no negative daily chill).
-    Season cumulative = running sum of daily CU from season_start.
-
-    Two series:
-      - open_field: on-site Air1 temperature
-      - under_panels: open_field × 1.1 (panels buffer nighttime → more chill)
-    Per Research/chill_hours/ANALYSIS_EXPLAINED.md.
-    """
-    import numpy as np
-    import pandas as pd
-
-    PANEL_MULTIPLIER = 1.1
-
-    try:
-        from src.data.thingsboard_client import ThingsBoardClient
-        from zoneinfo import ZoneInfo
-
-        client = ThingsBoardClient()
-        tz = ZoneInfo("Asia/Jerusalem")
-        start = pd.Timestamp(season_start, tz="UTC")
-        end = pd.Timestamp.now(tz="UTC")
-
-        # Fetch Air1 temperature in 7-day chunks (TB rejects large ranges)
-        chunks = []
-        cursor = start
-        while cursor < end:
-            chunk_end = min(cursor + pd.Timedelta(days=7), end)
-            try:
-                df = client.get_timeseries(
-                    "Air1", ["airTemperature"],
-                    start=cursor.to_pydatetime(), end=chunk_end.to_pydatetime(),
-                    interval_ms=0, agg="NONE", limit=10000,
-                )
-                if not df.empty:
-                    chunks.append(df)
-            except Exception:
-                pass
-            cursor = chunk_end
-
-        if not chunks:
-            return {"error": "No Air1 temperature data available from ThingsBoard"}
-
-        full = pd.concat(chunks).sort_index()
-        full = full[~full.index.duplicated(keep="first")]
-        full = full.tz_convert(tz)
-
-        hourly = full["airTemperature"].resample("1h").mean().dropna()
-        if hourly.empty:
-            return {"error": "No hourly temperature after resampling"}
-
-        temps = hourly.values
-        chill_hourly = np.select(
-            [temps <= 7.0, (temps > 7.0) & (temps <= 10.0),
-             (temps > 10.0) & (temps <= 18.0), temps > 18.0],
-            [1.0, 0.5, 0.0, -1.0],
-        )
-
-        daily_chill = pd.Series(chill_hourly, index=hourly.index).resample("D").sum().clip(lower=0)
-        cu_open = daily_chill.cumsum()
-        cu_panels = (daily_chill * PANEL_MULTIPLIER).cumsum()
-
-        daily = [
-            {
-                "date": ts.strftime("%Y-%m-%d"),
-                "under_panels": round(float(cu_panels.loc[ts]), 1),
-                "open_field": round(float(cu_open.loc[ts]), 1),
-            }
-            for ts in daily_chill.index
-        ]
-
-        return {
-            "season_start": season_start,
-            "latest_under_panels": round(float(cu_panels.iloc[-1]), 1) if len(cu_panels) else 0,
-            "latest_open_field": round(float(cu_open.iloc[-1]), 1) if len(cu_open) else 0,
-            "days_counted": len(daily_chill),
-            "daily": daily,
-        }
-    except Exception as exc:
-        log.error("Chill units failed: %s", exc)
-        return {"error": f"Chill units failed: {exc}"}
+    """Accumulated chill units (Utah model, 6h TTL cache)."""
+    return hub.biology.get_chill_units(season_start=season_start)

backend/api/routes/chatbot.py

@@ -83,6 +83,19 @@ async def chat_message(request: Request, req: ChatRequest, hub: DataHub = Depend
         raise HTTPException(status_code=500, detail="Chatbot processing failed")
 
 
+@router.get("/briefing")
+async def chat_briefing(hub: DataHub = Depends(get_datahub)):
+    """Return a short live status briefing for the advisor panel."""
+    bot = _get_chatbot(hub)
+    if bot is None:
+        raise HTTPException(status_code=503, detail="Chatbot unavailable")
+    try:
+        return {"briefing": bot._build_status_briefing()}
+    except Exception as exc:
+        log.error("Briefing error: %s", exc)
+        return {"briefing": ""}
+
+
 @router.post("/feedback")
 @limiter.limit("60/minute")
 async def chat_feedback(request: Request, req: FeedbackRequest):

backend/api/routes/control.py

@@ -1,12 +1,13 @@
-"""Control system endpoints — reads state from Redis."""
+"""Control system endpoints — reads state from Redis + DataHub services."""
 
 from __future__ import annotations
 
 import logging
 
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, Depends, HTTPException
 
 from backend.api.deps import get_datahub, get_redis_client
+from src.data.data_providers import DataHub
 
 log = logging.getLogger(__name__)
 router = APIRouter()
@@ -56,48 +57,13 @@ async def control_budget():
 
 
 @router.get("/trackers/details")
-async def tracker_details():
-    """Tracker entity table — name, mode, angle, last activity for all trackers."""
-    try:
-        from src.data.thingsboard_client import ThingsBoardClient, TRACKER_KEYS
-        client = ThingsBoardClient()
-        trackers = []
-        for name in ["Tracker501", "Tracker502", "Tracker503", "Tracker509"]:
-            try:
-                vals = client.get_latest_telemetry(name, list(TRACKER_KEYS))
-                trackers.append({
-                    "name": name,
-                    "label": name.replace("Tracker", "Row "),
-                    "angle": round(float(vals.get("angle", 0)), 1) if vals.get("angle") is not None else None,
-                    "manual_mode": vals.get("manualMode"),
-                    "set_angle": round(float(vals.get("setAngle", 0)), 1) if vals.get("setAngle") is not None else None,
-                    "set_mode": vals.get("setMode"),
-                })
-            except Exception as exc:
-                trackers.append({"name": name, "label": name, "error": str(exc)})
-        return {"trackers": trackers}
-    except Exception as exc:
-        log.error("Tracker details failed: %s", exc)
-        return {"trackers": [], "error": str(exc)}
-
-
-_tracker_cache: dict = {"data": None, "expires": 0.0}
+async def tracker_details(hub: DataHub = Depends(get_datahub)):
+    """Tracker entity table — name, mode, angle for all trackers (cached)."""
+    return hub.vine_sensors.get_tracker_details()
 
 
 @router.get("/trackers")
-async def control_trackers():
-    """Live tracker angles from ThingsBoard (5-min TTL cache)."""
-    import time as _time
-    if _tracker_cache["data"] and _time.monotonic() < _tracker_cache["expires"]:
-        return _tracker_cache["data"]
-    try:
-        from src.tracker_dispatcher import TrackerDispatcher
-        dispatcher = TrackerDispatcher()
-        angles = dispatcher.read_current_angles()
-        result = {"trackers": angles, "source": "ThingsBoard"}
-        _tracker_cache["data"] = result
-        _tracker_cache["expires"] = _time.monotonic() + 300  # 5 min
-        return result
-    except Exception as exc:
-        log.error("Tracker fetch failed: %s", exc)
-        raise HTTPException(status_code=502, detail=f"Tracker fetch failed: {exc}")
+async def control_trackers(hub: DataHub = Depends(get_datahub)):
+    """Live tracker angles (cached via VineSensorService)."""
+    result = hub.vine_sensors.get_tracker_details()
+    return {"trackers": result.get("trackers", []), "source": "ThingsBoard"}

backend/api/routes/events.py

@@ -0,0 +1,55 @@
+"""
+SSE endpoint — pushes lightweight version updates to connected browsers.
+
+The client receives a JSON dict of topic versions whenever data changes.
+It then invalidates the matching TanStack Query cache, triggering a
+refetch via the existing REST endpoints.
+
+No new Redis commands — everything is in-memory via EventBus.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+
+from fastapi import APIRouter, Request
+from starlette.responses import StreamingResponse
+
+from backend.api.events import event_bus
+
+log = logging.getLogger("solarwine.sse")
+router = APIRouter()
+
+
+@router.get("/stream")
+async def event_stream(request: Request):
+    """Server-Sent Events stream.
+
+    Sends a ``versions`` message whenever the EventBus is notified,
+    or a keep-alive comment every ~25 s to prevent proxy timeouts.
+    """
+
+    async def generate():
+        # Send current versions immediately on connect
+        yield f"data: {json.dumps(event_bus.versions)}\n\n"
+        last_versions = dict(event_bus.versions)
+        while True:
+            if await request.is_disconnected():
+                break
+            versions = await event_bus.wait(timeout=25)
+            if versions != last_versions:
+                yield f"data: {json.dumps(versions)}\n\n"
+                last_versions = dict(versions)
+            else:
+                # Keep-alive comment to prevent proxy idle timeout
+                yield ": heartbeat\n\n"
+
+    return StreamingResponse(
+        generate(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",  # disable nginx/proxy buffering
+        },
+    )

backend/api/routes/sensors.py

@@ -1,4 +1,4 @@
-"""Vine sensor endpoints — wraps VineSensorService."""
+"""Vine sensor endpoints — all data routed through VineSensorService."""
 
 from __future__ import annotations
 
@@ -69,164 +69,79 @@ async def sensors_history(
 @router.get("/soil-moisture")
 async def soil_moisture_history(
     hours: int = Query(168, ge=1, le=8760, description="Hours of history"),
+    hub: DataHub = Depends(get_datahub),
 ):
-    """Hourly soil moisture time-series from ThingsBoard (treatment Soil1)."""
-    try:
-        from datetime import datetime, timezone, timedelta
-        from src.data.thingsboard_client import ThingsBoardClient
-
-        client = ThingsBoardClient()
-        end = datetime.now(tz=timezone.utc)
-        start = end - timedelta(hours=hours)
-
-        df = client.get_timeseries(
-            "Soil1", ["soilMoisture", "soilMoisture2"],
-            start=start, end=end,
-            interval_ms=3_600_000, agg="AVG", limit=2000,
-        )
-        if df.empty:
-            return []
-        rows = []
-        for ts, row in df.iterrows():
-            r = {"timestamp": ts.isoformat()}
-            for col in df.columns:
-                val = row[col]
-                if val is not None and val == val:  # not NaN
-                    r[col] = round(float(val), 1)
-            rows.append(r)
-        return rows
-    except Exception as exc:
-        log.error("Soil moisture history failed: %s", exc)
-        return []
-
-
-def _tb_timeseries(device: str, keys: list, hours: int):
-    """Fetch TB time-series with fallback: try AVG aggregation first, then NONE."""
-    from datetime import datetime, timezone, timedelta
-    from src.data.thingsboard_client import ThingsBoardClient
-    client = ThingsBoardClient()
-    end = datetime.now(tz=timezone.utc)
-    start = end - timedelta(hours=hours)
-
-    # Try aggregated first
-    df = client.get_timeseries(device, keys, start=start, end=end,
-        interval_ms=3_600_000, agg="AVG", limit=2000)
-    if not df.empty:
-        return df
-
-    # Fallback: raw data, resample locally
-    import pandas as pd
-    df = client.get_timeseries(device, keys, start=start, end=end,
-        interval_ms=0, agg="NONE", limit=10000)
-    if df.empty:
-        return df
-    return df.resample("1h").mean(numeric_only=True).dropna(how="all")
+    """Hourly soil moisture time-series (treatment Soil1)."""
+    rows = hub.vine_sensors.get_device_timeseries(
+        "Soil1", ["soilMoisture", "soilMoisture2"], hours_back=hours,
+    )
+    return rows
 
 
 @router.get("/air-leaf-delta")
 async def air_leaf_delta(
     hours: int = Query(168, ge=1, le=8760, description="Hours of history"),
+    hub: DataHub = Depends(get_datahub),
 ):
-    """Hourly air-leaf temperature delta from ThingsBoard (Air2 under panels)."""
-    try:
-        df = _tb_timeseries("Air1", ["airLeafDeltaT"], hours)
-        if df.empty:
-            return []
-        return [{"timestamp": ts.isoformat(), "value": round(float(row["airLeafDeltaT"]), 2)}
-                for ts, row in df.iterrows() if row.get("airLeafDeltaT") is not None and row["airLeafDeltaT"] == row["airLeafDeltaT"]]
-    except Exception as exc:
-        log.error("Air/leaf delta failed: %s", exc)
-        return []
+    """Hourly air-leaf temperature delta (Air1)."""
+    rows = hub.vine_sensors.get_device_timeseries(
+        "Air1", ["airLeafDeltaT"], hours_back=hours,
+    )
+    # Reshape to {timestamp, value} for frontend compatibility
+    return [{"timestamp": r["timestamp"], "value": r.get("airLeafDeltaT")}
+            for r in rows if r.get("airLeafDeltaT") is not None]
 
 
 @router.get("/temperature-humidity")
 async def temp_humidity(
     hours: int = Query(168, ge=1, le=8760, description="Hours of history"),
+    hub: DataHub = Depends(get_datahub),
 ):
-    """Hourly temperature + humidity from ThingsBoard (Air2 under panels)."""
-    try:
-        df = _tb_timeseries("Air1", ["airTemperature", "airHumidity"], hours)
-        if df.empty:
-            return []
-        rows = []
-        for ts, row in df.iterrows():
-            r = {"timestamp": ts.isoformat()}
-            for col in ["airTemperature", "airHumidity"]:
-                v = row.get(col)
-                if v is not None and v == v:  # NaN check (NaN != NaN)
-                    r[col] = round(float(v), 1)
-            rows.append(r)
-        return rows
-    except Exception as exc:
-        log.error("Temp/humidity failed: %s", exc)
-        return []
+    """Hourly temperature + humidity (Air1)."""
+    return hub.vine_sensors.get_device_timeseries(
+        "Air1", ["airTemperature", "airHumidity"], hours_back=hours,
+    )
 
 
 @router.get("/ndvi")
 async def ndvi_history(
     hours: int = Query(168, ge=1, le=8760, description="Hours of history"),
+    hub: DataHub = Depends(get_datahub),
 ):
-    """Hourly NDVI from ThingsBoard — treatment (Crop3) and ambient (Air1)."""
-    try:
-        rows = []
-        for device, label in [("Air1", "ambient"), ("Crop3", "treatment")]:
-            try:
-                df = _tb_timeseries(device, ["NDVI"], min(hours, 168))
-                if not df.empty and "NDVI" in df.columns:
-                    for ts, row in df.iterrows():
-                        v = row.get("NDVI")
-                        if v is not None and v == v:  # NaN check (NaN != NaN)
-                            rows.append({"timestamp": ts.isoformat(), "device": label, "ndvi": round(float(v), 4)})
-            except Exception:
-                pass
-        return rows
-    except Exception as exc:
-        log.error("NDVI failed: %s", exc)
-        return []
+    """Hourly NDVI — treatment (Crop3) and ambient (Air1)."""
+    rows = []
+    for device, label in [("Air1", "ambient"), ("Crop3", "treatment")]:
+        ts_rows = hub.vine_sensors.get_device_timeseries(
+            device, ["NDVI"], hours_back=min(hours, 168),
+        )
+        for r in ts_rows:
+            v = r.get("NDVI")
+            if v is not None:
+                rows.append({"timestamp": r["timestamp"], "device": label, "ndvi": round(v, 4)})
+    return rows
 
 
 @router.get("/vpd")
 async def vpd_history(
     hours: int = Query(168, ge=1, le=8760, description="Hours of history"),
+    hub: DataHub = Depends(get_datahub),
 ):
-    """Hourly VPD from ThingsBoard (Air2 under panels)."""
-    try:
-        df = _tb_timeseries("Air1", ["VPD"], hours)
-        if df.empty:
-            return []
-        return [{"timestamp": ts.isoformat(), "value": round(float(row["VPD"]), 2)}
-                for ts, row in df.iterrows() if row.get("VPD") is not None and row["VPD"] == row["VPD"]]
-    except Exception as exc:
-        log.error("VPD failed: %s", exc)
-        return []
+    """Hourly VPD (Air1)."""
+    rows = hub.vine_sensors.get_device_timeseries(
+        "Air1", ["VPD"], hours_back=hours,
+    )
+    return [{"timestamp": r["timestamp"], "value": r.get("VPD")}
+            for r in rows if r.get("VPD") is not None]
 
 
 @router.get("/rain")
 async def rain_history(
     hours: int = Query(168, ge=1, le=8760, description="Hours of history"),
+    hub: DataHub = Depends(get_datahub),
 ):
-    """Hourly rain time-series from ThingsBoard ambient station (Air1)."""
-    try:
-        from datetime import datetime, timezone, timedelta
-        from src.data.thingsboard_client import ThingsBoardClient
-
-        client = ThingsBoardClient()
-        end = datetime.now(tz=timezone.utc)
-        start = end - timedelta(hours=hours)
-
-        df = client.get_timeseries(
-            "Air1", ["rain"],
-            start=start, end=end,
-            interval_ms=3_600_000, agg="SUM", limit=2000,
-        )
-        if df.empty:
-            return []
-        rows = []
-        for ts, row in df.iterrows():
-            val = row.get("rain")
-            if val is not None and val == val:
-                rows.append({"timestamp": ts.isoformat(), "rain_mm": round(float(val), 2)})
-        return rows
-    except Exception as exc:
-        log.error("Rain history failed: %s", exc)
-        return []
+    """Hourly rain (Air1, SUM aggregation)."""
+    rows = hub.vine_sensors.get_device_timeseries(
+        "Air1", ["rain"], hours_back=hours, agg="SUM",
+    )
+    return [{"timestamp": r["timestamp"], "rain_mm": r.get("rain", 0)}
+            for r in rows if r.get("rain") is not None]

src/chatbot/vineyard_chatbot.py

@@ -557,11 +557,24 @@ class VineyardChatbot:
         """Build Gemini multi-turn message list with sliding context window.
 
         Strategy:
+          - Inject live status briefing as pinned context (from cached data)
           - Keep the most recent 6 messages verbatim (for conversational flow)
           - Summarize older messages into a single context message
-          - Always include pinned context (current date, season)
         """
         messages = []
+
+        # Inject live status briefing so the LLM has immediate context
+        briefing = self._build_status_briefing()
+        if briefing:
+            messages.append({
+                "role": "user",
+                "parts": [{"text": f"[System status — do not repeat verbatim, use as context]\n{briefing}"}],
+            })
+            messages.append({
+                "role": "model",
+                "parts": [{"text": "Got it, I have the current status."}],
+            })
+
         n = len(history)
 
         if n > self._RECENT_MESSAGES:
@@ -685,6 +698,108 @@ class VineyardChatbot:
             pass
         return ctx
 
+    # ------------------------------------------------------------------
+    # Live status briefing — injected at conversation start
+    # ------------------------------------------------------------------
+
+    def _build_status_briefing(self) -> str:
+        """Assemble a short system status from cached DataHub data.
+
+        Uses only already-cached values (no new API calls), so it adds
+        zero latency. Returns an empty string if nothing is available.
+        """
+        from datetime import datetime
+        import zoneinfo
+
+        lines: list[str] = []
+        tz = zoneinfo.ZoneInfo("Asia/Jerusalem")
+        now = datetime.now(tz=tz)
+        lines.append(f"CURRENT STATUS ({now.strftime('%Y-%m-%d %H:%M')} IST):")
+
+        # Weather
+        try:
+            wx = self.hub.weather.get_current()
+            if wx and "error" not in wx:
+                t = wx.get("air_temperature_c")
+                ghi = wx.get("ghi_w_m2")
+                rh = wx.get("rh_percent")
+                wind = wx.get("wind_speed_ms")
+                parts = []
+                if t is not None:
+                    parts.append(f"T={float(t):.1f}°C")
+                if ghi is not None:
+                    parts.append(f"GHI={float(ghi):.0f} W/m²")
+                if rh is not None:
+                    parts.append(f"RH={float(rh):.0f}%")
+                if wind is not None:
+                    parts.append(f"wind={float(wind):.1f} m/s")
+                if parts:
+                    lines.append(f"  Weather: {', '.join(parts)}")
+                age = wx.get("age_minutes")
+                if age is not None and float(age) > 30:
+                    lines.append(f"  (weather data is {int(float(age))} min old)")
+        except Exception:
+            pass
+
+        # Sensors
+        try:
+            snap = self.hub.vine_sensors.get_snapshot(light=True)
+            if snap and "error" not in snap:
+                parts = []
+                for key, label in [
+                    ("treatment_air_temp_c", "air"),
+                    ("treatment_crop_par_umol", "PAR"),
+                    ("treatment_soil_moisture_pct", "soil"),
+                ]:
+                    v = snap.get(key)
+                    if v is not None:
+                        if "temp" in key:
+                            parts.append(f"{label}={float(v):.1f}°C")
+                        elif "par" in key:
+                            parts.append(f"{label}={float(v):.0f} µmol")
+                        else:
+                            parts.append(f"{label}={float(v):.0f}%")
+                if parts:
+                    lines.append(f"  Sensors (treatment): {', '.join(parts)}")
+                stale = snap.get("staleness_minutes")
+                if stale is not None and float(stale) > 15:
+                    lines.append(f"  (sensors {int(float(stale))} min old)")
+        except Exception:
+            pass
+
+        # Energy
+        try:
+            en = self.hub.energy.get_current()
+            if en and "error" not in en:
+                pw = en.get("power_kw")
+                if pw is not None:
+                    lines.append(f"  Energy: {float(pw):.1f} kW now")
+        except Exception:
+            pass
+
+        # Phenology
+        try:
+            from src.phenology import estimate_stage_for_date
+            from datetime import date
+            stage = estimate_stage_for_date(date.today())
+            lines.append(f"  Phenology: {stage.name} ({stage.id})")
+        except Exception:
+            pass
+
+        # Control status (from Redis via hub — no direct Redis import)
+        try:
+            ctrl = self.hub.advisory.get_status()
+            if ctrl and "error" not in ctrl:
+                mode = ctrl.get("mode") or ctrl.get("action")
+                if mode:
+                    lines.append(f"  Control: {mode}")
+        except Exception:
+            pass
+
+        if len(lines) <= 1:
+            return ""
+        return "\n".join(lines)
+
     # ------------------------------------------------------------------
     # Main chat method
     # ------------------------------------------------------------------

src/data/data_providers.py

@@ -38,9 +38,12 @@ Loose coupling guarantees:
 
 from __future__ import annotations
 
+import logging
 import math
 import time
 import traceback
+
+log = logging.getLogger("solarwine.data_providers")
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
 from datetime import date, datetime, timedelta, timezone
@@ -335,6 +338,8 @@ 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._ts_cache = TTLCache(ttl_seconds=900, redis_prefix="vine_ts:")  # 15 min for time-series
+        self._tracker_cache = TTLCache(ttl_seconds=300, redis_prefix="tracker:")  # 5 min for trackers
         self._breaker = CircuitBreaker(threshold=3, cooldown_sec=300)
 
     def _client(self):
@@ -430,6 +435,99 @@ class VineSensorService(BaseService):
             log.error("Sensor history query failed: %s", exc)
             return {"error": f"Sensor history failed: {exc}"}
 
+    def get_device_timeseries(
+        self,
+        device: str,
+        keys: List[str],
+        hours_back: int = 168,
+        agg: str = "AVG",
+    ) -> List[Dict[str, Any]]:
+        """Hourly time-series for a specific device + keys (15-min TTL cache).
+
+        Returns a list of ``{timestamp, key1, key2, ...}`` dicts.
+        Used by sensor history endpoints (soil moisture, VPD, NDVI, etc.).
+        """
+        cache_key = f"{device}:{','.join(sorted(keys))}:{hours_back}:{agg}"
+        cached = self._ts_cache.get(cache_key)
+        if cached is not None:
+            return cached
+        if self._breaker.is_open:
+            return []
+        try:
+            end = datetime.now(tz=timezone.utc)
+            start = end - timedelta(hours=hours_back)
+            client = self._client()
+
+            # Try server-side aggregation first
+            df = client.get_timeseries(
+                device, keys, start=start, end=end,
+                interval_ms=3_600_000, agg=agg, limit=2000,
+            )
+            if df.empty:
+                # Fallback: raw data, resample locally
+                df = client.get_timeseries(
+                    device, keys, start=start, end=end,
+                    interval_ms=0, agg="NONE", limit=10000,
+                )
+                if not df.empty:
+                    df = df.resample("1h").mean(numeric_only=True).dropna(how="all")
+
+            if df.empty:
+                self._breaker.record_success()
+                result: List[Dict[str, Any]] = []
+                self._ts_cache.set(cache_key, result)
+                return result
+
+            rows: List[Dict[str, Any]] = []
+            for ts, row in df.iterrows():
+                r: Dict[str, Any] = {"timestamp": ts.isoformat()}
+                for col in df.columns:
+                    val = row[col]
+                    if val is not None and val == val:  # NaN check
+                        r[col] = round(float(val), 2)
+                rows.append(r)
+
+            self._breaker.record_success()
+            self._ts_cache.set(cache_key, rows)
+            return rows
+        except Exception as exc:
+            self._breaker.record_failure()
+            log.error("Device timeseries failed (%s): %s", device, exc)
+            return []
+
+    def get_tracker_details(self) -> Dict[str, Any]:
+        """Latest tracker angles/modes for all 4 trackers (5-min TTL cache)."""
+        cached = self._tracker_cache.get("details")
+        if cached is not None:
+            return cached
+        if self._breaker.is_open:
+            return {"trackers": [], "error": "ThingsBoard circuit breaker open"}
+        try:
+            client = self._client()
+            tracker_keys = ["angle", "manualMode", "setAngle", "setMode"]
+            trackers = []
+            for name in ["Tracker501", "Tracker502", "Tracker503", "Tracker509"]:
+                try:
+                    vals = client.get_latest_telemetry(name, tracker_keys)
+                    trackers.append({
+                        "name": name,
+                        "label": name.replace("Tracker", "Row "),
+                        "angle": round(float(vals.get("angle", 0)), 1) if vals.get("angle") is not None else None,
+                        "manual_mode": vals.get("manualMode"),
+                        "set_angle": round(float(vals.get("setAngle", 0)), 1) if vals.get("setAngle") is not None else None,
+                        "set_mode": vals.get("setMode"),
+                    })
+                except Exception as exc:
+                    trackers.append({"name": name, "label": name, "error": str(exc)})
+            result = {"trackers": trackers}
+            self._breaker.record_success()
+            self._tracker_cache.set("details", result)
+            return result
+        except Exception as exc:
+            self._breaker.record_failure()
+            log.error("Tracker details failed: %s", exc)
+            return {"trackers": [], "error": str(exc)}
+
 
 # ═══════════════════════════════════════════════════════════════════════
 # 3. PhotosynthesisService  (FvCB + ML + forecast)
@@ -1157,15 +1255,23 @@ class AdvisoryService(BaseService):
 # ═══════════════════════════════════════════════════════════════════════
 
 class BiologyService(BaseService):
-    """Biology rules lookup — pure in-memory, no API calls."""
+    """Biology rules lookup + chill unit computation."""
 
     service_name = "biology"
 
-    def __init__(self, rules: Optional[Dict[str, str]] = None):
+    def __init__(self, rules: Optional[Dict[str, str]] = None, tb_client: Any = None):
         if rules is None:
             from src.vineyard_chatbot import BIOLOGY_RULES
             rules = BIOLOGY_RULES
         self._rules = rules
+        self._tb = tb_client
+        self._chill_cache = TTLCache(ttl_seconds=21600, redis_prefix="biology:")  # 6h TTL
+
+    def _client(self):
+        if self._tb is None:
+            from src.data.thingsboard_client import ThingsBoardClient
+            self._tb = ThingsBoardClient()
+        return self._tb
 
     def explain_rule(self, rule_name: str) -> Dict[str, Any]:
         key = rule_name.lower().strip()
@@ -1176,6 +1282,91 @@ class BiologyService(BaseService):
     def list_rules(self) -> Dict[str, Any]:
         return {"rules": list(self._rules.keys())}
 
+    def get_chill_units(self, season_start: str = "2025-11-01") -> Dict[str, Any]:
+        """Accumulated chill units from ThingsBoard Air1 temperature (Utah model, 6h TTL).
+
+        Richardson et al. 1974:
+          T <= 7°C   → +1.0 CU/hour
+          7 < T <= 10 → +0.5
+          10 < T <= 18 → 0.0
+          T > 18      → -1.0
+        """
+        cache_key = f"chill:{season_start}"
+        cached = self._chill_cache.get(cache_key)
+        if cached is not None:
+            return cached
+
+        try:
+            import numpy as np
+            from zoneinfo import ZoneInfo
+
+            tz = ZoneInfo("Asia/Jerusalem")
+            client = self._client()
+            start = pd.Timestamp(season_start, tz="UTC")
+            end = pd.Timestamp.now(tz="UTC")
+
+            # Fetch Air1 temperature in 7-day chunks
+            chunks = []
+            cursor = start
+            while cursor < end:
+                chunk_end = min(cursor + pd.Timedelta(days=7), end)
+                try:
+                    df = client.get_timeseries(
+                        "Air1", ["airTemperature"],
+                        start=cursor.to_pydatetime(), end=chunk_end.to_pydatetime(),
+                        interval_ms=0, agg="NONE", limit=10000,
+                    )
+                    if not df.empty:
+                        chunks.append(df)
+                except Exception:
+                    pass
+                cursor = chunk_end
+
+            if not chunks:
+                return {"error": "No Air1 temperature data available from ThingsBoard"}
+
+            full = pd.concat(chunks).sort_index()
+            full = full[~full.index.duplicated(keep="first")]
+            full = full.tz_convert(tz)
+
+            hourly = full["airTemperature"].resample("1h").mean().dropna()
+            if hourly.empty:
+                return {"error": "No hourly temperature after resampling"}
+
+            PANEL_MULTIPLIER = 1.1
+            temps = hourly.values
+            chill_hourly = np.select(
+                [temps <= 7.0, (temps > 7.0) & (temps <= 10.0),
+                 (temps > 10.0) & (temps <= 18.0), temps > 18.0],
+                [1.0, 0.5, 0.0, -1.0],
+            )
+
+            daily_chill = pd.Series(chill_hourly, index=hourly.index).resample("D").sum().clip(lower=0)
+            cu_open = daily_chill.cumsum()
+            cu_panels = (daily_chill * PANEL_MULTIPLIER).cumsum()
+
+            daily = [
+                {
+                    "date": ts.strftime("%Y-%m-%d"),
+                    "under_panels": round(float(cu_panels.loc[ts]), 1),
+                    "open_field": round(float(cu_open.loc[ts]), 1),
+                }
+                for ts in daily_chill.index
+            ]
+
+            result = {
+                "season_start": season_start,
+                "latest_under_panels": round(float(cu_panels.iloc[-1]), 1) if len(cu_panels) else 0,
+                "latest_open_field": round(float(cu_open.iloc[-1]), 1) if len(cu_open) else 0,
+                "days_counted": len(daily_chill),
+                "daily": daily,
+            }
+            self._chill_cache.set(cache_key, result)
+            return result
+        except Exception as exc:
+            log.error("Chill units failed: %s", exc)
+            return {"error": f"Chill units failed: {exc}"}
+
 
 # ═══════════════════════════════════════════════════════════════════════
 # DataHub  (service registry)