ManikandanandClaude Opus 5.5
ba15a574ab
Snapshot of the working project as of 2026-10-23 (local 2d4d4fc), one commit with no earlier history: the tracked files were de-identified first (synthetic claim ids and patient data only; see Context.md section 7). Credentials, run logs, screenshots, claim reports and client workbooks are gitignored and not included. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
381 lines
14 KiB
Python
381 lines
14 KiB
Python
r"""
|
||
conditions/medicare_pps_code.py — document #25: a PPS Payer's UB claim must
|
||
carry the PPS Code its billed Qualifying Codes map to on the client's
|
||
Medicare G code crosswalk (G0466–G0470; operator, 2026-10-23, spec EEWA-34).
|
||
|
||
python tests\test_conditions_medicare_pps_code.py
|
||
|
||
Terms (GLOSSARY.md): PPS Code, Qualifying Code, PPS Payer. Claim numbers
|
||
here are made up.
|
||
"""
|
||
|
||
import io
|
||
import sys
|
||
import pathlib
|
||
import inspect
|
||
import contextlib
|
||
|
||
ROOT = pathlib.Path(__file__).resolve().parent.parent
|
||
sys.path.insert(0, str(ROOT / "src"))
|
||
|
||
from ecw_bot.conditions import medicare_pps_code as pps # noqa: E402
|
||
|
||
CASES = []
|
||
|
||
|
||
def case(fn):
|
||
CASES.append(fn)
|
||
return fn
|
||
|
||
|
||
def _quiet(fn, *a, **k):
|
||
with contextlib.redirect_stdout(io.StringIO()) as out:
|
||
result = fn(*a, **k)
|
||
return result, out.getvalue()
|
||
|
||
|
||
def _row(code, billed=True, fee="120.00"):
|
||
return {"code": code, "billed": billed, "billed_fee": fee, "units": "1"}
|
||
|
||
|
||
def _data(codes, payer="Medicare", screen="ub"):
|
||
cpt = [c if isinstance(c, dict) else _row(c) for c in codes]
|
||
return {"cpt": cpt, "screen": screen,
|
||
"insurance": [{"Name": payer}] if payer is not None else []}
|
||
|
||
|
||
# ── decisions (no browser) ───────────────────────────────────────────────────
|
||
|
||
@case
|
||
def an_office_visit_without_its_pps_code_is_detected():
|
||
fired, _ = _quiet(pps.detect, "C1", _data(["99213"]))
|
||
assert fired is True
|
||
|
||
|
||
def _adds(data):
|
||
return [pps_code for pps_code, _ in pps.decide(data)["add"]]
|
||
|
||
|
||
@case
|
||
def each_visit_kind_maps_to_its_pps_code():
|
||
for codes, want in ((["99203"], "G0466"), (["92002"], "G0466"),
|
||
(["99341"], "G0466H"), (["99213"], "G0467"),
|
||
(["Q0091"], "G0467"), (["99309"], "G0467NH"),
|
||
(["99318"], "G0467NH"), (["99348"], "G0467H"),
|
||
(["99336"], "G0467"), (["G0439"], "G0468")):
|
||
assert _adds(_data(codes)) == [want], (codes, pps.decide(_data(codes)))
|
||
assert pps.decide(_data(["99203"]))["add"] == [("G0466", ["99203"])]
|
||
|
||
|
||
@case
|
||
def an_awv_or_ippe_takes_g0468_even_with_an_office_visit():
|
||
assert _adds(_data(["99213", "G0438"])) == ["G0468"]
|
||
|
||
|
||
@case
|
||
def the_right_pps_code_already_billed_needs_nothing():
|
||
for codes in (["99213", "G0467"], ["99203", "G0466"],
|
||
["99309", "G0467NH"], ["G0439", "G0468"]):
|
||
data = _data(codes)
|
||
fired, _ = _quiet(pps.detect, "C1", data)
|
||
assert fired is False, codes
|
||
assert pps.needs_review("C1", data) is None, codes
|
||
|
||
|
||
@case
|
||
def a_code_that_does_not_end_billed_does_not_qualify():
|
||
assert _adds(_data([_row("99213", billed=False, fee="0.00")])) == []
|
||
assert _adds(_data([_row("99213", fee="0.00")])) == []
|
||
assert _adds(_data(["36415"])) == [], "not on the crosswalk"
|
||
|
||
|
||
def _review(codes, **kw):
|
||
data = _data(codes, **kw)
|
||
fired, _ = _quiet(pps.detect, "C1", data)
|
||
assert fired is False, (codes, pps.decide(data))
|
||
return pps.needs_review("C1", data) or ""
|
||
|
||
|
||
@case
|
||
def codes_on_both_lists_only_are_review_new_or_established():
|
||
for codes in (["G0444"], ["97802"], ["G0444", "G0101"]):
|
||
why = _review(codes)
|
||
assert "new or established" in why and "G0466" in why \
|
||
and "G0467" in why and codes[0] in why, (codes, why)
|
||
|
||
|
||
@case
|
||
def a_code_on_one_list_settles_the_ones_on_both():
|
||
assert _adds(_data(["99213", "G0444"])) == ["G0467"]
|
||
assert pps.decide(_data(["99213", "G0444"]))["add"] == \
|
||
[("G0467", ["99213"])]
|
||
|
||
|
||
@case
|
||
def conflicting_visit_codes_are_review_and_nothing_is_added():
|
||
for codes, a, b in ((["99203", "99213"], "G0466", "G0467"),
|
||
(["99213", "99348"], "G0467", "G0467H")):
|
||
why = _review(codes)
|
||
assert "conflicting" in why and a in why and b in why, (codes, why)
|
||
assert _adds(_data(codes)) == [], codes
|
||
|
||
|
||
@case
|
||
def conflicting_codes_stay_review_with_one_of_their_pps_codes_billed():
|
||
why = _review(["99203", "99213", "G0467"])
|
||
assert "conflicting" in why and "G0466" in why, why
|
||
|
||
|
||
@case
|
||
def a_different_pps_code_billed_is_review_and_not_changed():
|
||
why = _review(["99203", "G0467"])
|
||
assert "G0467 billed" in why and "G0466" in why, why
|
||
|
||
|
||
@case
|
||
def a_different_pps_code_next_to_the_right_one_is_review_too():
|
||
why = _review(["99213", "G0467", "G0466"])
|
||
assert "G0466 billed" in why and "crosswalk says G0467" in why, why
|
||
why = _review(["G0439", "G0468", "G0467"])
|
||
assert "G0467 billed" in why and "crosswalk says G0468" in why, why
|
||
|
||
|
||
@case
|
||
def mental_health_is_review_until_the_client_rules():
|
||
why = _review(["90834"])
|
||
assert "G0469 or G0470" in why and "90834" in why, why
|
||
for billed in ("G0469", "G0470"):
|
||
data = _data(["90834", billed])
|
||
assert pps.needs_review("C1", data) is None, billed
|
||
|
||
|
||
@case
|
||
def hcfa_is_review_by_hand_and_dental_is_skipped():
|
||
why = _review(["99213"], screen="hcfa")
|
||
assert "by hand" in why and "PPS" in why and "G0467" in why, why
|
||
why = _review(["99213", "G0467"], screen="hcfa")
|
||
assert "hcfa" in why and "99213" in why and "by hand" in why, \
|
||
"the wrong form is caught with the PPS code there too"
|
||
assert _review(["36415"], screen="hcfa") == "", "no Qualifying Code"
|
||
data = _data(["99213"], screen="dental")
|
||
assert _quiet(pps.detect, "C1", data)[0] is False
|
||
assert pps.needs_review("C1", data) is None
|
||
|
||
|
||
@case
|
||
def g0468_on_a_payer_whose_history_is_never_checked_is_review():
|
||
"""Operator, 2026-10-23: the Cigna plans are Commercial for #27–#29,
|
||
so their preventive visit history is never read — Review, not an add.
|
||
On Medicare #27–#29 reads it first, so the claim still fires."""
|
||
for payer in ("Cigna Healthspring", "Cigna Health Spring/ Bravo"):
|
||
why = _review(["G0439"], payer=payer)
|
||
assert "G0468 not added" in why and "history not checked" in why \
|
||
and "G0439" in why, (payer, why)
|
||
assert _adds(_data(["G0439"])) == ["G0468"]
|
||
|
||
|
||
@case
|
||
def only_pps_payers_by_their_exact_names_cigna_healthspring_included():
|
||
for payer in ("Medicare", "Medicare-Novitas", "HCFA Medicare Novitas",
|
||
"Humana Medicare", "Devoted MCR UB", "Cigna Healthspring",
|
||
"Cigna Health Spring/ Bravo", " medicare "):
|
||
fired, _ = _quiet(pps.detect, "C1", _data(["99213"], payer=payer))
|
||
assert fired is True, payer
|
||
for payer in ("Medicaid", "Cigna Medicare Access Plan", "Devoted",
|
||
"Humana Claims", "", None):
|
||
data = _data(["99213"], payer=payer)
|
||
fired, _ = _quiet(pps.detect, "C1", data)
|
||
assert fired is False, payer
|
||
assert pps.needs_review("C1", data) is None, payer
|
||
|
||
|
||
# ── apply (browser) ─────────────────────────────────────────────────────────
|
||
|
||
FIXTURE = pathlib.Path(__file__).resolve().parent / "fixtures" / "vaccine_claim.html"
|
||
|
||
from ecw_bot import settings as cfg # noqa: E402
|
||
from ecw_bot.conditions.base import ConditionResult # noqa: E402
|
||
from ecw_bot.pages import claim_form # noqa: E402
|
||
|
||
_VISIT = [{"code": "99213", "title": "OFFICE VISIT EST LEVEL 3", "fee": "120.00"}]
|
||
|
||
|
||
def _fresh(page, rows, payer="Medicare"):
|
||
page.goto(FIXTURE.as_uri())
|
||
page.evaluate("rows => window.seedCptRows(rows)", rows)
|
||
data = claim_form.extract(page)
|
||
data["insurance"] = [{"Name": payer}]
|
||
return data
|
||
|
||
|
||
def _rows_for(page, code):
|
||
return [r for r in claim_form.extract(page)["cpt"] if r["code"] == code]
|
||
|
||
|
||
@case
|
||
def verify_adds_nothing_and_says_which_pps_code_to_add_by_hand(page):
|
||
data = _fresh(page, _VISIT)
|
||
saved = cfg.SAVE_CORRECTIONS
|
||
cfg.SAVE_CORRECTIONS = False
|
||
try:
|
||
result, _ = _quiet(pps.apply, page, "C1", data)
|
||
finally:
|
||
cfg.SAVE_CORRECTIONS = saved
|
||
assert isinstance(result, ConditionResult)
|
||
assert result.changed is False, result
|
||
assert "G0467" in result.review and "99213" in result.review \
|
||
and "by hand" in result.review, result.review
|
||
assert _rows_for(page, "G0467") == []
|
||
|
||
|
||
@case
|
||
def live_write_adds_the_pps_code_once_with_units_1(page):
|
||
result, out = _quiet(pps.apply, page, "C1", _fresh(page, _VISIT))
|
||
assert result.changed is True and result.review == "", (result, out)
|
||
added = _rows_for(page, "G0467")
|
||
assert len(added) == 1 and added[0]["units"] == "1", added
|
||
assert len(claim_form.extract(page)["cpt"]) == 2, "nothing else changed"
|
||
assert pps.summary_comment(result) == \
|
||
"CPT Code Missing (PPS code G0467 added (99213))", \
|
||
pps.summary_comment(result)
|
||
text = pps.describe(result)
|
||
assert "G0467" in text and "99213" in text and "Added" in text, text
|
||
|
||
|
||
@case
|
||
def a_pps_code_the_picker_cannot_find_is_review_add_by_hand(page):
|
||
data = _fresh(page, [{"code": "99348", "title": "HOME VISIT EST",
|
||
"fee": "130.00"}])
|
||
result, _ = _quiet(pps.apply, page, "C1", data)
|
||
assert result.changed is False, result
|
||
assert "G0467H could not be added" in result.review \
|
||
and "99348" in result.review, result.review
|
||
assert _rows_for(page, "G0467H") == []
|
||
|
||
|
||
from ecw_bot.conditions import CONDITIONS # noqa: E402
|
||
from ecw_bot.conditions.base import run_all # noqa: E402
|
||
from ecw_bot.core import charge_summary as cs # noqa: E402
|
||
|
||
|
||
def _outcome(page, data, live_write):
|
||
data["progress_notes"] = {"opened": "modal", "text": "Office visit."}
|
||
saved = cfg.SAVE_CORRECTIONS
|
||
cfg.SAVE_CORRECTIONS = live_write
|
||
try:
|
||
report = {}
|
||
results, out = _quiet(run_all, page, "C1", data, report=report)
|
||
finally:
|
||
cfg.SAVE_CORRECTIONS = saved
|
||
data["_conditions"] = {"fired": [r.condition_name for r in results],
|
||
"saved": False, "failed": report["failed"],
|
||
"review": report["review"],
|
||
"comments": report["comments"]}
|
||
return cs.claim_outcome(True, data), report, out
|
||
|
||
|
||
@case
|
||
def it_is_registered_after_the_wellness_history_and_before_priced_line_check():
|
||
names = [m.NAME for m in CONDITIONS]
|
||
i = names.index("medicare_pps_code")
|
||
assert names.index("medicare_wellness_history") < i < names.index(
|
||
"priced_line_check"), names
|
||
|
||
|
||
@case
|
||
def a_verify_run_makes_the_claim_review_naming_the_pps_code(page):
|
||
outcome, report, out = _outcome(page, _fresh(page, _VISIT), False)
|
||
assert outcome["Category"] == "Review", (outcome, out)
|
||
mine = [r for r in report["review"] if r["name"] == "medicare_pps_code"]
|
||
assert mine and "G0467" in mine[0]["reason"], report["review"]
|
||
assert "G0467" in outcome["Comments"], outcome
|
||
|
||
|
||
@case
|
||
def a_live_write_run_adds_it_and_the_claim_is_fail_saying_so(page):
|
||
outcome, report, out = _outcome(page, _fresh(page, _VISIT), True)
|
||
assert outcome["Category"] == "Fail", (outcome, out)
|
||
assert "PPS code G0467 added (99213)" in outcome["Comments"], outcome
|
||
assert not [r for r in report["review"]
|
||
if r["name"] == "medicare_pps_code"], report["review"]
|
||
assert len(_rows_for(page, "G0467")) == 1
|
||
|
||
|
||
@case
|
||
def g0468_is_held_back_when_the_wellness_history_found_a_problem(page):
|
||
awv = [{"code": "G0439", "title": "ANNUAL WELLNESS VISIT SUBSEQ",
|
||
"fee": "160.00"}]
|
||
for verdict, words in (
|
||
({"read": True, "conflicts": [
|
||
"G0439 already billed on 01/05/2026 (claim 9000001) — once "
|
||
"a year"]}, "already billed"),
|
||
({"read": False, "conflicts": []}, "history not read")):
|
||
data = _fresh(page, awv)
|
||
data["_wellness_history"] = verdict
|
||
result, _ = _quiet(pps.apply, page, "C1", data)
|
||
assert result.changed is False, (verdict, result)
|
||
assert "G0468 not added" in result.review and words in \
|
||
result.review, result.review
|
||
assert _rows_for(page, "G0468") == [], verdict
|
||
data = _fresh(page, awv)
|
||
data["_wellness_history"] = {"read": True, "conflicts": []}
|
||
result, _ = _quiet(pps.apply, page, "C1", data)
|
||
assert result.changed is True and _rows_for(page, "G0468"), result
|
||
|
||
|
||
@case
|
||
def g0468_with_no_wellness_verdict_by_apply_is_not_added(page):
|
||
"""#27–#29 raised or never reached the claim: its history is unchecked."""
|
||
data = _fresh(page, [{"code": "G0439", "title": "ANNUAL WELLNESS VISIT "
|
||
"SUBSEQ", "fee": "160.00"}])
|
||
result, _ = _quiet(pps.apply, page, "C1", data)
|
||
assert result.changed is False, result
|
||
assert "G0468 not added" in result.review, result.review
|
||
assert _rows_for(page, "G0468") == []
|
||
|
||
|
||
def _run_cases() -> int:
|
||
from playwright.sync_api import sync_playwright
|
||
|
||
failures = 0
|
||
with sync_playwright() as pw:
|
||
browser = pw.chromium.launch(headless=True)
|
||
page = browser.new_page()
|
||
for fn in CASES:
|
||
name = fn.__name__.replace("_", " ")
|
||
try:
|
||
if len(inspect.signature(fn).parameters) == 0:
|
||
fn()
|
||
else:
|
||
fn(page)
|
||
print(f" ok {name}")
|
||
except AssertionError as exc:
|
||
print(f" FAIL {name}: {exc}")
|
||
failures += 1
|
||
except Exception as exc:
|
||
print(f" ERROR {name}: {type(exc).__name__}: {exc}")
|
||
failures += 1
|
||
browser.close()
|
||
|
||
print(f"\n{len(CASES) - failures}/{len(CASES)} passed")
|
||
return 1 if failures else 0
|
||
|
||
|
||
def main() -> int:
|
||
"""shot() writes into the operator's Screenshots\\ folder; not here.
|
||
The add cases run as LIVE-WRITE, the way a run that adds does."""
|
||
import tempfile
|
||
from ecw_bot import settings as _cfg
|
||
orig = (_cfg.SCREENSHOTS, _cfg.SAVE_CORRECTIONS)
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
_cfg.SCREENSHOTS = pathlib.Path(tmp)
|
||
_cfg.SAVE_CORRECTIONS = True
|
||
try:
|
||
return _run_cases()
|
||
finally:
|
||
_cfg.SCREENSHOTS, _cfg.SAVE_CORRECTIONS = orig
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|