homelab/stack
1
0
Fork 0
Code Issues 70 Pull Requests 1 Actions Packages Projects 1 Releases Wiki Activity

feat(pfs): NPRM Addendum E loader -> pfs.gpci_proposed + ingest wiring (closes #629)

Browse Source
This commit is contained in:
kert
2026-08-17 12:16:17 -04:00
parent f68a16806c
commit 39dfaa20ae
3 changed files with 305 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
dev/scripts/ingest_pfs.py
View File

@@ -64,7 +64,11 @@ def main() -> int:
)
if args.nprm:
from pfs.nprm import NPRM_SOURCES, load_rvu_proposed
from pfs.nprm import (
NPRM_SOURCES,
load_gpci_proposed,
load_rvu_proposed,
)
print("\n--- NPRM Addendum B (proposed RVUs) ---")
for year, _tag, cms_rule_id, _pincite_key in NPRM_SOURCES:
@@ -82,6 +86,22 @@ def main() -> int:
run_id=run_id,
)
print("\n--- NPRM Addendum E (proposed GPCIs) ---")
for year, _tag, cms_rule_id, _pincite_key in NPRM_SOURCES:
gpci_out = load_gpci_proposed(con, year=year)
print(f" pfs.gpci_proposed [{cms_rule_id}]: {gpci_out}")
ingest_log.log_ingest(
con,
module="pfs",
table_name="pfs.gpci_proposed",
rows=gpci_out["rows"],
source_file=gpci_out["source_file"],
rule_id=gpci_out["cms_rule_id"],
fr_citation=gpci_out["fr_citation"],
pincite_key=gpci_out["pincite_key"],
run_id=run_id,
)
replica = publish_replica("aco")
print(f"replica → {replica}")

191
src/pfs/nprm.py
View File

