Update app.py

app.py

@@ -1,365 +0,0 @@
-# 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
-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;
-}
-.table-wrap .dataframe {
-    font-size: 14px;
-    width: 100% !important;
-    table-layout: fixed !important;  /* distribute width evenly */
-    border-collapse: collapse;
-}
-.table-wrap .dataframe th,
-.table-wrap .dataframe td {
-    white-space: normal !important;  /* allow wrapping */
-    word-wrap: break-word !important;
-    max-width: 240px !important;     /* readable column width */
-    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,
-                height=400,                 # taller, easier to read
-                elem_classes=["table-wrap"],
-            )
-            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()