# spine_coder_core_pro_elite.py # Vertebro FINAL-v2.3-PRO-Elite (Block 1/4) # Core utilities, text normalization, level parsing, region classification import re, json from typing import Dict, Any, List, Tuple # ───────────────────────────────────────────── # TEXT NORMALIZATION # ───────────────────────────────────────────── def _norm(s: str) -> str: if not s: return "" s = (s.replace("\u2013", "-").replace("\u2014", "-") .replace("—", "-").replace("–", "-").replace("\u00A0", " ")) return re.sub(r"\s+", " ", s.strip().lower()) def _has(text: str, pat: str) -> bool: return re.search(pat, text, flags=re.I) is not None # ───────────────────────────────────────────── # LEVEL AND REGION LOGIC # ───────────────────────────────────────────── _ORDER = ["C", "T", "L", "S"] _MAXNUM = {"C": 7, "T": 12, "L": 5, "S": 5} 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, r, num = [], _ORDER.index(p1), n1 while True: out.append(f"{_ORDER[r]}{num}") 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 return out def _extract_levels(t: str) -> List[str]: 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}") else: 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) def _count_interspaces(levels: List[str]) -> int: 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} for lv in levels: 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" if c and not (t or l or s): return "cervical" if t and not (c or l or s): return "thoracic" if l and not (c or t or s): return "lumbar" if s and not (c or t or l): return "sacral" return "mixed" # Laterality placeholder (case-level modifiers add real ones) def _laterality_and_modifiers(note: str) -> Tuple[str, List[str]]: return "midline", [] # ───────────────────────────────────────────── # KEYWORD GROUPS (v2.3 adds VCR/CDA/Exposure terms) # ───────────────────────────────────────────── 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|stealth|7d)\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|iliac crest bone|icbg)\b" # New v2.3 detector phrases ANTERIOR_KW = r"\banterior (approach|exposure|cervical|lumbar|thoracic)\b|\bacdf\b|\balif\b|\bsmith[- ]?robinson\b" POSTERIOR_KW = r"\bposterior (approach|exposure)\b|\btlif\b|\bplif\b|\bposterolateral fusion\b|\bpedicle screws?\b|\brods?\b" EXPOSURE_ONLY_KW= r"\b(exposure|approach)\b.*\b(no|without)\b.*\b(fusion|interbody|instrument|implant|cage|plate|screw)\b" VCR_KW = r"\bvertebral column resection\b|\bVCR\b|\b3[- ]column resection\b" INTRADURAL_KW = r"\bintradural\b|(intra[- ]?dural).*tumou?r|\bmeningioma\b|\bschwannoma\b|\bneurinoma\b|\bfilum\b" CERVICAL_ARTHRO = r"\b(cervical (arthroplasty|disc replacement|artificial disc)|\bcda\b)\b" LUMBAR_ARTHRO = r"\b(lumbar (arthroplasty|disc replacement)|\btdr\b)\b" # Vertebro FINAL-v2.3-PRO-Elite (Block 2/4) # CPT reasoning engine + specialty packs (extended) def _inst_code_by_span(span: int, anterior: bool) -> str: if span <= 1: return "" if anterior: return "22845" if span <= 3 else ("22846" if span <= 7 else "22847") return "22842" if span <= 3 else ("22843" if span <= 7 else "22844") def _infer_cpts(note: str, region: str, levels: List[str]) -> List[Dict[str, Any]]: t = _norm(note) out: List[Dict[str, Any]] = [] inters = _count_interspaces(levels) span = max(inters + 1, 2) 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}) # Washout-only guard washout_only = _has(t, r"(washout|irrigation and debridement|i\&d)") and not _has(t, FUSION_KW) # 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) TLIF/PLIF (explicit or implicit) tlif_like = ( _has(t, r"\btlif\b|\bplif\b|posterior interbody fusion") or (_has(t, r"\bfacetectom(y|ies)\b|complete facetectomy|transforaminal") and _has(t, r"\b(interbody (cage|device|spacer)|peek (cage|spacer)|titanium (cage|spacer)|allograft spacer)\b") and _has(t, r"\bpedicle\s+screws?\b")) ) if tlif_like and not washout_only and region in {"lumbar", "thoracic"}: add("22633", "Posterior/posterolateral + posterior interbody, single level", "TLIF/PLIF pattern: interbody device + pedicle screws.", "TLIF/PLIF", 0.92, True) if inters > 0: add("22634", f"Posterior interbody each additional interspace ×{inters}", "Multi-level TLIF/PLIF.", "TLIF/PLIF add-on", 0.88) code = _inst_code_by_span(span, anterior=False) if code: desc = {"22842":"Posterior segmental instrumentation (2–3 segments)", "22843":"Posterior segmental instrumentation (4–7 segments)", "22844":"Posterior segmental instrumentation (8+ segments)"}[code] add(code, desc, "Posterior instrumentation detected.", "instrumentation", 0.83) # 3) ALIF (+ optional plate) if _has(t, r"\balif\b|anterior lumbar interbody fusion") and not washout_only: add("22558", "Anterior lumbar interbody fusion, single interspace", "ALIF detected.", "ALIF", 0.9, True) if inters > 0: add("22585", f"ALIF each additional interspace ×{inters}", "Multi-level ALIF.", "ALIF add-on", 0.86) if _has(t, r"\b(anterior (plate|plating)|plate fixed)\b|\bplate\b"): code = _inst_code_by_span(span, anterior=True) if code: desc = {"22845":"Anterior instrumentation (2–3 segments)", "22846":"Anterior instrumentation (4–7 segments)", "22847":"Anterior instrumentation (8+ segments)"}[code] add(code, desc, "Anterior plate present; span estimated from levels.", "instrumentation", 0.8) # 4) Posterolateral/posterior fusion (no interbody) if _has(t, r"posterolateral\b.*\bfusion|posterior\b.*\bfusion|in situ\b.*\bfusion") \ and not _has(t, r"\btlif\b|\bplif\b|posterior interbody") and not washout_only: base_map = {"cervical":"22600", "thoracic":"22610", "lumbar":"22612", "lumbosacral":"22612", "cervicothoracic":"22600"} base = base_map.get(region, "22612") add(base, f"Posterolateral/posterior fusion, first level ({region})", "Posterior fusion without interbody.", "posterior_fusion", 0.78, True) if inters > 0: add("22614", f"Posterior fusion each additional segment ×{inters}", "Multi-level posterior fusion.", "posterior_fusion add-on", 0.72) if _has(t, INSTR_KW): code = _inst_code_by_span(span, anterior=False) if code: desc = {"22842":"Posterior segmental instrumentation (2–3 segments)", "22843":"Posterior segmental instrumentation (4–7 segments)", "22844":"Posterior segmental instrumentation (8+ segments)"}[code] add(code, desc, "Posterior instrumentation detected.", "instrumentation", 0.8) # 5) Instrumentation w/o explicit fusion (posterior) if _has(t, INSTR_KW) and not washout_only: code = _inst_code_by_span(span, anterior=False) if code and not any(r["category"] == "instrumentation" for r in out): desc = {"22842":"Posterior segmental instrumentation (2–3 segments)", "22843":"Posterior segmental instrumentation (4–7 segments)", "22844":"Posterior segmental instrumentation (8+ segments)"}[code] add(code, desc, "Posterior instrumentation documented.", "instrumentation", 0.82) # 6) Navigation if _has(t, NAV_KW): add("61783", "Intraoperative navigation (image-guided)", "Navigation terms detected (O-arm/3D spin/Stealth/7D).", "navigation", 0.82) # 7) Bone grafts if _has(t, AUTO_SEP_KW): add("20937", "Autograft (separate incision)", "Iliac crest or separate-site autograft.", "graft", 0.8) elif _has(t, AUTO_LOCAL_KW): add("20936", "Autograft, local (same incision)", "Local autograft retained.", "graft", 0.8) if _has(t, ALLO_KW): add("20930", "Allograft, morselized / DBM", "Allograft/DBM used.", "graft", 0.8) # 8) Hardware removal if _has(t, r"(remov(ed|al)|explant).*(instrument|hardware|plate|rod|screw)"): if _has(t, r"\banterior\b|acdf|plate"): add("22855", "Removal of anterior instrumentation", "Anterior hardware removal.", "hardware_removal", 0.84, True) else: add("22852", "Removal of posterior instrumentation", "Posterior hardware removal.", "hardware_removal", 0.84, True) return out # ───────────────────────────────────────────── # SPECIALTY PACKS (adds 360°, VCR, Intradural, Exposure-only) # ───────────────────────────────────────────── def _apply_specialty_packs(note: str, region: str, inters: int, levels: List[str]) -> List[Dict[str, Any]]: 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 (kept from v2.2) if _has(t, r"corpectomy|tumou?r|metastatic|metastasis|en bloc"): base = "63081" if region.startswith("cervical") or "cervico" in region else "63085" add(base, "Vertebral corpectomy, first segment", "Corpectomy/tumor resection.", "tumor/corpectomy", 0.88, True) if inters > 0: addon = "63082" if base == "63081" else "63086" add(addon, f"Each additional segment ×{inters}", "Multi-segment corpectomy.", "tumor/corpectomy add-on", 0.83) # Deformity / SPO (kept) 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 with SPO.", "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 (kept) if _has(t, r"stimulator|paddle lead|scs"): add("63655", "Laminectomy for implantation of neurostimulator paddle", "Paddle lead placement.", "stimulator", 0.86, True) if _has(t, r"\bipg\b|pulse generator|battery"): add("63685", "Insertion or replacement of IPG", "Pulse generator placed.", "stimulator add-on", 0.82) # Kyphoplasty / Vertebral augmentation (kept) 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", "Kypho/vertebroplasty terms found.", "augmentation", 0.84, True) if inters > 0: add("22515", f"Each additional vertebral body ×{inters}", "Multi-level augmentation.", "augmentation add-on", 0.8) # Revision instrumentation (kept) if _has(t, r"revision of instrumentation|reinsertion|remove and replace|re-?insert"): add("22849", "Revision/reinsertion of spinal instrumentation", "Instrumentation revised/reinserted.", "revision", 0.84, True) # Pelvic fixation (kept) if _has(t, r"pelvic fixation|iliac bolt|iliac screw|s2ai"): add("22848", "Pelvic fixation (attach instrumentation to pelvis)", "Iliac/S2AI fixation.", "pelvic_fixation", 0.83) # NEW: 360° fusion bundler (anterior + posterior in same episode) if _has(t, ANTERIOR_KW) and _has(t, POSTERIOR_KW): add("BUNDLE-360", "360° (circumferential) construct detected", "Anterior + posterior fusion/instrumentation in same setting; verify distinct CPT families applied correctly.", "bundle", 0.9, False) # NEW: Vertebral Column Resection (flag for coder review) if _has(t, VCR_KW): add("REVIEW-VCR", "Vertebral Column Resection (3-column)", "VCR language detected; confirm exact CPT family and approach-level specifics before final billing.", "deformity_vcr_review", 0.95, True) # NEW: Intradural tumor resection (flag for coder review) if _has(t, INTRADURAL_KW): add("REVIEW-INTRADURAL", "Intradural tumor resection suspected", "Intradural tumor/excision terms detected; map to exact intradural CPT by level and approach.", "intradural_review", 0.93, True) # NEW: Exposure-only (guard produces explicit output row) if _has(t, EXPOSURE_ONLY_KW) and not _has(t, FUSION_KW) and not _has(t, INSTR_KW): add("EXPOSURE-ONLY", "Approach/exposure only (no fusion/instrumentation performed)", "Explicit documentation of exposure without implantation or arthrodesis.", "exposure_only", 0.9, True) # Arthroplasty (kept but using the new keywords) if _has(t, CERVICAL_ARTHRO): add("22856", "Cervical disc arthroplasty, single level", "Cervical disc arthroplasty documented.", "arthroplasty", 0.87, True) if _has(t, LUMBAR_ARTHRO): add("22857", "Lumbar disc arthroplasty, single level", "Lumbar disc arthroplasty documented.", "arthroplasty", 0.86, True) return extra # Vertebro FINAL-v2.3-PRO-Elite (Block 3/4) # Case-level modifiers & complications detector (same as your v2.2, labeled v2.3) def _detect_case_modifiers_and_complications(note: str) -> Tuple[List[Dict[str, str]], Dict[str, bool], List[str]]: t = _norm(note) modifiers: List[Dict[str, str]] = [] complications_map: Dict[str, bool] = {} complications_list: List[str] = [] mod_patterns = [ ("22", r"(complex|technically (difficult|demanding)|difficult dissection|extensive adhesiolysis|" r"severe deformity|morbid obesity|revision (case|exposure)|re-?operative field|" r"dense scar tissue|prolonged (exposure|procedure)|ossified p(?:l|ll)\b)", "Increased procedural service (complexity)."), ("52", r"\b(partial|limited|reduced (service|extent))\b", "Reduced service performed."), ("53", r"\b(aborted|terminated|discontinued|stopped prior to completion)\b", "Procedure discontinued for patient safety."), ("62", r"\b(co[- ]?surgeon|two surgeons|co-surgeons)\b", "Two surgeons (co-surgeons) documented."), ("76", r"\brepeat(ed)? procedure\b.*\b(same (surgeon|physician))\b", "Repeat procedure/service by the same physician."), ("77", r"\brepeat(ed)? procedure\b.*\b(another|different) (surgeon|physician)\b", "Repeat procedure by another physician."), ("78", r"(\bunplanned return\b.*\b(operating|op)\s*room\b)|" r"(\breturn to (the )?(operating|op)\s*room\b.*\b(postoperative|global period)\b)", "Unplanned return to OR during postoperative period."), ("79", r"\bunrelated procedure\b.*\b(postoperative|global period)\b", "Unrelated procedure during postoperative period."), ] for code, pat, reason in mod_patterns: if _has(t, pat): modifiers.append({"modifier": code, "reason": reason}) # Assistant: AS vs -80 (mutually exclusive); -82 supersedes -80 has_pa_np = _has(t, r"\b(pa[- ]?c|physician assistant|pa-c|nurse practitioner|np|advanced practice provider|app)\b") has_md_do = _has(t, r"\b(dr\.?|m\.?d\.?|d\.?o\.?)\b") or _has(t, r"\bassistant surgeon\b") if _has(t, r"assistant\(s\):") or _has(t, r"\bassistant(s)?[:\-]"): has_md_do = True if has_pa_np: modifiers.append({"modifier": "AS", "reason": "Non-physician assistant at surgery."}) elif has_md_do: modifiers.append({"modifier": "80", "reason": "Assistant surgeon documented."}) if any(m["modifier"] == "82" for m in modifiers): modifiers = [m for m in modifiers if m["modifier"] != "80"] if any(m["modifier"] == "AS" for m in modifiers): modifiers = [m for m in modifiers if m["modifier"] != "80"] if any(m["modifier"] == "53" for m in modifiers): modifiers = [m for m in modifiers if m["modifier"] != "52"] # Complications comp_rules = [ ("dural_tear_or_durotomy", r"\b(dural tear|durotom(y|ies)|csf leak|cerebrospinal fluid leak)\b"), ("neuromonitoring_change", r"(loss|significant (decrease|change)).*(ssep|mep|meps|tcem|neuromonitor)"), ("neurologic_deficit", r"\b(new|worsen(ed|ing)).*(weakness|deficit|paresthesia|paralysis|foot drop)\b"), ("vascular_injury", r"\b(vertebral artery|carotid|venous|arterial) (injury|laceration|avulsion)\b"), ("epidural_or_wound_hematoma", r"\b(epidural hematoma|wound hematoma|compressive hematoma|hematoma evacuation)\b"), ("infection_or_washout", r"\b(infection|purulence|washout|irrigation and debridement|i\&d)\b"), ("hardware_malposition_or_revision", r"\b(malposition(ed)? (screw|hardware)|reposition(ed)? screw|re-?insert(ed)? screw)\b"), ("cage_migration_or_retropulsion", r"\b(cage|interbody).*(migrat|retropuls|backed out)\b"), ("iatrogenic_fracture", r"\biatrogenic (fracture|fx)\b|\bendplate fracture\b"), ("positioning_injury", r"\b(pressure (injury|ulcer)|brachial plexus|ulnar neuropathy|peroneal neuropathy)\b"), ("transfusion", r"\btransfus(ed|ion)|prbc\b|\bcell saver\b"), ("massive_blood_loss", r"\bebl\b.*\b(> ?800\s?ml|>\s?1(\.|,)?0?00\s?ml|> ?1\s?l|> ?1000\s?cc)\b"), ("unplanned_return_to_or", r"\bunplanned return\b.*\b(operating|op)\s*room\b"), ("wound_dehiscence", r"\bwound dehiscence\b|\bdehisced\b"), ] for key, pat in comp_rules: hit = _has(t, pat) complications_map[key] = hit if hit: complications_list.append(key.replace("_", " ").title()) if _has(t, r"no significant changes in (motor|sensory) evoked potentials|neuromonitoring.*no (significant )?changes"): complications_map["neuromonitoring_change"] = False complications_list = [c for c in complications_list if c != "Neuromonitoring Change"] return modifiers, complications_map, complications_list # Vertebro FINAL-v2.3-PRO-Elite (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 + modifiers/complications.""" t = _norm(note) levels = _extract_levels(t) region = _classify_region(levels) inters = _count_interspaces(levels) laterality, mods_stub = _laterality_and_modifiers(t) # Inference base_rows = _infer_cpts(t, region, levels) extra_rows = _apply_specialty_packs(t, region, inters, levels) rows = base_rows + extra_rows # Merge duplicates (max confidence, concat rationale) merged: Dict[tuple, 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"] = (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"])) # Tech flags flags_map = { "microscope": _has(t, r"\bmicroscope\b|microdissection"), "nav": _has(t, r"\bnavigation\b|o-?arm|ziehm|3d\s+spin|stealth|7d"), "io_monitor": _has(t, r"\bneuromonitor|ssep|mep|meps|tcem\b|monitoring\b"), "fluoro": _has(t, r"\bfluoro|c[- ]?arm|fluoroscop"), } tech_flags = [k for k, v in flags_map.items() if v] # 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() # Case-level modifiers & complications case_modifiers, complications_map, complications_list = _detect_case_modifiers_and_complications(t) if not case_modifiers and mods_stub: case_modifiers = [{"modifier": m, "reason": "Laterality-derived modifier."} for m in mods_stub] # Propagate -53 only to primary rows if any(m["modifier"] == "53" for m in case_modifiers): for r in rows: if r.get("primary", False): r.setdefault("modifiers", []) if "53" not in r["modifiers"]: r["modifiers"].append("53") else: if r.get("modifiers"): r["modifiers"] = [m for m in r["modifiers"] if m != "53"] # Top-K cap try: k = int(top_k) if k > 0: rows = rows[:k] except Exception: pass return { "payer": payer, "region": region, "levels": levels, "interspaces_est": inters, "laterality": laterality, "case_modifiers": case_modifiers, "flags": tech_flags, "flags_map": flags_map, "complications_flags": complications_map, "complications": complications_list, "suggestions": rows, "build": "FINAL-v2.3-PRO-Elite", "mode": "standard", } # Backward compatibility alias suggest_with_cpt_billing = vertebro_infer # CLI smoke test if __name__ == "__main__": sample = """ 360 fusion example: ALIF L4–L5 with structural allograft and plate; posterior pedicle screws L4–L5 with posterolateral fusion; fluoroscopy + O-arm. """ print(json.dumps(vertebro_infer(sample, top_k=15), indent=2)) # Vertebro FINAL-v2.3-PRO-Elite (Block 5/5) # Test harness, batch helper, and lightweight Space shim (optional) from typing import Iterable # ───────────────────────────────────────────── # Batch helper # ───────────────────────────────────────────── def batch_infer(notes: Iterable[str], payer: str = "Medicare", top_k: int = 10): """Run vertebro_infer over an iterable of op notes. Returns list of dicts.""" results = [] for i, note in enumerate(notes, 1): try: out = vertebro_infer(note, payer=payer, top_k=top_k) results.append({ "idx": i, "region": out.get("region"), "levels": out.get("levels"), "interspaces": out.get("interspaces_est"), "flags": out.get("flags"), "case_modifiers": out.get("case_modifiers"), "complications": out.get("complications"), "primary_codes": [ {"cpt": r["cpt"], "desc": r["desc"], "cat": r["category"], "conf": r.get("confidence")} for r in out.get("suggestions", []) if r.get("primary") ], "all_codes": [ {"cpt": r["cpt"], "desc": r["desc"], "cat": r["category"], "conf": r.get("confidence")} for r in out.get("suggestions", []) ], "raw": out, }) except Exception as e: results.append({"idx": i, "error": str(e)}) return results # ───────────────────────────────────────────── # Pretty printer for CLI debugging # ───────────────────────────────────────────── def _pp_row(row: dict): if "error" in row: print(f"[{row['idx']}] ERROR: {row['error']}") return print(f"\n[{row['idx']}] Region={row['region']} Levels={row['levels']} Interspaces={row['interspaces']}") if row.get("flags"): print(f" Tech: {', '.join(row['flags'])}") if row.get("case_modifiers"): mods = ', '.join([m['modifier'] for m in row["case_modifiers"]]) print(f" Case Modifiers: {mods}") if row.get("complications"): print(f" Complications: {', '.join(row['complications'])}") prims = row.get("primary_codes", []) print(" Primary:") if not prims: print(" (none)") for p in prims: print(f" {p['cpt']} {p['cat']} conf={p['conf']} — {p['desc']}") allc = row.get("all_codes", []) extra = [a for a in allc if a not in prims] if extra: print(" Add-ons / Extras:") for e in extra: print(f" {e['cpt']} {e['cat']} conf={e['conf']} — {e['desc']}") # ───────────────────────────────────────────── # Smoke test set (Jason-style) # ───────────────────────────────────────────── def _smoke_notes(): return [ # 1) Cervicothoracic posterior fusion + decomp + nav + grafts """Preop: Cervical stenosis with myelopathy. Proc: C2–T2 posterolateral arthrodesis; posterior instrumentation C2–T2; C3–C6 laminectomy with bilateral medial facetectomies/foraminotomies; navigation (Ziehm 3D spin) and fluoroscopy; local autograft + DBM. Assistant(s): Dr. Amber Parker.""", # 2) ACDF C4–C6 with plate """ACDF C4–C6 with structural allograft; anterior plate spanning C4–C6; nav + fluoro; PA-C assisting.""", # 3) TLIF L4–L5 with screws/rods """Right facetectomy and TLIF L4–L5 with PEEK cage; posterior pedicle screws L4–L5 with rods; posterolateral arthrodesis; microscope + fluoro; PA-C assistant.""", # 4) ALIF L4–S1 with plate """ALIF L4–S1 with structural allograft spacers and anterior plating; vascular exposure; PA-C present; fluoroscopy.""", # 5) Deformity PSO + pelvic fixation """L3 pedicle subtraction osteotomy; posterior segmental instrumentation L2–S1 with S2AI pelvic fixation; posterolateral fusion L2–S1; O-arm navigation; EBL 1200 mL; MD assistant.""", # 6) Tumor corpectomy T7 + long construct """T7 corpectomy for metastatic tumor with expandable cage reconstruction; posterior instrumentation T5–T9; laminectomy T7; navigation + fluoroscopy; MD assistant.""", # 7) Postop epidural hematoma evacuation (take-back) """Unplanned return to OR POD#1: L4–L5 laminectomy for evacuation of epidural hematoma; Hemovac placed; PA-C assist; fluoroscopy.""", # 8) I&D depth to bone (infection) """Irrigation and debridement of lumbar wound to bone with pulse-lavage; VAC applied; PA-C assistant.""", # 9) Exposure-only (guard) """Left retroperitoneal exposure L4–S1 by vascular surgeon only; no interbody, no fusion, no screws/plate/cage placed; closed; to be staged later.""", # 10) Cervical arthroplasty """C5–C6 cervical disc arthroplasty (CDA); microscope, Ziehm 3D spin, fluoroscopy; PA-C assistant.""", # 11) Lumbar TDR """L5–S1 total disc replacement via anterior approach; vascular exposure; fluoroscopy; PA-C present.""", # 12) SCS paddle + IPG """T9–T10 laminectomy for paddle lead; IPG in right gluteal pocket; fluoroscopy; PA-C assistant.""", # 13) NEW: 360° fusion bundler case """ALIF L4–L5 with structural allograft and anterior plate; in same setting posterior approach with bilateral pedicle screws L4–L5 and posterolateral fusion; O-arm nav; fluoroscopy.""", # 14) NEW: VCR review case """Rigid kyphosis: Vertebral Column Resection (VCR) at T8 (3-column), cage reconstruction, posterior segmental instrumentation T6–T10; neuromonitoring; fluoroscopy; navigation.""", # 15) NEW: Intradural tumor review case """C6–C7 intradural extramedullary tumor (likely meningioma). Microscope; intradural tumor resection with dural repair; IONM; fluoroscopy.""", ] # ───────────────────────────────────────────── # Public smoke runner # ───────────────────────────────────────────── def run_smoke_tests(top_k: int = 12): """Run a fixed set of diverse Jason-style notes through the engine and print.""" notes = _smoke_notes() results = batch_infer(notes, top_k=top_k) for row in results: _pp_row(row) return results # ───────────────────────────────────────────── # Minimal Space shim (no hard dependency on gradio) # Provides a callable `infer_for_space(note, payer, top_k)` used by app.py # ───────────────────────────────────────────── def infer_for_space(note: str, payer: str = "Medicare", top_k: int = 10) -> str: """ Stable text output for UI: returns pretty JSON string. Import this in app.py and wire to your textbox -> JSON panel. """ res = vertebro_infer(note, payer=payer, top_k=top_k) try: return json.dumps(res, indent=2) except Exception: # Fallback minimal serializer return str(res) # ───────────────────────────────────────────── # When run directly, execute smoke tests # ───────────────────────────────────────────── if __name__ == "__main__": print("Running Vertebro v2.3-PRO-Elite smoke tests…") _ = run_smoke_tests(top_k=15)