106 lines
3.8 KiB
Python
106 lines
3.8 KiB
Python
"""pfs.extract — deterministic parse + injected classifier → rows and review queue."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pfs.descriptors import DescriptorRun, Para
|
|
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 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
|
|
)
|