homelab/stack
1
0
Fork 0
Code Issues 169 Pull Requests 1 Actions Packages Projects 1 Releases 1 Wiki Activity

feat(pfs): CPT Changes years as lineage events; CPT required-elements lists as an extraction source (refs #685 #686)

Browse Source
This commit is contained in:
kert
2026-09-09 20:08:42 -04:00
parent e9c3a5251f
commit c30173c5b4
7 changed files with 479 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/cli/pfs.py
View File

@@ -174,7 +174,12 @@ def elements(
def _print_events(events: list[Any]) -> None:
for e in events:
arrow = f"{e.from_codes or '·'}→{e.to_codes or '·'}"
anchor = f"{e.item_key}{e.p_id}" if e.item_key else f"rvu {e.note}".strip()
if e.source == "cpt":
anchor = f"cpt {e.year} {e.item_key}"
elif e.item_key:
anchor = f"{e.item_key}{e.p_id}"
else:
anchor = f"rvu {e.note}".strip()
flag = "" if e.anchored else " UNANCHORED"
typer.echo(f"{e.year} {e.kind:<18}{arrow:<16}{anchor}{flag}")

11
src/pfs/codetables.py
View File

@@ -46,7 +46,7 @@ class ElementRow:
item_key: str
p_id: int
page: int
source: str # "fr" | "hcpcs" | "rvu"
source: str # "fr" | "cpt" | "hcpcs" | "rvu"
@dataclass(frozen=True)
@@ -69,7 +69,7 @@ class EventRow:
item_key: str
p_id: int
page: int
source: str # "fr" | "rvu"
source: str # "fr" | "rvu" | "cpt"
anchored: bool
note: str
@@ -465,6 +465,13 @@ def read_cpt_instructions(con: Any, year: int) -> list[CptInstructionRow]:
return [CptInstructionRow(*r) for r in rows]
def read_cpt_references(con: Any, year: int) -> list[CptReferenceRow]:
rows = con.execute(
"SELECT * FROM pfs.cpt_reference WHERE edition_year = ? ORDER BY code", [year]
).fetchall()
return [CptReferenceRow(*r) for r in rows]
def cpt_years(con: Any) -> list[int]:
rows = con.execute(
"SELECT DISTINCT edition_year FROM pfs.cpt_section ORDER BY edition_year"

99
src/pfs/extract.py
View File

@@ -4,7 +4,11 @@ model second (closed vocabulary only), a review queue for the rest.
``extract_run`` works on an FR descriptor run (stem + element
paragraphs); ``extract_text`` on a flat descriptor (HCPCS long
description, RVU short description); ``extract_code`` gathers every
source for a code. No I/O here beyond what the caller hands in.
source for a code — FR, then the newest CPT codebook edition's own
``pfs.cpt_code`` row (stem + tail, then each required-elements list
item), then HCPCS, then RVU — with FR rows winning over CPT and CPT
winning over HCPCS/RVU on a duplicate ``(type, value, detail)``. No I/O
here beyond what the caller hands in.
"""
from __future__ import annotations
@@ -88,11 +92,96 @@ def extract_text(code: str, text: str, *, year: int, source: str) -> Extraction:
return Extraction(code, tuple(rows), ())
def _cpt_elements(
con: Any, code: str, *, classify: Classifier | None = None
) -> tuple[list[ElementRow], list[ReviewRow]]:
"""The newest CPT codebook edition's own row for *code*: one pass of
``parse_descriptor`` over ``stem + " " + tail`` (anchored to the
stem), then one pass per required-elements list item (each item's
trailing ``,``/``;`` stripped first) — an item nothing can place goes
to *classify* if given, else to the review queue. The ``parent``
code's stem/elements already apply to add-on rows (the CPT parser
inherits them onto the child's own ``pfs.cpt_code`` row), so nothing
extra is done here for add-ons."""
row = con.execute(
"SELECT edition_year, item_key, stem, elements, tail FROM pfs.cpt_code "
"WHERE code = ? ORDER BY edition_year DESC LIMIT 1",
[code],
).fetchone()
if row is None:
return [], []
edition_year, item_key, stem, elements, tail = row
stem_text = (stem or "").strip()
tail_text = (tail or "").strip()
rows: list[ElementRow] = []
for e in parse_descriptor(f"{stem_text} {tail_text}".strip()):
rows.append(
ElementRow(
code,
edition_year,
e.type.value,
e.value,
e.detail,
stem_text,
item_key,
0,
0,
"cpt",
)
)
reviews: list[ReviewRow] = []
for raw in elements or ():
text = raw.strip()
if text.endswith((",", ";")):
text = text[:-1].strip()
if not text:
continue
found = parse_descriptor(raw)
if found:
for e in found:
rows.append(
ElementRow(
code,
edition_year,
e.type.value,
e.value,
e.detail,
text,
item_key,
0,
0,
"cpt",
)
)
continue
choice = classify(text, _CHOICES) if classify else None
if choice is not None and choice in _TYPE_OF:
e = Element(_TYPE_OF[choice], choice)
rows.append(
ElementRow(
code,
edition_year,
e.type.value,
e.value,
e.detail,
text,
item_key,
0,
0,
"cpt",
)
)
else:
reviews.append(ReviewRow(code, text, "", "", item_key, 0))
return rows, reviews
def extract_code(
store: Any, con: Any, code: str, *, classify: Classifier | None = None
) -> Extraction:
"""All sources for *code*; FR rows win over HCPCS/RVU rows for the same
element, and each source contributes its own year."""
"""All sources for *code*; FR rows win over CPT, and CPT wins over
HCPCS/RVU rows for the same element — each source contributes its own
year."""
code = code.upper()
merged: dict[tuple[str, str, str], ElementRow] = {}
reviews: list[ReviewRow] = []
@@ -101,6 +190,10 @@ def extract_code(
for r in x.rows:
merged.setdefault((r.type, r.value, r.detail), r)
reviews.extend(x.reviews)
cpt_rows, cpt_reviews = _cpt_elements(con, code, classify=classify)
for r in cpt_rows:
merged.setdefault((r.type, r.value, r.detail), r)
reviews.extend(cpt_reviews)
long_desc = hcpcs_long_description(con, code)
years = rvu_descriptions(con, code)
if long_desc:

92
src/pfs/lineage.py
View File

@@ -1,14 +1,22 @@
"""Lineage events for a code: what happened, when, and where the record
says so.
Two independent sources: the RVU files (``pfs.rvu`` base rows year over
Three independent sources: the RVU files (``pfs.rvu`` base rows year over
year — appeared / disappeared / status_change / descriptor_change /
revalued) and the Federal Register (``fr_anchors`` paragraphs that name
the code next to an event verb — created / adopted_cpt / replaces /
replaced_by / deleted / crosswalk / bundled / telehealth_list).
``lineage`` merges them and marks each RVU event ``anchored`` when an FR
event for the same code lies within ±1 rule year; the rest are
``unanchored`` for review, never silently accepted.
revalued), the Federal Register (``fr_anchors`` paragraphs that name the
code next to an event verb — created / adopted_cpt / replaces /
replaced_by / deleted / crosswalk / bundled / telehealth_list), and the
CPT codebook's own "CPT Changes" citations (``pfs.cpt_reference`` kind
``cpt-changes`` on the newest ingested edition's row for the code — one
``cpt_changed`` event per year the AMA itself revised the code, plus a
``cpt_deleted`` event when a code an older edition carried is absent from
the newest one). CPT Changes years are *effective* years — the AMA's own
change year — while FR events are dated by the *mentioning* rule's
payment year; the two can and do disagree, and surfacing that
disagreement is exactly the cross-check this module exists for.
``lineage`` merges all three and marks each RVU event ``anchored`` when
an FR **or** CPT event for the same code lies within ±1 rule year; the
rest are ``unanchored`` for review, never silently accepted.
"""
from __future__ import annotations
@@ -17,7 +25,7 @@ import dataclasses
import re
from typing import Any
from pfs.codetables import EventRow
from pfs.codetables import EventRow, cpt_years
from pfs.descriptors import rule_year_of
from pfs.families import FR_CITE_RE, find_codes
@@ -257,11 +265,73 @@ def fr_events(store: Any, code: str) -> list[EventRow]:
return sorted(out, key=lambda e: (e.year, FR_KINDS.index(e.kind), e.p_id))
def cpt_events(con: Any, code: str) -> list[EventRow]:
"""CPT codebook lineage for *code*: one ``cpt_changed`` event per year
the newest ingested edition's ``cpt-changes`` reference row names
(``cpt_present`` is deliberately not an event — too noisy), or a
single ``cpt_deleted`` event when *code* is present in an older
edition's ``pfs.cpt_code`` but absent from the newest one."""
code = code.upper()
years = cpt_years(con)
if not years:
return []
newest = max(years)
ref = con.execute(
"SELECT item_key, years FROM pfs.cpt_reference "
"WHERE edition_year = ? AND code = ? AND kind = 'cpt-changes'",
[newest, code],
).fetchone()
if ref is not None:
item_key, yrs = ref
return [
EventRow(
code,
int(y),
"cpt_changed",
"",
"",
item_key,
0,
0,
"cpt",
True,
f"CPT Changes {int(y)}",
)
for y in sorted({int(y) for y in (yrs or [])})
]
present = con.execute(
"SELECT edition_year, item_key FROM pfs.cpt_code WHERE code = ? "
"ORDER BY edition_year",
[code],
).fetchall()
if present and int(present[-1][0]) < newest:
last_year, last_key = present[-1]
return [
EventRow(
code,
int(last_year) + 1,
"cpt_deleted",
"",
"",
last_key,
0,
0,
"cpt",
True,
f"absent from CPT {newest}",
)
]
return []
def lineage(con: Any, store: Any, code: str) -> list[EventRow]:
fr = fr_events(store, code)
fr_years = {e.year for e in fr}
cpt = cpt_events(con, code)
anchor_years = {e.year for e in fr} | {e.year for e in cpt}
rvu = [
dataclasses.replace(ev, anchored=any(abs(ev.year - y) <= 1 for y in fr_years))
dataclasses.replace(
ev, anchored=any(abs(ev.year - y) <= 1 for y in anchor_years)
)
for ev in rvu_events(con, code)
]
return sorted([*rvu, *fr], key=lambda e: (e.year, e.source != "fr", e.kind))
return sorted([*rvu, *fr, *cpt], key=lambda e: (e.year, e.source != "fr", e.kind))

20
tests/cli/test_pfs_cli.py
View File

@@ -213,6 +213,26 @@ class TestLineage:
assert read_events(con, "G2058")[0].to_codes == "99439"
assert con.published == [True]
def test_prints_cpt_event_with_cpt_anchor_text(self, con, monkeypatch):
ev = EventRow(
"99490",
2015,
"cpt_changed",
"",
"",
"GQGTPGYV",
0,
0,
"cpt",
True,
"CPT Changes 2015",
)
monkeypatch.setattr(pfs_cli, "lineage", lambda c, s, code: [ev])
res = runner.invoke(app, ["pfs", "lineage", "--code", "99490", "--write"])
assert res.exit_code == 0, res.output
assert "2015 cpt_changed" in res.output
assert "cpt 2015 GQGTPGYV" in res.output
def test_without_write_reads_and_never_takes_the_write_lock(self, con, monkeypatch):
# I4: `stack pfs lineage` without --write must read the replica,
# not open a RW duckdb_batch connection (the DuckDB single-writer

175
tests/pfs/test_extract.py
View File

@@ -4,8 +4,11 @@ from __future__ import annotations
import sqlite3
import duckdb
import pytest
from pfs.descriptors import DescriptorRun, Para, descriptor_runs
from pfs.extract import Extraction, extract_run, extract_text
from pfs.extract import Extraction, extract_code, extract_run, extract_text
STEM = Para(
"JJ6AM5HJ",
@@ -151,6 +154,176 @@ class TestAnchorIntegrity:
con.close()
def _sqlite_store_con():
con = sqlite3.connect(":memory:")
con.row_factory = sqlite3.Row
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);"
)
return con
class _Store:
def __init__(self, con):
self._c = con
def _con(self):
return self._c
def _duckdb_con():
c = duckdb.connect(":memory:")
c.execute("CREATE SCHEMA pfs")
c.execute(
"CREATE TABLE pfs.rvu (hcpcs VARCHAR, mod VARCHAR, description VARCHAR, "
"status_code VARCHAR, non_fac_total DOUBLE, year INTEGER)"
)
c.execute("CREATE SCHEMA terminology")
c.execute(
"CREATE TABLE terminology.hcpcs_level_2 (hcpcs VARCHAR, long_description VARCHAR, "
"seqnum VARCHAR, recid VARCHAR)"
)
c.execute(
"CREATE TABLE pfs.cpt_code (edition_year INTEGER, item_key VARCHAR, code VARCHAR, "
"sec_id VARCHAR, category VARCHAR, descriptor VARCHAR, stem VARCHAR, "
"elements VARCHAR[], tail VARCHAR, parent VARCHAR, addon BOOLEAN, resequenced BOOLEAN, "
"new BOOLEAN, revised BOOLEAN, telemedicine BOOLEAN, mod51_exempt BOOLEAN, "
"audio_only BOOLEAN, fda_pending BOOLEAN, pla BOOLEAN)"
)
return c
def _insert_cpt_code(con, year, item_key, code, *, stem, elements, tail):
con.execute(
"INSERT INTO pfs.cpt_code VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
[
year,
item_key,
code,
"sec1",
"I",
f"{stem}; {tail}",
stem,
list(elements),
tail,
"",
False,
False,
False,
False,
False,
False,
False,
False,
False,
],
)
class TestExtractCodeCpt:
@pytest.fixture
def store(self):
con = _sqlite_store_con()
yield _Store(con)
con.close()
@pytest.fixture
def con(self):
c = _duckdb_con()
yield c
c.close()
def test_cpt_row_yields_source_cpt_rows_and_a_review_row(self, store, con):
_insert_cpt_code(
con,
2024,
"GQGTPGYV",
"99490",
stem="Chronic care management services",
elements=[
"Consent;",
"Something the vocabulary does not know about at all;",
],
tail="first 20 minutes, per calendar month.",
)
x = extract_code(store, con, "99490")
cpt_rows = [r for r in x.rows if r.source == "cpt"]
assert cpt_rows
assert all(r.item_key == "GQGTPGYV" and r.year == 2024 for r in cpt_rows)
assert any(r.value == "consent" for r in cpt_rows)
stem_row = next(r for r in cpt_rows if r.value == "calendar-month")
assert stem_row.text == "Chronic care management services"
assert stem_row.p_id == 0 and stem_row.page == 0
assert [(r.text, r.item_key, r.p_id) for r in x.reviews] == [
(
"Something the vocabulary does not know about at all",
"GQGTPGYV",
0,
)
]
def test_no_cpt_code_row_yields_no_cpt_rows(self, store, con):
x = extract_code(store, con, "99490")
assert not any(r.source == "cpt" for r in x.rows)
assert x.reviews == ()
def test_fr_wins_over_cpt_on_duplicate(self, con):
store_con = _sqlite_store_con()
try:
store_con.execute(
"INSERT INTO items VALUES ('XFGGRBDH', "
"'Medicare and Medicaid Programs; CY 2025 Payment Policies', '2024-07-31')"
)
rows = [
(
900,
60000,
"CPT code 99490 (Chronic care management services, with the "
"following required elements:",
),
(901, 60000, "Consent;"),
]
store_con.executemany(
"INSERT INTO fr_anchors VALUES (?,?,?,?,?)",
[("XFGGRBDH", p, pg, p, t) for p, pg, t in rows],
)
_insert_cpt_code(
con,
2024,
"GQGTPGYV",
"99490",
stem="Chronic care management services",
elements=["Consent;"],
tail="with the following required elements.",
)
x = extract_code(_Store(store_con), con, "99490")
consent_rows = [r for r in x.rows if r.value == "consent"]
assert len(consent_rows) == 1
assert consent_rows[0].source == "fr" and consent_rows[0].p_id == 901
finally:
store_con.close()
def test_cpt_wins_over_hcpcs_on_duplicate(self, store, con):
con.execute(
"INSERT INTO terminology.hcpcs_level_2 VALUES (?,?,?,?)",
["99490", "… per calendar month, consent …", "1", "1"],
)
_insert_cpt_code(
con,
2024,
"GQGTPGYV",
"99490",
stem="Chronic care management services",
elements=["Consent;"],
tail="per calendar month.",
)
x = extract_code(store, con, "99490")
consent_rows = [r for r in x.rows if r.value == "consent"]
assert len(consent_rows) == 1
assert consent_rows[0].source == "cpt"
class TestText:
def test_hcpcs_long_description(self):
x = extract_text(

95
tests/pfs/test_lineage.py
View File

@@ -7,15 +7,64 @@ import sqlite3
import duckdb
import pytest
from pfs.lineage import _expand_ranges, fr_events, lineage, rvu_events, rvu_year_span
from pfs.codetables import ensure_tables
from pfs.lineage import (
_expand_ranges,
cpt_events,
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"
def _cpt_section(con, year, item_key):
con.execute(
"INSERT INTO pfs.cpt_section VALUES (?,?,?,?,?,?,?,?,?,?)",
[year, item_key, "sec1", 1, "T", [], "T", "", "", ""],
)
def _cpt_reference(con, year, item_key, code, years, text="text"):
con.execute(
"INSERT INTO pfs.cpt_reference VALUES (?,?,?,?,?,?)",
[year, item_key, code, "cpt-changes", years, text],
)
def _cpt_code(con, year, item_key, code):
con.execute(
"INSERT INTO pfs.cpt_code VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
[
year,
item_key,
code,
"sec1",
"I",
"d",
"s",
[],
"t",
"",
False,
False,
False,
False,
False,
False,
False,
False,
False,
],
)
@pytest.fixture
def con():
c = duckdb.connect(":memory:")
c.execute("CREATE SCHEMA pfs")
ensure_tables(c)
c.execute(f"CREATE TABLE pfs.rvu ({RVU_COLS})")
rows = [
# table spans 2015..2026 (filler code keeps the span honest)
@@ -219,7 +268,49 @@ class TestExpandRanges:
assert _expand_ranges("G0008-G0010") == "G0008 G0009 G0010"
class TestCptEvents:
def test_cpt_changed_one_per_year_from_newest_edition(self, con):
_cpt_section(con, 2019, "ITEM2019")
_cpt_section(con, 2024, "GQGTPGYV")
_cpt_reference(con, 2019, "ITEM2019", "99490", [2015])
_cpt_reference(con, 2024, "GQGTPGYV", "99490", [2015, 2021, 2022])
ev = cpt_events(con, "99490")
assert [(e.year, e.kind, e.item_key, e.note) for e in ev] == [
(2015, "cpt_changed", "GQGTPGYV", "CPT Changes 2015"),
(2021, "cpt_changed", "GQGTPGYV", "CPT Changes 2021"),
(2022, "cpt_changed", "GQGTPGYV", "CPT Changes 2022"),
]
assert all(e.source == "cpt" and e.anchored is True for e in ev)
assert all(e.code == "99490" for e in ev)
def test_cpt_deleted_when_absent_from_newest_edition(self, con):
_cpt_section(con, 2019, "ITEM2019")
_cpt_section(con, 2024, "GQGTPGYV")
_cpt_code(con, 2019, "ITEM2019", "99999")
ev = cpt_events(con, "99999")
assert [(e.year, e.kind, e.item_key, e.note) for e in ev] == [
(2020, "cpt_deleted", "ITEM2019", "absent from CPT 2024")
]
def test_no_events_for_a_code_absent_everywhere(self, con):
_cpt_section(con, 2024, "GQGTPGYV")
assert cpt_events(con, "00000") == []
def test_no_cpt_tables_yet_is_empty(self, con):
assert cpt_events(con, "99490") == []
class TestCrossCheck:
def test_rvu_event_anchored_by_cpt_changed_within_one_year(self, con, store):
# 99487 has no FR paragraph in `store`, but a cpt_changed at 2016
# (Y-1) must still anchor the 2017 status_change.
_cpt_section(con, 2024, "GQGTPGYV")
_cpt_reference(con, 2024, "GQGTPGYV", "99487", [2016])
ev = lineage(con, store, "99487")
by = {(e.kind, e.year): e for e in ev}
assert by[("status_change", 2017)].anchored is True
assert ("cpt_changed", 2016) in by
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}