Update app.py

app.py

@@ -1,11 +1,12 @@
 # app.py — Spine Coder (Chatbot + Feedback + Session Logs) — Gradio 4.x
 # ------------------------------------------------------------------------------
 # FINAL-v2.1 UI build:
+# - Bulletproof core loader (module + direct file path) to avoid stale caches
+# - Startup PROBE print to logs + in-UI Diagnostics/Probe button
 # - Clean CPT table (no per-row modifier columns)
 # - Case Modifiers panel (-22, -50, -53, -80, -82, ...)
-# - Build tag + Core file path shown in UI chips and logged
-# - Force-import + reload of core to avoid stale cache
-# - Structured logs/export, graceful errors, modern Gradio usage
+# - Build tag + Core file path in UI chips and logs
+# - Structured logs/export, modern Gradio usage
 # ------------------------------------------------------------------------------
 
 import os
@@ -19,49 +20,63 @@ 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
+# ==== Bulletproof Core Import ==================================================
+import importlib, importlib.util, importlib.machinery, sys as __sys, os as __os, inspect as __inspect
 
-# Try normal imports (still useful for local dev)
-try:
-    from spine_coder.spine_coder.spine_coder_core import suggest_with_cpt_billing as _maybe_suggest
-except Exception:
-    try:
-        from spine_coder.spine_coder_core import suggest_with_cpt_billing as _maybe_suggest
-    except Exception:
-        _maybe_suggest = None
+__sys.path.insert(0, __os.path.abspath("."))  # ensure repo root is on sys.path
 
-# --- Force-load the v2.1 core and avoid stale caches (HF Space sometimes caches modules) ---
-import importlib
+def _load_core_from_file(path: str):
+    """Load spine_coder_core.py directly from a given path (bypass module cache)."""
+    if not __os.path.isabs(path):
+        path = __os.path.abspath(path)
+    if not __os.path.exists(path):
+        return None
+    name = "spine_coder_core_dynamic"
+    loader = importlib.machinery.SourceFileLoader(name, path)
+    spec = importlib.util.spec_from_loader(name, loader)
+    mod = importlib.util.module_from_spec(spec)
+    loader.exec_module(mod)
+    return importlib.reload(mod)
 
 def _force_import_core():
-    mod = None
-    try:
-        # Preferred package layout
-        mod = importlib.import_module("spine_coder.spine_coder.spine_coder_core")
-    except Exception:
-        # Fallback to flat layout
-        mod = importlib.import_module("spine_coder.spine_coder_core")
-    # Hard reload to bust any stale cache
-    mod = importlib.reload(mod)
-    return mod
-
-try:
-    _core = _force_import_core()
-    suggest_with_cpt_billing = _core.suggest_with_cpt_billing
-except Exception:
-    # Last resort: use whatever we imported earlier (if any)
-    if _maybe_suggest is None:
-        raise
-    suggest_with_cpt_billing = _maybe_suggest
+    """Try package modules first; then known file paths. Reload to bust caches."""
+    # 1) Try module layouts
+    for modname in [
+        "spine_coder.spine_coder.spine_coder_core",
+        "spine_coder.spine_coder_core",
+    ]:
+        try:
+            mod = importlib.import_module(modname)
+            return importlib.reload(mod)
+        except Exception:
+            pass
+    # 2) Try likely file paths (repo variants)
+    for rel in [
+        "spine_coder/spine_coder/spine_coder_core.py",
+        "spine_coder/spine_coder_core.py",
+        "spine_coder_core.py",
+    ]:
+        mod = _load_core_from_file(rel)
+        if mod:
+            return mod
+    raise ImportError("Unable to locate spine_coder_core.py via modules or file paths.")
+
+_core = _force_import_core()
+suggest_with_cpt_billing = _core.suggest_with_cpt_billing
 
-# Prove which file is actually loaded (visible in Space logs)
 try:
-    print("[CORE] loaded from:", _inspect.getsourcefile(suggest_with_cpt_billing))
+    print("[CORE] loaded from:", __inspect.getsourcefile(suggest_with_cpt_billing))
 except Exception:
     pass
 
+# ---- One-time startup probe (prints to Space logs) ----------------------------
+try:
+    _probe_note = "Discectomies at C4–C5, C5–C6, and C6–C7. Interbody cages and anterior plate spanning C4–C7."
+    _probe = suggest_with_cpt_billing(_probe_note, payer="Medicare", top_k=5)
+    print("[PROBE] build:", _probe.get("build"))
+    print("[PROBE] first_suggestion:", (_probe.get("suggestions") or [{}])[0])
+except Exception as e:
+    print("[PROBE] failed:", e)
 
 # ==== Config ==================================================================
 os.environ.setdefault("GRADIO_ANALYTICS_ENABLED", "False")
@@ -94,18 +109,15 @@ def export_session(session_id: str) -> str:
         json.dump(data, f, indent=2, ensure_ascii=False)
     return out_path
 
