479 lines
18 KiB
Python
479 lines
18 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):
|
|
# ── 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, 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."
|
|
),
|
|
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, 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 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:
|
|
_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.ui.table(plain_years(_rows), label=f"pfs.code_family — {_key}"),
|
|
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."
|
|
),
|
|
]
|
|
)
|
|
_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, q):
|
|
# ── 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"
|
|
)
|
|
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."
|
|
),
|
|
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.md(
|
|
"\n".join(f"- {k}: {v}" for k, v in NOTES.items())
|
|
if NOTES
|
|
else "_All sources available._"
|
|
),
|
|
]
|
|
)
|
|
return
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run()
|