@@ -22,6 +22,7 @@ from pathlib import Path
from typing import Any
_ADDENDUM_B_RE = re.compile(r"(?i)addendum[_ ]?b")
_ADDENDUM_E_RE = re.compile(r"(?i)addendum[_ ]?e")
# pfs.pipe._RVU_COLUMNS normalizes raw headers to these intermediate
# names, which already match the pfs.rvu_proposed target schema (#607
@@ -65,12 +66,12 @@ def _source_for(year: int) -> tuple[int, str, str, str]:
raise ValueError(f"No NPRM source registered for year {year} (known: {known})")
def _discover_addendum_b(tag: str) -> Path:
"""Locate the NPRM Addendum B attachment in the bib store.
def _discover_attachment(tag: str, pattern: re.Pattern[str], label: str) -> Path:
"""Locate an NPRM addendum attachment in the bib store.
Queries attachments of items tagged ``tag`` for one whose filename
matches ``addendum[_ ]?b`` (case-insensitive), mirroring the join
in ``llm/source.py::_attachment_text``.
matches ``pattern``, mirroring the join in
``llm/source.py::_attachment_text``.
"""
from conf import connect
@@ -90,13 +91,23 @@ def _discover_addendum_b(tag: str) -> Path:
)
for row in rows:
filename = row["filename"] or ""
if _ADDENDUM_B_RE.search(filename):
if pattern.search(filename):
return Path(row["storage_path"])
raise FileNotFoundError(
f"No Addendum B attachment found among bib items tagged {tag!r}"
f"No {label} attachment found among bib items tagged {tag!r}"
)
def _discover_addendum_b(tag: str) -> Path:
"""Locate the NPRM Addendum B (proposed RVUs) attachment."""
return _discover_attachment(tag, _ADDENDUM_B_RE, "Addendum B")
def _discover_addendum_e(tag: str) -> Path:
"""Locate the NPRM Addendum E (proposed GPCIs) attachment."""
return _discover_attachment(tag, _ADDENDUM_E_RE, "Addendum E")
def _schema_matches_target(con: Any) -> bool:
"""True if ``pfs.rvu_proposed`` doesn't exist yet, or its columns match.
@@ -226,3 +237,171 @@ def load_rvu_proposed(con: Any, *, year: int) -> dict:
"fr_citation": fr_citation,
"pincite_key": pincite_key,
}
# ── Addendum E → pfs.gpci_proposed (#629) ─────────────────────────
# One row per (locality, gpci_year, cms_rule_id). GPCIs update
# biennially, so a single NPRM's Addendum E can propose more than one
# year: the CY2026 workbook (CMS-1832-P) carries proposed 2026 AND
# phase-in 2027 columns (plus a 2025 baseline restatement we skip);
# the CY2027 workbook (CMS-1848-P) carries 2027 only. Proposed PW
# GPCIs are published WITHOUT the statutory 1.0 work floor — values
# load as published; floor labeling is presentation-layer work.
_GPCI_TARGET_COLUMNS = [
"mac",
"state",
"locality",
"locality_name",
"work_gpci",
"pe_gpci",
"mp_gpci",
"gpci_year",
"cms_rule_id",
]
_GPCI_HEADER_RE = re.compile(r"(?i)^medicare administrative")
_GPCI_YEAR_COL_RE = re.compile(r"(?i)^(\d{4})\s+(PW|PE|MP)\s+GPCI")
_GPCI_MAC_RE = re.compile(r"^\d{5}$")
def _gpci_schema_matches(con: Any) -> bool:
"""True if ``pfs.gpci_proposed`` doesn't exist yet, or matches target."""
rows = con.execute(
"SELECT column_name FROM information_schema.columns "
"WHERE table_schema = 'pfs' AND table_name = 'gpci_proposed' "
"ORDER BY ordinal_position"
).fetchall()
if not rows:
return True
return [r[0] for r in rows] == _GPCI_TARGET_COLUMNS
def _to_float(value: Any) -> float | None:
try:
return float(str(value).strip())
except (TypeError, ValueError):
return None
def load_gpci_proposed(con: Any, *, year: int) -> dict:
"""Load ``year``'s NPRM Addendum E workbook into ``pfs.gpci_proposed``.
Same contract and partition discipline as :func:`load_rvu_proposed`:
resolves tag/rule-id/pincite from ``NPRM_SOURCES``, deletes only its
own ``cms_rule_id`` partition before inserting, and returns
``{"rows", "source_file", "cms_rule_id", "fr_citation",
"pincite_key"}``. Year-columns earlier than the NPRM's own rule
year (baseline restatements) are skipped.
"""
import polars as pl
from pfs.rules import proposed_for
_, tag, cms_rule_id, pincite_key = _source_for(year)
rule = proposed_for(year)
fr_citation = rule.federal_register_citation if rule is not None else ""
path = _discover_addendum_e(tag)
if path.suffix.lower() == ".csv":
raw = pl.read_csv(path, has_header=False, infer_schema_length=0)
else:
raw = pl.read_excel(path, has_header=False)
header_idx = None
for i, v in enumerate(raw.get_column(raw.columns[0])):
if v is not None and _GPCI_HEADER_RE.match(str(v).strip()):
header_idx = i
break
if header_idx is None:
raise ValueError(f"{path}: no 'Medicare Administrative ...' header row found")
headers = [str(v).strip() if v is not None else "" for v in raw.row(header_idx)]
def _col(prefix: str) -> int:
for i, h in enumerate(headers):
if h.lower().startswith(prefix):
return i
raise ValueError(f"{path}: no {prefix!r} column in header {headers!r}")
mac_i = _col("medicare administrative")
state_i = _col("state")
loc_i = _col("locality number")
locname_i = _col("locality name")
year_cols: dict[int, dict[str, int]] = {}
for i, h in enumerate(headers):
m = _GPCI_YEAR_COL_RE.match(h)
if m:
year_cols.setdefault(int(m.group(1)), {})[m.group(2).upper()] = i
load_years = sorted(
yy
for yy, cols in year_cols.items()
if yy >= year and {"PW", "PE", "MP"} <= set(cols)
)
if not load_years:
raise ValueError(
f"{path}: no complete PW/PE/MP GPCI column triplet for year >= {year}"
)
records: list[dict[str, Any]] = []
for row in raw.slice(header_idx + 1).iter_rows():
mac = str(row[mac_i]).strip() if row[mac_i] is not None else ""
# Footnote/blank rows never carry a 5-digit MAC.
if not _GPCI_MAC_RE.match(mac):
continue
for yy in load_years:
cols = year_cols[yy]
records.append(
{
"mac": mac,
"state": str(row[state_i]).strip(),
"locality": str(row[loc_i]).strip(),
"locality_name": str(row[locname_i]).strip(),
"work_gpci": _to_float(row[cols["PW"]]),
"pe_gpci": _to_float(row[cols["PE"]]),
"mp_gpci": _to_float(row[cols["MP"]]),
"gpci_year": yy,
"cms_rule_id": cms_rule_id,
}
)
df = pl.DataFrame(records, schema_overrides={"gpci_year": pl.Int64}).select(
_GPCI_TARGET_COLUMNS
)
con.execute("CREATE SCHEMA IF NOT EXISTS pfs")
if not _gpci_schema_matches(con):
con.execute("DROP TABLE IF EXISTS pfs.gpci_proposed")
con.execute(
"""
CREATE TABLE IF NOT EXISTS pfs.gpci_proposed (
mac VARCHAR,
state VARCHAR,
locality VARCHAR,
locality_name VARCHAR,
work_gpci DOUBLE,
pe_gpci DOUBLE,
mp_gpci DOUBLE,
gpci_year BIGINT,
cms_rule_id VARCHAR
)
"""
)
con.execute("DELETE FROM pfs.gpci_proposed WHERE cms_rule_id = ?", [cms_rule_id])
con.register("_gpci_df", df)
try:
con.execute(
"INSERT INTO pfs.gpci_proposed SELECT "
"mac, state, locality, locality_name, work_gpci, pe_gpci, "
"mp_gpci, gpci_year, cms_rule_id FROM _gpci_df"
)
finally:
con.unregister("_gpci_df")
return {
"rows": df.height,
"source_file": str(path),
"cms_rule_id": cms_rule_id,
"fr_citation": fr_citation,
"pincite_key": pincite_key,
}

99
tests/pfs/test_nprm.py
View File

