224 lines
8.2 KiB
Python
224 lines
8.2 KiB
Python
"""pfs.lineage — dated events from RVU diffs and FR paragraphs, cross-checked."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
|
|
import duckdb
|
|
import pytest
|
|
|
|
from pfs.lineage import _expand_ranges, fr_events, lineage, rvu_events, rvu_year_span
|
|
|
|
RVU_COLS = "hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, non_fac_total DOUBLE, year INTEGER"
|
|
|
|
|
|
@pytest.fixture
|
|
def con():
|
|
c = duckdb.connect(":memory:")
|
|
c.execute("CREATE SCHEMA pfs")
|
|
c.execute(f"CREATE TABLE pfs.rvu ({RVU_COLS})")
|
|
rows = [
|
|
# table spans 2015..2026 (filler code keeps the span honest)
|
|
*[("00000", None, "filler", "A", 1.0, y) for y in range(2015, 2027)],
|
|
# 99487: B in 2015-2016, A from 2017; description rename 2021; +25% revalue 2022
|
|
("99487", None, "Cmplx chron care w/o pt vsit", "B", 0.0, 2015),
|
|
("99487", None, "Cmplx chron care w/o pt vsit", "B", 0.0, 2016),
|
|
("99487", None, "Cmplx chron care w/o pt vsit", "A", 2.00, 2017),
|
|
("99487", None, "Cmplx chron care w/o pt vsit", "A", 2.02, 2018),
|
|
("99487", None, "Cmplx chron care w/o pt vsit", "A", 2.05, 2019),
|
|
("99487", None, "Cmplx chron care w/o pt vsit", "A", 2.10, 2020),
|
|
("99487", None, "Cplx chrnc care 1st 60 min", "A", 2.12, 2021),
|
|
("99487", None, "Cplx chrnc care 1st 60 min", "A", 2.65, 2022),
|
|
*[
|
|
("99487", None, "Cplx chrnc care 1st 60 min", "A", 2.65, y)
|
|
for y in range(2023, 2027)
|
|
],
|
|
("99487", "26", "modifier row ignored", "A", 99.0, 2022),
|
|
# G2058: 2020 only
|
|
("G2058", None, "Ccm add 20min", "A", 1.0, 2020),
|
|
# 99439: 2021 onward
|
|
*[
|
|
("99439", "", "Chrnc care mgmt svc ea addl", "A", 1.0, y)
|
|
for y in range(2021, 2027)
|
|
],
|
|
]
|
|
c.executemany("INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)", rows)
|
|
yield c
|
|
c.close()
|
|
|
|
|
|
class TestSpan:
|
|
def test_span(self, con):
|
|
assert rvu_year_span(con) == (2015, 2026)
|
|
|
|
|
|
class TestRvuEvents:
|
|
def test_status_descriptor_and_revalue(self, con):
|
|
ev = rvu_events(con, "99487")
|
|
kinds = [(e.year, e.kind, e.note) for e in ev]
|
|
assert (2017, "status_change", "B→A") in kinds
|
|
assert (
|
|
2021,
|
|
"descriptor_change",
|
|
"Cmplx chron care w/o pt vsit → Cplx chrnc care 1st 60 min",
|
|
) in kinds
|
|
assert (2022, "revalued", "+25.0%") in kinds
|
|
assert not any(
|
|
k == "appeared" for _, k, _ in kinds
|
|
) # present from the first table year
|
|
assert all(
|
|
e.source == "rvu" and e.anchored is False and e.code == "99487" for e in ev
|
|
)
|
|
|
|
def test_appeared_and_disappeared(self, con):
|
|
ev = rvu_events(con, "G2058")
|
|
assert [(e.year, e.kind) for e in ev] == [
|
|
(2020, "appeared"),
|
|
(2021, "disappeared"),
|
|
]
|
|
|
|
def test_appeared_only(self, con):
|
|
ev = rvu_events(con, "99439")
|
|
assert [(e.year, e.kind) for e in ev] == [(2021, "appeared")]
|
|
|
|
def test_unknown_code(self, con):
|
|
assert rvu_events(con, "99999") == []
|
|
|
|
|
|
class _Store:
|
|
def __init__(self):
|
|
self.con = sqlite3.connect(":memory:")
|
|
self.con.row_factory = sqlite3.Row
|
|
self.con.executescript(
|
|
"CREATE TABLE items (key TEXT PRIMARY KEY, title TEXT, date_published TEXT);"
|
|
"CREATE TABLE fr_anchors (item_key TEXT, p_id INTEGER, page INTEGER, ordinal INTEGER, text TEXT);"
|
|
)
|
|
|
|
def _con(self):
|
|
return self.con
|
|
|
|
def close(self):
|
|
self.con.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def store():
|
|
s = _Store()
|
|
s.con.executemany(
|
|
"INSERT INTO items VALUES (?,?,?)",
|
|
[
|
|
("DE2VH9PD", "Medicare Program; CY 2015 PFS Final Rule", "2014-11-13"),
|
|
(
|
|
"YBM4IZUS",
|
|
"Medicare Program; CY 2021 Payment Policies Under the PFS",
|
|
"2020-12-28",
|
|
),
|
|
(
|
|
"XFGGRBDH",
|
|
"Medicare and Medicaid Programs; CY 2025 Payment Policies",
|
|
"2024-07-31",
|
|
),
|
|
],
|
|
)
|
|
paras = [
|
|
(
|
|
"DE2VH9PD",
|
|
1251,
|
|
67716,
|
|
"Accordingly, we will adopt CPT code 99490 for Medicare CCM services, effective January 1, 2015 instead of the G code.",
|
|
),
|
|
(
|
|
"YBM4IZUS",
|
|
686,
|
|
84547,
|
|
"We also are finalizing our proposal to allow HCPCS code G2058 (which we are finalizing in this rule as new CPT code 99439, see the codes in section II.H.) to be billed concurrently with TCM.",
|
|
),
|
|
(
|
|
"YBM4IZUS",
|
|
1578,
|
|
84639,
|
|
"At the January 2020 RUC meeting, specialty societies requested a temporary crosswalk through CY 2021 between the value established by CMS for HCPCS code G2058 and the value of new CPT code 99439 (with a descriptor identical to G2058).",
|
|
),
|
|
(
|
|
"XFGGRBDH",
|
|
489,
|
|
61652,
|
|
"The CPT Editorial Panel also deleted three codes (99441-99443) for reporting telephone E/M services. We note that CPT codes 99441, 99442, and 99443, each are assigned provisional status on the Medicare telehealth services list, and would return to bundled status when the telehealth flexibilities expire.",
|
|
),
|
|
(
|
|
"XFGGRBDH",
|
|
512,
|
|
61700,
|
|
"The CPT Editorial Panel also deleted three codes (90867-90869) for reporting transcranial magnetic stimulation treatment.",
|
|
),
|
|
]
|
|
s.con.executemany(
|
|
"INSERT INTO fr_anchors VALUES (?,?,?,?,?)",
|
|
[(k, p, pg, p, t) for k, p, pg, t in paras],
|
|
)
|
|
yield s
|
|
s.close()
|
|
|
|
|
|
class TestFrEvents:
|
|
def test_adopted_cpt(self, store):
|
|
ev = fr_events(store, "99490")
|
|
assert [(e.year, e.kind, e.item_key, e.p_id) for e in ev] == [
|
|
(2015, "adopted_cpt", "DE2VH9PD", 1251)
|
|
]
|
|
assert ev[0].anchored is True and ev[0].source == "fr" and ev[0].page == 67716
|
|
assert not any(e.kind == "replaces" for e in ev)
|
|
|
|
def test_replaced_by_and_crosswalk(self, store):
|
|
ev = fr_events(store, "G2058")
|
|
kinds = {e.kind for e in ev}
|
|
assert "created" not in kinds
|
|
assert "replaces" not in kinds
|
|
assert "replaced_by" in kinds and "crosswalk" in kinds
|
|
to_kinds = {(e.year, e.kind, e.to_codes) for e in ev}
|
|
assert (2021, "replaced_by", "99439") in to_kinds
|
|
|
|
def test_replaces_from_the_successor_side(self, store):
|
|
ev = fr_events(store, "99439")
|
|
kinds = {(e.year, e.kind) for e in ev}
|
|
assert (2021, "created") in kinds # ¶686 "new CPT code 99439"
|
|
rep = [e for e in ev if e.kind == "replaces"]
|
|
assert rep and rep[0].from_codes == "G2058"
|
|
|
|
def test_deleted_bundled_and_telehealth(self, store):
|
|
kinds = {e.kind for e in fr_events(store, "99441")}
|
|
assert {"deleted", "bundled", "telehealth_list"} <= kinds
|
|
|
|
def test_deleted_via_range_prefilter_middle_member(self, store):
|
|
# 90868 is a middle member of a "90867-90869" range and appears
|
|
# nowhere else in the paragraph literally — the SQL prefilter must
|
|
# still surface the row for a range-block LIKE match.
|
|
ev = fr_events(store, "90868")
|
|
assert any(e.kind == "deleted" for e in ev)
|
|
|
|
|
|
class TestExpandRanges:
|
|
def test_letter_range(self):
|
|
assert _expand_ranges("G0008-G0010") == "G0008 G0009 G0010"
|
|
|
|
|
|
class TestCrossCheck:
|
|
def test_rvu_events_anchored_by_nearby_fr_event(self, con, store):
|
|
ev = lineage(con, store, "G2058")
|
|
by_kind = {e.kind: e for e in ev}
|
|
assert (
|
|
by_kind["appeared"].year == 2020 and by_kind["appeared"].anchored is True
|
|
) # 2021 is within ±1
|
|
assert (
|
|
by_kind["disappeared"].year == 2021
|
|
and by_kind["disappeared"].anchored is True
|
|
) # 2021 replaced_by
|
|
assert "replaced_by" in by_kind
|
|
assert all(
|
|
not e.anchored for e in lineage(con, store, "99487") if e.source == "rvu"
|
|
)
|
|
|
|
def test_sorted_by_year_then_source(self, con, store):
|
|
ev = lineage(con, store, "99439")
|
|
assert [e.year for e in ev] == sorted(e.year for e in ev)
|