247 lines
8.1 KiB
Python
247 lines
8.1 KiB
Python
"""pfs.families — code-family registry + deterministic code detection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import duckdb
|
|
|
|
from pfs.codetables import (
|
|
ElementRow,
|
|
EventRow,
|
|
FamilyRow,
|
|
ensure_tables,
|
|
write_families,
|
|
)
|
|
from pfs.families import (
|
|
FAMILIES,
|
|
HAND_FAMILIES,
|
|
Detection,
|
|
derive_families,
|
|
detect_codes,
|
|
family_of,
|
|
find_codes,
|
|
load_families,
|
|
refresh_from,
|
|
stem_tokens,
|
|
)
|
|
|
|
|
|
class TestFindCodes:
|
|
def test_hcpcs_and_cpt(self):
|
|
assert find_codes("Codes G0556 and 99490 apply; see g0557.") == (
|
|
"99490",
|
|
"G0556",
|
|
"G0557",
|
|
)
|
|
|
|
def test_lower_case_codes_are_normalised(self):
|
|
assert find_codes("what is g0556?") == ("G0556",)
|
|
|
|
def test_no_false_hits_on_years_and_fr_pages(self):
|
|
# A Federal Register citation is stripped before scanning, so its page
|
|
# number (43842 — a well-formed CPT code) never reads as a code.
|
|
assert find_codes("91 FR 43842, CY2026") == ()
|
|
|
|
def test_fr_citation_alone_yields_nothing(self):
|
|
assert find_codes("What does 91 FR 43842 say?") == ()
|
|
|
|
def test_dedupes_and_sorts(self):
|
|
assert find_codes("G0558 G0556 G0558") == ("G0556", "G0558")
|
|
|
|
def test_empty(self):
|
|
assert find_codes("") == ()
|
|
|
|
|
|
class TestRegistry:
|
|
def test_apcm_family(self):
|
|
f = FAMILIES["APCM"]
|
|
assert f.codes == ("G0556", "G0557", "G0558")
|
|
assert "advanced primary care management" in f.synonyms
|
|
assert family_of("G0557") is f
|
|
assert family_of("00000") is None
|
|
|
|
def test_all_families_present(self):
|
|
assert set(FAMILIES) == {"ACP", "CCM", "PCM", "TCM", "APCM"}
|
|
|
|
|
|
class TestDetectCodes:
|
|
def test_family_name_expands_to_codes(self):
|
|
d = detect_codes("What is APCM and how is it valued?")
|
|
assert d == Detection(
|
|
codes=("G0556", "G0557", "G0558"), families=("APCM",), explicit=()
|
|
)
|
|
|
|
def test_synonym_case_insensitive_word_boundary(self):
|
|
d = detect_codes("Tell me about Advanced Primary Care Management.")
|
|
assert d.families == ("APCM",)
|
|
assert detect_codes("the apcmx code").families == ()
|
|
|
|
def test_single_member_code_expands_family(self):
|
|
d = detect_codes("How much does G0557 pay?")
|
|
assert d.explicit == ("G0557",)
|
|
assert d.families == ("APCM",)
|
|
assert d.codes == ("G0556", "G0557", "G0558")
|
|
|
|
def test_unregistered_code_stays_alone(self):
|
|
d = detect_codes("value of 99213")
|
|
assert d == Detection(codes=("99213",), families=(), explicit=("99213",))
|
|
|
|
def test_multiple_families(self):
|
|
d = detect_codes("compare CCM and TCM")
|
|
assert d.families == ("CCM", "TCM")
|
|
assert d.codes == tuple(sorted(FAMILIES["CCM"].codes + FAMILIES["TCM"].codes))
|
|
|
|
def test_nothing(self):
|
|
assert detect_codes("what did commenters say about telehealth?") == Detection(
|
|
(), (), ()
|
|
)
|
|
|
|
|
|
def _el(code, type_, value, detail=""):
|
|
return ElementRow(code, 2025, type_, value, detail, "", "", 0, 0, "fr")
|
|
|
|
|
|
def _ev(code, year, kind, frm="", to=""):
|
|
return EventRow(code, year, kind, frm, to, "", 0, 0, "fr", True, "")
|
|
|
|
|
|
class TestStemTokens:
|
|
def test_strips_time_and_actor_noise(self):
|
|
assert stem_tokens("Chrnc care mgmt staff 1st 20") == frozenset(
|
|
{"chrnc", "care", "mgmt"}
|
|
)
|
|
assert stem_tokens("Chrnc care mgmt phys ea addl") == frozenset(
|
|
{"chrnc", "care", "mgmt"}
|
|
)
|
|
assert stem_tokens("Adv prim care mgmt lvl 1") == frozenset(
|
|
{"adv", "prim", "care", "mgmt"}
|
|
)
|
|
|
|
|
|
class TestDerive:
|
|
def test_ccm_reproduced_with_predecessor_and_addon_roles(self):
|
|
elements = {
|
|
"99490": [
|
|
_el("99490", "actor", "clinical-staff-directed"),
|
|
_el("99490", "activity", "comprehensive-care-plan"),
|
|
],
|
|
"99439": [
|
|
_el("99439", "relation", "addon-of", "99490"),
|
|
_el("99439", "relation", "not-with", "99487"),
|
|
],
|
|
"99487": [_el("99487", "activity", "comprehensive-care-plan")],
|
|
"99489": [_el("99489", "relation", "addon-of", "99487")],
|
|
"99491": [
|
|
_el("99491", "actor", "physician-or-qhp-personally"),
|
|
_el("99491", "activity", "comprehensive-care-plan"),
|
|
],
|
|
"99437": [_el("99437", "relation", "addon-of", "99491")],
|
|
"G2058": [_el("G2058", "relation", "addon-of", "99490")],
|
|
"99497": [_el("99497", "activity", "advance-directive-discussion")],
|
|
"99498": [_el("99498", "relation", "addon-of", "99497")],
|
|
}
|
|
events = {
|
|
"G2058": [
|
|
_ev("G2058", 2020, "appeared"),
|
|
_ev("G2058", 2021, "disappeared"),
|
|
_ev("G2058", 2021, "replaced_by", to="99439"),
|
|
],
|
|
"99439": [
|
|
_ev("99439", 2021, "appeared"),
|
|
_ev("99439", 2021, "replaces", frm="G2058"),
|
|
],
|
|
}
|
|
descriptions = {
|
|
"99490": "Chrnc care mgmt staff 1st 20",
|
|
"99439": "Chrnc care mgmt staf ea addl",
|
|
"99487": "Cplx chrnc care 1st 60 min",
|
|
"99489": "Cplx chrnc care ea addl 30",
|
|
"99491": "Chrnc care mgmt phys 1st 30",
|
|
"99437": "Chrnc care mgmt phys ea addl",
|
|
"G2058": "Ccm add 20min",
|
|
"99497": "Advncd care plan 30 min",
|
|
"99498": "Advncd care plan addl 30 min",
|
|
}
|
|
rows = derive_families(elements, events, descriptions)
|
|
ccm = {r.code: r for r in rows if r.key == "CCM"}
|
|
assert set(ccm) == {
|
|
"99490",
|
|
"99439",
|
|
"99487",
|
|
"99489",
|
|
"99491",
|
|
"99437",
|
|
"G2058",
|
|
}
|
|
assert ccm["99439"].role == "add-on" and ccm["G2058"].role == "predecessor"
|
|
assert ccm["G2058"].since == 2020 and ccm["G2058"].until == 2021
|
|
assert ccm["99490"].role == "base"
|
|
acp = {r.code for r in rows if r.key == "ACP"}
|
|
assert acp == {"99497", "99498"}
|
|
assert all(
|
|
r.name == HAND_FAMILIES[r.key].name for r in rows if r.key in HAND_FAMILIES
|
|
)
|
|
|
|
def test_unknown_family_gets_stem_key(self):
|
|
elements = {
|
|
"99453": [_el("99453", "activity", "device-data-review")],
|
|
"99454": [_el("99454", "activity", "device-data-review")],
|
|
}
|
|
rows = derive_families(
|
|
elements,
|
|
{},
|
|
{
|
|
"99453": "Rem mntr physiol param setup",
|
|
"99454": "Rem mntr physiol param dev supl",
|
|
},
|
|
)
|
|
keys = {r.key for r in rows}
|
|
assert len(keys) == 1 and next(iter(keys)).startswith("REM-MNTR")
|
|
|
|
|
|
class TestLoadAndRefresh:
|
|
def test_load_and_refresh_in_place(self):
|
|
from pfs import families as mod
|
|
|
|
con = duckdb.connect(":memory:")
|
|
try:
|
|
ensure_tables(con)
|
|
assert load_families(con) == {}
|
|
before = dict(mod.FAMILIES)
|
|
assert refresh_from(con) == 0 and mod.FAMILIES == before
|
|
write_families(
|
|
con,
|
|
[
|
|
FamilyRow(
|
|
"CCM",
|
|
"Chronic Care Management",
|
|
"99490",
|
|
"base",
|
|
None,
|
|
None,
|
|
"",
|
|
0,
|
|
),
|
|
FamilyRow(
|
|
"CCM",
|
|
"Chronic Care Management",
|
|
"G2058",
|
|
"predecessor",
|
|
2020,
|
|
2021,
|
|
"",
|
|
0,
|
|
),
|
|
],
|
|
)
|
|
n = refresh_from(con)
|
|
assert (
|
|
n == 1
|
|
and set(mod.FAMILIES) == {"CCM"}
|
|
and "G2058" in mod.FAMILIES["CCM"].codes
|
|
)
|
|
# restore for other tests
|
|
mod.FAMILIES.clear()
|
|
mod.FAMILIES.update(before)
|
|
finally:
|
|
con.close()
|