"""
Vineyard Advisor tab — farmer-friendly dashboard + AI chatbot.
Loading strategy:
- Static sections render instantly (vine stage, shading text)
- Group A (IMS-only, ~200ms): weather, forecast, last rain
- Group B (network, ~1-2s): TB sensors, irrigation, energy — fills in async
Sections:
1. Vine stage — phenological stage + countdown to next
2. Weather forecast — week-ahead daily highs, rain/wind events
3. Panel shading — current conditions assessment
4. Tomorrow's plan — day-ahead DP trajectory with explanation
5. Energy generation — yesterday / today / tomorrow kWh
6. Irrigation — last water, soil status, when next needed
7. Chatbot — AI advisor
"""
from __future__ import annotations
import logging
import math
import threading
from datetime import date as _date_cls, timedelta as _td
import pandas as pd
import streamlit as st
from ui.bootstrap import ensure_ims_refresh_background, now_israel as _now_israel
def _get_plotly():
"""Lazy-load plotly only when a chart is actually rendered."""
try:
import plotly.express as px
return px
except ImportError:
return None
# ---------------------------------------------------------------------------
# Helpers — farmer-friendly formatting
# ---------------------------------------------------------------------------
def _temp_word(t: float) -> str:
if t >= 40: return "extreme heat"
if t >= 35: return "very hot"
if t >= 30: return "hot"
if t >= 25: return "warm"
if t >= 15: return "mild"
if t >= 5: return "cool"
return "cold"
def _soil_word(pct: float) -> str:
if pct < 12: return "very dry"
if pct < 18: return "dry"
if pct < 25: return "adequate"
if pct < 35: return "moist"
return "wet"
def _hours_ago_str(h: float) -> str:
if h < 1: return "less than 1 hour ago"
if h < 24: return f"{h:.0f} hours ago"
days = h / 24
if days < 2: return "yesterday"
return f"{days:.0f} days ago"
# ---------------------------------------------------------------------------
# Shared DataHub
# ---------------------------------------------------------------------------
def _vs_hub():
if "vs_hub" not in st.session_state:
from src.data_providers import DataHub
st.session_state["vs_hub"] = DataHub.default()
return st.session_state["vs_hub"]
# ---------------------------------------------------------------------------
# Shading gate assessment (uses same logic as control loop)
# ---------------------------------------------------------------------------
def _run_intervention_gate(
stage_id: str,
temp_c: float,
ghi: float | None,
hour: int,
) -> dict:
"""Run the InterventionGate logic and return a farmer-friendly result.
Returns dict with 'status' ("dormant", "blocked", "passed") and 'reason'.
Uses the same thresholds as the real-time control loop.
"""
from config.settings import (
NO_SHADE_BEFORE_HOUR,
SEMILLON_TRANSITION_TEMP_C,
SHADE_ELIGIBLE_CWSI_ABOVE,
SHADE_ELIGIBLE_GHI_ABOVE,
)
# Dormant — no canopy, shading is irrelevant (not harmful, just pointless)
if stage_id == "winter_dormancy":
return {"status": "dormant", "reason": "vine dormant"}
# Before allowed hour — morning light is critical for carbon fixation
if hour < NO_SHADE_BEFORE_HOUR:
return {
"status": "blocked",
"reason": (
f"morning light (before {NO_SHADE_BEFORE_HOUR}:00) is critical "
f"for carbon fixation. No shading before {NO_SHADE_BEFORE_HOUR}:00"
),
}
# Night / overcast — no meaningful sun
if ghi is not None and ghi < 100:
return {
"status": "blocked",
"reason": f"low radiation ({ghi:.0f} W/m²) — no meaningful sunlight to shade from",
}
# Below Rubisco transition — vine is light-limited (RuBP regime)
if temp_c < SEMILLON_TRANSITION_TEMP_C:
return {
"status": "blocked",
"reason": (
f"at {temp_c:.0f}°C the vine is light-limited (below {SEMILLON_TRANSITION_TEMP_C:.0f}°C "
f"Rubisco transition). Every photon helps — shading would reduce photosynthesis"
),
}
# GHI below stress threshold — not enough radiation to cause heat damage
if ghi is not None and ghi < SHADE_ELIGIBLE_GHI_ABOVE:
return {
"status": "blocked",
"reason": (
f"radiation ({ghi:.0f} W/m²) is below stress threshold "
f"({SHADE_ELIGIBLE_GHI_ABOVE:.0f} W/m²). Panels at full tracking"
),
}
# Heat stress proxy from air temperature (not true CWSI — no leaf temp available here)
heat_stress = max(0.0, min(1.0, (temp_c - 30.0) / 10.0))
if heat_stress < SHADE_ELIGIBLE_CWSI_ABOVE:
return {
"status": "blocked",
"reason": (
f"at {temp_c:.0f}°C, heat stress is moderate (index {heat_stress:.2f} "
f"< {SHADE_ELIGIBLE_CWSI_ABOVE:.2f}). Panels at full tracking"
),
}
# All conditions met — gate passes
return {
"status": "passed",
"reason": (
f"temperature {temp_c:.0f}°C above Rubisco transition, "
f"radiation {ghi:.0f} W/m², heat stress index {heat_stress:.2f}"
),
}
# ---------------------------------------------------------------------------
# Tomorrow's plan computation
# ---------------------------------------------------------------------------
def _compute_tomorrow_plan(tomorrow: _date_cls) -> dict:
"""Run DayAheadPlanner on tomorrow using IMS data for forecast proxy.
Uses last available day's temperature/GHI profile as a naive forecast
(persistence forecast). Returns dict with 'plan' key or 'error'.
"""
try:
hub = _vs_hub()
ims_df = hub.weather.get_dataframe()
if ims_df.empty:
return {"error": "No IMS data available."}
df = ims_df.copy()
if "timestamp_utc" in df.columns:
df["timestamp_utc"] = pd.to_datetime(df["timestamp_utc"], utc=True)
df = df.set_index("timestamp_utc")
# Use last full day as forecast proxy (persistence forecast)
last_day = df.index.max().normalize()
day_data = df[df.index.normalize() == last_day]
if len(day_data) < 10:
# Try day before
last_day = last_day - pd.Timedelta(days=1)
day_data = df[df.index.normalize() == last_day]
if len(day_data) < 10:
return {"error": "Not enough recent data for forecast."}
# Build 96-slot arrays from the proxy day
temps = [25.0] * 96
ghis = [0.0] * 96
for _, row in day_data.iterrows():
slot = row.name.hour * 4 + row.name.minute // 15
if 0 <= slot < 96:
if "air_temperature_c" in day_data.columns:
t = row.get("air_temperature_c")
if pd.notna(t):
temps[slot] = float(t)
if "ghi_w_m2" in day_data.columns:
g = row.get("ghi_w_m2")
if pd.notna(g):
ghis[slot] = float(g)
# Compute daily budget (simplified: use typical daily allocation)
from src.energy_budget import EnergyBudgetPlanner
planner_budget = EnergyBudgetPlanner()
annual = planner_budget.compute_annual_plan(tomorrow.year)
month_budget = annual["monthly_budgets"].get(tomorrow.month, 0.5)
daily_budget = month_budget / 30.0 # approximate daily share
# Run the day-ahead planner
from src.day_ahead_planner import DayAheadPlanner
planner = DayAheadPlanner()
plan = planner.plan_day(
target_date=tomorrow,
forecast_temps=temps,
forecast_ghi=ghis,
daily_budget_kwh=max(daily_budget, 0.1),
)
return {"plan": plan}
except Exception as exc:
return {"error": str(exc)}
# ---------------------------------------------------------------------------
# Group A: Fast data (IMS CSV only, ~200ms, no network)
# ---------------------------------------------------------------------------
def _fetch_group_a(today: _date_cls) -> dict:
"""Weather current + week view + last rain. All from cached IMS CSV."""
hub = _vs_hub()
ims_df = hub.weather.get_dataframe() # disk read (~100ms first time, cached after)
result: dict = {}
# Current weather
result["weather"] = hub.weather.get_current()
# Week view (last 7 days as proxy)
try:
if ims_df.empty:
result["week_forecast"] = {"error": "No IMS data."}
else:
df = ims_df
if "timestamp_utc" in df.columns:
df = df.copy()
df["timestamp_utc"] = pd.to_datetime(df["timestamp_utc"], utc=True)
df = df.set_index("timestamp_utc")
end = df.index.max()
start = end - pd.Timedelta(days=7)
week = df.loc[start:end]
if week.empty:
result["week_forecast"] = {"error": "Not enough data."}
else:
temp_col = next((c for c in week.columns if "temp" in c.lower() and "dew" not in c.lower() and "soil" not in c.lower()), None)
rain_col = next((c for c in week.columns if "rain" in c.lower()), None)
wind_col = next((c for c in week.columns if "wind" in c.lower() and "angle" not in c.lower()), None)
days_out = []
for day_date, day_data in week.resample("1D"):
if day_data.empty:
continue
d = {"date": day_date.strftime("%a %d/%m")}
if temp_col:
d["high"] = round(float(day_data[temp_col].max()), 1)
d["low"] = round(float(day_data[temp_col].min()), 1)
d["label"] = _temp_word(d["high"])
if rain_col:
total_rain = float(day_data[rain_col].sum())
if total_rain > 0.5:
d["rain_mm"] = round(total_rain, 1)
if wind_col:
max_wind = float(day_data[wind_col].max())
if max_wind > 8:
d["wind_ms"] = round(max_wind, 1)
days_out.append(d)
result["week_forecast"] = {"days": days_out, "note": "Based on recent IMS measurements (last 7 days)."}
except Exception as exc:
result["week_forecast"] = {"error": str(exc)}
# Last rain
try:
if not ims_df.empty:
df = ims_df
if "timestamp_utc" in df.columns:
df = df.copy()
df["timestamp_utc"] = pd.to_datetime(df["timestamp_utc"], utc=True)
df = df.set_index("timestamp_utc")
rain_col = next((c for c in df.columns if "rain" in c.lower()), None)
if rain_col:
rain = df[df[rain_col] > 0.2]
if not rain.empty:
last_ts = rain.index.max()
result["last_rain"] = {
"mm": round(float(rain.loc[last_ts, rain_col]), 1),
"hours_ago": round((pd.Timestamp.now(tz="UTC") - last_ts).total_seconds() / 3600, 1),
}
except Exception:
pass
return result
# ---------------------------------------------------------------------------
# Group B: Slow data (ThingsBoard + energy, ~1-2s network)
# ---------------------------------------------------------------------------
def _fetch_group_b(today: _date_cls) -> dict:
"""TB sensors + irrigation + energy predictions. Network-bound."""
from concurrent.futures import ThreadPoolExecutor, as_completed
hub = _vs_hub()
# Pre-load IMS DF once for all 3 energy predictions (avoids 3x CSV re-read)
ims_df = hub.weather.get_dataframe()
yesterday = str(today - _td(days=1))
today_str = str(today)
tomorrow = str(today + _td(days=1))
tasks = {
"sensors": lambda: hub.vine_sensors.get_snapshot(mode="dashboard"),
"energy_yesterday": lambda: hub.energy.predict(yesterday),
"energy_today": lambda: hub.energy.predict(today_str),
"energy_tomorrow": lambda: hub.energy.predict(tomorrow, ims_df=ims_df),
}
# Irrigation history from ThingsBoard
def _fetch_irrigation():
from datetime import datetime as _dt, timezone as _tz
from src.thingsboard_client import IRRIGATION_KEYS
try:
tb = hub.vine_sensors._client()
end = _dt.now(tz=_tz.utc)
start = end - _td(hours=48)
df = tb.get_timeseries("Irrigation1", IRRIGATION_KEYS, start, end)
if df.empty:
return {"error": "No irrigation data in last 48 hours."}
vol_col = next((c for c in df.columns if "volume" in c.lower()), None)
flow_col = next((c for c in df.columns if "flow" in c.lower()), None)
check_col = vol_col or flow_col
last_event = None
if check_col:
nonzero = df[df[check_col] > 0]
if not nonzero.empty:
last_ts = nonzero.index.max()
last_event = {
"hours_ago": round((pd.Timestamp.now(tz="UTC") - last_ts).total_seconds() / 3600, 1),
}
if vol_col and pd.notna(nonzero.loc[last_ts, vol_col]):
last_event["volume_l"] = round(float(nonzero.loc[last_ts, vol_col]), 1)
return {"last_event": last_event}
except Exception as exc:
return {"error": str(exc)}
tasks["irrigation"] = _fetch_irrigation
results = {}
with ThreadPoolExecutor(max_workers=6) as pool:
futures = {pool.submit(fn): name for name, fn in tasks.items()}
for f in as_completed(futures):
name = futures[f]
try:
results[name] = f.result()
except Exception as exc:
results[name] = {"error": str(exc)}
return results
# ---------------------------------------------------------------------------
# Main render function
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# Control System Verification section (folded)
# ---------------------------------------------------------------------------
def _render_verification_section() -> None:
"""Render the control-system verification report as a folded expander."""
import json
from pathlib import Path
report_path = Path(__file__).resolve().parent.parent / "Data" / "verification_report.json"
with st.expander("Control System Verification Report", expanded=False):
# ── Methodology explanation ──────────────────────────────
st.markdown("""
**What this report does**
The verification script (`scripts/verify_control_system.py`) compares what the
DayAheadPlanner *would have commanded* against what the trackers *actually did*
over the last 10 days. It runs 10 automated checks:
| Step | Check | Data source |
|------|-------|-------------|
| 1 | Pull actual tracker angles | ThingsBoard Tracker501-509 |
| 2 | Pull actual energy production | ThingsBoard Plant asset (Wh) |
| 3 | Pull historical weather | Air1 ambient sensor (temp, PAR) |
| 4 | Run DayAheadPlanner per day | Real weather as perfect forecast |
| 5 | Compute astronomical angles | pvlib solar position model |
| 6 | Compare planned vs actual angles | MAE, max deviation, % within 2 deg |
| 7 | Validate InterventionGate decisions | No shade before 10:00, temp/GHI thresholds |
| 8 | Verify energy budget compliance | Daily cost <= daily budget ceiling |
| 9 | Cross-validate FvCB photosynthesis | A range, shading_helps consistency |
| 10 | Generate scorecard | Per-day and overall PASS/WARN/FAIL |
**How to interpret the results**
- **Angle alignment**: the MAE between the astronomical model and the real tracker.
Deviations > 2 deg are mostly at sunrise/sunset where the model clips at +/-60 deg
but the physical tracker can reach +/-73 deg.
- **Gate compliance**: verifies that no shading was triggered when it should not
have been (before 10:00, below 30 deg C, low GHI, May).
- **Budget compliance**: the planner must never exceed the daily energy ceiling.
- **FvCB consistency**: the Farquhar model must return physically plausible A values
and the `shading_helps` flag must agree with the temperature-based transition rule.
""")
st.markdown("---")
# ── Load and display report ──────────────────────────────
if not report_path.exists():
st.info(
"No verification report found. Run the verification script to generate one:\n\n"
"```\npython scripts/verify_control_system.py --days 10 "
"--output Data/verification_report.json\n```"
)
return
try:
report = json.loads(report_path.read_text())
except Exception as exc:
st.error(f"Failed to load report: {exc}")
return
sc = report.get("scorecard", {})
checks = sc.get("checks", {})
overall = sc.get("overall", "N/A")
generated = sc.get("generated_at", "unknown")
days = sc.get("days_analyzed", 0)
# ── Overall badge ────────────────────────────────────────
_badge_colors = {"PASS": "green", "WARN": "orange", "FAIL": "red"}
_badge_emoji = {"PASS": "white_check_mark", "WARN": "warning", "FAIL": "x"}
badge_color = _badge_colors.get(overall, "gray")
badge_emoji = _badge_emoji.get(overall, "question")
st.markdown(
f"### Overall: :{badge_emoji}: **{overall}**\n\n"
f"*{days} days analyzed — generated {generated[:16].replace('T', ' ')} UTC*"
)
# ── Per-check results ────────────────────────────────────
cols = st.columns(len(checks))
for col, (name, info) in zip(cols, checks.items()):
if not isinstance(info, dict):
continue
status = info.get("status", "N/A")
color = _badge_colors.get(status, "gray")
emoji = _badge_emoji.get(status, "question")
label = name.replace("_", " ").title()
with col:
st.markdown(f"**{label}**")
st.markdown(f":{emoji}: {status}")
# Show key metric
if "mae_deg" in info:
st.caption(f"MAE {info['mae_deg']}deg, {info.get('within_tolerance_pct', 0):.0f}% within 2deg")
elif "critical" in info:
st.caption(f"{info['critical']} critical, {info['warnings']} warn")
elif "days_exceeded" in info:
st.caption(f"{info['days_exceeded']} days exceeded")
elif "issues" in info:
st.caption(f"{info['issues']} issues, {info.get('errors', 0)} errors")
# ── Per-day table ────────────────────────────────────────
plans = report.get("plans", {})
violations = report.get("violations", [])
budget_results = report.get("budget_results", [])
if plans:
st.markdown("#### Per-Day Breakdown")
rows = []
budget_lookup = {r["date"]: r for r in budget_results}
for day_str in sorted(plans.keys()):
plan = plans[day_str]
if "error" in plan:
rows.append({
"Date": day_str, "Interventions": "ERR",
"Budget Used (%)": "-", "Gate Violations": "-",
})
continue
n_interv = plan.get("n_intervention_slots", 0)
util = plan.get("budget_utilisation_pct", 0)
day_viol = sum(1 for v in violations
if v.get("date") == day_str and v.get("severity") == "CRITICAL")
rows.append({
"Date": day_str,
"Interventions": n_interv,
"Budget Used (%)": f"{util:.1f}",
"Gate Violations": day_viol,
})
st.dataframe(pd.DataFrame(rows), hide_index=True, use_container_width=True)
# ── Angle alignment detail ───────────────────────────────
comp_summary = report.get("comparison_summary", {})
angle_info = checks.get("angle_alignment", {})
if angle_info:
st.markdown("#### Angle Alignment Detail")
st.markdown(
f"Compared **{comp_summary.get('matched_records', 0)}** planned slots "
f"against actual Tracker501 telemetry.\n\n"
f"- **MAE**: {angle_info.get('mae_deg', 'N/A')} deg\n"
f"- **Within +/-2 deg tolerance**: {angle_info.get('within_tolerance_pct', 'N/A')}%\n\n"
f"*Large deviations (>50 deg) are concentrated at sunrise/sunset slots "
f"where the pvlib model clips at +/-60 deg but the physical tracker "
f"reaches +/-73 deg. Midday alignment is typically within 1-2 deg.*"
)
# ── Re-run button ────────────────────────────────────────
st.markdown("---")
st.caption(
"To refresh this report, run from the terminal:\n\n"
"`python scripts/verify_control_system.py --days 10 "
"--output Data/verification_report.json`"
)
def render_tab_advisor() -> None:
today = _date_cls.today()
_now_vs = st.session_state.get("current_time_israel") or _now_israel()
st.header("Vineyard Advisor")
# ── Session state ─────────────────────────────────────────────────
if "chat_history" not in st.session_state:
st.session_state.chat_history = []
if "chatbot" not in st.session_state:
st.session_state.chatbot = None
if "chatbot_available" not in st.session_state:
st.session_state.chatbot_available = None
_GRP_A = "vs_grp_a"
_GRP_B = "vs_grp_b"
for k in (_GRP_A, _GRP_B):
if k not in st.session_state:
st.session_state[k] = None
# ── Pre-warm chatbot in background ────────────────────────────────
if "chatbot_warmup_started" not in st.session_state:
st.session_state.chatbot_warmup_started = True
def _warmup():
try:
from src.vineyard_chatbot import VineyardChatbot
bot = VineyardChatbot(hub=_vs_hub())
st.session_state.chatbot = bot
st.session_state.chatbot_available = True
# Try to warm up Gemini client (may fail without API key — that's OK)
if bot.has_api_key:
_ = bot.client
except Exception:
pass
threading.Thread(target=_warmup, daemon=True).start()
# ══════════════════════════════════════════════════════════════════
# 1. VINE STAGE (instant — no IO)
# ══════════════════════════════════════════════════════════════════
from src.phenology import estimate_stage_for_date, next_stage_for_date
stage = estimate_stage_for_date(today)
next_label, _, days_until = next_stage_for_date(today)
# ── Live vineyard camera + vine stage side by side ────────────
_cam_col, _stage_col = st.columns([1.2, 1])
with _cam_col:
import streamlit.components.v1 as _components
_components.html(
"""
""",
height=300,
)
st.caption("Live picture from our site in Yeruham. Vine: Chenin Blanc. Energy capacity: 100 kW.")
with _stage_col:
st.subheader(f"Vine stage: {stage.label}")
st.markdown(stage.description)
if days_until > 0:
st.caption(f"Next stage: **{next_label}** in ~{days_until} days")
st.caption(f"{_now_vs.strftime('%A %d %B %Y, %H:%M')} Israel time")
st.divider()
# ══════════════════════════════════════════════════════════════════
# GROUP A: Weather + Forecast (fast, ~200ms from cached CSV)
# ══════════════════════════════════════════════════════════════════
if st.session_state[_GRP_A] is None:
st.session_state[_GRP_A] = _fetch_group_a(today)
a = st.session_state[_GRP_A]
_wx = a.get("weather", {})
# Auto-refresh stale IMS data (>60 min) — at most once per session
_wx_age = _wx.get("age_minutes")
if _wx_age is not None and _wx_age > 60:
_ims_status = ensure_ims_refresh_background(days=3)
if _ims_status.get("running"):
st.info(f"Weather data is {_wx_age:.0f} min old — refreshing...")
elif _ims_status.get("ok") and not st.session_state.get("vs_ims_refreshed"):
st.session_state["vs_ims_refreshed"] = True
_vs_hub().weather._df_cache.invalidate("ims")
st.session_state[_GRP_A] = None
st.session_state[_GRP_B] = None
st.rerun()
elif _wx_age > 60:
st.caption(f"Weather data is {_wx_age:.0f} min old (IMS refresh attempted).")
# ── 2. WEATHER FORECAST ───────────────────────────────────────────
st.subheader("Weather — recent week")
_forecast = a.get("week_forecast", {})
if "error" in _forecast:
st.caption(f"Weather: {_forecast['error']}")
else:
days_list = _forecast.get("days", [])
if days_list:
n_cols = min(len(days_list), 7)
cols = st.columns(n_cols)
for i, day in enumerate(days_list[:7]):
with cols[i]:
st.markdown(f"**{day['date']}**")
high = day.get("high")
low = day.get("low")
if high is not None:
st.metric("High", f"{high:.0f} C", label_visibility="collapsed")
st.caption(f"{low:.0f}–{high:.0f} C ({day.get('label', '')})")
events = []
if day.get("rain_mm"):
events.append(f"Rain {day['rain_mm']} mm")
if day.get("wind_ms"):
events.append(f"Wind {day['wind_ms']:.0f} m/s")
if events:
st.caption(" | ".join(events))
if _forecast.get("note"):
st.caption(_forecast["note"])
# Current conditions — compact one-liner
# Extract once — used later in irrigation VPD calc
_rh = _wx.get("rh_percent") if "error" not in _wx else None
if "error" not in _wx:
_t = _wx.get("air_temperature_c")
_ws = _wx.get("wind_speed_ms")
parts = []
if _t is not None:
parts.append(f"**{_t:.0f} C** ({_temp_word(_t)})")
if _rh is not None:
parts.append(f"Humidity {_rh:.0f}%")
if _ws is not None:
parts.append(f"Wind {_ws:.1f} m/s")
age_str = f" — {_wx_age:.0f} min ago" if _wx_age else ""
if parts:
st.markdown("Now: " + " | ".join(parts) + age_str)
st.divider()
# ══════════════════════════════════════════════════════════════════
# 5. SHADING (instant — uses InterventionGate, same logic as control loop)
# ══════════════════════════════════════════════════════════════════
st.subheader("Panel shading")
_shading_assessed = False
if "error" not in _wx:
_t_now = _wx.get("air_temperature_c")
_ghi_now = _wx.get("ghi_w_m2")
_rh_now = _wx.get("rh_percent")
_wind_now = _wx.get("wind_speed_ms")
if _t_now is not None:
_shading_assessed = True
_gate_result = _run_intervention_gate(
stage_id=stage.id, temp_c=_t_now, ghi=_ghi_now, hour=_now_israel().hour,
)
if _gate_result["status"] == "dormant":
st.markdown(
"Panels at **full tracking** — maximum energy generation. "
"The vine is dormant (no leaves, no canopy) so panel position "
"has no effect on the plant. All energy goes to the grid."
)
elif _gate_result["status"] == "blocked":
reason = _gate_result["reason"]
st.markdown(f"Panels at **full tracking** — {reason}")
else:
# Gate passed — stress conditions met
st.markdown(
f"**Stress conditions detected** — "
f"{_gate_result['reason']}. "
f"The system may activate minimum-dose shading to protect "
f"the fruiting zone while preserving top-canopy productivity."
)
# Wind stow check
if _wind_now is not None:
from src.command_arbiter import CommandArbiter
wind_override = CommandArbiter.check_wind_stow(_wind_now)
if wind_override:
st.warning(
f"Wind stow active: {_wind_now:.1f} m/s — "
f"panels flat (0°) for structural safety. "
f"Overrides all other commands."
)
# Detailed gate explanation
with st.expander("How shading decisions work"):
st.markdown(
"Panels **always default to full astronomical tracking** "
"(following the sun for maximum energy). Shading is a rare, "
"minimum-dose intervention that activates only when **all** "
"of these conditions are true simultaneously:\n\n"
f"| Condition | Threshold | Current |\n"
f"|---|---|---|\n"
f"| Sunlight (GHI) | ≥ 400 W/m² | {_ghi_now:.0f} W/m² {'✓' if _ghi_now and _ghi_now >= 400 else '✗'} |\n"
f"| Leaf temperature | ≥ 30°C (Rubisco transition) | {_t_now:.0f}°C {'✓' if _t_now >= 30 else '✗'} |\n"
f"| Water stress (CWSI) | ≥ 0.4 | estimated from temp |\n"
f"| Morning light | After 10:00 | {'✓' if _now_israel().hour >= 10 else '✗'} |\n"
f"| Biology (FvCB model) | Shading helps photosynthesis | {'✓' if _t_now >= 30 and (_ghi_now or 0) >= 400 else '✗'} |\n\n"
"**Why?** Below 30°C, the vine is RuBP-limited — light is the "
"bottleneck, so reducing light (shading) hurts photosynthesis. "
"Above 30°C, the vine becomes Rubisco-limited — the enzyme is the "
"bottleneck, and reducing excess light can actually help by "
"lowering heat load on the fruiting zone."
)
if not _shading_assessed:
st.caption("No weather data available — cannot assess shading conditions.")
st.divider()
# ══════════════════════════════════════════════════════════════════
# 5b. TOMORROW'S PLAN (computed from IMS forecast + DayAheadPlanner)
# ══════════════════════════════════════════════════════════════════
st.subheader("Tomorrow's plan")
_tomorrow = today + _td(days=1)
_PLAN_KEY = "vs_tomorrow_plan"
try:
if _PLAN_KEY not in st.session_state:
st.session_state[_PLAN_KEY] = _compute_tomorrow_plan(_tomorrow)
_plan = st.session_state[_PLAN_KEY]
if _plan and "error" not in _plan:
plan_obj = _plan["plan"]
# Summary sentence
if plan_obj.n_intervention_slots == 0:
st.markdown(
f"**Full tracking all day** — panels follow the sun for maximum energy. "
f"No heat stress expected (forecast peak below shading threshold)."
)
else:
# Find intervention time window
int_slots = [s for s in plan_obj.slots if s.offset_deg > 0]
first_time = int_slots[0].time
last_time = int_slots[-1].time
avg_offset = sum(s.offset_deg for s in int_slots) / len(int_slots)
st.markdown(
f"**{plan_obj.n_intervention_slots} slots with shade** "
f"({first_time}–{last_time} UTC), "
f"average {avg_offset:.0f}° offset from tracking. "
f"Energy cost: {plan_obj.total_energy_cost_kwh:.3f} kWh "
f"({plan_obj.budget_utilisation_pct:.1f}% of daily budget)."
)
# Why this plan?
with st.expander("Why this plan?"):
st.markdown(
"The day-ahead planner uses dynamic programming to find the "
"optimal panel tilt trajectory for tomorrow. It balances two goals:\n\n"
"1. **Maximise energy** — panels track the sun (default)\n"
"2. **Protect the vine** — tilt panels to shade the fruiting zone "
"when heat stress is forecast\n\n"
"The planner only activates shading when **all** conditions are met:\n"
f"- Air temperature ≥ 30°C (Rubisco transition)\n"
f"- Solar radiation ≥ 400 W/m² (meaningful heat load)\n"
f"- Water stress indicator (CWSI) above threshold\n"
f"- Before 10:00 — no shading (morning light is critical)\n\n"
f"**Vine stage:** {plan_obj.stage_id.replace('_', ' ').title()}\n\n"
f"**Budget:** {plan_obj.daily_budget_kwh:.2f} kWh available "
f"(from hierarchical Year → Month → Week → Day allocation)"
)
# Slot-level detail
if plan_obj.slots:
slot_data = []
for s in plan_obj.slots:
if s.gate_passed or s.offset_deg > 0:
slot_data.append({
"Time (UTC)": s.time,
"Offset": f"{s.offset_deg:.0f}°",
"Energy cost": f"{s.energy_cost_kwh:.4f} kWh",
"Status": "Shade" if s.offset_deg > 0 else "Track",
})
if slot_data:
st.dataframe(pd.DataFrame(slot_data), hide_index=True)
else:
st.caption("All slots at full tracking — no interventions needed.")
else:
_err = _plan.get("error", "Could not compute plan.") if _plan else "No IMS data for planning."
st.caption(f"Plan unavailable: {_err}")
except Exception as _exc:
st.caption(f"Plan unavailable: {_exc}")
st.divider()
# ══════════════════════════════════════════════════════════════════
# GROUP B: Energy + Irrigation + Sensors (slow, ~1-2s network)
# ══════════════════════════════════════════════════════════════════
if st.session_state[_GRP_B] is None:
with st.spinner("Loading energy & irrigation data..."):
st.session_state[_GRP_B] = _fetch_group_b(today)
b = st.session_state[_GRP_B]
_sns = b.get("sensors", {})
# ── 3. ENERGY GENERATION ──────────────────────────────────────────
st.subheader("Energy generation")
_en_y = b.get("energy_yesterday", {})
_en_t = b.get("energy_today", {})
_en_m = b.get("energy_tomorrow", {})
en_cols = st.columns(3)
for col, label, data in [
(en_cols[0], "Yesterday", _en_y),
(en_cols[1], "Today (so far)", _en_t),
(en_cols[2], "Tomorrow (est.)", _en_m),
]:
with col:
st.markdown(f"**{label}**")
kwh = data.get("daily_kwh")
if kwh is None:
st.caption(data.get("error", "No data"))
else:
peak = data.get("peak_hour", "?")
st.metric("kWh", f"{kwh:.1f}")
st.caption(f"Peak at {peak}:00")
px = _get_plotly()
if px is not None:
chart_data = []
for label, data in [("Yesterday", _en_y), ("Today", _en_t), ("Tomorrow", _en_m)]:
kwh = data.get("daily_kwh")
if kwh is not None:
chart_data.append({"Day": label, "kWh": kwh})
if chart_data:
fig = px.bar(pd.DataFrame(chart_data), x="Day", y="kWh",
color_discrete_sequence=["#FF9800"])
fig.update_layout(height=200, margin=dict(t=10, b=10, l=40, r=10), showlegend=False)
st.plotly_chart(fig, use_container_width=True)
# Source attribution
src_y = _en_y.get("source", "")
src_t = _en_t.get("source", "")
if "ThingsBoard" in src_t:
st.caption("Source: live inverter data (ThingsBoard).")
else:
st.caption("Source: analytical estimate (48 kW system).")
st.caption("No energy was lost to shading today — panels at full tracking.")
st.divider()
# ── 4. IRRIGATION ─────────────────────────────────────────────────
st.subheader("Irrigation")
irr_cols = st.columns(3)
with irr_cols[0]:
st.markdown("**Last water**")
_irr = b.get("irrigation", {})
_last_rain = a.get("last_rain")
_irr_event = _irr.get("last_event") if isinstance(_irr, dict) and "error" not in _irr else None
if _irr_event:
ago = _irr_event.get("hours_ago", 0)
vol = _irr_event.get("volume_l")
vol_str = f" ({vol:.0f} L)" if vol else ""
st.markdown(f"Irrigated **{_hours_ago_str(ago)}**{vol_str}")
else:
st.caption("No irrigation in last 48 hours")
if _last_rain:
st.markdown(f"Rain **{_hours_ago_str(_last_rain['hours_ago'])}** ({_last_rain['mm']} mm)")
else:
st.caption("No recent rain")
with irr_cols[1]:
st.markdown("**Soil now**")
soil_m = _sns.get("treatment_soil_moisture_pct") if isinstance(_sns, dict) and "error" not in _sns else None
if soil_m is not None:
st.metric("Soil moisture", f"{soil_m:.0f}%")
st.caption(_soil_word(soil_m))
else:
st.caption("No soil data")
if _t_now is not None and _rh is not None:
es = 0.6108 * math.exp(17.27 * _t_now / (_t_now + 237.3))
vpd = max(es * (1 - _rh / 100), 0.0)
if vpd > 3.0:
st.caption("Air is **very dry** — high evaporation")
elif vpd > 2.0:
st.caption("Air is dry — moderate evaporation")
with irr_cols[2]:
st.markdown("**When next?**")
if soil_m is not None:
tomorrow_high = None
forecast_days = _forecast.get("days", []) if isinstance(_forecast, dict) else []
if forecast_days:
tomorrow_high = forecast_days[-1].get("high")
if soil_m < 12:
st.markdown("**Now** — soil is very dry")
elif soil_m < 18:
if tomorrow_high and tomorrow_high > 33:
st.markdown("**Tomorrow morning** — soil drying + hot weather ahead")
else:
st.markdown("**1-2 days** — soil getting dry")
elif soil_m < 25:
if tomorrow_high and tomorrow_high > 36:
st.markdown("**Tomorrow** possible — extreme heat forecast")
else:
st.markdown("**Not yet** — soil is fine")
else:
st.markdown("**Not needed** — soil is well-watered")
else:
st.caption("Cannot estimate — no soil data")
st.divider()
# ── REFRESH ───────────────────────────────────────────────────────
if st.button("Refresh all data", key="vs_refresh_all"):
st.session_state[_GRP_A] = None
st.session_state[_GRP_B] = None
st.session_state.pop("vs_ims_refreshed", None)
st.rerun()
st.divider()
# ══════════════════════════════════════════════════════════════════
# 6. CHATBOT
# ══════════════════════════════════════════════════════════════════
st.subheader("Ask the advisor")
st.markdown(
"Ask about shading decisions, vine health, weather, or irrigation. "
"The advisor uses live sensor data and models."
)
_qa_cols = st.columns(4)
with _qa_cols[0]:
if st.button("Should I shade?", key="qa_shade_now"):
st.session_state.chat_history.append(
{"role": "user", "content": "Should I shade right now? Check the current weather and tell me."})
with _qa_cols[1]:
if st.button("Tomorrow's outlook", key="qa_forecast"):
st.session_state.chat_history.append(
{"role": "user", "content": "What does tomorrow look like for the vineyard? Should I prepare for anything?"})
with _qa_cols[2]:
if st.button("Irrigation advice", key="qa_irrigation"):
st.session_state.chat_history.append(
{"role": "user", "content": "Check the soil moisture and weather. When should I irrigate next?"})
with _qa_cols[3]:
if st.button("Clear chat", key="qa_clear"):
st.session_state.chat_history = []
st.rerun()
st.divider()
# Lazy-init chatbot
def _ensure_chatbot():
if "chatbot" not in st.session_state:
st.session_state.chatbot = None
if "chatbot_available" not in st.session_state:
st.session_state.chatbot_available = None
if st.session_state.chatbot is not None:
return st.session_state.chatbot
try:
from src.vineyard_chatbot import VineyardChatbot
st.session_state.chatbot = VineyardChatbot(hub=_vs_hub())
st.session_state.chatbot_available = True
return st.session_state.chatbot
except Exception:
st.session_state.chatbot = None
st.session_state.chatbot_available = False
return None
def _render_tool_data(data: dict) -> None:
"""Farmer-friendly rendering — no 3D scenes, DLI, or technical charts."""
if not data or "error" in data:
return
# Energy prediction — simple metric
if "daily_kwh" in data and data["daily_kwh"] is not None:
st.metric("Daily energy", f"{data['daily_kwh']:.1f} kWh")
return
# Stress advisory summary
if "stress_profile" in data and data["stress_profile"].get("summary"):
st.info(data["stress_profile"]["summary"])
return
# Shading comparison — plain numbers, no technical labels
if "A_unshaded" in data and "A_shaded" in data:
change = data.get("A_change_pct", 0)
if change < -5:
st.warning(f"Shading would reduce vine activity by {abs(change):.0f}%")
elif change > 5:
st.success(f"Shading would help — vine activity improves by {change:.0f}%")
else:
st.info("Shading makes little difference right now.")
return
# Schedule table (e.g. irrigation schedule)
if "schedule" in data and isinstance(data["schedule"], list):
st.dataframe(pd.DataFrame(data["schedule"]), hide_index=True)
return
# Tabular data
if "rows" in data and isinstance(data["rows"], list) and data["rows"]:
st.dataframe(pd.DataFrame(data["rows"]), hide_index=True)
return
# Skip 3D scenes, DLI charts, angle sweeps, heatmaps — not farmer-friendly
if "scene_3d_html" in data or "angles" in data:
return
# Data summary
if "summarised" in data and data.get("columns"):
rows = [{"Column": k, **v} for k, v in data["columns"].items()]
if rows:
st.dataframe(pd.DataFrame(rows), hide_index=True)
return
def _render_grounding_metadata(entry: dict, msg_index: int = -1) -> None:
"""Render confidence badge, sources, caveats, and feedback under a response."""
confidence = entry.get("confidence", "")
sources = entry.get("sources", [])
caveats = entry.get("caveats", [])
violations = entry.get("rule_violations", [])
response_mode = entry.get("response_mode", "info")
# Dual-channel mode indicator
if response_mode == "advisory":
st.caption(":orange[Advisory — recommendation with reasoning]")
elif response_mode == "info" and confidence:
st.caption(":blue[Information — factual data]")
# Confidence badge
if confidence:
badge_colors = {
"high": "green",
"medium": "orange",
"low": "red",
"insufficient_data": "red",
}
color = badge_colors.get(confidence, "gray")
label = confidence.replace("_", " ").title()
st.caption(f":{color}[Confidence: {label}]")
# Sources
if sources:
st.caption(f"Source: {', '.join(sources)}")
# Caveats
for caveat in caveats:
st.caption(f"Note: {caveat}")
# Rule violations (warnings only — blocks replace the message)
for v in violations:
if v.get("severity") == "warn":
st.caption(f"Rule check ({v.get('rule', '?')}): {v.get('message', '')}")
# Feedback buttons
if msg_index >= 0:
_render_feedback_buttons(entry, msg_index)
def _render_feedback_buttons(entry: dict, msg_index: int) -> None:
"""Render thumbs up/down and flag buttons for a response."""
fb_key = f"fb_{msg_index}"
# Check if already rated
if entry.get("_feedback"):
fb = entry["_feedback"]
label = {"thumbs_up": "Helpful", "thumbs_down": "Not helpful", "flag_incorrect": "Flagged"}
st.caption(f"Feedback: {label.get(fb, fb)}")
return
cols = st.columns([1, 1, 1, 7])
with cols[0]:
if st.button("👍", key=f"{fb_key}_up", help="Helpful"):
_submit_feedback(entry, msg_index, "thumbs_up")
with cols[1]:
if st.button("👎", key=f"{fb_key}_down", help="Not helpful"):
_submit_feedback(entry, msg_index, "thumbs_down")
with cols[2]:
if st.button("🚩", key=f"{fb_key}_flag", help="Flag as incorrect"):
_submit_feedback(entry, msg_index, "flag_incorrect")
def _submit_feedback(entry: dict, msg_index: int, feedback_type: str) -> None:
"""Log feedback and mark the message as rated."""
try:
from src.chatbot.feedback import log_feedback
# Find the preceding user message
query = ""
if msg_index > 0:
for i in range(msg_index - 1, -1, -1):
prev = st.session_state.chat_history[i]
if prev.get("role") == "user":
query = prev.get("content", "")
break
log_feedback(
query=query,
response=entry.get("content", ""),
feedback=feedback_type,
confidence=entry.get("confidence", ""),
sources=entry.get("sources", []),
tool_calls=entry.get("tool_calls", []),
rule_violations=entry.get("rule_violations", []),
response_mode=entry.get("response_mode", ""),
)
except Exception as exc:
logging.getLogger(__name__).warning("Failed to log feedback: %s", exc)
# Mark as rated in session state
st.session_state.chat_history[msg_index]["_feedback"] = feedback_type
st.rerun()
# Display chat history
for _msg_idx, entry in enumerate(st.session_state.chat_history):
role = entry.get("role", "user")
with st.chat_message(role):
st.markdown(entry.get("content", ""))
if role == "assistant":
if entry.get("data"):
_render_tool_data(entry["data"])
_render_grounding_metadata(entry, msg_index=_msg_idx)
# Process pending message
_needs_response = (
st.session_state.chat_history
and st.session_state.chat_history[-1].get("role") == "user"
)
if _needs_response:
last_msg = st.session_state.chat_history[-1]["content"]
_hist_for_bot = st.session_state.chat_history[:-1]
with st.chat_message("assistant"):
import time as _time
t0 = _time.perf_counter()
spinner_msg = "Loading advisor..." if st.session_state.chatbot is None else "Thinking..."
try:
with st.spinner(spinner_msg):
chatbot = _ensure_chatbot()
resp = chatbot.chat(last_msg, history=_hist_for_bot) if chatbot else None
except Exception as _chat_exc:
resp = None
st.error(f"Advisor error: {_chat_exc}")
elapsed = _time.perf_counter() - t0
st.caption(f"{elapsed:.1f}s")
if resp is None:
st.markdown("Could not load the advisor. Check that a Google API key is set, or ask about vine biology rules.")
st.session_state.chat_history.append({
"role": "assistant",
"content": "Could not load the advisor. Check that a Google API key is set, or ask about vine biology rules.",
"data": {}, "tool_calls": [],
"confidence": "insufficient_data", "sources": [], "caveats": [],
})
else:
st.markdown(resp.message)
if resp.data:
_render_tool_data(resp.data)
_render_grounding_metadata({
"confidence": resp.confidence,
"sources": resp.sources,
"caveats": resp.caveats,
"rule_violations": resp.rule_violations,
})
st.session_state.chat_history.append({
"role": "assistant", "content": resp.message,
"data": resp.data if resp.data else {},
"tool_calls": resp.tool_calls,
"confidence": resp.confidence,
"sources": resp.sources,
"caveats": resp.caveats,
"rule_violations": resp.rule_violations,
"response_mode": getattr(resp, "response_mode", "info"),
})
st.rerun()
_user_input = st.chat_input("Ask the vineyard advisor...")
if _user_input:
st.session_state.chat_history.append({"role": "user", "content": _user_input})
st.rerun()
# ══════════════════════════════════════════════════════════════════
# CONTROL SYSTEM VERIFICATION (folded section at bottom)
# ══════════════════════════════════════════════════════════════════
st.divider()
_render_verification_section()