Update spine_coder/spine_coder_core.py

spine_coder/spine_coder_core.py

@@ -1,13 +1,11 @@
 # spine_coder_core_pro.py
-# Vertebro FINAL-v2.2-PRO  (Block 1 / 3)
+# Vertebro FINAL-v2.2-PRO   (Block 1/4)
 # Core utilities, text normalization, level parsing, region classification
 
 import re, json
 from typing import Dict, Any, List, Tuple
 
-# ─────────────────────────────────────────────
-# TEXT NORMALIZATION
-# ─────────────────────────────────────────────
+# ── Text normalization ────────────────────────────────────────────────────────
 def _norm(s: str) -> str:
     """Clean unicode dashes, normalize spaces, lowercase text."""
     if not s:
@@ -21,9 +19,7 @@ def _has(text: str, pat: str) -> bool:
     return re.search(pat, text, flags=re.I) is not None
 
 
-# ─────────────────────────────────────────────
-# LEVEL AND REGION LOGIC
-# ─────────────────────────────────────────────
+# ── Levels & region ──────────────────────────────────────────────────────────
 _ORDER = ["C", "T", "L", "S"]
 _MAXNUM = {"C": 7, "T": 12, "L": 5, "S": 5}
 
@@ -33,16 +29,14 @@ def _level_sort_key(lv: str) -> Tuple[int, int]:
     num = int(re.sub(r"\D", "", lv) or 0)
     return (band, num)
 
-_SPAN = re.compile(r"\b([CTLS])\s?(\d{1,2})\s*[-–]\s*([CTLS])?\s?(\d{1,2})\b", re.I)
+_SPAN   = re.compile(r"\b([CTLS])\s?(\d{1,2})\s*[-–]\s*([CTLS])?\s?(\d{1,2})\b", re.I)
 _SINGLE = re.compile(r"\b([CTLS])\s?(\d{1,2})\b", re.I)
 
 def _expand_across_regions(p1: str, n1: int, p2: str, n2: int) -> List[str]:
     """Inclusive sequence from (p1,n1) to (p2,n2), crossing C/T/L/S boundaries."""
     p1, p2 = p1.upper(), p2.upper()
     out = []
-    i = _ORDER.index(p1)
-    j = _ORDER.index(p2)
-    r = i
+    r = _ORDER.index(p1)
     num = n1
     while True:
         out.append(f"{_ORDER[r]}{num}")
@@ -99,9 +93,7 @@ def _classify_region(levels: List[str]) -> str:
     return "mixed"
 
 
-# ─────────────────────────────────────────────
-# LATERALITY AND MODIFIERS
-# ─────────────────────────────────────────────
+# ── Laterality & modifiers (spine midline default) ───────────────────────────
 def _laterality_and_modifiers(note: str) -> Tuple[str, List[str]]:
     """Force midline for spine work; suppress invalid -50."""
     t = _norm(note)
@@ -115,25 +107,23 @@ def _laterality_and_modifiers(note: str) -> Tuple[str, List[str]]:
     return "midline", []
 
 
-# ─────────────────────────────────────────────
-# BASE FLAG PATTERNS
-# ─────────────────────────────────────────────
-FUSION_KW = r"\b(arthrodesis|fusion|t?lif|alif|plif|xlif|interbody\s+cage|peek\s+cage|structural\s+cage)\b"
-INSTR_KW  = r"\b(pedicle\s+screws?|lateral\s+mass\s+screws?|rods?|set\s+screws?|instrument(?:ed|ation))\b"
-NAV_KW    = r"\b(navigation|navigated|o-?arm|ziehm|3d\s+spin)\b"
-ALLO_KW   = r"\b(allograft|dbm|demineralized\s+bone\s+matrix)\b"
-AUTO_LOCAL_KW = r"\b(local\s+autograft|spinous\s+process\s+bone|lamina\s+bone\s+retained|morselized\s+autograft)\b"
-AUTO_SEP_KW   = r"\b(iliac\s+crest|separate\s+incision|rib\s+graft|harvested\s+from\s+iliac)\b"
+# ── Base flag patterns ───────────────────────────────────────────────────────
+FUSION_KW      = r"\b(arthrodesis|fusion|t?lif|alif|plif|xlif|interbody\s+cage|peek\s+cage|structural\s+cage)\b"
+INSTR_KW       = r"\b(pedicle\s+screws?|lateral\s+mass\s+screws?|rods?|set\s+screws?|instrument(?:ed|ation))\b"
+NAV_KW         = r"\b(navigation|navigated|o-?arm|ziehm|3d\s+spin)\b"
+ALLO_KW        = r"\b(allograft|dbm|demineralized\s+bone\s+matrix)\b"
+AUTO_LOCAL_KW  = r"\b(local\s+autograft|spinous\s+process\s+bone|lamina\s+bone\s+retained|morselized\s+autograft)\b"
+AUTO_SEP_KW    = r"\b(iliac\s+crest|separate\s+incision|rib\s+graft|harvested\s+from\s+iliac)\b"
 
