feat(pfs): CMS comment/response summaries with quantifier, action and stance — kept apart from our counts
This commit is contained in:
kert
@@ -23,6 +23,7 @@ from typing import Any
|
||||
|
||||
import typer
|
||||
|
||||
from pfs.cms_summaries import summaries as cms_summaries_rows
|
||||
from pfs.codetables import (
|
||||
cpt_years,
|
||||
ensure_tables,
|
||||
@@ -36,6 +37,7 @@ from pfs.codetables import (
|
||||
read_exposures,
|
||||
read_guidance,
|
||||
read_reviews,
|
||||
write_cms_summaries,
|
||||
write_elements,
|
||||
write_events,
|
||||
write_exposures,
|
||||
@@ -1160,3 +1162,56 @@ def origin(
|
||||
n = write_origin(con, fam.key, rows)
|
||||
typer.echo(f"wrote {n} origins")
|
||||
_publish()
|
||||
|
||||
|
||||
@app.command("cms-summaries")
|
||||
def cms_summaries(
|
||||
family: str = typer.Option(..., "--family", help="Curated family key."),
|
||||
no_llm: bool = typer.Option(
|
||||
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)."),
|
||||
retry_failed: bool = typer.Option(
|
||||
False, "--retry-failed", help="Re-run ledger entries that failed."
|
||||
),
|
||||
write: bool = typer.Option(False, "--write"),
|
||||
) -> None:
|
||||
"""CMS's own Comment:/Response: summaries (pfs.cms_comment_summary):
|
||||
CMS's quantifier, the stance of the summarized comments and CMS's
|
||||
response action, per pair a rule prints. Kept apart from our own
|
||||
commenter counts."""
|
||||
fam = curated_family(normalize_key(family))
|
||||
if fam is None:
|
||||
raise typer.BadParameter(f"unknown curated family {family!r}")
|
||||
store = _store()
|
||||
model = _model_name()
|
||||
counts: dict[str, int] = {}
|
||||
rows = cms_summaries_rows(
|
||||
store,
|
||||
fam,
|
||||
classify=None if no_llm else _classifier(),
|
||||
model=model,
|
||||
log=RunLog(ledger_path("cms-summaries", fam.key)),
|
||||
limit=limit,
|
||||
retry_failed=retry_failed,
|
||||
replay_model=model if no_llm else None,
|
||||
counts=counts,
|
||||
)
|
||||
by_year: dict[tuple[int, str], int] = {}
|
||||
for r in rows:
|
||||
by_year[(r.rule_year, r.rule_kind)] = (
|
||||
by_year.get((r.rule_year, r.rule_kind), 0) + 1
|
||||
)
|
||||
for (year, kind), n in sorted(by_year.items()):
|
||||
typer.echo(f"CY{year} {kind}: {n} summaries")
|
||||
typer.echo(
|
||||
f"summaries: {len(rows)}; done {counts.get('done', 0)}, "
|
||||
f"skipped {counts.get('skipped', 0)}, failed {counts.get('failed', 0)}, "
|
||||
f"failed_pending {counts.get('failed_pending', 0)}"
|
||||
)
|
||||
if write:
|
||||
with _batch() as con:
|
||||
ensure_tables(con)
|
||||
n = write_cms_summaries(con, fam.key, rows)
|
||||
typer.echo(f"wrote {n} summaries")
|
||||
_publish()
|
||||
|
||||
179
src/pfs/cms_summaries.py
Normal file
179
src/pfs/cms_summaries.py
Normal file
@@ -0,0 +1,179 @@
|
||||
"""CMS's own summaries of public comments (``pfs.cms_comment_summary``,
|
||||
dossier task 7).
|
||||
|
||||
The Federal Register prints a rule's public-comment summaries as
|
||||
``Comment:`` / ``Response:`` paragraph pairs. Each pair that names a
|
||||
family code or one of the family's synonym phrases becomes one row
|
||||
carrying CMS's quantifier ("a majority of commenters"), the stance of
|
||||
the summarized comments and CMS's response action.
|
||||
|
||||
These are CMS's words about its own comments. They are stored apart
|
||||
from our own commenter counts (``pfs.comment_stance``, task 9) and the
|
||||
dossier shows the two side by side, never merged. A row holds labels and
|
||||
paragraph ids only, no comment text.
|
||||
|
||||
The quantifier and the response action are regex over the paragraph. The
|
||||
stance goes through the local closed-vocabulary classifier, one call per
|
||||
pair, through the ``pfs.runlog`` ledger (resumable, replayable with
|
||||
``--no-llm``). Corrections are not read (``rounds.rule_items``).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any, Callable, Sequence
|
||||
|
||||
from pfs.codetables import CmsSummaryRow
|
||||
from pfs.reaction import fr_pair_spans
|
||||
from pfs.rounds import _anchor_rows, rule_items
|
||||
from pfs.runlog import RunLog, input_hash, ledger_path, run_units
|
||||
|
||||
TASK = "cms-summaries"
|
||||
SUMMARY_PROMPT_VERSION = "v1"
|
||||
SUMMARY_STANCES = ("support", "modify", "oppose", "mixed", "unclear")
|
||||
_PROMPT_PREFIX = "CMS's summary of public comments:"
|
||||
_PROMPT_CHARS = 1500
|
||||
|
||||
# first match wins, in this order
|
||||
_QUANTIFIERS = (
|
||||
("all", re.compile(r"\ball (?:of the )?commenters\b", re.I)),
|
||||
(
|
||||
"majority",
|
||||
re.compile(r"\bmajority of (?:the )?commenters\b|\bmost commenters\b", re.I),
|
||||
),
|
||||
("many", re.compile(r"\bmany\b|\bnumerous commenters\b", re.I)),
|
||||
("several", re.compile(r"\bseveral\b", re.I)),
|
||||
("some", re.compile(r"\bsome\b", re.I)),
|
||||
("few", re.compile(r"\b(?:a )?few commenters\b", re.I)),
|
||||
("one", re.compile(r"\b(?:one|a|another) commenter\b", re.I)),
|
||||
)
|
||||
|
||||
_ACTIONS = (
|
||||
(
|
||||
"finalize_as_proposed",
|
||||
re.compile(
|
||||
r"(?<!not )\bfinaliz\w+[^.]{0,120}as proposed|\bcontinue to believe\b|\bmaintain\b",
|
||||
re.I,
|
||||
),
|
||||
),
|
||||
(
|
||||
"finalize_modified",
|
||||
re.compile(
|
||||
r"\bfinaliz\w+[^.]{0,120}(?:with (?:a )?modification|\brevis)"
|
||||
r"|\bwe have assigned\b|\bwe are assigning\b",
|
||||
re.I,
|
||||
),
|
||||
),
|
||||
(
|
||||
"not_finalized",
|
||||
re.compile(r"\bnot (?:to )?finaliz|\bare not adopting\b|\bdecline", re.I),
|
||||
),
|
||||
("acknowledged", re.compile(r"\bthank|\bappreciat", re.I)),
|
||||
)
|
||||
|
||||
|
||||
def quantifier_of(comment_text: str) -> str:
|
||||
"""CMS's quantifier for the summarized commenters, else ``unspecified``."""
|
||||
for name, pattern in _QUANTIFIERS:
|
||||
if pattern.search(comment_text):
|
||||
return name
|
||||
return "unspecified"
|
||||
|
||||
|
||||
def response_action_of(response_text: str) -> str:
|
||||
"""What CMS did in response, else ``unclear``. ``acknowledged`` means
|
||||
the response only thanks or appreciates."""
|
||||
for name, pattern in _ACTIONS:
|
||||
if pattern.search(response_text):
|
||||
return name
|
||||
return "unclear"
|
||||
|
||||
|
||||
def _prompt(comment_text: str) -> str:
|
||||
return f"{_PROMPT_PREFIX}\n{comment_text.strip()[:_PROMPT_CHARS]}"
|
||||
|
||||
|
||||
def summary_stance(
|
||||
comment_text: str, classify: Callable[[str, Sequence[str]], str | None]
|
||||
) -> str:
|
||||
"""Stance of the summarized comments; an answer outside the closed
|
||||
vocabulary counts as ``unclear``."""
|
||||
stance = classify(_prompt(comment_text), SUMMARY_STANCES)
|
||||
return stance if stance in SUMMARY_STANCES else "unclear"
|
||||
|
||||
|
||||
def summaries(
|
||||
store: Any,
|
||||
family: Any,
|
||||
*,
|
||||
classify: Callable[[str, Sequence[str]], str | None] | None = None,
|
||||
model: str | None = None,
|
||||
log: RunLog | None = None,
|
||||
limit: int = 0,
|
||||
retry_failed: bool = False,
|
||||
replay_model: str | None = None,
|
||||
counts: dict[str, int] | None = None,
|
||||
) -> list[CmsSummaryRow]:
|
||||
"""One row per Comment:/Response: pair in a PFS rule (corrections
|
||||
left out) naming a family code or synonym phrase. With *classify*
|
||||
(and its *model* name) the stances are asked through the ledger
|
||||
*log*; without it (``--no-llm``) the stances recorded for
|
||||
*replay_model* are replayed, and every other row is stance
|
||||
``unclear`` with extractor ``regex``. *counts* receives the run
|
||||
counts of the model calls."""
|
||||
kinds = {
|
||||
it["item_key"]: it["kind"]
|
||||
for it in rule_items(store, _anchor_rows(store, family.codes, family.synonyms))
|
||||
}
|
||||
spans = [
|
||||
s
|
||||
for s in fr_pair_spans(store, family.codes, family.synonyms)
|
||||
if s.item_key in kinds
|
||||
]
|
||||
units = {f"{s.item_key}:{s.comment_p_id}": s for s in spans}
|
||||
hashes = {u: input_hash(_prompt(s.comment_text)) for u, s in units.items()}
|
||||
model_name = model if classify is not None else replay_model
|
||||
recorded: dict[str, str] = {}
|
||||
if model_name is not None:
|
||||
log = log or RunLog(ledger_path(TASK, family.key))
|
||||
if classify is not None:
|
||||
result = run_units(
|
||||
[
|
||||
(
|
||||
u,
|
||||
hashes[u],
|
||||
lambda s=s: {
|
||||
"stance": summary_stance(s.comment_text, classify)
|
||||
},
|
||||
)
|
||||
for u, s in units.items()
|
||||
],
|
||||
log,
|
||||
model=model_name,
|
||||
version=SUMMARY_PROMPT_VERSION,
|
||||
limit=limit,
|
||||
retry_failed=retry_failed,
|
||||
)
|
||||
if counts is not None:
|
||||
counts.update(result)
|
||||
for u in units:
|
||||
entry = log.get(u, hashes[u], model_name, SUMMARY_PROMPT_VERSION)
|
||||
if entry is not None and entry.output is not None:
|
||||
recorded[u] = entry.output["stance"]
|
||||
extractor = f"model:{model_name}@{SUMMARY_PROMPT_VERSION}"
|
||||
return [
|
||||
CmsSummaryRow(
|
||||
family=family.key,
|
||||
item_key=s.item_key,
|
||||
rule_year=s.rule_year,
|
||||
rule_kind=kinds[s.item_key],
|
||||
comment_p_id=s.comment_p_id,
|
||||
response_p_id=s.response_p_id,
|
||||
codes=",".join(s.codes),
|
||||
quantifier=quantifier_of(s.comment_text),
|
||||
stance=recorded.get(u, "unclear"),
|
||||
response_action=response_action_of(s.response_text),
|
||||
extractor=extractor if u in recorded else "regex",
|
||||
)
|
||||
for u, s in units.items()
|
||||
]
|
||||
@@ -328,6 +328,30 @@ class CptCodeAltRow:
|
||||
reason: str # "guidelines-reprint" | "lower-score" (C9)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CmsSummaryRow:
|
||||
"""CMS's own summary of public comments (``pfs.cms_comment_summary``):
|
||||
one ``Comment:``/``Response:`` paragraph pair a rule prints. The
|
||||
quantifier ("a majority of commenters"), the stance and the response
|
||||
action are CMS's wording read by regex or a local model — never our
|
||||
own count of comments (``pfs.comment_stance``), never merged with it.
|
||||
Only labels and paragraph ids are stored, no comment text. ``codes``
|
||||
is empty for a pair matched by a family synonym alone."""
|
||||
|
||||
family: str
|
||||
item_key: str
|
||||
rule_year: int
|
||||
rule_kind: str
|
||||
comment_p_id: int
|
||||
response_p_id: int
|
||||
codes: str
|
||||
quantifier: str
|
||||
stance: str
|
||||
response_action: str
|
||||
#: ``regex`` (stance unclassified) or ``model:<model>@<prompt version>``
|
||||
extractor: str
|
||||
|
||||
|
||||
ORIGIN_KINDS = (
|
||||
"statute",
|
||||
"regulation",
|
||||
@@ -431,6 +455,10 @@ CREATE TABLE IF NOT EXISTS pfs.code_origin (
|
||||
family VARCHAR, code VARCHAR, kind VARCHAR, reference VARCHAR, year INTEGER,
|
||||
summary VARCHAR, cite_kind VARCHAR, item_key VARCHAR, p_id INTEGER,
|
||||
page_label VARCHAR, locator VARCHAR, extractor VARCHAR, ref_p_id INTEGER);
|
||||
CREATE TABLE IF NOT EXISTS pfs.cms_comment_summary (
|
||||
family VARCHAR, item_key VARCHAR, rule_year INTEGER, rule_kind VARCHAR,
|
||||
comment_p_id INTEGER, response_p_id INTEGER, codes VARCHAR, quantifier VARCHAR,
|
||||
stance VARCHAR, response_action VARCHAR, extractor VARCHAR);
|
||||
"""
|
||||
|
||||
|
||||
@@ -846,3 +874,18 @@ def read_origin(con: Any, family: str) -> list[OriginRow]:
|
||||
).fetchall()
|
||||
# Rows written before ``ref_p_id`` existed read back NULL for it.
|
||||
return [OriginRow(*r[:-1], r[-1] or 0) for r in rows]
|
||||
|
||||
|
||||
def write_cms_summaries(con: Any, family: str, rows: Sequence[CmsSummaryRow]) -> int:
|
||||
ensure_tables(con)
|
||||
con.execute("DELETE FROM pfs.cms_comment_summary WHERE family = ?", [family])
|
||||
return _insert(con, "pfs.cms_comment_summary", rows)
|
||||
|
||||
|
||||
def read_cms_summaries(con: Any, family: str) -> list[CmsSummaryRow]:
|
||||
rows = con.execute(
|
||||
"SELECT * FROM pfs.cms_comment_summary WHERE family = ? "
|
||||
"ORDER BY rule_year, item_key, comment_p_id",
|
||||
[family],
|
||||
).fetchall()
|
||||
return [CmsSummaryRow(*r) for r in rows]
|
||||
|
||||
@@ -33,6 +33,7 @@ from __future__ import annotations
|
||||
import json
|
||||
import re
|
||||
from collections import Counter
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Callable, Sequence
|
||||
|
||||
from sqlalchemy import text
|
||||
@@ -171,6 +172,100 @@ _COMMENT_RE = re.compile(r"^Comment:", re.I)
|
||||
_RESPONSE_RE = re.compile(r"^Response:", re.I)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PairSpan:
|
||||
"""One ``Comment:``/``Response:`` pair (Ruling A12) of an FR rule item:
|
||||
the paragraph ids of its two ends, their texts, and the family codes
|
||||
the whole span names (empty when only a synonym phrase qualified it)."""
|
||||
|
||||
item_key: str
|
||||
rule_year: int
|
||||
comment_p_id: int
|
||||
response_p_id: int
|
||||
comment_text: str
|
||||
response_text: str
|
||||
codes: tuple[str, ...]
|
||||
|
||||
|
||||
def _scan_pairs(
|
||||
store: Any, codes: Sequence[str], phrases: Sequence[str]
|
||||
) -> tuple[list[PairSpan], dict[str, int]]:
|
||||
"""Every qualifying pair, and each rule item's total ``Comment:``
|
||||
paragraph count (only items with a qualifying pair appear in it)."""
|
||||
if not codes and not phrases:
|
||||
return [], {}
|
||||
con = store._con() # noqa: SLF001 — same pattern as pfs.descriptors/pfs.guidance
|
||||
likes = [f"%{c.upper()}%" for c in codes] + [f"%{x}%" for x in phrases]
|
||||
like_sql = "".join(" OR a.text LIKE ?" for _ in likes)
|
||||
rows = con.execute(
|
||||
"SELECT a.item_key, a.p_id, a.text, i.title, i.date_published "
|
||||
"FROM fr_anchors a JOIN items i ON i.key = a.item_key "
|
||||
f"WHERE a.text LIKE 'Comment:%' OR a.text LIKE 'Response:%'{like_sql} "
|
||||
"ORDER BY a.item_key, a.p_id",
|
||||
likes,
|
||||
).fetchall()
|
||||
|
||||
pattern = code_pattern(codes) if codes else None
|
||||
phrase_re = (
|
||||
re.compile(r"\b(?:" + "|".join(re.escape(x) for x in phrases) + r")\b", re.I)
|
||||
if phrases
|
||||
else None
|
||||
)
|
||||
by_item: dict[str, list[tuple[int, str]]] = {}
|
||||
meta: dict[str, tuple[str, str]] = {}
|
||||
for item_key, p_id, para_text, title, date_published in rows:
|
||||
by_item.setdefault(item_key, []).append((p_id, para_text or ""))
|
||||
meta[item_key] = (title, date_published)
|
||||
|
||||
spans: list[PairSpan] = []
|
||||
totals: dict[str, int] = {}
|
||||
for item_key, paras in by_item.items():
|
||||
found: list[PairSpan] = []
|
||||
opened: list[tuple[int, str]] | None = None # None: no comment open
|
||||
for p_id, para_text in paras:
|
||||
if _COMMENT_RE.match(para_text):
|
||||
# A new Comment: abandons any still-open, unresolved one
|
||||
# (Ruling A12: unpaired, never counted).
|
||||
opened = [(p_id, para_text)]
|
||||
continue
|
||||
if _RESPONSE_RE.match(para_text):
|
||||
if opened is not None:
|
||||
opened.append((p_id, para_text))
|
||||
whole = " ".join(t for _, t in opened)
|
||||
named = codes_in(whole, pattern) if pattern else ()
|
||||
if named or (phrase_re and phrase_re.search(whole)):
|
||||
title, date_published = meta[item_key]
|
||||
found.append(
|
||||
PairSpan(
|
||||
item_key,
|
||||
rule_year_of(title, date_published),
|
||||
opened[0][0],
|
||||
p_id,
|
||||
opened[0][1],
|
||||
para_text,
|
||||
named,
|
||||
)
|
||||
)
|
||||
opened = None
|
||||
continue
|
||||
if opened is not None:
|
||||
opened.append((p_id, para_text)) # continuation paragraph
|
||||
if found:
|
||||
spans += found
|
||||
totals[item_key] = sum(1 for _, t in paras if _COMMENT_RE.match(t))
|
||||
return spans, totals
|
||||
|
||||
|
||||
def fr_pair_spans(
|
||||
store: Any, codes: Sequence[str], phrases: Sequence[str] = ()
|
||||
) -> list[PairSpan]:
|
||||
"""Every ``Comment:``/``Response:`` pair (Ruling A12, as ``fr_pairs``)
|
||||
whose span names one of *codes* or, when given, matches one of
|
||||
*phrases* (case-insensitive, word-bounded), oldest rule first."""
|
||||
spans, _ = _scan_pairs(store, codes, phrases)
|
||||
return sorted(spans, key=lambda x: (x.rule_year, x.item_key, x.comment_p_id))
|
||||
|
||||
|
||||
def fr_pairs(store: Any, codes: Sequence[str]) -> list[tuple[str, int, int, int]]:
|
||||
"""``(item_key, rule_year, n_pairs, n_comment_paras)`` per rule item —
|
||||
the pre-2017 proxy for public reaction, since regulations.gov comment
|
||||
@@ -206,56 +301,10 @@ def fr_pairs(store: Any, codes: Sequence[str]) -> list[tuple[str, int, int, int]
|
||||
space before the colon) so the Python pass never counts a paragraph
|
||||
the SQL wouldn't have fetched as an opener in the first place.
|
||||
"""
|
||||
if not codes:
|
||||
return []
|
||||
con = store._con() # noqa: SLF001 — same pattern as pfs.descriptors/pfs.guidance
|
||||
code_likes = " OR ".join("a.text LIKE ?" for _ in codes)
|
||||
params = [f"%{c.upper()}%" for c in codes]
|
||||
rows = con.execute(
|
||||
"SELECT a.item_key, a.p_id, a.text, i.title, i.date_published "
|
||||
"FROM fr_anchors a JOIN items i ON i.key = a.item_key "
|
||||
f"WHERE a.text LIKE 'Comment:%' OR a.text LIKE 'Response:%' OR ({code_likes}) "
|
||||
"ORDER BY a.item_key, a.p_id",
|
||||
params,
|
||||
).fetchall()
|
||||
|
||||
pattern = code_pattern(codes)
|
||||
by_item: dict[str, list[str]] = {}
|
||||
meta: dict[str, tuple[str, str]] = {}
|
||||
for item_key, _p_id, para_text, title, date_published in rows:
|
||||
by_item.setdefault(item_key, []).append(para_text or "")
|
||||
meta[item_key] = (title, date_published)
|
||||
|
||||
out: list[tuple[str, int, int, int]] = []
|
||||
for item_key, paras in by_item.items():
|
||||
n_pairs = 0
|
||||
span: list[str] | None = None # None: no comment currently open
|
||||
for para_text in paras:
|
||||
if _COMMENT_RE.match(para_text):
|
||||
# A new Comment: abandons any still-open, unresolved one
|
||||
# (Ruling A12: unpaired, never counted).
|
||||
span = [para_text]
|
||||
continue
|
||||
if _RESPONSE_RE.match(para_text):
|
||||
if span is not None:
|
||||
span.append(para_text)
|
||||
if codes_in(" ".join(span), pattern):
|
||||
n_pairs += 1
|
||||
span = None
|
||||
continue
|
||||
if span is not None:
|
||||
span.append(para_text) # continuation paragraph
|
||||
if n_pairs:
|
||||
n_comment_paras = sum(1 for p in paras if _COMMENT_RE.match(p))
|
||||
title, date_published = meta[item_key]
|
||||
out.append(
|
||||
(
|
||||
item_key,
|
||||
rule_year_of(title, date_published),
|
||||
n_pairs,
|
||||
n_comment_paras,
|
||||
)
|
||||
)
|
||||
spans, totals = _scan_pairs(store, codes, ())
|
||||
n_pairs = Counter(x.item_key for x in spans)
|
||||
years = {x.item_key: x.rule_year for x in spans}
|
||||
out = [(k, years[k], n, totals[k]) for k, n in n_pairs.items()]
|
||||
return sorted(out, key=lambda r: (r[1], r[0]))
|
||||
|
||||
|
||||
|
||||
@@ -1621,3 +1621,120 @@ class TestOrigin:
|
||||
assert res.exit_code == 0, res.output
|
||||
assert "2 candidate Federal Register paragraphs" in res.output
|
||||
assert "3ANRT854: 3 pages" in res.output and "no model calls made" in res.output
|
||||
|
||||
|
||||
class TestCmsSummaries:
|
||||
def _rows(self):
|
||||
from pfs.codetables import CmsSummaryRow
|
||||
|
||||
return [
|
||||
CmsSummaryRow(
|
||||
"HOME-VISITS",
|
||||
"K1",
|
||||
2023,
|
||||
"final",
|
||||
1679,
|
||||
1680,
|
||||
"",
|
||||
"majority",
|
||||
"support",
|
||||
"acknowledged",
|
||||
"regex",
|
||||
),
|
||||
CmsSummaryRow(
|
||||
"HOME-VISITS",
|
||||
"K1",
|
||||
2023,
|
||||
"final",
|
||||
1681,
|
||||
1682,
|
||||
"99341",
|
||||
"few",
|
||||
"modify",
|
||||
"finalize_as_proposed",
|
||||
"regex",
|
||||
),
|
||||
CmsSummaryRow(
|
||||
"HOME-VISITS",
|
||||
"K0",
|
||||
2008,
|
||||
"final",
|
||||
2241,
|
||||
2242,
|
||||
"99326",
|
||||
"unspecified",
|
||||
"oppose",
|
||||
"finalize_modified",
|
||||
"regex",
|
||||
),
|
||||
]
|
||||
|
||||
def _patch(self, monkeypatch, seen, counts=None):
|
||||
def fake(store, fam, **kw):
|
||||
seen.update(kw, key=fam.key)
|
||||
kw["counts"].update(
|
||||
{"done": 2, "skipped": 1, "failed": 0, "failed_pending": 3}
|
||||
if counts is None
|
||||
else counts
|
||||
)
|
||||
return self._rows()
|
||||
|
||||
monkeypatch.setattr(pfs_cli, "cms_summaries_rows", fake)
|
||||
monkeypatch.setattr(pfs_cli, "_classifier", lambda: "CLS")
|
||||
monkeypatch.setattr(pfs_cli, "_model_name", lambda: "qwen-x")
|
||||
monkeypatch.setattr(
|
||||
pfs_cli, "ledger_path", lambda t, f: Path("/nonexistent") / f"{t}-{f}"
|
||||
)
|
||||
|
||||
def test_writes_and_publishes(self, con, monkeypatch):
|
||||
from pfs.codetables import read_cms_summaries
|
||||
|
||||
seen = {}
|
||||
self._patch(monkeypatch, seen)
|
||||
res = runner.invoke(
|
||||
app,
|
||||
[
|
||||
"pfs",
|
||||
"cms-summaries",
|
||||
"--family",
|
||||
"home-visits",
|
||||
"--limit",
|
||||
"4",
|
||||
"--retry-failed",
|
||||
"--write",
|
||||
],
|
||||
)
|
||||
assert res.exit_code == 0, res.output
|
||||
assert seen["classify"] == "CLS" and seen["model"] == "qwen-x"
|
||||
assert (
|
||||
seen["limit"] == 4 and seen["retry_failed"] and seen["replay_model"] is None
|
||||
)
|
||||
assert seen["key"] == "HOME-VISITS"
|
||||
assert "CY2008 final: 1 summaries" in res.output
|
||||
assert "CY2023 final: 2 summaries" in res.output
|
||||
assert "done 2, skipped 1, failed 0, failed_pending 3" in res.output
|
||||
assert "wrote 3 summaries" in res.output
|
||||
assert con.published == [True]
|
||||
assert len(read_cms_summaries(con._real, "HOME-VISITS")) == 3
|
||||
|
||||
def test_without_write_writes_nothing(self, con, monkeypatch):
|
||||
self._patch(monkeypatch, {})
|
||||
monkeypatch.setattr(pfs_cli, "_batch", lambda: 1 / 0)
|
||||
res = runner.invoke(app, ["pfs", "cms-summaries", "--family", "HOME-VISITS"])
|
||||
assert res.exit_code == 0, res.output
|
||||
assert "wrote" not in res.output
|
||||
|
||||
def test_no_llm_builds_no_classifier_and_replays(self, con, monkeypatch):
|
||||
seen = {}
|
||||
self._patch(monkeypatch, seen, counts={})
|
||||
monkeypatch.setattr(pfs_cli, "_classifier", lambda: 1 / 0)
|
||||
res = runner.invoke(
|
||||
app, ["pfs", "cms-summaries", "--family", "HOME-VISITS", "--no-llm"]
|
||||
)
|
||||
assert res.exit_code == 0, res.output
|
||||
assert seen["classify"] is None and seen["replay_model"] == "qwen-x"
|
||||
assert "done 0, skipped 0, failed 0, failed_pending 0" in res.output
|
||||
|
||||
def test_unknown_family(self, con):
|
||||
res = runner.invoke(app, ["pfs", "cms-summaries", "--family", "NOPE"])
|
||||
assert res.exit_code != 0 and "unknown curated family" in res.output
|
||||
|
||||
17
tests/pfs/fixtures/fr_comment_pairs.json
Normal file
17
tests/pfs/fixtures/fr_comment_pairs.json
Normal file
@@ -0,0 +1,17 @@
|
||||
{
|
||||
"PETM225D:1673": "The following is a summary of the comments we received and our responses.",
|
||||
"PETM225D:1674": "Comment: The commenters supported our proposal to accept the CPT coding changes for Home or Residence Services codes.",
|
||||
"PETM225D:1675": "Response: We appreciate the feedback from the commenters regarding our proposal, and are finalizing as proposed to adopt the CPT merger of the home and domiciliary visits into one family, and adopt the CPT codes as revised for reporting these services.",
|
||||
"PETM225D:1676": "The RUC survey time includes pre-service time 3 days before the date of encounter, intraservice time on the date of encounter, and post-service time that includes 7 days after the date of encounter. These eight CPT codes were reviewed at the October 2021 RUC meeting with recommendations submitted to CMS for the CY 2023 rule cycle. 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. We proposed the RUC-recommended work RVU for all eight CPT codes in the Home or Residence Services CPT code family. We proposed a work RVU of 1.00 for CPT code 99341 ( Home or residence visit for the evaluation and management of a new patient, which requires a medically appropriate history and/or examination and straightforward medical decision making), a work RVU of 1.65 for CPT code 99342 ( Home or residence visit for the evaluation and management of a new patient, which requires a medically appropriate history and/or examination and low level of medical decision making), a work RVU of 2.87 for CPT code 99344 ( Home or residence visit for the evaluation and management of a new patient, which requires a medically appropriate history and/or examination and moderate level of medical decision making), a work RVU of 3.88 for CPT code 99345 ( Home or residence visit for the evaluation and management of a new patient, which requires a medically appropriate history and/or examination and high level of medical decision making), a work RVU of 0.90 for CPT code 99347 ( Home or residence visit for the evaluation and management of an established patient, which requires a medically appropriate history and/or examination straightforward medical decision making), a work RVU of 1.50 for CPT code 99348 ( Home or residence visit for the evaluation and management of an established patient, which requires a medically appropriate history and/or examination and low level of medical decision making), a work RVU of 2.44 for CPT code 99349 ( Home or residence visit for the evaluation and management of an established patient, which requires a medically appropriate history and/or examination and moderate level of medical decision making), and a work RVU of 3.60 for CPT code 99350 ( Home or residence visit for the evaluation and management of an established patient, which requires a medically appropriate history and/or examination and high level of medical decision making).",
|
||||
"PETM225D:1677": "We proposed the RUC-recommended direct PE inputs for CPT codes 99345 and 99347-99350 without refinement. For CPT codes 99341 and 99342, we are refining the direct PE inputs by removing supply item SK062 (patient education booklet). For CPT code 99344, we are refining the direct PE inputs by removing supply items SK062 (patient education booklet), SJ053 (swab-pad, alcohol), and SJ061 (tongue depressor). Per the PE Summary of Recommendations provided by the RUC, CPT codes 99341, 99342, 99344, and 99347 would typically have procedures performed on the same date of service. For those CPT codes, the RUC stated that they removed supplies that would be duplicative, such as gloves, alcohol wipes, booklet, and tongue depressor. However, we found that not all of these duplicative supplies had been removed from CPT codes ( printed page 69609) 99341, 99342, and 99344 by the RUC. Therefore, we proposed to remove these duplicative supplies from CPT codes 99341, 99342, and 99344, and accept the remaining RUC-recommended direct PE inputs without refinement.",
|
||||
"PETM225D:1678": "The following is a summary of the comments we received and our responses.",
|
||||
"PETM225D:1679": "Comment: The majority of commenters supported our proposal to accept the RUC-recommended RVUs for all of the Home or Residence Services codes.",
|
||||
"PETM225D:1680": "Response: We appreciate the feedback from the commenters regarding our proposal.",
|
||||
"PETM225D:1681": "Comment: A few commenters stated the RVUs for the Home or Residence Services code family were too low, and fail to adequately account for travel, addressing social determinants of health, and other comprehensive care. One commenter suggested that we maintain the current RVUs for codes 99341, 99344, 99345, 99347, and 99348, for which the RUC-recommended RVUs decreased, while supporting our proposal to accept the increased RVUs for codes 99342, 99349, and 99350 recommended by the RUC. Another commenter suggested that we increase the RVUs for the entire code family to reflect travel expenses, and expressed concern that unreimbursed travel will be detrimental to the care provided to very ill patients and threaten the financial viability of physician practices that provide home or residence visits.",
|
||||
"PETM225D:1682": "Response: We continue to believe that the RUC-recommended RVUs are the appropriate values for CPT codes 99341, 99342, 99344, 99345, and 99347-99350. We appreciate the commenters' remarks, and we acknowledge the concerns regarding providing patient care and addressing social determinants of health and other issues during home or residence visits. Historically, travel costs incurred by the physician or practitioner are not included in the valuation of E/M codes, since travel time and/or mileage is not considered a resource involved in furnishing the service. The RUC survey for the Home or Residence Services code family did not include information on physician travel or mileage. Also, the CPT E/M guidelines for selecting the level of home or residence service, when performed based on time, specifically says not to count the time spent on travel (2023 CPT Codebook, p. 26). Therefore, we are finalizing the RUC-recommended work RVUs as proposed.",
|
||||
"PETM225D:1683": "Comment: One commenter requested that we not refine the direct PE inputs for codes 99341, 99342, and 99344 as proposed. For CPT codes 99341 and 99342, we proposed removing supply item SK062 (patient education booklet). For CPT code 99344, we proposed removing supply items SK062 (patient education booklet), SJ053 (swab-pad, alcohol), and SJ061 (tongue depressor).",
|
||||
"PETM225D:1684": "Response: Per the PE Summary of Recommendations provided by the RUC, CPT codes 99341, 99342, 99344, and 99347 would typically have procedures performed on the same date of service, which would already include some of the same supply items. For these CPT codes, the RUC stated that they had removed the PE inputs for supplies that would be duplicative with the procedure on the same day, such as gloves, alcohol wipes, a booklet, and a tongue depressor. Although the RUC did remove the duplicative supplies for CPT code 99347 from the direct PE inputs they recommended, we found that the RUC had not removed all of these duplicative supplies from CPT codes 99341, 99342, and 99344. Therefore, we proposed to remove these duplicative supplies from the direct PE inputs for CPT codes 99341, 99342, and 99344. We continue to believe these supplies are duplicative with a same-day procedure, and are, therefore, finalizing the direct PE inputs for CPT codes 99341, 99342, 99344, and 99347 as proposed.",
|
||||
"PETM225D:1685": "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.",
|
||||
"54GTKR5G:2241": "Comment: Commenters disagreed with the RUC-recommended work values for these services, which we had accepted. The commenters disagreed with the RUC-recommended work RVUs and believed the services were undervalued. The commenters also believed that the home visit work RVUs should remain “relatively” the same with respect to office visit codes as they did prior to the five-year review and requested that CMS reject the RUC recommended work RVUs and follow their survey values. Based on these comments, we referred these codes to the multi-specialty validation panel for review.",
|
||||
"54GTKR5G:2242": "Response: As a result of the statistical analysis of the 2007 multi-specialty validation panel ratings, we have assigned 2.63 work RVUs to CPT code 99326; 3.46 work RVUs to CPT code 99327; 4.09 work RVUs to CPT code 99328; 1.07 work RVUs to CPT code 99334; 1.72 work RVUs to CPT code 99335; 2.46 work RVUs to CPT code 99336; 3.58 work RVUs to CPT code 99337; 2.53 work RVUs to CPT code 99343; 3.38 work RVUs to CPT code 99344; 4.09 work RVUs to CPT code 99345; 1.00 work RVUs to CPT code 99347; 1.56 work RVUs to CPT code 99348; 2.33 work RVUs to CPT code 99349; and 3.28 work RVUs to CPT code 99350."
|
||||
}
|
||||
324
tests/pfs/test_cms_summaries.py
Normal file
324
tests/pfs/test_cms_summaries.py
Normal file
@@ -0,0 +1,324 @@
|
||||
"""pfs.cms_summaries — CMS's own Comment:/Response: summaries (dossier task 7)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import duckdb
|
||||
import pytest
|
||||
|
||||
from bib.item import Rule
|
||||
from bib.store import Store
|
||||
from pfs.cms_summaries import (
|
||||
SUMMARY_PROMPT_VERSION,
|
||||
SUMMARY_STANCES,
|
||||
quantifier_of,
|
||||
response_action_of,
|
||||
summaries,
|
||||
summary_stance,
|
||||
)
|
||||
from pfs.codetables import (
|
||||
CmsSummaryRow,
|
||||
ensure_tables,
|
||||
read_cms_summaries,
|
||||
write_cms_summaries,
|
||||
)
|
||||
from pfs.reaction import PairSpan, fr_pair_spans
|
||||
from pfs.runlog import RunLog
|
||||
|
||||
_RAW = json.loads(
|
||||
(Path(__file__).parent / "fixtures" / "fr_comment_pairs.json").read_text()
|
||||
)
|
||||
# Live Federal Register paragraph text, keyed by paragraph id per rule item.
|
||||
CY2023 = {int(k.split(":")[1]): v for k, v in _RAW.items() if k.startswith("PETM")}
|
||||
CY2008 = {int(k.split(":")[1]): v for k, v in _RAW.items() if k.startswith("54GT")}
|
||||
TEXT = {**CY2023, **CY2008}
|
||||
|
||||
FAM = SimpleNamespace(
|
||||
key="HOME-VISITS",
|
||||
codes=("99341", "99342", "99350", "99326"),
|
||||
synonyms=("home or residence services",),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"p,q",
|
||||
[(1679, "majority"), (1681, "few"), (1674, "unspecified"), (2241, "unspecified")],
|
||||
)
|
||||
def test_quantifier(p, q):
|
||||
assert quantifier_of(TEXT[p]) == q
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"text,q",
|
||||
[
|
||||
("Comment: All commenters agreed.", "all"),
|
||||
("Comment: All of the commenters agreed.", "all"),
|
||||
("Comment: Most commenters agreed.", "majority"),
|
||||
("Comment: Many commenters agreed.", "many"),
|
||||
("Comment: Numerous commenters agreed.", "many"),
|
||||
("Comment: Several commenters agreed.", "several"),
|
||||
("Comment: Some commenters agreed.", "some"),
|
||||
("Comment: One commenter agreed.", "one"),
|
||||
("Comment: Another commenter agreed.", "one"),
|
||||
("Comment: A commenter agreed.", "one"),
|
||||
("Comment: Commenters in somewhere else.", "unspecified"),
|
||||
("Comment: Manyfold commenters.", "unspecified"),
|
||||
],
|
||||
)
|
||||
def test_quantifier_words(text, q):
|
||||
assert quantifier_of(text) == q
|
||||
|
||||
|
||||
def test_a_few_is_not_one():
|
||||
assert quantifier_of("A few commenters said") == "few"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"p,a",
|
||||
[
|
||||
(1675, "finalize_as_proposed"),
|
||||
(1680, "acknowledged"),
|
||||
(1682, "finalize_as_proposed"),
|
||||
(2242, "finalize_modified"),
|
||||
],
|
||||
)
|
||||
def test_response_action(p, a):
|
||||
assert response_action_of(TEXT[p]) == a
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"text,a",
|
||||
[
|
||||
("Response: We are finalizing it with modifications.", "finalize_modified"),
|
||||
(
|
||||
"Response: We are finalizing the value, revised from 2.0.",
|
||||
"finalize_modified",
|
||||
),
|
||||
("Response: We are not finalizing this as proposed.", "not_finalized"),
|
||||
("Response: We decline to adopt it.", "not_finalized"),
|
||||
("Response: We are not adopting that.", "not_finalized"),
|
||||
("Response: Thank you for the comment.", "acknowledged"),
|
||||
("Response: We disagree.", "unclear"),
|
||||
],
|
||||
)
|
||||
def test_response_action_words(text, a):
|
||||
assert response_action_of(text) == a
|
||||
|
||||
|
||||
def _store():
|
||||
s = Store(":memory:")
|
||||
con = s._con() # noqa: SLF001
|
||||
keys = {}
|
||||
for tag, title, date, texts in (
|
||||
("c23", "Medicare Program; CY 2023 PFS Final Rule", "2022-11-18", CY2023),
|
||||
("c08", "Medicare Program; CY 2008 PFS Final Rule", "2007-12-27", CY2008),
|
||||
):
|
||||
key = s.create(Rule(title=title, date_published=date))
|
||||
keys[tag] = key
|
||||
con.executemany(
|
||||
"INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)",
|
||||
[(key, p, 1, i, t) for i, (p, t) in enumerate(sorted(texts.items()))],
|
||||
)
|
||||
con.commit()
|
||||
return s, keys
|
||||
|
||||
|
||||
def test_fr_pair_spans_by_code_and_phrase():
|
||||
s, keys = _store()
|
||||
try:
|
||||
by_code = fr_pair_spans(s, ("99341",))
|
||||
assert [(x.comment_p_id, x.response_p_id) for x in by_code] == [
|
||||
(1681, 1682),
|
||||
(1683, 1684),
|
||||
]
|
||||
sp = by_code[0]
|
||||
assert isinstance(sp, PairSpan) and sp.rule_year == 2023
|
||||
assert sp.item_key == keys["c23"] and sp.codes[0] == "99341"
|
||||
assert sp.comment_text == TEXT[1681] and sp.response_text == TEXT[1682]
|
||||
both = fr_pair_spans(s, ("99341",), ("Home or Residence Services",))
|
||||
assert [x.comment_p_id for x in both] == [1674, 1679, 1681, 1683]
|
||||
phrase_only = [x for x in both if x.comment_p_id == 1679][0]
|
||||
assert phrase_only.codes == ()
|
||||
assert fr_pair_spans(s, (), ()) == []
|
||||
finally:
|
||||
s.close()
|
||||
|
||||
|
||||
def test_fr_pair_spans_phrase_only_input():
|
||||
s, _ = _store()
|
||||
try:
|
||||
out = fr_pair_spans(s, (), ("home or residence services",))
|
||||
assert [x.comment_p_id for x in out] == [1674, 1679, 1681]
|
||||
finally:
|
||||
s.close()
|
||||
|
||||
|
||||
def test_summaries_pair_and_classify(tmp_path):
|
||||
s, keys = _store()
|
||||
seen = []
|
||||
|
||||
def classify(text, choices):
|
||||
seen.append((text, tuple(choices)))
|
||||
return "support" if "majority" in text.lower() else "modify"
|
||||
|
||||
try:
|
||||
rows = summaries(
|
||||
s, FAM, classify=classify, model="m", log=RunLog(tmp_path / "l.jsonl")
|
||||
)
|
||||
by = {(r.item_key, r.comment_p_id): r for r in rows}
|
||||
r = by[(keys["c23"], 1679)]
|
||||
assert (r.response_p_id, r.quantifier, r.stance) == (
|
||||
1680,
|
||||
"majority",
|
||||
"support",
|
||||
)
|
||||
assert r.family == "HOME-VISITS" and r.rule_year == 2023
|
||||
assert r.rule_kind == "final" and r.codes == ""
|
||||
assert (
|
||||
r.response_action == "acknowledged"
|
||||
and r.extractor == f"model:m@{SUMMARY_PROMPT_VERSION}"
|
||||
)
|
||||
assert by[(keys["c23"], 1681)].codes == "99341,99342,99350"
|
||||
assert by[(keys["c08"], 2241)].codes == "99326,99350"
|
||||
assert by[(keys["c08"], 2241)].response_action == "finalize_modified"
|
||||
assert seen[0][0].startswith("CMS's summary of public comments:")
|
||||
assert seen[0][1] == SUMMARY_STANCES
|
||||
finally:
|
||||
s.close()
|
||||
|
||||
|
||||
def test_summaries_without_classifier_are_regex_unclear():
|
||||
s, _ = _store()
|
||||
try:
|
||||
rows = summaries(s, FAM)
|
||||
assert rows and {r.stance for r in rows} == {"unclear"}
|
||||
assert {r.extractor for r in rows} == {"regex"}
|
||||
finally:
|
||||
s.close()
|
||||
|
||||
|
||||
def test_summary_stance_outside_vocabulary_is_unclear():
|
||||
assert summary_stance("x", lambda t, c: "support") == "support"
|
||||
assert summary_stance("x", lambda t, c: "banana") == "unclear"
|
||||
assert summary_stance("x", lambda t, c: None) == "unclear"
|
||||
|
||||
|
||||
def test_summaries_ledger_resume_and_replay(tmp_path):
|
||||
s, keys = _store()
|
||||
log = RunLog(tmp_path / "l.jsonl")
|
||||
calls = []
|
||||
|
||||
def classify(text, choices):
|
||||
calls.append(text)
|
||||
return "oppose"
|
||||
|
||||
try:
|
||||
counts: dict = {}
|
||||
first = summaries(s, FAM, classify=classify, model="m", log=log, counts=counts)
|
||||
n = len(first)
|
||||
assert counts["done"] == n and len(calls) == n
|
||||
assert any(f"{keys['c23']}:1679" == e.unit for e in log._load()) # noqa: SLF001
|
||||
again: dict = {}
|
||||
summaries(s, FAM, classify=classify, model="m", log=log, counts=again)
|
||||
assert again["skipped"] == n and len(calls) == n # no new calls
|
||||
# --no-llm replay: same rows, zero calls
|
||||
replay = summaries(s, FAM, log=log, replay_model="m")
|
||||
assert replay == first and len(calls) == n
|
||||
# unrecorded under a different model: unclear / regex
|
||||
other = summaries(s, FAM, log=log, replay_model="other")
|
||||
assert {r.stance for r in other} == {"unclear"}
|
||||
assert {r.extractor for r in other} == {"regex"}
|
||||
# limit caps new calls; the rest stay regex
|
||||
fresh = RunLog(tmp_path / "l2.jsonl")
|
||||
capped = summaries(s, FAM, classify=classify, model="m", log=fresh, limit=1)
|
||||
assert sum(r.extractor != "regex" for r in capped) == 1
|
||||
finally:
|
||||
s.close()
|
||||
|
||||
|
||||
def test_summaries_failed_unit_stays_regex(tmp_path):
|
||||
s, _ = _store()
|
||||
log = RunLog(tmp_path / "l.jsonl")
|
||||
try:
|
||||
# a logged failure (output None) is not replayed as a stance
|
||||
rows = summaries(s, FAM, classify=lambda t, c: "support", model="m", log=log)
|
||||
unit = f"{rows[0].item_key}:{rows[0].comment_p_id}"
|
||||
e = log.get(unit, *_hash_of(log, unit), "m", SUMMARY_PROMPT_VERSION)
|
||||
assert e.output == {"stance": "support"}
|
||||
log2 = RunLog(tmp_path / "l3.jsonl")
|
||||
log2.put(unit, e.input_hash, "m", SUMMARY_PROMPT_VERSION, None)
|
||||
out = summaries(s, FAM, log=log2, replay_model="m")
|
||||
assert out[0].stance == "unclear" and out[0].extractor == "regex"
|
||||
finally:
|
||||
s.close()
|
||||
|
||||
|
||||
def _hash_of(log, unit):
|
||||
e = next(e for e in log._load() if e.unit == unit) # noqa: SLF001
|
||||
return (e.input_hash,)
|
||||
|
||||
|
||||
def test_summaries_skip_corrections_and_non_pfs_duplicates():
|
||||
s, keys = _store()
|
||||
try:
|
||||
s.create(
|
||||
Rule(
|
||||
title="Medicare Program; CY 2023 PFS Correction",
|
||||
date_published="2023-01-05",
|
||||
)
|
||||
)
|
||||
# a correction item carrying the same text must add no rows
|
||||
corr = [
|
||||
k
|
||||
for k in [
|
||||
s.create(
|
||||
Rule(
|
||||
title="CY 2023 PFS Correction notice",
|
||||
date_published="2023-01-06",
|
||||
)
|
||||
)
|
||||
]
|
||||
][0]
|
||||
s._con().execute( # noqa: SLF001
|
||||
"INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)",
|
||||
(corr, 1, 1, 1, "Comment: A commenter on 99341."),
|
||||
)
|
||||
s._con().execute( # noqa: SLF001
|
||||
"INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)",
|
||||
(corr, 2, 1, 2, "Response: We thank the commenter."),
|
||||
)
|
||||
s._con().commit() # noqa: SLF001
|
||||
assert corr not in {r.item_key for r in summaries(s, FAM)}
|
||||
finally:
|
||||
s.close()
|
||||
|
||||
|
||||
def test_write_read_roundtrip():
|
||||
con = duckdb.connect(":memory:")
|
||||
ensure_tables(con)
|
||||
row = CmsSummaryRow(
|
||||
"HOME-VISITS",
|
||||
"K",
|
||||
2023,
|
||||
"final",
|
||||
1679,
|
||||
1680,
|
||||
"",
|
||||
"majority",
|
||||
"support",
|
||||
"acknowledged",
|
||||
"regex",
|
||||
)
|
||||
other = CmsSummaryRow(
|
||||
"OTHER", "K", 2023, "final", 1, 2, "99341", "one", "unclear", "unclear", "regex"
|
||||
)
|
||||
assert write_cms_summaries(con, "HOME-VISITS", [row]) == 1
|
||||
write_cms_summaries(con, "OTHER", [other])
|
||||
assert write_cms_summaries(con, "HOME-VISITS", [row]) == 1 # delete-then-insert
|
||||
assert read_cms_summaries(con, "HOME-VISITS") == [row]
|
||||
assert write_cms_summaries(con, "HOME-VISITS", []) == 0
|
||||
assert read_cms_summaries(con, "HOME-VISITS") == []
|
||||
assert read_cms_summaries(con, "OTHER") == [other]
|
||||
@@ -387,3 +387,16 @@ class TestReactionTable:
|
||||
write_reaction(con, "G2211", [self._row(family="G2211")])
|
||||
write_reaction(con, "CCM", [])
|
||||
assert len(read_reaction(con, "G2211")) == 1
|
||||
|
||||
|
||||
class TestFrPairSpansRefactor:
|
||||
def test_counts_derive_from_spans(self, store_with_pairs):
|
||||
from pfs.reaction import fr_pair_spans
|
||||
|
||||
s, item_key = store_with_pairs
|
||||
spans = fr_pair_spans(s, ("99490",))
|
||||
assert [
|
||||
(x.item_key, x.rule_year, x.comment_p_id, x.response_p_id) for x in spans
|
||||
] == [(item_key, 2015, 10, 11)]
|
||||
assert spans[0].codes == ("99490",)
|
||||
assert fr_pairs(s, ("99490",)) == [(item_key, 2015, len(spans), 3)]
|
||||
|
||||
Reference in New Issue
Block a user