120 lines
3.9 KiB
Python
120 lines
3.9 KiB
Python
"""Ingest CMS PFS reference data into DuckDB, then replica + lake.
|
|
|
|
Wraps ``pfs.pipe.load_all`` (RVU / GPCI / carrier locality / labor /
|
|
equipment / supply files, sourced from Zotero attachments) with the
|
|
standard ingest plumbing: lock-preflighted write connection (#508),
|
|
read-replica refresh (#510), and DuckLake publish (#514). PFS loads
|
|
were previously ad-hoc ``load_all`` calls with none of that.
|
|
|
|
Usage:
|
|
uv run python dev/scripts/ingest_pfs.py
|
|
uv run python dev/scripts/ingest_pfs.py --years 2020 2026 --no-lake
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--years",
|
|
nargs=2,
|
|
type=int,
|
|
metavar=("FROM", "TO"),
|
|
help="inclusive year range (default: all available)",
|
|
)
|
|
parser.add_argument(
|
|
"--no-lake",
|
|
action="store_true",
|
|
help="skip publishing to the DuckLake lakehouse",
|
|
)
|
|
parser.add_argument(
|
|
"--nprm",
|
|
action="store_true",
|
|
help=(
|
|
"also load every registered NPRM Addendum B proposed-RVU "
|
|
"partition (pfs.nprm.NPRM_SOURCES) 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
|
|
|
|
run_id = ingest_log.new_run_id()
|
|
|
|
print("Ingesting CMS PFS files into DuckDB ...")
|
|
with duckdb_batch("aco") as con:
|
|
summary = load_all(con, years=tuple(args.years) if args.years else None)
|
|
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 (
|
|
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:
|
|
nprm_out = load_rvu_proposed(con, year=year)
|
|
print(f" pfs.rvu_proposed [{cms_rule_id}]: {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=nprm_out["cms_rule_id"],
|
|
fr_citation=nprm_out["fr_citation"],
|
|
pincite_key=nprm_out["pincite_key"],
|
|
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}")
|
|
|
|
if not args.no_lake:
|
|
print("\n--- Publishing to DuckLake ---")
|
|
import _lake
|
|
|
|
_lake.publish_lake(("pfs", "cms"))
|
|
|
|
print("\nDone.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|