stack/dev/scripts/gen_bcda_v3_crosswalk.py

"""Flatten the BCDA v3 Data Dictionary into crosswalk seed CSVs.

Reads the vendored ``dev/vendor/bcda-site/BCDA_v3_Data_Dictionary.xlsx``
(synced by ``vendor_bcda_app.py``) and writes three seeds that
``stack load seed`` lands in ``reference_data``:

- ``dev/seeds/bcda_v3_data_dictionary.csv`` — one row per v3 field:
  resource, claim type, field name, fhirPath, IDR ``sourceColumn``,
  ``ccwMapping`` (the CCW/v2 variable the ``bcda.*`` table columns are
  named after), ``cclfMapping`` (CCLF file.element), profiles.  This is
  the v2→v3 crosswalk the flatteners in ``bcda.express.v3`` follow.
- ``dev/seeds/bcda_v3_cclf_crosswalk.csv`` — the ``CCLF-BCDA v3
  Mapping`` sheet: every CCLF element with the v3 field + fhirPath that
  replaces it (``#N/A`` → empty).
- ``dev/seeds/bcda_v3_cclf_absent.csv`` — the CCLF elements BCDA v3
  does not carry (HICNs, FIPS county, address lines, BETOS, …).

Idempotent; rerun after ``--sync`` moves the dictionary.

    uv run python dev/scripts/gen_bcda_v3_crosswalk.py
"""

from __future__ import annotations

import csv
import re
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
XLSX = ROOT / "dev" / "vendor" / "bcda-site" / "BCDA_v3_Data_Dictionary.xlsx"
SEEDS = ROOT / "dev" / "seeds"

_NA = {None, "", "#N/A", "None"}


def _clean(v: object) -> str:
    if v in _NA:
        return ""
    return " ".join(str(v).split())


def _list(v: object) -> str:
    """``"['A', 'B']"`` → ``"A;B"`` (the sheet stores python-ish lists)."""
    s = _clean(v)
    if s.startswith("[") and s.endswith("]"):
        inner = s[1:-1]
        return ";".join(p.strip().strip("'\"") for p in inner.split(",") if p.strip())
    return s


def dictionary_rows(wb) -> list[dict]:
    ws = wb["Data Dictionary"]
    rows = ws.iter_rows(values_only=True)
    header = [str(h) for h in next(rows)]
    out = []
    for r in rows:
        d = dict(zip(header, r, strict=False))
        if not _clean(d.get("Field Name")):
            continue
        out.append(
            {
                "row": _clean(d.get("Row")),
                "fhir_resource": _clean(d.get("FHIR Resource")),
                "claim_type": _list(d.get("Coverage / Claim Type")),
                "field_name": _clean(d.get("Field Name")),
                "description": _clean(d.get("Description")),
                "fhir_path": _clean(d.get("fhirPath")),
                "example": _clean(d.get("example")),
                "source_view": _clean(d.get("sourceView")),
                "source_column": _clean(d.get("sourceColumn")),
                "sources": _list(d.get("sources")),
                "reference_table": _clean(d.get("referenceTable")),
                "cclf_mapping": _list(d.get("cclfMapping")),
                "ccw_mapping": _list(d.get("ccwMapping")),
                "profiles": _list(d.get("profiles")),
            }
        )
    return out


_IDENT = re.compile(r"^[A-Z0-9_]+$")


def _label_name(label: object, name: object) -> tuple[str, str]:
    """(prose label, UPPER_SNAKE variable).

    The sheet header says ``Field Label`` then ``Field Name`` but the
    data rows carry the CCLF variable (``CLM_CNTL_NUM``) under *Label*
    and the prose under *Name* — except the CCLF0 header rows, which
    are the other way round.  Normalise by shape so ``field_name`` is
    always the variable.
    """
    a, b = _clean(label), _clean(name)
    if _IDENT.match(a) and not _IDENT.match(b):
        return b, a
    return a, b


def cclf_rows(wb) -> list[dict]:
    ws = wb["CCLF-BCDA v3 Mapping"]
    rows = ws.iter_rows(values_only=True)
    next(rows)
    out = []
    for r in rows:
        if not _clean(r[0]):
            continue
        label, name = _label_name(r[2], r[3])
        out.append(
            {
                "cclf_file": _clean(r[0]),
                "element": _clean(r[1]),
                "field_label": label,
                "field_name": name,
                "start_position": _clean(r[4]),
                "end_position": _clean(r[5]),
                "data_length": _clean(r[6]),
                "format": _clean(r[7]),
                "field_description": _clean(r[8]),
                "bcda_v3_field": _clean(r[12]),
                "fhir_path": _clean(r[13]),
            }
        )
    return out


def absent_rows(wb) -> list[dict]:
    ws = wb["CCLF fields absent from BCDA v3"]
    rows = ws.iter_rows(values_only=True)
    next(rows)
    out = []
    for r in rows:
        if r[1] in _NA:
            continue
        out.append(
            {
                "cclf_file": _clean(r[0]),
                "element": _clean(r[1]),
                "field_name": _clean(r[2]),
                "field_description": _clean(r[3]),
            }
        )
    return out


def _write(path: Path, rows: list[dict]) -> int:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
        w.writeheader()
        w.writerows(rows)
    return len(rows)


def main(xlsx: Path = XLSX, seeds: Path = SEEDS) -> dict[str, int]:
    import openpyxl

    # data_only: the CCLF sheet is XLOOKUP formulas; we want their cached
    # results, not the formula text.
    wb = openpyxl.load_workbook(xlsx, read_only=True, data_only=True)
    counts = {
        "bcda_v3_data_dictionary": _write(
            seeds / "bcda_v3_data_dictionary.csv", dictionary_rows(wb)
        ),
        "bcda_v3_cclf_crosswalk": _write(
            seeds / "bcda_v3_cclf_crosswalk.csv", cclf_rows(wb)
        ),
        "bcda_v3_cclf_absent": _write(
            seeds / "bcda_v3_cclf_absent.csv", absent_rows(wb)
        ),
    }
    return counts


if __name__ == "__main__":  # pragma: no cover
    for name, n in main().items():
        print(f"{name}: {n} rows")
    sys.exit(0)