feat(pfs): NPRM Addendum B loader -> pfs.rvu_proposed + ingest wiring (refs #592)

dev/scripts/ingest_pfs.py

@@ -33,10 +33,19 @@ def main() -> int:
         action="store_true",
         help="skip publishing to the DuckLake lakehouse",
     )
+    parser.add_argument(
+        "--nprm",
+        action="store_true",
+        help="also load the NPRM Addendum B proposed RVUs into pfs.rvu_proposed",
+    )
     args = parser.parse_args()
 
+    from cms import ingest_log
     from conf.connect import duckdb_batch, publish_replica
     from pfs.pipe import load_all
+    from pfs.rules import RULES
+
+    run_id = ingest_log.new_run_id()
 
     print("Ingesting CMS PFS files into DuckDB ...")
     with duckdb_batch("aco") as con:
@@ -44,6 +53,31 @@ def main() -> int:
         print("\n--- PFS tables ---")
         for name, info in sorted(summary.items()):
             print(f"  pfs.{name:24s}: {info}")
+            ingest_log.log_ingest(
+                con,
+                module="pfs",
+                table_name=name,
+                rows=info.get("rows", 0) if isinstance(info, dict) else 0,
+                run_id=run_id,
+            )
+
+        if args.nprm:
+            from pfs.nprm import load_rvu_proposed
+
+            print("\n--- NPRM Addendum B (proposed RVUs) ---")
+            nprm_out = load_rvu_proposed(con)
+            print(f"  pfs.rvu_proposed        : {nprm_out}")
+            ingest_log.log_ingest(
+                con,
+                module="pfs",
+                table_name="pfs.rvu_proposed",
+                rows=nprm_out["rows"],
+                source_file=nprm_out["source_file"],
+                rule_id="CMS-1832-P",
+                fr_citation=RULES[2026].proposed.federal_register_citation,
+                pincite_key="2KVJ2HKX",
+                run_id=run_id,
+            )
 
     replica = publish_replica("aco")
     print(f"replica → {replica}")
@@ -52,7 +86,7 @@ def main() -> int:
         print("\n--- Publishing to DuckLake ---")
         import _lake
 
-        _lake.publish_lake(("pfs",))
+        _lake.publish_lake(("pfs", "cms"))
 
     print("\nDone.")
     return 0

src/pfs/nprm.py

