import os import re import json import time import traceback from pathlib import Path from typing import Dict, Any, List, Optional, Tuple import pandas as pd import gradio as gr # papermill imported lazily inside run_notebook() to avoid startup hang # Optional LLM (HuggingFace Inference API) try: from huggingface_hub import InferenceClient except Exception: InferenceClient = None # ========================================================= # CONFIG # ========================================================= BASE_DIR = Path(__file__).resolve().parent NB1 = os.environ.get("NB1", "FILE_1_Python.ipynb").strip() NB2 = os.environ.get("NB2", "FILE_2_Python.ipynb").strip() NB3 = os.environ.get("NB3", "FILE_3_R.ipynb").strip() RUNS_DIR = BASE_DIR / "runs" ART_DIR = BASE_DIR / "artifacts" PY_FIG_DIR = ART_DIR / "py" / "figures" PY_TAB_DIR = ART_DIR / "py" / "tables" R_FIG_DIR = ART_DIR / "r" / "figures" R_TAB_DIR = ART_DIR / "r" / "tables" PAPERMILL_TIMEOUT = int(os.environ.get("PAPERMILL_TIMEOUT", "1800")) MAX_PREVIEW_ROWS = int(os.environ.get("MAX_FILE_PREVIEW_ROWS", "50")) MAX_LOG_CHARS = int(os.environ.get("MAX_LOG_CHARS", "8000")) HF_API_KEY = os.environ.get("HF_API_KEY", "").strip() MODEL_NAME = os.environ.get("MODEL_NAME", "mistralai/Mistral-7B-Instruct-v0.3").strip() HF_PROVIDER = os.environ.get("HF_PROVIDER", "hf-inference").strip() LLM_ENABLED = bool(HF_API_KEY) and InferenceClient is not None llm_client = ( InferenceClient(provider=HF_PROVIDER, api_key=HF_API_KEY) if LLM_ENABLED else None ) # ========================================================= # HELPERS # ========================================================= def ensure_dirs(): for p in [RUNS_DIR, ART_DIR, PY_FIG_DIR, PY_TAB_DIR, R_FIG_DIR, R_TAB_DIR]: p.mkdir(parents=True, exist_ok=True) def stamp(): return time.strftime("%Y%m%d-%H%M%S") def tail(text: str, n: int = MAX_LOG_CHARS) -> str: return (text or "")[-n:] def _ls(dir_path: Path, exts: Tuple[str, ...]) -> List[str]: if not dir_path.is_dir(): return [] return sorted(p.name for p in dir_path.iterdir() if p.is_file() and p.suffix.lower() in exts) def _read_csv(path: Path) -> pd.DataFrame: return pd.read_csv(path, nrows=MAX_PREVIEW_ROWS) def _read_json(path: Path): with path.open(encoding="utf-8") as f: return json.load(f) def artifacts_index() -> Dict[str, Any]: return { "python": { "figures": _ls(PY_FIG_DIR, (".png", ".jpg", ".jpeg")), "tables": _ls(PY_TAB_DIR, (".csv", ".json")), }, "r": { "figures": _ls(R_FIG_DIR, (".png", ".jpg", ".jpeg")), "tables": _ls(R_TAB_DIR, (".csv", ".json")), }, } # ========================================================= # PIPELINE RUNNERS # ========================================================= def run_notebook(nb_name: str, kernel: str = "python3") -> str: import papermill as pm # lazy import ensure_dirs() nb_in = BASE_DIR / nb_name if not nb_in.exists(): return f"ERROR: {nb_name} not found in {BASE_DIR}." nb_out = RUNS_DIR / f"run_{stamp()}_{nb_name}" pm.execute_notebook( input_path=str(nb_in), output_path=str(nb_out), kernel_name=kernel, cwd=str(BASE_DIR), log_output=True, progress_bar=False, request_save_on_cell_execute=True, execution_timeout=PAPERMILL_TIMEOUT, ) return f"Executed {nb_name}" def run_datacreation() -> str: try: log = run_notebook(NB1, kernel="python3") csvs = [f.name for f in BASE_DIR.glob("*.csv")] return ( f"✅ Step 1 DONE — {log}\n\n" f"CSVs present in /app:\n" + "\n".join(f" • {c}" for c in sorted(csvs)) ) except Exception as e: return f"❌ Step 1 FAILED\n\n{e}\n\n{traceback.format_exc()[-2000:]}" def run_pythonanalysis() -> str: try: log = run_notebook(NB2, kernel="python3") idx = artifacts_index() figs = idx["python"]["figures"] tabs = idx["python"]["tables"] return ( f"✅ Step 2a DONE — {log}\n\n" f"Figures ({len(figs)}): {', '.join(figs) or '(none)'}\n" f"Tables ({len(tabs)}): {', '.join(tabs) or '(none)'}" ) except Exception as e: return f"❌ Step 2a FAILED\n\n{e}\n\n{traceback.format_exc()[-2000:]}" def run_r() -> str: try: log = run_notebook(NB3, kernel="ir") idx = artifacts_index() figs = idx["r"]["figures"] tabs = idx["r"]["tables"] return ( f"✅ Step 2b DONE — {log}\n\n" f"Figures ({len(figs)}): {', '.join(figs) or '(none)'}\n" f"Tables ({len(tabs)}): {', '.join(tabs) or '(none)'}" ) except Exception as e: return f"❌ Step 2b FAILED\n\n{e}\n\n{traceback.format_exc()[-2000:]}" def run_full_pipeline() -> str: parts = [] parts.append("=" * 60) parts.append("STEP 1/3 — Data Creation") parts.append("=" * 60) parts.append(run_datacreation()) parts.append("") parts.append("=" * 60) parts.append("STEP 2/3 — Python Analysis (EDA + ML)") parts.append("=" * 60) parts.append(run_pythonanalysis()) parts.append("") parts.append("=" * 60) parts.append("STEP 3/3 — R Statistical Validation") parts.append("=" * 60) parts.append(run_r()) return "\n".join(parts) # ========================================================= # GALLERY LOADERS # ========================================================= def _load_all_figures() -> List[Tuple[str, str]]: items = [] for p in sorted(PY_FIG_DIR.glob("*.png")): items.append((str(p), f"Python | {p.stem.replace('_', ' ').title()}")) for p in sorted(R_FIG_DIR.glob("*.png")): items.append((str(p), f"R | {p.stem.replace('_', ' ').title()}")) return items def _load_table_safe(path: Path) -> pd.DataFrame: try: if path.suffix == ".json": obj = _read_json(path) if isinstance(obj, dict): return pd.DataFrame([obj]) return pd.DataFrame(obj) return _read_csv(path) except Exception as e: return pd.DataFrame([{"error": str(e)}]) def refresh_gallery(): figures = _load_all_figures() idx = artifacts_index() table_choices = [] for scope in ("python", "r"): for name in idx[scope]["tables"]: table_choices.append(f"{scope}/{name}") default_df = pd.DataFrame() if table_choices: parts = table_choices[0].split("/", 1) base = PY_TAB_DIR if parts[0] == "python" else R_TAB_DIR default_df = _load_table_safe(base / parts[1]) return ( figures if figures else [], gr.update(choices=table_choices, value=table_choices[0] if table_choices else None), default_df, ) def on_table_select(choice: str): if not choice or "/" not in choice: return pd.DataFrame([{"hint": "Select a table above."}]) scope, name = choice.split("/", 1) base = {"python": PY_TAB_DIR, "r": R_TAB_DIR}.get(scope) if not base: return pd.DataFrame([{"error": f"Unknown scope: {scope}"}]) path = base / name if not path.exists(): return pd.DataFrame([{"error": f"File not found: {path}"}]) return _load_table_safe(path) # ========================================================= # KPI LOADER # ========================================================= def load_kpis() -> Dict[str, Any]: for candidate in [PY_TAB_DIR / "kpis.json"]: if candidate.exists(): try: return _read_json(candidate) except Exception: pass return {} # ========================================================= # AI DASHBOARD # ========================================================= DASHBOARD_SYSTEM = """\ You are an expert AI data analyst for RX12 — Airbnb Pricing Intelligence, a data science project analyzing Airbnb pricing across 10 European cities. The project has two analysis pipelines: - Python (EDA + Random Forest ML): charts on pricing by city, room type, weekend vs weekday, bedrooms, ML feature importance - R (Statistical Inference): t-tests, ANOVA, and multiple linear regression on pricing drivers AVAILABLE ARTIFACTS: {artifacts_json} KPI SUMMARY: {kpis_json} YOUR ROLE: 1. Answer data questions in 2-4 sentences using the KPIs and your Airbnb pricing knowledge. 2. At the END of your response, output EXACTLY ONE JSON block (fenced with ```json ... ```) that tells the dashboard which artifact to display: {{"show": "figure"|"table"|"none", "scope": "python"|"r", "filename": "..."}} ARTIFACT ROUTING RULES: - "price by city" or "avg price" → figure: 02_avg_price_by_city.png (python) - "rating" or "satisfaction by city" → figure: 01_avg_rating_by_city.png (python) - "room type distribution" → figure: 03_room_type_distribution.png (python) - "weekend" or "weekday" EDA chart → figure: 04_weekday_vs_weekend.png (python) - "bedrooms" → figure: 05_bedrooms_vs_price.png (python) - "booking window" → figure: 06_booking_window_satisfaction.png (python) - "feature importance" or "ML" or "random forest" or "what drives price" → figure: 07_ml_feature_importance.png (python) - "t-test" or "weekend premium significant" → table: ttest_weekday_vs_weekend.csv (r) - "anova" or "room type significant" → table: anova_room_type.csv (r) - "regression" or "coefficients" or "lm" → table: regression_coefficients.csv (r) - "model fit" or "R-squared" → table: regression_model_fit.csv (r) - "city day" or "heatmap" → table: city_day_avg_price.csv (r) - general overview / KPI → table: avg_price_by_city.csv (python) """ JSON_BLOCK_RE = re.compile(r"```json\s*(\{.*?\})\s*```", re.DOTALL) FALLBACK_JSON_RE = re.compile(r"\{[^{}]*\"show\"[^{}]*\}", re.DOTALL) def _parse_display_directive(text: str) -> Dict[str, str]: m = JSON_BLOCK_RE.search(text) if m: try: return json.loads(m.group(1)) except json.JSONDecodeError: pass m = FALLBACK_JSON_RE.search(text) if m: try: return json.loads(m.group(0)) except json.JSONDecodeError: pass return {"show": "none"} def _clean_response(text: str) -> str: return JSON_BLOCK_RE.sub("", text).strip() def ai_chat(user_msg: str, history: list): if not user_msg or not user_msg.strip(): return history, "", None, None idx = artifacts_index() kpis = load_kpis() if not LLM_ENABLED: reply, directive = _keyword_fallback(user_msg, idx, kpis) else: system = DASHBOARD_SYSTEM.format( artifacts_json=json.dumps(idx, indent=2), kpis_json=json.dumps(kpis, indent=2) if kpis else "(no KPIs yet — run pipeline first)", ) msgs = [{"role": "system", "content": system}] for entry in (history or [])[-6:]: msgs.append(entry) msgs.append({"role": "user", "content": user_msg}) try: r = llm_client.chat_completion( model=MODEL_NAME, messages=msgs, temperature=0.3, max_tokens=600, stream=False, ) raw = ( r["choices"][0]["message"]["content"] if isinstance(r, dict) else r.choices[0].message.content ) directive = _parse_display_directive(raw) reply = _clean_response(raw) except Exception as e: reply_fb, directive = _keyword_fallback(user_msg, idx, kpis) reply = f"LLM error: {e}\n\n{reply_fb}" # Resolve artifact paths fig_out = None tab_out = None show = directive.get("show", "none") scope = directive.get("scope", "") fname = directive.get("filename", "") if show == "figure" and scope and fname: base = {"python": PY_FIG_DIR, "r": R_FIG_DIR}.get(scope) if base and (base / fname).exists(): fig_out = str(base / fname) else: reply += f"\n\n*(Figure not found yet — run the pipeline first: {scope}/{fname})*" if show == "table" and scope and fname: base = {"python": PY_TAB_DIR, "r": R_TAB_DIR}.get(scope) if base and (base / fname).exists(): tab_out = _load_table_safe(base / fname) else: reply += f"\n\n*(Table not found yet — run the pipeline first: {scope}/{fname})*" new_history = (history or []) + [ {"role": "user", "content": user_msg}, {"role": "assistant", "content": reply}, ] return new_history, "", fig_out, tab_out def _keyword_fallback(msg: str, idx: Dict, kpis: Dict) -> Tuple[str, Dict]: ml = msg.lower() no_artifacts = not any(idx[s]["figures"] or idx[s]["tables"] for s in ("python", "r")) kpi_text = "" if kpis: kpi_text = ( f"Quick stats: **{kpis.get('n_cities','?')} cities**, " f"**{kpis.get('total_listings','?'):,} listings**, " f"avg price **€{kpis.get('avg_price_eur','?')}**, " f"weekend premium **+{kpis.get('weekend_premium_pct','?')}%**." ) if no_artifacts: return ( "No artifacts found yet. Run the pipeline in Tab 1 first, then ask me anything about the data.", {"show": "none"}, ) if any(w in ml for w in ["rating", "satisfaction", "review score"]): return (f"Here's the average guest rating by city. {kpi_text}", {"show": "figure", "scope": "python", "filename": "01_avg_rating_by_city.png"}) if any(w in ml for w in ["price city", "avg price", "price by city", "expensive"]): return (f"Here's the average price per city across Europe. {kpi_text}", {"show": "figure", "scope": "python", "filename": "02_avg_price_by_city.png"}) if any(w in ml for w in ["room type", "entire", "private", "distribution"]): return (f"Here's the room type distribution by city. {kpi_text}", {"show": "figure", "scope": "python", "filename": "03_room_type_distribution.png"}) if any(w in ml for w in ["weekend", "weekday", "day type"]): if any(w in ml for w in ["t-test", "significant", "p-value", "statistical"]): return ("The t-test confirms the weekend premium is statistically significant (p < 0.001).", {"show": "table", "scope": "r", "filename": "ttest_weekday_vs_weekend.csv"}) return (f"Here's the weekday vs weekend price comparison by city. {kpi_text}", {"show": "figure", "scope": "python", "filename": "04_weekday_vs_weekend.png"}) if any(w in ml for w in ["bedroom", "size", "capacity"]): return (f"Here's how number of bedrooms affects price across cities. {kpi_text}", {"show": "figure", "scope": "python", "filename": "05_bedrooms_vs_price.png"}) if any(w in ml for w in ["booking window", "advance", "lead time"]): return ("Here's how booking lead time relates to guest satisfaction.", {"show": "figure", "scope": "python", "filename": "06_booking_window_satisfaction.png"}) if any(w in ml for w in ["feature importance", "ml", "random forest", "drives price", "what drives", "model"]): return (f"Here are the key determinants of Airbnb price from our Random Forest model. {kpi_text}", {"show": "figure", "scope": "python", "filename": "07_ml_feature_importance.png"}) if any(w in ml for w in ["anova", "room type significant"]): return ("ANOVA confirms room type has a highly significant effect on price (p < 2e-16).", {"show": "table", "scope": "r", "filename": "anova_room_type.csv"}) if any(w in ml for w in ["regression", "coefficient", "lm result", "linear model"]): return ("Here are the regression coefficients from the R multiple linear regression.", {"show": "table", "scope": "r", "filename": "regression_coefficients.csv"}) if any(w in ml for w in ["r squared", "r2", "model fit", "accuracy"]): return ("Here's the overall model fit summary from R.", {"show": "table", "scope": "r", "filename": "regression_model_fit.csv"}) if any(w in ml for w in ["overview", "summary", "dashboard", "kpi"]): return (f"Dashboard overview: {kpi_text}\n\nAsk me about: prices, ratings, room types, weekend premiums, bedrooms, booking windows, feature importance, or statistical tests.", {"show": "table", "scope": "python", "filename": "avg_price_by_city.csv"}) return ( f"I can help analyze Airbnb pricing across Europe. {kpi_text}\n\n" "Try: **price by city**, **weekend premium**, **feature importance**, " "**regression results**, **ANOVA**, or **rating by city**.", {"show": "none"}, ) # ========================================================= # UI # ========================================================= ensure_dirs() body, .gradio-container { font-family: 'DM Sans', 'Segoe UI', sans-serif !important; background: #0f1117 !important; color: #e8eaf0 !important; } .gr-padded { background: #161b27 !important; } #rx12_header { background: linear-gradient(135deg, #1a2540 0%, #0f1117 60%, #1a1f2e 100%); border: 1px solid #2a3550; border-radius: 12px; padding: 28px 36px 20px 36px; margin-bottom: 12px; } #rx12_header h1 { font-size: 1.9rem !important; font-weight: 700 !important; color: #e2e8f0 !important; margin-bottom: 4px !important; letter-spacing: -0.5px; } #rx12_header p { color: #8892a4 !important; font-size: 0.92rem !important; } .tab-nav button { background: #161b27 !important; color: #8892a4 !important; border-radius: 8px 8px 0 0 !important; font-weight: 500 !important; font-size: 0.88rem !important; } .tab-nav button.selected { background: #1e2a42 !important; color: #60a5fa !important; border-bottom: 2px solid #60a5fa !important; } .step-card { background: #161b27; border: 1px solid #2a3550; border-radius: 10px; padding: 16px; } .step-card h3 { color: #60a5fa !important; margin-bottom: 6px; font-size: 0.95rem; } .step-card p { color: #8892a4 !important; font-size: 0.82rem; line-height: 1.5; } button.primary { background: linear-gradient(90deg, #2563eb, #3b82f6) !important; color: white !important; border-radius: 8px !important; font-weight: 600 !important; } button.secondary { background: #1e2a42 !important; color: #93c5fd !important; border: 1px solid #2a3a5a !important; border-radius: 8px !important; } """ with gr.Blocks(title="RX12 — Airbnb Pricing Intelligence", css=CSS) as demo: with gr.Group(elem_id="rx12_header"): gr.HTML("""

