homelab/stack
1
0
Fork 0
Code Issues 214 Pull Requests 6 Actions Packages Projects 1 Releases 1 Wiki Activity

fix(pfs): rounds — deterministic percentiles, proposed/final rules, final_p_id, all PFS tables, ledger replay

Browse Source
This commit is contained in:
kert
2026-11-12 00:55:09 -05:00
parent 5d304c6490
commit 63b9b99fcf
6 changed files with 939 additions and 429 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/cli/pfs.py
View File

@@ -131,6 +131,14 @@ def _asker() -> Any:
return json_extractor(cfg, HostPool.from_config(cfg))
def _model_name() -> str:
"""The instruct model's name, without touching the GPU pool: what a
ledger replay (``--no-llm``) keys the recorded outputs by."""
from llm import config as llm_config
return llm_config.load().instruct_model
def _engine() -> Any:
# llm.index/llm.evidence expose no public "give me a bare SQLAlchemy
# engine over the pgvector DB" helper — cli/bib.py and cli/llm.py
@@ -991,10 +999,12 @@ def cpt_ingest_cmd(
def rounds(
family: str = typer.Option(..., "--family", help="Curated family key."),
tables_only: bool = typer.Option(
False, "--tables-only", help="FR tables only: no paragraphs, no model."
False,
"--tables-only",
help="No model calls (as --no-llm); the finished ledger is replayed.",
),
no_llm: bool = typer.Option(
False, "--no-llm", help="Tables and regex only; make no model calls."
False, "--no-llm", help="Make no model calls; the finished ledger is replayed."
),
limit: int = typer.Option(0, "--limit", help="Cap new model calls (0 = all)."),
dry_run: bool = typer.Option(
@@ -1019,7 +1029,8 @@ def rounds(
f"{len(table)} table rows; no model calls made"
)
return
ask = None if (tables_only or no_llm) else _asker()
replay = tables_only or no_llm
ask = None if replay else _asker()
log = RunLog(ledger_path("rounds", fam.key))
con = _read()
try:
@@ -1032,6 +1043,7 @@ def rounds(
limit=limit,
tables_only=tables_only,
retry_failed=retry_failed,
replay_model=_model_name() if replay else None,
)
finally:
con.close()

8
src/pfs/codetables.py
View File

@@ -309,6 +309,8 @@ class RoundRow:
rationale_quote: str
pe_note: str
extractor: str
#: the paragraph that supplies the final value (R23); 0 = none/unknown
final_p_id: int = 0
@dataclass(frozen=True)
@@ -375,7 +377,7 @@ CREATE TABLE IF NOT EXISTS pfs.valuation_round (
survey_total_min DOUBLE, comparators VARCHAR, surveying_societies VARCHAR,
requested_work_rvu DOUBLE, proposed_work_rvu DOUBLE, final_work_rvu DOUBLE,
decision VARCHAR, rationale_p_id INTEGER, rationale_quote VARCHAR,
pe_note VARCHAR, extractor VARCHAR);
pe_note VARCHAR, extractor VARCHAR, final_p_id INTEGER);
"""
@@ -411,6 +413,10 @@ def ensure_tables(con: Any) -> None:
)
# #705 item 2: the located IOM section page.
con.execute("ALTER TABLE pfs.code_guidance ADD COLUMN IF NOT EXISTS page INTEGER")
# R23: the paragraph that supplies a round's final value.
con.execute(
"ALTER TABLE pfs.valuation_round ADD COLUMN IF NOT EXISTS final_p_id INTEGER"
)
def _insert(con: Any, table: str, rows: Sequence[Any]) -> int:

689
src/pfs/rounds.py
View File

@@ -18,10 +18,12 @@ from dataclasses import dataclass, fields, replace
from pathlib import Path
from typing import Any
from bib.frlink import rule_kind as frlink_rule_kind
from conf import ROOT
from pfs.codetables import RoundRow
from pfs.descriptors import rule_year_of
from pfs.families import code_pattern, codes_in
from pfs.lineage import _RE_RANGE, _expand_ranges
from pfs.runlog import RunLog, input_hash, ledger_path, run_units
_PAGE = re.compile(r"^\[\[Page (\d+)\]\]\s*$")
@@ -70,14 +72,18 @@ def _rows_resume(lines: list[str], k: int) -> bool:
return len(cells) >= 2 and bool(_CODE.match(cells[0]))
def _header_columns(lines: list[str]) -> tuple[str, ...]:
def _header_columns(lines: list[str], data_line: str | None = None) -> tuple[str, ...]:
"""Join stacked header lines column-wise by character span: the line
with the most segments fixes the columns; every other segment joins
the column it overlaps most (else the nearest by centre)."""
with the most segments fixes the columns (or, with *data_line*, the
first body row's cells do); every other segment joins the column it
overlaps most (else the nearest by centre)."""
segs = [
[(m.start(), m.end(), m.group()) for m in _SEG.finditer(ln)] for ln in lines
]
anchor = max(segs, key=len)
if data_line is not None:
anchor = [(m.start(), m.end(), m.group()) for m in _SEG.finditer(data_line)]
else:
anchor = max(segs, key=len)
cols: list[list[tuple[int, str]]] = [[] for _ in anchor]
for li, line_segs in enumerate(segs):
for s, e, txt in line_segs:
@@ -128,6 +134,7 @@ def find_tables(text: str) -> list[FrTable]:
j += 1
rows: list[list[str]] = []
row_pages: list[int] = []
first_line = ""
while j < n and not (_RULE.match(lines[j]) and not _rows_resume(lines, j + 1)):
pm = _PAGE.match(lines[j])
if pm:
@@ -142,15 +149,21 @@ def find_tables(text: str) -> list[FrTable]:
else:
rows.append(_cells(lines[j]))
row_pages.append(page)
first_line = first_line or lines[j]
j += 1
if j >= n:
i += 1
continue
cols = _header_columns(header)
if rows and len(rows[0]) != len(cols):
# stacked header cells split finer than the data (CY2008 NPRM
# Table 10): let the first data row fix the columns
cols = _header_columns(header, first_line)
out.append(
FrTable(
t.group(1),
tpage,
_header_columns(header),
cols,
tuple(map(tuple, rows)),
tuple(row_pages),
)
@@ -166,17 +179,20 @@ def _num(s: str) -> float | None:
return None
def _role(h: str) -> str:
def _role(h: str, rule_year: int = 0) -> str:
if re.search(r"RUC", h) and re.search(r"rec", h, re.I):
return "ruc"
if re.search(r"proposed", h, re.I):
return "proposed"
if re.search(r"request", h, re.I):
return "requested"
if re.search(r"20\d\d work RVU|final|CMS", h, re.I):
return "final"
if re.search(r"basis|decision", h, re.I):
if re.search(r"basis|decision|agree", h, re.I):
return "basis"
m = re.search(r"(20\d\d) work RVU", h)
if m: # only the rule's own year is its result; an earlier year is the old value
return "final" if rule_year in (0, int(m.group(1))) else ""
if re.search(r"final|CMS", h, re.I):
return "final"
return ""
@@ -205,7 +221,7 @@ def table_rounds(
) -> list[RoundRow]:
roles: dict[str, int] = {}
for i, h in enumerate(table.header):
if i and (r := _role(h)):
if i and (r := _role(h, rule_year)):
roles.setdefault(r, i)
if not {"ruc", "proposed", "final"} & roles.keys():
return []
@@ -219,7 +235,7 @@ def table_rounds(
if row[0] not in codes or len(row) < len(table.header):
continue
ruc, final = _num(cell(row, "ruc")), _num(cell(row, "final"))
basis = cell(row, "basis")
basis = re.sub(r"\.{3,}\s*$", "", cell(row, "basis"))
interim = " interim" if re.search(r"interim", table.title, re.I) else ""
decision = decision_of(ruc, final, basis + interim)
note = ""
@@ -285,16 +301,29 @@ def fr_text_path(store: Any, item_key: str) -> Path | None:
_VAL_A = re.compile(r"work RVU of (\d+\.\d+) for (?:CPT|HCPCS) code (\w{5})")
_VAL_B = re.compile(r"(\d+\.\d+) work RVUs (?:to|for) (?:CPT|HCPCS) code (\w{5})")
_PROPOSE = re.compile(r"\b(?:we|are|is)\s+(?:(?:are|is)\s+)?propos(?:e|ed|ing)\b", re.I)
_FINALIZE = re.compile(
r"(?<!not )(?<!not be )finaliz\w+|\bwe have assigned\b|\bwe are assigning\b",
re.I,
)
_NEGATED = re.compile(
r"\bnot (?:be )?(?:finaliz\w+|adopt\w*|assign\w*)|\bdo not (?:finaliz|adopt)\w*",
re.I,
)
_RUC_VERB = re.compile(r"RUC[- ]recommended|RUC recommends", re.I)
_VERBS = (
(
"proposed_work_rvu",
re.compile(r"\bwe (?:are )?propos(?:e|ed|ing)\b", re.I),
),
(
"final_work_rvu",
re.compile(r"finaliz\w+|\bwe have assigned\b|\bwe are assigning\b", re.I),
),
("ruc_work_rvu", re.compile(r"RUC[- ]recommend\w*", re.I)),
("proposed", _PROPOSE),
("final", _FINALIZE),
("neg", _NEGATED),
("ruc", _RUC_VERB),
)
_FIELD = {
"proposed": "proposed_work_rvu",
"final": "final_work_rvu",
"ruc": "ruc_work_rvu",
}
_CONDITIONAL = re.compile(
r"\bwould (?:need|be|increase|decrease)\b|\bif we were to\b", re.I
)
_FAMILY_AS_PROPOSED = re.compile(
r"finaliz\w+ the work RVU values? for the .{0,80}code family as proposed", re.I
@@ -302,37 +331,49 @@ _FAMILY_AS_PROPOSED = re.compile(
_RUC_AS_PROPOSED = re.compile(
r"(?:propos\w+|finaliz\w+) the RUC-recommended work RVUs? for all", re.I
)
_SENT = re.compile(r"(?<=[.;])\s+(?=[A-Z])")
_CODE_TOKEN = r"(?:\d{5}|[A-Z]\d{4})"
def _sentences(text: str) -> list[str]:
return _SENT.split(text)
def para_values(
text: str, codes: set[str], *, rule_kind: str
) -> list[tuple[str, str, float]]:
"""``(code, field, value)`` for each "work RVU of X for CPT code Y"
clause naming one of *codes*. The field comes from the nearest
preceding verb in the paragraph: "we propose[d]" is always a
proposed value, even in a final rule; "finalize"/"assigned" is
final; "RUC-recommended" is the RUC's. *rule_kind* is accepted for
the callers' symmetry: the verb alone decides, so a final rule's
"We proposed" can never become a final value."""
del rule_kind
verbs = sorted(
(m.start(), field) for field, rx in _VERBS for m in rx.finditer(text)
)
found = sorted(
(m.start(), m.group(2), float(m.group(1)))
for rx in (_VAL_A, _VAL_B)
for m in rx.finditer(text)
)
clause naming one of *codes*. Within a sentence, the nearest verb
before the clause decides: "we propose[d]" is a proposed value (even
in a final rule), "RUC-recommended" the RUC's ("We proposed the
RUC-recommended work RVU of X" is both), "finalize"/"assigned" a
final value only in a ``final`` rule (a proposed rule's "we
finalized" is a recap of an earlier rule); a negated verb ("we are
not finalizing") gives nothing."""
out: list[tuple[str, str, float]] = []
for pos, code, value in found:
if code not in codes:
continue
prior = [f for p, f in verbs if p < pos]
if not prior:
continue
triple = (code, prior[-1], value)
if triple not in out:
out.append(triple)
for sent in _sentences(text):
verbs = sorted(
(m.start(), cls) for cls, rx in _VERBS for m in rx.finditer(sent)
)
found = sorted(
(m.start(), m.group(2), float(m.group(1)))
for rx in (_VAL_A, _VAL_B)
for m in rx.finditer(sent)
)
for pos, code, value in found:
prior = [c for p, c in verbs if p < pos]
if code not in codes or not prior:
continue
classes = [prior[-1]]
if prior[-1] == "ruc" and len(prior) > 1:
classes.insert(0, prior[-2]) # "We proposed the RUC-recommended …"
for cls in classes:
if cls == "neg" or (cls == "final" and rule_kind != "final"):
continue
if cls == "ruc" or cls in _FIELD:
triple = (code, _FIELD[cls], value)
if triple not in out:
out.append(triple)
return out
@@ -344,17 +385,6 @@ def ruc_as_proposed(text: str) -> bool:
return bool(_RUC_AS_PROPOSED.search(text))
def rule_kind_of(title: str, date_published: str) -> str:
"""``final`` or ``proposed``: the title says so, else a PFS rule
published before October is the proposed rule."""
if re.search(r"\b(?:final|interim)\b", title or "", re.I):
return "final"
if re.search(r"\bproposed\b", title or "", re.I):
return "proposed"
month = int((date_published or "0000-12")[5:7] or 12)
return "proposed" if month <= 9 else "final"
_MENTION = re.compile(r"RUC|work RVU|survey|percentile|crosswalk|refine|direct PE")
@@ -370,33 +400,66 @@ def _anchor_rows(
f"WHERE {where} ORDER BY i.date_published, a.item_key, a.p_id",
likes,
).fetchall()
return [
dict(
zip(
("item_key", "p_id", "page", "text", "title", "date_published"),
r,
strict=True,
)
keys = ("item_key", "p_id", "page", "text", "title", "date_published")
return [dict(zip(keys, r, strict=True)) for r in rows]
def rule_items(store: Any, anchors: Sequence[dict]) -> list[dict]:
"""Every PFS rule item to read: those *anchors* came from plus every
PFS-titled item with anchors. One entry per FR document (bib items
sharing a document number are one rule), corrections left out
(``bib.frlink.rule_kind``). Each entry carries ``kind``."""
con = store._con() # noqa: SLF001
found = {r["item_key"]: (r["title"], r["date_published"]) for r in anchors}
for k, title, dp in con.execute(
"SELECT DISTINCT i.key, i.title, i.date_published FROM items i "
"JOIN fr_anchors a ON a.item_key = i.key "
"WHERE i.title LIKE '%Physician Fee Schedule%' OR i.title LIKE '%PFS%'"
).fetchall():
found.setdefault(k, (title, dp))
out: list[dict] = []
docs: set[str] = set()
for key, (title, dp) in sorted(found.items(), key=lambda kv: (kv[1][1], kv[0])):
doc = con.execute(
"SELECT document_number FROM fr_anchor_docs WHERE item_key = ?", (key,)
).fetchone()
if doc and doc[0] in docs:
continue
kind = frlink_rule_kind(store, key, title=title)
if kind == "correction":
continue
if doc:
docs.add(doc[0])
out.append(
{"item_key": key, "title": title, "date_published": dp, "kind": kind}
)
for r in rows
]
return out
def _named(text: str, pattern: re.Pattern[str]) -> tuple[str, ...]:
return codes_in(_expand_ranges(text), pattern)
def candidate_paragraphs(
store: Any, codes: Sequence[str], phrases: Sequence[str]
) -> list[dict]:
"""``fr_anchors`` paragraphs naming a family code or phrase and
talking valuation, each with ``context`` = the two preceding
paragraphs of the same item."""
"""``fr_anchors`` paragraphs of the (de-duplicated, non-correction)
rules naming a family code or phrase and talking valuation, each with
``rule_kind`` and ``context`` = the two preceding paragraphs of the
same item."""
if not codes and not phrases:
return []
pattern = code_pattern(codes) if codes else None
low = [x.lower() for x in phrases]
con = store._con() # noqa: SLF001
anchors = _anchor_rows(store, codes, phrases)
kinds = {it["item_key"]: it["kind"] for it in rule_items(store, anchors)}
out = []
for r in _anchor_rows(store, codes, phrases):
for r in anchors:
text = r["text"] or ""
named = bool(pattern and codes_in(text, pattern)) or any(
if r["item_key"] not in kinds:
continue
named = bool(pattern and _named(text, pattern)) or any(
x in text.lower() for x in low
)
if not named or not _MENTION.search(text):
@@ -406,10 +469,97 @@ def candidate_paragraphs(
"ORDER BY p_id DESC LIMIT 2",
(r["item_key"], r["p_id"]),
).fetchall()
out.append({**r, "context": [p[0] or "" for p in reversed(prior)]})
out.append(
{
**r,
"rule_kind": kinds[r["item_key"]],
"context": [p[0] or "" for p in reversed(prior)],
}
)
return out
# ── survey percentiles: deterministic (R21) ──────────────────────────
_TERM = re.compile(
r"(?P<p>25th|75th) percentile"
r"|\bmedian\b(?!\s+(?:[\w-]+\s+){0,2}(?:time|intra-service|minutes))",
re.I,
)
_LIST = rf"{_CODE_TOKEN}(?:\s*(?:,|and|or)\s*(?:and\s+|or\s+)?{_CODE_TOKEN})*"
_FOR_CODES = re.compile(rf"\bfor (?:the )?(?:CPT |HCPCS )?codes? ({_LIST})", re.I)
_COLLECTIVE_FOR = re.compile(
r"\bfor all (?:(?:\d+|one|two|three|four|five|six|seven|eight|nine|ten|the)\s+)?"
r"(?:[\w-]+\s+){0,2}?(?:CPT |HCPCS )?codes?\b",
re.I,
)
_EXCEPT = re.compile(rf"\bexcept (?:for )?(?:CPT |HCPCS )?codes? ({_LIST})", re.I)
_AT_TERM = re.compile(
r"\b(?:at|of) the (?:survey )?(?=(?:25th|75th) percentile|median)", re.I
)
_ANY_CODE = re.compile(rf"\b{_CODE_TOKEN}\b")
def _label(m: re.Match[str]) -> str:
return "median" if m.group("p") is None else m.group("p").lower()
def _list_codes(text: str) -> list[str]:
return _ANY_CODE.findall(text)
def _percentiles(text: str) -> dict[str, str]:
"""``{code: "25th"|"median"|"75th"}`` for the codes the paragraph
states a survey value for, and only those: (P1) a value followed by
``for [CPT] code(s) <list>`` in one clause, (P2) ``<code> … at the
<value>`` before the next code, (P3) ``<value> for all … codes,
except <list> … <value2>``. A code the paragraph names as deleted
never gets one; a code with two different statements gets none."""
text = _expand_ranges(text)
found: dict[str, set[str]] = {}
deleted: set[str] = set()
for sent in _sentences(text):
if re.search(r"\bdelet", sent, re.I):
deleted.update(_list_codes(sent))
for sent in _sentences(text):
terms = [(m.start(), m.end(), _label(m)) for m in _TERM.finditer(sent)]
if not terms:
continue
coll = _COLLECTIVE_FOR.search(sent)
if coll and terms[0][0] < coll.start():
term1 = [t for t in terms if t[0] < coll.start()][-1][2]
exc = _EXCEPT.search(sent, coll.end())
excepted = _list_codes(exc.group(1)) if exc else []
later = [t for t in terms if exc and t[0] >= exc.end()]
for code in set(_list_codes(text)) - set(excepted):
found.setdefault(code, set()).add(term1)
for code in excepted if later else []:
found.setdefault(code, set()).add(later[0][2])
continue
claimed: set[str] = set()
for i, (start, end, label) in enumerate(terms):
stop = terms[i + 1][0] if i + 1 < len(terms) else len(sent)
m = _FOR_CODES.search(sent[end:stop])
if m:
for code in _list_codes(m.group(1)):
found.setdefault(code, set()).add(label)
claimed.add(code)
mentions = list(_ANY_CODE.finditer(sent))
for i, m in enumerate(mentions):
if m.group() in claimed:
continue
stop = mentions[i + 1].start() if i + 1 < len(mentions) else len(sent)
window = sent[m.end() : stop]
at = _AT_TERM.search(window)
if at:
t = _TERM.match(window, at.end())
if t:
found.setdefault(m.group(), set()).add(_label(t))
return {
c: next(iter(v)) for c, v in found.items() if len(v) == 1 and c not in deleted
}
# ── paragraph source: local model ────────────────────────────────────
ROUND_PROMPT_VERSION = "rounds-v1"
@@ -466,25 +616,27 @@ ROUND_SYSTEM = (
"codes. Extract ONLY what the PARAGRAPH states; the CONTEXT paragraphs "
'are for understanding only. Use null (or "none", or an empty string) '
"for anything the paragraph does not state; never infer, compute or "
"recall a value, and copy numbers exactly as written. Fields: "
"ruc_work_rvu = the work RVU the RUC recommended; ruc_percentile = which "
"survey value the RUC recommended for THAT code (25th percentile, median, "
"75th percentile); survey_intra_min / survey_total_min = survey "
"intra-service / total minutes; comparators = reference or crosswalk "
"codes cited; proposed_work_rvu = a work RVU CMS proposed (in a FINAL "
'rule, a sentence that says "we proposed" gives a PROPOSED value, never '
"a final one); final_work_rvu = a work RVU CMS is finalizing or has "
"assigned; decision = CMS's decision on the RUC recommendation (accept, "
"refine, reject, interim); pe_note = one short note on direct practice-"
"expense inputs; surveying_societies = specialty societies that "
"surveyed; rationale_quote = one sentence copied verbatim from the "
"PARAGRAPH giving CMS's reason, else an empty string."
"recall a value, and copy numbers exactly as written. A work RVU is "
"proposed or final only when the paragraph says CMS proposed/is "
"proposing or finalized/is finalizing it for that code: an illustration "
'("would need to increase from X to Y") or a comparison is neither. '
"Fields: ruc_work_rvu = the work RVU the RUC recommended; ruc_percentile "
"= which survey value the RUC recommended for THAT code (25th "
"percentile, median, 75th percentile); survey_intra_min / "
"survey_total_min = survey intra-service / total minutes; comparators = "
"reference or crosswalk codes cited; proposed_work_rvu = a work RVU CMS "
'proposed (in a FINAL rule, a sentence that says "we proposed" gives a '
"PROPOSED value, never a final one); final_work_rvu = a work RVU CMS is "
"finalizing or has assigned; decision = CMS's decision on the RUC "
"recommendation (accept, refine, reject, interim); pe_note = one short "
"note on direct practice-expense inputs; surveying_societies = specialty "
"societies that surveyed; rationale_quote = one sentence copied "
"verbatim from the PARAGRAPH giving CMS's reason, else an empty string. "
"A code range such as 99347-99350 names every code in it; the paragraph "
"is followed by the expanded ranges."
)
_PCT = re.compile(r"(25th|75th) percentile|\bmedian\b", re.I)
_NUMBER = re.compile(r"\d+(?:,\d{3})*(?:\.\d+)?")
_COLLECTIVE = re.compile(r"\ball\b|\bfamily\b|\beach\b|\bevery\b", re.I)
_SENT = re.compile(r"(?<=[.;])\s+(?=[A-Z])")
_NUM_FIELDS = (
"ruc_work_rvu",
"survey_intra_min",
@@ -495,47 +647,21 @@ _NUM_FIELDS = (
)
def _pct_terms(sentence: str) -> list[tuple[int, str]]:
return [
(m.start(), "median" if m.group(1) is None else m.group(1).lower())
for m in _PCT.finditer(sentence)
]
def _allowed_percentiles(code: str, text: str) -> set[str]:
"""The survey values the paragraph states for *code*: sentences naming
the code speak for it (what follows its last mention if anything
does); otherwise a collective sentence ("all codes in the family,
except 99350, ...") speaks for the codes it does not except."""
named = re.compile(rf"\b{re.escape(code)}\b")
own: set[str] = set()
collective: set[str] = set()
for s in _SENT.split(text):
terms = _pct_terms(s)
if not terms:
continue
mentions = [m.start() for m in named.finditer(s)]
if mentions:
after = {label for pos, label in terms if pos > mentions[-1]}
own |= after or {label for _, label in terms}
elif _COLLECTIVE.search(s):
cut = re.search(r"\bexcept\b", s, re.I)
collective |= {
label for pos, label in terms if cut is None or pos < cut.start()
}
return own or collective
def _numbers(text: str) -> set[float]:
return {float(n.replace(",", "")) for n in _NUMBER.findall(text)}
def guard(row: RoundRow, text: str) -> RoundRow:
"""Clear whatever *text* does not literally say: numbers (compared
numerically, so ``1.0`` is ``1.00``), the percentile (per code), the
numerically, so ``1.0`` is ``1.00``), a percentile other than the
one the paragraph states for this code (``_percentiles``), the
rationale quote (must be a substring), societies and comparators."""
present = {float(n.replace(",", "")) for n in _NUMBER.findall(text)}
present = _numbers(text)
low = text.lower()
changes: dict[str, Any] = {
f: None for f in _NUM_FIELDS if getattr(row, f) not in present
}
if row.ruc_percentile not in _allowed_percentiles(row.code, text):
if row.ruc_percentile != _percentiles(text).get(row.code):
changes["ruc_percentile"] = ""
if row.rationale_quote and row.rationale_quote not in text:
changes["rationale_quote"] = ""
@@ -547,10 +673,7 @@ def guard(row: RoundRow, text: str) -> RoundRow:
def _para_meta(para: dict) -> tuple[int, str]:
return (
rule_year_of(para["title"], para["date_published"]),
rule_kind_of(para["title"], para["date_published"]),
)
return rule_year_of(para["title"], para["date_published"]), para["rule_kind"]
def _blank(family: str, code: str, para: dict, extractor: str) -> RoundRow:
@@ -587,80 +710,66 @@ def regex_rounds(para: dict, codes: set[str], *, family: str) -> list[RoundRow]:
rows: dict[str, RoundRow] = {}
for code, field, value in para_values(para["text"], codes, rule_kind=kind):
row = rows.get(code) or _blank(family, code, para, "regex")
rows[code] = replace(row, **{field: value})
extra = {"final_p_id": para["p_id"]} if field == "final_work_rvu" else {}
rows[code] = replace(row, **{field: value}, **extra)
for code, pct in _percentiles(para["text"]).items():
if code in codes:
row = rows.get(code) or _blank(family, code, para, "regex")
rows[code] = replace(row, ruc_percentile=pct)
return list(rows.values())
def _range_note(text: str) -> str:
notes = [
f"{m.group(0)} = {_expand_ranges(m.group(0))}" for m in _RE_RANGE.finditer(text)
]
return "\nCode ranges: " + "; ".join(notes) if notes else ""
def model_user(para: dict, codes: Sequence[str]) -> str:
ctx = "\n\n".join(para.get("context") or []) or "(none)"
return (
f"Codes: {', '.join(codes)}\nRule: {para['title']}\n\n"
f"CONTEXT:\n{ctx}\n\nPARAGRAPH:\n{para['text']}"
f"CONTEXT:\n{ctx}\n\nPARAGRAPH:\n{para['text']}{_range_note(para['text'])}"
)
def rows_from_output(
out: dict, para: dict, codes: set[str], *, family: str, model: str
) -> list[RoundRow]:
"""Build guarded rows from one model reply (a ledger output)."""
_, kind = _para_meta(para)
extractor = f"model:{model}@{ROUND_PROMPT_VERSION}"
regex = para_values(para["text"], codes, rule_kind=kind)
quote = out.get("rationale_quote") or ""
societies = "; ".join(out.get("surveying_societies") or [])
rows: list[RoundRow] = []
for d in out.get("codes") or []:
code = d.get("code")
if code not in codes:
continue
pct, dec = d.get("ruc_percentile"), d.get("decision")
final = None if kind == "proposed" else d.get("final_work_rvu")
if final is not None and (code, "proposed_work_rvu", final) in regex:
# "we proposed X" sits in this paragraph: X is the proposal,
# whatever the model called it
final = None
row = replace(
_blank(family, code, para, extractor),
ruc_work_rvu=d.get("ruc_work_rvu"),
ruc_percentile=pct if pct in ("25th", "median", "75th") else "",
survey_intra_min=d.get("survey_intra_min"),
survey_total_min=d.get("survey_total_min"),
comparators="; ".join(c for c in d.get("comparators") or [] if c != code),
surveying_societies=societies,
proposed_work_rvu=d.get("proposed_work_rvu"),
final_work_rvu=final,
decision="" if dec == "none" else dec or "",
rationale_p_id=para["p_id"] if quote else 0,
rationale_quote=quote,
pe_note=_pe_note(d.get("pe_note")),
)
row = guard(row, para["text"])
stated = _allowed_percentiles(code, para["text"])
if not row.ruc_percentile and len(stated) == 1:
# the paragraph states one survey value for this code (e.g.
# "25th percentile for all codes except 99350") and the model
# said nothing: take what the text says
row = replace(row, ruc_percentile=next(iter(stated)))
if row.ruc_work_rvu is not None and row.final_work_rvu is not None:
# both numbers are in the paragraph: they decide, not the
# model's label ("accept" next to 2.27 -> 2.63 is a refinement)
row = replace(row, decision=decision_of(*_pair(row), ""))
if _has_content(row):
rows.append(row)
return rows
def _supported(code: str, value: float, field: str, text: str, kind: str) -> bool:
"""Is *value* stated as a proposal/finalization for *code*: a sentence
naming the code and the value carries a proposing (or finalizing, in a
final rule) verb and no "would need to"/"if we were to" hedge."""
verb = _PROPOSE if field == "proposed_work_rvu" else _FINALIZE
if field == "final_work_rvu" and kind != "final":
return False
named = re.compile(rf"\b{re.escape(code)}\b")
for sent in _sentences(_expand_ranges(text)):
if (
named.search(sent)
and verb.search(sent)
and not _CONDITIONAL.search(sent)
and not (field == "final_work_rvu" and _NEGATED.search(sent))
and value in _numbers(sent)
):
return True
return False
def _has_content(row: RoundRow) -> bool:
"""A reply that names a code but states nothing about it is no round."""
return any(
(
*(getattr(row, f) is not None for f in _NUM_FIELDS),
row.ruc_percentile,
row.decision,
row.comparators,
row.pe_note,
)
)
def _keep(
code: str,
field: str,
value: float | None,
regex: list[tuple[str, str, float]],
text: str,
kind: str,
) -> float | None:
"""A model proposed/final value survives only when the regex read it
too, or the code's own sentence proposes/finalizes it (R20); a "final"
that the paragraph gives as the proposal is dropped (R22)."""
if value is None or (code, field, value) in regex:
return value
if field == "final_work_rvu" and (code, "proposed_work_rvu", value) in regex:
return None
return value if _supported(code, value, field, text, kind) else None
def _pe_note(note: str | None) -> str:
@@ -669,14 +778,72 @@ def _pe_note(note: str | None) -> str:
return "" if note.lower().startswith("none") else note
def _pair(row: RoundRow) -> tuple[float | None, float | None]:
return row.ruc_work_rvu, row.final_work_rvu
def has_value(row: RoundRow) -> bool:
return any(getattr(row, f) is not None for f in _NUM_FIELDS)
def has_content(row: RoundRow) -> bool:
"""R20: a round states a value or a decision; comparators and PE
notes alone never make one."""
return has_value(row) or bool(row.decision)
def rows_from_output(
out: dict, para: dict, codes: set[str], *, family: str, model: str
) -> list[RoundRow]:
"""Build guarded rows from one model reply (a ledger output)."""
_, kind = _para_meta(para)
text = para["text"]
extractor = f"model:{model}@{ROUND_PROMPT_VERSION}"
regex = para_values(text, codes, rule_kind=kind)
pcts = _percentiles(text)
quote = out.get("rationale_quote") or ""
societies = "; ".join(out.get("surveying_societies") or [])
rows: list[RoundRow] = []
for d in out.get("codes") or []:
code = d.get("code")
if code not in codes:
continue
dec = d.get("decision")
vals = {
f: _keep(code, f, d.get(f), regex, text, kind)
for f in ("proposed_work_rvu", "final_work_rvu")
}
row = replace(
_blank(family, code, para, extractor),
ruc_work_rvu=d.get("ruc_work_rvu"),
ruc_percentile=pcts.get(code, ""),
survey_intra_min=d.get("survey_intra_min"),
survey_total_min=d.get("survey_total_min"),
comparators="; ".join(c for c in d.get("comparators") or [] if c != code),
surveying_societies=societies,
proposed_work_rvu=vals["proposed_work_rvu"],
final_work_rvu=vals["final_work_rvu"],
decision="" if dec == "none" else dec or "",
rationale_p_id=para["p_id"] if quote else 0,
rationale_quote=quote,
pe_note=_pe_note(d.get("pe_note")),
)
row = guard(row, text)
if row.final_work_rvu is not None:
row = replace(row, final_p_id=para["p_id"])
if not has_value(row):
row = replace(row, decision="") # a bare label is not a round
elif row.ruc_work_rvu is not None and row.final_work_rvu is not None:
# both numbers are in the paragraph: they decide, not the
# model's label ("accept" next to 2.27 -> 2.63 is a refinement)
row = replace(
row,
decision=decision_of(row.ruc_work_rvu, row.final_work_rvu, ""),
)
rows.append(row)
return rows
def model_rounds(
para: dict, codes: set[str], ask: Any, *, family: str
) -> list[RoundRow]:
named = sorted(codes_in(para["text"], code_pattern(sorted(codes))))
named = sorted(_named(para["text"], code_pattern(sorted(codes))))
out = ask(ROUND_SYSTEM, model_user(para, named), ROUND_SCHEMA)
if out is None:
return []
@@ -744,17 +911,20 @@ def merge_rounds(*sources: Sequence[RoundRow]) -> list[RoundRow]:
def apply_paragraph_rules(
rows: Sequence[RoundRow], paras: Sequence[dict]
) -> list[RoundRow]:
"""Paragraph-level statements that apply beyond the codes they name:
"finalizing the work RVU values for the family as proposed" gives
each round of that rule without a final its proposed value as the
final; "proposed the RUC-recommended work RVU for all" gives the
codes the paragraph names their proposal as the RUC value. An empty
decision is then derived from the numbers."""
"""Paragraph-level statements that apply beyond the codes they name,
in FINAL rules only: "finalizing the work RVU values for the family
as proposed" gives each round of that rule without a final its
proposed value as the final (and cites that paragraph for it);
"proposed the RUC-recommended work RVU for all" gives the codes the
paragraph names their proposal as the RUC value. An empty decision is
then derived from the numbers."""
out = list(rows)
for para in paras:
if para.get("rule_kind") != "final":
continue
text = para["text"]
if family_as_proposed(text):
quote = _SENT.split(text)[0]
quote = _sentences(text)[0]
for i, r in enumerate(out):
if (
r.item_key == para["item_key"]
@@ -764,16 +934,21 @@ def apply_paragraph_rules(
out[i] = replace(
r,
final_work_rvu=r.proposed_work_rvu,
rationale_p_id=r.rationale_p_id or para["p_id"],
rationale_quote=r.rationale_quote or quote[:300],
final_p_id=para["p_id"],
rationale_p_id=para["p_id"],
rationale_quote=quote[:300],
)
if ruc_as_proposed(text):
named = re.compile(
r"\b(?:%s)\b" % "|".join(re.escape(r.code) for r in out) or "(?!)"
)
present = set(named.findall(_expand_ranges(text)))
for i, r in enumerate(out):
if (
r.item_key == para["item_key"]
and r.ruc_work_rvu is None
and r.proposed_work_rvu is not None
and re.search(rf"\b{re.escape(r.code)}\b", text)
and r.code in present
):
out[i] = replace(r, ruc_work_rvu=r.proposed_work_rvu)
return [
@@ -787,44 +962,48 @@ def apply_paragraph_rules(
# ── the whole family ─────────────────────────────────────────────────
def table_rows_for(store: Any, rows: Sequence[dict], family: Any) -> list[RoundRow]:
"""Rounds from every FR table in the rules *rows* (``_anchor_rows``)
belong to; one pass per rule item."""
def table_rows_for(store: Any, items: Sequence[dict], family: Any) -> list[RoundRow]:
"""Rounds from every FR table in the rule *items* (``rule_items``)."""
codes = set(family.codes)
out: list[RoundRow] = []
seen: set[str] = set()
for r in rows:
if r["item_key"] in seen:
continue
seen.add(r["item_key"])
path = fr_text_path(store, r["item_key"])
for it in items:
path = fr_text_path(store, it["item_key"])
if path is None:
continue
year = rule_year_of(r["title"], r["date_published"])
kind = rule_kind_of(r["title"], r["date_published"])
year = rule_year_of(it["title"], it["date_published"])
for t in find_tables(path.read_text(encoding="utf-8", errors="replace")):
out += table_rounds(
t,
family=family.key,
item_key=r["item_key"],
item_key=it["item_key"],
rule_year=year,
rule_kind=kind,
rule_kind=it["kind"],
codes=codes,
)
return out
def _naming(store: Any, family: Any) -> tuple[list[dict], list[dict]]:
pattern = code_pattern(family.codes)
paras = candidate_paragraphs(store, family.codes, family.synonyms)
return paras, [p for p in paras if _named(p["text"], pattern)]
def rounds_inventory(store: Any, family: Any) -> tuple[list[dict], list[RoundRow]]:
"""What a run would work on, without any model call: the candidate
paragraphs naming a family code, and the FR-table rounds."""
anchors = _anchor_rows(store, family.codes, ())
pattern = code_pattern(family.codes)
paras = [
p
for p in candidate_paragraphs(store, family.codes, family.synonyms)
if codes_in(p["text"], pattern)
]
return paras, table_rows_for(store, anchors, family)
_, naming = _naming(store, family)
return naming, table_rows_for(store, rule_items(store, anchors), family)
def _unit(p: dict) -> str:
return f"{p['item_key']}:{p['p_id']}"
def _model_input(p: dict, family: Any) -> str:
named = sorted(_named(p["text"], code_pattern(family.codes)))
return model_user(p, named)
def family_rounds(
@@ -837,61 +1016,59 @@ def family_rounds(
limit: int = 0,
tables_only: bool = False,
retry_failed: bool = False,
replay_model: str | None = None,
) -> tuple[list[RoundRow], dict[str, int]]:
"""Table + regex + model rounds for *family*, merged, with the
paragraph rules applied. Returns ``(rows, run counts)``; the model
runs through the ledger *log* (resumable, ``limit`` caps new calls).
*con* is accepted for the callers' symmetry (nothing is read from it)."""
paragraph rules applied and contentless rounds dropped. Returns
``(rows, run counts)``. The model runs through the ledger *log*
(resumable, ``limit`` caps new calls). Without *ask* (``--no-llm``,
``tables_only``) no call is made, but the completed ledger for
*replay_model* is replayed so earlier model rows are never lost.
*con* is accepted for the callers' symmetry (nothing is read)."""
del con
codes = set(family.codes)
counts = {"done": 0, "skipped": 0, "failed": 0, "failed_pending": 0}
anchors = _anchor_rows(store, family.codes, ())
table = table_rows_for(store, anchors, family)
if tables_only:
return apply_paragraph_rules(merge_rounds(table), []), counts
paras = candidate_paragraphs(store, family.codes, family.synonyms)
naming = [p for p in paras if codes_in(p["text"], code_pattern(family.codes))]
table = table_rows_for(store, rule_items(store, anchors), family)
paras, naming = _naming(store, family)
regex: list[RoundRow] = []
for p in naming:
regex += regex_rounds(p, codes, family=family.key)
model: list[RoundRow] = []
if ask is not None:
model_name = ask.model if ask is not None else replay_model
if model_name is not None:
log = log or RunLog(ledger_path("rounds", family.key))
def hash_of(p: dict) -> str:
return input_hash(p["text"], "\n".join(p["context"]), ROUND_SYSTEM)
units = [
(
f"{p['item_key']}:{p['p_id']}",
hash_of(p),
lambda p=p: ask(
ROUND_SYSTEM,
model_user(
p, sorted(codes_in(p["text"], code_pattern(family.codes)))
hashes = {
_unit(p): input_hash(_model_input(p, family), ROUND_SYSTEM) for p in naming
}
if ask is not None and not tables_only:
units = [
(
_unit(p),
hashes[_unit(p)],
lambda p=p: ask(
ROUND_SYSTEM,
_model_input(p, family),
ROUND_SCHEMA,
),
ROUND_SCHEMA,
),
)
for p in naming
]
counts = run_units(
units,
log,
model=model_name,
version=ROUND_PROMPT_VERSION,
limit=limit,
retry_failed=retry_failed,
)
for p in naming
]
counts = run_units(
units,
log,
model=ask.model,
version=ROUND_PROMPT_VERSION,
limit=limit,
retry_failed=retry_failed,
)
for p in naming:
entry = log.get(
f"{p['item_key']}:{p['p_id']}",
hash_of(p),
ask.model,
ROUND_PROMPT_VERSION,
_unit(p), hashes[_unit(p)], model_name, ROUND_PROMPT_VERSION
)
if entry is not None and entry.output is not None:
model += rows_from_output(
entry.output, p, codes, family=family.key, model=ask.model
entry.output, p, codes, family=family.key, model=model_name
)
return apply_paragraph_rules(merge_rounds(table, regex, model), paras), counts
merged = apply_paragraph_rules(merge_rounds(table, regex, model), paras)
return [r for r in merged if has_content(r)], counts

19
tests/cli/test_pfs_cli.py
View File

@@ -1420,9 +1420,28 @@ class TestRounds:
def test_no_model_flags_build_no_asker(self, con, monkeypatch, flag):
seen = self._patch(monkeypatch, [])
monkeypatch.setattr(pfs_cli, "_asker", lambda: 1 / 0)
monkeypatch.setattr(pfs_cli, "_model_name", lambda: "qwen-x")
res = runner.invoke(app, ["pfs", "rounds", "--family", "HOME-VISITS", flag])
assert res.exit_code == 0, res.output
assert seen["ask"] is None and seen["tables_only"] == (flag == "--tables-only")
# the finished ledger is replayed, so a --write cannot delete model rows
assert seen["replay_model"] == "qwen-x"
def test_model_run_does_not_replay(self, con, monkeypatch):
seen = self._patch(monkeypatch, [])
monkeypatch.setattr(pfs_cli, "_model_name", lambda: 1 / 0)
res = runner.invoke(app, ["pfs", "rounds", "--family", "HOME-VISITS"])
assert res.exit_code == 0, res.output
assert seen["replay_model"] is None
def test_model_name_reads_config_without_a_pool(self, monkeypatch):
from llm import config as llm_config
class Cfg:
instruct_model = "qwen-test"
monkeypatch.setattr(llm_config, "load", lambda: Cfg())
assert pfs_cli._model_name() == "qwen-test"
def test_rounds_unknown_family(self, con):
res = runner.invoke(app, ["pfs", "rounds", "--family", "NOPE"])

16
tests/pfs/fixtures/fr_table_cy2008_nprm_t10.txt Normal file
View File

@@ -0,0 +1,16 @@
[[Page 38147]]
Table 10.--Remaining Codes From Five-Year Review of Work Relative Value Units
--------------------------------------------------------------------------------------------------------------------------------------------------------
2008
2007 work Requested CMS proposal (agree/ Proposed
CPT \1\/ HCPCS code Mod Descriptor RVU work RVU RUC REC disagree) work RVU
\2\
--------------------------------------------------------------------------------------------------------------------------------------------------------
99337.............................. ................ Domicil/r-home visit 3.03 4.05 3.03 Agree................ 3.03
est pat.
99343.............................. ................ Home visit, new 2.27 2.65 2.27 Agree................ 2.27
patient.
99344.............................. ................ Home visit, new 3.03 3.60 3.03 Agree................ 3.03
patient.
--------------------------------------------------------------------------------------------------------------------------------------------------------

618
tests/pfs/test_rounds.py
View File

@@ -16,6 +16,7 @@ from pfs.rounds import (
ROUND_PROMPT_VERSION,
ROUND_SCHEMA,
FrTable,
_percentiles,
apply_paragraph_rules,
candidate_paragraphs,
decision_of,
@@ -24,6 +25,7 @@ from pfs.rounds import (
find_tables,
fr_text_path,
guard,
has_content,
merge_rounds,
model_rounds,
para_values,
@@ -31,7 +33,7 @@ from pfs.rounds import (
rounds_inventory,
rows_from_output,
ruc_as_proposed,
rule_kind_of,
rule_items,
split_pages,
table_rounds,
)
@@ -288,7 +290,48 @@ def test_rounds_roundtrip():
assert isinstance(got[0], RoundRow)
# ── paragraph source (task 5) ────────────────────────────────────────
def test_cy2008_nprm_table_10_parses_with_the_data_row_columns():
"""4PYQ2WWF (07-3274) Table 10: stacked header cells split finer than
the data, so the first data row fixes the columns."""
[t] = find_tables(
(
Path(__file__).parent / "fixtures" / "fr_table_cy2008_nprm_t10.txt"
).read_text()
)
assert t.header[3:] == (
"2007 work RVU",
"Requested work RVU",
"RUC REC",
"CMS proposal (agree/ disagree)",
"2008 Proposed work RVU \\2\\",
)
rows = {
r.code: r
for r in table_rounds(
t,
family="F",
item_key="4PYQ2WWF",
rule_year=2008,
rule_kind="proposed",
codes={"99343", "99344"},
)
}
r = rows["99343"]
assert (r.proposed_work_rvu, r.requested_work_rvu, r.ruc_work_rvu) == (
2.27,
2.65,
2.27,
)
assert (r.final_work_rvu, r.decision, r.rationale_quote, r.page) == (
None,
"accept",
"Agree",
38147,
)
assert rows["99344"].proposed_work_rvu == 3.03
# ── paragraph source ─────────────────────────────────────────────────
DESC = "( Home or residence visit for the evaluation and management of a patient)"
P1676 = (
@@ -303,15 +346,30 @@ P1676 = (
f"a work RVU of 1.65 for CPT code 99342 {DESC}, "
f"and a work RVU of 3.60 for CPT code 99350 {DESC}."
)
P877 = (
"In February 2021, the CPT Editorial Panel deleted the nine CPT codes in "
"the Domiciliary, Rest Home family (CPT codes 99324-99328, and "
"99334-99337), and one CPT code in the Home Services family (CPT code "
"99343), to merge these services with the eight remaining home visit "
"services. The eight remaining home services CPT codes (99341, 99342, "
"99344, 99345, and 99347-99350) were revised. These eight CPT codes were "
"reviewed at the October 2021 RUC meeting. The RUC recommended the survey "
"25th percentile value for all CPT codes in the Home or Residence "
"Services code family, except for CPT code 99350, for which the RUC "
"recommended the median value."
)
P1685 = (
"After consideration of the public comments, we are finalizing the work RVU "
"values for the Home or Residence Services code family as proposed. We are "
"also finalizing the direct PE inputs for these codes as proposed."
)
TITLE23 = "Medicare Program; CY 2023 PFS Final Rule"
TITLE23P = "Medicare Program; CY 2023 PFS Proposed Rule"
def para(p_id, text, *, item="PETM225D", title=TITLE23, date="2022-11-18"):
def para(
p_id, text, *, item="PETM225D", title=TITLE23, date="2022-11-18", kind="final"
):
return {
"item_key": item,
"p_id": p_id,
@@ -319,10 +377,17 @@ def para(p_id, text, *, item="PETM225D", title=TITLE23, date="2022-11-18"):
"text": text,
"title": title,
"date_published": date,
"rule_kind": kind,
"context": ["earlier one", "earlier two"],
}
def pp(text, kind="proposed"):
return para(
1, text, title="CY 2027 PFS Proposed Rule", date="2026-07-16", kind=kind
)
CODES23 = {"99341", "99342", "99350"}
@@ -343,43 +408,109 @@ def test_para_values_verbs_patterns_and_filters():
"assigning 1.10 work RVUs to HCPCS code G0001, "
"a work RVU of 9.99 for CPT code 11111 (not ours)."
)
got = para_values(t, {"99343", "G0001"}, rule_kind="final")
assert got == [
assert para_values(t, {"99343", "G0001"}, rule_kind="final") == [
("99343", "ruc_work_rvu", 2.27),
("99343", "final_work_rvu", 2.53),
("G0001", "final_work_rvu", 1.10),
]
# no verb before the clause: nothing to classify
t2 = "We are assigning 1.10 work RVUs to HCPCS code G0001."
assert para_values(t2, {"G0001"}, rule_kind="final") == [
("G0001", "final_work_rvu", 1.10)
]
assert (
para_values("a work RVU of 1.00 for CPT code 99341", {"99341"}, rule_kind="x")
== []
)
# the same clause twice is one triple
twice = "We propose a work RVU of 1.00 for CPT code 99341. " * 2
assert len(para_values(twice, {"99341"}, rule_kind="proposed")) == 1
def test_paragraph_predicates_and_rule_kind():
def test_we_proposed_the_ruc_recommended_work_rvu_is_proposed_and_ruc():
t = "We proposed the RUC-recommended work RVU of 1.00 for CPT code 99341."
got = para_values(t, {"99341"}, rule_kind="final")
assert sorted(got) == [
("99341", "proposed_work_rvu", 1.0),
("99341", "ruc_work_rvu", 1.0),
]
t = "We are finalizing the RUC-recommended work RVU of 1.10 for CPT code 99341."
assert sorted(para_values(t, {"99341"}, rule_kind="final")) == [
("99341", "final_work_rvu", 1.1),
("99341", "ruc_work_rvu", 1.1),
]
# in a proposed rule the finalize half is a recap
assert para_values(t, {"99341"}, rule_kind="proposed") == [
("99341", "ruc_work_rvu", 1.1)
]
# a model "final" equal to that proposal is dropped
out = {
"codes": [_d("99341", final_work_rvu=1.0, proposed_work_rvu=1.0)],
"surveying_societies": [],
"rationale_quote": "",
}
t = "We proposed the RUC-recommended work RVU of 1.00 for CPT code 99341."
[r] = rows_from_output(out, para(3, t), {"99341"}, family="F", model="m")
assert (r.proposed_work_rvu, r.final_work_rvu) == (1.0, None)
def test_proposed_rules_make_no_finals_and_negation_gives_none():
t = "In the CY 2023 PFS final rule, we finalized a work RVU of 1.00 for CPT code 99341."
assert para_values(t, {"99341"}, rule_kind="proposed") == []
assert [x.final_work_rvu for x in regex_rounds(pp(t), {"99341"}, family="F")] == []
for neg in (
"We are not finalizing the proposed work RVU of 1.00 for CPT code 99341.",
"We are not adopting a work RVU of 1.00 for CPT code 99341.",
"We do not finalize a work RVU of 1.00 for CPT code 99341.",
):
assert para_values(neg, {"99341"}, rule_kind="final") == [], neg
t = "We disagree with the RUC recommendation and are proposing a work RVU of 1.50 for CPT code 99341."
assert para_values(t, {"99341"}, rule_kind="proposed") == [
("99341", "proposed_work_rvu", 1.5)
]
def test_paragraph_rules_only_in_final_rules():
prop = regex_rounds(
pp("We are proposing a work RVU of 1.10 for CPT code 99341."),
{"99341"},
family="F",
)
hist = pp(
"In the CY 2023 PFS final rule, we finalized the work RVU values for the Home or "
"Residence Services code family as proposed (87 FR 69608).",
kind="proposed",
)
[r] = apply_paragraph_rules(prop, [hist])
assert (r.proposed_work_rvu, r.final_work_rvu, r.decision) == (1.1, None, "")
ruc = pp(
"We proposed the RUC-recommended work RVU for all eight CPT codes: 99341.",
kind="proposed",
)
[r] = apply_paragraph_rules(prop, [ruc])
assert r.ruc_work_rvu is None
def test_paragraph_predicates():
assert family_as_proposed(P1685) and not family_as_proposed(P1676)
assert ruc_as_proposed(P1676) and not ruc_as_proposed(P1685)
assert rule_kind_of("CY 2008 PFS Final Rule", "2007-11-27") == "final"
assert rule_kind_of("... Interim Final Rule", "2020-12-28") == "final"
assert rule_kind_of("CY 2020 PFS Proposed Rule", "2019-08-14") == "proposed"
assert rule_kind_of("CY 2023 Payment Policies", "2022-07-29") == "proposed"
assert rule_kind_of("CY 2023 Payment Policies", "2022-11-18") == "final"
assert rule_kind_of("CY 2023 Payment Policies", "") == "final"
def _regex_1676():
return regex_rounds(para(1676, P1676), CODES23, family="HOME-VISITS")
def test_regex_rounds_carry_anchor_and_year():
def test_regex_rounds_carry_anchor_year_and_percentile():
rows = {r.code: r for r in _regex_1676()}
r = rows["99341"]
assert (r.rule_year, r.rule_kind, r.p_id, r.page) == (2023, "final", 1676, 69608)
assert (r.source, r.extractor, r.proposed_work_rvu) == ("fr-para", "regex", 1.00)
assert r.final_work_rvu is None
assert (r.final_work_rvu, r.ruc_percentile, r.final_p_id) == (None, "25th", 0)
assert rows["99350"].ruc_percentile == "median"
fin = regex_rounds(
para(7, "We are finalizing a work RVU of 2.0 for CPT code 99341."),
{"99341"},
family="F",
)
assert (fin[0].final_work_rvu, fin[0].final_p_id) == (2.0, 7)
def _applied():
@@ -389,17 +520,18 @@ def _applied():
}
def test_family_finalize_as_proposed_propagates():
def test_family_finalize_as_proposed_propagates_and_cites_the_finalizing_paragraph():
r = _applied()
assert (r["99341"].final_work_rvu, r["99350"].final_work_rvu) == (1.00, 3.60)
assert r["99341"].rationale_p_id == 1685
assert (r["99341"].final_p_id, r["99341"].rationale_p_id) == (1685, 1685)
assert r["99341"].rationale_quote.startswith("After consideration")
assert r["99341"].p_id == 1676 # the proposal stays cited to its paragraph
def test_ruc_recommended_proposal_sets_ruc():
r = _applied()["99341"]
assert (r.ruc_work_rvu, r.final_work_rvu, r.decision) == (1.00, 1.00, "accept")
assert r.proposed_work_rvu == 1.00
assert r.proposed_work_rvu == 1.00 and r.ruc_percentile == "25th"
def test_paragraph_rules_leave_other_rules_and_existing_values_alone():
@@ -414,6 +546,75 @@ def test_paragraph_rules_leave_other_rules_and_existing_values_alone():
assert (out[1].final_work_rvu, out[1].ruc_work_rvu) == (7.0, 9.0)
assert out[2].final_work_rvu is None and out[2].decision == ""
assert out[1].decision == "refine"
assert apply_paragraph_rules([], [para(1, P1685)]) == []
# ── percentiles (R21) ────────────────────────────────────────────────
def test_percentiles_real_1676_text():
p = _percentiles(P1676)
assert p["99350"] == "median"
assert p == {c: "25th" for c in ("99341", "99342")} | {"99350": "median"}
def test_percentiles_real_877_deleted_codes_get_none():
p = _percentiles(P877)
for dead in (
"99324",
"99325",
"99326",
"99327",
"99328",
"99334",
"99335",
"99336",
"99337",
"99343",
):
assert dead not in p, dead
assert {
c: p[c] for c in ("99341", "99342", "99344", "99345", "99347", "99348", "99349")
} == {
c: "25th"
for c in ("99341", "99342", "99344", "99345", "99347", "99348", "99349")
}
assert p["99350"] == "median"
def test_percentiles_probe_sentences():
both = "The RUC recommended the 25th percentile for CPT code 99341 and the median for CPT code 99342."
assert _percentiles(both) == {"99341": "25th", "99342": "median"}
# a bare exception names no value for 99350
assert "99350" not in _percentiles(
"The RUC recommended the 25th percentile for all codes except CPT code 99350."
)
# a stray "all" is not collective; 99341 gets nothing
stray = "After reviewing all survey data, the RUC recommended the median work RVU for CPT code 99350."
assert _percentiles(stray) == {"99350": "median"}
# median as a time statistic
assert (
_percentiles(
"The survey median intra-service time for CPT code 99343 was 45 minutes."
)
== {}
)
# P2: "<code> at the <term>"
assert _percentiles(
"The RUC set CPT code 99341 at the 75th percentile, and CPT code 99342 at the median."
) == {"99341": "75th", "99342": "median"}
# contradictory statements give nothing
assert (
_percentiles(
"The 25th percentile for CPT code 99341. The median for CPT code 99341."
)
== {}
)
assert _percentiles("No survey statistic here for CPT code 99341.") == {}
# a list of codes
assert _percentiles(
"The 75th percentile for CPT codes 99341, 99342 and 99344."
) == {"99341": "75th", "99342": "75th", "99344": "75th"}
# ── guard ────────────────────────────────────────────────────────────
@@ -432,11 +633,14 @@ def test_guard_drops_numbers_not_in_text():
def test_guard_keeps_median_only_when_stated():
r50 = guard(_row(code="99350", ruc_percentile="median"), P1676)
assert r50.ruc_percentile == "median"
assert (
guard(_row(code="99350", ruc_percentile="median"), P1676).ruc_percentile
== "median"
)
assert guard(_row(code="99350", ruc_percentile="25th"), P1676).ruc_percentile == ""
r41 = guard(_row(code="99341", ruc_percentile="25th"), P1676)
assert r41.ruc_percentile == "25th"
assert (
guard(_row(code="99341", ruc_percentile="25th"), P1676).ruc_percentile == "25th"
)
for bad in ("median", "75th"):
assert guard(_row(code="99341", ruc_percentile=bad), P1676).ruc_percentile == ""
assert (
@@ -445,22 +649,6 @@ def test_guard_keeps_median_only_when_stated():
)
def test_guard_percentile_sentence_naming_the_code():
text = "The RUC used the median for 99341. Other codes: the 25th percentile."
assert (
guard(_row(code="99341", ruc_percentile="median"), text).ruc_percentile
== "median"
)
# a sentence about one code does not speak for another
text = "For code 99342 the RUC used the median value."
assert guard(_row(code="99341", ruc_percentile="median"), text).ruc_percentile == ""
# collective sentence without an exception speaks for every code
text = "The RUC used the 75th percentile for each code in the family."
assert (
guard(_row(code="99341", ruc_percentile="75th"), text).ruc_percentile == "75th"
)
def test_guard_quote_societies_comparators():
quote = "These eight CPT codes were reviewed at the October 2021 RUC meeting."
r = guard(
@@ -509,33 +697,44 @@ def _d(code, **kw):
return {**base, **kw}
def _reply(*codes, quote="", societies=()):
return {
"codes": list(codes),
"surveying_societies": list(societies),
"rationale_quote": quote,
}
def test_schema_accepts_a_good_reply():
from llm.structured import validate
out = {"codes": [_d("99341")], "surveying_societies": [], "rationale_quote": ""}
assert validate(out, ROUND_SCHEMA) == []
assert validate(_reply(_d("99341")), ROUND_SCHEMA) == []
assert ROUND_PROMPT_VERSION == "rounds-v1"
def test_model_rounds_guards_and_maps_enums():
out = {
"codes": [
_d(
"99341",
ruc_work_rvu=1.75,
ruc_percentile="25th",
proposed_work_rvu=1.0,
decision="accept",
comparators=["99350"],
pe_note="removed SK062",
),
_d("99350", ruc_percentile="median", final_work_rvu=3.6, decision="none"),
_d("99342", ruc_percentile="other"),
_d("99999", ruc_work_rvu=1.0),
],
"surveying_societies": ["AAFP"],
"rationale_quote": "We proposed the RUC-recommended work RVU for all eight CPT codes",
}
out = _reply(
_d(
"99341",
ruc_work_rvu=1.75,
ruc_percentile="75th",
proposed_work_rvu=1.0,
decision="accept",
comparators=["99350"],
pe_note="removed SK062",
),
_d(
"99350",
ruc_percentile="25th",
final_work_rvu=3.6,
proposed_work_rvu=3.6,
decision="none",
),
_d("99342", ruc_percentile="other", proposed_work_rvu=1.65),
_d("99999", ruc_work_rvu=1.0),
quote="We proposed the RUC-recommended work RVU for all eight CPT codes in the Home or Residence Services CPT code family.",
societies=["AAFP"],
)
ask = FakeAsk(out)
rows = {
r.code: r for r in model_rounds(para(1676, P1676), CODES23, ask, family="HV")
@@ -543,50 +742,114 @@ def test_model_rounds_guards_and_maps_enums():
assert set(rows) == CODES23 # 99999 is not a family code
r = rows["99341"]
assert r.ruc_work_rvu is None # 1.75 is not in the text
# the percentile is the paragraph's (25th), not the model's 75th
assert (r.ruc_percentile, r.proposed_work_rvu, r.decision) == (
"25th",
1.0,
"accept",
)
assert r.comparators == "99350" # in the text
assert r.surveying_societies == "" # "AAFP" is not
assert r.comparators == "99350" and r.surveying_societies == ""
assert r.extractor == "model:qwen-test@rounds-v1"
assert (r.p_id, r.rationale_p_id, r.source) == (1676, 1676, "fr-para")
assert rows["99350"].ruc_percentile == "median" and rows["99350"].decision == ""
assert rows["99342"].ruc_percentile == "25th" # "other" -> what the text states
assert rows["99350"].final_work_rvu is None # "we proposed" 3.60 is the proposal
assert rows["99342"].ruc_percentile == "25th"
assert "99341, 99342, 99350" in ask.calls[0][1]
def test_model_final_that_is_a_proposal_is_dropped():
out = {
"codes": [_d("99341", proposed_work_rvu=1.0, final_work_rvu=1.0)],
"surveying_societies": [],
"rationale_quote": "",
}
[r] = model_rounds(para(1676, P1676), CODES23, FakeAsk(out), family="HV")
assert (r.proposed_work_rvu, r.final_work_rvu) == (1.0, None)
def test_proposed_rule_never_has_a_final():
text = (
"The RUC recommended a work RVU of 1.10; in CY2020 we finalized 1.10 for 99341."
def test_model_only_values_need_a_verb_in_the_codes_sentence():
"""R20: the CY2020 99348 "would need to increase from 1.56 to 2.00" case."""
t = (
"To maintain the same proportional value, the work RVU for CPT code 99348 "
"would need to increase from 1.56 to 2.00. We are proposing a work "
"RVU of 1.25 for CPT code 99349 and we propose 4.5 for CPT code 99350. "
"In CY2020 we finalized 3.0 for CPT code 99342. We are not finalizing 2.5 for CPT code 99344."
)
out = {
"codes": [_d("99341", ruc_work_rvu=1.10, final_work_rvu=1.10)],
"surveying_societies": [],
"rationale_quote": "",
}
p = para(5, text, title="CY 2023 PFS Proposed Rule", date="2022-07-29")
[r] = rows_from_output(out, p, {"99341"}, family="HV", model="m")
assert r.rule_kind == "proposed" and r.final_work_rvu is None
out = _reply(
_d("99348", proposed_work_rvu=2.0, final_work_rvu=1.56, decision="accept"),
_d("99349", proposed_work_rvu=1.25),
_d("99350", proposed_work_rvu=4.5),
_d("99342", final_work_rvu=3.0),
_d("99344", final_work_rvu=2.5, comparators=["G2211"], pe_note="x"),
)
codes = {"99348", "99349", "99350", "99342", "99344"}
p = para(
9, t, title="CY 2020 PFS Proposed Rule", date="2019-08-14", kind="proposed"
)
rows = {r.code: r for r in rows_from_output(out, p, codes, family="F", model="m")}
assert (
rows["99348"].proposed_work_rvu is None and rows["99348"].final_work_rvu is None
)
assert rows["99348"].decision == "" # a bare label with no value is no round
assert rows["99349"].proposed_work_rvu == 1.25
assert rows["99350"].proposed_work_rvu == 4.5 # its sentence says "we propose"
assert rows["99342"].final_work_rvu is None # a proposed rule's recap
assert not any(
has_content(r) for k, r in rows.items() if k in ("99348", "99342", "99344")
)
# the same paragraph in a final rule: finals need the finalize verb
pf = {**p, "rule_kind": "final"}
rows = {r.code: r for r in rows_from_output(out, pf, codes, family="F", model="m")}
assert rows["99342"].final_work_rvu == 3.0 and rows["99342"].final_p_id == 9
assert rows["99344"].final_work_rvu is None # negated
assert rows["99348"].final_work_rvu is None
def test_model_rounds_failed_and_empty_replies():
def test_has_content_ignores_comparators_and_pe_notes():
r = _row(
code="99341",
proposed_work_rvu=None,
ruc_work_rvu=None,
final_work_rvu=None,
comparators="G2211",
pe_note="removed SK062",
decision="",
)
assert not has_content(r)
assert has_content(replace(r, decision="accept"))
assert has_content(replace(r, survey_intra_min=30.0))
assert not rounds_mod.has_value(replace(r, decision="accept"))
def test_model_failed_and_empty_replies():
assert model_rounds(para(1, P1676), CODES23, FakeAsk(None), family="HV") == []
assert rows_from_output({}, para(1, P1676), CODES23, family="HV", model="m") == []
out = {"codes": [{"code": "99341"}]}
[r] = rows_from_output(out, para(1, P1676), CODES23, family="HV", model="m")
[r] = rows_from_output(
{"codes": [{"code": "99341"}]}, para(1, P1676), CODES23, family="HV", model="m"
)
assert r.decision == "" and r.comparators == "" and r.rationale_p_id == 0
assert r.ruc_percentile == "25th"
def test_model_final_with_both_numbers_gets_a_numeric_decision():
text = (
"The RUC recommended a work RVU of 2.27 for CPT code 99326; we are "
"finalizing a work RVU of 2.63 for CPT code 99326 after refinement."
)
out = _reply(
_d(
"99326",
ruc_work_rvu=2.27,
final_work_rvu=2.63,
decision="accept",
comparators=["99326", "99350"],
pe_note="none (no PE inputs)",
)
)
[r] = rows_from_output(out, para(9, text), {"99326"}, family="HV", model="m")
assert (r.decision, r.pe_note, r.comparators) == ("refine", "", "")
out["codes"][0]["pe_note"] = "removed SK062"
[r] = rows_from_output(out, para(9, text), {"99326"}, family="HV", model="m")
assert r.pe_note == "removed SK062"
def test_model_prompt_expands_code_ranges():
ask = FakeAsk(_reply())
model_rounds(para(877, P877), {"99349", "99350", "99341"}, ask, family="HV")
user = ask.calls[0][1]
assert "99347-99350 = 99347 99348 99349 99350" in user and "Code ranges:" in user
assert "99341, 99349, 99350" in user # range members are named codes
assert "Code ranges" not in rounds_mod.model_user(para(1, "no ranges"), [])
# ── merge ────────────────────────────────────────────────────────────
@@ -627,11 +890,17 @@ def test_merge_prefers_table_over_model():
page=1,
rationale_p_id=77,
rationale_quote="because",
final_p_id=77,
)
[m] = merge_rounds([table], [], [model])
assert m.final_work_rvu == 2.53 and m.decision == "refine"
assert m.proposed_work_rvu == 2.27 and m.ruc_percentile == "25th"
assert (m.p_id, m.rationale_p_id, m.rationale_quote) == (77, 77, "because")
assert (m.p_id, m.rationale_p_id, m.rationale_quote, m.final_p_id) == (
77,
77,
"because",
77,
)
assert m.page == 66361 and m.source == "fr-table"
assert "conflict: final_work_rvu 2.6 (model:m@rounds-v1)" in m.pe_note
assert "conflict: decision accept (model:m@rounds-v1)" in m.pe_note
@@ -654,27 +923,49 @@ def test_merge_keys_and_same_extractor():
# ── store-backed: candidates, family run ─────────────────────────────
def _store(tmp_path, with_text=True):
def _store(tmp_path, with_text=True, dup=False):
s = Store(":memory:", storage_dir=tmp_path / "st")
key = s.create(
Rule(title=TITLE23, document_number="22-0001", date_published="2022-11-18")
)
con = s._con()
rows = [
anchors = [
(key, 1, 10, 1, "Unrelated opening paragraph."),
(key, 2, 10, 2, "Context two."),
(key, 1676, 69608, 3, P1676),
(key, 1685, 69609, 4, P1685),
(key, 1690, 69609, 5, "Code 99341 is named in passing here."),
]
keys = [key]
if dup: # a second bib item for the same FR document
k2 = s.create(
Rule(title=TITLE23, document_number="22-0001", date_published="2022-11-19")
)
keys.append(k2)
anchors += [(k2, 1676, 69608, 3, P1676)]
# a correction notice quoting the rule
kc = s.create(
Rule(
title="CY 2023 PFS Final Rule; Correction",
document_number="22-9",
date_published="2023-01-01",
)
)
anchors.append((kc, 5, 1, 1, P1676))
con.executemany(
"INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)",
rows,
anchors,
)
for k in keys:
con.execute(
"INSERT INTO fr_anchor_docs (item_key, document_number, html_url, start_page, end_page, fr_volume)"
" VALUES (?,'22-0001','u',1,2,87)",
(k,),
)
con.commit()
if with_text:
txt = tmp_path / "rule.txt"
txt.write_text(FIXTURE.replace("99343", "99341"))
txt.write_text(FIXTURE.replace("99343", "99341").replace("2008", "2023"))
s.attach_file(key, txt)
return s, key
@@ -686,28 +977,34 @@ FAM = SimpleNamespace(
)
def test_rule_items_dedupe_documents_and_drop_corrections(tmp_path):
s, key = _store(tmp_path, dup=True)
items = rule_items(s, [])
assert [i["item_key"] for i in items] == [key] # dup and correction gone
assert items[0]["kind"] == "final"
# PFS-titled items with anchors are included even when no anchor names a code
assert [i["item_key"] for i in rule_items(s, [])] == [key]
def test_candidate_paragraphs_filter_and_context(tmp_path):
s, key = _store(tmp_path)
s, key = _store(tmp_path, dup=True)
assert candidate_paragraphs(s, (), ()) == []
got = candidate_paragraphs(s, FAM.codes, FAM.synonyms)
assert [p["p_id"] for p in got] == [1676, 1685] # 1690 has no valuation word
assert [(p["item_key"], p["p_id"]) for p in got] == [(key, 1676), (key, 1685)]
assert got[0]["context"] == ["Unrelated opening paragraph.", "Context two."]
assert got[0]["title"] == TITLE23 and got[0]["date_published"] == "2022-11-18"
only_phrase = candidate_paragraphs(s, (), FAM.synonyms)
assert [p["p_id"] for p in only_phrase] == [1676, 1685]
only_code = candidate_paragraphs(s, FAM.codes, ())
assert [p["p_id"] for p in only_code] == [1676]
assert got[0]["rule_kind"] == "final" and got[0]["title"] == TITLE23
assert [p["p_id"] for p in candidate_paragraphs(s, (), FAM.synonyms)] == [
1676,
1685,
]
assert [p["p_id"] for p in candidate_paragraphs(s, FAM.codes, ())] == [1676]
def _good_reply():
return {
"codes": [
_d("99341", ruc_percentile="25th", proposed_work_rvu=1.0),
_d("99350", ruc_percentile="median", proposed_work_rvu=3.6),
],
"surveying_societies": [],
"rationale_quote": "",
}
return _reply(
_d("99341", ruc_percentile="25th", proposed_work_rvu=1.0),
_d("99350", ruc_percentile="median", proposed_work_rvu=3.6),
)
def test_family_rounds_end_to_end_and_resume(tmp_path):
@@ -720,14 +1017,12 @@ def test_family_rounds_end_to_end_and_resume(tmp_path):
r = by["99341"]
assert (r.proposed_work_rvu, r.final_work_rvu, r.ruc_work_rvu) == (1.0, 1.0, 1.0)
assert (r.ruc_percentile, r.decision, r.rule_year) == ("25th", "accept", 2023)
assert (r.p_id, r.final_p_id) == (1676, 1685)
assert by["99350"].ruc_percentile == "median"
assert len(ask.calls) == 1
# second run: the ledger answers, the model is not asked
again, counts = family_rounds(None, s, FAM, ask=ask, log=log)
assert counts["skipped"] == 1 and counts["done"] == 0 and len(ask.calls) == 1
assert again == rows
# limit caps new calls: a fresh ledger with limit 0 runs all, so test with
# a failing reply and retry
bad = FakeAsk(None)
log2 = RunLog(tmp_path / "l2.jsonl")
rows2, c2 = family_rounds(None, s, FAM, ask=bad, log=log2)
@@ -738,7 +1033,29 @@ def test_family_rounds_end_to_end_and_resume(tmp_path):
assert c4["failed"] == 1 and len(bad.calls) == 2
def test_family_rounds_default_ledger_and_no_model(tmp_path, monkeypatch):
def test_no_llm_and_tables_only_replay_the_finished_ledger(tmp_path):
s, key = _store(tmp_path, with_text=False)
ask = FakeAsk(_good_reply())
log = RunLog(tmp_path / "ledger.jsonl")
full, _ = family_rounds(None, s, FAM, ask=ask, log=log)
assert any("model:" in r.extractor for r in full)
for kw in ({}, {"tables_only": True}):
replay, counts = family_rounds(
None, s, FAM, ask=None, log=log, replay_model="qwen-test", **kw
)
assert replay == full and counts["done"] == 0
assert len(ask.calls) == 1
# a different model has no recorded outputs: regex rows only, nothing deleted that regex made
other, _ = family_rounds(None, s, FAM, ask=None, log=log, replay_model="other")
assert {r.extractor for r in other} == {"regex"} or all(
"model:" not in r.extractor for r in other
)
# no model name at all
plain, _ = family_rounds(None, s, FAM, ask=None, log=None)
assert all("model:" not in r.extractor for r in plain)
def test_family_rounds_default_ledger(tmp_path, monkeypatch):
s, _ = _store(tmp_path, with_text=False)
monkeypatch.setattr(
rounds_mod, "ledger_path", lambda t, f: tmp_path / f"{t}-{f}.jsonl"
@@ -746,22 +1063,19 @@ def test_family_rounds_default_ledger_and_no_model(tmp_path, monkeypatch):
ask = FakeAsk(_good_reply())
family_rounds(None, s, FAM, ask=ask, log=None)
assert (tmp_path / "rounds-HOME-VISITS.jsonl").exists()
rows, counts = family_rounds(None, s, FAM, ask=None, log=None)
assert counts["done"] == 0 and len(ask.calls) == 1
assert {r.extractor for r in rows} == {"regex"} # no table rows for these ids
def test_family_rounds_tables_only_reads_the_fr_table(tmp_path):
def test_family_rounds_reads_the_fr_table_and_merges_it_first(tmp_path):
s, key = _store(tmp_path)
ask = FakeAsk(_good_reply())
rows, counts = family_rounds(None, s, FAM, ask=ask, log=None, tables_only=True)
assert ask.calls == [] and counts["done"] == 0
rows, counts = family_rounds(None, s, FAM, ask=None, log=None)
[r] = [r for r in rows if r.code == "99341"]
assert (r.source, r.rule_year, r.rule_kind) == ("fr-table", 2023, "final")
assert r.final_work_rvu == 2.53
# an item with no text file contributes no table rows
assert r.source == "fr-table" and (r.rule_year, r.rule_kind) == (2023, "final")
assert r.final_work_rvu == 2.53 and r.p_id == 1676 # table + paragraph anchor
s2, _ = _store(tmp_path / "x", with_text=False)
assert family_rounds(None, s2, FAM, ask=None, log=None, tables_only=True)[0] == []
assert all(
r.source != "fr-table"
for r in family_rounds(None, s2, FAM, ask=None, log=None)[0]
)
def test_rounds_inventory(tmp_path):
@@ -771,52 +1085,18 @@ def test_rounds_inventory(tmp_path):
assert any(r.code == "99341" for r in table)
def test_model_labels_that_numbers_contradict_or_junk_are_cleaned():
text = (
"The RUC recommended a work RVU of 2.27 for CPT code 99326; we are "
"assigning 2.63 work RVUs to CPT code 99326 after refinement."
)
out = {
"codes": [
_d(
"99326",
ruc_work_rvu=2.27,
final_work_rvu=2.63,
decision="accept",
comparators=["99326", "99350"],
pe_note="none (no PE inputs) Note: blah",
)
],
"surveying_societies": [],
"rationale_quote": "",
}
[r] = rows_from_output(out, para(9, text), {"99326"}, family="HV", model="m")
assert (r.decision, r.pe_note, r.comparators) == ("refine", "", "")
out["codes"][0]["pe_note"] = "removed SK062"
[r] = rows_from_output(out, para(9, text), {"99326"}, family="HV", model="m")
assert r.pe_note == "removed SK062"
def test_model_silent_percentile_is_taken_from_the_text_and_empty_rounds_dropped():
out = {
"codes": [_d("99341"), _d("99350"), _d("99342", ruc_percentile="75th")],
"surveying_societies": [],
"rationale_quote": "",
}
rows = {
r.code: r
for r in rows_from_output(
out, para(1676, P1676), CODES23, family="HV", model="m"
)
}
# 99341: the family sentence says 25th (except 99350); 99350 says median
assert rows["99341"].ruc_percentile == "25th"
assert rows["99350"].ruc_percentile == "median"
assert rows["99342"].ruc_percentile == "25th" # model's 75th contradicted
blank = {"codes": [_d("99341")], "surveying_societies": [], "rationale_quote": ""}
assert (
rows_from_output(
blank, para(1, "Only 99341 here."), CODES23, family="HV", model="m"
)
== []
def test_final_p_id_roundtrips_and_old_tables_gain_the_column():
con = duckdb.connect(":memory:")
con.execute("CREATE SCHEMA pfs")
con.execute(
"CREATE TABLE pfs.valuation_round (family VARCHAR, code VARCHAR, rule_year INTEGER, item_key VARCHAR,"
" rule_kind VARCHAR, p_id INTEGER, page INTEGER, source VARCHAR, table_title VARCHAR,"
" ruc_work_rvu DOUBLE, ruc_percentile VARCHAR, survey_intra_min DOUBLE, survey_total_min DOUBLE,"
" comparators VARCHAR, surveying_societies VARCHAR, requested_work_rvu DOUBLE,"
" proposed_work_rvu DOUBLE, final_work_rvu DOUBLE, decision VARCHAR, rationale_p_id INTEGER,"
" rationale_quote VARCHAR, pe_note VARCHAR, extractor VARCHAR)"
)
ensure_tables(con)
row = replace(_row(code="99341"), final_p_id=1685)
write_rounds(con, "F", [row])
assert read_rounds(con, "F")[0].final_p_id == 1685