feat(pfs): descriptor sources — FR stem+element paragraph runs, HCPCS long and RVU short descriptors (refs #685)

src/pfs/descriptors.py

@@ -0,0 +1,127 @@
+"""Where a code's descriptor text comes from.
+
+1. The Federal Register prints a descriptor as a *stem* paragraph
+   ("…CPT code 99490 (Chronic care management services … with the
+   following required elements:") followed by one paragraph per element
+   (``fr_anchors`` keeps them as consecutive ``p_id`` rows — CY2015 final
+   rule ¶1244–1247 for 99490, CY2025 final rule ¶1163–1185 for G0556).
+2. ``terminology.hcpcs_level_2.long_description`` for HCPCS level II
+   codes (G-codes carry the full element list; CPT long descriptors are
+   not distributed).
+3. ``pfs.rvu.description`` — the short descriptor per year (tracks
+   renames such as "Chron care mgmt srvc 20 min" → "Chrnc care mgmt
+   staff 1st 20").
+
+Function-level ``bib`` access only (a ``Store``-like object exposing
+``_con()``) so this module imports without SQLite files present.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from typing import Any
+
+from pfs.elements import is_element_paragraph
+
+
+@dataclass(frozen=True)
+class Para:
+    item_key: str
+    p_id: int
+    page: int
+    text: str
+
+
+@dataclass(frozen=True)
+class DescriptorRun:
+    code: str
+    item_key: str
+    rule_year: int
+    stem: Para
+    elements: tuple[Para, ...]
+
+    @property
+    def text(self) -> str:
+        return " ".join([self.stem.text, *(p.text for p in self.elements)])
+
+
+_RE_CY = re.compile(r"\bCY\s?(\d{4})\b")
+
+
+def rule_year_of(title: str, date_published: str) -> int:
+    """The payment year a rule governs: ``CY nnnn`` in the title, else the
+    publication year + 1 (PFS rules publish in the second half of the
+    prior year)."""
+    m = _RE_CY.search(title or "")
+    if m:
+        return int(m.group(1))
+    return int((date_published or "1970")[:4]) + 1
+
+
+def _stem_pattern(code: str) -> re.Pattern[str]:
+    # "code 99490 (Chronic…", "CPT code 99490 (", "HCPCS code G0556 ( Advanced…"
+    return re.compile(rf"\b{re.escape(code)}\s*\(\s*[A-Z]", re.I)
+
+
+def descriptor_runs(
+    store: Any, code: str, *, max_elements: int = 40
+) -> list[DescriptorRun]:
+    """Every FR paragraph that opens *code*'s descriptor, with the element
+    paragraphs that follow it (stops at the first prose paragraph)."""
+    con = store._con()
+    pat = _stem_pattern(code.upper())
+    stems = con.execute(
+        "SELECT a.item_key, a.p_id, a.page, a.text, i.title, i.date_published "
+        "FROM fr_anchors a JOIN items i ON i.key = a.item_key "
+        "WHERE a.text LIKE ? ORDER BY i.date_published, a.p_id",
+        (f"%{code.upper()}%",),
+    ).fetchall()
+    runs: list[DescriptorRun] = []
+    for row in stems:
+        if not pat.search(row["text"]):
+            continue
+        stem = Para(row["item_key"], row["p_id"], row["page"], row["text"])
+        following = con.execute(
+            "SELECT p_id, page, text FROM fr_anchors WHERE item_key = ? AND p_id > ? "
+            "ORDER BY p_id LIMIT ?",
+            (stem.item_key, stem.p_id, max_elements),
+        ).fetchall()
+        elements: list[Para] = []
+        for f in following:
+            if not is_element_paragraph(f["text"]):
+                break
+            elements.append(Para(stem.item_key, f["p_id"], f["page"], f["text"]))
+            if f["text"].rstrip().endswith((")", ").")):
+                break  # closing parenthesis ends the descriptor
+        runs.append(
+            DescriptorRun(
+                code=code.upper(),
+                item_key=stem.item_key,
+                rule_year=rule_year_of(row["title"], row["date_published"]),
+                stem=stem,
+                elements=tuple(elements),
+            )
+        )
+    return runs
+
+
+def hcpcs_long_description(con: Any, code: str) -> str:
+    row = con.execute(
+        "SELECT long_description FROM terminology.hcpcs_level_2 WHERE hcpcs = ? LIMIT 1",
+        [code.upper()],
+    ).fetchone()
+    return (row[0] or "") if row else ""
+
+
+def rvu_descriptions(con: Any, code: str) -> list[tuple[int, str, str]]:
+    """(year, status_code, description) for the base (no-modifier) row of
+    each year, ascending."""
+    rows = con.execute(
+        "SELECT year, status_code, description FROM pfs.rvu "
+        "WHERE hcpcs = ? AND (mod IS NULL OR mod = '') "
+        "QUALIFY row_number() OVER (PARTITION BY year ORDER BY mod NULLS FIRST) = 1 "
+        "ORDER BY year",
+        [code.upper()],
+    ).fetchall()
+    return [(int(y), s or "", d or "") for y, s, d in rows]

tests/pfs/test_descriptors.py

@@ -0,0 +1,159 @@
+"""pfs.descriptors — where descriptor text comes from."""
+
+from __future__ import annotations
+
+import sqlite3
+
+import duckdb
+import pytest
+
+from pfs.descriptors import (
+    DescriptorRun,
+    descriptor_runs,
+    hcpcs_long_description,
+    rule_year_of,
+    rvu_descriptions,
+)
+
+
+class _Store:
+    """Minimal stand-in for bib.Store: only ``_con()`` is used."""
+
+    def __init__(self) -> None:
+        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
+
+
+@pytest.fixture
+def store():
+    s = _Store()
+    s.con.execute(
+        "INSERT INTO items VALUES ('DE2VH9PD', 'Medicare Program; CY 2015 PFS Final Rule', '2014-11-13')"
+    )
+    rows = [
+        (
+            "DE2VH9PD",
+            1243,
+            67716,
+            "Comment: commenters noted the CPT panel created a code.",
+        ),
+        (
+            "DE2VH9PD",
+            1244,
+            67716,
+            "These commenters suggested that we use the new CPT code 99490 (Chronic care management services, at least 20 minutes of clinical staff time directed by a physician or other qualified health care professional, per calendar month, with the following required elements:",
+        ),
+        (
+            "DE2VH9PD",
+            1245,
+            67716,
+            "Multiple (two or more) chronic conditions expected to last at least 12 months, or until the death of the patient;",
+        ),
+        (
+            "DE2VH9PD",
+            1246,
+            67716,
+            "Chronic conditions place the patient at significant risk of death, acute exacerbation/decompensation, or functional decline;",
+        ),
+        (
+            "DE2VH9PD",
+            1247,
+            67716,
+            "Comprehensive care plan established, implemented, revised, or monitored).",
+        ),
+        (
+            "DE2VH9PD",
+            1248,
+            67716,
+            "Many of these commenters expressed a preference for the per calendar month period.",
+        ),
+        ("DE2VH9PD", 1249, 67716, "Response: It is our preference to use CPT codes."),
+    ]
+    s.con.executemany(
+        "INSERT INTO fr_anchors VALUES (?,?,?,?,?)",
+        [(k, p, pg, p, t) for k, p, pg, t in rows],
+    )
+    return s
+
+
+@pytest.fixture
+def con():
+    c = duckdb.connect(":memory:")
+    c.execute("CREATE SCHEMA terminology; CREATE SCHEMA pfs")
+    c.execute(
+        "CREATE TABLE terminology.hcpcs_level_2 (hcpcs VARCHAR, seqnum VARCHAR, recid VARCHAR, long_description VARCHAR, short_description VARCHAR)"
+    )
+    c.execute(
+        "INSERT INTO terminology.hcpcs_level_2 VALUES ('G0556','10','3','Advanced primary care management services … per calendar month','Adv prim care mgmt lvl 1')"
+    )
+    c.execute(
+        "CREATE TABLE pfs.rvu (hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, year INTEGER)"
+    )
+    c.executemany(
+        "INSERT INTO pfs.rvu VALUES (?,?,?,?,?)",
+        [
+            ("99490", None, "Chron care mgmt srvc 20 min", "A", 2015),
+            ("99490", "26", "ignored modifier row", "A", 2015),
+            ("99490", "", "Chrnc care mgmt staff 1st 20", "A", 2022),
+        ],
+    )
+    return c
+
+
+class TestRuleYear:
+    def test_from_title(self):
+        assert (
+            rule_year_of("Medicare Program; CY 2015 PFS Final Rule", "2014-11-13")
+            == 2015
+        )
+        assert (
+            rule_year_of(
+                "Medicare and Medicaid Programs; CY 2027 Payment Policies", "2026-07-16"
+            )
+            == 2027
+        )
+
+    def test_fallback_to_date_plus_one(self):
+        assert rule_year_of("Revisions to Payment Policies", "2017-07-21") == 2018
+
+
+class TestDescriptorRuns:
+    def test_stem_and_following_elements(self, store):
+        runs = descriptor_runs(store, "99490")
+        assert len(runs) == 1
+        run = runs[0]
+        assert isinstance(run, DescriptorRun)
+        assert run.item_key == "DE2VH9PD" and run.rule_year == 2015
+        assert run.stem.p_id == 1244
+        assert [p.p_id for p in run.elements] == [1245, 1246, 1247]
+        assert (
+            "per calendar month" in run.text and "Comprehensive care plan" in run.text
+        )
+
+    def test_no_stem_no_run(self, store):
+        assert descriptor_runs(store, "99491") == []
+
+    def test_max_elements_cap(self, store):
+        run = descriptor_runs(store, "99490", max_elements=2)[0]
+        assert [p.p_id for p in run.elements] == [1245, 1246]
+
+
+class TestDuckDBSources:
+    def test_long_description(self, con):
+        assert hcpcs_long_description(con, "G0556").startswith("Advanced primary care")
+        assert hcpcs_long_description(con, "99490") == ""
+
+    def test_rvu_descriptions_base_rows_only(self, con):
+        assert rvu_descriptions(con, "99490") == [
+            (2015, "A", "Chron care mgmt srvc 20 min"),
+            (2022, "A", "Chrnc care mgmt staff 1st 20"),
+        ]