748 lines
25 KiB
Python
748 lines
25 KiB
Python
"""pfs.origin — origin and formation rows (dossier task 6).
|
|
|
|
The EPUBs below are built in ``tmp_path`` from synthetic strings: the repo
|
|
holds no AMA text.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import zipfile
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import duckdb
|
|
import pytest
|
|
|
|
from bib.item import Rule
|
|
from bib.store import Store
|
|
from pfs.codetables import (
|
|
ORIGIN_KINDS,
|
|
GuidanceRow,
|
|
OriginRow,
|
|
ensure_tables,
|
|
read_origin,
|
|
write_guidance,
|
|
write_origin,
|
|
)
|
|
from pfs.origin import (
|
|
ORIGIN_PROMPT_VERSION,
|
|
ORIGIN_SCHEMA,
|
|
book_candidates,
|
|
extract_origin,
|
|
family_origin,
|
|
fr_origin_candidates,
|
|
guidance_origins,
|
|
origin_inventory,
|
|
origins_from_output,
|
|
resolve_book,
|
|
shares_run,
|
|
)
|
|
from pfs.runlog import RunLog
|
|
|
|
FAM = SimpleNamespace(
|
|
key="HOME-VISITS",
|
|
codes=("99341", "99342", "99350"),
|
|
synonyms=("home or residence services",),
|
|
)
|
|
CODES = FAM.codes
|
|
|
|
_CONTAINER = (
|
|
'<?xml version="1.0"?><container version="1.0" '
|
|
'xmlns="urn:oasis:names:tc:opendocument:xmlns:container"><rootfiles>'
|
|
'<rootfile full-path="OPS/book.opf" media-type="application/oebps-package+xml"/>'
|
|
"</rootfiles></container>"
|
|
)
|
|
_OPF = (
|
|
'<?xml version="1.0"?><package xmlns="http://www.idpf.org/2007/opf" version="3.0">'
|
|
'<manifest><item id="c1" href="c1.xhtml" media-type="application/xhtml+xml"/>'
|
|
'</manifest><spine><itemref idref="c1"/></spine></package>'
|
|
)
|
|
|
|
|
|
def make_epub(path: Path, pages: dict[str, str]) -> Path:
|
|
body = "".join(
|
|
f'<span id="page_{label}"/><p>{text}</p>' for label, text in pages.items()
|
|
)
|
|
with zipfile.ZipFile(path, "w") as zf:
|
|
zf.writestr("mimetype", "application/epub+zip")
|
|
zf.writestr("META-INF/container.xml", _CONTAINER)
|
|
zf.writestr("OPS/book.opf", _OPF)
|
|
zf.writestr("OPS/c1.xhtml", f"<html><body>{body}</body></html>")
|
|
return path
|
|
|
|
|
|
# synthetic stand-ins; the sentences are invented for the tests
|
|
BOOK_PAGES = {
|
|
"ix": "Preface with no code at all.",
|
|
"12": (
|
|
"The panel adopted new home visit family entries beginning with 99341 "
|
|
"and 99350 for 2023, and the former domiciliary codes were deleted."
|
|
),
|
|
"13": "Another page about 99342 only in an unrelated table.",
|
|
}
|
|
|
|
FR_STATUTE = (
|
|
"Section 1861(m) of the Act establishes the home health benefit, and CPT "
|
|
"code 99341 describes a home visit for a new patient."
|
|
)
|
|
FR_RECAP = (
|
|
"In the CY 2019 PFS final rule, we established the homebound exception "
|
|
"for CPT code 99342."
|
|
)
|
|
FR_REQUEST = (
|
|
"Commenters requested that section 1834(m) of the Act cover CPT code 99341."
|
|
)
|
|
FR_HYPO = (
|
|
"If we were to repeal section 1862(a)(1) of the Act, CPT code 99350 "
|
|
"would need to be revised."
|
|
)
|
|
FR_PLAIN = "CPT code 99350 is not subject to the demonstration."
|
|
|
|
|
|
def ask_with(items, calls=None):
|
|
def ask(system, user, schema):
|
|
if calls is not None:
|
|
calls.append((system, user, schema))
|
|
return {"origins": items}
|
|
|
|
ask.model = "qwen-test"
|
|
return ask
|
|
|
|
|
|
def item(**kw):
|
|
base = dict(
|
|
kind="statute",
|
|
reference="Section 1861(m) of the Act",
|
|
year=None,
|
|
codes=["99341"],
|
|
summary="The Act sets up the home health benefit.",
|
|
)
|
|
return {**base, **kw}
|
|
|
|
|
|
# ── pure guards ──────────────────────────────────────────────────────
|
|
|
|
|
|
def test_shares_run_is_case_and_punctuation_insensitive():
|
|
text = "The Panel, in 2022, adopted these Home-Visit entries for everyone."
|
|
assert shares_run(
|
|
"it said: the panel in 2022 adopted these home visit entries", text
|
|
)
|
|
assert not shares_run("the panel adopted entries about home visits", text)
|
|
assert not shares_run("too short", "too short") # fewer words than n
|
|
|
|
|
|
def test_book_summary_copy_guard():
|
|
page = BOOK_PAGES["12"]
|
|
copied = item(
|
|
kind="cpt_panel",
|
|
reference="CPT Editorial Panel",
|
|
summary="The panel adopted new home visit family entries beginning with 99341.",
|
|
)
|
|
fresh = item(
|
|
kind="cpt_panel",
|
|
reference="CPT Editorial Panel",
|
|
codes=["99350"],
|
|
summary="Editors created a fresh family of home visit entries.",
|
|
)
|
|
out = origins_from_output(
|
|
{"origins": [copied, fresh]}, page, source="book", codes=CODES
|
|
)
|
|
assert [o["summary"] for o in out] == [fresh["summary"]]
|
|
# the reference is guarded as well
|
|
leaky = item(
|
|
kind="cpt_panel",
|
|
reference="adopted new home visit family entries beginning with 99341",
|
|
)
|
|
assert origins_from_output({"origins": [leaky]}, page, source="book") == []
|
|
|
|
|
|
def test_fr_reference_must_appear():
|
|
ok = item()
|
|
missing = item(reference="section 1861(s)(1) of the Act")
|
|
out = origins_from_output(
|
|
{"origins": [missing, ok]}, FR_STATUTE, source="fr", codes=CODES
|
|
)
|
|
assert [o["reference"] for o in out] == ["Section 1861(m) of the Act"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"text",
|
|
[FR_RECAP, FR_REQUEST, FR_HYPO],
|
|
ids=["recap", "commenter-request", "hypothetical"],
|
|
)
|
|
def test_fr_reference_must_sit_in_a_plain_sentence(text):
|
|
ref = {
|
|
FR_RECAP: "the homebound exception",
|
|
FR_REQUEST: "section 1834(m) of the Act",
|
|
FR_HYPO: "section 1862(a)(1) of the Act",
|
|
}[text]
|
|
code = {FR_RECAP: "99342", FR_REQUEST: "99341", FR_HYPO: "99350"}[text]
|
|
out = origins_from_output(
|
|
{"origins": [item(kind="program", reference=ref, codes=[code])]},
|
|
text,
|
|
source="fr",
|
|
codes=CODES,
|
|
rule_year=2023,
|
|
)
|
|
assert out == []
|
|
|
|
|
|
def test_fr_recap_of_the_same_year_is_a_plain_statement():
|
|
text = "In the CY 2023 PFS final rule, we established the homebound exception for CPT code 99342."
|
|
out = origins_from_output(
|
|
{
|
|
"origins": [
|
|
item(
|
|
kind="program", reference="the homebound exception", codes=["99342"]
|
|
)
|
|
]
|
|
},
|
|
text,
|
|
source="fr",
|
|
codes=CODES,
|
|
rule_year=2023,
|
|
)
|
|
assert len(out) == 1
|
|
|
|
|
|
def test_codes_outside_the_family_or_not_named_are_dropped():
|
|
out = origins_from_output(
|
|
{"origins": [item(codes=["99341", "99999", "99342", "99341"])]},
|
|
FR_STATUTE,
|
|
source="fr",
|
|
codes=CODES,
|
|
)
|
|
assert out[0]["codes"] == ["99341"] # 99999 foreign, 99342 not in the text
|
|
gone = origins_from_output(
|
|
{"origins": [item(codes=["99999"])]}, FR_STATUTE, source="fr", codes=CODES
|
|
)
|
|
assert gone == []
|
|
|
|
|
|
def test_code_ranges_in_the_text_name_every_code():
|
|
text = "Section 1861(m) of the Act covers CPT codes 99341-99342."
|
|
out = origins_from_output(
|
|
{"origins": [item(codes=["99342"])]}, text, source="fr", codes=CODES
|
|
)
|
|
assert len(out) == 1
|
|
|
|
|
|
def test_year_and_numbers_must_be_in_the_text():
|
|
ok = item(year=2023, summary="Section 1861 and 2023 set it up.")
|
|
text = "Section 1861(m) of the Act, effective 2023, sets it up for CPT code 99341."
|
|
[row] = origins_from_output({"origins": [ok]}, text, source="fr", codes=CODES)
|
|
assert row["year"] == 2023
|
|
# a year the text never states is dropped to null, not guessed
|
|
[row] = origins_from_output(
|
|
{"origins": [item(year=1965)]}, text, source="fr", codes=CODES
|
|
)
|
|
assert row["year"] is None
|
|
# a summary number the text lacks drops the item
|
|
assert (
|
|
origins_from_output(
|
|
{"origins": [item(summary="It began in 1988 under the Act.")]},
|
|
text,
|
|
source="fr",
|
|
codes=CODES,
|
|
)
|
|
== []
|
|
)
|
|
|
|
|
|
def test_origins_from_output_tolerates_missing_codes_filter_and_empty_output():
|
|
out = origins_from_output({"origins": [item()]}, FR_STATUTE, source="fr")
|
|
assert out[0]["codes"] == ["99341"]
|
|
assert origins_from_output({}, FR_STATUTE, source="fr") == []
|
|
assert (
|
|
origins_from_output({"origins": [item(codes=[])]}, FR_STATUTE, source="fr")
|
|
== []
|
|
)
|
|
|
|
|
|
def test_schema_and_version():
|
|
assert ORIGIN_PROMPT_VERSION == "origin-v1"
|
|
kinds = ORIGIN_SCHEMA["properties"]["origins"]["items"]["properties"]["kind"]
|
|
assert tuple(kinds["enum"]) == ORIGIN_KINDS
|
|
assert ORIGIN_KINDS == (
|
|
"statute",
|
|
"regulation",
|
|
"program",
|
|
"demonstration",
|
|
"cpt_panel",
|
|
"cms_gcode",
|
|
"coverage",
|
|
)
|
|
|
|
|
|
def test_extract_origin_asks_with_the_schema_and_filters():
|
|
calls: list = []
|
|
out = extract_origin(
|
|
FR_STATUTE,
|
|
ask_with([item(), item(reference="not in the text")], calls),
|
|
source="fr",
|
|
codes=CODES,
|
|
)
|
|
assert [o["reference"] for o in out] == ["Section 1861(m) of the Act"]
|
|
system, user, schema = calls[0]
|
|
assert schema is ORIGIN_SCHEMA and "99341" in user and FR_STATUTE in user
|
|
assert "never quote" in system.lower() or "paraphras" in system.lower()
|
|
|
|
|
|
def test_extract_origin_failed_reply_is_empty():
|
|
ask = lambda s, u, sc: None # noqa: E731
|
|
assert extract_origin(FR_STATUTE, ask, source="fr") == []
|
|
|
|
|
|
def test_extract_origin_book_system_differs_from_fr():
|
|
calls: list = []
|
|
extract_origin(BOOK_PAGES["12"], ask_with([], calls), source="book", codes=CODES)
|
|
extract_origin(FR_STATUTE, ask_with([], calls), source="fr", codes=CODES)
|
|
assert calls[0][0] != calls[1][0]
|
|
assert "8" in calls[0][0] or "eight" in calls[0][0].lower()
|
|
|
|
|
|
def test_extract_origin_rejects_an_unknown_source():
|
|
with pytest.raises(ValueError):
|
|
extract_origin("x", ask_with([]), source="web")
|
|
|
|
|
|
# ── candidates ───────────────────────────────────────────────────────
|
|
|
|
|
|
def test_book_candidates(tmp_path):
|
|
path = make_epub(tmp_path / "b.epub", {"": "Lead 99341 text.", **BOOK_PAGES})
|
|
got = book_candidates(path, CODES)
|
|
assert [label for label, _ in got] == ["12", "13"] # unlabeled text can't be cited
|
|
assert book_candidates(path, ()) == []
|
|
|
|
|
|
def test_book_candidates_see_code_ranges(tmp_path):
|
|
path = make_epub(tmp_path / "b.epub", {"4": "Codes 99341-99342 were revised."})
|
|
assert [label for label, _ in book_candidates(path, ("99342",))] == ["4"]
|
|
|
|
|
|
TITLE = "Medicare Program; CY 2023 PFS Final Rule"
|
|
|
|
|
|
def _store(tmp_path):
|
|
s = Store(":memory:", storage_dir=tmp_path / "st")
|
|
key = s.create(
|
|
Rule(title=TITLE, document_number="22-1", date_published="2022-11-18")
|
|
)
|
|
other = s.create(
|
|
Rule(
|
|
title="Home Health Rule",
|
|
document_number="22-2",
|
|
date_published="2022-06-01",
|
|
)
|
|
)
|
|
corr = s.create(
|
|
Rule(
|
|
title="CY 2023 PFS Final Rule; Correction",
|
|
document_number="22-3",
|
|
date_published="2023-01-01",
|
|
)
|
|
)
|
|
anchors = [
|
|
(key, 1, 100, 1, "An unrelated opening paragraph."),
|
|
(key, 2, 100, 2, FR_STATUTE),
|
|
(key, 3, 101, 3, "CPT code 99341 appears here with nothing about origins."),
|
|
(
|
|
key,
|
|
4,
|
|
101,
|
|
4,
|
|
"The home or residence services family includes a new benefit.",
|
|
),
|
|
(
|
|
other,
|
|
7,
|
|
55,
|
|
1,
|
|
"Section 1895 of the Act created the demonstration for 99350.",
|
|
),
|
|
(corr, 9, 1, 1, FR_STATUTE),
|
|
]
|
|
s._con().executemany(
|
|
"INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)",
|
|
anchors,
|
|
)
|
|
for k, doc in ((key, "22-1"), (other, "22-2"), (corr, "22-3")):
|
|
s._con().execute(
|
|
"INSERT INTO fr_anchor_docs (item_key, document_number, html_url, start_page,"
|
|
" end_page, fr_volume) VALUES (?,?,'u',1,2,87)",
|
|
(k, doc),
|
|
)
|
|
s._con().commit()
|
|
return s, key, other
|
|
|
|
|
|
def test_fr_origin_candidates_filter(tmp_path):
|
|
s, key, other = _store(tmp_path)
|
|
got = fr_origin_candidates(s, CODES, FAM.synonyms)
|
|
assert [(g["item_key"], g["p_id"]) for g in got] == [(other, 7), (key, 2), (key, 4)]
|
|
first = got[1]
|
|
assert first["page"] == 100 and first["rule_year"] == 2023
|
|
assert first["rule_kind"] == "final" and first["text"] == FR_STATUTE
|
|
assert fr_origin_candidates(s, (), ()) == []
|
|
|
|
|
|
def test_resolve_book(tmp_path):
|
|
s = Store(":memory:", storage_dir=tmp_path / "st")
|
|
from bib.item import Source # noqa: PLC0415
|
|
|
|
key = s.create(Source(title="Synthetic Changes 2023"))
|
|
assert resolve_book(s, key) is None # no attachment
|
|
epub = make_epub(tmp_path / "book.epub", BOOK_PAGES)
|
|
s.attach_file(key, epub)
|
|
label, path = resolve_book(s, key)
|
|
assert label == "Synthetic Changes 2023" and path.name.endswith(".epub")
|
|
txt = tmp_path / "n.txt"
|
|
txt.write_text("x")
|
|
k2 = s.create(Source(title="No epub"))
|
|
s.attach_file(k2, txt)
|
|
assert resolve_book(s, k2) is None
|
|
|
|
|
|
# ── guidance ─────────────────────────────────────────────────────────
|
|
|
|
|
|
def _con():
|
|
c = duckdb.connect(":memory:")
|
|
ensure_tables(c)
|
|
return c
|
|
|
|
|
|
def test_guidance_origins_map_kinds():
|
|
c = _con()
|
|
g = [
|
|
GuidanceRow(
|
|
"HOME-VISITS", "99341", "cfr", "42 CFR 410.78(a)(3)", "", "K1", 5, 100
|
|
),
|
|
GuidanceRow(
|
|
"HOME-VISITS", "99341", "cfr", "42 CFR 410.78(a)(3)", "", "K2", 9, 200
|
|
),
|
|
GuidanceRow(
|
|
"HOME-VISITS",
|
|
"99341",
|
|
"iom",
|
|
"100-04 ch.12 §30.6.4",
|
|
"IOM1",
|
|
"K1",
|
|
6,
|
|
100,
|
|
3,
|
|
),
|
|
GuidanceRow("HOME-VISITS", "99350", "mln", "MLN 907166", "", "K3", 0, 0),
|
|
]
|
|
write_guidance(c, "HOME-VISITS", g)
|
|
rows = guidance_origins(c, "HOME-VISITS")
|
|
by = {(r.code, r.cite_kind): r for r in rows}
|
|
assert len(rows) == 3 # the CFR cite repeated by two paragraphs collapses
|
|
cfr = by[("99341", "cfr")]
|
|
assert (cfr.kind, cfr.reference, cfr.extractor) == (
|
|
"regulation",
|
|
"42 CFR 410.78(a)(3)",
|
|
"guidance",
|
|
)
|
|
assert (cfr.item_key, cfr.p_id, cfr.page_label, cfr.year) == ("K1", 5, "100", None)
|
|
assert cfr.locator == "42 CFR 410.78(a)(3)" and len(cfr.summary) <= 240
|
|
assert by[("99341", "iom")].kind == "coverage"
|
|
mln = by[("99350", "mln")]
|
|
assert mln.kind == "coverage" and mln.page_label == "" and mln.p_id == 0
|
|
|
|
|
|
# ── family run ───────────────────────────────────────────────────────
|
|
|
|
|
|
def test_origin_rows_carry_page_label_for_book(tmp_path):
|
|
s, key, other = _store(tmp_path)
|
|
epub = make_epub(tmp_path / "b.epub", BOOK_PAGES)
|
|
book_item = ("BOOK1", "Synthetic Changes 2023", epub)
|
|
paraphrase = item(
|
|
kind="cpt_panel",
|
|
reference="CPT Editorial Panel action",
|
|
codes=["99341", "99350"],
|
|
summary="Editors introduced a new group of home visit entries.",
|
|
year=2023,
|
|
)
|
|
ask = ask_with([paraphrase])
|
|
c = _con()
|
|
rows, counts = family_origin(
|
|
c, s, FAM, ask=ask, log=RunLog(tmp_path / "l.jsonl"), books=[book_item]
|
|
)
|
|
book = [r for r in rows if r.cite_kind == "cpt_book"]
|
|
assert {(r.code, r.page_label, r.p_id, r.item_key) for r in book} == {
|
|
("99341", "12", 0, "BOOK1"),
|
|
("99350", "12", 0, "BOOK1"),
|
|
}
|
|
r = book[0]
|
|
assert r.locator == "Synthetic Changes 2023, p. 12" and r.year == 2023
|
|
assert r.extractor == f"model:qwen-test@{ORIGIN_PROMPT_VERSION}"
|
|
assert all(len(r.summary) <= 240 for r in rows)
|
|
assert counts["failed"] == 0
|
|
|
|
|
|
def test_family_origin_fr_rows_cite_the_paragraph_and_dedupe(tmp_path):
|
|
s, key, other = _store(tmp_path)
|
|
ask = ask_with(
|
|
[
|
|
item(codes=["99341"]),
|
|
item(reference="not in the paragraph", codes=["99341"]),
|
|
]
|
|
)
|
|
c = _con()
|
|
write_guidance(
|
|
c,
|
|
"HOME-VISITS",
|
|
[GuidanceRow("HOME-VISITS", "99341", "cfr", "42 CFR 410.78", "", key, 2, 100)],
|
|
)
|
|
rows, counts = family_origin(
|
|
c, s, FAM, ask=ask, log=RunLog(tmp_path / "l.jsonl"), books=[]
|
|
)
|
|
fr = [r for r in rows if r.cite_kind == "fr_para"]
|
|
assert [(r.code, r.item_key, r.p_id, r.page_label) for r in fr] == [
|
|
("99341", key, 2, "100")
|
|
]
|
|
assert fr[0].locator == "FR p. 100, ¶2" and fr[0].kind == "statute"
|
|
assert [r.cite_kind for r in rows if r.extractor == "guidance"] == ["cfr"]
|
|
assert counts == {"done": 2, "skipped": 0, "failed": 0, "failed_pending": 0}
|
|
|
|
|
|
def test_family_origin_resumes_and_replays_without_a_model(tmp_path):
|
|
s, key, other = _store(tmp_path)
|
|
epub = make_epub(tmp_path / "b.epub", BOOK_PAGES)
|
|
books = [("BOOK1", "Synthetic Changes 2023", epub)]
|
|
log = RunLog(tmp_path / "l.jsonl")
|
|
calls: list = []
|
|
ask = ask_with([item()], calls)
|
|
first, c1 = family_origin(_con(), s, FAM, ask=ask, log=log, books=books)
|
|
n = len(calls)
|
|
assert c1["done"] == n and n > 0
|
|
# a second run with the same ledger asks nothing
|
|
again, c2 = family_origin(_con(), s, FAM, ask=ask, log=log, books=books)
|
|
assert len(calls) == n and c2["skipped"] == n and again == first
|
|
# no model: the finished ledger is replayed, model rows are kept
|
|
replay, c3 = family_origin(
|
|
_con(),
|
|
s,
|
|
FAM,
|
|
ask=None,
|
|
log=RunLog(tmp_path / "l.jsonl"),
|
|
books=books,
|
|
replay_model="qwen-test",
|
|
)
|
|
assert replay == first and c3 == {
|
|
"done": 0,
|
|
"skipped": 0,
|
|
"failed": 0,
|
|
"failed_pending": 0,
|
|
}
|
|
# no model and nothing to replay: only the deterministic guidance rows
|
|
none, _ = family_origin(_con(), s, FAM, ask=None, log=None, books=books)
|
|
assert none == []
|
|
|
|
|
|
def test_family_origin_ledger_holds_no_copied_book_text(tmp_path):
|
|
s, key, other = _store(tmp_path)
|
|
epub = make_epub(tmp_path / "b.epub", BOOK_PAGES)
|
|
copied = item(
|
|
kind="cpt_panel",
|
|
reference="CPT Editorial Panel",
|
|
codes=["99341"],
|
|
summary="adopted new home visit family entries beginning with 99341 and 99350",
|
|
)
|
|
log = RunLog(tmp_path / "l.jsonl")
|
|
rows, _ = family_origin(
|
|
_con(),
|
|
s,
|
|
FAM,
|
|
ask=ask_with([copied]),
|
|
log=log,
|
|
books=[("BOOK1", "Synthetic Changes 2023", epub)],
|
|
)
|
|
assert not [r for r in rows if r.cite_kind == "cpt_book"]
|
|
lines = (tmp_path / "l.jsonl").read_text().splitlines()
|
|
book = [ln for ln in lines if '"unit": "book:BOOK1:12' in ln]
|
|
assert book and all("adopted new home visit" not in ln for ln in book)
|
|
|
|
|
|
def test_family_origin_limit_failed_and_retry(tmp_path):
|
|
s, key, other = _store(tmp_path)
|
|
log = RunLog(tmp_path / "l.jsonl")
|
|
ask = ask_with([item()])
|
|
_, c = family_origin(_con(), s, FAM, ask=ask, log=log, books=[], limit=1)
|
|
assert c["done"] == 1
|
|
failing = lambda sy, u, sc: None # noqa: E731
|
|
failing.model = "qwen-test"
|
|
_, c = family_origin(_con(), s, FAM, ask=failing, log=log, books=[])
|
|
assert c["failed"] == 1 and c["skipped"] == 1
|
|
_, c = family_origin(_con(), s, FAM, ask=failing, log=log, books=[])
|
|
assert c["failed_pending"] == 1
|
|
_, c = family_origin(_con(), s, FAM, ask=ask, log=log, books=[], retry_failed=True)
|
|
assert c["done"] == 1
|
|
|
|
|
|
def test_origin_inventory(tmp_path):
|
|
s, key, other = _store(tmp_path)
|
|
epub = make_epub(tmp_path / "b.epub", BOOK_PAGES)
|
|
paras, pages = origin_inventory(s, FAM, [("BOOK1", "Synthetic Changes 2023", epub)])
|
|
assert len(paras) == 2 and pages == {"BOOK1": ["12", "13"]}
|
|
|
|
|
|
# ── writers ──────────────────────────────────────────────────────────
|
|
|
|
|
|
def _row(**kw):
|
|
base = dict(
|
|
family="HOME-VISITS",
|
|
code="99341",
|
|
kind="statute",
|
|
reference="Section 1861(m) of the Act",
|
|
year=None,
|
|
summary="A paraphrase.",
|
|
cite_kind="fr_para",
|
|
item_key="K",
|
|
p_id=2,
|
|
page_label="100",
|
|
locator="FR p. 100, ¶2",
|
|
extractor="m",
|
|
)
|
|
return OriginRow(**{**base, **kw})
|
|
|
|
|
|
def test_write_read_origin_round_trip_and_replace():
|
|
c = duckdb.connect(":memory:")
|
|
rows = [_row(), _row(code="99350", year=2023), _row(year=1965, reference="A")]
|
|
assert write_origin(c, "HOME-VISITS", rows) == 3
|
|
write_origin(c, "OTHER", [_row(family="OTHER")])
|
|
got = read_origin(c, "HOME-VISITS")
|
|
assert len(got) == 3 and got[0].year is None and got[1].year == 1965
|
|
assert write_origin(c, "HOME-VISITS", [_row()]) == 1
|
|
assert len(read_origin(c, "HOME-VISITS")) == 1 and len(read_origin(c, "OTHER")) == 1
|
|
assert write_origin(c, "HOME-VISITS", []) == 0
|
|
|
|
|
|
def _fr(it, text, **kw):
|
|
return origins_from_output({"origins": [it]}, text, source="fr", codes=CODES, **kw)
|
|
|
|
|
|
def test_kind_follows_what_the_reference_is():
|
|
# a section of the Act is a statute whatever the model called it
|
|
[row] = _fr(item(kind="regulation"), FR_STATUTE)
|
|
assert row["kind"] == "statute"
|
|
text = "Section 425.20 of 42 CFR defines primary care services for CPT code 99341."
|
|
[row] = _fr(item(kind="statute", reference="of 42 CFR"), text)
|
|
assert row["kind"] == "regulation"
|
|
# a statute/regulation claim no reference supports is dropped
|
|
assert (
|
|
_fr(
|
|
item(kind="regulation", reference="the CY 2026 MPFS"),
|
|
"The CY 2026 MPFS covers CPT code 99341.",
|
|
)
|
|
== []
|
|
)
|
|
# a G-code creation must name a HCPCS code
|
|
g = "CMS created HCPCS code G2211 to pay for visits with CPT code 99341."
|
|
assert _fr(item(kind="cms_gcode", reference="HCPCS code G2211"), g)
|
|
assert (
|
|
_fr(
|
|
item(kind="cms_gcode", reference="a new add-on"),
|
|
"A new add-on exists for CPT code 99341.",
|
|
)
|
|
== []
|
|
)
|
|
|
|
|
|
def test_pointers_into_the_same_rule_and_proposals_are_no_origin():
|
|
ptr = "Section II.F. of this proposed rule covers the program for CPT code 99341."
|
|
assert (
|
|
_fr(item(kind="program", reference="section II.F. of this proposed rule"), ptr)
|
|
== []
|
|
)
|
|
text = "The program for CPT code 99341 is described in the telehealth section."
|
|
assert (
|
|
_fr(
|
|
item(
|
|
kind="program",
|
|
reference="The program",
|
|
summary="Proposes to add the codes.",
|
|
),
|
|
text,
|
|
)
|
|
== []
|
|
)
|
|
assert _fr(
|
|
item(kind="program", reference="The program", summary="It adds the codes."),
|
|
text,
|
|
)
|
|
|
|
|
|
def test_codes_must_sit_near_the_sentence_that_writes_the_reference():
|
|
text = (
|
|
"Section 1861(m) of the Act establishes the home health benefit. "
|
|
"That is all it does. Unrelated sentence here. "
|
|
"Later we discuss CPT code 99341 in a different context."
|
|
)
|
|
# 99341 is three sentences from the reference sentence: no origin for it
|
|
assert _fr(item(codes=["99341"]), text) == []
|
|
near = (
|
|
"Section 1861(m) of the Act establishes the home health benefit. "
|
|
"It covers CPT code 99341."
|
|
)
|
|
assert _fr(item(codes=["99341"]), near)
|
|
|
|
|
|
def test_book_rows_are_kept_per_page_and_fr_rows_deduped(tmp_path):
|
|
s, key, other = _store(tmp_path)
|
|
pages = {
|
|
"5": "Rationale: 99341 was restructured last year.",
|
|
"6": "Rationale: 99341 and 99350 were revised for 2023.",
|
|
}
|
|
epub = make_epub(tmp_path / "b.epub", pages)
|
|
it = item(
|
|
kind="cpt_panel",
|
|
reference="CPT Editorial Panel action",
|
|
codes=["99341"],
|
|
summary="Editors revised the visit codes.",
|
|
)
|
|
rows, _ = family_origin(
|
|
_con(),
|
|
s,
|
|
FAM,
|
|
ask=ask_with([it]),
|
|
log=RunLog(tmp_path / "l.jsonl"),
|
|
books=[("B", "Synthetic Changes 2023", epub)],
|
|
)
|
|
assert [(r.page_label) for r in rows if r.cite_kind == "cpt_book"] == ["5", "6"]
|
|
|
|
|
|
def test_a_summary_naming_codes_applies_only_to_those_codes():
|
|
text = "Section 1861(m) of the Act covers CPT codes 99341, 99342 and 99350."
|
|
out = _fr(
|
|
item(
|
|
codes=["99341", "99342", "99350"],
|
|
summary="The Act covers codes 99341-99342.",
|
|
),
|
|
text,
|
|
)
|
|
assert out[0]["codes"] == ["99341", "99342"]
|
|
# a summary naming no code keeps every named code
|
|
assert _fr(item(codes=["99341", "99350"]), text)[0]["codes"] == ["99341", "99350"]
|
|
|
|
|
|
def test_a_book_page_without_prose_states_no_origin():
|
|
table = "99341 Home visit for a new patient, 20 minutes. 99350 Home visit."
|
|
it = item(
|
|
kind="cpt_panel",
|
|
reference="CPT Editorial Panel action",
|
|
codes=["99341"],
|
|
summary="Editors created the code.",
|
|
)
|
|
assert (
|
|
origins_from_output({"origins": [it]}, table, source="book", codes=CODES) == []
|
|
)
|
|
prose = table + " Rationale: 99341 was restructured."
|
|
assert origins_from_output({"origins": [it]}, prose, source="book", codes=CODES)
|