🏙️ RX12 — Airbnb Pricing Intelligence

European cities · EDA + Machine Learning (Python) · Statistical Inference (R) · AI Assistant

""") # =================================================== # TAB 1 — Pipeline Runner # =================================================== with gr.Tab("🔁 Pipeline Runner"): gr.HTML("""
Execute the three notebooks in sequence. Outputs are saved to artifacts/ and will appear in the Results Gallery.
""") with gr.Row(): with gr.Column(): gr.HTML("""

Step 1 — Data Creation

Runs FILE_1_Python.ipynb
Merges raw Airbnb CSVs, engineers features, synthesizes data, produces FINAL_merged_clean_synthetic.csv

""") btn_nb1 = gr.Button("▶ Run Step 1", variant="secondary") with gr.Column(): gr.HTML("""

Step 2a — Python Analysis

Runs FILE_2_Python.ipynb
8 EDA charts · Random Forest pricing model · Feature importance · Exports PNG figures + CSV tables to artifacts/py/

""") btn_nb2 = gr.Button("▶ Run Step 2a", variant="secondary") with gr.Column(): gr.HTML("""

Step 2b — R Statistical Inference

Runs FILE_3_R.ipynb
T-test (weekend premium) · ANOVA (room type) · Multiple Linear Regression · Exports CSV tables to artifacts/r/

