Spaces:
Sleeping
Sleeping
| import os | |
| import json | |
| import random | |
| from datetime import datetime, timedelta | |
| from typing import Dict, Any, List, Tuple | |
| import gradio as gr | |
| import pandas as pd | |
| # ============================ | |
| # Branding | |
| # ============================ | |
| PROCELEVATE_BLUE = "#0F2C59" | |
| CUSTOM_CSS = f""" | |
| /* Primary buttons */ | |
| .gr-button.gr-button-primary, | |
| button.primary {{ | |
| background: {PROCELEVATE_BLUE} !important; | |
| border-color: {PROCELEVATE_BLUE} !important; | |
| color: white !important; | |
| font-weight: 650 !important; | |
| }} | |
| .gr-button.gr-button-primary:hover, | |
| button.primary:hover {{ | |
| filter: brightness(0.92); | |
| }} | |
| /* Tabs selected */ | |
| button[data-testid="tab-button"][aria-selected="true"] {{ | |
| border-bottom: 3px solid {PROCELEVATE_BLUE} !important; | |
| color: {PROCELEVATE_BLUE} !important; | |
| font-weight: 750 !important; | |
| }} | |
| /* Subtle modern rounding */ | |
| .block, .gr-box, .gr-panel {{ | |
| border-radius: 14px !important; | |
| }} | |
| """ | |
| # ============================ | |
| # Settings / Paths | |
| # ============================ | |
| DATA_DIR = "data" | |
| OPS_FILE = os.path.join(DATA_DIR, "ops_events.json") | |
| ADMIN_PIN = os.environ.get("ADMIN_PIN", "2580") # demo PIN | |
| # ============================ | |
| # Demo: generate operational events | |
| # ============================ | |
| DEPARTMENTS = ["Front Office", "Housekeeping", "F&B", "Maintenance", "Security"] | |
| EVENT_TYPES = [ | |
| "Check-in delay", | |
| "Self check-in success", | |
| "Concierge question", | |
| "Room service order", | |
| "Housekeeping request", | |
| "Towel request", | |
| "Maintenance issue", | |
| "Noise complaint", | |
| "Wi-Fi complaint", | |
| "Late checkout request", | |
| "Breakfast query", | |
| "Dinner menu query", | |
| ] | |
| SENTIMENTS = ["Positive", "Neutral", "Negative"] | |
| def ensure_data_dir(): | |
| os.makedirs(DATA_DIR, exist_ok=True) | |
| def load_events() -> List[Dict[str, Any]]: | |
| ensure_data_dir() | |
| if not os.path.exists(OPS_FILE): | |
| return [] | |
| try: | |
| with open(OPS_FILE, "r", encoding="utf-8") as f: | |
| data = json.load(f) | |
| return data if isinstance(data, list) else [] | |
| except Exception: | |
| return [] | |
| def save_events(events: List[Dict[str, Any]]): | |
| ensure_data_dir() | |
| with open(OPS_FILE, "w", encoding="utf-8") as f: | |
| json.dump(events, f, ensure_ascii=False, indent=2) | |
| def dt_now_str(): | |
| return datetime.now().strftime("%Y-%m-%d %H:%M") | |
| def date_str(d: datetime): | |
| return d.strftime("%Y-%m-%d") | |
| def simulate_events(days: int = 7, seed: int = 42) -> List[Dict[str, Any]]: | |
| random.seed(seed) | |
| base = datetime.now().date() | |
| events = [] | |
| for i in range(days): | |
| d = base - timedelta(days=(days - 1 - i)) | |
| # Vary volumes by day (simulate peaks) | |
| base_volume = random.randint(80, 140) | |
| peak_multiplier = 1.15 if d.weekday() in [4, 5] else 1.0 # Fri/Sat peaks | |
| volume = int(base_volume * peak_multiplier) | |
| for _ in range(volume): | |
| evt_type = random.choices( | |
| EVENT_TYPES, | |
| weights=[7, 10, 18, 7, 14, 12, 7, 5, 5, 5, 6, 8], | |
| k=1 | |
| )[0] | |
| dept = "Front Office" | |
| if evt_type in ["Housekeeping request", "Towel request"]: | |
| dept = "Housekeeping" | |
| elif evt_type in ["Dinner menu query", "Breakfast query", "Room service order"]: | |
| dept = "F&B" | |
| elif evt_type in ["Maintenance issue", "Wi-Fi complaint"]: | |
| dept = "Maintenance" | |
| elif evt_type in ["Noise complaint"]: | |
| dept = "Security" | |
| sentiment = random.choices(SENTIMENTS, weights=[35, 45, 20], k=1)[0] | |
| if evt_type in ["Noise complaint", "Wi-Fi complaint", "Maintenance issue", "Check-in delay"]: | |
| sentiment = random.choices(SENTIMENTS, weights=[10, 35, 55], k=1)[0] | |
| if evt_type in ["Self check-in success"]: | |
| sentiment = random.choices(SENTIMENTS, weights=[70, 25, 5], k=1)[0] | |
| # Simulated timestamps (spread within day) | |
| hour = random.randint(6, 23) | |
| minute = random.randint(0, 59) | |
| ts = datetime(d.year, d.month, d.day, hour, minute) | |
| # Extra attributes for some events | |
| wait_mins = None | |
| if evt_type == "Check-in delay": | |
| wait_mins = random.randint(8, 25) | |
| req_priority = None | |
| if evt_type in ["Maintenance issue", "Noise complaint"]: | |
| req_priority = random.choices(["Normal", "Urgent"], weights=[70, 30], k=1)[0] | |
| events.append({ | |
| "timestamp": ts.strftime("%Y-%m-%d %H:%M"), | |
| "date": ts.strftime("%Y-%m-%d"), | |
| "department": dept, | |
| "event_type": evt_type, | |
| "sentiment": sentiment, | |
| "wait_mins": wait_mins, | |
| "priority": req_priority, | |
| "channel": random.choice(["Front Desk", "Phone", "WhatsApp", "Web/App", "Concierge Agent"]), | |
| }) | |
| return events | |
| # ============================ | |
| # Analytics + Pulse generation | |
| # ============================ | |
| def events_to_df(events: List[Dict[str, Any]]) -> pd.DataFrame: | |
| if not events: | |
| return pd.DataFrame(columns=["timestamp", "date", "department", "event_type", "sentiment", "wait_mins", "priority", "channel"]) | |
| df = pd.DataFrame(events) | |
| return df | |
| def compute_kpis(df: pd.DataFrame, target_date: str) -> Dict[str, Any]: | |
| if df.empty: | |
| return { | |
| "target_date": target_date, | |
| "total_events": 0, | |
| "neg_sentiment_rate": 0.0, | |
| "self_checkin_success": 0, | |
| "checkin_delays": 0, | |
| "avg_delay_mins": None, | |
| "hk_requests": 0, | |
| "wifi_complaints": 0, | |
| "maintenance_issues": 0, | |
| "dinner_queries": 0, | |
| } | |
| ddf = df[df["date"] == target_date].copy() | |
| if ddf.empty: | |
| return { | |
| "target_date": target_date, | |
| "total_events": 0, | |
| "neg_sentiment_rate": 0.0, | |
| "self_checkin_success": 0, | |
| "checkin_delays": 0, | |
| "avg_delay_mins": None, | |
| "hk_requests": 0, | |
| "wifi_complaints": 0, | |
| "maintenance_issues": 0, | |
| "dinner_queries": 0, | |
| } | |
| total = len(ddf) | |
| neg_rate = (ddf["sentiment"].eq("Negative").sum() / total) if total else 0.0 | |
| delays = ddf[ddf["event_type"] == "Check-in delay"] | |
| avg_delay = None | |
| if not delays.empty and delays["wait_mins"].notna().any(): | |
| avg_delay = float(delays["wait_mins"].dropna().mean()) | |
| return { | |
| "target_date": target_date, | |
| "total_events": int(total), | |
| "neg_sentiment_rate": float(neg_rate), | |
| "self_checkin_success": int((ddf["event_type"] == "Self check-in success").sum()), | |
| "checkin_delays": int((ddf["event_type"] == "Check-in delay").sum()), | |
| "avg_delay_mins": avg_delay, | |
| "hk_requests": int(ddf["event_type"].isin(["Housekeeping request", "Towel request"]).sum()), | |
| "wifi_complaints": int((ddf["event_type"] == "Wi-Fi complaint").sum()), | |
| "maintenance_issues": int((ddf["event_type"] == "Maintenance issue").sum()), | |
| "dinner_queries": int((ddf["event_type"] == "Dinner menu query").sum()), | |
| } | |
| def compare_to_prev_day(df: pd.DataFrame, target_date: str) -> Dict[str, Any]: | |
| t = datetime.strptime(target_date, "%Y-%m-%d").date() | |
| prev = t - timedelta(days=1) | |
| prev_date = prev.strftime("%Y-%m-%d") | |
| k_today = compute_kpis(df, target_date) | |
| k_prev = compute_kpis(df, prev_date) | |
| def delta(a, b): | |
| if a is None or b is None: | |
| return None | |
| return a - b | |
| return { | |
| "prev_date": prev_date, | |
| "today": k_today, | |
| "prev": k_prev, | |
| "delta_total_events": delta(k_today["total_events"], k_prev["total_events"]), | |
| "delta_neg_rate_pp": delta(k_today["neg_sentiment_rate"]*100, k_prev["neg_sentiment_rate"]*100), | |
| "delta_checkin_delays": delta(k_today["checkin_delays"], k_prev["checkin_delays"]), | |
| "delta_hk_requests": delta(k_today["hk_requests"], k_prev["hk_requests"]), | |
| "delta_maintenance": delta(k_today["maintenance_issues"], k_prev["maintenance_issues"]), | |
| "delta_wifi": delta(k_today["wifi_complaints"], k_prev["wifi_complaints"]), | |
| "delta_dinner_queries": delta(k_today["dinner_queries"], k_prev["dinner_queries"]), | |
| } | |
| def build_alerts_and_actions(k: Dict[str, Any], comp: Dict[str, Any]) -> Tuple[pd.DataFrame, List[str], List[str]]: | |
| alerts = [] | |
| actions = [] | |
| positives = [] | |
| # Thresholds (demo defaults) | |
| neg_rate = k["neg_sentiment_rate"] | |
| delays = k["checkin_delays"] | |
| avg_delay = k["avg_delay_mins"] | |
| hk = k["hk_requests"] | |
| wifi = k["wifi_complaints"] | |
| maint = k["maintenance_issues"] | |
| dinner = k["dinner_queries"] | |
| # Alerts | |
| if neg_rate >= 0.30: | |
| alerts.append(("RED", "Guest dissatisfaction spike", f"Negative sentiment rate is {neg_rate*100:.0f}% today.")) | |
| actions.append("GM to review top complaints today; run 10-min standup with FO/HK/F&B leads.") | |
| elif neg_rate >= 0.22: | |
| alerts.append(("AMBER", "Guest dissatisfaction rising", f"Negative sentiment rate is {neg_rate*100:.0f}% today.")) | |
| actions.append("Supervisor to spot-check service recovery for negative interactions.") | |
| if delays >= 8: | |
| details = f"{delays} check-in delay events today." | |
| if avg_delay is not None: | |
| details += f" Avg delay ~{avg_delay:.0f} mins." | |
| alerts.append(("RED", "Front desk check-in delays", details)) | |
| actions.append("Add 1 staff during peak arrival window; use express/self-check flow for pre-arrivals.") | |
| elif delays >= 4: | |
| alerts.append(("AMBER", "Check-in delays observed", f"{delays} check-in delay events today.")) | |
| actions.append("Review arrival peaks; pre-assign rooms for early arrivals.") | |
| if hk >= 25: | |
| alerts.append(("AMBER", "High housekeeping load", f"{hk} housekeeping-related requests today.")) | |
| actions.append("Temporarily re-balance HK routes; pre-stage linens/towels for speed.") | |
| if wifi >= 6: | |
| alerts.append(("AMBER", "Wi-Fi issues", f"{wifi} Wi-Fi complaints today.")) | |
| actions.append("Check AP health in hotspot floors; proactive message with Wi-Fi steps to guests.") | |
| if maint >= 6: | |
| alerts.append(("AMBER", "Maintenance load high", f"{maint} maintenance issues today.")) | |
| actions.append("Prioritize urgent issues; schedule preventive checks during low occupancy hours.") | |
| # Trend alerts vs previous day | |
| if comp and comp.get("delta_checkin_delays") is not None and comp["delta_checkin_delays"] >= 4: | |
| alerts.append(("AMBER", "Delays increased vs yesterday", f"Check-in delays up by {comp['delta_checkin_delays']} vs {comp['prev_date']}.")) | |
| if comp and comp.get("delta_hk_requests") is not None and comp["delta_hk_requests"] >= 8: | |
| alerts.append(("AMBER", "HK requests increased vs yesterday", f"HK-related requests up by {comp['delta_hk_requests']} vs {comp['prev_date']}.")) | |
| # Positives | |
| if k["self_checkin_success"] >= 15: | |
| positives.append(f"Self check-in adoption is strong ({k['self_checkin_success']} successful self check-ins).") | |
| if delays <= 2 and k["total_events"] > 0: | |
| positives.append("Front desk flow looks stable today (low check-in delays).") | |
| if maint == 0 and k["total_events"] > 0: | |
| positives.append("No maintenance issues recorded today.") | |
| if neg_rate <= 0.15 and k["total_events"] > 0: | |
| positives.append("Guest sentiment is healthy today (low negative rate).") | |
| # If no alerts, add a default positive note | |
| if not alerts and k["total_events"] > 0: | |
| positives.append("No major operational risks detected. Continue monitoring peak windows.") | |
| alerts_df = pd.DataFrame(alerts, columns=["Severity", "Category", "Detail"]) if alerts else pd.DataFrame(columns=["Severity", "Category", "Detail"]) | |
| return alerts_df, actions, positives | |
| def generate_pulse_text(k: Dict[str, Any], comp: Dict[str, Any], alerts_df: pd.DataFrame, actions: List[str], positives: List[str]) -> str: | |
| td = k["target_date"] | |
| prev = comp.get("prev_date") if comp else None | |
| # Small deltas summary | |
| delta_bits = [] | |
| if comp: | |
| if comp.get("delta_total_events") is not None: | |
| delta_bits.append(f"Total activity {'+' if comp['delta_total_events']>=0 else ''}{comp['delta_total_events']} vs {prev}") | |
| if comp.get("delta_neg_rate_pp") is not None: | |
| delta_bits.append(f"Neg. sentiment {'+' if comp['delta_neg_rate_pp']>=0 else ''}{comp['delta_neg_rate_pp']:.0f} pp vs {prev}") | |
| if comp.get("delta_checkin_delays") is not None: | |
| delta_bits.append(f"Check-in delays {'+' if comp['delta_checkin_delays']>=0 else ''}{comp['delta_checkin_delays']} vs {prev}") | |
| delta_line = " | ".join(delta_bits) if delta_bits else "Trend comparison not available." | |
| # Compose narrative | |
| top_alerts = "" | |
| if not alerts_df.empty: | |
| # show up to 3 alerts in text | |
| top = alerts_df.head(3).to_dict(orient="records") | |
| lines = [] | |
| for a in top: | |
| icon = "π΄" if a["Severity"] == "RED" else "π " | |
| lines.append(f"{icon} **{a['Category']}** β {a['Detail']}") | |
| top_alerts = "\n".join(lines) | |
| else: | |
| top_alerts = "π’ No major operational risks detected." | |
| # Actions list (up to 4) | |
| action_lines = "\n".join([f"β {a}" for a in actions[:4]]) if actions else "β Maintain current staffing and monitor peaks." | |
| # Positives (up to 3) | |
| pos_lines = "\n".join([f"π’ {p}" for p in positives[:3]]) if positives else "π’ Stable day expected." | |
| avg_delay_str = f"{k['avg_delay_mins']:.0f} mins" if k["avg_delay_mins"] is not None else "N/A" | |
| pulse = f""" | |
| ## π Hotel Operations Pulse β {td} | |
| **Snapshot** | |
| - Total operational signals captured: **{k['total_events']}** | |
| - Negative sentiment rate: **{k['neg_sentiment_rate']*100:.0f}%** | |
| - Check-in delays: **{k['checkin_delays']}** (avg delay: **{avg_delay_str}**) | |
| - Housekeeping-related requests: **{k['hk_requests']}** | |
| - Maintenance issues: **{k['maintenance_issues']}** | |
| - Wi-Fi complaints: **{k['wifi_complaints']}** | |
| - Dinner/menu queries: **{k['dinner_queries']}** | |
| - Self check-in successes: **{k['self_checkin_success']}** | |
| **Trend vs {prev if prev else 'previous day'}** | |
| - {delta_line} | |
| ### π¦ Key Alerts | |
| {top_alerts} | |
| ### β Recommended Actions (Manager / Supervisor) | |
| {action_lines} | |
| ### π Positive Signals | |
| {pos_lines} | |
| **Note:** This is a demo pulse generated from sample operational signals. In production, this can connect to PMS / POS / housekeeping logs / guest feedback channels. | |
| """ | |
| return pulse.strip() | |
| def kpis_table(k: Dict[str, Any], comp: Dict[str, Any]) -> pd.DataFrame: | |
| def fmt_delta(x): | |
| if x is None: | |
| return "" | |
| return f"{'+' if x>=0 else ''}{x}" | |
| rows = [ | |
| ("Total signals", k["total_events"], fmt_delta(comp.get("delta_total_events") if comp else None)), | |
| ("Negative sentiment (%)", round(k["neg_sentiment_rate"]*100), f"{fmt_delta(round(comp.get('delta_neg_rate_pp')))} pp" if comp and comp.get("delta_neg_rate_pp") is not None else ""), | |
| ("Check-in delays (#)", k["checkin_delays"], fmt_delta(comp.get("delta_checkin_delays") if comp else None)), | |
| ("Avg delay (mins)", (round(k["avg_delay_mins"]) if k["avg_delay_mins"] is not None else "N/A"), ""), | |
| ("HK requests (#)", k["hk_requests"], fmt_delta(comp.get("delta_hk_requests") if comp else None)), | |
| ("Maintenance issues (#)", k["maintenance_issues"], fmt_delta(comp.get("delta_maintenance") if comp else None)), | |
| ("Wi-Fi complaints (#)", k["wifi_complaints"], fmt_delta(comp.get("delta_wifi") if comp else None)), | |
| ("Dinner/menu queries (#)", k["dinner_queries"], fmt_delta(comp.get("delta_dinner_queries") if comp else None)), | |
| ("Self check-in successes (#)", k["self_checkin_success"], ""), | |
| ] | |
| return pd.DataFrame(rows, columns=["Metric", "Today", "Ξ vs Yesterday"]) | |
| # ============================ | |
| # Ops Assistant (simple NL routing) | |
| # ============================ | |
| def ops_assistant_answer(question: str, k: Dict[str, Any], comp: Dict[str, Any], alerts_df: pd.DataFrame, actions: List[str]) -> str: | |
| q = (question or "").strip().lower() | |
| if not q: | |
| return "Ask something like: βWhat needs attention today?β or βAny issues in housekeeping?β" | |
| if "attention" in q or "focus" in q or "urgent" in q or "risk" in q: | |
| if alerts_df.empty: | |
| return "π’ No major risks detected. Focus on peak arrival windows and keep service recovery readiness." | |
| top = alerts_df.head(3).to_dict(orient="records") | |
| lines = [] | |
| for a in top: | |
| icon = "π΄" if a["Severity"] == "RED" else "π " | |
| lines.append(f"{icon} {a['Category']}: {a['Detail']}") | |
| return "Here are the top items needing attention:\n- " + "\n- ".join(lines) | |
| if "housekeeping" in q or "towel" in q: | |
| return f"Housekeeping load today: {k['hk_requests']} HK-related requests. " + ( | |
| "Recommendation: re-balance routes and pre-stage linens/towels during peak." | |
| if k["hk_requests"] >= 20 else | |
| "Load looks manageable; keep monitoring peak hours." | |
| ) | |
| if "front" in q or "check-in" in q or "lobby" in q: | |
| avg_delay = f"{k['avg_delay_mins']:.0f} mins" if k["avg_delay_mins"] is not None else "N/A" | |
| return f"Front desk today: {k['checkin_delays']} check-in delay signals (avg: {avg_delay}). " + ( | |
| "Recommendation: add 1 staff during peak + push self-check pre-arrival." | |
| if k["checkin_delays"] >= 4 else | |
| "Flow looks stable; keep express check-in visible." | |
| ) | |
| if "wifi" in q: | |
| return f"Wi-Fi complaints today: {k['wifi_complaints']}. " + ( | |
| "Recommendation: check AP health + proactive guest message with Wi-Fi steps." | |
| if k["wifi_complaints"] >= 4 else | |
| "Low complaint volume; continue monitoring." | |
| ) | |
| if "recommend" in q or "action" in q or "do next" in q: | |
| if not actions: | |
| return "Recommended actions: maintain staffing plan, monitor peaks, and review any negative feedback quickly." | |
| return "Recommended actions:\n- " + "\n- ".join(actions[:5]) | |
| if "compare" in q or "yesterday" in q or "trend" in q: | |
| if not comp: | |
| return "Trend comparison not available." | |
| msg = ( | |
| f"Compared to {comp['prev_date']}:\n" | |
| f"- Total signals: {comp['delta_total_events']:+d}\n" | |
| f"- Check-in delays: {comp['delta_checkin_delays']:+d}\n" | |
| f"- HK requests: {comp['delta_hk_requests']:+d}\n" | |
| f"- Maintenance issues: {comp['delta_maintenance']:+d}\n" | |
| f"- Wi-Fi complaints: {comp['delta_wifi']:+d}\n" | |
| ) | |
| if comp.get("delta_neg_rate_pp") is not None: | |
| msg += f"- Negative sentiment: {comp['delta_neg_rate_pp']:+.0f} pp\n" | |
| return msg | |
| return "I can help with: risks, priorities, department issues (front desk/housekeeping/F&B/maintenance), trends vs yesterday, and recommended actions. Try: βWhat needs attention today?β" | |
| # ============================ | |
| # UI Actions | |
| # ============================ | |
| def refresh_pulse(selected_date: str) -> Tuple[str, pd.DataFrame, pd.DataFrame, str, Dict[str, Any]]: | |
| events = load_events() | |
| df = events_to_df(events) | |
| if not selected_date: | |
| # default to latest date in dataset | |
| if df.empty: | |
| selected_date = date_str(datetime.now()) | |
| else: | |
| selected_date = sorted(df["date"].unique())[-1] | |
| k = compute_kpis(df, selected_date) | |
| comp = compare_to_prev_day(df, selected_date) if not df.empty else {} | |
| alerts_df, actions, positives = build_alerts_and_actions(k, comp) | |
| pulse_md = generate_pulse_text(k, comp, alerts_df, actions, positives) | |
| kpi_df = kpis_table(k, comp) | |
| quick_summary = ( | |
| f"Today: {k['total_events']} signals | Neg: {k['neg_sentiment_rate']*100:.0f}% | " | |
| f"Delays: {k['checkin_delays']} | HK: {k['hk_requests']} | Maint: {k['maintenance_issues']}" | |
| ) | |
| return pulse_md, kpi_df, alerts_df, quick_summary, {"k": k, "comp": comp, "alerts": alerts_df.to_dict(orient="records"), "actions": actions} | |
| def answer_ops(question: str, state: Dict[str, Any]) -> str: | |
| if not state or "k" not in state: | |
| return "Please generate the pulse first." | |
| k = state["k"] | |
| comp = state.get("comp", {}) | |
| alerts_df = pd.DataFrame(state.get("alerts", [])) | |
| actions = state.get("actions", []) | |
| return ops_assistant_answer(question, k, comp, alerts_df, actions) | |
| def admin_unlock(pin: str): | |
| if (pin or "").strip() == ADMIN_PIN: | |
| return gr.update(visible=False), gr.update(visible=True), "β Admin access granted." | |
| return gr.update(visible=True), gr.update(visible=False), "β Incorrect PIN." | |
| def admin_generate(days: int, seed: int): | |
| events = simulate_events(days=int(days), seed=int(seed)) | |
| save_events(events) | |
| return f"β Generated {len(events)} demo operational events across last {days} day(s). Updated at {dt_now_str()}." | |
| def admin_clear(pin: str): | |
| if (pin or "").strip() != ADMIN_PIN: | |
| return "β Incorrect PIN. Cannot clear data." | |
| save_events([]) | |
| return f"β Cleared demo data at {dt_now_str()}." | |
| # ============================ | |
| # Build UI | |
| # ============================ | |
| with gr.Blocks(title="AI Hotel Operations Pulse (Prototype)", css=CUSTOM_CSS) as demo: | |
| gr.Markdown( | |
| """ | |
| # π AI Hotel Operations Pulse (Prototype) | |
| A manager/owner-focused assistant that summarizes hotel health, flags risks, and recommends actions β **without reading long reports**. | |
| **Outputs:** Daily Pulse β’ KPI Snapshot β’ Alerts β’ Recommended Actions β’ Ops Assistant Q&A | |
| **Note:** Demo uses sample operational signals. In production, this can connect to PMS/POS/housekeeping logs/guest feedback systems. | |
| """ | |
| ) | |
| state = gr.State({}) | |
| with gr.Tab("Manager / Owner Pulse"): | |
| with gr.Row(): | |
| selected_date = gr.Textbox(label="Pulse Date (YYYY-MM-DD)", placeholder="Leave blank to use latest available date") | |
| btn = gr.Button("Generate Pulse", variant="primary") | |
| quick = gr.Markdown("") | |
| pulse_md = gr.Markdown("") | |
| with gr.Row(): | |
| kpi_table_out = gr.Dataframe(label="KPI Snapshot", interactive=False, wrap=True) | |
| alerts_out = gr.Dataframe(label="Alerts (Red/Amber)", interactive=False, wrap=True) | |
| gr.Markdown("### π§ Ask the Ops Assistant") | |
| q = gr.Textbox(label="Ask a manager-style question", placeholder="e.g., What needs my attention today? Any housekeeping issues? Compare vs yesterday.") | |
| ask_btn = gr.Button("Ask", variant="primary") | |
| a = gr.Textbox(label="Answer", lines=6, interactive=False) | |
| btn.click(refresh_pulse, inputs=[selected_date], outputs=[pulse_md, kpi_table_out, alerts_out, quick, state]) | |
| ask_btn.click(answer_ops, inputs=[q, state], outputs=[a]) | |
| with gr.Tab("Admin (Demo Data)"): | |
| gr.Markdown("### Admin access (PIN protected)") | |
| pin_box = gr.Textbox(label="Enter Admin PIN", type="password", placeholder="PIN") | |
| unlock_btn = gr.Button("Unlock Admin Tools", variant="primary") | |
| unlock_status = gr.Markdown("") | |
| admin_tools = gr.Column(visible=False) | |
| with admin_tools: | |
| gr.Markdown("Generate realistic demo operational signals.") | |
| with gr.Row(): | |
| days = gr.Slider(3, 21, value=7, step=1, label="Days of demo data") | |
| seed = gr.Slider(1, 999, value=42, step=1, label="Random seed (for repeatability)") | |
| gen_btn = gr.Button("Generate / Refresh Demo Data", variant="primary") | |
| gen_out = gr.Markdown("") | |
| gr.Markdown("---") | |
| clear_btn = gr.Button("Clear Demo Data (PIN required)") | |
| clear_out = gr.Markdown("") | |
| gen_btn.click(admin_generate, inputs=[days, seed], outputs=[gen_out]) | |
| clear_btn.click(admin_clear, inputs=[pin_box], outputs=[clear_out]) | |
| unlock_btn.click(admin_unlock, inputs=[pin_box], outputs=[pin_box, admin_tools, unlock_status]) | |
| demo.launch() |