Update spine_coder/spine_coder_core.py

spine_coder/spine_coder_core.py

@@ -1,72 +1,94 @@
-# spine_coder_core.py
-# Vertebro FINAL-v2.2 — Production Hybrid Engine
+# spine_coder_core_pro.py
+# Vertebro FINAL-v2.2-PRO  (Block 1 / 3)
+# Core utilities, text normalization, level parsing, region classification
 
 import re, json
 from typing import Dict, Any, List, Tuple
 
 # ─────────────────────────────────────────────
-# NORMALIZATION UTILITIES
+# TEXT NORMALIZATION
 # ─────────────────────────────────────────────
 def _norm(s: str) -> str:
+    """Clean unicode dashes, normalize spaces, lowercase text."""
     if not s:
         return ""
     s = (s.replace("\u2013", "-").replace("\u2014", "-")
-           .replace("—", "-").replace("–", "-")
-           .replace("\u00A0", " "))
+           .replace("—", "-").replace("–", "-").replace("\u00A0", " "))
     return re.sub(r"\s+", " ", s.strip().lower())
 
 def _has(text: str, pat: str) -> bool:
+    """Case-insensitive regex presence check."""
     return re.search(pat, text, flags=re.I) is not None
 
+
 # ─────────────────────────────────────────────
-# LEVEL / REGION LOGIC
+# LEVEL AND REGION LOGIC
 # ─────────────────────────────────────────────
 _ORDER = ["C", "T", "L", "S"]
-_MAXNUM = {"C":7, "T":12, "L":5, "S":5}
+_MAXNUM = {"C": 7, "T": 12, "L": 5, "S": 5}
 
-def _level_sort_key(lv:str)->Tuple[int,int]:
-    rank={"C":100,"T":200,"L":300,"S":400}
-    return (rank.get(lv[0].upper(),999), int(re.sub(r"\D","",lv) or 0))
+def _level_sort_key(lv: str) -> Tuple[int, int]:
+    band_rank = {"C": 100, "T": 200, "L": 300, "S": 400}
+    band = band_rank.get(lv[0].upper(), 999)
+    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)
 _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]:
-    p1,p2=p1.upper(),p2.upper()
-    out=[]; i=_ORDER.index(p1); r=i; num=n1
+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
+    num = n1
     while True:
         out.append(f"{_ORDER[r]}{num}")
-        if _ORDER[r]==p2 and num==n2: break
-        if num<_MAXNUM[_ORDER[r]]:
-            num+=1
+        if _ORDER[r] == p2 and num == n2:
+            break
+        if num < _MAXNUM[_ORDER[r]]:
+            num += 1
         else:
-            r+=1
-            if r>=len(_ORDER): break
-            num=1
+            r += 1
+            if r >= len(_ORDER):
+                break
+            num = 1
     return out
 
-def _extract_levels(t:str)->List[str]:
-    t=_norm(t); levels=set()
+def _extract_levels(t: str) -> List[str]:
+    """Return all vertebral levels found in text (handles cross-region spans)."""
+    t = _norm(t)
+    levels = set()
     for m in _SPAN.finditer(t):
-        p1,n1=m.group(1).upper(),int(m.group(2))
-        p2=(m.group(3) or p1).upper(); n2=int(m.group(4))
-        if p1==p2:
-            for k in range(min(n1,n2),max(n1,n2)+1): levels.add(f"{p1}{k}")
+        p1, n1 = m.group(1).upper(), int(m.group(2))
+        p2 = (m.group(3) or p1).upper()
+        n2 = int(m.group(4))
+        if p1 == p2:
+            for k in range(min(n1, n2), max(n1, n2) + 1):
+                levels.add(f"{p1}{k}")
         else:
-            for lv in _expand_across_regions(p1,n1,p2,n2): levels.add(lv)
+            for lv in _expand_across_regions(p1, n1, p2, n2):
+                levels.add(lv)
     for m in _SINGLE.finditer(t):
         levels.add(f"{m.group(1).upper()}{int(m.group(2))}")
-    return sorted(levels,key=_level_sort_key)
+    return sorted(levels, key=_level_sort_key)
 
-def _count_interspaces(levels:List[str])->int:
-    if not levels: return 0
-    return max(0,len(sorted(set(levels),key=_level_sort_key))-1)
+def _count_interspaces(levels: List[str]) -> int:
+    """Estimate motion segments between unique sequential levels."""
+    if not levels:
+        return 0
+    lv_sorted = sorted(set(levels), key=_level_sort_key)
+    return max(0, len(lv_sorted) - 1)
 
