diff --git a/fieldExtraction/crosswalk/README.md b/fieldExtraction/crosswalk/README.md new file mode 100644 index 0000000..7828d59 --- /dev/null +++ b/fieldExtraction/crosswalk/README.md @@ -0,0 +1,46 @@ +This framework allows flexible serialization and application of crosswalk tables. + +### Features +- Create crosswalks from Excel files or dictionaries +- Save crosswalks as JSON for easy reuse +- Apply crosswalks with customizable default behavior +- Metadata tracking for column mappings + +### Creating a Crosswalk +``` +from crosswalk_utils import CrosswalkBuilder + +# From Excel +builder = CrosswalkBuilder() +builder.from_excel( + path="CROSSWALK_PROVIDER_TYPE.20220106.xlsx", + sheet_name='Prac', + from_col="Practitioner Specialty", + to_col="Taxonomy" +).save('crosswalk_provider_type.json') + +# You can also create from a dictionary +builder.from_dict( + mapping={'Acupuncturist': '171100000X'}, + from_col='Specialty', + to_col='Taxonomy' +) +``` + +### Applying a Crosswalk +``` +import json +from crosswalk_utils import apply_crosswalk + +# Load the mapping +with open('crosswalk_provider_type.json') as f: + data = json.load(f) + +# Apply the crosswalk +result = apply_crosswalk( + val="Acupuncturist", + mapping=data['mapping'], + default=None # Optional: specify default for unmapped values +) +# Result: '171100000X' +``` diff --git a/fieldExtraction/crosswalk/crosswalk_utils.py b/fieldExtraction/crosswalk/crosswalk_utils.py new file mode 100644 index 0000000..de78e1f --- /dev/null +++ b/fieldExtraction/crosswalk/crosswalk_utils.py @@ -0,0 +1,107 @@ +import json + +import pandas as pd + + +class CrosswalkBuilder: + def __init__(self): + self.mapping: dict[str, str] = {} + self.metadata: dict[str, str | None] = {} + + def from_excel( + self, path: str, from_col: str, to_col: str, sheet_name: str | None = None + ) -> "CrosswalkBuilder": + """Load mapping from an excel file + + Args: + path (str): location for the excel file + from_col (str): Source column (for keys) + to_col (str): Destination column (for values) + sheet_name (str|None, optional): Sheet name in the excel file. Defaults to None. + + Returns: + CrosswalkBuilder: the builder instance for method chaining + """ + if sheet_name is None: + # If sheet_name is not provided, read the first sheet + df = pd.read_excel(path, sheet_name=0) + else: + df = pd.read_excel(path, sheet_name=sheet_name) + + if from_col not in df.columns: + raise ValueError(f"Column '{from_col}' not found in the excel file") + elif to_col not in df.columns: + raise ValueError(f"Column '{to_col}' not found in the excel file") + + self.metadata = { + "from_col": from_col, + "to_col": to_col, + } + + self.mapping = dict(zip(df[from_col], df[to_col])) + + return self + + def from_dict( + self, + mapping: dict[str, str], + from_col: str | None = None, + to_col: str | None = None, + ) -> "CrosswalkBuilder": + """Load mapping from a dictionary + + Args: + mapping (dict[str, str]): Mapping dictionary + + Returns: + CrosswalkBuilder: the builder instance for method chaining + """ + self.metadata = { + "from_col": from_col, + "to_col": to_col, + } + self.mapping = mapping + return self + + def from_json(self, path: str) -> "CrosswalkBuilder": + """Load mapping from a JSON file + + Args: + path (str): Path to the JSON file + + Returns: + CrosswalkBuilder: the builder instance for method chaining + """ + with open(path, "r") as f: + data = json.load(f) + + self.metadata = data.get("metadata", {}) + self.mapping = data.get("mapping", {}) + + return self + + def save(self, output_path: str) -> None: + """Save the crosswalk to JSON. + + Args: + output_path (str): Path to save the crosswalk + """ + output_dict = {"metadata": self.metadata, "mapping": self.mapping} + with open(output_path, "w") as f: + json.dump(output_dict, f, indent=4) + + +def apply_crosswalk( + val: str, mapping: dict[str, str], default: str | None = None +) -> str: + """Apply a crosswalk to a value + + Args: + val (str): Input value + mapping (dict[str, str]): Mapping dictionary + default (str | None, optional): Default value; if None and val is not found in the mapping, val is returned. Defaults to None. + + Returns: + str: Crosswalked value + """ + return mapping.get(val, default if default is not None else val) diff --git a/fieldExtraction/crosswalk/test_crosswalk_utils.py b/fieldExtraction/crosswalk/test_crosswalk_utils.py new file mode 100644 index 0000000..bc92c8d --- /dev/null +++ b/fieldExtraction/crosswalk/test_crosswalk_utils.py @@ -0,0 +1,91 @@ +import json + +import pandas as pd +import pytest + +from crosswalk_utils import CrosswalkBuilder, apply_crosswalk + + +@pytest.fixture +def sample_excel_file(tmp_path): + data = {"from_col": ["A", "B", "C"], "to_col": ["one", "two", "three"]} + df = pd.DataFrame(data) + file_path = tmp_path / "sample.xlsx" + df.to_excel(file_path, index=False) + return file_path + + +@pytest.fixture +def sample_json_file(tmp_path): + data = { + "metadata": {"from_col": "from_col", "to_col": "to_col"}, + "mapping": {"A": "1", "B": "2", "C": "3"}, + } + file_path = tmp_path / "sample.json" + with open(file_path, "w") as f: + json.dump(data, f) + return file_path + + +def test_from_excel(sample_excel_file): + builder = CrosswalkBuilder() + builder.from_excel(sample_excel_file, "from_col", "to_col") + assert builder.mapping == {"A": "one", "B": "two", "C": "three"} + assert builder.metadata == {"from_col": "from_col", "to_col": "to_col"} + + +def test_from_excel_missing_column(sample_excel_file): + builder = CrosswalkBuilder() + with pytest.raises(ValueError, match="Column 'nonexistent' not found"): + builder.from_excel(sample_excel_file, "nonexistent", "to_col") + + +def test_from_dict(): + mapping = {"A": "1", "B": "2", "C": "3"} + builder = CrosswalkBuilder() + builder.from_dict(mapping) + assert builder.mapping == mapping + assert builder.metadata == {"from_col": None, "to_col": None} + + +def test_from_json(sample_json_file): + builder = CrosswalkBuilder() + builder.from_json(sample_json_file) + assert builder.mapping == {"A": "1", "B": "2", "C": "3"} + assert builder.metadata == {"from_col": "from_col", "to_col": "to_col"} + + +def test_json_roundtrip(tmp_path): + # Create and save a crosswalk + original = CrosswalkBuilder() + original.from_dict({"A": "1"}, "source", "target") + json_path = tmp_path / "test.json" + original.save(json_path) + + # Load it back + loaded = CrosswalkBuilder().from_json(json_path) + + # Check everything matches + assert loaded.mapping == original.mapping + assert loaded.metadata == original.metadata + + +def test_save(tmp_path): + mapping = {"A": "1", "B": "2", "C": "3"} + builder = CrosswalkBuilder() + builder.from_dict(mapping, "from_col", "to_col") + output_path = tmp_path / "output.json" + builder.save(output_path) + + with open(output_path, "r") as f: + data = json.load(f) + + assert data["mapping"] == mapping + assert data["metadata"] == {"from_col": "from_col", "to_col": "to_col"} + + +def test_apply_crosswalk(): + mapping = {"A": "1", "B": "2", "C": "3"} + assert apply_crosswalk("A", mapping) == "1" + assert apply_crosswalk("D", mapping) == "D" + assert apply_crosswalk("D", mapping, default="0") == "0"