-### --- END OF BLOCK 1 --- ###
+# ACDF recognition
+ACDF_KW      = r"(anterior cervical discectomy|acdf|smith[- ]?robinson|anterior cervical fusion)"
+ACDF_SUPPORT = r"(interbody|cage|peek|dbm|allograft|arthrodesis|plate|plating)"
 
 # spine_coder_core_pro.py
-# Vertebro FINAL-v2.2-PRO  (Block 2 / 3)
+# Vertebro FINAL-v2.2-PRO   (Block 2/4)
 # CPT reasoning logic and inference tree
 
-# ─────────────────────────────────────────────
-# INSTRUMENTATION SPAN HELPER
-# ─────────────────────────────────────────────
+# ── Instrumentation span helper ──────────────────────────────────────────────
 def _inst_code_by_span(span: int, anterior: bool) -> str:
     """Return instrumentation CPT code based on span length."""
     if span <= 1:
@@ -144,28 +134,21 @@ def _inst_code_by_span(span: int, anterior: bool) -> str:
         return "22842" if span <= 3 else ("22843" if span <= 7 else "22844")
 
 
-# ─────────────────────────────────────────────
-# CORE CPT INFERENCE ENGINE
-# ─────────────────────────────────────────────
+# ── Core CPT inference engine ────────────────────────────────────────────────
 def _infer_cpts(note: str, region: str, levels: List[str]) -> List[Dict[str, Any]]:
     """Return CPT dictionary list with rationale and confidence."""
     t = _norm(note)
     out: List[Dict[str, Any]] = []
     inters = _count_interspaces(levels)
     span = max(inters + 1, 2)
-    conf = 0.85
 
     def add(cpt: str, desc: str, rationale: str, cat: str, conf: float = 0.85, primary: bool = False):
         out.append({
-            "cpt": cpt,
-            "desc": desc,
-            "rationale": rationale,
-            "category": cat,
-            "confidence": round(conf, 2),
-            "primary": primary
+            "cpt": cpt, "desc": desc, "rationale": rationale,
+            "category": cat, "confidence": round(conf, 2), "primary": primary
         })
 
-    # ----- 1. Decompression -----
+    # 1) Decompression
     if _has(t, r"laminectomy|decompression|facetectomy|foraminotomy"):
         base_map = {"cervical": "63045", "thoracic": "63046", "lumbar": "63047"}
         base = base_map.get(region, "63047")
