""" Kirana AI — UI renderer. Loads Jinja2 templates and builds the per-page context from database state. """ import datetime from collections import defaultdict from pathlib import Path from jinja2 import Environment, FileSystemLoader, select_autoescape import os import kirana_db as db from seasonal_calendar import ( get_upcoming_festivals, get_seasonal_context, build_seasonal_summary, ) def is_model_available() -> bool: return bool(os.getenv("MODAL_RECEIPT_ENDPOINT", "").strip()) # ────────────────────────────────────────────────────────────────────────────── TEMPLATES_DIR = Path(__file__).parent / "templates" env = Environment( loader=FileSystemLoader(str(TEMPLATES_DIR)), autoescape=select_autoescape(["html"]), trim_blocks=True, lstrip_blocks=True, ) # ────────────────────────────────────────────────────────────────────────────── # SVG icon library (1.6px stroke, rounded — SF Symbols-inspired) # ────────────────────────────────────────────────────────────────────────────── def _svg(body: str, size: int = 18) -> str: return ( f'' ) ICON = { "dashboard": _svg('' '' '' ''), "inventory": _svg('' '' ''), "add": _svg('' ''), "orders": _svg('' '' ''), "analytics": _svg('' '' '' ''), "seasonal": _svg('' '' ''), "settings": _svg('' ''), "play": _svg('', 14), "refresh": _svg('', 14), "sparkles": _svg('' '', 14), "plus": _svg('', 14), "edit": _svg(''), "rupee": _svg(''), "trash": _svg('' ''), "search": _svg(''), "filter": _svg(''), "mic": _svg('' ''), "camera": _svg('' ''), "check": _svg('', 14), "x": _svg('', 14), "package": _svg('' ''), "alert": _svg(''), "clock": _svg(''), "trend": _svg(''), "trophy": _svg('' ''), "store": _svg('' '' '' ''), "info": _svg(''), "pulse": _svg(''), "shipped": _svg('' ''), "medal": _svg('' ''), "bell": _svg('' ''), "sun": _svg('' '', 16), "moon": _svg('', 16), } LOGO_SVG = ( '' ) # ────────────────────────────────────────────────────────────────────────────── # Navigation definition # ────────────────────────────────────────────────────────────────────────────── NAV_PAGES = [ {"key": "dashboard", "label": "Overview", "icon": ICON["dashboard"], "title": "Overview", "subtitle": "Real-time snapshot of your store"}, {"key": "inventory", "label": "Inventory", "icon": ICON["inventory"], "title": "Inventory", "subtitle": "Browse, search, and update your full stock"}, {"key": "add", "label": "Add Product", "icon": ICON["add"], "title": "Add Product", "subtitle": "Photo, voice, or manual entry"}, {"key": "orders", "label": "Orders", "icon": ICON["orders"], "title": "Restock Orders", "subtitle": "AI-suggested orders awaiting your approval"}, {"key": "analytics", "label": "Analytics", "icon": ICON["analytics"], "title": "Analytics", "subtitle": "Category mix and best-selling products"}, {"key": "seasonal", "label": "Seasonal", "icon": ICON["seasonal"], "title": "Seasonal Forecast", "subtitle": "Upcoming festivals and demand spikes"}, {"key": "settings", "label": "Settings", "icon": ICON["settings"], "title": "Settings", "subtitle": "Shop info, thresholds, model status"}, ] STATUS_FILTERS = ["All", "OK", "Low Stock", "Expiring", "Expired"] # ────────────────────────────────────────────────────────────────────────────── # Helpers # ────────────────────────────────────────────────────────────────────────────── def _status_of(p: dict) -> tuple[str, str, str]: """Return (label, kind, raw_key) for a product.""" today = datetime.date.today().isoformat() week = (datetime.date.today() + datetime.timedelta(days=7)).isoformat() exp = p.get("expiry_date") or "" if exp and exp < today: return "Expired", "danger", "Expired" if exp and exp <= week: return "Expiring", "warn", "Expiring" if (p.get("quantity") or 0) <= (p.get("min_stock") or 0): return "Low", "warn", "Low Stock" return "OK", "success", "OK" def _nav_with_badges() -> list[dict]: s = db.get_summary() badges = { "dashboard": s.get("low_stock", 0) or 0, "orders": len(db.get_pending_orders()), } result = [] for item in NAV_PAGES: copy = dict(item) n = badges.get(item["key"], 0) if n: copy["badge"] = n copy["badge_kind"] = "warn" result.append(copy) return result # ────────────────────────────────────────────────────────────────────────────── # Page builders # ────────────────────────────────────────────────────────────────────────────── def _build_kpis() -> list[dict]: s = db.get_summary() total = s.get("total", 0) or 0 low = s.get("low_stock", 0) or 0 expd = s.get("expired", 0) or 0 exp7 = s.get("expiring_soon", 0) or 0 val = s.get("total_value", 0) or 0 cost = s.get("cost_value", 0) or 0 margin = val - cost needs = expd + exp7 if expd and exp7: needs_foot = f"{expd} expired · {exp7} expiring 7d" elif expd: needs_foot = f"{expd} expired, clear today" elif exp7: needs_foot = f"{exp7} expiring within 7 days" else: needs_foot = "Nothing on the clock" return [ {"label": "Products", "value": total, "icon": ICON["package"], "color": "primary", "foot": "across all categories"}, {"label": "Low stock", "value": low, "icon": ICON["trend"], "color": "warn" if low else "success", "foot": f"{low} need restocking" if low else "All stocked"}, {"label": "Needs attention", "value": needs, "icon": ICON["alert"], "color": "danger" if expd else ("warn" if exp7 else "success"), "foot": needs_foot}, {"label": "Inventory value", "value": f"₹{val:,.0f}", "icon": ICON["rupee"], "color": "info", "foot": f"Margin ≈ ₹{margin:,.0f}"}, ] def _build_health() -> dict: products = db.get_all_products() today = datetime.date.today().isoformat() week = (datetime.date.today() + datetime.timedelta(days=7)).isoformat() healthy = low = expiring = expired = 0 for p in products: exp = p.get("expiry_date") or "" if exp and exp < today: expired += 1 elif exp and exp <= week: expiring += 1 elif (p["quantity"] or 0) <= (p["min_stock"] or 0): low += 1 else: healthy += 1 total = max(len(products), 1) def pct(n): return max(round((n / total) * 100, 1), 2 if n else 1) return { "total": len(products), "segments": [ {"label": "Healthy", "value": healthy, "cls": "healthy", "flex": pct(healthy)}, {"label": "Low", "value": low, "cls": "low", "flex": pct(low)}, {"label": "Expiring", "value": expiring, "cls": "expiring", "flex": pct(expiring)}, {"label": "Expired", "value": expired, "cls": "expired", "flex": pct(expired)}, ], } def _build_alerts() -> dict: low_items = db.get_low_stock() expir_items = db.get_expiring_soon(7) expd_items = db.get_expired() n_expd, n_exp7, n_low = len(expd_items), len(expir_items), len(low_items) total = n_expd + n_exp7 + n_low chips = [ {"label": "Expired", "value": n_expd, "kind": "danger"}, {"label": "Expiring 7d", "value": n_exp7, "kind": "warn"}, {"label": "Low stock", "value": n_low, "kind": "warn"}, ] # Top 3 most urgent: expired first, then soonest expiry, then lowest %-of-min top = [] for p in expd_items[:3]: top.append({ "kind": "danger", "icon": "🚨", "tag": "EXPIRED", "title": p["name"], "sub": f"{p['quantity']} {p['unit']} · Expired {p['expiry_date']}", }) if len(top) < 3: for p in expir_items[: 3 - len(top)]: days = (datetime.date.fromisoformat(p["expiry_date"]) - datetime.date.today()).days top.append({ "kind": "warn", "icon": "⏰", "tag": f"{days}d LEFT", "title": p["name"], "sub": f"{p['quantity']} {p['unit']} · Expires {p['expiry_date']}", }) if len(top) < 3: for p in low_items[: 3 - len(top)]: pct = int((p["quantity"] / max(p["min_stock"], 0.01)) * 100) top.append({ "kind": "warn", "icon": "📉", "tag": "LOW", "title": p["name"], "sub": f"{p['quantity']} {p['unit']} ({pct}% of min {p['min_stock']}) · {p['supplier'] or '—'}", }) overflow = max(total - len(top), 0) badge_cls = ("badge-danger" if expd_items else "badge-warn" if (low_items or expir_items) else "badge-success") return { "chips": chips, "top": top, "overflow": overflow, "total": total, "badge_cls": badge_cls, # back-compat alias for any template still reading `.entries` "entries": top, } def _build_festivals() -> dict: ctx = get_seasonal_context() upcoming = get_upcoming_festivals(30) cards = [{ "name": f["name"], "name_te": f["name_te"], "mult": f["demand_multiplier"], "demand_items": ", ".join(f["demand_items"][:6]), "tips": f["tips"], } for f in upcoming[:6]] return { "season": ctx["season"], "note": ctx["note"], "cards": cards, "count": len(upcoming), } def _build_pulse(days: int = 14) -> dict: """Daily revenue series for the dashboard hero sparkline. Returns a path-ready SVG polyline + area, indexed counts, peak/avg summary, and a "today vs prior-week" delta so the viz can pulse and label itself without any client-side math. """ conn = db.get_conn() rows = conn.execute( "SELECT date(sale_date) AS d, " " COALESCE(SUM(qty_sold * price_per_unit), 0) AS rev, " " COALESCE(SUM(qty_sold), 0) AS units " "FROM sales WHERE sale_date >= date('now', ?) " "GROUP BY date(sale_date)", (f"-{days - 1} days",), ).fetchall() conn.close() by_day = {r["d"]: (float(r["rev"] or 0), float(r["units"] or 0)) for r in rows} today = datetime.date.today() series: list[dict] = [] for i in range(days): d = today - datetime.timedelta(days=days - 1 - i) rev, units = by_day.get(d.isoformat(), (0.0, 0.0)) series.append({ "date": d.isoformat(), "label": d.strftime("%a"), "rev": rev, "units": units, }) revs = [pt["rev"] for pt in series] peak = max(revs) if revs else 0 total = sum(revs) avg = total / max(len(revs), 1) today_rev = revs[-1] if revs else 0 prior_week = sum(revs[:7]) # days 1..7 of the 14-day window last_week = sum(revs[7:]) # days 8..14 if prior_week > 0: delta_pct = round(((last_week - prior_week) / prior_week) * 100, 1) else: delta_pct = 100.0 if last_week else 0.0 # SVG path generation — 100-wide × 36-tall normalized viewBox W, H, PAD = 100.0, 36.0, 2.0 n = len(series) if n < 2 or peak <= 0: line_path = f"M{PAD},{H - PAD}L{W - PAD},{H - PAD}" area_path = (f"M{PAD},{H - PAD}L{W - PAD},{H - PAD}" f"L{W - PAD},{H}L{PAD},{H}Z") points = [] else: step = (W - PAD * 2) / (n - 1) coords = [] for i, pt in enumerate(series): x = PAD + step * i y = (H - PAD) - (pt["rev"] / peak) * (H - PAD * 2) coords.append((x, y)) # Smooth Catmull-Rom-ish path via Bezier def _seg(p0, p1, p2, p3): t = 0.18 c1x = p1[0] + (p2[0] - p0[0]) * t c1y = p1[1] + (p2[1] - p0[1]) * t c2x = p2[0] - (p3[0] - p1[0]) * t c2y = p2[1] - (p3[1] - p1[1]) * t return f" C{c1x:.2f},{c1y:.2f} {c2x:.2f},{c2y:.2f} {p2[0]:.2f},{p2[1]:.2f}" parts = [f"M{coords[0][0]:.2f},{coords[0][1]:.2f}"] for i in range(len(coords) - 1): p0 = coords[i - 1] if i > 0 else coords[i] p1 = coords[i] p2 = coords[i + 1] p3 = coords[i + 2] if i + 2 < len(coords) else coords[i + 1] parts.append(_seg(p0, p1, p2, p3)) line_path = "".join(parts) area_path = (line_path + f"L{coords[-1][0]:.2f},{H}L{coords[0][0]:.2f},{H}Z") points = [{ "x": round(x, 2), "y": round(y, 2), "rev": pt["rev"], "label": pt["label"], "date": pt["date"], } for (x, y), pt in zip(coords, series)] return { "series": series, "points": points, "line_path": line_path, "area_path": area_path, "viewbox": f"0 0 {int(W)} {int(H)}", "today_rev_fmt": f"₹{today_rev:,.0f}", "avg_rev_fmt": f"₹{avg:,.0f}", "peak_rev_fmt": f"₹{peak:,.0f}", "total_rev_fmt": f"₹{total:,.0f}", "delta_pct": delta_pct, "delta_kind": "up" if delta_pct > 0 else ("down" if delta_pct < 0 else "flat"), "delta_label": f"{'+' if delta_pct > 0 else ''}{delta_pct}%", "has_data": peak > 0, "days": days, } def _build_categories() -> dict: products = db.get_all_products() if not products: return {"rows": [], "total_value": 0, "total_count": 0} by_cat = defaultdict(lambda: {"count": 0, "value": 0.0}) for p in products: by_cat[p["category"]]["count"] += 1 by_cat[p["category"]]["value"] += (p["quantity"] or 0) * (p["sell_price"] or 0) rows_raw = sorted(by_cat.items(), key=lambda kv: kv[1]["value"], reverse=True) grand = sum(v["value"] for _, v in rows_raw) or 1 max_v = max((v["value"] for _, v in rows_raw), default=1) or 1 rows = [{ "name": cat, "count": v["count"], "value": v["value"], "value_fmt": f"{v['value']:,.0f}", "share": round((v["value"] / grand) * 100, 1), # % of total "bar_pct": round((v["value"] / max_v) * 100, 1), # bar width (vs leader) } for cat, v in rows_raw] return { "rows": rows, "total_value": grand, "total_value_fmt": f"{grand:,.0f}", "total_count": sum(r["count"] for r in rows), } def _build_sellers(days: int = 30) -> dict: raw = db.get_top_sellers(n=10, days=days) if not raw: return {"rows": [], "total_revenue": 0} revs = [(t.get("revenue") or 0) for t in raw] total_rev = sum(revs) or 1 max_rev = max(revs) or 1 rows = [{ "name": t["name"], "name_local": t.get("name_local") or "", "sold": f"{t['total_sold']:.1f}", "unit": t["unit"], "revenue": t.get("revenue") or 0, "revenue_fmt": f"{(t.get('revenue') or 0):,.0f}", "share": round(((t.get("revenue") or 0) / total_rev) * 100, 1), "bar_pct": round(((t.get("revenue") or 0) / max_rev) * 100, 1), } for t in raw] return {"rows": rows, "total_revenue": total_rev, "total_revenue_fmt": f"{total_rev:,.0f}"} # ────────────────────────────────────────────────────────────────────────────── # Per-page context # ────────────────────────────────────────────────────────────────────────────── # ── AI insights cache (auto-run, 5-min TTL) ─────────────────────────────────── _INSIGHTS_TTL = 300 _INSIGHTS_CACHE: dict = {"ts": None, "data": None, "running": False} def invalidate_insights() -> None: """Mark the insights cache stale so the next dashboard render re-runs analysis.""" _INSIGHTS_CACHE["ts"] = None _INSIGHTS_CACHE["data"] = None def _confidence_band(pct: int) -> tuple[str, str]: if pct >= 85: return "high", "High confidence" if pct >= 70: return "med", "Medium confidence" return "low", "Low confidence" def _build_structured_insights(agent_state: dict) -> dict: """Convert the agent's KiranaState into typed dashboard rows. Each row carries the verb + noun + qty + reason needed for Principle 1 (AI suggestions must be concrete). No row renders without all four. """ today = datetime.date.today() # ── Reorder queue (ranked) ──────────────────────────────────────────────── reorders: list[dict] = [] for o in agent_state.get("suggested_orders", []) or []: pid = o.get("product_id") prod = db.get_product(pid) if pid else None name_te = (prod or {}).get("name_local") or "" on_hand = (prod or {}).get("quantity") unit = o.get("unit") or ((prod or {}).get("unit") or "") qty = o.get("qty_needed") or 0 if not o.get("product_name") or not qty or not unit: continue conf_pct = int(round((o.get("ai_confidence") or 0) * 100)) conf_cls, conf_label = _confidence_band(conf_pct) reorders.append({ "pid": pid, "name": o["product_name"], "name_te": name_te, "qty": qty, "qty_fmt": (f"{qty:g}"), "unit": unit, "on_hand": f"{on_hand:g} {unit}" if on_hand is not None else "", "reason": o.get("reason") or "Below minimum stock.", "conf_pct": conf_pct, "conf_cls": conf_cls, "conf_label": conf_label, }) reorders.sort(key=lambda r: -r["conf_pct"]) # ── Expiry liquidation (stacked) ────────────────────────────────────────── liquidation: list[dict] = [] for p in agent_state.get("expired_items", []) or []: liquidation.append({ "pid": p.get("id"), "name": p["name"], "name_te": p.get("name_local") or "", "qty_fmt": f"{p['quantity']:g} {p['unit']}", "days_left": -1, "kind": "expired", "chip_label": "Past expiry", "expiry_date": p.get("expiry_date") or "", "action_verb": "Pull from godown", "action_hint": "Past expiry. Remove from godown today.", }) for p in agent_state.get("expiring_items", []) or []: try: days = (datetime.date.fromisoformat(p["expiry_date"]) - today).days except (ValueError, TypeError): days = 99 if days <= 2: kind, discount, route = "danger", "15%", "route-3 kiranas" elif days <= 4: kind, discount, route = "warn", "12%", "your nearest route" else: kind, discount, route = "warn", "10%", "route-3 kiranas" chip = "Expires today" if days == 0 else ("Expires tomorrow" if days == 1 else f"{days} days left") liquidation.append({ "pid": p.get("id"), "name": p["name"], "name_te": p.get("name_local") or "", "qty_fmt": f"{p['quantity']:g} {p['unit']}", "days_left": days, "kind": kind, "chip_label": chip, "expiry_date": p["expiry_date"], "action_verb": f"Offer {discount} off", "action_hint": f"Discount {discount} to {route} to clear {p['quantity']:g} {p['unit']} before expiry.", }) liquidation.sort(key=lambda r: (r["days_left"] if r["days_left"] >= 0 else -1)) # ── Festival demand strip (timeline) ────────────────────────────────────── inv = agent_state.get("all_products", []) or [] inv_names = {p["name"].lower() for p in inv} fest_strip: list[dict] = [] for f in (agent_state.get("upcoming_festivals", []) or [])[:4]: demand_items = f.get("demand_items", []) or [] missing = [i for i in demand_items if i.lower() not in inv_names] stocked = [i for i in demand_items if i.lower() in inv_names] prep_days = f.get("prep_days", 14) # Approximate timing: festival is "this month" or "next month". # Without exact dates we render "within prep window" / "stocking starts soon". this_month = today.month in (f.get("months") or []) urgent = this_month # stocking window has opened eta_label = "This month" if this_month else "Next month" fest_strip.append({ "key": f.get("key"), "name": f["name"], "name_te": f.get("name_te", ""), "mult": f.get("demand_multiplier", 1.0), "mult_fmt": f"{f.get('demand_multiplier', 1.0):g}×", "prep_days": prep_days, "eta_label": eta_label, "urgent": urgent, "stocked_n": len(stocked), "demand_n": len(demand_items), "missing_n": len(missing), "missing_preview": ", ".join(missing[:3]), "tip": f.get("tips", ""), }) return { "reorders": reorders, "liquidation": liquidation, "festival_strip": fest_strip, "counts": { "reorders": len(reorders), "liquidation": len(liquidation), "festivals": len(fest_strip), }, } def _get_insights(force: bool = False) -> tuple[dict, datetime.datetime | None, str | None]: """Return (structured_data, last_run_ts, error). Auto-runs analysis on TTL miss.""" now = datetime.datetime.now() cache = _INSIGHTS_CACHE fresh = ( cache["data"] is not None and cache["ts"] is not None and (now - cache["ts"]).total_seconds() < _INSIGHTS_TTL ) if fresh and not force: return cache["data"], cache["ts"], None try: from frontend_backend import run_analysis # local import: avoid circular at module load agent_state = run_analysis() data = _build_structured_insights(agent_state) cache["ts"] = now cache["data"] = data return data, now, None except Exception as e: # noqa: BLE001 # Fall back to last good data if available return (cache["data"] or _empty_insights()), cache["ts"], str(e) def _empty_insights() -> dict: return { "reorders": [], "liquidation": [], "festival_strip": [], "counts": {"reorders": 0, "liquidation": 0, "festivals": 0}, } def _format_ts(ts: datetime.datetime | None) -> str: if ts is None: return "Never analysed" delta = datetime.datetime.now() - ts secs = int(delta.total_seconds()) if secs < 60: return "just now" if secs < 3600: return f"{secs // 60} min ago" if secs < 86400: return f"{secs // 3600} hr ago" return ts.strftime("%b %d, %H:%M") def _ctx_dashboard(state: dict) -> dict: insights, ts, err = _get_insights() return { "kpis": _build_kpis(), "pulse": _build_pulse(14), "health": _build_health(), "alerts": _build_alerts(), "festivals": _build_festivals(), "icons": { "health": ICON["pulse"], "alerts": ICON["bell"], "festivals": ICON["seasonal"], "ai": ICON["sparkles"], "play": ICON["play"], "refresh": ICON["refresh"], "package": ICON["package"], "search": ICON["search"], "trophy": ICON["trophy"], "plus": ICON["plus"], "clock": ICON["clock"], "trend": ICON["trend"], "shipped": ICON["shipped"], }, "insights": insights, "insights_meta": { "last_run": _format_ts(ts), "stale": ts is None, "error": err, }, } def _ctx_inventory(state: dict) -> dict: f = state.get("filters") or {} q = f.get("q", "") or "" category = f.get("category", "All") or "All" status = f.get("status", "All") or "All" products = db.search_products(q) if q.strip() else db.get_all_products() if category != "All": products = [p for p in products if p["category"] == category] if status != "All": products = [p for p in products if _status_of(p)[2] == status] rows = [] for p in products: label, kind, _ = _status_of(p) rows.append({ "id": p["id"], "name": p["name"], "name_local": p["name_local"], "category": p["category"], "qty_str": f"{p['quantity']} {p['unit']}", "min_str": f"{p['min_stock']} {p['unit']}", "sell_price": p["sell_price"], "expiry": p.get("expiry_date") or "", "supplier": p.get("supplier") or "", "status_label": label, "status_kind": kind, }) return { "products": rows, "categories": ["All"] + db.CATEGORIES, "statuses": STATUS_FILTERS, "filters": {"q": q, "category": category, "status": status}, } def _ctx_add(state: dict) -> dict: return { "categories": db.CATEGORIES, "units": db.UNITS, "active_method": state.get("active_method", "manual"), "photo_result": state.get("photo_result"), "voice_result": state.get("voice_result"), } def _ctx_orders(state: dict) -> dict: active_filter = state.get("orders_filter", "pending") all_orders = db.get_all_orders() if active_filter != "all": all_orders = [o for o in all_orders if o["status"] == active_filter] status_kind = {"pending": "warn", "approved": "success", "received": "success", "rejected": "danger"} rows = [{ "id": o["id"], "product_name": o["product_name"], "product_id": o["product_id"], "qty_needed": o["qty_needed"], "unit": o["unit"], "qty_str": f"{o['qty_needed']} {o['unit']}", "reason": (o["reason"] or "")[:80], "confidence": f"{o['ai_confidence']*100:.0f}", "status_raw": o["status"], "status": o["status"].upper(), "status_kind": status_kind.get(o["status"], "neutral"), "created_at": (o["created_at"] or "")[:16], } for o in all_orders] return { "orders": rows, "order_filters": ["pending", "approved", "received", "rejected", "all"], "active_filter": active_filter, } def _ctx_analytics(state: dict) -> dict: try: days = int(state.get("analytics_days", 30)) except (TypeError, ValueError): days = 30 if days not in {7, 30, 90}: days = 30 s = db.get_summary() val = s.get("total_value", 0) or 0 cost = s.get("cost_value", 0) or 0 margin = val - cost margin_pct = (margin / val * 100) if val else 0 top = db.get_top_sellers(n=5, days=days) units_sold = sum(t["total_sold"] for t in top) if top else 0 revenue = sum((t.get("revenue") or 0) for t in top) if top else 0 return { "categories": _build_categories(), "sellers": _build_sellers(days), "analytics_days": days, "analytics_ranges": [7, 30, 90], "icon_chart": ICON["analytics"], "icon_trophy": ICON["trophy"], "metrics": [ {"label": f"{days}-Day Revenue", "value": f"₹{revenue:,.0f}", "icon": ICON["rupee"], "color": "success", "foot": "from top 5 sellers"}, {"label": "Units Sold", "value": f"{units_sold:.0f}", "icon": ICON["shipped"], "color": "primary", "foot": f"last {days} days"}, {"label": "Gross Margin", "value": f"{margin_pct:.1f}%", "icon": ICON["trend"], "color": "info", "foot": f"₹{margin:,.0f} held on shelf"}, ], } def _ctx_seasonal(_state: dict) -> dict: return { "festivals": _build_festivals(), "brief": build_seasonal_summary().replace("\n", "
"), } def _ctx_settings(_state: dict) -> dict: return { "s": { "shop_name": db.get_setting("shop_name", ""), "owner_name": db.get_setting("owner_name", ""), "region": db.get_setting("region", ""), "low_stock_days_ahead": db.get_setting("low_stock_days_ahead", "7"), "expiry_warn_days": db.get_setting("expiry_warn_days", "7"), }, "vision_online": is_model_available(), "vision_status": ( "Vision model available (LLaVA)" if is_model_available() else "Vision model not loaded — run setup.sh to download" ), } _CTX = { "dashboard": _ctx_dashboard, "inventory": _ctx_inventory, "add": _ctx_add, "orders": _ctx_orders, "analytics": _ctx_analytics, "seasonal": _ctx_seasonal, "settings": _ctx_settings, } # ────────────────────────────────────────────────────────────────────────────── # Top-bar action buttons # ────────────────────────────────────────────────────────────────────────────── def _theme_toggle() -> str: return ( '' ) def _topbar_actions(page: str) -> str: today = datetime.date.today().strftime("%a, %d %b %Y") badge = f'{today}' toggle = _theme_toggle() if page == "dashboard": return ( badge + '' '' + toggle ) if page == "inventory": return ( badge + '' + toggle ) if page == "orders": return ( badge + '' + toggle ) return badge + toggle # ────────────────────────────────────────────────────────────────────────────── # Public render entry # ────────────────────────────────────────────────────────────────────────────── def render(page: str, state: dict | None = None, toast: str = "") -> str: state = state or {} if page not in _CTX: page = "dashboard" nav_info = NAV_PAGES_BY_KEY[page] body_template = env.get_template(f"{page}.html") ctx = _CTX[page](state) ctx.setdefault("icons", {}) for _k in ("seasonal", "sparkles", "package", "bell", "info", "clock", "edit", "trash", "store", "settings", "mic", "camera", "plus", "refresh", "search", "trend", "pulse", "rupee", "shipped", "trophy", "alert", "check", "play", "x"): if _k in ICON: ctx["icons"].setdefault(_k, ICON[_k]) # Aliases used by dashboard _aliases = {"health": "pulse", "alerts": "bell", "festivals": "seasonal", "ai": "sparkles"} for _a, _src in _aliases.items(): ctx["icons"].setdefault(_a, ICON[_src]) body = body_template.render(**ctx) return env.get_template("base.html").render( page=page, nav=_nav_with_badges(), title=nav_info["title"], subtitle=nav_info["subtitle"], topbar_actions=_topbar_actions(page), body=body, logo_svg=LOGO_SVG, shop_name=db.get_setting("shop_name", "My Kirana Store"), region=db.get_setting("region", "India"), vision_online=is_model_available(), toast=toast, ) NAV_PAGES_BY_KEY = {p["key"]: p for p in NAV_PAGES}