""") btn_r = gr.Button("▶ Run Step 2b", variant="secondary") with gr.Row(): btn_all = gr.Button("🚀 Run Full Pipeline (Steps 1 → 2a → 2b)", variant="primary") run_log = gr.Textbox( label="Execution Log", lines=20, max_lines=35, interactive=False, placeholder="Pipeline output will appear here...", ) btn_nb1.click(run_datacreation, outputs=[run_log]) btn_nb2.click(run_pythonanalysis, outputs=[run_log]) btn_r.click(run_r, outputs=[run_log]) btn_all.click(run_full_pipeline, outputs=[run_log]) # =================================================== # TAB 2 — Results Gallery # =================================================== with gr.Tab("📊 Results Gallery"): gr.HTML("""
After running the pipeline, click Refresh to load all generated figures and tables.
""") refresh_btn = gr.Button("🔄 Refresh Gallery", variant="primary") gr.Markdown("#### Figures") gallery = gr.Gallery( label="EDA + ML Charts (Python) · Statistical (R)", columns=2, height=520, object_fit="contain", ) gr.Markdown("#### Tables") table_dropdown = gr.Dropdown( label="Select a table to preview", choices=[], interactive=True, ) table_display = gr.Dataframe(label="Table Preview", interactive=False) refresh_btn.click( refresh_gallery, outputs=[gallery, table_dropdown, table_display], ) table_dropdown.change( on_table_select, inputs=[table_dropdown], outputs=[table_display], ) # =================================================== # TAB 3 — AI Dashboard # =================================================== with gr.Tab("🤖 AI Dashboard"): gr.HTML(f"""
Ask questions about Airbnb pricing data. The AI will answer and display the relevant chart or table. {"● LLM active" if LLM_ENABLED else "⚠ No HF_API_KEY — using keyword matching. Add your key to Space secrets for full LLM mode."}
""") with gr.Row(equal_height=True): with gr.Column(scale=1): chatbot = gr.Chatbot(label="Conversation", height=400) user_input = gr.Textbox( label="Ask about Airbnb pricing data", placeholder="e.g. Which city has the highest average price?", lines=1, ) gr.Examples( examples=[ "Which city is most expensive?", "Show me the weekend vs weekday price difference", "What drives Airbnb prices? (feature importance)", "Is the weekend premium statistically significant?", "Show me the regression coefficients from R", "What does the ANOVA say about room types?", "Give me a dashboard overview", ], inputs=user_input, ) with gr.Column(scale=1): ai_figure = gr.Image(label="Chart", height=360) ai_table = gr.Dataframe(label="Data Table", interactive=False) user_input.submit( ai_chat, inputs=[user_input, chatbot], outputs=[chatbot, user_input, ai_figure, ai_table], ) demo.launch(allowed_paths=[str(BASE_DIR)])