210 lines
6.8 KiB
Python
210 lines
6.8 KiB
Python
"""stack pfs — code elements, lineage events and derived families.
|
|
|
|
uv run stack pfs elements --code G0556 --code 99490
|
|
uv run stack pfs elements --family CCM
|
|
uv run stack pfs elements --all-payable --no-llm --dry-run
|
|
uv run stack pfs lineage --code G2058 [--write]
|
|
uv run stack pfs families [--write]
|
|
uv run stack pfs review [--code G0556]
|
|
|
|
Writes go through ``duckdb_batch`` (single-writer rule) and republish
|
|
the read-only replica so the chat and notebooks see them.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import typer
|
|
|
|
from pfs.codetables import (
|
|
ensure_tables,
|
|
read_elements,
|
|
read_events,
|
|
write_elements,
|
|
write_events,
|
|
write_families,
|
|
)
|
|
from pfs.extract import extract_code
|
|
from pfs.families import FAMILIES, derive_families, refresh_from
|
|
from pfs.lineage import lineage
|
|
|
|
app = typer.Typer(no_args_is_help=True)
|
|
|
|
|
|
# ── indirections the tests monkeypatch ───────────────────────────────
|
|
def _batch() -> Any:
|
|
from conf.connect import duckdb_batch
|
|
|
|
return duckdb_batch("aco")
|
|
|
|
|
|
def _read() -> Any:
|
|
from conf.connect import duckdb
|
|
|
|
return duckdb("aco", read_only=True)
|
|
|
|
|
|
def _store() -> Any:
|
|
from conf.connect import bib
|
|
|
|
return bib()
|
|
|
|
|
|
def _publish() -> None:
|
|
from conf.connect import publish_replica
|
|
|
|
typer.echo(f"replica → {publish_replica('aco')}")
|
|
|
|
|
|
def _classifier() -> Any:
|
|
from llm import config as llm_config
|
|
from llm.classify import closed_vocab_classifier
|
|
from llm.pool import HostPool
|
|
|
|
cfg = llm_config.load()
|
|
return closed_vocab_classifier(cfg, HostPool.from_config(cfg))
|
|
|
|
|
|
def _run_elements(
|
|
con: Any, store: Any, targets: list[str], classify: Any, *, write: bool
|
|
) -> None:
|
|
for c in targets:
|
|
x = extract_code(store, con, c, classify=classify)
|
|
if write:
|
|
write_elements(con, c, x.rows, x.reviews)
|
|
typer.echo(f"{c}: {len(x.rows)} elements, {len(x.reviews)} for review")
|
|
|
|
|
|
def _codes_for(con: Any, codes: list[str], family: str, all_payable: bool) -> list[str]:
|
|
out: list[str] = [c.upper() for c in codes]
|
|
if family:
|
|
refresh_from(con)
|
|
fam = FAMILIES.get(family.upper())
|
|
if fam is None:
|
|
raise typer.BadParameter(
|
|
f"unknown family {family!r}; known: {', '.join(FAMILIES)}"
|
|
)
|
|
out.extend(fam.codes)
|
|
if all_payable:
|
|
rows = con.execute(
|
|
"SELECT DISTINCT hcpcs FROM pfs.rvu WHERE status_code IN ('A','R','T') "
|
|
"AND year = (SELECT max(year) FROM pfs.rvu) ORDER BY hcpcs"
|
|
).fetchall()
|
|
out.extend(r[0] for r in rows)
|
|
if not out:
|
|
raise typer.BadParameter("pass --code, --family or --all-payable")
|
|
return sorted(set(out))
|
|
|
|
|
|
@app.command()
|
|
def elements(
|
|
code: list[str] = typer.Option([], "--code", help="HCPCS/CPT code (repeatable)."),
|
|
family: str = typer.Option("", help="Expand a registered family (CCM, APCM, …)."),
|
|
all_payable: bool = typer.Option(
|
|
False, "--all-payable", help="Every A/R/T code in the newest RVU year."
|
|
),
|
|
no_llm: bool = typer.Option(
|
|
False,
|
|
"--no-llm",
|
|
help="Skip the local-model classifier (unknown lines go to review).",
|
|
),
|
|
dry_run: bool = typer.Option(
|
|
False, "--dry-run", help="Extract and report; write nothing."
|
|
),
|
|
) -> None:
|
|
"""Extract typed elements for codes into pfs.code_element (+ review queue)."""
|
|
store = _store()
|
|
classify = None if no_llm else _classifier()
|
|
if dry_run:
|
|
# A preview never contends for the DuckDB single-writer lock a
|
|
# notebook may be holding (#508-#514) — read the replica, don't
|
|
# open a batch writer.
|
|
con = _read()
|
|
try:
|
|
targets = _codes_for(con, code, family, all_payable)
|
|
_run_elements(con, store, targets, classify, write=False)
|
|
finally:
|
|
con.close()
|
|
else:
|
|
with _batch() as con:
|
|
ensure_tables(con)
|
|
targets = _codes_for(con, code, family, all_payable)
|
|
_run_elements(con, store, targets, classify, write=True)
|
|
_publish()
|
|
|
|
|
|
@app.command("lineage")
|
|
def lineage_cmd(
|
|
code: str = typer.Option(..., "--code"),
|
|
write: bool = typer.Option(
|
|
False, "--write", help="Persist to pfs.code_event and republish."
|
|
),
|
|
) -> None:
|
|
"""Timeline of a code: RVU-file diffs and FR paragraphs, cross-checked."""
|
|
store = _store()
|
|
with _batch() as con:
|
|
ensure_tables(con)
|
|
events = lineage(con, store, code.upper())
|
|
for e in events:
|
|
arrow = f"{e.from_codes or '·'}→{e.to_codes or '·'}"
|
|
anchor = (
|
|
f"{e.item_key} ¶{e.p_id}" if e.item_key else f"rvu {e.note}".strip()
|
|
)
|
|
flag = "" if e.anchored else " UNANCHORED"
|
|
typer.echo(f"{e.year} {e.kind:<18}{arrow:<16}{anchor}{flag}")
|
|
if write:
|
|
write_events(con, code.upper(), events)
|
|
if write:
|
|
_publish()
|
|
|
|
|
|
@app.command()
|
|
def families(write: bool = typer.Option(False, "--write")) -> None:
|
|
"""Derive families from pfs.code_element / pfs.code_event / pfs.rvu."""
|
|
with _batch() as con:
|
|
ensure_tables(con)
|
|
codes = [
|
|
r[0]
|
|
for r in con.execute(
|
|
"SELECT DISTINCT code FROM pfs.code_element UNION SELECT DISTINCT code FROM pfs.code_event "
|
|
"UNION SELECT DISTINCT hcpcs FROM pfs.rvu WHERE year = (SELECT max(year) FROM pfs.rvu) AND status_code IN ('A','R','T')"
|
|
).fetchall()
|
|
]
|
|
elements_by_code = {c: read_elements(con, c) for c in codes}
|
|
events = {c: read_events(con, c) for c in codes}
|
|
descriptions = {
|
|
r[0]: r[1]
|
|
for r in con.execute(
|
|
"SELECT hcpcs, arg_max(description, year) FROM pfs.rvu WHERE mod IS NULL OR mod = '' GROUP BY hcpcs"
|
|
).fetchall()
|
|
}
|
|
rows = derive_families(elements_by_code, events, descriptions)
|
|
by_key: dict[str, list[str]] = {}
|
|
for r in rows:
|
|
by_key.setdefault(r.key, []).append(f"{r.code}({r.role})")
|
|
for key, members in sorted(by_key.items()):
|
|
typer.echo(f"{key}: {' '.join(members)}")
|
|
if write:
|
|
write_families(con, rows)
|
|
refresh_from(con)
|
|
if write:
|
|
_publish()
|
|
|
|
|
|
@app.command()
|
|
def review(code: str = typer.Option("", "--code")) -> None:
|
|
"""Element lines the classifier could not place."""
|
|
con = _read()
|
|
sql = (
|
|
"SELECT code, text, proposed_value, item_key, p_id FROM pfs.code_element_review"
|
|
)
|
|
params: list[Any] = []
|
|
if code:
|
|
sql += " WHERE code = ?"
|
|
params.append(code.upper())
|
|
for c, text, proposed, key, p_id in con.execute(
|
|
sql + " ORDER BY code, p_id", params
|
|
).fetchall():
|
|
typer.echo(f"{c} {key} ¶{p_id} [{proposed or '?'}] {text[:120]}")
|