| """ |
| 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 |
|
|
| |
| _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: |
| 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: |
| 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: |
| _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: |
| _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)) |
| return datetime.now(tz) |
|
|
