202 lines
7.2 KiB
Python
202 lines
7.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 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.",
|
|
),
|
|
]
|
|
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
|
|
|
|
def test_replaced_by_and_crosswalk(self, store):
|
|
ev = fr_events(store, "G2058")
|
|
kinds = {(e.year, e.kind, e.to_codes) for e in ev}
|
|
assert (2021, "replaced_by", "99439") in kinds
|
|
assert (2021, "crosswalk", "") in kinds or any(
|
|
e.kind == "crosswalk" for e in ev
|
|
)
|
|
|
|
def test_replaces_from_the_successor_side(self, store):
|
|
ev = fr_events(store, "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
|
|
|
|
|
|
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)
|