diff --git a/src/pfs/extract.py b/src/pfs/extract.py new file mode 100644 index 0000000..b2de539 --- /dev/null +++ b/src/pfs/extract.py @@ -0,0 +1,116 @@ +"""Element extraction for one code: deterministic parse first, the local +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. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Callable, Sequence + +from pfs.codetables import ElementRow, ReviewRow +from pfs.descriptors import ( + DescriptorRun, + Para, + descriptor_runs, + hcpcs_long_description, + rvu_descriptions, +) +from pfs.elements import VOCAB, Element, ElementType, parse_descriptor + +Classifier = Callable[[str, Sequence[str]], "str | None"] + +#: Types the classifier may assign to a free-standing element line. +_CLASSIFIABLE = (ElementType.ACTIVITY, ElementType.POPULATION, ElementType.BILLING) +_CHOICES: tuple[str, ...] = tuple(v for t in _CLASSIFIABLE for v in VOCAB[t]) +_TYPE_OF: dict[str, ElementType] = {v: t for t in _CLASSIFIABLE for v in VOCAB[t]} + + +@dataclass(frozen=True) +class Extraction: + code: str + rows: tuple[ElementRow, ...] + reviews: tuple[ReviewRow, ...] + + +def _row(code: str, e: Element, para: Para, *, year: int, source: str) -> ElementRow: + return ElementRow( + code, + year, + e.type.value, + e.value, + e.detail, + para.text, + para.item_key, + para.p_id, + para.page, + source, + ) + + +def extract_run( + run: DescriptorRun, *, classify: Classifier | None = None +) -> Extraction: + rows: dict[Element, ElementRow] = {} + reviews: list[ReviewRow] = [] + # 1. deterministic, per element paragraph (anchors to that paragraph) + for para in run.elements: + found = parse_descriptor(para.text) + for e in found: + rows.setdefault(e, _row(run.code, e, para, year=run.rule_year, source="fr")) + if found: + continue + # 2. classifier for lines nothing matched + choice = classify(para.text, _CHOICES) if classify else None + if choice is not None and choice in _TYPE_OF: + e = Element(_TYPE_OF[choice], choice) + rows.setdefault(e, _row(run.code, e, para, year=run.rule_year, source="fr")) + else: + reviews.append( + ReviewRow(run.code, para.text, "", "", para.item_key, para.p_id) + ) + # 3. deterministic on the stem (and anything the whole text reveals) + for e in parse_descriptor(run.text): + rows.setdefault(e, _row(run.code, e, run.stem, year=run.rule_year, source="fr")) + ordered = sorted(rows.values(), key=lambda r: (r.type, r.value, r.detail)) + return Extraction(run.code, tuple(ordered), tuple(reviews)) + + +def extract_text( + code: str, text: str, *, year: int, source: str, classify: Classifier | None = None +) -> Extraction: + rows = [ + ElementRow(code, year, e.type.value, e.value, e.detail, text, "", 0, 0, source) + for e in parse_descriptor(text) + ] + return Extraction(code, tuple(rows), ()) + + +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.""" + code = code.upper() + merged: dict[tuple[str, str, str], ElementRow] = {} + reviews: list[ReviewRow] = [] + for run in descriptor_runs(store, code): + x = extract_run(run, classify=classify) + for r in x.rows: + merged.setdefault((r.type, r.value, r.detail), r) + reviews.extend(x.reviews) + long_desc = hcpcs_long_description(con, code) + years = rvu_descriptions(con, code) + if long_desc: + y = years[-1][0] if years else 0 + for r in extract_text(code, long_desc, year=y, source="hcpcs").rows: + merged.setdefault((r.type, r.value, r.detail), r) + for year, _status, desc in years: + for r in extract_text(code, desc, year=year, source="rvu").rows: + merged.setdefault((r.type, r.value, r.detail), r) + ordered = sorted(merged.values(), key=lambda r: (r.type, r.value, r.detail)) + return Extraction(code, tuple(ordered), tuple(reviews)) diff --git a/tests/pfs/test_extract.py b/tests/pfs/test_extract.py new file mode 100644 index 0000000..50be90c --- /dev/null +++ b/tests/pfs/test_extract.py @@ -0,0 +1,105 @@ +"""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 + )