File size: 4,603 Bytes
13fc29d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 | """
Shared bootstrap helpers for the SolarWine Streamlit app.
Cached data loaders, AI/IMS health checks, and common utilities.
"""
from __future__ import annotations
import base64
import threading
import streamlit as st
_BRAND_GREEN = "#00BD3E"
_BRAND_DARK = "#1A1A1A"
try:
import plotly.express
import plotly.graph_objects
from plotly.subplots import make_subplots
_HAS_PLOTLY = True
except ImportError:
_HAS_PLOTLY = False
from pathlib import Path
import pandas as pd
# Threading state for IMS background refresh (shared with app.py via module-level vars)
_IMS_REFRESH_LOCK = threading.Lock()
_IMS_REFRESH_STARTED = False
_IMS_REFRESH_STATUS: dict = {"ok": None, "error": None, "running": False}
@st.cache_data
def load_labels(path: str) -> pd.DataFrame:
return pd.read_csv(path, index_col=0, parse_dates=True)
@st.cache_data
def load_metrics(path: str) -> pd.DataFrame:
return pd.read_csv(path)
def _check_ai_health_impl() -> dict:
"""Actual Gemini ping; cached so we only run once."""
from src.genai_utils import get_genai_client
try:
client = get_genai_client()
except Exception as exc: # noqa: BLE001
return {"ok": False, "error": f"client init failed: {exc}"}
try:
resp = client.models.generate_content(
model="gemini-2.5-flash",
contents="ping",
)
text = getattr(resp, "text", "") or ""
return {"ok": bool(text.strip()), "error": None}
except Exception as exc: # noqa: BLE001
return {"ok": False, "error": f"generate_content failed: {exc}"}
@st.cache_resource(show_spinner=False)
def _cached_ai_ping() -> dict:
"""Cached Gemini health check (used by check_ai_health with timeout)."""
return _check_ai_health_impl()
def check_ai_health(timeout_seconds: float = 1.5) -> dict:
"""
Lightweight startup validation that Gemini is reachable.
Waits at most timeout_seconds so app load stays under ~5s.
Returns a dict with:
- ok: bool
- error: str | None
"""
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
with ThreadPoolExecutor(max_workers=1) as pool:
future = pool.submit(_cached_ai_ping)
try:
return future.result(timeout=timeout_seconds)
except FuturesTimeoutError:
return {"ok": False, "error": "Health check timed out. Try refreshing."}
def _run_ims_refresh(days: int = 7) -> None:
"""Background worker to refresh IMS cache without blocking the UI."""
from datetime import date, timedelta
from src.ims_client import IMSClient
global _IMS_REFRESH_STATUS
try:
ims = IMSClient()
except Exception as exc: # noqa: BLE001
_IMS_REFRESH_STATUS = {
"ok": False,
"error": f"IMS init failed: {exc}",
"running": False,
}
return
try:
today = date.today()
start = (today - timedelta(days=days)).strftime("%Y-%m-%d")
end = today.strftime("%Y-%m-%d")
df = ims.fetch_and_cache(start, end)
if df.empty:
_IMS_REFRESH_STATUS = {
"ok": False,
"error": "IMS returned no data for the requested range.",
"running": False,
}
else:
_IMS_REFRESH_STATUS = {"ok": True, "error": None, "running": False}
except Exception as exc: # noqa: BLE001
_IMS_REFRESH_STATUS = {
"ok": False,
"error": f"IMS fetch_and_cache failed: {exc}",
"running": False,
}
def ensure_ims_refresh_background(days: int = 7) -> dict:
"""
Kick off IMS cache refresh in a background thread if not already started.
Returns the latest known status dict without blocking page load.
"""
global _IMS_REFRESH_STARTED, _IMS_REFRESH_STATUS
with _IMS_REFRESH_LOCK:
if not _IMS_REFRESH_STARTED:
_IMS_REFRESH_STARTED = True
_IMS_REFRESH_STATUS = {"ok": None, "error": None, "running": True}
t = threading.Thread(target=_run_ims_refresh, args=(days,), daemon=True)
t.start()
return dict(_IMS_REFRESH_STATUS)
def img_to_base64(path: Path) -> str:
return base64.b64encode(path.read_bytes()).decode()
def now_israel():
from datetime import datetime
try:
from zoneinfo import ZoneInfo
tz = ZoneInfo("Asia/Jerusalem")
except ImportError:
import datetime as _dt
tz = _dt.timezone(_dt.timedelta(hours=2)) # fallback UTC+2
return datetime.now(tz)
|