@@ -176,7 +159,7 @@ def _infer_cpts(note: str, region: str, levels: List[str]) -> List[Dict[str, Any
             add("63048", f"Each additional level ×{inters}",
                 "Multi-level decompression inferred.", "decompression add-on", 0.82)
 
-    # ----- 2. Fusion -----
+    # 2) Generic posterior/posterolateral fusion (will be suppressed if TLIF/PLIF hits)
     if _has(t, FUSION_KW):
         base_map = {"cervicothoracic": "22600", "cervical": "22600",
                     "thoracic": "22610", "lumbar": "22630", "lumbosacral": "22612"}
@@ -187,7 +170,43 @@ def _infer_cpts(note: str, region: str, levels: List[str]) -> List[Dict[str, Any
             add("22614", f"Each additional level ×{inters}",
                 "Multi-level fusion extension.", "fusion add-on", 0.84)
 
-    # ----- 3. Instrumentation -----
+    # 2b) ACDF (22551/22552) + optional anterior instrumentation
+    if _has(t, ACDF_KW) and _has(t, ACDF_SUPPORT) and region in {"cervical", "cervicothoracic"}:
+        cerv_levels = [lv for lv in levels if lv.startswith("C")]
+        cerv_inters = max(1, len(sorted(set(cerv_levels), key=_level_sort_key)) - 1) if cerv_levels else max(1, inters or 1)
+        add("22551", "ACDF, first interspace (includes discectomy)",
+            "Anterior cervical discectomy and fusion pattern detected.", "ACDF", 0.95, True)
+        if cerv_inters > 1:
+            add("22552", f"ACDF, each additional interspace ×{cerv_inters-1}",
+                "Multi-level ACDF.", "ACDF add-on", 0.90)
+        if _has(t, r"\bplate|plating|anterior plate\b"):
+            span_est = max(2, cerv_inters + 1)  # segments = interspaces + 1
+            a_code = _inst_code_by_span(span_est, anterior=True)
+            if a_code:
+                a_desc = {"22845":"Anterior instrumentation (2–3 segments)",
+                          "22846":"Anterior instrumentation (4–7 segments)",
+                          "22847":"Anterior instrumentation (8+ segments)"}[a_code]
+                add(a_code, a_desc, "Anterior plate present; span estimated from cervical levels.", "instrumentation", 0.84)
+
+    # 2c) TLIF/PLIF (22633/22634) with suppression of generic posterior fusion add-ons
+    tlif_plif_hit = (
+        _has(t, r"\b(tlif|plif|posterior interbody fusion)\b")
+        or (_has(t, r"facetectom(y|ies)|hemifacetectomy|transforaminal")
+            and _has(t, r"interbody (cage|device|spacer)|peek (cage|spacer)|titanium (cage|spacer)")
+            and _has(t, r"pedicle\s+screws?"))
+    )
+    if tlif_plif_hit and region in {"lumbar", "thoracic", "thoracolumbar", "lumbosacral"}:
+        n = max(1, inters or 1)
+        add("22633", "Posterior/posterolateral + posterior interbody fusion, single interspace",
+            "TLIF/PLIF construct detected (facetectomy + interbody device + pedicle screws).",
+            "TLIF/PLIF", 0.93, True)
+        if n > 1:
+            add("22634", f"Posterior interbody, each additional interspace ×{n-1}",
+                "Multi-level TLIF/PLIF.", "TLIF/PLIF add-on", 0.89)
+        # internal flag to suppress generic posterior fusion rows
+        out.append({"cpt":"__SUPPRESS_PF__", "desc":"", "rationale":"", "category":"internal", "confidence":1.0, "primary":False})
+
+    # 3) Posterior instrumentation (span-based)
     if _has(t, INSTR_KW):
         code = _inst_code_by_span(span, anterior=False)
         desc = {
@@ -195,36 +214,46 @@ def _infer_cpts(note: str, region: str, levels: List[str]) -> List[Dict[str, Any
             "22843": "Posterior segmental instrumentation (4–7 segments)",
             "22844": "Posterior segmental instrumentation (8+ segments)"
         }.get(code, "Posterior instrumentation")
-        add(code, desc, "Posterior instrumentation detected (pedicle screws/rods).",
-            "instrumentation", 0.88)
+        if code:
+            add(code, desc, "Posterior instrumentation detected (pedicle screws/rods).",
+                "instrumentation", 0.88)
 
-    # ----- 4. Navigation -----
+    # 4) Navigation
     if _has(t, NAV_KW):
         add("61783", "Intraoperative navigation (image-guided)",
             "Navigation terms detected (Ziehm/O-arm/3D spin).", "navigation", 0.82)
 
-    # ----- 5. Bone grafts -----
+    # 5) Bone grafts (local vs separate incision; structural preference)
     if _has(t, AUTO_SEP_KW):
         add("20937", "Autograft, separate incision",
-            "Iliac crest or separate-site autograft documented.", "graft", 0.8)
+            "Iliac crest or separate-site autograft documented.", "graft", 0.80)
     elif _has(t, AUTO_LOCAL_KW):
         add("20936", "Autograft, local (same incision)",
-            "Local autograft material retained.", "graft", 0.8)
-    if _has(t, ALLO_KW):
+            "Local autograft material retained.", "graft", 0.80)
+    if _has(t, r"structural allograft|fibular strut|strut graft|allograft spacer"):
+        # prefer 20931 over 20930 when structural is present
+        out = [r for r in out if r.get("cpt") != "20930"]
+        add("20931", "Structural allograft",
+            "Structural allograft (strut/cage) documented.", "graft", 0.82)
+    elif _has(t, ALLO_KW):
         add("20930", "Allograft, morselized / DBM",
-            "Allograft or DBM usage detected.", "graft", 0.8)
+            "Allograft or DBM usage detected.", "graft", 0.80)
 
-    # ----- 6. Hardware removal -----
+    # 6) Hardware removal
     if _has(t, r"remov(ed|al).*instrument"):
         add("22852", "Removal of posterior instrumentation",
             "Hardware removal/explantation documented.", "hardware_removal", 0.84, True)
 
+    # TLIF guard: remove generic posterior fusion rows if TLIF/PLIF detected
+    if any(r.get("cpt") == "__SUPPRESS_PF__" for r in out):
+        out = [r for r in out if r["category"] not in {"fusion", "fusion add-on"} and r.get("cpt") != "__SUPPRESS_PF__"]
+
     return out
 
+# spine_coder_core_pro.py
+# Vertebro FINAL-v2.2-PRO   (Block 3/4)
+# Specialty packs
 
-# ─────────────────────────────────────────────
-# SPECIALTY PACKS (Tumor / Deformity / Stimulator / Kyphoplasty / Revision)
-# ─────────────────────────────────────────────
 def _apply_specialty_packs(note: str, region: str, inters: int, levels: List[str]) -> List[Dict[str, Any]]:
     """Return CPTs from specialty packs layered onto base logic."""
     t = _norm(note)
@@ -232,12 +261,8 @@ def _apply_specialty_packs(note: str, region: str, inters: int, levels: List[str
 
     def add(cpt, desc, rationale, cat, conf=0.84, primary=False):
         extra.append({
-            "cpt": cpt,
-            "desc": desc,
-            "rationale": rationale,
-            "category": cat,
-            "confidence": round(conf, 2),
-            "primary": primary
+            "cpt": cpt, "desc": desc, "rationale": rationale,
+            "category": cat, "confidence": round(conf, 2), "primary": primary
         })
 
     # Tumor / corpectomy
@@ -251,13 +276,13 @@ def _apply_specialty_packs(note: str, region: str, inters: int, levels: List[str
                 "Multi-segment corpectomy.", "tumor/corpectomy add-on", 0.83)
 
     # Deformity / osteotomy
-    if _has(t, r"spo\b|smith[- ]?petersen|posterior column osteotomy|osteotomy"):
+    if _has(t, r"\bspo\b|smith[- ]?petersen|posterior column osteotomy|osteotomy"):
         base = {"cervical": "22210", "thoracic": "22214", "lumbar": "22206"}.get(region, "22206")
         add(base, "Posterior column osteotomy, first level",
             "Deformity correction (SPO/osteotomy) documented.", "deformity", 0.84, True)
         if inters > 0:
             add(str(int(base) + 1), f"Each additional level ×{inters}",
-                "Multi-level osteotomy.", "deformity add-on", 0.8)
+                "Multi-level osteotomy.", "deformity add-on", 0.80)
 
     # Stimulator
     if _has(t, r"stimulator|paddle lead|scs"):
@@ -275,7 +300,7 @@ def _apply_specialty_packs(note: str, region: str, inters: int, levels: List[str
             "Kyphoplasty/vertebroplasty terms found.", "augmentation", 0.84, True)
         if inters > 0:
             add("22515", f"Each additional level ×{inters}",
-                "Multi-level augmentation.", "augmentation add-on", 0.8)
+                "Multi-level augmentation.", "augmentation add-on", 0.80)
 
     # Revision instrumentation
     if _has(t, r"revision of instrumentation|reinsert|remove and replace"):
@@ -289,18 +314,10 @@ def _apply_specialty_packs(note: str, region: str, inters: int, levels: List[str
 
     return extra
 
-### --- END OF BLOCK 2 --- ###
-
 # spine_coder_core_pro.py
-# Vertebro FINAL-v2.2-PRO  (Block 3 / 3)
-# Output builder, rationale composer, and main entrypoint
-
-# ─────────────────────────────────────────────
-# MAIN OUTPUT ASSEMBLER
-# ─────────────────────────────────────────────
-# ─────────────────────────────────────────────
-# MAIN OUTPUT ASSEMBLER (v2.2-PRO with top_k + BC alias)
-# ─────────────────────────────────────────────
+# Vertebro FINAL-v2.2-PRO   (Block 4/4)
+# Output builder & main entrypoint with top_k + backward-compat alias
+
 def vertebro_infer(note: str, payer: str = "Medicare", top_k: int = 10) -> Dict[str, Any]:
     """Master inference wrapper — merges base logic + specialty packs.
     `top_k` kept for HF UI/backward compatibility; rows are truncated to top_k after scoring.
@@ -311,12 +328,12 @@ def vertebro_infer(note: str, payer: str = "Medicare", top_k: int = 10) -> Dict[
     inters = _count_interspaces(levels)
     laterality, mods = _laterality_and_modifiers(t)
 
-    # --- run inference ---
-    base_rows = _infer_cpts(t, region, levels)
+    # Inference
+    base_rows  = _infer_cpts(t, region, levels)
     extra_rows = _apply_specialty_packs(t, region, inters, levels)
     rows = base_rows + extra_rows
 
-    # --- merge duplicates and compute confidence max + rationale concat ---
+    # Merge duplicates (max confidence, concat rationale)
     merged: Dict[tuple, Dict[str, Any]] = {}
     for r in rows:
         key = (r["cpt"], r["category"])
@@ -324,28 +341,28 @@ def vertebro_infer(note: str, payer: str = "Medicare", top_k: int = 10) -> Dict[
             merged[key] = r.copy()
         else:
             merged[key]["confidence"] = round(max(merged[key]["confidence"], r["confidence"]), 2)
-            merged[key]["rationale"] = (merged[key]["rationale"] + " / " + r["rationale"]).strip()
+            merged[key]["rationale"]  = (merged[key]["rationale"] + " / " + r["rationale"]).strip()
 
     rows = list(merged.values())
     rows.sort(key=lambda r: (-r.get("confidence", 0.0), r.get("primary", False) is False, r["cpt"]))
 
-    # --- flag map (tech) ---
+    # Tech flags
     flags_map = {
         "microscope": _has(t, r"\bmicroscope\b|microdissection"),
-        "nav": _has(t, r"\bnavigation\b|o-?arm|ziehm|3d\s+spin"),
+        "nav":        _has(t, r"\bnavigation\b|o-?arm|ziehm|3d\s+spin"),
         "io_monitor": _has(t, r"\bneuromonitor|ssep|emg|monitoring\b"),
-        "fluoro": _has(t, r"\bfluoro|c[- ]?arm|fluoroscop"),
+        "fluoro":     _has(t, r"\bfluoro|c[- ]?arm|fluoroscop"),
     }
     tech_flags = [k for k, v in flags_map.items() if v]
 
-    # --- attach tech info to rationale for primary row(s) ---
+    # Attach tech info to primary rows
     if tech_flags:
         tech_txt = f" (Tech: {', '.join(tech_flags)})"
         for r in rows:
             if r.get("primary", False):
                 r["rationale"] = (r.get("rationale", "") + tech_txt).strip()
 
-    # --- Top-K cap (HF UI slider compatibility) ---
+    # Top-K cap
     try:
         k = int(top_k)
         if k > 0:
@@ -353,7 +370,7 @@ def vertebro_infer(note: str, payer: str = "Medicare", top_k: int = 10) -> Dict[
     except Exception:
         pass
 
-    # --- build final JSON output ---
+    # Output
     return {
         "payer": payer,
         "region": region,
@@ -372,24 +389,15 @@ def vertebro_infer(note: str, payer: str = "Medicare", top_k: int = 10) -> Dict[
 suggest_with_cpt_billing = vertebro_infer
 
 
-
-# ─────────────────────────────────────────────
-# CLI / HF Space TEST ENTRYPOINT
-# ─────────────────────────────────────────────
+# CLI / HF Space quick test
 if __name__ == "__main__":
     sample = """
     Preoperative Diagnosis: C3–C6 stenosis with myelopathy.
     Procedure:
-    1. C3–C6 laminectomy and medial facetectomies.
-    2. C2–T1 posterolateral arthrodesis.
-    3. C2–T1 posterior instrumentation.
-    4. Use of navigation and intra-operative fluoroscopy.
-    5. Local autograft and allograft placed.
+      1. C3–C6 laminectomy and medial facetectomies.
+      2. C2–T1 posterolateral arthrodesis.
+      3. C2–T1 posterior instrumentation with pedicle screws and rods.
+      4. Use of navigation and intra-operative fluoroscopy.
+      5. Local autograft and allograft placed.
     """
-    result = vertebro_infer(sample)
-    print(json.dumps(result, indent=2))
-
-
-# Backward compatibility for old app versions
-suggest_with_cpt_billing = vertebro_infer
-
+    print(json.dumps(vertebro_infer(sample), indent=2))