c8ba1bbefa
`_stem_pattern`'s `re.I` made `[A-Z]` match any letter, so an enumeration item like "( 9) 99439 (code for non-complex chronic care management)." read as the opening of 99439's own descriptor; `descriptor_runs` then consumed the following list item as an element paragraph, and `extract_run` minted rows anchored to a paragraph naming a different code entirely. Live table already held wrong rows from this (an APCM paragraph under 99439, a TCM enumeration line under 99490). Keep the stem match case-insensitive (some rules print descriptors lowercase) but reject the cross-reference form with a negative lookahead on "code(s)". `is_element_paragraph` also rejects lines that open with an enumeration marker or contain "code(s) for" — the same enumeration items were passing its ")."-ending heuristic too. Claude-Session: https://claude.ai/code/session_01Aum3pEMAM3yQVdFSdVe6Gc
166 lines
6.3 KiB
Python
166 lines
6.3 KiB
Python
"""pfs.extract — deterministic parse + injected classifier → rows and review queue."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
|
|
from pfs.descriptors import DescriptorRun, Para, descriptor_runs
|
|
from pfs.extract import Extraction, extract_run, extract_text
|
|
|
|
STEM = Para(
|
|
"JJ6AM5HJ",
|
|
1163,
|
|
97864,
|
|
"HCPCS code G0556 ( Advanced primary care management services provided by clinical staff and directed by a "
|
|
"physician or other qualified health care professional who is responsible for all primary care and serves as "
|
|
"the continuing focal point for all needed health care services, per calendar month, with the following elements, as appropriate:",
|
|
)
|
|
ELS = (
|
|
Para("JJ6AM5HJ", 1164, 97864, "Consent;"),
|
|
Para(
|
|
"JJ6AM5HJ",
|
|
1168,
|
|
97864,
|
|
"Provide 24/7 access for urgent needs to care team/practitioner;",
|
|
),
|
|
Para(
|
|
"JJ6AM5HJ",
|
|
1170,
|
|
97864,
|
|
"Deliver care in alternative ways to traditional office visits to best meet the patient's needs, such as home visits and/or expanded hours;",
|
|
),
|
|
Para(
|
|
"JJ6AM5HJ",
|
|
1185,
|
|
97865,
|
|
"Be assessed through performance measurement of primary care quality, total cost of care, and meaningful use of Certified EHR Technology).",
|
|
),
|
|
Para(
|
|
"JJ6AM5HJ", 1199, 97865, "Something the vocabulary does not know about at all;"
|
|
),
|
|
)
|
|
RUN = DescriptorRun("G0556", "JJ6AM5HJ", 2025, STEM, ELS)
|
|
|
|
|
|
def _vals(x, type_):
|
|
return sorted(r.value for r in x.rows if r.type == type_)
|
|
|
|
|
|
class TestDeterministic:
|
|
def test_rows_anchor_to_the_paragraph_that_contains_them(self):
|
|
x = extract_run(RUN)
|
|
assert isinstance(x, Extraction) and x.code == "G0556"
|
|
by_value = {r.value: r for r in x.rows}
|
|
assert by_value["consent"].p_id == 1164
|
|
assert by_value["24-7-access"].p_id == 1168
|
|
assert by_value["performance-measurement"].p_id == 1185
|
|
assert by_value["clinical-staff-directed"].p_id == 1163 # stem-level
|
|
assert (
|
|
by_value["calendar-month"].source == "fr"
|
|
and by_value["calendar-month"].year == 2025
|
|
)
|
|
|
|
def test_unknown_paragraph_goes_to_review_without_classifier(self):
|
|
x = extract_run(RUN)
|
|
assert [r.p_id for r in x.reviews] == [1199]
|
|
assert x.reviews[0].proposed_value == ""
|
|
|
|
def test_home_setting_from_element_line(self):
|
|
assert "home" in _vals(extract_run(RUN), "setting")
|
|
|
|
|
|
class TestClassifier:
|
|
def test_classifier_places_unknown_line(self):
|
|
seen = []
|
|
|
|
def classify(text, choices):
|
|
seen.append((text, tuple(choices)))
|
|
return "community-coordination" if "vocabulary" in text else None
|
|
|
|
x = extract_run(RUN, classify=classify)
|
|
assert x.reviews == ()
|
|
row = next(r for r in x.rows if r.value == "community-coordination")
|
|
assert row.type == "activity" and row.p_id == 1199
|
|
assert seen and "consent" in seen[0][1] and "significant-risk" in seen[0][1]
|
|
|
|
def test_classifier_none_means_review_with_no_proposal(self):
|
|
x = extract_run(RUN, classify=lambda t, c: None)
|
|
assert len(x.reviews) == 1 and x.reviews[0].proposed_value == ""
|
|
|
|
def test_classifier_not_called_for_deterministic_lines(self):
|
|
calls = []
|
|
extract_run(RUN, classify=lambda t, c: calls.append(t) or None)
|
|
assert calls == ["Something the vocabulary does not know about at all;"]
|
|
|
|
|
|
class TestAnchorIntegrity:
|
|
def test_no_row_anchored_to_a_foreign_code_paragraph(self):
|
|
# C1 live-corpus incident: an enumeration/cross-reference line for
|
|
# a *different* code ("( 9) 99457 and 99458 (codes for remote
|
|
# physiologic monitoring, each additional 20 minutes).") must
|
|
# never contribute an ElementRow to a 99439 run. The fix lives in
|
|
# `descriptor_runs`/`is_element_paragraph` (it stops the run
|
|
# before absorbing such a line) — this test drives the real
|
|
# `descriptor_runs` to build the run, then checks `extract_run`'s
|
|
# output end to end: no row may be anchored to a paragraph whose
|
|
# text contains a different 5-char code and none of 99439.
|
|
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);"
|
|
)
|
|
con.execute(
|
|
"INSERT INTO items VALUES ('XFGGRBDH', "
|
|
"'Medicare and Medicaid Programs; CY 2025 Payment Policies', '2024-07-31')"
|
|
)
|
|
rows = [
|
|
(
|
|
832,
|
|
61652,
|
|
"CPT code 99439 (Chronic care management services, each additional 20 "
|
|
"minutes of clinical staff time directed by a physician or other "
|
|
"qualified health care professional, per calendar month, with the "
|
|
"following required elements:",
|
|
),
|
|
(833, 61652, "Consent;"),
|
|
(
|
|
834,
|
|
61652,
|
|
"( 9) 99457 and 99458 (codes for remote physiologic monitoring, "
|
|
"each additional 20 minutes).",
|
|
),
|
|
]
|
|
con.executemany(
|
|
"INSERT INTO fr_anchors VALUES (?,?,?,?,?)",
|
|
[("XFGGRBDH", p, pg, p, t) for p, pg, t in rows],
|
|
)
|
|
|
|
class _S:
|
|
def _con(self):
|
|
return con
|
|
|
|
try:
|
|
run = descriptor_runs(_S(), "99439")[0]
|
|
# the enumeration line for 99457/99458 never joins the run
|
|
assert [p.p_id for p in run.elements] == [833]
|
|
x = extract_run(run)
|
|
for r in x.rows:
|
|
assert "99457" not in r.text and "99458" not in r.text
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
class TestText:
|
|
def test_hcpcs_long_description(self):
|
|
x = extract_text(
|
|
"G0556",
|
|
"… per calendar month, with the following elements: consent; 24/7 access …",
|
|
year=2025,
|
|
source="hcpcs",
|
|
)
|
|
assert {r.value for r in x.rows} >= {"calendar-month", "consent", "24-7-access"}
|
|
assert all(
|
|
r.item_key == "" and r.p_id == 0 and r.source == "hcpcs" for r in x.rows
|
|
)
|