@@ -206,3 +206,102 @@ def test_load_rvu_proposed_drops_footnote_rows(tmp_path, monkeypatch):
"SELECT count(*) FROM pfs.rvu_proposed WHERE hcpcs LIKE '1 CPT%'"
).fetchone()[0]
assert footnote == 0
# ── load_gpci_proposed (#629) ──────────────────────────────────────
def _gpci_fixture_2027(tmp_path):
"""CY2027 Addendum E layout: title row, header row, data, footnote."""
p = tmp_path / "Addendum_E_CY2027.csv"
p.write_text(
"ADDENDUM E. CY 2027 GEOGRAPHIC PRACTICE COST INDICES,,,,,,\n"
"Medicare Administrative Contractor,State,Locality Number,Locality Name,"
"2027 PW GPCI (without 1.0 Floor),2027 PE GPCI,2027 MP GPCI\n"
"10112,AL,00,ALABAMA,0.975,0.88,0.557\n"
"02102,AK,01,ALASKA*,1.5,1.049,0.511\n"
"*Note: localities marked with an asterisk...,,,,,,\n"
)
return p
def _gpci_fixture_2026_wide(tmp_path):
"""CY2026 Addendum E layout: 2025 baseline + 2026 and 2027 proposed."""
p = tmp_path / "Addendum_E_CY2026.csv"
p.write_text(
"ADDENDUM E. PROPOSED CY 2026 GPCIs,,,,,,,,,,,,\n"
"Medicare Administrative Contractor,State,Locality Number,Locality Name,"
"2025 PW GPCI (with 1.0 Floor),2025 PE GPCI,2025 MP GPCI,"
"2026 PW GPCI (without 1.0 Floor),2026 PE GPCI,2026 MP GPCI,"
"2027 PW GPCI (without 1.0 Floor),2027 PE GPCI,2027 MP GPCI\n"
"10112,AL,00,ALABAMA,1,0.869,0.575,0.988,0.875,0.566,0.975,0.88,0.557\n"
"02102,AK,01,ALASKA*,1.5,1.081,0.592,1.5,1.065,0.551,1.5,1.049,0.511\n"
)
return p
def _patch_gpci_discovery(monkeypatch, mapping):
def _fake(tag):
try:
return mapping[tag]
except KeyError:
raise FileNotFoundError(
f"No Addendum E attachment found among bib items tagged {tag!r}"
) from None
monkeypatch.setattr(nprm, "_discover_addendum_e", _fake)
def test_load_gpci_proposed_cy2027_single_year(tmp_path, monkeypatch):
path = _gpci_fixture_2027(tmp_path)
_patch_gpci_discovery(monkeypatch, {"sup:2027_PFS_NPRM": path})
con = duckdb.connect()
out = nprm.load_gpci_proposed(con, year=2027)
assert out["rows"] == 2
assert out["cms_rule_id"] == "CMS-1848-P"
rows = con.execute(
"SELECT mac, state, locality, locality_name, work_gpci, pe_gpci, "
"mp_gpci, gpci_year, cms_rule_id FROM pfs.gpci_proposed ORDER BY mac"
).fetchall()
assert rows == [
("02102", "AK", "01", "ALASKA*", 1.5, 1.049, 0.511, 2027, "CMS-1848-P"),
("10112", "AL", "00", "ALABAMA", 0.975, 0.88, 0.557, 2027, "CMS-1848-P"),
]
def test_load_gpci_proposed_cy2026_wide_melts_two_years(tmp_path, monkeypatch):
path = _gpci_fixture_2026_wide(tmp_path)
_patch_gpci_discovery(monkeypatch, {"sup:2026_PFS_NPRM": path})
con = duckdb.connect()
out = nprm.load_gpci_proposed(con, year=2026)
# 2 localities x 2 proposed years (2026, 2027) — the 2025 baseline
# restatement is skipped.
assert out["rows"] == 4
years = con.execute(
"SELECT DISTINCT gpci_year FROM pfs.gpci_proposed ORDER BY gpci_year"
).fetchall()
assert years == [(2026,), (2027,)]
pe = con.execute(
"SELECT gpci_year, pe_gpci FROM pfs.gpci_proposed "
"WHERE state='AL' ORDER BY gpci_year"
).fetchall()
assert pe == [(2026, 0.875), (2027, 0.88)]
def test_load_gpci_proposed_partitions_isolated(tmp_path, monkeypatch):
_patch_gpci_discovery(
monkeypatch,
{
"sup:2026_PFS_NPRM": _gpci_fixture_2026_wide(tmp_path),
"sup:2027_PFS_NPRM": _gpci_fixture_2027(tmp_path),
},
)
con = duckdb.connect()
nprm.load_gpci_proposed(con, year=2026)
nprm.load_gpci_proposed(con, year=2027)
nprm.load_gpci_proposed(con, year=2027) # reload one partition
counts = con.execute(
"SELECT cms_rule_id, count(*) FROM pfs.gpci_proposed "
"GROUP BY cms_rule_id ORDER BY cms_rule_id"
).fetchall()
assert counts == [("CMS-1832-P", 4), ("CMS-1848-P", 2)]