Waystar Charge Billed Report bot. Reads CMS-1500 claim PDFs from the text layer (src/cms1500_text.py) instead of Google Document AI. Dependencies managed with uv (Python 3.12). Secrets, PHI and build output are git-ignored. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
198 lines
8.0 KiB
Python
198 lines
8.0 KiB
Python
"""Unit tests for the CMS-1500 text-layer reader.
|
|
|
|
Words are synthetic (made-up codes and numbers placed at real CMS-1500
|
|
positions), so no patient data lives in this file.
|
|
"""
|
|
from decimal import Decimal
|
|
|
|
from cms1500_text import Word, parse_diagnosis_codes, parse_service_lines
|
|
|
|
|
|
def W(text, x, y):
|
|
return Word(text=text, x=x, y=y)
|
|
|
|
|
|
# Slot labels as Waystar prints them in box 21 (x, y are normalized 0-1).
|
|
DX_LABELS = [
|
|
W("A.", 0.047, 0.602), W("B.", 0.199, 0.603), W("C.", 0.352, 0.602), W("D.", 0.505, 0.604),
|
|
W("E.", 0.047, 0.617), W("F.", 0.200, 0.618), W("G.", 0.352, 0.618), W("H.", 0.505, 0.619),
|
|
W("I.", 0.045, 0.632), W("J.", 0.199, 0.632), W("K.", 0.352, 0.631), W("L.", 0.504, 0.632),
|
|
W("21.", 0.046, 0.581), W("DIAGNOSIS", 0.082, 0.581),
|
|
]
|
|
|
|
|
|
def test_dx_codes_are_keyed_by_slot_letter_in_slot_order():
|
|
words = DX_LABELS + [
|
|
# Printed slightly above their labels, with jittery y — the old OCR
|
|
# sort by (y, x) scrambled exactly this kind of row.
|
|
W("Z9999", 0.248, 0.5975), # B
|
|
W("M1111", 0.095, 0.5985), # A
|
|
W("R2222", 0.400, 0.5980), # C
|
|
W("S81812A", 0.095, 0.6130), # E
|
|
]
|
|
assert parse_diagnosis_codes(words) == [
|
|
("A", "M1111"), ("B", "Z9999"), ("C", "R2222"), ("E", "S81812A"),
|
|
]
|
|
|
|
|
|
def line_words(y, mm="01", dd="05", yy="26", pos="12", cpt="97110", mods=("GP",),
|
|
ptr="AB", charge="9000", units="2", npi="1234567890", row_no=None):
|
|
"""One box-24 service line at Waystar's x positions. None = not printed."""
|
|
ws = [W("NPI", 0.801, y)]
|
|
if row_no:
|
|
ws.append(W(row_no, 0.026, y - 0.009)) # gutter number sits in the shaded row above
|
|
for text, x in ((mm, 0.048), (dd, 0.083), (yy, 0.118), (pos, 0.259), (cpt, 0.348),
|
|
(ptr, 0.565), (charge, 0.683), (units, 0.724), (npi, 0.883)):
|
|
if text is not None:
|
|
ws.append(W(text, x, y))
|
|
for i, m in enumerate(mods):
|
|
ws.append(W(m, 0.424 + 0.035 * i, y))
|
|
return ws
|
|
|
|
|
|
BOX_24_HEADER = [W("24.", 0.046, 0.642), W("UNITS", 0.736, 0.661)]
|
|
BOX_25_ROW = [W("25.", 0.046, 0.854), W("44680", 0.324, 0.869), W("14439", 0.395, 0.869),
|
|
W("63000", 0.701, 0.869)]
|
|
|
|
|
|
def test_service_line_fields_are_read_from_their_columns():
|
|
words = BOX_24_HEADER + BOX_25_ROW + line_words(0.689, mods=("GP", "CQ", "59"))
|
|
[line] = parse_service_lines(words)
|
|
assert line.line_no == 1
|
|
assert line.dos_from == "01/05/2026"
|
|
assert line.dos_to == ""
|
|
assert line.pos == "12"
|
|
assert line.cpt == "97110"
|
|
assert line.modifiers == ("GP", "CQ", "59")
|
|
assert line.dx_pointer == "AB"
|
|
assert line.charge == Decimal("90.00")
|
|
assert line.units == "2"
|
|
assert line.npi == "1234567890"
|
|
|
|
|
|
def test_lines_are_numbered_top_to_bottom_and_ignore_gutter_and_box_26():
|
|
words = (BOX_24_HEADER + BOX_25_ROW
|
|
+ line_words(0.719, cpt="97140", row_no="2")
|
|
+ line_words(0.689, cpt="97110", row_no="1"))
|
|
lines = parse_service_lines(words)
|
|
assert [(l.line_no, l.cpt) for l in lines] == [(1, "97110"), (2, "97140")]
|
|
assert all(l.dos_from == "01/05/2026" for l in lines) # gutter "1"/"2" never read as MM
|
|
|
|
|
|
def test_a_missing_value_stays_blank_on_its_own_line():
|
|
words = (BOX_24_HEADER
|
|
+ line_words(0.689, units="2")
|
|
+ line_words(0.719, units=None)
|
|
+ line_words(0.750, units="1"))
|
|
assert [l.units for l in parse_service_lines(words)] == ["2", "", "1"]
|
|
|
|
|
|
def test_charge_without_decimal_point_has_implied_cents():
|
|
words = (BOX_24_HEADER
|
|
+ line_words(0.689, charge="12000")
|
|
+ line_words(0.719, charge="45.50"))
|
|
assert [l.charge for l in parse_service_lines(words)] == [Decimal("120.00"), Decimal("45.50")]
|
|
|
|
|
|
def test_hcpcs_line_with_dos_to_and_large_units():
|
|
words = BOX_24_HEADER + line_words(0.689, cpt="Q4250", mods=(), ptr="A", units="120")
|
|
words += [W("01", 0.155, 0.689), W("07", 0.186, 0.689), W("26", 0.224, 0.689)]
|
|
[line] = parse_service_lines(words)
|
|
assert (line.cpt, line.units, line.modifiers) == ("Q4250", "120", ())
|
|
assert line.dos_to == "01/07/2026"
|
|
|
|
|
|
from cms1500_text import Claim, build_claim, parse_header, to_report_row # noqa: E402
|
|
|
|
HEADER_WORDS = [
|
|
W("1a.", 0.619, 0.128), W("NUMBER", 0.723, 0.128),
|
|
W("1AB2CD3EF45", 0.677, 0.145),
|
|
W("3.", 0.381, 0.158), W("MM", 0.402, 0.165), W("DD", 0.439, 0.165), W("YY", 0.482, 0.165),
|
|
W("07", 0.401, 0.174), W("04", 0.436, 0.174), W("1950", 0.483, 0.174), W("M", 0.524, 0.174),
|
|
]
|
|
|
|
|
|
def test_header_fields_come_from_their_boxes():
|
|
h = parse_header(HEADER_WORDS + BOX_25_ROW)
|
|
assert h == {
|
|
"insured_id": "1AB2CD3EF45",
|
|
"dob": "07/04/1950",
|
|
"patient_account": "44680 14439",
|
|
"total_charge": Decimal("630.00"),
|
|
}
|
|
|
|
|
|
def page(*line_specs, total="18000"):
|
|
"""A full synthetic page: header + dx + given lines + box 25/28 row."""
|
|
ws = HEADER_WORDS + DX_LABELS + [W("M1111", 0.095, 0.598), W("Z9999", 0.248, 0.598)]
|
|
ws += BOX_24_HEADER + [W("25.", 0.046, 0.854), W("44680", 0.324, 0.869),
|
|
W("14439", 0.395, 0.869), W(total, 0.701, 0.869)]
|
|
for i, kw in enumerate(line_specs):
|
|
ws += line_words(0.689 + 0.030 * i, **kw)
|
|
return ws
|
|
|
|
|
|
def test_clean_claim_has_no_review_notes():
|
|
claim = build_claim([page({}, {"cpt": "97140"})])
|
|
assert isinstance(claim, Claim)
|
|
assert [l.cpt for l in claim.lines] == ["97110", "97140"]
|
|
assert claim.issues == []
|
|
|
|
|
|
def test_line_charges_must_add_up_to_box_28():
|
|
claim = build_claim([page({}, {}, total="20000")])
|
|
assert claim.issues == ["Line charges 180.00 != box 28 total 200.00"]
|
|
|
|
|
|
def test_missing_line_values_and_bad_pointer_are_flagged():
|
|
claim = build_claim([page({"units": None, "ptr": "AC"}, {"npi": None})])
|
|
assert claim.issues == [
|
|
"Line 1: no Units",
|
|
"Line 1: Dx pointer C has no diagnosis code",
|
|
"Line 2: no Rendering NPI",
|
|
]
|
|
|
|
|
|
def test_claim_with_no_service_lines_or_no_text_is_flagged():
|
|
assert build_claim([page()]).issues == ["No service lines found", "Line charges 0.00 != box 28 total 180.00"]
|
|
assert build_claim([[]]).issues == ["No text layer on PDF (scanned?) - check by hand"]
|
|
|
|
|
|
def test_lines_continue_across_pages_and_totals_add_up():
|
|
claim = build_claim([page({}, {}, {}, {}, {}, {}, total="54000"), page({}, total="9000")])
|
|
assert [l.line_no for l in claim.lines] == [1, 2, 3, 4, 5, 6, 7]
|
|
assert claim.total_charge == Decimal("630.00")
|
|
assert claim.issues == []
|
|
|
|
|
|
def test_report_row_keeps_legacy_columns_and_alignment():
|
|
claim = build_claim([page({"units": "2"}, {"cpt": "97140", "units": None, "mods": ("GP", "59")})])
|
|
row = to_report_row(claim, claim_id="999", insurance="PAYER", patient_name="DOE, J",
|
|
signature_physician="Jo Doe", fallback_account="")
|
|
assert list(row)[:15] == [
|
|
"Claim ID", "Rendering Provider #", "Signature of Physician", "Patient's Account No",
|
|
"Date of Birth", "Insurance Name", "Insured ID Number", "Patient's Name",
|
|
"Date of Service", "Diagnosis Codes", "CPT", "POS", "Modifier", "Units", "Charges",
|
|
]
|
|
assert row["CPT"] == "CPT 1: 97110; CPT 2: 97140"
|
|
assert row["Units"] == "Units 1: 2; Units 2: " # blank stays on line 2
|
|
assert row["Modifier"] == "Modifier 1: GP; Modifier 2: GP 59"
|
|
assert row["Diagnosis Codes"] == "M1111, Z9999 (Total: 2)"
|
|
assert row["Patient's Account No"] == "44680 14439"
|
|
assert row["Charges"] == "180.00"
|
|
assert row["Line Charges"] == "Charge 1: 90.00; Charge 2: 90.00"
|
|
assert row["Review Notes"] == "Line 2: no Units"
|
|
|
|
|
|
def test_report_row_for_single_line_claim_uses_plain_values():
|
|
row = to_report_row(build_claim([page({}, total="9000")]), claim_id="1")
|
|
assert (row["CPT"], row["Units"], row["Rendering Provider #"]) == ("97110", "2", "1234567890")
|
|
assert row["Review Notes"] == ""
|
|
|
|
|
|
def test_report_row_falls_back_to_grid_charges_when_box_28_is_blank():
|
|
claim = build_claim([page({})])
|
|
claim.total_charge = None
|
|
row = to_report_row(claim, claim_id="1", fallback_charges="90")
|
|
assert row["Charges"] == "90"
|