feat(pfs): closed element vocabulary + deterministic descriptor parser (refs #684)

src/pfs/elements.py

@@ -0,0 +1,355 @@
+"""Closed, typed vocabulary for the logical elements of a PFS code
+descriptor, and a deterministic parser for the parts of a descriptor that
+regular expressions can read (minutes, periods, code references, the
+recurring actor / population / activity / modality phrases).
+
+CMS treats a code as a bundle of elements — telehealth Step 3 is "Review
+the elements of the service as described by the HCPCS code" (90 FR 32389).
+Types are fixed here; *values* are a curated list that grows only by
+review (an unknown phrase is queued, see ``pfs.extract``). Pure: no I/O,
+importable inside the ``llm`` container.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from enum import Enum
+
+from pfs.families import find_codes
+
+
+class ElementType(str, Enum):
+    ACTOR = "actor"
+    TIME = "time"
+    PERIOD = "period"
+    POPULATION = "population"
+    ACTIVITY = "activity"
+    MODALITY = "modality"
+    RELATION = "relation"
+    SETTING = "setting"
+    BILLING = "billing"
+
+
+@dataclass(frozen=True, order=True)
+class Element:
+    type: ElementType
+    value: str
+    detail: str = ""  # "20" for TIME, the target code for RELATION, "" otherwise
+
+
+#: Closed value lists. Seeded from the descriptors the FR prints for
+#: 99490/99487/99489/99491/99437/99439, 99495/99496, 99497/99498,
+#: 99424–99427, G0556–G0558, 99441–99443/98008–98016, G2211, 99457/99458.
+VOCAB: dict[ElementType, tuple[str, ...]] = {
+    ElementType.ACTOR: (
+        "clinical-staff-directed",
+        "physician-or-qhp-personally",
+        "rhc-fqhc",
+        "care-team",
+    ),
+    ElementType.TIME: ("first", "each-additional", "at-least", "total"),
+    ElementType.PERIOD: (
+        "calendar-month",
+        "30-days",
+        "per-visit",
+        "14-days-post-discharge",
+        "7-days-post-discharge",
+        "per-day",
+    ),
+    ElementType.POPULATION: (
+        "multiple-chronic-conditions-12-months",
+        "single-high-risk-condition",
+        "significant-risk",
+        "qualified-medicare-beneficiary",
+        "new-patient",
+        "established-patient",
+        "one-or-fewer-chronic-conditions",
+    ),
+    ElementType.ACTIVITY: (
+        "comprehensive-care-plan",
+        "consent",
+        "initiating-visit",
+        "24-7-access",
+        "continuity-of-care",
+        "alternative-care-delivery",
+        "comprehensive-care-management",
+        "needs-assessment",
+        "preventive-services",
+        "medication-reconciliation",
+        "care-transitions",
+        "community-coordination",
+        "asynchronous-communication",
+        "population-data-analysis",
+        "risk-stratification",
+        "performance-measurement",
+        "medical-decision-making",
+        "interactive-contact-post-discharge",
+        "advance-directive-discussion",
+        "device-data-review",
+    ),
+    ElementType.MODALITY: (
+        "face-to-face",
+        "interactive-telecommunications",
+        "audio-video",
+        "audio-only",
+        "asynchronous",
+        "non-face-to-face",
+    ),
+    ElementType.RELATION: (
+        "addon-of",
+        "not-with",
+        "replaces",
+        "replaced-by",
+        "defined-by-reference-to",
+        "crosswalk-valued-to",
+    ),
+    ElementType.SETTING: ("non-facility", "facility", "home", "office"),
+    ElementType.BILLING: (
+        "list-separately",
+        "one-practitioner-per-month",
+        "focal-point-all-primary-care",
+        "once-per-period",
+    ),
+}
+
+_MIN = r"(\d{1,3})\s*(?:minutes|min)"
+_RE_FIRST = re.compile(rf"\b(?:first|initial)\s+{_MIN}", re.I)
+_RE_AT_LEAST = re.compile(rf"\bat least\s+{_MIN}", re.I)
+_RE_EACH_ADDL = re.compile(rf"\beach additional\s+{_MIN}", re.I)
+_RE_TOTAL = re.compile(rf"\btotal (?:time )?(?:of )?{_MIN}", re.I)
+
+_PERIOD_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
+    (re.compile(r"\bper calendar month\b", re.I), "calendar-month"),
+    (re.compile(r"\bper 30 days\b", re.I), "30-days"),
+    (
+        re.compile(r"\bwithin 14 (?:calendar )?days of discharge\b", re.I),
+        "14-days-post-discharge",
+    ),
+    (
+        re.compile(r"\bwithin 7 (?:calendar )?days of discharge\b", re.I),
+        "7-days-post-discharge",
+    ),
+    (re.compile(r"\bper day\b", re.I), "per-day"),
+    (re.compile(r"\bper visit\b", re.I), "per-visit"),
+)
+
+_PHRASES: tuple[tuple[ElementType, str, re.Pattern[str]], ...] = (
+    (
+        ElementType.ACTOR,
+        "clinical-staff-directed",
+        re.compile(r"clinical staff(?: time)?[^.;]{0,40}directed by a physician", re.I),
+    ),
+    (
+        ElementType.ACTOR,
+        "physician-or-qhp-personally",
+        re.compile(r"(?:provided|furnished) personally by a physician", re.I),
+    ),
+    (ElementType.ACTOR, "rhc-fqhc", re.compile(r"\b(?:RHC|FQHC)s?\b")),
+    (
+        ElementType.POPULATION,
+        "multiple-chronic-conditions-12-months",
+        re.compile(r"multiple \(two or more\) chronic conditions", re.I),
+    ),
+    (
+        ElementType.POPULATION,
+        "one-or-fewer-chronic-conditions",
+        re.compile(r"one chronic condition[^;]{0,120}or fewer", re.I),
+    ),
+    (
+        ElementType.POPULATION,
+        "significant-risk",
+        re.compile(r"significant risk of death", re.I),
+    ),
+    (
+        ElementType.POPULATION,
+        "qualified-medicare-beneficiary",
+        re.compile(r"qualified medicare beneficiary", re.I),
+    ),
+    (ElementType.POPULATION, "new-patient", re.compile(r"\bnew patient\b", re.I)),
+    (
+        ElementType.POPULATION,
+        "established-patient",
+        re.compile(r"\bestablished patient\b", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "comprehensive-care-plan",
+        re.compile(r"comprehensive care plan", re.I),
+    ),
+    (ElementType.ACTIVITY, "consent", re.compile(r"\bconsent\b", re.I)),
+    (
+        ElementType.ACTIVITY,
+        "initiating-visit",
+        re.compile(r"initiat\w+ (?:during a )?(?:qualifying )?visit", re.I),
+    ),
+    (ElementType.ACTIVITY, "24-7-access", re.compile(r"24/7 access", re.I)),
+    (
+        ElementType.ACTIVITY,
+        "continuity-of-care",
+        re.compile(r"continuity of care", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "alternative-care-delivery",
+        re.compile(r"alternative ways to traditional office visits", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "comprehensive-care-management",
+        re.compile(r"comprehensive care management", re.I),
+    ),
+    (ElementType.ACTIVITY, "needs-assessment", re.compile(r"needs assessment", re.I)),
+    (
+        ElementType.ACTIVITY,
+        "preventive-services",
+        re.compile(r"preventive services", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "medication-reconciliation",
+        re.compile(r"medication reconciliation", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "care-transitions",
+        re.compile(r"care transitions|transitions? of care", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "community-coordination",
+        re.compile(r"community-based", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "asynchronous-communication",
+        re.compile(r"asynchronous", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "population-data-analysis",
+        re.compile(r"patient population data", re.I),
+    ),
+    (ElementType.ACTIVITY, "risk-stratification", re.compile(r"risk stratif", re.I)),
+    (
+        ElementType.ACTIVITY,
+        "performance-measurement",
+        re.compile(r"performance measurement", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "medical-decision-making",
+        re.compile(r"medical decision making", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "interactive-contact-post-discharge",
+        re.compile(r"interactive contact", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "advance-directive-discussion",
+        re.compile(r"advance directive", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "device-data-review",
+        re.compile(r"physiologic (?:parameter|data)", re.I),
+    ),
+    (
+        ElementType.MODALITY,
+        "interactive-telecommunications",
+        re.compile(r"interactive telecommunications", re.I),
+    ),
+    (
+        ElementType.MODALITY,
+        "audio-video",
+        re.compile(r"audio-video|audio and video", re.I),
+    ),
+    (ElementType.MODALITY, "audio-only", re.compile(r"audio-only|telephone", re.I)),
+    (ElementType.MODALITY, "face-to-face", re.compile(r"(?<!non-)face-to-face", re.I)),
+    (ElementType.MODALITY, "non-face-to-face", re.compile(r"non-face-to-face", re.I)),
+    (ElementType.SETTING, "home", re.compile(r"\bhome visits?\b", re.I)),
+    (
+        ElementType.BILLING,
+        "list-separately",
+        re.compile(r"list separately in addition", re.I),
+    ),
+    (
+        ElementType.BILLING,
+        "one-practitioner-per-month",
+        re.compile(r"only one practitioner can furnish", re.I),
+    ),
+    (
+        ElementType.BILLING,
+        "focal-point-all-primary-care",
+        re.compile(r"continuing focal point", re.I),
+    ),
+)
+
+_RE_IN_CONJUNCTION = re.compile(r"in conjunction with\s+([^)]+)", re.I)
+_RE_NOT_WITH = re.compile(
+    r"do not report[^)]*?(?:same (?:calendar )?month|in conjunction|with)\s+(?:as\s+)?([^)]+)",
+    re.I,
+)
+_RE_ELEMENTS_OF = re.compile(
+    r"with the elements included in\s+([A-Z]\d{4}|\d{5})", re.I
+)
+
+
+def parse_descriptor(text: str) -> tuple[Element, ...]:
+    """Every element a regex can read from *text*, sorted and unique."""
+    found: set[Element] = set()
+    for pat, value in (
+        (_RE_FIRST, "first"),
+        (_RE_EACH_ADDL, "each-additional"),
+        (_RE_TOTAL, "total"),
+    ):
+        for m in pat.finditer(text):
+            found.add(Element(ElementType.TIME, value, m.group(1)))
+    for m in _RE_AT_LEAST.finditer(text):
+        # "at least N minutes" is the first-unit threshold unless an
+        # each-additional phrase already claims N.
+        value = "first" if not _RE_EACH_ADDL.search(text) else "at-least"
+        found.add(Element(ElementType.TIME, value, m.group(1)))
+    for pat, value in _PERIOD_PATTERNS:
+        if pat.search(text):
+            found.add(Element(ElementType.PERIOD, value))
+    for type_, value, pat in _PHRASES:
+        if pat.search(text):
+            found.add(Element(type_, value))
+    for m in _RE_IN_CONJUNCTION.finditer(text):
+        for code in find_codes(m.group(1)):
+            found.add(Element(ElementType.RELATION, "addon-of", code))
+    for m in _RE_NOT_WITH.finditer(text):
+        for code in find_codes(m.group(1)):
+            found.add(Element(ElementType.RELATION, "not-with", code))
+    for m in _RE_ELEMENTS_OF.finditer(text):
+        found.add(
+            Element(ElementType.RELATION, "defined-by-reference-to", m.group(1).upper())
+        )
+    return tuple(sorted(found, key=lambda e: (e.type.value, e.value, e.detail)))
+
+
+_RE_PROSE_START = re.compile(
+    r"^(?:Comment|Response|We|In the|After|For|As|The|CMS|Section)\b"
+)
+
+
+def is_element_paragraph(text: str) -> bool:
+    """True for the short, fragment-like lines the FR prints inside a
+    descriptor (``Consent;``, ``++ Document …``), false for prose."""
+    t = text.strip()
+    if not t or len(t) > 600:
+        return False
+    if t.startswith("++"):
+        return True
+    if _RE_PROSE_START.match(t):
+        return False
+    return t.endswith((";", ")", ").", ":")) or (
+        t[0].islower() is False and t.count(". ") <= 1 and len(t) < 300
+    )
+
+
+def slug(e: Element) -> str:
+    return f"{e.type.value}={e.value}" + (f":{e.detail}" if e.detail else "")

tests/pfs/test_elements.py

@@ -0,0 +1,154 @@
+"""pfs.elements — closed element vocabulary + deterministic descriptor parse."""
+
+from __future__ import annotations
+
+from pfs.elements import (
+    VOCAB,
+    Element,
+    ElementType,
+    is_element_paragraph,
+    parse_descriptor,
+    slug,
+)
+
+CCM_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: multiple (two or more) chronic "
+    "conditions expected to last at least 12 months, or until the death of the patient; "
+    "chronic conditions place the patient at significant risk of death, acute "
+    "exacerbation/decompensation, or functional decline; comprehensive care plan "
+    "established, implemented, revised, or monitored."
+)
+G2058 = (
+    "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 "
+    "(List separately in addition to code for primary procedure). (Use G2058 in conjunction "
+    "with 99490). (Do not report 99490, G2058 in the same calendar month as 99487, 99489, 99491)"
+)
+APCM_G0556_STEM = (
+    "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:"
+)
+
+
+def _has(elements, type_, value, detail=None):
+    return any(
+        e.type == type_ and e.value == value and (detail is None or e.detail == detail)
+        for e in elements
+    )
+
+
+class TestVocab:
+    def test_every_type_has_values(self):
+        for t in ElementType:
+            assert VOCAB[t], t
+
+    def test_values_are_slugs(self):
+        for values in VOCAB.values():
+            for v in values:
+                assert v == v.lower() and " " not in v, v
+
+
+class TestTimeAndPeriod:
+    def test_first_20_minutes_per_calendar_month(self):
+        els = parse_descriptor(CCM_99490)
+        assert _has(els, ElementType.TIME, "first", "20")
+        assert _has(els, ElementType.PERIOD, "calendar-month")
+
+    def test_each_additional(self):
+        els = parse_descriptor(G2058)
+        assert _has(els, ElementType.TIME, "each-additional", "20")
+
+    def test_per_30_days(self):
+        els = parse_descriptor(
+            "Chronic care management, at least 20 minutes, per 30 days"
+        )
+        assert _has(els, ElementType.PERIOD, "30-days")
+
+
+class TestActorAndPopulation:
+    def test_clinical_staff_directed(self):
+        els = parse_descriptor(CCM_99490)
+        assert _has(els, ElementType.ACTOR, "clinical-staff-directed")
+
+    def test_personally_by_physician(self):
+        els = parse_descriptor(
+            "Chronic care management services, provided personally by a physician or other "
+            "qualified health care professional, at least 30 minutes, per calendar month"
+        )
+        assert _has(els, ElementType.ACTOR, "physician-or-qhp-personally")
+
+    def test_two_or_more_chronic_conditions(self):
+        els = parse_descriptor(CCM_99490)
+        assert _has(
+            els, ElementType.POPULATION, "multiple-chronic-conditions-12-months"
+        )
+        assert _has(els, ElementType.POPULATION, "significant-risk")
+
+
+class TestActivityAndRelation:
+    def test_care_plan_activity(self):
+        els = parse_descriptor(CCM_99490)
+        assert _has(els, ElementType.ACTIVITY, "comprehensive-care-plan")
+
+    def test_addon_and_exclusions(self):
+        els = parse_descriptor(G2058)
+        assert _has(els, ElementType.RELATION, "addon-of", "99490")
+        for code in ("99487", "99489", "99491"):
+            assert _has(els, ElementType.RELATION, "not-with", code)
+        assert _has(els, ElementType.BILLING, "list-separately")
+
+    def test_apcm_stem_billing_and_actor(self):
+        els = parse_descriptor(APCM_G0556_STEM)
+        assert _has(els, ElementType.ACTOR, "clinical-staff-directed")
+        assert _has(els, ElementType.PERIOD, "calendar-month")
+        assert _has(els, ElementType.BILLING, "focal-point-all-primary-care")
+
+
+class TestModality:
+    def test_telecommunications(self):
+        els = parse_descriptor(
+            "Synchronous audio-video visit for the evaluation and management of a new patient, "
+            "furnished using an interactive telecommunications system"
+        )
+        assert _has(els, ElementType.MODALITY, "interactive-telecommunications")
+        assert _has(els, ElementType.MODALITY, "audio-video")
+
+    def test_audio_only(self):
+        els = parse_descriptor(
+            "Telephone evaluation and management service, audio-only"
+        )
+        assert _has(els, ElementType.MODALITY, "audio-only")
+
+
+class TestHelpers:
+    def test_sorted_unique(self):
+        els = parse_descriptor(G2058 + " " + G2058)
+        assert list(els) == sorted(
+            set(els), key=lambda e: (e.type.value, e.value, e.detail)
+        )
+
+    def test_element_paragraph_heuristic(self):
+        assert is_element_paragraph("Consent;")
+        assert is_element_paragraph(
+            "++ Document in patient's medical record that consent was obtained."
+        )
+        assert is_element_paragraph(
+            "Comprehensive care plan established, implemented, revised, or monitored)."
+        )
+        assert not is_element_paragraph(
+            "Comment: Several commenters noted that the CPT Editorial Panel created a new code"
+        )
+        assert not is_element_paragraph(
+            "Response: It is our preference to use CPT codes unless Medicare has a programmatic need"
+        )
+        assert not is_element_paragraph(
+            "We proposed that HCPCS codes G0556 through G0558 would describe APCM services furnished per calendar month."
+        )
+
+    def test_slug(self):
+        assert slug(Element(ElementType.TIME, "first", "20")) == "time=first:20"
+        assert slug(Element(ElementType.ACTIVITY, "consent")) == "activity=consent"