-
-# ==== Helpers =================================================================
+# ==== UI Helpers ==============================================================
 
 def _core_path() -> str:
-    """Absolute path of the loaded core, for UI + logs."""
     try:
-        p = _inspect.getsourcefile(suggest_with_cpt_billing) or "unknown"
-        return p
+        return __inspect.getsourcefile(suggest_with_cpt_billing) or "unknown"
     except Exception:
         return "unknown"
 
-# NOTE: Modifiers columns removed from the CPT table on purpose.
+# NOTE: Per-row modifiers removed from the CPT table on purpose.
 SUGG_COLS = [
     "CPT", "Description", "Rationale",
     "Confidence", "Primary", "Category", "Laterality", "Units"
@@ -210,7 +222,6 @@ def _summary_md(meta: Dict[str, Any]) -> str:
     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']}`")
-    # Show only the basename of the core file in UI chips (full path is in logs)
     try:
         core_base = os.path.basename(meta.get("core_path","")) if meta.get("core_path") else ""
         if core_base:
@@ -222,7 +233,6 @@ def _summary_md(meta: Dict[str, Any]) -> str:
 def new_session() -> str:
     return str(uuid.uuid4())[:8]
 
-
 # ==== Core actions ============================================================
 
 def run_inference(note: str, payer: str, top_k: int, session_id: str):
@@ -234,7 +244,6 @@ def run_inference(note: str, payer: str, top_k: int, session_id: str):
         )
 
     _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:
@@ -262,7 +271,6 @@ def run_inference(note: str, payer: str, top_k: int, session_id: str):
         "event": "response",
         "meta": {
             **meta,
-            # compact string for quick scan in logs
             "case_modifiers": ", ".join([f"-{m}" for m in [cm.get("modifier","") for cm in (result.get("case_modifiers") or [])] if m]) or ""
         },
         "rows_len": int(len(sugg_df) if hasattr(sugg_df, "__len__") else 0)
@@ -289,6 +297,46 @@ def on_clear():
         new_session()
     )
 
+# ==== Diagnostics / Probe =====================================================
+
+def run_probe(session_id: str) -> Tuple[str, str]:
+    """Run a live diagnostic: show core path & build and 3 smoke tests."""
+    info_lines = []
+    try:
+        core_path = _core_path()
+        # Three canonical tests
+        tests = [
+            ("Implicit TLIF",
+             "Left facetectomy L4–L5 with PEEK interbody cage and pedicle screws; rods secured.",
+             ["22633"],  # expect
+            ),
+            ("ACDF chain w/ plate",
+             "Discectomies at C4–C5, C5–C6, and C6–C7. Interbody cages and anterior plate spanning C4–C7.",
+             ["22551","22552","22846"],
+            ),
+            ("Exposure-only",
+             "Anterior exposure of L4–S1 performed by vascular surgeon for access. No fusion performed.",
+             ["00000"],
+            ),
+        ]
+        # Run once to get build
+        probe = suggest_with_cpt_billing(tests[1][1], payer="Medicare", top_k=10)
+        build = probe.get("build","")
+        info_lines.append(f"**Core path:** `{core_path}`  \n**Build:** `{build}`")
+
+        # Execute each test
+        for label, note, expect_codes in tests:
+            res = suggest_with_cpt_billing(note, payer="Medicare", top_k=10)
+            codes = [s.get("cpt") for s in (res.get("suggestions") or [])]
+            ok = all(any(ec == c for c in codes) for ec in expect_codes)
+            info_lines.append(f"- **{label}** → codes: `{', '.join(codes) or '∅'}` — **{'PASS' if ok else 'CHECK'}**")
+        md = "\n".join(info_lines)
+        _append_log(session_id, {"event":"probe","details": md})
+        return md, ""
+    except Exception as e:
+        tb = traceback.format_exc()
+        _append_log(session_id, {"event":"probe_error","error":repr(e),"traceback":tb})
+        return "", f"⚠️ Probe failed: {e}"
 
 # ==== Examples ================================================================
 
@@ -453,6 +501,10 @@ with gr.Blocks(theme=THEME, css=CUSTOM_CSS, title="Spine Coder — CPT Billing")
                 export_btn = gr.Button("Export Session as JSON")
                 export_file = gr.File(label="Download", interactive=False)
 
+            with gr.Accordion("Diagnostics", open=False):
+                probe_btn = gr.Button("Run Probe (core path + 3 tests)")
+                probe_md = gr.Markdown("")
+
         # Results
         with gr.Column(scale=7):
             gr.Markdown("#### Results")
@@ -504,6 +556,8 @@ with gr.Blocks(theme=THEME, css=CUSTOM_CSS, title="Spine Coder — CPT Billing")
     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])
 
+    probe_btn.click(run_probe, inputs=[session_id], outputs=[probe_md, warn_md])
+
 if __name__ == "__main__":
     # If running locally: set server_name to 0.0.0.0 for external access; Space ignores.
     demo.launch()