diff --git a/src/cli/pfs.py b/src/cli/pfs.py index f239d10..11dedd7 100644 --- a/src/cli/pfs.py +++ b/src/cli/pfs.py @@ -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() diff --git a/src/pfs/codetables.py b/src/pfs/codetables.py index f7658be..b93c215 100644 --- a/src/pfs/codetables.py +++ b/src/pfs/codetables.py @@ -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: diff --git a/src/pfs/rounds.py b/src/pfs/rounds.py index bb0d422..6981e9d 100644 --- a/src/pfs/rounds.py +++ b/src/pfs/rounds.py @@ -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"(? 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

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) `` in one clause, (P2) `` … at the + `` before the next code, (P3) `` for all … codes, + except … ``. 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 diff --git a/tests/cli/test_pfs_cli.py b/tests/cli/test_pfs_cli.py index a2cf130..1c1479b 100644 --- a/tests/cli/test_pfs_cli.py +++ b/tests/cli/test_pfs_cli.py @@ -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"]) diff --git a/tests/pfs/fixtures/fr_table_cy2008_nprm_t10.txt b/tests/pfs/fixtures/fr_table_cy2008_nprm_t10.txt new file mode 100644 index 0000000..27dbd41 --- /dev/null +++ b/tests/pfs/fixtures/fr_table_cy2008_nprm_t10.txt @@ -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. +-------------------------------------------------------------------------------------------------------------------------------------------------------- diff --git a/tests/pfs/test_rounds.py b/tests/pfs/test_rounds.py index a79ffaa..2a79b01 100644 --- a/tests/pfs/test_rounds.py +++ b/tests/pfs/test_rounds.py @@ -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: " at the " + 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