feat(pfs): public-reaction series per code family — pfs.code_reaction + stack pfs reaction (refs #690)
This commit is contained in:
kert
132
src/cli/pfs.py
132
src/cli/pfs.py
@@ -6,6 +6,7 @@
|
||||
uv run stack pfs lineage --code G2058 [--write]
|
||||
uv run stack pfs families [--write]
|
||||
uv run stack pfs guidance --family CCM [--family …] [--write]
|
||||
uv run stack pfs reaction --family CCM [--code G2211] [--stance-sample 30] [--write]
|
||||
uv run stack pfs review [--code G0556]
|
||||
uv run stack pfs cpt-ingest --edition 2024 [--edition 2022 …] [--dry-run]
|
||||
uv run stack pfs cpt-ingest --all
|
||||
@@ -35,12 +36,14 @@ from pfs.codetables import (
|
||||
write_events,
|
||||
write_families,
|
||||
write_guidance,
|
||||
write_reaction,
|
||||
)
|
||||
from pfs.cpt_load import ingest as cpt_ingest
|
||||
from pfs.extract import extract_code
|
||||
from pfs.families import FAMILIES, HAND_FAMILIES, derive_families, refresh_from
|
||||
from pfs.guidance import build as build_guidance
|
||||
from pfs.lineage import lineage
|
||||
from pfs.reaction import series as reaction_series
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
@@ -92,6 +95,13 @@ def _classifier() -> Any:
|
||||
return closed_vocab_classifier(cfg, HostPool.from_config(cfg))
|
||||
|
||||
|
||||
def _engine() -> Any:
|
||||
from llm import config as llm_config
|
||||
from llm.index import _engine as pgvector_engine
|
||||
|
||||
return pgvector_engine(llm_config.load())
|
||||
|
||||
|
||||
def _run_elements(
|
||||
con: Any, store: Any, targets: list[str], classify: Any, *, write: bool
|
||||
) -> None:
|
||||
@@ -409,6 +419,128 @@ def guidance(
|
||||
con.close()
|
||||
|
||||
|
||||
def _print_reaction(rows: list[Any]) -> None:
|
||||
for r in sorted(rows, key=lambda r: (r.family, r.period_kind, r.year, r.period)):
|
||||
if r.period_kind == "docket":
|
||||
stance = ""
|
||||
if (
|
||||
r.stance_support
|
||||
or r.stance_oppose
|
||||
or r.stance_modify
|
||||
or r.stance_unclear
|
||||
):
|
||||
stance = (
|
||||
f" stance support={r.stance_support} oppose={r.stance_oppose} "
|
||||
f"modify={r.stance_modify} unclear={r.stance_unclear}"
|
||||
)
|
||||
typer.echo(
|
||||
f"{r.family} {r.period} ({r.year}) {r.n_items}/{r.n_total} commenters{stance}"
|
||||
)
|
||||
else:
|
||||
typer.echo(
|
||||
f"{r.family} {r.period} CY{r.year} "
|
||||
f"{r.n_items} comment/response pairs"
|
||||
)
|
||||
|
||||
|
||||
def _reaction_targets(
|
||||
con: Any, family: list[str], code: list[str]
|
||||
) -> list[tuple[str, tuple[str, ...]]]:
|
||||
refresh_from(con)
|
||||
targets: list[tuple[str, tuple[str, ...]]] = []
|
||||
for f in family:
|
||||
key = f.upper()
|
||||
fam = FAMILIES.get(key)
|
||||
if fam is None:
|
||||
raise typer.BadParameter(
|
||||
f"unknown family {key!r}; known: {', '.join(HAND_FAMILIES)}"
|
||||
)
|
||||
targets.append((key, fam.codes))
|
||||
for c in code:
|
||||
key = c.upper()
|
||||
targets.append((key, (key,)))
|
||||
return targets
|
||||
|
||||
|
||||
@app.command()
|
||||
def reaction(
|
||||
family: list[str] = typer.Option(
|
||||
[], "--family", help="Family key (CCM, APCM, …); repeatable."
|
||||
),
|
||||
code: list[str] = typer.Option(
|
||||
[],
|
||||
"--code",
|
||||
help="A single code as its own pseudo-family, keyed by the code; repeatable.",
|
||||
),
|
||||
stance_sample: int = typer.Option(
|
||||
0,
|
||||
"--stance-sample",
|
||||
help="Classify the newest N commenters per docket into support/oppose/"
|
||||
"modify/unclear via the self-hosted classifier (0 = skip; no stance "
|
||||
"sample means no LLM call at all).",
|
||||
),
|
||||
write: bool = typer.Option(False, "--write"),
|
||||
) -> None:
|
||||
"""Public-reaction series per code family (pfs.code_reaction): comment
|
||||
counts per docket (share of the docket's commenters mentioning the
|
||||
family), an optional stance sample, and the FR Comment:/Response:
|
||||
pair counts per rule as the pre-2017 proxy. Always recomputed live
|
||||
from pgvector/bib — there is no persisted-table fallback to read
|
||||
without --write, so --write only adds persisting the result."""
|
||||
if not family and not code:
|
||||
raise typer.BadParameter("pass --family or --code (repeatable)")
|
||||
store = _store()
|
||||
|
||||
if write:
|
||||
# Mirrors `elements --write` (I9): the stance classifier — the
|
||||
# only genuinely slow part, ~1-2s/sample — runs inside the batch
|
||||
# here too, the same tradeoff that command already makes.
|
||||
with _batch() as con:
|
||||
ensure_tables(con)
|
||||
targets = _reaction_targets(con, family, code)
|
||||
classify = _classifier() if stance_sample > 0 else None
|
||||
engine = _engine()
|
||||
rows: list[Any] = []
|
||||
for key, codes in targets:
|
||||
fam_rows = reaction_series(
|
||||
engine,
|
||||
store,
|
||||
key,
|
||||
codes,
|
||||
classify=classify,
|
||||
stance_sample_n=stance_sample,
|
||||
)
|
||||
write_reaction(con, key, fam_rows)
|
||||
rows.extend(fam_rows)
|
||||
_publish()
|
||||
_print_reaction(rows)
|
||||
return
|
||||
|
||||
# No --write: resolving family keys is the only thing that touches
|
||||
# DuckDB, and it's a plain read (I4) — never the write lock a
|
||||
# notebook may be holding (#508-#514).
|
||||
con = _read()
|
||||
try:
|
||||
targets = _reaction_targets(con, family, code)
|
||||
finally:
|
||||
con.close()
|
||||
classify = _classifier() if stance_sample > 0 else None
|
||||
engine = _engine()
|
||||
rows = []
|
||||
for key, codes in targets:
|
||||
rows.extend(
|
||||
reaction_series(
|
||||
engine,
|
||||
store,
|
||||
key,
|
||||
codes,
|
||||
classify=classify,
|
||||
stance_sample_n=stance_sample,
|
||||
)
|
||||
)
|
||||
_print_reaction(rows)
|
||||
|
||||
|
||||
@app.command()
|
||||
def review(code: str = typer.Option("", "--code")) -> None:
|
||||
"""Element lines the classifier could not place."""
|
||||
|
||||
@@ -116,6 +116,32 @@ class GuidanceRow:
|
||||
page_src: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ReactionRow:
|
||||
"""One period's public-reaction counts for a code family
|
||||
(``pfs.code_reaction``, #690) — ``period_kind`` is ``"docket"``
|
||||
(``period`` is a regulations.gov docket id, ``year`` the docket's
|
||||
``comment_end_date`` year) or ``"fr-pairs"`` (``period`` is the FR
|
||||
rule ``item_key``, ``year`` its ``rule_year_of``). ``n_items``/
|
||||
``n_total`` mean distinct commenters mentioning the family / total
|
||||
distinct commenters for a docket row, or qualifying Comment:/
|
||||
Response: paragraph pairs for an fr-pairs row. ``stance_*`` and
|
||||
``sample_json`` (``[{"item_key": …, "stance": …}]``) stay zero/
|
||||
``"[]"`` unless a stance sample was classified for that docket."""
|
||||
|
||||
family: str
|
||||
period: str
|
||||
period_kind: str # "docket" | "fr-pairs"
|
||||
year: int
|
||||
n_items: int
|
||||
n_total: int
|
||||
stance_support: int
|
||||
stance_oppose: int
|
||||
stance_modify: int
|
||||
stance_unclear: int
|
||||
sample_json: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CptSectionRow:
|
||||
edition_year: int
|
||||
@@ -217,6 +243,10 @@ CREATE TABLE IF NOT EXISTS pfs.code_family (
|
||||
CREATE TABLE IF NOT EXISTS pfs.code_guidance (
|
||||
family VARCHAR, code VARCHAR, kind VARCHAR, locator VARCHAR, item_key VARCHAR,
|
||||
item_key_src VARCHAR, p_id_src INTEGER, page_src INTEGER);
|
||||
CREATE TABLE IF NOT EXISTS pfs.code_reaction (
|
||||
family VARCHAR, period VARCHAR, period_kind VARCHAR, year INTEGER,
|
||||
n_items INTEGER, n_total INTEGER, stance_support INTEGER, stance_oppose INTEGER,
|
||||
stance_modify INTEGER, stance_unclear INTEGER, sample_json VARCHAR);
|
||||
CREATE TABLE IF NOT EXISTS pfs.cpt_section (
|
||||
edition_year INTEGER, item_key VARCHAR, sec_id VARCHAR, level INTEGER, title VARCHAR,
|
||||
path VARCHAR[], path_key VARCHAR, code_lo VARCHAR, code_hi VARCHAR, guideline VARCHAR);
|
||||
@@ -360,6 +390,21 @@ def read_guidance(con: Any, family: str) -> list[GuidanceRow]:
|
||||
return [GuidanceRow(*r) for r in rows]
|
||||
|
||||
|
||||
def write_reaction(con: Any, family: str, rows: Sequence[ReactionRow]) -> int:
|
||||
ensure_tables(con)
|
||||
con.execute("DELETE FROM pfs.code_reaction WHERE family = ?", [family])
|
||||
return _insert(con, "pfs.code_reaction", rows)
|
||||
|
||||
|
||||
def read_reaction(con: Any, family: str) -> list[ReactionRow]:
|
||||
rows = con.execute(
|
||||
"SELECT * FROM pfs.code_reaction WHERE family = ? "
|
||||
"ORDER BY period_kind, year, period",
|
||||
[family],
|
||||
).fetchall()
|
||||
return [ReactionRow(*r) for r in rows]
|
||||
|
||||
|
||||
_CPT_TABLES = (
|
||||
"pfs.cpt_section",
|
||||
"pfs.cpt_code",
|
||||
|
||||
281
src/pfs/reaction.py
Normal file
281
src/pfs/reaction.py
Normal file
@@ -0,0 +1,281 @@
|
||||
"""Public-reaction series per code family (``pfs.code_reaction``, #690,
|
||||
task 7 of the 2026-09-09 code-family-anchors plan).
|
||||
|
||||
Two period kinds, one ``ReactionRow`` shape:
|
||||
|
||||
* ``"docket"`` — one row per regulations.gov docket (2017-2026, ten
|
||||
sealed dockets, ``bib.Store.dockets()``): how many distinct
|
||||
commenters (``item_key``) mentioned the family, out of the docket's
|
||||
total commenters, straight from the ``comments`` pgvector
|
||||
collection's chunk metadata (``docket``, ``item_key``, ``codes``,
|
||||
``families``) — never from ``Docket.counts_json``, which counts raw
|
||||
comment documents, not distinct commenters who named the family.
|
||||
When ``--stance-sample N`` is passed, the newest *N* commenters' first
|
||||
matching chunk is classified into ``STANCES`` by the self-hosted
|
||||
closed-vocabulary classifier (``llm.classify``); an answer outside
|
||||
the vocabulary counts as ``"unclear"``.
|
||||
* ``"fr-pairs"`` — one row per FR rule item, back to 2001: how many
|
||||
``Comment:``/``Response:`` paragraph pairs in that rule's
|
||||
``fr_anchors`` name a family code — the pre-2017 proxy, since
|
||||
regulations.gov comment text itself isn't indexed for rules that old.
|
||||
|
||||
``series`` builds both halves for one family (or one ``--code``
|
||||
pseudo-family, a single-code family keyed by the code itself) and
|
||||
returns them as one list, dockets first (docket id order) then FR pairs
|
||||
(rule item_key order); the CLI (``stack pfs reaction``) prints and
|
||||
optionally writes them.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from collections import Counter
|
||||
from typing import Any, Callable, Sequence
|
||||
|
||||
from sqlalchemy import text
|
||||
|
||||
from pfs.descriptors import rule_year_of
|
||||
from pfs.guidance import _code_pattern, _codes_in
|
||||
|
||||
#: Closed vocabulary the stance classifier answers from — anything else
|
||||
#: (a refusal, a hedge, garbage) counts as "unclear" (Resolutions).
|
||||
STANCES = ("support", "oppose", "modify", "unclear")
|
||||
|
||||
#: One docket's distinct-commenter count, with vs. without the family —
|
||||
#: matches by literal code presence (``codes`` metadata) OR family
|
||||
#: membership (``families`` metadata), the same two-way match
|
||||
#: ``llm.evidence``'s code-cited/family-cited sources use, so a chunk
|
||||
#: the indexer only stamped via family membership (no literal code
|
||||
#: string in that particular chunk) still counts.
|
||||
_DOCKET_COUNTS_SQL = text(
|
||||
"SELECT docket, "
|
||||
"count(DISTINCT item_key) FILTER (WHERE matched) AS n_items, "
|
||||
"count(DISTINCT item_key) AS n_total "
|
||||
"FROM ( "
|
||||
"SELECT e.cmetadata->>'docket' AS docket, e.cmetadata->>'item_key' AS item_key, "
|
||||
"(string_to_array(COALESCE(e.cmetadata->>'codes', ''), ' ') && CAST(:codes AS text[])) "
|
||||
"OR (string_to_array(COALESCE(e.cmetadata->>'families', ''), ' ') && ARRAY[:family_key]) "
|
||||
"AS matched "
|
||||
"FROM langchain_pg_embedding e "
|
||||
"WHERE e.collection_id = (SELECT uuid FROM langchain_pg_collection WHERE name = :collection) "
|
||||
") t "
|
||||
"WHERE docket IS NOT NULL AND docket <> '' "
|
||||
"GROUP BY docket ORDER BY docket"
|
||||
)
|
||||
|
||||
#: The newest N distinct items' first matching chunk in one docket — the
|
||||
#: same one-chunk-per-item windowing ``llm.evidence``'s family-cited SQL
|
||||
#: uses (stage one: lowest ``seq`` per item_key; stage two: newest
|
||||
#: ``date`` first), text truncated to 1,500 chars for the classifier
|
||||
#: (Resolutions).
|
||||
_STANCE_SAMPLE_SQL = text(
|
||||
"SELECT item_key, LEFT(document, 1500) FROM ( "
|
||||
"SELECT e.cmetadata->>'item_key' AS item_key, e.document AS document, "
|
||||
"e.cmetadata->>'date' AS date, "
|
||||
"row_number() OVER ( "
|
||||
"PARTITION BY e.cmetadata->>'item_key' "
|
||||
"ORDER BY COALESCE(NULLIF(e.cmetadata->>'seq', '')::int, 0) ASC "
|
||||
") AS item_rn "
|
||||
"FROM langchain_pg_embedding e "
|
||||
"WHERE e.collection_id = (SELECT uuid FROM langchain_pg_collection WHERE name = :collection) "
|
||||
"AND e.cmetadata->>'docket' = :docket "
|
||||
"AND ( "
|
||||
"(string_to_array(COALESCE(e.cmetadata->>'codes', ''), ' ') && CAST(:codes AS text[])) "
|
||||
"OR (string_to_array(COALESCE(e.cmetadata->>'families', ''), ' ') && ARRAY[:family_key]) "
|
||||
") "
|
||||
") one_per_item "
|
||||
"WHERE item_rn = 1 "
|
||||
"ORDER BY date DESC NULLS LAST "
|
||||
"LIMIT :n"
|
||||
)
|
||||
|
||||
_COMMENTS_COLLECTION = "comments"
|
||||
|
||||
|
||||
def docket_counts(
|
||||
engine: Any, family_key: str, codes: Sequence[str]
|
||||
) -> list[tuple[str, int, int]]:
|
||||
"""``(docket, n_items_with_family, n_items_total)`` for every docket
|
||||
that has any chunks in the ``comments`` collection — one round trip,
|
||||
``GROUP BY docket``."""
|
||||
upper = [c.upper() for c in codes]
|
||||
with engine.begin() as conn:
|
||||
rows = conn.execute(
|
||||
_DOCKET_COUNTS_SQL,
|
||||
{
|
||||
"codes": upper,
|
||||
"family_key": family_key.upper(),
|
||||
"collection": _COMMENTS_COLLECTION,
|
||||
},
|
||||
).fetchall()
|
||||
return [
|
||||
(docket, int(n_items or 0), int(n_total or 0))
|
||||
for docket, n_items, n_total in rows
|
||||
]
|
||||
|
||||
|
||||
def stance_sample(
|
||||
engine: Any, family_key: str, codes: Sequence[str], docket: str, *, n: int
|
||||
) -> list[tuple[str, str]]:
|
||||
"""``(item_key, text)`` for the newest *n* distinct commenters in
|
||||
*docket* who mentioned the family, first matching chunk, newest
|
||||
first, text capped to 1,500 chars."""
|
||||
if n <= 0:
|
||||
return []
|
||||
upper = [c.upper() for c in codes]
|
||||
with engine.begin() as conn:
|
||||
rows = conn.execute(
|
||||
_STANCE_SAMPLE_SQL,
|
||||
{
|
||||
"codes": upper,
|
||||
"family_key": family_key.upper(),
|
||||
"collection": _COMMENTS_COLLECTION,
|
||||
"docket": docket,
|
||||
"n": n,
|
||||
},
|
||||
).fetchall()
|
||||
return [(item_key, text or "") for item_key, text in rows]
|
||||
|
||||
|
||||
def _prompt_text(family_key: str, codes: Sequence[str], chunk: str) -> str:
|
||||
return (
|
||||
f"Comment on Medicare payment for {family_key} "
|
||||
f"(codes {' '.join(codes)}):\n{chunk}"
|
||||
)
|
||||
|
||||
|
||||
def _stance_of(text: str, classify: Callable[[str, Sequence[str]], str | None]) -> str:
|
||||
stance = classify(text, STANCES)
|
||||
return stance if stance in STANCES else "unclear"
|
||||
|
||||
|
||||
def classify_stances(
|
||||
sample: Sequence[tuple[str, str]],
|
||||
classify: Callable[[str, Sequence[str]], str | None],
|
||||
) -> Counter:
|
||||
"""Tally ``STANCES`` for a stance sample — an answer outside the
|
||||
closed vocabulary counts as ``"unclear"`` (Resolutions)."""
|
||||
return Counter(_stance_of(text, classify) for _, text in sample)
|
||||
|
||||
|
||||
#: "Comment:" / "Response:" paragraph openers — a docket-comment excerpt
|
||||
#: reproduced verbatim inside an FR rule's own preamble, always at the
|
||||
#: very start of the paragraph.
|
||||
_COMMENT_RE = re.compile(r"^\s*Comment\s*:", re.I)
|
||||
_RESPONSE_RE = re.compile(r"^\s*Response\s*:", re.I)
|
||||
|
||||
|
||||
def fr_pairs(store: Any, codes: Sequence[str]) -> list[tuple[str, int, int, int]]:
|
||||
"""``(item_key, rule_year, n_comment_paras, n_response_paras)`` per
|
||||
rule item: a ``Comment:`` paragraph immediately followed (next
|
||||
``p_id`` in that item) by a ``Response:`` paragraph, where either
|
||||
paragraph names one of *codes* — the pre-2017 proxy for public
|
||||
reaction, since regulations.gov comment text isn't indexed that far
|
||||
back. Rules with zero qualifying pairs are omitted."""
|
||||
if not codes:
|
||||
return []
|
||||
con = store._con() # noqa: SLF001 — same pattern as pfs.descriptors/pfs.guidance
|
||||
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 "
|
||||
"ORDER BY a.item_key, a.p_id"
|
||||
).fetchall()
|
||||
|
||||
pattern = _code_pattern(codes)
|
||||
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))
|
||||
meta[item_key] = (title, date_published)
|
||||
|
||||
out: list[tuple[str, int, int, int]] = []
|
||||
for item_key, paras in by_item.items():
|
||||
n_comment = 0
|
||||
n_response = 0
|
||||
for i, (_p_id, para_text) in enumerate(paras):
|
||||
if not _COMMENT_RE.match(para_text or ""):
|
||||
continue
|
||||
if i + 1 >= len(paras):
|
||||
continue
|
||||
_next_p_id, next_text = paras[i + 1]
|
||||
if not _RESPONSE_RE.match(next_text or ""):
|
||||
continue
|
||||
combined = f"{para_text} {next_text}"
|
||||
if not _codes_in(combined, pattern):
|
||||
continue
|
||||
n_comment += 1
|
||||
n_response += 1
|
||||
if n_comment:
|
||||
title, date_published = meta[item_key]
|
||||
out.append(
|
||||
(item_key, rule_year_of(title, date_published), n_comment, n_response)
|
||||
)
|
||||
return sorted(out, key=lambda r: (r[1], r[0]))
|
||||
|
||||
|
||||
def series(
|
||||
engine: Any,
|
||||
store: Any,
|
||||
family: str,
|
||||
codes: Sequence[str],
|
||||
*,
|
||||
classify: Callable[[str, Sequence[str]], str | None] | None = None,
|
||||
stance_sample_n: int = 0,
|
||||
) -> list[Any]:
|
||||
"""Every ``ReactionRow`` for one family (or ``--code`` pseudo-family):
|
||||
a docket row per docket with any comment chunks, then an fr-pairs row
|
||||
per rule with qualifying Comment/Response pairs. Stance columns stay
|
||||
zero (and ``sample_json`` ``"[]"``) unless *classify* is given and
|
||||
*stance_sample_n* > 0."""
|
||||
from pfs.codetables import ReactionRow
|
||||
|
||||
docket_years = {d.id: d.comment_end_date[:4] for d in store.dockets()}
|
||||
|
||||
rows: list[ReactionRow] = []
|
||||
for docket, n_items, n_total in docket_counts(engine, family, codes):
|
||||
year_str = docket_years.get(docket, "")
|
||||
year = int(year_str) if year_str.isdigit() else 0
|
||||
stance_counts: Counter = Counter()
|
||||
sample_json = "[]"
|
||||
if classify is not None and stance_sample_n > 0:
|
||||
sample = stance_sample(engine, family, codes, docket, n=stance_sample_n)
|
||||
stances = [
|
||||
(item_key, _stance_of(_prompt_text(family, codes, chunk), classify))
|
||||
for item_key, chunk in sample
|
||||
]
|
||||
stance_counts = Counter(s for _, s in stances)
|
||||
sample_json = json.dumps([{"item_key": k, "stance": s} for k, s in stances])
|
||||
rows.append(
|
||||
ReactionRow(
|
||||
family=family,
|
||||
period=docket,
|
||||
period_kind="docket",
|
||||
year=year,
|
||||
n_items=n_items,
|
||||
n_total=n_total,
|
||||
stance_support=stance_counts.get("support", 0),
|
||||
stance_oppose=stance_counts.get("oppose", 0),
|
||||
stance_modify=stance_counts.get("modify", 0),
|
||||
stance_unclear=stance_counts.get("unclear", 0),
|
||||
sample_json=sample_json,
|
||||
)
|
||||
)
|
||||
|
||||
for item_key, rule_year, n_comment, n_response in fr_pairs(store, codes):
|
||||
rows.append(
|
||||
ReactionRow(
|
||||
family=family,
|
||||
period=item_key,
|
||||
period_kind="fr-pairs",
|
||||
year=rule_year,
|
||||
n_items=n_comment,
|
||||
n_total=n_response,
|
||||
stance_support=0,
|
||||
stance_oppose=0,
|
||||
stance_modify=0,
|
||||
stance_unclear=0,
|
||||
sample_json="[]",
|
||||
)
|
||||
)
|
||||
return rows
|
||||
@@ -1,4 +1,4 @@
|
||||
"""stack pfs — elements / lineage / families / guidance / review."""
|
||||
"""stack pfs — elements / lineage / families / guidance / reaction / review."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -12,11 +12,13 @@ from pfs.codetables import (
|
||||
ElementRow,
|
||||
EventRow,
|
||||
GuidanceRow,
|
||||
ReactionRow,
|
||||
ensure_tables,
|
||||
read_elements,
|
||||
read_events,
|
||||
read_families,
|
||||
read_guidance,
|
||||
read_reaction,
|
||||
write_cpt_edition,
|
||||
write_elements,
|
||||
write_events,
|
||||
@@ -621,6 +623,108 @@ class TestGuidance:
|
||||
assert res.exit_code != 0
|
||||
|
||||
|
||||
class TestReaction:
|
||||
def _row(
|
||||
self, family="CCM", period="CMS-2023-0121", period_kind="docket", year=2023
|
||||
):
|
||||
return ReactionRow(family, period, period_kind, year, 5, 20, 1, 0, 0, 1, "[]")
|
||||
|
||||
def test_write_builds_writes_and_publishes(self, con, monkeypatch):
|
||||
seen = {}
|
||||
|
||||
def fake_series(
|
||||
engine, store_, family, codes, *, classify=None, stance_sample_n=0
|
||||
):
|
||||
seen["engine"] = engine
|
||||
seen["store"] = store_
|
||||
seen["family"] = family
|
||||
seen["codes"] = codes
|
||||
seen["classify"] = classify
|
||||
seen["n"] = stance_sample_n
|
||||
return [self._row(family=family)]
|
||||
|
||||
monkeypatch.setattr(pfs_cli, "reaction_series", fake_series)
|
||||
monkeypatch.setattr(pfs_cli, "_engine", lambda: "ENGINE")
|
||||
res = runner.invoke(app, ["pfs", "reaction", "--family", "ccm", "--write"])
|
||||
assert res.exit_code == 0, res.output
|
||||
assert seen["family"] == "CCM"
|
||||
assert seen["engine"] == "ENGINE"
|
||||
assert seen["classify"] is None # no --stance-sample: no LLM at all
|
||||
assert seen["n"] == 0
|
||||
assert set(seen["codes"]) == set(pfs_cli.HAND_FAMILIES["CCM"].codes)
|
||||
got = read_reaction(con, "CCM")
|
||||
assert [r.period for r in got] == ["CMS-2023-0121"]
|
||||
assert con.published == [True]
|
||||
assert "CCM CMS-2023-0121 (2023) 5/20 commenters" in res.output
|
||||
|
||||
def test_code_makes_a_single_code_pseudo_family(self, con, monkeypatch):
|
||||
seen = {}
|
||||
|
||||
def fake_series(
|
||||
engine, store_, family, codes, *, classify=None, stance_sample_n=0
|
||||
):
|
||||
seen["family"] = family
|
||||
seen["codes"] = codes
|
||||
return [self._row(family=family)]
|
||||
|
||||
monkeypatch.setattr(pfs_cli, "reaction_series", fake_series)
|
||||
monkeypatch.setattr(pfs_cli, "_engine", lambda: "ENGINE")
|
||||
res = runner.invoke(app, ["pfs", "reaction", "--code", "g2211", "--write"])
|
||||
assert res.exit_code == 0, res.output
|
||||
assert seen["family"] == "G2211"
|
||||
assert seen["codes"] == ("G2211",)
|
||||
assert [r.family for r in read_reaction(con, "G2211")] == ["G2211"]
|
||||
|
||||
def test_stance_sample_builds_the_classifier(self, con, monkeypatch):
|
||||
seen = {}
|
||||
monkeypatch.setattr(
|
||||
pfs_cli,
|
||||
"reaction_series",
|
||||
lambda engine, store_, family, codes, *, classify=None, stance_sample_n=0: (
|
||||
seen.update(classify=classify, n=stance_sample_n) or []
|
||||
),
|
||||
)
|
||||
monkeypatch.setattr(pfs_cli, "_engine", lambda: "ENGINE")
|
||||
monkeypatch.setattr(pfs_cli, "_classifier", lambda: "CLASSIFIER")
|
||||
res = runner.invoke(
|
||||
app, ["pfs", "reaction", "--family", "CCM", "--stance-sample", "5"]
|
||||
)
|
||||
assert res.exit_code == 0, res.output
|
||||
assert seen["classify"] == "CLASSIFIER"
|
||||
assert seen["n"] == 5
|
||||
|
||||
def test_without_write_never_opens_a_batch_connection(self, con, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
pfs_cli,
|
||||
"reaction_series",
|
||||
lambda engine, store_, family, codes, *, classify=None, stance_sample_n=0: [
|
||||
self._row(family=family)
|
||||
],
|
||||
)
|
||||
monkeypatch.setattr(pfs_cli, "_engine", lambda: "ENGINE")
|
||||
|
||||
def fail_batch():
|
||||
raise AssertionError("must not open a RW duckdb_batch connection")
|
||||
|
||||
monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
|
||||
res = runner.invoke(app, ["pfs", "reaction", "--family", "CCM"])
|
||||
assert res.exit_code == 0, res.output
|
||||
assert "CCM CMS-2023-0121 (2023) 5/20 commenters" in res.output
|
||||
assert con.published == [] # nothing persisted, batch never opened
|
||||
|
||||
def test_unknown_family_error_lists_hand_families_only(self, con, monkeypatch):
|
||||
pfs_cli.FAMILIES["ZZZZ-99999"] = pfs_cli.FAMILIES["CCM"]
|
||||
res = runner.invoke(app, ["pfs", "reaction", "--family", "NOPE", "--write"])
|
||||
assert res.exit_code != 0
|
||||
assert "ZZZZ-99999" not in res.output
|
||||
for key in pfs_cli.HAND_FAMILIES:
|
||||
assert key in res.output
|
||||
|
||||
def test_requires_family_or_code(self, con):
|
||||
res = runner.invoke(app, ["pfs", "reaction"])
|
||||
assert res.exit_code != 0
|
||||
|
||||
|
||||
class TestReview:
|
||||
def test_lists_queue(self, con):
|
||||
from pfs.codetables import ReviewRow
|
||||
|
||||
285
tests/pfs/test_reaction.py
Normal file
285
tests/pfs/test_reaction.py
Normal file
@@ -0,0 +1,285 @@
|
||||
"""pfs.reaction — public-reaction series per code family (#690, task 7)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import duckdb
|
||||
import pytest
|
||||
|
||||
from bib.dockets import Docket
|
||||
from bib.item import Rule
|
||||
from bib.store import Store
|
||||
from pfs.codetables import ReactionRow, ensure_tables, read_reaction, write_reaction
|
||||
from pfs.reaction import (
|
||||
STANCES,
|
||||
classify_stances,
|
||||
docket_counts,
|
||||
fr_pairs,
|
||||
series,
|
||||
stance_sample,
|
||||
)
|
||||
|
||||
# ── fake pgvector engine ──────────────────────────────────────────────
|
||||
|
||||
|
||||
def _engine(rows):
|
||||
engine = MagicMock()
|
||||
conn = engine.begin.return_value.__enter__.return_value
|
||||
conn.execute.return_value.fetchall.return_value = rows
|
||||
return engine, conn
|
||||
|
||||
|
||||
def _multi_engine(docket_rows=None, stance_rows=None):
|
||||
"""A fake engine whose rows depend on which of the two queries is
|
||||
running — distinguished by the ``docket``/``n`` bind params that
|
||||
only ``_STANCE_SAMPLE_SQL`` carries."""
|
||||
engine = MagicMock()
|
||||
conn = engine.begin.return_value.__enter__.return_value
|
||||
|
||||
def _execute(_sql, params):
|
||||
result = MagicMock()
|
||||
if "docket" in params:
|
||||
result.fetchall.return_value = stance_rows or []
|
||||
else:
|
||||
result.fetchall.return_value = docket_rows or []
|
||||
return result
|
||||
|
||||
conn.execute.side_effect = _execute
|
||||
return engine, conn
|
||||
|
||||
|
||||
class TestDocketCounts:
|
||||
def test_maps_rows_and_binds_params(self):
|
||||
engine, conn = _engine([("CMS-2023-0121", 40, 200), ("CMS-2024-0453", 3, 150)])
|
||||
out = docket_counts(engine, "CCM", ("99490", "99439"))
|
||||
assert out == [("CMS-2023-0121", 40, 200), ("CMS-2024-0453", 3, 150)]
|
||||
params = conn.execute.call_args.args[1]
|
||||
assert params["codes"] == ["99490", "99439"]
|
||||
assert params["family_key"] == "CCM"
|
||||
assert params["collection"] == "comments"
|
||||
|
||||
def test_null_counts_become_zero(self):
|
||||
engine, _conn = _engine([("CMS-2023-0121", None, 0)])
|
||||
assert docket_counts(engine, "CCM", ()) == [("CMS-2023-0121", 0, 0)]
|
||||
|
||||
|
||||
class TestStanceSample:
|
||||
def test_maps_rows_and_binds_params(self):
|
||||
engine, conn = _engine([("C1", "chunk one"), ("C2", "chunk two")])
|
||||
out = stance_sample(engine, "CCM", ("99490",), "CMS-2023-0121", n=5)
|
||||
assert out == [("C1", "chunk one"), ("C2", "chunk two")]
|
||||
params = conn.execute.call_args.args[1]
|
||||
assert params["docket"] == "CMS-2023-0121"
|
||||
assert params["n"] == 5
|
||||
assert params["family_key"] == "CCM"
|
||||
|
||||
def test_n_zero_short_circuits_without_a_query(self):
|
||||
engine, conn = _engine([("C1", "x")])
|
||||
assert stance_sample(engine, "CCM", ("99490",), "CMS-2023-0121", n=0) == []
|
||||
conn.execute.assert_not_called()
|
||||
|
||||
|
||||
class TestClassifyStances:
|
||||
def test_tallies_known_slugs(self):
|
||||
sample = [("C1", "t1"), ("C2", "t2"), ("C3", "t3")]
|
||||
answers = {"t1": "support", "t2": "oppose", "t3": "support"}
|
||||
counts = classify_stances(sample, lambda text, choices: answers[text])
|
||||
assert counts["support"] == 2 and counts["oppose"] == 1
|
||||
|
||||
def test_out_of_vocabulary_answer_counts_as_unclear(self):
|
||||
sample = [("C1", "t1")]
|
||||
counts = classify_stances(sample, lambda text, choices: "maybe-ish")
|
||||
assert counts["unclear"] == 1
|
||||
|
||||
def test_none_answer_counts_as_unclear(self):
|
||||
sample = [("C1", "t1")]
|
||||
counts = classify_stances(sample, lambda text, choices: None)
|
||||
assert counts["unclear"] == 1
|
||||
|
||||
def test_every_stance_is_a_known_slug(self):
|
||||
assert set(STANCES) == {"support", "oppose", "modify", "unclear"}
|
||||
|
||||
|
||||
# ── fr_pairs ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def store_with_pairs():
|
||||
s = Store(":memory:")
|
||||
item_key = s.create(Rule(title="CY2015 PFS final rule"))
|
||||
con = s._con() # noqa: SLF001
|
||||
con.executemany(
|
||||
"INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)",
|
||||
[
|
||||
(
|
||||
item_key,
|
||||
10,
|
||||
100,
|
||||
1,
|
||||
"Comment: Commenters asked why CPT code 99490 (chronic care "
|
||||
"management) is valued as proposed.",
|
||||
),
|
||||
(
|
||||
item_key,
|
||||
11,
|
||||
100,
|
||||
2,
|
||||
"Response: We are finalizing the RVUs for 99490 as proposed.",
|
||||
),
|
||||
(item_key, 12, 101, 3, "Comment: A commenter raised an unrelated point."),
|
||||
(item_key, 13, 101, 4, "Response: We disagree with the commenter."),
|
||||
# a Comment: with no immediately-following Response: (next para
|
||||
# is prose) must never be counted.
|
||||
(item_key, 14, 102, 5, "Comment: Another commenter mentioned 99490."),
|
||||
(item_key, 15, 102, 6, "We did not adopt this suggestion."),
|
||||
],
|
||||
)
|
||||
con.commit()
|
||||
yield s, item_key
|
||||
s.close()
|
||||
|
||||
|
||||
class TestFrPairs:
|
||||
def test_counts_only_pairs_naming_a_code(self, store_with_pairs):
|
||||
s, item_key = store_with_pairs
|
||||
out = fr_pairs(s, ("99490",))
|
||||
assert out == [(item_key, 2015, 1, 1)]
|
||||
|
||||
def test_no_codes_returns_empty(self, store_with_pairs):
|
||||
s, _item_key = store_with_pairs
|
||||
assert fr_pairs(s, ()) == []
|
||||
|
||||
def test_unmatched_code_yields_no_rows(self, store_with_pairs):
|
||||
s, _item_key = store_with_pairs
|
||||
assert fr_pairs(s, ("99437",)) == []
|
||||
|
||||
def test_dangling_comment_without_response_is_not_counted(self, store_with_pairs):
|
||||
# ¶14's "Comment:" is followed by ¶15, which is not a "Response:"
|
||||
# paragraph — it must not inflate the count even though it names
|
||||
# 99490.
|
||||
s, item_key = store_with_pairs
|
||||
out = fr_pairs(s, ("99490",))
|
||||
assert out == [(item_key, 2015, 1, 1)]
|
||||
|
||||
|
||||
# ── series ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestSeries:
|
||||
@pytest.fixture
|
||||
def store(self, store_with_pairs):
|
||||
s, item_key = store_with_pairs
|
||||
s.docket_upsert(
|
||||
Docket(id="CMS-2023-0121", rule_cms_id="R1", comment_end_date="2023-06-01")
|
||||
)
|
||||
return s, item_key
|
||||
|
||||
def test_docket_and_fr_pairs_rows_without_stance(self, store):
|
||||
s, item_key = store
|
||||
engine, _conn = _multi_engine(docket_rows=[("CMS-2023-0121", 5, 20)])
|
||||
rows = series(engine, s, "CCM", ("99490",))
|
||||
by_kind = {r.period_kind: r for r in rows}
|
||||
docket_row = by_kind["docket"]
|
||||
assert (
|
||||
docket_row.period,
|
||||
docket_row.year,
|
||||
docket_row.n_items,
|
||||
docket_row.n_total,
|
||||
) == (
|
||||
"CMS-2023-0121",
|
||||
2023,
|
||||
5,
|
||||
20,
|
||||
)
|
||||
assert docket_row.stance_support == 0 and docket_row.sample_json == "[]"
|
||||
fr_row = by_kind["fr-pairs"]
|
||||
assert (fr_row.period, fr_row.year, fr_row.n_items, fr_row.n_total) == (
|
||||
item_key,
|
||||
2015,
|
||||
1,
|
||||
1,
|
||||
)
|
||||
|
||||
def test_stance_sample_classifies_and_fills_sample_json(self, store):
|
||||
s, _item_key = store
|
||||
engine, _conn = _multi_engine(
|
||||
docket_rows=[("CMS-2023-0121", 5, 20)],
|
||||
stance_rows=[("C1", "we support this"), ("C2", "we oppose this")],
|
||||
)
|
||||
|
||||
def fake_classify(text, choices):
|
||||
return "support" if "support" in text else "oppose"
|
||||
|
||||
rows = series(
|
||||
engine, s, "CCM", ("99490",), classify=fake_classify, stance_sample_n=2
|
||||
)
|
||||
docket_row = next(r for r in rows if r.period_kind == "docket")
|
||||
assert docket_row.stance_support == 1 and docket_row.stance_oppose == 1
|
||||
sample = json.loads(docket_row.sample_json)
|
||||
assert {s_["item_key"] for s_ in sample} == {"C1", "C2"}
|
||||
assert {s_["stance"] for s_ in sample} == {"support", "oppose"}
|
||||
|
||||
def test_zero_stance_sample_skips_classification(self, store):
|
||||
s, _item_key = store
|
||||
engine, conn = _multi_engine(docket_rows=[("CMS-2023-0121", 5, 20)])
|
||||
rows = series(
|
||||
engine,
|
||||
s,
|
||||
"CCM",
|
||||
("99490",),
|
||||
classify=lambda t, c: "support",
|
||||
stance_sample_n=0,
|
||||
)
|
||||
docket_row = next(r for r in rows if r.period_kind == "docket")
|
||||
assert docket_row.stance_support == 0
|
||||
# only the docket_counts query ran — no stance_sample round trip
|
||||
assert conn.execute.call_count == 1
|
||||
|
||||
def test_code_pseudo_family(self, store):
|
||||
s, _item_key = store
|
||||
engine, conn = _multi_engine(docket_rows=[("CMS-2023-0121", 2, 20)])
|
||||
rows = series(engine, s, "G2211", ("G2211",))
|
||||
assert [r.family for r in rows if r.period_kind == "docket"] == ["G2211"]
|
||||
params = conn.execute.call_args.args[1]
|
||||
assert params["codes"] == ["G2211"]
|
||||
|
||||
|
||||
# ── DDL / write / read round trip ────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def con():
|
||||
c = duckdb.connect(":memory:")
|
||||
ensure_tables(c)
|
||||
yield c
|
||||
c.close()
|
||||
|
||||
|
||||
class TestReactionTable:
|
||||
def _row(
|
||||
self, family="CCM", period="CMS-2023-0121", period_kind="docket", year=2023
|
||||
):
|
||||
return ReactionRow(family, period, period_kind, year, 5, 20, 1, 0, 0, 1, "[]")
|
||||
|
||||
def test_write_then_read_round_trips(self, con):
|
||||
rows = [self._row(), self._row(period="R1", period_kind="fr-pairs", year=2015)]
|
||||
n = write_reaction(con, "CCM", rows)
|
||||
assert n == 2
|
||||
got = read_reaction(con, "CCM")
|
||||
assert {(r.period, r.period_kind) for r in got} == {
|
||||
("CMS-2023-0121", "docket"),
|
||||
("R1", "fr-pairs"),
|
||||
}
|
||||
|
||||
def test_write_is_delete_then_insert_per_family(self, con):
|
||||
write_reaction(con, "CCM", [self._row(period="CMS-2023-0121")])
|
||||
write_reaction(con, "CCM", [self._row(period="CMS-2024-0453")])
|
||||
got = read_reaction(con, "CCM")
|
||||
assert [r.period for r in got] == ["CMS-2024-0453"]
|
||||
|
||||
def test_other_family_untouched(self, con):
|
||||
write_reaction(con, "G2211", [self._row(family="G2211")])
|
||||
write_reaction(con, "CCM", [])
|
||||
assert len(read_reaction(con, "G2211")) == 1
|
||||
Reference in New Issue
Block a user