-def _classify_region(levels:List[str])->str:
-    b={"C":False,"T":False,"L":False,"S":False}
+def _classify_region(levels: List[str]) -> str:
+    """Return spinal region classification based on detected level bands."""
+    b = {"C": False, "T": False, "L": False, "S": False}
     for lv in levels:
-        if lv: b[lv[0].upper()]=True
-    c,t,l,s=b["C"],b["T"],b["L"],b["S"]
+        if lv:
+            b[lv[0].upper()] = True
+    c, t, l, s = b["C"], b["T"], b["L"], b["S"]
     if c and t and not (l or s): return "cervicothoracic"
     if t and l and not (c or s): return "thoracolumbar"
     if l and s and not (c or t): return "lumbosacral"
@@ -76,134 +98,279 @@ def _classify_region(levels:List[str])->str:
     if s and not (c or t or l): return "sacral"
     return "mixed"
 
+
 # ─────────────────────────────────────────────
-# LATERALITY / MODIFIERS
+# LATERALITY AND MODIFIERS
 # ─────────────────────────────────────────────
-def _laterality_and_modifiers(note:str)->Tuple[str,List[str]]:
-    t=_norm(note)
-    midline_kw=[
-        r"bilateral medial facetectom",r"bilateral foraminotom",
-        r"bilateral decompression",r"bilateral pedicle screws",
-        r"bilateral rods?",r"left hemilaminectomy",r"right hemilaminectomy"
+def _laterality_and_modifiers(note: str) -> Tuple[str, List[str]]:
+    """Force midline for spine work; suppress invalid -50."""
+    t = _norm(note)
+    midline_kw = [
+        r"bilateral medial facetectom", r"bilateral foraminotom",
+        r"bilateral decompression", r"bilateral pedicle screws",
+        r"bilateral rods?", r"left hemilaminectomy", r"right hemilaminectomy"
     ]
-    if any(_has(t,p) for p in midline_kw):
-        return "midline",[]
-    return "midline",[]
+    if any(_has(t, p) for p in midline_kw):
+        return "midline", []
+    return "midline", []
+
 
 # ─────────────────────────────────────────────
-# FLAG PATTERNS
+# 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"
+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 --- ###
+
+# spine_coder_core_pro.py
+# Vertebro FINAL-v2.2-PRO  (Block 2 / 3)
+# CPT reasoning logic and inference tree
 
 # ─────────────────────────────────────────────
 # INSTRUMENTATION SPAN HELPER
 # ─────────────────────────────────────────────
-def _inst_code_by_span(span:int, anterior:bool)->str:
-    if span<=1: return ""
+def _inst_code_by_span(span: int, anterior: bool) -> str:
+    """Return instrumentation CPT code based on span length."""
+    if span <= 1:
+        return ""
     if anterior:
-        return "22845" if span<=3 else ("22846" if span<=7 else "22847")
+        return "22845" if span <= 3 else ("22846" if span <= 7 else "22847")
     else:
-        return "22842" if span<=3 else ("22843" if span<=7 else "22844")
-
-# ─────────────────────────────────────────────
-# PRIMARY CPT LOGIC (streamlined from v2.1)
-# ─────────────────────────────────────────────
-def _infer_cpts(note:str, region:str, levels:List[str])->List[str]:
-    t=_norm(note); cpts=[]
-    inters=_count_interspaces(levels)
-
-    # Decompression
-    if re.search(r"laminectomy|decompression|facetectomy|foraminotomy",t):
-        base={"cervical":"63045","thoracic":"63046","lumbar":"63047"}.get(region,"63047")
-        extras=max(0,inters-1)
-        cpts.append(base)
-        if extras>0: cpts.extend(["63048"]*extras)
-
-    # Fusions (posterior/interbody)
-    if re.search(FUSION_KW,t):
-        if region.startswith("cervico"): base="22600"
-        elif region.startswith("thoraco"): base="22610"
-        elif region.startswith("lumbo"): base="22612"
-        elif region=="lumbar": base="22630"
-        else: base="22612"
-        extras=max(0,inters)
-        cpts.append(base)
-        if extras>0: cpts.extend(["22614"]*extras)
-
-    # Instrumentation
-    if re.search(INSTR_KW,t):
-        segs=len(levels)
-        if segs<=2: cpts.append("22840")
-        elif segs<=6: cpts.append("22842")
-        elif segs<=12: cpts.append("22843")
-        else: cpts.append("22844")
-
-    # Navigation
-    if re.search(NAV_KW,t): cpts.append("61783")
-
-    # Grafts
-    if re.search(AUTO_SEP_KW,t): cpts.append("20937")
-    elif re.search(AUTO_LOCAL_KW,t): cpts.append("20936")
-    if re.search(ALLO_KW,t): cpts.append("20930")
-
-    # Hardware removal
-    if re.search(r"remov(ed|al).*instrument",t): cpts.append("22852")
-
-    return sorted(set(cpts))
-
-# ─────────────────────────────────────────────
-# MAIN WRAPPER
-# ─────────────────────────────────────────────
-def vertebro_infer(note:str, payer:str="Medicare")->Dict[str,Any]:
-    t=_norm(note)
-    levels=_extract_levels(t)
-    region=_classify_region(levels)
-    inter=_count_interspaces(levels)
-    laterality,mods=_laterality_and_modifiers(t)
-
-    # Flag summary
-    flags=[]
-    for pat,label in [(FUSION_KW,"arthrodesis"),(INSTR_KW,"instrumentation"),
-                      (NAV_KW,"nav"),(ALLO_KW,"allograft"),
-                      (AUTO_LOCAL_KW,"autograft"),(AUTO_SEP_KW,"autograft_sep")]:
-        if re.search(pat,t): flags.append(label)
-
-    cpts=_infer_cpts(t,region,levels)
-
-    # Tech flags
-    flags_map={
-        "microscope":_has(t,r"\bmicroscope\b|microdissection"),
-        "nav":_has(t,r"\bnavigation\b|o-?arm|stealth|7d|image[- ]?guided"),
-        "io_monitor":_has(t,r"\bneuromonitor|ssep|emg|monitoring\b"),
-        "fluoro":_has(t,r"\bfluoro|c[- ]?arm|fluoroscop"),
+        return "22842" if span <= 3 else ("22843" if span <= 7 else "22844")
+
+
+# ─────────────────────────────────────────────
+# 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
+        })
+
+    # ----- 1. Decompression -----
+    if _has(t, r"laminectomy|decompression|facetectomy|foraminotomy"):
+        base_map = {"cervical": "63045", "thoracic": "63046", "lumbar": "63047"}
+        base = base_map.get(region, "63047")
+        add(base, "Decompression, first level",
+            "Detected decompression terms (laminectomy/facetectomy/foraminotomy).",
+            "decompression", 0.86, True)
+        if inters > 0:
+            add("63048", f"Each additional level ×{inters}",
+                "Multi-level decompression inferred.", "decompression add-on", 0.82)
+
+    # ----- 2. Fusion -----
+    if _has(t, FUSION_KW):
+        base_map = {"cervicothoracic": "22600", "cervical": "22600",
+                    "thoracic": "22610", "lumbar": "22630", "lumbosacral": "22612"}
+        base = base_map.get(region, "22612")
+        add(base, "Fusion, first level",
+            f"Fusion/arthrodesis detected ({region} region).", "fusion", 0.9, True)
+        if inters > 0:
+            add("22614", f"Each additional level ×{inters}",
+                "Multi-level fusion extension.", "fusion add-on", 0.84)
+
+    # ----- 3. Instrumentation -----
+    if _has(t, INSTR_KW):
+        code = _inst_code_by_span(span, anterior=False)
+        desc = {
+            "22842": "Posterior segmental instrumentation (2–3 segments)",
+            "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)
+
+    # ----- 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 -----
+    if _has(t, AUTO_SEP_KW):
+        add("20937", "Autograft, separate incision",
+            "Iliac crest or separate-site autograft documented.", "graft", 0.8)
+    elif _has(t, AUTO_LOCAL_KW):
+        add("20936", "Autograft, local (same incision)",
+            "Local autograft material retained.", "graft", 0.8)
+    if _has(t, ALLO_KW):
+        add("20930", "Allograft, morselized / DBM",
+            "Allograft or DBM usage detected.", "graft", 0.8)
+
+    # ----- 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)
+
+    return out
+
+
+# ─────────────────────────────────────────────
+# 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)
+    extra: List[Dict[str, Any]] = []
+
+    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
+        })
+
+    # Tumor / corpectomy
+    if _has(t, r"corpectomy|tumou?r|metastatic|metastasis|en bloc"):
+        base = "63081" if region.startswith("cervical") else "63085"
+        add(base, "Vertebral corpectomy, first segment",
+            "Corpectomy or tumor resection detected.", "tumor/corpectomy", 0.88, True)
+        if inters > 0:
+            addon = "63082" if region.startswith("cervical") else "63086"
+            add(addon, f"Each additional segment ×{inters}",
+                "Multi-segment corpectomy.", "tumor/corpectomy add-on", 0.83)
+
+    # Deformity / osteotomy
+    if _has(t, r"spo\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)
+
+    # Stimulator
+    if _has(t, r"stimulator|paddle lead|scs"):
+        add("63655", "Laminectomy for implantation of neurostimulator paddle",
+            "Paddle lead implantation detected.", "stimulator", 0.86, True)
+        if _has(t, r"ipg|pulse generator|battery"):
+            add("63685", "Insertion of implantable pulse generator (IPG)",
+                "Pulse generator/battery placement documented.", "stimulator add-on", 0.82)
+
+    # Kyphoplasty / Vertebral augmentation
+    if _has(t, r"kyphoplasty|vertebroplasty|cement"):
+        base_map = {"cervical": "22510", "thoracic": "22513", "lumbar": "22514"}
+        base = base_map.get(region, "22514")
+        add(base, "Percutaneous vertebral augmentation, first level",
+            "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)
+
+    # Revision instrumentation
+    if _has(t, r"revision of instrumentation|reinsert|remove and replace"):
+        add("22849", "Revision/reinsertion of instrumentation",
+            "Instrumentation revised/reinserted after removal.", "revision", 0.84, True)
+
+    # Pelvic fixation
+    if _has(t, r"pelvic fixation|iliac bolt|iliac screw|s2ai"):
+        add("22848", "Pelvic fixation (attach instrumentation to pelvis)",
+            "Iliac/S2AI fixation described.", "pelvic_fixation", 0.83)
+
+    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
+# ─────────────────────────────────────────────
+def vertebro_infer(note: str, payer: str = "Medicare") -> Dict[str, Any]:
+    """Master inference wrapper — merges base logic + specialty packs."""
+    t = _norm(note)
+    levels = _extract_levels(t)
+    region = _classify_region(levels)
+    inters = _count_interspaces(levels)
+    laterality, mods = _laterality_and_modifiers(t)
+
+    # --- run 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 average ---
+    merged: Dict[str, Dict[str, Any]] = {}
+    for r in rows:
+        key = (r["cpt"], r["category"])
+        if key not in merged:
+            merged[key] = r.copy()
+        else:
+            merged[key]["confidence"] = round(
+                max(merged[key]["confidence"], r["confidence"]), 2)
+            merged[key]["rationale"] += " / " + r["rationale"]
+
+    rows = list(merged.values())
+    rows.sort(key=lambda r: (-r.get("confidence", 0.0), r["cpt"]))
+
+    # --- flag map (tech) ---
+    flags_map = {
+        "microscope": _has(t, r"\bmicroscope\b|microdissection"),
+        "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"),
     }
-    tech_flags=[k for k,v in flags_map.items() if v]
-
-    return {
-        "payer":payer,
-        "region":region,
-        "levels":levels,
-        "interspaces_est":inter,
-        "laterality":laterality,
-        "case_modifiers":mods,
-        "flags":list(set(flags+tech_flags)),
-        "flags_map":flags_map,
-        "suggested_cpts":cpts,
-        "build":"FINAL-v2.2",
-        "mode":"standard"
+    tech_flags = [k for k, v in flags_map.items() if v]
+
+    # --- attach tech info to rationale for primaries ---
+    if tech_flags:
+        for r in rows:
+            if r.get("primary", False):
+                r["rationale"] = (
+                    r.get("rationale", "") + f" (Tech: {', '.join(tech_flags)})"
+                ).strip()
+
+    # --- build final JSON output ---
+    result = {
+        "payer": payer,
+        "region": region,
+        "levels": levels,
+        "interspaces_est": inters,
+        "laterality": laterality,
+        "case_modifiers": mods,
+        "flags": tech_flags,
+        "flags_map": flags_map,
+        "suggestions": rows,
+        "build": "FINAL-v2.2-PRO",
+        "mode": "standard",
     }
+    return result
+
 
 # ─────────────────────────────────────────────
-# CLI TEST
+# CLI / HF Space TEST ENTRYPOINT
 # ─────────────────────────────────────────────
-if __name__=="__main__":
-    sample = """Preop Dx: C3–C6 stenosis with myelopathy.
-    Procedure: C3–C6 laminectomy and C2–T1 posterior fusion with instrumentation.
-    Navigation used; local autograft and allograft placed."""
+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.
+    """
     result = vertebro_infer(sample)
     print(json.dumps(result, indent=2))