78a56f593e
Section 0 states the two organizing principles in order (CPT hierarchy first, FR second) with the replica's CPT edition years. Section 1 gains "Where the manual files this code": breadcrumb from cpt_section.path_key, symbols, the manual's own stem/elements/tail, and its parenthetical instructions — a HCPCS pick (G-codes) renders a "not in the CPT book" note instead of erroring. Section 3 explains CPT Changes years (AMA's own revision years) versus FR mentioning-rule dating. Section 4 shows a family's CPT heading, sibling headings under the same parent, and summary counts (total/multi-code/CPT-named families by top-level chapter). Section 8 adds pfs.cpt_* row counts per edition and the edition bib items with titles. Claude-Session: https://claude.ai/code/session_01Aum3pEMAM3yQVdFSdVe6Gc
704 lines
29 KiB
Python
704 lines
29 KiB
Python
import marimo
|
|
|
|
__generated_with = "0.23.13"
|
|
app = marimo.App(width="medium")
|
|
|
|
|
|
@app.cell(hide_code=True)
|
|
def _():
|
|
import marimo as mo
|
|
|
|
return (mo,)
|
|
|
|
|
|
@app.cell(hide_code=True)
|
|
def _(mo):
|
|
mo.md(
|
|
"""
|
|
# Code families as first-class objects
|
|
|
|
A physician fee schedule code is not a number — it is a **bundle of logical elements**:
|
|
who furnishes the service, for how long, per what period, to which patients, doing which
|
|
activities, by which modality. This notebook walks through how the stack turns that idea
|
|
into tables you can query and cite: elements → extraction → lineage → families → anchors
|
|
→ guidance → public reaction. Every number on this page is read live from the replica and
|
|
the bibliography; every claim links to the Federal Register paragraph it came from.
|
|
"""
|
|
)
|
|
return
|
|
|
|
|
|
@app.cell(hide_code=True)
|
|
def _():
|
|
# ── Setup ──
|
|
import altair as alt
|
|
import polars as pl
|
|
|
|
from conf import connect, path
|
|
from conf.display import plain_years
|
|
|
|
connect.theme()
|
|
NOTES = {}
|
|
|
|
# data/replica/<name>.ro.duckdb — same layout connect._replica_path resolves,
|
|
# computed here (not imported) because it's a private helper.
|
|
_primary = path("db.aco")
|
|
REPLICA_PATH = _primary.parent / "replica" / f"{_primary.stem}.ro.duckdb"
|
|
|
|
def _open_replica():
|
|
try:
|
|
return connect.duckdb("aco", read_only=True)
|
|
except Exception as e: # noqa: BLE001 — degrade, never crash the page
|
|
NOTES["replica"] = f"replica unavailable: {e}"
|
|
return None
|
|
|
|
def _open_bib():
|
|
try:
|
|
return connect.bib()
|
|
except Exception as e: # noqa: BLE001
|
|
NOTES["bib"] = f"bibliography unavailable: {e}"
|
|
return None
|
|
|
|
con = _open_replica()
|
|
store = _open_bib()
|
|
|
|
def q(sql, params=()):
|
|
if con is None:
|
|
return pl.DataFrame()
|
|
try:
|
|
return con.execute(sql, list(params)).pl()
|
|
except Exception as e: # noqa: BLE001 — a missing table is a "not built yet"
|
|
NOTES[sql[:40]] = str(e)
|
|
return pl.DataFrame()
|
|
|
|
def fr_md(item_key, p_id):
|
|
if store is None:
|
|
return f"{item_key} ¶{p_id}"
|
|
try:
|
|
from bib.frlink import md_link
|
|
|
|
return md_link(
|
|
f"p-{p_id}", store=store, item_key=item_key, text=f"{item_key} ¶{p_id}"
|
|
)
|
|
except Exception: # noqa: BLE001
|
|
return f"{item_key} ¶{p_id}"
|
|
|
|
def not_built(cmd):
|
|
return f"_Not built yet — run `{cmd}` and republish the replica._"
|
|
|
|
return NOTES, REPLICA_PATH, alt, con, fr_md, not_built, pl, plain_years, q, store
|
|
|
|
|
|
@app.cell(hide_code=True)
|
|
def _(fr_md, mo, store):
|
|
# ── 0. Why a code is a bundle of elements ──
|
|
_steps = []
|
|
if store is not None:
|
|
try:
|
|
_con = store._con() # noqa: SLF001
|
|
for _p in (394, 396, 398):
|
|
_row = _con.execute(
|
|
"SELECT text FROM fr_anchors WHERE item_key = ? AND p_id = ?",
|
|
("2KVJ2HKX", _p),
|
|
).fetchone()
|
|
if _row:
|
|
_steps.append(f"> {_row[0][:400]}… — {fr_md('2KVJ2HKX', _p)}")
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
mo.md(
|
|
"## 0. Why a code is a bundle of elements\n\n"
|
|
"CMS says so itself. When it decides whether a service can be furnished by telehealth it "
|
|
"walks three steps, and the third is literally *review the elements of the service as "
|
|
"described by the HCPCS code* (CY2026 proposed rule, 90 FR 32389):\n\n"
|
|
+ (
|
|
"\n\n".join(_steps)
|
|
if _steps
|
|
else "_(bibliography unavailable — quotes omitted)_"
|
|
)
|
|
+ "\n\nThat one sentence is the whole design: a code is not an opaque five-character "
|
|
"string CMS prices as a unit. It is a bundle of *who* furnishes it, *how long* it takes, "
|
|
"*how often* it can be billed, *which patients* qualify, *which activities* it covers, "
|
|
"and *by what modality* — and CMS itself reasons about codes at that level of detail, one "
|
|
"element at a time. Everything below builds machine-readable tables out of that same "
|
|
"bundle: parse the descriptor into typed elements, extract them with their FR anchor, "
|
|
"trace how a code's identity changes over the years (lineage), and group codes that share "
|
|
"one clinical program into a family."
|
|
)
|
|
return
|
|
|
|
|
|
@app.cell(hide_code=True)
|
|
def _(mo, not_built, q):
|
|
# Two organizing principles behind every table on this page
|
|
_editions = q(
|
|
"SELECT DISTINCT edition_year FROM pfs.cpt_code ORDER BY edition_year"
|
|
)
|
|
if _editions.is_empty():
|
|
_view = mo.md(not_built("stack pfs cpt-ingest --all"))
|
|
else:
|
|
_years = ", ".join(str(y) for y in _editions["edition_year"].to_list())
|
|
_view = mo.md(
|
|
"**Two organizing principles, in that order.** The CPT codebook is the **first**: "
|
|
"the American Medical Association's own hierarchy — Section → subsection → category "
|
|
"→ subcategory → codes, with guideline text at every section, symbols marking new, "
|
|
"revised, add-on and telemedicine codes, and parenthetical instructions "
|
|
"cross-referencing related codes. That hierarchy is what groups codes into clinical "
|
|
"families below, and it is what a descriptor's elements are drawn from. The Federal "
|
|
"Register is the **second**: it is where CMS decides whether Medicare pays for a code "
|
|
"the AMA has already defined, and how much — the FR reprices codes, it does not "
|
|
f"reorganize the code set. CPT editions on this replica: {_years}."
|
|
)
|
|
_view
|
|
return
|
|
|
|
|
|
@app.cell(hide_code=True)
|
|
def _(mo):
|
|
# ── 1. Reading a descriptor ──
|
|
from pfs.families import HAND_FAMILIES
|
|
|
|
_codes = sorted({c for f in HAND_FAMILIES.values() for c in f.codes})
|
|
code_picker = mo.ui.dropdown(options=_codes, value="99490", label="Code")
|
|
mo.vstack(
|
|
[
|
|
mo.md(
|
|
"## 1. Reading a descriptor\n\n"
|
|
"The Federal Register prints a code's descriptor as a *stem* paragraph — the "
|
|
"sentence that opens with the code number and a parenthesis — followed by one "
|
|
"paragraph per required element, each ending in a semicolon or closing "
|
|
"parenthesis. `descriptor_runs` finds every place a rule prints that pattern for "
|
|
"a code and pairs the stem with the element paragraphs that immediately follow "
|
|
"it; `parse_descriptor` then reads whatever a regex can read out of that text — "
|
|
"minutes, billing periods, populations, activities, modalities — against a "
|
|
"**closed vocabulary** that only grows by human review. Pick a code from any of "
|
|
"the five hand-registered families below."
|
|
),
|
|
code_picker,
|
|
]
|
|
)
|
|
return (code_picker,)
|
|
|
|
|
|
@app.cell(hide_code=True)
|
|
def _(code_picker, fr_md, mo, pl, store):
|
|
from pfs.descriptors import descriptor_runs
|
|
from pfs.elements import VOCAB, parse_descriptor
|
|
|
|
code = code_picker.value
|
|
runs = descriptor_runs(store, code) if store is not None else []
|
|
if not runs:
|
|
_view = mo.md(
|
|
f"_No Federal Register descriptor run found for {code} (bibliography unavailable "
|
|
"or code never printed as a stem)._"
|
|
)
|
|
elements = ()
|
|
else:
|
|
# The original codification, preferring the earliest run that carries its own
|
|
# element paragraphs (a later rule often just cites the code inline mid-sentence,
|
|
# with no paragraph break) — for 99490 this is the CY2015 final rule, stem ¶1244.
|
|
run = next((r for r in runs if r.elements), runs[0])
|
|
_paras = pl.DataFrame(
|
|
{
|
|
"p_id": [run.stem.p_id, *[p.p_id for p in run.elements]],
|
|
"role": ["stem", *["element"] * len(run.elements)],
|
|
"text": [run.stem.text[:300], *[p.text[:300] for p in run.elements]],
|
|
}
|
|
)
|
|
elements = parse_descriptor(run.text)
|
|
_els = pl.DataFrame(
|
|
{
|
|
"type": [e.type.value for e in elements],
|
|
"value": [e.value for e in elements],
|
|
"detail": [e.detail for e in elements],
|
|
}
|
|
)
|
|
_vocab = pl.DataFrame(
|
|
{
|
|
"type": [t.value for t in VOCAB for _ in VOCAB[t]],
|
|
"value": [v for t in VOCAB for v in VOCAB[t]],
|
|
}
|
|
)
|
|
_view = mo.vstack(
|
|
[
|
|
mo.md(
|
|
f"**{code}** as printed in {run.item_key} (CY{run.rule_year}), stem "
|
|
f"{fr_md(run.item_key, run.stem.p_id)}: the stem paragraph opens the "
|
|
"descriptor and each following paragraph is one element."
|
|
),
|
|
mo.ui.table(_paras, label="Descriptor paragraphs"),
|
|
mo.md(
|
|
"The deterministic parser reads what a regex can read — minutes, periods, "
|
|
"code references, and the recurring phrases:"
|
|
),
|
|
mo.ui.table(_els, label="Typed elements"),
|
|
mo.accordion(
|
|
{
|
|
"The closed vocabulary (values grow only by review)": mo.ui.table(
|
|
_vocab
|
|
)
|
|
}
|
|
),
|
|
]
|
|
)
|
|
_view
|
|
return code, elements
|
|
|
|
|
|
@app.cell(hide_code=True)
|
|
def _(code, mo, not_built, pl, q):
|
|
# Where the manual files this code
|
|
_ed = q("SELECT max(edition_year) y FROM pfs.cpt_code")
|
|
if _ed.is_empty():
|
|
_view = mo.md(
|
|
"**Where the manual files this code.**\n\n"
|
|
+ not_built("stack pfs cpt-ingest --all")
|
|
)
|
|
else:
|
|
_year = _ed.item(0, "y")
|
|
_cpt = q(
|
|
"SELECT c.stem, c.elements, c.tail, c.addon, c.resequenced, c.new, c.revised, "
|
|
"c.telemedicine, c.mod51_exempt, c.audio_only, c.fda_pending, c.pla, s.path_key "
|
|
"FROM pfs.cpt_code c JOIN pfs.cpt_section s USING (edition_year, item_key, sec_id) "
|
|
"WHERE c.edition_year = ? AND c.code = ?",
|
|
(_year, code),
|
|
)
|
|
if _cpt.is_empty():
|
|
_view = mo.md(
|
|
f"**Where the manual files this code.** `{code}` is a HCPCS Level II code — "
|
|
"CMS's own coding system, used when Medicare has a programmatic need CPT does "
|
|
f"not cover — and is **not in the CPT {_year} book**; the manual's hierarchy "
|
|
"below does not apply to it."
|
|
)
|
|
else:
|
|
_row = _cpt.row(0, named=True)
|
|
_crumb = " → ".join(_row["path_key"].split(" > "))
|
|
_flags = [
|
|
label
|
|
for label, on in (
|
|
("add-on ✚", _row["addon"]),
|
|
("resequenced #", _row["resequenced"]),
|
|
("new ●", _row["new"]),
|
|
("revised ▲", _row["revised"]),
|
|
("telemedicine ★", _row["telemedicine"]),
|
|
("modifier-51 exempt ⦸", _row["mod51_exempt"]),
|
|
("audio-only", _row["audio_only"]),
|
|
("FDA-pending ⚡", _row["fda_pending"]),
|
|
("PLA", _row["pla"]),
|
|
)
|
|
if on
|
|
]
|
|
_instr = q(
|
|
"SELECT kind, text, targets FROM pfs.cpt_instruction WHERE edition_year = ? "
|
|
"AND code = ? ORDER BY kind, text",
|
|
(_year, code),
|
|
)
|
|
_view = mo.vstack(
|
|
[
|
|
mo.md(
|
|
"**Where the manual files this code.** The CPT hierarchy is the first "
|
|
f"organizing principle: in the CY{_year} book `{code}` sits under "
|
|
f"**{_crumb}**."
|
|
+ (
|
|
f" Symbols: {', '.join(_flags)}."
|
|
if _flags
|
|
else " No symbols set."
|
|
)
|
|
),
|
|
mo.ui.table(
|
|
pl.DataFrame(
|
|
{
|
|
"part": [
|
|
"stem",
|
|
*["element"] * len(_row["elements"]),
|
|
"tail",
|
|
],
|
|
"text": [_row["stem"], *_row["elements"], _row["tail"]],
|
|
}
|
|
),
|
|
label="The manual's own stem / elements / tail",
|
|
),
|
|
(
|
|
mo.ui.table(_instr, label="Parenthetical instructions")
|
|
if not _instr.is_empty()
|
|
else mo.md(
|
|
"_No parenthetical instructions for this code in this edition._"
|
|
)
|
|
),
|
|
]
|
|
)
|
|
_view
|
|
return
|
|
|
|
|
|
@app.cell(hide_code=True)
|
|
def _(code, mo, not_built, q):
|
|
# ── 2. What the extractor wrote ──
|
|
_els = q(
|
|
"SELECT type, value, detail, source, item_key, p_id, page FROM pfs.code_element "
|
|
"WHERE code = ? ORDER BY type, value",
|
|
(code,),
|
|
)
|
|
_rev = q(
|
|
"SELECT text, proposed_value, item_key, p_id FROM pfs.code_element_review "
|
|
"WHERE code = ? ORDER BY p_id",
|
|
(code,),
|
|
)
|
|
if _els.is_empty():
|
|
_view = mo.md(
|
|
"## 2. What the extractor wrote\n\n"
|
|
+ not_built(f"stack pfs elements --code {code}")
|
|
)
|
|
else:
|
|
_view = mo.vstack(
|
|
[
|
|
mo.md(
|
|
"## 2. What the extractor wrote\n\n"
|
|
"Three passes, in order: the regex parser above finds what it can; a local "
|
|
"model then reads every remaining candidate line and chooses **one slug from "
|
|
"the closed list, or `none`**; whatever neither pass can place is queued for "
|
|
"human review rather than guessed. Nothing enters `pfs.code_element` unless "
|
|
"it is a member of the closed vocabulary, and every row keeps the exact "
|
|
"Federal Register paragraph it came from."
|
|
),
|
|
mo.ui.table(
|
|
_els, label=f"pfs.code_element — {code} ({_els.height} rows)"
|
|
),
|
|
(
|
|
mo.ui.table(
|
|
_rev,
|
|
label=f"pfs.code_element_review — {code} ({_rev.height} lines)",
|
|
)
|
|
if not _rev.is_empty()
|
|
else mo.md("_Review queue empty for this code._")
|
|
),
|
|
]
|
|
)
|
|
_view
|
|
return
|
|
|
|
|
|
@app.cell(hide_code=True)
|
|
def _(alt, code, fr_md, mo, not_built, pl, plain_years, q):
|
|
# ── 3. Lineage ──
|
|
from pfs.families import family_of as _family_of
|
|
|
|
_fam = _family_of(code)
|
|
_codes = list(_fam.codes) if _fam else [code]
|
|
_ev = q(
|
|
"SELECT code, year, kind, from_codes, to_codes, source, anchored, item_key, p_id, note "
|
|
"FROM pfs.code_event WHERE code IN (" + ",".join("?" * len(_codes)) + ") "
|
|
"ORDER BY year, code, kind",
|
|
_codes,
|
|
)
|
|
if _ev.is_empty():
|
|
_view = mo.md(
|
|
"## 3. Lineage\n\n" + not_built(f"stack pfs lineage --code {code} --write")
|
|
)
|
|
else:
|
|
_chart = (
|
|
alt.Chart(_ev.to_pandas())
|
|
.mark_circle(size=90)
|
|
.encode(
|
|
x=alt.X("year:O", title="Rule year"),
|
|
y=alt.Y("kind:N", title=None),
|
|
color=alt.Color("source:N", title="Source"),
|
|
shape=alt.Shape("anchored:N", title="Anchored"),
|
|
tooltip=[
|
|
"code",
|
|
"year",
|
|
"kind",
|
|
"from_codes",
|
|
"to_codes",
|
|
"item_key",
|
|
"p_id",
|
|
"note",
|
|
],
|
|
)
|
|
.properties(height=260, width=640)
|
|
)
|
|
# Ruling A2: build the anchor column with a plain list comprehension, not
|
|
# DataFrame.map_rows.
|
|
_anchor_col = [
|
|
fr_md(item_key, p_id) if item_key else "rvu"
|
|
for item_key, p_id in zip(_ev["item_key"].to_list(), _ev["p_id"].to_list())
|
|
]
|
|
_links = _ev.with_columns(pl.Series("anchor", _anchor_col))
|
|
_view = mo.vstack(
|
|
[
|
|
mo.md(
|
|
"## 3. Lineage\n\n"
|
|
f"Every dated event for the **{_fam.name if _fam else code}** codes. "
|
|
"RVU-file events (`source=rvu`) are dated by the fee-schedule year; Federal "
|
|
"Register events are dated by the **rule that mentions them** — a later rule "
|
|
"recounting a code's creation adds a later `created` row, which is why the "
|
|
"earliest anchored event is the origin, not the latest one. An RVU event is "
|
|
"`anchored` when a Federal Register event for the same code lies within one "
|
|
"rule year of it; an unanchored RVU event is evidence CMS never wrote a "
|
|
"sentence about, and is weaker to cite. CPT-source events (`source=cpt`) are "
|
|
"dated a third way: `cpt_changed` rows use the AMA's own CPT Changes edition "
|
|
"year — the year the AMA revised the code, independent of whether or when any "
|
|
"FR rule mentions it — and both an FR event and a CPT event for the same code "
|
|
"can anchor the same RVU change from two independent directions."
|
|
),
|
|
mo.ui.altair_chart(_chart),
|
|
mo.ui.table(
|
|
plain_years(_links.drop("item_key", "p_id")), label="pfs.code_event"
|
|
),
|
|
]
|
|
)
|
|
_view
|
|
return
|
|
|
|
|
|
@app.cell(hide_code=True)
|
|
def _(code, mo, not_built, pl, plain_years, q):
|
|
# ── 4. Families ──
|
|
from pfs.families import family_of as _family_of
|
|
|
|
_hand = _family_of(code)
|
|
_key = _hand.key if _hand else ""
|
|
_rows = (
|
|
q(
|
|
"SELECT key, name, code, role, since, until, item_key, p_id, note "
|
|
"FROM pfs.code_family WHERE key = ? ORDER BY code",
|
|
(_key,),
|
|
)
|
|
if _key
|
|
else None
|
|
)
|
|
if _rows is None or _rows.is_empty():
|
|
_view = mo.md("## 4. Families\n\n" + not_built("stack pfs families --write"))
|
|
else:
|
|
# The family's CPT heading — the organizing principle that classified its members
|
|
# (Task 6 context: "note" is per-code, so different members can carry different
|
|
# headings when the hand list or a derived edge pulls in a neighboring heading).
|
|
_notes = [n for n in _rows["note"].to_list() if n]
|
|
_note = _notes[0] if _notes else ""
|
|
_siblings = pl.DataFrame()
|
|
if _note:
|
|
_parts = _note.split(" > ")
|
|
_parent_prefix = " > ".join(_parts[:-1])
|
|
_yr = q("SELECT max(edition_year) y FROM pfs.cpt_section")
|
|
_year = _yr.item(0, "y") if not _yr.is_empty() else None
|
|
if _year is not None:
|
|
_lvl_row = q(
|
|
"SELECT min(level) lvl FROM pfs.cpt_section WHERE edition_year = ? "
|
|
"AND path_key = ?",
|
|
(_year, _note),
|
|
)
|
|
_lvl = (
|
|
_lvl_row.item(0, "lvl")
|
|
if not _lvl_row.is_empty() and _lvl_row.item(0, "lvl") is not None
|
|
else None
|
|
)
|
|
if _lvl is not None:
|
|
_siblings = q(
|
|
"SELECT DISTINCT title FROM pfs.cpt_section WHERE edition_year = ? "
|
|
"AND path_key LIKE ? AND level = ? ORDER BY title",
|
|
(_year, _parent_prefix + " > %", _lvl),
|
|
)
|
|
_summary = q(
|
|
"WITH fam AS (SELECT key, count(*) n, bool_or(note <> '') has_note "
|
|
"FROM pfs.code_family GROUP BY key) "
|
|
"SELECT count(*) total, count(*) FILTER (WHERE n > 1) multi, "
|
|
"count(*) FILTER (WHERE has_note) cpt_named FROM fam"
|
|
)
|
|
_by_chapter = q(
|
|
"SELECT split_part(note, ' > ', 1) AS chapter, count(DISTINCT key) AS families "
|
|
"FROM pfs.code_family WHERE note <> '' GROUP BY 1 ORDER BY families DESC"
|
|
)
|
|
_view = mo.vstack(
|
|
[
|
|
mo.md(
|
|
"## 4. Families\n\n"
|
|
"A family is a connected component over four kinds of edge: an **add-on** "
|
|
"relation element (`in conjunction with 99490`), a **defined-by-reference** "
|
|
"relation (`with the elements included in 99490`), a **single-target "
|
|
"replacement** lineage event (one code's `replaced_by` names exactly one "
|
|
"successor), and **stem similarity with an identical activity set** (two "
|
|
"descriptors share half their service-naming words and every activity "
|
|
"element). The hand-written registry is a floor, never a ceiling: "
|
|
f"**{_key}** lists {len(_hand.codes)} hand codes; the derived table below "
|
|
f"shows {_rows.height}, because the connected-component search also reaches "
|
|
"codes the hand list never named."
|
|
),
|
|
mo.md(
|
|
f"**Where this family sits in the manual.** {_note}"
|
|
if _note
|
|
else "_No member of this family carries a CPT heading (HCPCS-only family)._"
|
|
),
|
|
mo.ui.table(plain_years(_rows), label=f"pfs.code_family — {_key}"),
|
|
(
|
|
mo.ui.table(
|
|
_siblings, label="Sibling headings under the same parent"
|
|
)
|
|
if not _siblings.is_empty()
|
|
else mo.md("_No sibling headings found._")
|
|
),
|
|
mo.md(
|
|
"**Example (single-target replacement):** HCPCS G2058, billable only in "
|
|
"2020, was replaced the following year by CPT 99439 — an identical "
|
|
"descriptor crosswalked at the same value. That `replaced_by` event names "
|
|
"exactly one target code, so it is unambiguous evidence, and G2058 joins CCM "
|
|
"as a *predecessor* rather than becoming a one-code family of its own."
|
|
),
|
|
(
|
|
mo.md(
|
|
f"**Across all families:** {_summary.item(0, 'total')} total, "
|
|
f"{_summary.item(0, 'multi')} multi-code, "
|
|
f"{_summary.item(0, 'cpt_named')} carry a CPT heading (`note <> ''`)."
|
|
)
|
|
if not _summary.is_empty()
|
|
else mo.md("")
|
|
),
|
|
(
|
|
mo.ui.table(
|
|
_by_chapter, label="CPT-named families by top-level chapter"
|
|
)
|
|
if not _by_chapter.is_empty()
|
|
else mo.md("")
|
|
),
|
|
]
|
|
)
|
|
_view
|
|
return
|
|
|
|
|
|
@app.cell(hide_code=True)
|
|
def _(mo, not_built):
|
|
# ── 5. Anchors in the corpus ──
|
|
mo.md(
|
|
"## 5. Anchors in the corpus\n\n"
|
|
"How many chunks in the RAG index — comments, guidance, Federal Register text — carry "
|
|
"this family's codes as metadata, and how many bibliography items carry a `family:` or "
|
|
"`code:` tag. That count lives in pgvector, not the DuckDB replica this notebook reads.\n\n"
|
|
+ not_built("stack llm restamp")
|
|
)
|
|
return
|
|
|
|
|
|
@app.cell(hide_code=True)
|
|
def _(mo, not_built):
|
|
# ── 6. Guidance ──
|
|
mo.md(
|
|
"## 6. Guidance\n\n"
|
|
"Sub-regulatory guidance — Medicare Learning Network articles, Internet-Only Manual "
|
|
"sections, MACs' local coverage determinations — that cites a family's codes, with CFR "
|
|
"cross-references where the guidance implements a rule.\n\n"
|
|
+ not_built("stack pfs guidance --family CCM --write")
|
|
)
|
|
return
|
|
|
|
|
|
@app.cell(hide_code=True)
|
|
def _(mo, not_built):
|
|
# ── 7. Reaction ──
|
|
mo.md(
|
|
"## 7. Reaction\n\n"
|
|
"Public comment volume on a family's codes over time — how many comment letters mention "
|
|
"them, in which rule years, and (on a sample) whether commenters supported or opposed the "
|
|
"proposal.\n\n" + not_built("stack pfs reaction --family CCM --write")
|
|
)
|
|
return
|
|
|
|
|
|
@app.cell(hide_code=True)
|
|
def _(NOTES, REPLICA_PATH, con, mo, pl, q, store):
|
|
# ── 8. Provenance ──
|
|
import datetime as _dt
|
|
|
|
if REPLICA_PATH.exists():
|
|
_mtime = _dt.datetime.fromtimestamp(REPLICA_PATH.stat().st_mtime).isoformat(
|
|
timespec="seconds"
|
|
)
|
|
_replica_line = f"`{REPLICA_PATH}` — last published {_mtime}"
|
|
else:
|
|
_replica_line = f"`{REPLICA_PATH}` — not found"
|
|
_counts = q(
|
|
"SELECT 'code_element' t, count(*) n FROM pfs.code_element "
|
|
"UNION ALL SELECT 'code_element_review', count(*) FROM pfs.code_element_review "
|
|
"UNION ALL SELECT 'code_event', count(*) FROM pfs.code_event "
|
|
"UNION ALL SELECT 'code_family', count(*) FROM pfs.code_family"
|
|
)
|
|
_log = q(
|
|
"SELECT run_id, ingested_at, module, table_name, rule_id, source_file, sha256, rows, "
|
|
"fr_citation, pincite_key FROM cms.ingest_log WHERE table_name LIKE 'pfs.%' "
|
|
"ORDER BY ingested_at DESC LIMIT 20"
|
|
)
|
|
# pfs.cpt_* comes from a separate ingest path (stack pfs cpt-ingest) with its own
|
|
# per-edition grain, so it gets its own counts table and its own bib provenance —
|
|
# one row per CPT edition on the replica, with the edition's bib title.
|
|
_cpt_counts = q(
|
|
"SELECT 'cpt_section' t, edition_year, count(*) n FROM pfs.cpt_section GROUP BY 1, 2 "
|
|
"UNION ALL SELECT 'cpt_code', edition_year, count(*) FROM pfs.cpt_code GROUP BY 1, 2 "
|
|
"UNION ALL SELECT 'cpt_instruction', edition_year, count(*) "
|
|
"FROM pfs.cpt_instruction GROUP BY 1, 2 "
|
|
"UNION ALL SELECT 'cpt_reference', edition_year, count(*) "
|
|
"FROM pfs.cpt_reference GROUP BY 1, 2 "
|
|
"UNION ALL SELECT 'cpt_crosswalk', edition_year, count(*) "
|
|
"FROM pfs.cpt_crosswalk GROUP BY 1, 2 "
|
|
"UNION ALL SELECT 'cpt_list', edition_year, count(*) FROM pfs.cpt_list GROUP BY 1, 2 "
|
|
"UNION ALL SELECT 'cpt_code_alt', edition_year, count(*) "
|
|
"FROM pfs.cpt_code_alt GROUP BY 1, 2 "
|
|
"ORDER BY edition_year, t"
|
|
)
|
|
_cpt_editions = q(
|
|
"SELECT DISTINCT edition_year, item_key FROM pfs.cpt_code ORDER BY edition_year"
|
|
)
|
|
_bib_rows = []
|
|
for _yr, _key in zip(
|
|
[] if _cpt_editions.is_empty() else _cpt_editions["edition_year"].to_list(),
|
|
[] if _cpt_editions.is_empty() else _cpt_editions["item_key"].to_list(),
|
|
):
|
|
if store is None:
|
|
_title = "(bibliography unavailable)"
|
|
else:
|
|
try:
|
|
_title = store.get(_key).title
|
|
except KeyError:
|
|
_title = "(item not found)"
|
|
_bib_rows.append({"edition_year": _yr, "item_key": _key, "title": _title})
|
|
_cpt_bib = pl.DataFrame(_bib_rows) if _bib_rows else pl.DataFrame()
|
|
mo.vstack(
|
|
[
|
|
mo.md(
|
|
"## 8. Provenance\n\n"
|
|
f"Replica: {_replica_line}"
|
|
+ (
|
|
" (connection open)"
|
|
if con is not None
|
|
else " (connection unavailable)"
|
|
)
|
|
+ ". Row counts and the most recent ingest-log entries for every `pfs.code_*` "
|
|
"table this notebook reads follow, so a reader can tell how fresh the family "
|
|
"data is without leaving the page. The CPT codebook tables (`pfs.cpt_*`) are "
|
|
"ingested separately, one edition at a time, so their counts and bib "
|
|
"provenance are broken out by edition below."
|
|
),
|
|
mo.ui.table(_counts, label="pfs.code_* row counts")
|
|
if not _counts.is_empty()
|
|
else mo.md("_replica unavailable_"),
|
|
mo.ui.table(_log, label="Ingest log — pfs.*")
|
|
if not _log.is_empty()
|
|
else mo.md("_no ingest log rows_"),
|
|
mo.ui.table(_cpt_counts, label="pfs.cpt_* row counts by edition")
|
|
if not _cpt_counts.is_empty()
|
|
else mo.md(
|
|
"_no CPT codebook tables built yet — run `stack pfs cpt-ingest --all`_"
|
|
),
|
|
mo.ui.table(_cpt_bib, label="CPT edition bib items")
|
|
if not _cpt_bib.is_empty()
|
|
else mo.md("_no CPT editions ingested_"),
|
|
mo.md(
|
|
"\n".join(f"- {k}: {v}" for k, v in NOTES.items())
|
|
if NOTES
|
|
else "_All sources available._"
|
|
),
|
|
]
|
|
)
|
|
return
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run()
|