8e7d3637f7
reaction.fr_pairs previously reported n_total equal to n_items (the matching Response: count, always the same number as the family's pair count) — it now reports the rule item's TOTAL Comment: paragraph count, family or not, matching the n_items-out-of-n_total shape the docket rows already use. _COMMENT_RE/_RESPONSE_RE are tightened to no leading whitespace and no space before the colon, so they agree exactly with the LIKE 'Comment:%'/'Response:%' SQL prefilter. Also moves the code-pattern/code-match helpers pfs.guidance and pfs.reaction both need (code_pattern/codes_in) into pfs.families next to find_codes/FR_CITE_RE, as public functions both modules import instead of reaction reaching into guidance's private names.
461 lines
16 KiB
Python
461 lines
16 KiB
Python
"""pfs.guidance — sub-regulatory crosswalk (CFR/IOM/MLN) per code family
|
|
(#689, task 6). Regex snippets are real ``fr_anchors`` text (see
|
|
task-6-report.md for item_key/p_id provenance)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import duckdb
|
|
import pytest
|
|
|
|
from bib.item import Manual, Regulation, Rule
|
|
from bib.store import Store
|
|
from pfs.codetables import (
|
|
GuidanceRow,
|
|
ensure_tables,
|
|
read_guidance,
|
|
write_cpt_edition,
|
|
write_guidance,
|
|
)
|
|
from pfs.cpt_model import CptCode, CptEdition, CptSection
|
|
from pfs.guidance import (
|
|
BARE_CFR_RE,
|
|
CFR_RE,
|
|
IOM_RE,
|
|
MLN_RE,
|
|
_cfr_refs,
|
|
build,
|
|
dedupe,
|
|
harvest,
|
|
harvest_cpt,
|
|
resolve_cfr,
|
|
resolve_iom,
|
|
)
|
|
|
|
# ── real fr_anchors snippets ──────────────────────────────────────────
|
|
|
|
BARE_CFR_TEXT = (
|
|
"Step 3. Review the elements of the service as described by the "
|
|
"HCPCS code and determine whether each of them is capable of being "
|
|
"furnished using an interactive telecommunications system as "
|
|
"defined in § 410.78(a)(3)."
|
|
) # 2KVJ2HKX ¶398
|
|
|
|
IOM_MANUAL_TEXT = (
|
|
"Consistent with the established Medicare HPSA physician bonus "
|
|
"program (Medicare Claims Processing Manual, Pub. 100-04, Chapter "
|
|
"12, Section 90.4.4) and the proposed Health Professional Shortage "
|
|
"Area Surgical Incentive Payment Program (HSIP) described in "
|
|
"section III.S.2. of this proposed rule, we are proposing that "
|
|
"PCIP payments would be calculated by the Medicare contractors."
|
|
) # T4N6JINW ¶1502
|
|
|
|
IOM_PUB_TEXT = (
|
|
"A physician may not bill Medicare for a service that is on the "
|
|
"list of “always therapy” services (see Pub. 100-04, the "
|
|
"Medicare Benefit Policy Manual, chapter 5, section 20) if the "
|
|
"service was done by staff that is not qualified to provide a "
|
|
"skilled therapy service."
|
|
) # WBS77GY5 ¶1771
|
|
|
|
|
|
# ── regexes ─────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestCfrRegex:
|
|
def test_bare_section_infers_no_title(self):
|
|
# BARE_CFR_RE never captures a title — resolve_cfr's caller
|
|
# supplies the inferred "42" (Resolutions: title_inferred kept
|
|
# only in memory, never as a stored flag).
|
|
m = BARE_CFR_RE.search(BARE_CFR_TEXT)
|
|
assert m is not None
|
|
assert m.groups() == ("410", "78", "(a)(3)")
|
|
|
|
def test_titled_cfr_captures_title_part_section_paras(self):
|
|
m = CFR_RE.search("under 42 CFR 410.78(a)(3) and elsewhere")
|
|
assert m is not None
|
|
assert (
|
|
m.group("title"),
|
|
m.group("part"),
|
|
m.group("dec"),
|
|
m.group("paras"),
|
|
) == (
|
|
"42",
|
|
"410",
|
|
"78",
|
|
"(a)(3)",
|
|
)
|
|
assert m.group("partword") is None
|
|
|
|
def test_titled_cfr_part_only(self):
|
|
m = CFR_RE.search("see 42 CFR Part 425 for details")
|
|
assert m is not None
|
|
assert (m.group("title"), m.group("partword"), m.group("part")) == (
|
|
"42",
|
|
"Part",
|
|
"425",
|
|
)
|
|
assert m.group("dec") is None and m.group("paras") is None
|
|
|
|
def test_titled_cfr_plural_parts_only(self):
|
|
m = CFR_RE.search("see 42 CFR parts 405 for details")
|
|
assert m is not None
|
|
assert (m.group("title"), m.group("partword"), m.group("part")) == (
|
|
"42",
|
|
"parts",
|
|
"405",
|
|
)
|
|
|
|
def test_dotted_cfr_with_section_sign_resolves_via_bare_fallback(self):
|
|
# CFR_RE (titled) has no "§" between "C.F.R." and the number —
|
|
# the bare regex's title-inferred fallback picks it up instead,
|
|
# landing on the same (title, section) either way.
|
|
assert CFR_RE.search("42 C.F.R. § 414.1425 governs this") is None
|
|
assert _cfr_refs("42 C.F.R. § 414.1425 governs this") == [("42", "414.1425")]
|
|
|
|
|
|
class TestCfrListContinuation:
|
|
"""Ruling A10: a titled CFR head followed by a list continuation
|
|
(comma/and/or/through-separated bare tokens) captures every
|
|
section/part in the run, not just the head."""
|
|
|
|
def test_comma_list_of_sections(self):
|
|
assert _cfr_refs(
|
|
"the current regulations at 42 CFR 410.20, 410.26, and 410.32 for CCM"
|
|
) == [("42", "410.20"), ("42", "410.26"), ("42", "410.32")]
|
|
|
|
def test_and_joined_paragraph_cites(self):
|
|
assert _cfr_refs("… 42 CFR 410.26(a)(3) and 410.26(b) to better define …") == [
|
|
("42", "410.26(a)(3)"),
|
|
("42", "410.26(b)"),
|
|
]
|
|
|
|
def test_plural_parts_list(self):
|
|
assert _cfr_refs("… 42 CFR parts 405, 414, and 426 …") == [
|
|
("42", "Part 405"),
|
|
("42", "Part 414"),
|
|
("42", "Part 426"),
|
|
]
|
|
|
|
def test_bare_double_section_plural(self):
|
|
assert _cfr_refs("§§ 410.26 and 410.32 govern this") == [
|
|
("42", "410.26"),
|
|
("42", "410.32"),
|
|
]
|
|
|
|
def test_no_continuation_across_prose(self):
|
|
# "the" is not a citation token — the continuation grammar must
|
|
# not fire on ordinary prose following "and".
|
|
assert _cfr_refs("42 CFR 410.78 and the physician") == [("42", "410.78")]
|
|
|
|
def test_singular_part_does_not_trigger_continuation(self):
|
|
# Only a plural "parts" head triggers the bare-part-number
|
|
# continuation grammar (Ruling A10) — a singular "Part 425"
|
|
# followed by an unrelated number must not absorb it.
|
|
assert _cfr_refs("42 CFR Part 425 and 42 other things") == [("42", "Part 425")]
|
|
|
|
|
|
class TestIomRegex:
|
|
def test_manual_name_form(self):
|
|
m = IOM_RE.search(IOM_MANUAL_TEXT)
|
|
assert m is not None
|
|
pub_digit, manual_name, chapter, section = m.groups()
|
|
# The pub-number branch fires (it starts later in the string but
|
|
# is the only branch the Manual-name gap can't cross — "Pub."
|
|
# has a literal period the [^.]{0,120}? gap can't skip past).
|
|
assert pub_digit == "4"
|
|
assert manual_name is None
|
|
assert chapter == "12"
|
|
assert section == "90.4.4"
|
|
|
|
def test_pub_number_form(self):
|
|
m = IOM_RE.search(IOM_PUB_TEXT)
|
|
assert m is not None
|
|
pub_digit, manual_name, chapter, section = m.groups()
|
|
assert pub_digit == "4"
|
|
assert chapter == "5"
|
|
assert section == "20"
|
|
|
|
def test_no_match_without_chapter(self):
|
|
assert IOM_RE.search("Pub. 100-04 discusses billing generally.") is None
|
|
|
|
|
|
class TestMlnRegex:
|
|
def test_matches_booklet_number(self):
|
|
m = MLN_RE.search("see MLN907166 for details")
|
|
assert m is not None
|
|
assert m.group(1) == "907166"
|
|
|
|
def test_case_insensitive_with_space(self):
|
|
assert MLN_RE.search("mln 1234567").group(1) == "1234567"
|
|
|
|
|
|
# ── resolvers ───────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
def store():
|
|
s = Store(":memory:")
|
|
s.create(
|
|
Regulation(
|
|
title="42 CFR Part 410",
|
|
url="https://www.ecfr.gov/current/title-42/part-410",
|
|
)
|
|
)
|
|
sec_key = s.create(
|
|
Regulation(
|
|
title="42 CFR 410.78",
|
|
url="https://www.ecfr.gov/current/title-42/section-410.78",
|
|
)
|
|
)
|
|
manual_key = s.create(
|
|
Manual(
|
|
title=(
|
|
"Medicare Claims Processing Manual — Chapter 12: "
|
|
"Physicians/Nonphysician Practitioners"
|
|
)
|
|
)
|
|
)
|
|
s.add_tag(manual_key, "pub:100-04")
|
|
yield s, sec_key, manual_key
|
|
s.close()
|
|
|
|
|
|
class TestResolveCfr:
|
|
def test_resolves_exact_section(self, store):
|
|
s, sec_key, _ = store
|
|
assert resolve_cfr(s, "42", "410.78(a)(3)") == sec_key
|
|
|
|
def test_unresolved_returns_empty(self, store):
|
|
s, _, _ = store
|
|
assert resolve_cfr(s, "45", "155.20") == ""
|
|
|
|
|
|
class TestResolveIom:
|
|
def test_resolves_chapter_by_pub_and_title(self, store):
|
|
s, _, manual_key = store
|
|
assert resolve_iom(s, "100-04", "12") == manual_key
|
|
|
|
def test_unresolved_pub_returns_empty(self, store):
|
|
s, _, _ = store
|
|
assert resolve_iom(s, "100-02", "12") == ""
|
|
|
|
def test_unresolved_chapter_returns_empty(self, store):
|
|
s, _, _ = store
|
|
assert resolve_iom(s, "100-04", "5") == ""
|
|
|
|
|
|
# ── harvest (FR paragraphs) ────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
def store_with_anchors(store):
|
|
s, sec_key, manual_key = store
|
|
con = s._con() # noqa: SLF001
|
|
item_key = s.create(Rule(title="CY2020 PFS final rule"))
|
|
con.executemany(
|
|
"INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)",
|
|
[
|
|
(item_key, 398, 32389, 1, f"For code 99490, {BARE_CFR_TEXT}"),
|
|
(item_key, 1502, 100, 2, f"Codes 99490 and 99439: {IOM_MANUAL_TEXT}"),
|
|
(item_key, 9, 5, 3, "no codes named here at all"),
|
|
],
|
|
)
|
|
con.commit()
|
|
yield s, item_key, sec_key, manual_key
|
|
s.close()
|
|
|
|
|
|
class TestHarvest:
|
|
def test_bare_cfr_paragraph_resolves(self, store_with_anchors):
|
|
s, item_key, sec_key, _manual_key = store_with_anchors
|
|
rows = harvest(s, ("99490", "99439"), family="CCM")
|
|
cfr_rows = [r for r in rows if r.kind == "cfr"]
|
|
assert len(cfr_rows) == 1
|
|
r = cfr_rows[0]
|
|
assert r.family == "CCM"
|
|
assert r.code == "99490"
|
|
assert r.locator == "42 CFR 410.78(a)(3)"
|
|
assert r.item_key == sec_key
|
|
assert r.item_key_src == item_key
|
|
assert r.p_id_src == 398
|
|
assert r.page_src == 32389
|
|
|
|
def test_iom_paragraph_names_both_codes(self, store_with_anchors):
|
|
s, item_key, _sec_key, manual_key = store_with_anchors
|
|
rows = harvest(s, ("99490", "99439"), family="CCM")
|
|
iom_rows = [r for r in rows if r.kind == "iom"]
|
|
codes = {r.code for r in iom_rows}
|
|
assert codes == {"99490", "99439"}
|
|
assert all(r.locator == "100-04 ch.12 §90.4.4" for r in iom_rows)
|
|
assert all(r.item_key == manual_key for r in iom_rows)
|
|
|
|
def test_paragraph_without_codes_is_skipped(self, store_with_anchors):
|
|
s, _item_key, _sec_key, _manual_key = store_with_anchors
|
|
rows = harvest(s, ("99490", "99439"), family="CCM")
|
|
assert all(r.p_id_src != 9 for r in rows)
|
|
|
|
def test_empty_codes_returns_empty(self, store_with_anchors):
|
|
s, *_rest = store_with_anchors
|
|
assert harvest(s, (), family="CCM") == []
|
|
|
|
def test_fr_citation_page_number_never_fires_as_a_code(self, store_with_anchors):
|
|
# A code equal to an FR citation's page number ("43842") must
|
|
# not be treated as present just because "91 FR 43842" appears.
|
|
s, item_key, _sec_key, _manual_key = store_with_anchors
|
|
con = s._con() # noqa: SLF001
|
|
con.execute(
|
|
"INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)",
|
|
(item_key, 500, 1, 4, "See 91 FR 43842 for background."),
|
|
)
|
|
con.commit()
|
|
rows = harvest(s, ("43842",), family="FAKE")
|
|
assert rows == []
|
|
|
|
|
|
# ── dedupe ──────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestDedupe:
|
|
def test_collapses_identical_keys(self):
|
|
r = GuidanceRow("CCM", "99490", "cfr", "42 CFR 410.78", "K", "SRC", 1, 2)
|
|
assert dedupe([r, r]) == [r]
|
|
|
|
def test_keeps_distinct_codes(self):
|
|
r1 = GuidanceRow("CCM", "99490", "cfr", "42 CFR 410.78", "K", "SRC", 1, 2)
|
|
r2 = GuidanceRow("CCM", "99439", "cfr", "42 CFR 410.78", "K", "SRC", 1, 2)
|
|
assert dedupe([r1, r2]) == [r1, r2]
|
|
|
|
|
|
# ── DDL / write / read round trip ────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
def con():
|
|
c = duckdb.connect(":memory:")
|
|
ensure_tables(c)
|
|
yield c
|
|
c.close()
|
|
|
|
|
|
class TestGuidanceTable:
|
|
def test_write_then_read_round_trips(self, con):
|
|
rows = [
|
|
GuidanceRow(
|
|
"CCM", "99490", "cfr", "42 CFR 410.78(a)(3)", "SEC", "K", 398, 32389
|
|
),
|
|
GuidanceRow(
|
|
"CCM", "99439", "iom", "100-04 ch.12 §90.4.4", "", "K", 1502, 100
|
|
),
|
|
]
|
|
n = write_guidance(con, "CCM", rows)
|
|
assert n == 2
|
|
got = read_guidance(con, "CCM")
|
|
assert {(r.code, r.kind) for r in got} == {("99490", "cfr"), ("99439", "iom")}
|
|
|
|
def test_write_is_delete_then_insert_per_family(self, con):
|
|
write_guidance(
|
|
con, "CCM", [GuidanceRow("CCM", "99490", "cfr", "x", "", "K", 1, 1)]
|
|
)
|
|
write_guidance(
|
|
con, "CCM", [GuidanceRow("CCM", "99439", "cfr", "y", "", "K", 2, 2)]
|
|
)
|
|
got = read_guidance(con, "CCM")
|
|
assert [r.code for r in got] == ["99439"]
|
|
|
|
def test_other_family_untouched(self, con):
|
|
write_guidance(
|
|
con, "APCM", [GuidanceRow("APCM", "G0556", "cfr", "x", "", "K", 1, 1)]
|
|
)
|
|
write_guidance(con, "CCM", [])
|
|
assert len(read_guidance(con, "APCM")) == 1
|
|
|
|
|
|
# ── harvest_cpt (non-empty: a real CFR cite in the guideline text) ───
|
|
|
|
|
|
def _cpt_code(code: str, sec_id: str) -> CptCode:
|
|
return CptCode(
|
|
code=code,
|
|
sec_id=sec_id,
|
|
descriptor="d",
|
|
stem="s",
|
|
elements=(),
|
|
tail="",
|
|
addon=False,
|
|
resequenced=False,
|
|
new=False,
|
|
revised=False,
|
|
telemedicine=False,
|
|
parent="",
|
|
mod51_exempt=False,
|
|
audio_only=False,
|
|
fda_pending=False,
|
|
pla=False,
|
|
category="I",
|
|
)
|
|
|
|
|
|
class TestHarvestCpt:
|
|
def test_guideline_cfr_cite_produces_a_row(self, store, con):
|
|
s, sec_key, _manual_key = store
|
|
edition = CptEdition(
|
|
year=2024,
|
|
sections=(
|
|
CptSection(
|
|
sec_id="sec_1",
|
|
level=2,
|
|
title="Chronic Care Management Services",
|
|
path=(
|
|
"Evaluation and Management",
|
|
"Chronic Care Management Services",
|
|
),
|
|
code_lo="99490",
|
|
code_hi="99490",
|
|
guideline=(
|
|
"See 42 CFR 410.78(a)(3) for telehealth conditions of payment."
|
|
),
|
|
),
|
|
),
|
|
codes=(_cpt_code("99490", "sec_1"),),
|
|
instructions=(),
|
|
references=(),
|
|
crosswalks=(),
|
|
lists=(),
|
|
alternates=(),
|
|
)
|
|
write_cpt_edition(con, edition, "CPTED2024")
|
|
rows = harvest_cpt(con, s, ("99490",), family="CCM")
|
|
assert len(rows) == 1
|
|
r = rows[0]
|
|
assert (r.family, r.code, r.kind, r.locator) == (
|
|
"CCM",
|
|
"99490",
|
|
"cfr",
|
|
"42 CFR 410.78(a)(3)",
|
|
)
|
|
assert r.item_key == sec_key
|
|
assert r.item_key_src == "CPTED2024"
|
|
assert r.p_id_src == 0
|
|
assert r.page_src == 0
|
|
|
|
|
|
# ── build (FR + CPT, no CPT tables ingested) ─────────────────────────
|
|
|
|
|
|
class TestBuild:
|
|
def test_build_tolerates_missing_cpt_tables(self, store_with_anchors):
|
|
# A bare connection (no ensure_tables!) so `cpt_years` actually
|
|
# raises a Catalog "does not exist" error and harvest_cpt's
|
|
# except-is_missing_table_error branch is exercised — the `con`
|
|
# fixture already has an (empty) pfs.cpt_section via
|
|
# ensure_tables, which only exercises the "if not years: return
|
|
# []" branch instead.
|
|
s, *_rest = store_with_anchors
|
|
bare = duckdb.connect(":memory:")
|
|
try:
|
|
rows = build(bare, s, ("99490", "99439"), family="CCM")
|
|
assert rows # the FR-sourced rows still come through
|
|
assert harvest_cpt(bare, s, ("99490",), family="CCM") == []
|
|
finally:
|
|
bare.close()
|