@@ -0,0 +1,171 @@
+"""NPRM Addendum B loader → ``pfs.rvu_proposed`` (#592).
+
+Loads the CY2026 NPRM Addendum B "Relative Value Units and Related
+Information" workbook — registered in bib under the
+``sup:2026_PFS_NPRM`` tag — into ``pfs.rvu_proposed``, the
+proposed-rule counterpart to the final-rule ``pfs.rvu`` table that
+``pfs.pipe.load_all`` populates. Reuses ``pfs.pipe``'s header-detect
+readers and column map instead of reimplementing them; the target
+schema is intentionally narrower than ``pfs.rvu`` (no PPRRVU-only
+columns) since Addendum B never carries them.
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+from typing import Any
+
+_ADDENDUM_B_TAG = "sup:2026_PFS_NPRM"
+_ADDENDUM_B_RE = re.compile(r"(?i)addendum[_ ]?b")
+
+# pfs.pipe._RVU_COLUMNS normalizes raw headers to these intermediate
+# names; map them onto the pfs.rvu_proposed target schema.
+_TARGET_RENAME = {
+    "hcpcs": "hcpcs",
+    "mod": "modifier",
+    "description": "description",
+    "status_code": "status_code",
+    "work_rvu": "work_rvu",
+    "non_fac_pe_rvu": "nonfac_pe_rvu",
+    "fac_pe_rvu": "fac_pe_rvu",
+    "mp_rvu": "mp_rvu",
+}
+
+_TARGET_COLUMNS = [
+    "hcpcs",
+    "modifier",
+    "description",
+    "status_code",
+    "work_rvu",
+    "nonfac_pe_rvu",
+    "fac_pe_rvu",
+    "mp_rvu",
+    "cms_rule_id",
+]
+
+# Intermediate (pre-rename) names as produced by pfs.pipe._normalize_columns.
+_RAW_NUMERIC_COLUMNS = ("work_rvu", "non_fac_pe_rvu", "fac_pe_rvu", "mp_rvu")
+
+
+def _discover_addendum_b() -> Path:
+    """Locate the NPRM Addendum B attachment in the bib store.
+
+    Queries attachments of items tagged ``sup:2026_PFS_NPRM`` for one
+    whose filename matches ``addendum[_ ]?b`` (case-insensitive),
+    mirroring the join in ``llm/source.py::_attachment_text``.
+    """
+    from conf import connect
+
+    store = connect.bib()
+    rows = (
+        store._con()
+        .execute(
+            "SELECT a.filename, a.storage_path FROM attachments a "
+            "JOIN items i ON i.id = a.item_id "
+            "JOIN item_tags it ON it.item_id = i.id "
+            "JOIN tags t ON t.id = it.tag_id "
+            "WHERE t.name = ?",
+            (_ADDENDUM_B_TAG,),
+        )
+        .fetchall()
+    )
+    for row in rows:
+        filename = row["filename"] or ""
+        if _ADDENDUM_B_RE.search(filename):
+            return Path(row["storage_path"])
+    raise FileNotFoundError(
+        f"No Addendum B attachment found among bib items tagged {_ADDENDUM_B_TAG!r}"
+    )
+
+
+def load_rvu_proposed(con: Any, *, cms_rule_id: str = "CMS-1832-P") -> dict:
+    """Load the NPRM Addendum B workbook into ``pfs.rvu_proposed``.
+
+    Idempotent: deletes any existing rows for ``cms_rule_id`` before
+    inserting, so re-running the ingest is safe.
+
+    Returns ``{"rows": int, "source_file": str}``.
+    """
+    import polars as pl
+
+    from pfs.pipe import (
+        _RVU_COLUMNS,
+        _normalize_columns,
+        _read_csv_with_header,
+        _read_excel,
+    )
+
+    path = _discover_addendum_b()
+    ext = path.suffix.lower()
+    if ext == ".csv":
+        df = _read_csv_with_header(str(path))
+    else:
+        df = _read_excel(str(path))
+
+    if "hcpcs" not in df.columns:
+        df = _normalize_columns(df, _RVU_COLUMNS)
+    if "hcpcs" not in df.columns:
+        raise ValueError(f"{path}: no hcpcs column found after normalization")
+
+    for col in (*_RAW_NUMERIC_COLUMNS, "mod", "description", "status_code"):
+        if col not in df.columns:
+            df = df.with_columns(pl.lit(None).alias(col))
+
+    df = df.with_columns(
+        # Addendum B numeric cells carry leading/trailing padding
+        # ("  0.00") that Polars' string->float cast won't tolerate —
+        # strip before casting or every padded value silently becomes
+        # null. Bare cast(strict=False) still nulls genuine non-numeric
+        # markers like "NA".
+        [
+            pl.col(c).cast(pl.Utf8).str.strip_chars().cast(pl.Float64, strict=False)
+            for c in _RAW_NUMERIC_COLUMNS
+        ]
+    )
+    df = df.with_columns(
+        pl.col(c).cast(pl.Utf8, strict=False).str.strip_chars()
+        for c in ("hcpcs", "mod", "description", "status_code")
+    )
+    # Addendum B workbooks append copyright/footnote text in column A
+    # below the data (e.g. "1 CPT codes and descriptors only are
+    # copyright ..."). Real HCPCS/CPT codes are always exactly 5
+    # alphanumeric characters, so that length check both drops blank
+    # rows and these footnotes without needing to special-case them.
+    df = df.filter(
+        pl.col("hcpcs").is_not_null()
+        & pl.col("hcpcs").str.contains(r"^[A-Za-z0-9]{5}$")
+    )
+
+    df = df.select(list(_TARGET_RENAME)).rename(_TARGET_RENAME)
+    df = df.with_columns(pl.lit(cms_rule_id).alias("cms_rule_id"))
+    df = df.select(_TARGET_COLUMNS)
+
+    con.execute("CREATE SCHEMA IF NOT EXISTS pfs")
+    con.execute(
+        """
+        CREATE TABLE IF NOT EXISTS pfs.rvu_proposed (
+            hcpcs VARCHAR,
+            modifier VARCHAR,
+            description VARCHAR,
+            status_code VARCHAR,
+            work_rvu DOUBLE,
+            nonfac_pe_rvu DOUBLE,
+            fac_pe_rvu DOUBLE,
+            mp_rvu DOUBLE,
+            cms_rule_id VARCHAR
+        )
+        """
+    )
+    con.execute("DELETE FROM pfs.rvu_proposed WHERE cms_rule_id = ?", [cms_rule_id])
+    con.register("_nprm_df", df)
+    try:
+        con.execute(
+            "INSERT INTO pfs.rvu_proposed SELECT "
+            "hcpcs, modifier, description, status_code, work_rvu, "
+            "nonfac_pe_rvu, fac_pe_rvu, mp_rvu, cms_rule_id FROM _nprm_df"
+        )
+    finally:
+        con.unregister("_nprm_df")
+
+    return {"rows": df.height, "source_file": str(path)}

tests/pfs/test_nprm.py

@@ -0,0 +1,32 @@
+"""Tests for pfs.nprm — NPRM Addendum B loader → pfs.rvu_proposed (#592)."""
+
+from __future__ import annotations
+
+import duckdb
+
+from pfs import nprm
+
+
+def _fixture(tmp_path):
+    p = tmp_path / "CY2026_NPRM_Addendum_B.csv"
+    p.write_text(
+        "HCPCS,MOD,DESCRIPTION,STATUS CODE,WORK RVU,"
+        "NON-FAC PE RVU,FAC PE RVU,MP RVU\n"
+        "99213,,Office visit est,A,1.3,1.5,0.55,0.1\n"
+        "0001A,,Admin covid,X,0.0,0.0,0.0,0.0\n"
+    )
+    return p
+
+
+def test_load_rvu_proposed_loads_and_reloads(tmp_path, monkeypatch):
+    path = _fixture(tmp_path)
+    monkeypatch.setattr(nprm, "_discover_addendum_b", lambda: path)
+    con = duckdb.connect()
+    out = nprm.load_rvu_proposed(con)
+    assert out["rows"] == 2
+    row = con.execute(
+        "SELECT hcpcs, work_rvu, cms_rule_id FROM pfs.rvu_proposed WHERE hcpcs='99213'"
+    ).fetchone()
+    assert row == ("99213", 1.3, "CMS-1832-P")
+    nprm.load_rvu_proposed(con)  # idempotent delete-and-reload
+    assert con.execute("SELECT count(*) FROM pfs.rvu_proposed").fetchone()[0] == 2