Update app.py

app.py

@@ -0,0 +1,370 @@
+# app.py — Spine Coder (Chatbot + Feedback + Session Logs) — Gradio 4.x
+# ------------------------------------------------------------------------------
+# Stable build: FINAL-v2.0 core wired, transitional regions, flags, laterality,
+# structured logs, graceful errors, and a clean UI.
+# ------------------------------------------------------------------------------
+
+import os
+import json
+import uuid
+import pathlib
+import traceback
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Tuple
+
+import pandas as pd
+import gradio as gr
+
+# ---- Core import --------------------------------------------------------------
+import os as _os, sys as _sys, inspect as _inspect
+_sys.path.insert(0, _os.path.abspath("."))  # ensure repo root is first on sys.path
+
+try:
+    # If your package layout is spine_coder/spine_coder/spine_coder_core.py
+    from spine_coder.spine_coder.spine_coder_core import suggest_with_cpt_billing
+except ImportError:
+    # Fallback to flat layout if you later flatten the folder tree
+    from spine_coder.spine_coder_core import suggest_with_cpt_billing
+
+# Prove which file is actually loaded (shows in Space logs)
+try:
+    print("[CORE] loaded from:", _inspect.getsourcefile(suggest_with_cpt_billing))
+except Exception:
+    pass
+
+
+# ==== Config ==================================================================
+os.environ.setdefault("GRADIO_ANALYTICS_ENABLED", "False")
+DEBUG = os.environ.get("DEBUG", "0") == "1"
+PAYER_CHOICES = ["Medicare", "BCBS", "Aetna", "Cigna", "UnitedHealthcare", "Other"]
+
+# ==== Logging =================================================================
+LOG_DIR = os.environ.get("LOG_DIR", "logs")
+pathlib.Path(LOG_DIR).mkdir(parents=True, exist_ok=True)
+
+def _log_path(session_id: str) -> str:
+    return os.path.join(LOG_DIR, f"{session_id}.jsonl")
+
+def _utcnow_iso() -> str:
+    return datetime.now(timezone.utc).isoformat(timespec="seconds")
+
+def _append_log(session_id: str, entry: Dict[str, Any]) -> None:
+    entry = {"ts": _utcnow_iso(), "session_id": session_id, **entry}
+    with open(_log_path(session_id), "a", encoding="utf-8") as f:
+        f.write(json.dumps(entry, ensure_ascii=False) + "\n")
+
+def export_session(session_id: str) -> str:
+    src = _log_path(session_id)
+    data: List[Dict[str, Any]] = []
+    if os.path.exists(src):
+        with open(src, "r", encoding="utf-8") as f:
+            data = [json.loads(l) for l in f if l.strip()]
+    out_path = os.path.join(LOG_DIR, f"export_{session_id}.json")
+    with open(out_path, "w", encoding="utf-8") as f:
+        json.dump(data, f, indent=2, ensure_ascii=False)
+    return out_path
+
+# ==== UI helpers ==============================================================
+
+TABLE_COLS = [
+    "CPT","Modifier","Modifiers","Description","Rationale",
+    "Confidence","Primary","Category","Laterality","Units"
+]
+EMPTY_DF = pd.DataFrame(columns=TABLE_COLS)
+
+def _coalesce_suggestions(result: Dict[str, Any]) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]:
+    """
+    Return (rows_for_table, meta_badges) from core result.
+    Robust to minor schema differences (list vs dict for levels/flags).
+    """
+    if not isinstance(result, dict):
+        return [], {}
+
+    # ---------- rows ----------
+    sugg = result.get("suggestions") or []
+    rows: List[Dict[str, Any]] = []
+    case_lat = (result.get("laterality") or "").strip().lower()
+
+    for s in sugg:
+        if not isinstance(s, dict):
+            continue
+
+        # derive row laterality from explicit field, LT/RT modifiers, or case-level fallback
+        row_lat = (s.get("laterality") or "").strip().lower()
+        mods = s.get("modifiers", []) or []
+        if not row_lat:
+            if isinstance(mods, list) and "LT" in mods:
+                row_lat = "left"
+            elif isinstance(mods, list) and "RT" in mods:
+                row_lat = "right"
+            elif case_lat in ("left", "right", "bilateral"):
+                row_lat = case_lat
+
+        conf_val = s.get("confidence")
+        conf_out = round(float(conf_val), 2) if isinstance(conf_val, (int, float)) else (conf_val or "")
+
+        rows.append({
+            "CPT": s.get("cpt", ""),
+            "Modifier": s.get("modifier") or "",
+            "Modifiers": ", ".join(mods) if isinstance(mods, list) else (s.get("modifiers") or ""),
+            "Description": s.get("desc", ""),
+            "Rationale": s.get("rationale", ""),
+            "Confidence": conf_out,
+            "Primary": "✓" if s.get("primary") else "",
+            "Category": s.get("category", ""),
+            "Laterality": row_lat,
+            "Units": s.get("units", 1),
+        })
+
+    # ---------- normalize levels ----------
+    segs: List[str] = []
+    inters_list: List[str] = []
+    lvl_lat: str = ""
+
+    levels_obj = result.get("levels")
+    if isinstance(levels_obj, dict):
+        segs = list(levels_obj.get("segments") or [])
+        inters_list = list(levels_obj.get("interspaces") or [])
+        lvl_lat = levels_obj.get("laterality", "") or ""
+    elif isinstance(levels_obj, list):
+        segs = [str(x) for x in levels_obj]
+
+    # ---------- normalize flags ----------
+    flags_obj = result.get("flags")
+    if isinstance(flags_obj, list):
+        flags_list = [str(x) for x in flags_obj]
+    elif isinstance(flags_obj, dict):
+        flags_list = [k for k, v in flags_obj.items() if v]
+    else:
+        flags_list = []
+
+    # ---------- meta ----------
+    meta = {
+        "payer": result.get("payer", ""),
+        "region": result.get("region", ""),
+        "laterality": result.get("laterality", "") or lvl_lat,
+        "levels_segments": ", ".join(segs),
+        # Prefer numeric estimate if present; else join the interspaces list
+        "levels_interspaces": (
+            str(result.get("interspaces_est", "")) if "interspaces_est" in result
+            else ", ".join(inters_list)
+        ),
+        "flags": ", ".join(sorted(flags_list)),
+        "build": result.get("build", ""),
+        "mode": result.get("mode", ""),
+    }
+    return rows, meta
+
+def _summary_md(meta: Dict[str, Any]) -> str:
+    chips = []
+    if meta.get("region"): chips.append(f"`Region: {meta['region']}`")
+    if meta.get("laterality"): chips.append(f"`Laterality: {meta['laterality']}`")
+    if meta.get("levels_segments"): chips.append(f"`Segments: {meta['levels_segments']}`")
+    if meta.get("levels_interspaces"): chips.append(f"`Interspaces: {meta['levels_interspaces']}`")
+    if meta.get("flags"): chips.append(f"`Flags: {meta['flags']}`")
+    if meta.get("build"): chips.append(f"`Build: {meta['build']}`")
+    if meta.get("mode"): chips.append(f"`Mode: {meta['mode']}`")
+    return " ".join(chips) if chips else "—"
+
+def new_session() -> str:
+    return str(uuid.uuid4())[:8]
+
+# ==== Core actions ============================================================
+
+def run_inference(note: str, payer: str, top_k: int, session_id: str):
+    if not note.strip():
+        return EMPTY_DF, "—", "", "", session_id
+
+    _append_log(session_id, {"event": "request", "payer": payer, "top_k": top_k, "note": note})
+    warn_text = ""
+    try:
+        result = suggest_with_cpt_billing(note=note, payer=payer, top_k=top_k)
+        if DEBUG:
+            print("[DEBUG] build/region/laterality/flags:",
+                  result.get("build"), result.get("region"),
+                  result.get("laterality"), result.get("flags"))
+    except Exception as e:
+        tb = traceback.format_exc()
+        _append_log(session_id, {"event": "error", "error": repr(e), "traceback": tb})
+        warn = f"⚠️ Error: {e}"
+        if DEBUG:
+            warn += f"\n\n```traceback\n{tb}\n```"
+        return EMPTY_DF, "—", "", warn, session_id
+
+    rows, meta = _coalesce_suggestions(result)
+    df = EMPTY_DF if not rows else pd.DataFrame(rows)
+    if not df.empty:
+        # ensure consistent column order
+        for col in TABLE_COLS:
+            if col not in df.columns:
+                df[col] = ""
+        df = df[TABLE_COLS]
+
+    summary = _summary_md(meta)
+    json_pretty = json.dumps(result, indent=2, ensure_ascii=False)
+
+    _append_log(session_id, {"event": "response", "meta": meta, "rows_len": len(rows)})
+    return df, summary, json_pretty, warn_text, session_id
+
+def record_feedback(session_id: str, vote: str, text: str):
+    if not vote and not text:
+        return "Please choose 👍/👎 or add a short note."
+    _append_log(session_id, {"event": "feedback", "vote": vote, "text": text})
+    return "Thanks! Your feedback was recorded."
+
+def do_export(session_id: str):
+    path = export_session(session_id)
+    _append_log(session_id, {"event": "export", "path": path})
+    return path
+
+def on_clear():
+    return "", "Medicare", 10, EMPTY_DF, "—", "", "", new_session()
+
+# ==== Examples ================================================================
+
+EXAMPLES = [
+    ["Left-sided TLIF L4–L5 with pedicle screws and interbody cage; posterolateral fusion performed. Navigation used.", "Medicare", 10],
+    ["ACDF C5–C6 with PEEK cage and anterior plate; microscope and neuromonitoring used.", "Medicare", 10],
+    ["Posterior cervical foraminotomy right C6–C7; no fusion or instrumentation.", "Medicare", 10],
+    ["ALIF L5–S1 with structural allograft; non-segmental instrumentation placed.", "Medicare", 10],
+    ["Removal of posterior segmental instrumentation T10–L2; no new hardware placed.", "Medicare", 10],
+]
+
+# ==== Theme / CSS =============================================================
+
+THEME = gr.themes.Soft(
+    primary_hue="indigo",
+    secondary_hue="blue",
+    neutral_hue="slate",
+).set(
+    body_text_color="#0f172a",
+    background_fill_primary="#ffffff",
+    button_primary_background_fill="#4f46e5",
+    input_background_fill="#ffffff",
+)
+
+# Wider, more readable table (no Dataframe.height arg; do it via CSS)
+CUSTOM_CSS = """
+:root { --radius-lg: 16px; }
+.gradio-container { font-family: ui-sans-serif, system-ui, -apple-system; }
+.header-card {
+    border-radius: 18px;
+    padding: 18px;
+    border: 1px solid #e5e7eb;
+    background: linear-gradient(180deg,#fafafa,#ffffff);
+}
+.badge-row code {
+    margin-right: 8px;
+    border-radius: 12px;
+    padding: 2px 8px;
+    background: #f1f5f9;
+}
+
+/* Make the dataframe area taller & scrollable */
+.table-wrap { max-height: 420px; overflow: auto; }
+
+/* Improve readability inside the table */
+.table-wrap .dataframe {
+    font-size: 14px;
+    width: 100% !important;
+    table-layout: fixed !important;
+    border-collapse: collapse;
+}
+.table-wrap .dataframe th,
+.table-wrap .dataframe td {
+    white-space: normal !important;
+    word-wrap: break-word !important;
+    max-width: 240px !important;
+    text-align: left;
+    vertical-align: top;
+    padding: 6px 8px;
+}
+
+.footer-note { color:#64748b; font-size:12px; }
+"""
+
+# ==== App Layout ==============================================================
+with gr.Blocks(theme=THEME, css=CUSTOM_CSS, title="Spine Coder — CPT Billing") as demo:
+    session_id = gr.State(new_session())
+
+    with gr.Row():
+        with gr.Column():
+            gr.Markdown("### 🦴 Spine Coder — CPT Billing & Operative Note NLP")
+            gr.Markdown(
+                '<div class="header-card">Structured CPT suggestions from spine operative notes — with payer-aware '
+                'modifiers, laterality detection, and rationales.<br/>'
+                '<span class="footer-note">No PHI is stored; inputs are session-scoped and ephemeral.</span></div>'
+            )
+
+    with gr.Row():
+        # Inputs
+        with gr.Column(scale=5):
+            note_in = gr.Textbox(
+                label="Operative Note",
+                placeholder="Paste an operative note here…",
+                lines=14,
+                autofocus=True,
+            )
+            with gr.Row():
+                payer_dd = gr.Dropdown(PAYER_CHOICES, value="Medicare", label="Payer")
+                topk = gr.Slider(1, 15, value=10, step=1, label="Top-K suggestions")
+            with gr.Row():
+                run_btn = gr.Button("Analyze Note", variant="primary")
+                clear_btn = gr.Button("Clear")
+
+            with gr.Accordion("Quick Examples", open=False):
+                gr.Examples(
+                    examples=EXAMPLES,
+                    inputs=[note_in, payer_dd, topk],
+                    label="Click a row to load an example"
+                )
+
+            with gr.Accordion("Feedback", open=False):
+                fb_choice = gr.Radio(choices=["👍", "👎"], label="Was this helpful?")
+                fb_text = gr.Textbox(label="Optional comment", lines=2, placeholder="Tell us what worked or what missed…")
+                fb_submit = gr.Button("Submit Feedback")
+                fb_status = gr.Markdown("")
+
+            with gr.Accordion("Session", open=False):
+                sid_show = gr.Textbox(label="Session ID", value="", interactive=False)
+                export_btn = gr.Button("Export Session as JSON")
+                export_file = gr.File(label="Download", interactive=False)
+
+        # Results
+        with gr.Column(scale=6):
+            gr.Markdown("#### Results")
+            summary_md = gr.Markdown("—", elem_classes=["badge-row"])
+            table = gr.Dataframe(
+                value=EMPTY_DF,
+                label="Suggestions",
+                interactive=False,
+                row_count=(0, "dynamic"),
+                wrap=True,
+                elem_classes=["table-wrap"],  # CSS above makes it tall/scrollable
+            )
+            with gr.Accordion("Raw JSON", open=False):
+                json_out = gr.Code(language="json", value="", interactive=False)
+            warn_md = gr.Markdown("")
+
+    # ---- Events / Wiring ----
+    def _on_load():
+        sid = new_session()
+        return sid, sid
+
+    demo.load(_on_load, outputs=[session_id, sid_show])
+
+    run_inputs = [note_in, payer_dd, topk, session_id]
+    run_outputs = [table, summary_md, json_out, warn_md, session_id]
+
+    run_btn.click(run_inference, inputs=run_inputs, outputs=run_outputs)
+    note_in.submit(run_inference, inputs=run_inputs, outputs=run_outputs)
+
+    clear_btn.click(on_clear, outputs=[note_in, payer_dd, topk, table, summary_md, json_out, warn_md, session_id])
+
+    fb_submit.click(record_feedback, inputs=[session_id, fb_choice, fb_text], outputs=fb_status)
+
+    export_btn.click(do_export, inputs=[session_id], outputs=[export_file])
+
+if __name__ == "__main__":
+    # If running locally: set server_name to 0.0.0.0 for external access; Space ignores.
+    demo.launch()