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()