Merged in bugfix/proc-code-lists (pull request #403)

Bugfix/proc code lists * Add admit type code * add proc mod description * Add bill type pos mapping * Merged main into feature/proc-desc * Uncomment test code * Bugfix * script optimisation for codes * modified is empty function in string utils * Remove test.py * bug fix to handle list of proc_codes * Merged main into bugfix/proc-code-lists * changed any method to all method in list handling * code mapping moved outside to run once * Merge branch 'feature/daip2-97' into bugfix/proc-code-lists * merge conflict fixed * merge conflicts fixed * incorporated latest changes from main * Merged main into bugfix/proc-code-lists * Move load all dataset to io_utils * Fix list is_empty * Updated poetry.lock * Docstring * Docstrings * Merge branch 'main' into bugfix/proc-code-lists * bug fix Approved-by: Mayank Aamseek Approved-by: Alex Galarce
2025-02-21 14:44:36 +00:00
parent 87223cc3b8
commit 92029cdabe
8 changed files with 1161 additions and 1062 deletions
@@ -139,3 +139,5 @@ VALID_CARVEOUTS = {
    "READMISSIONS" : "Describes payment for subsequent admission with the same diagnostic category of codes as the initial admission.",    
 }

+levels = ["cpt", "hcpcs", "cpt_level3", "cpt_level2", "hcpcs_level2", "cpt_level1", "hcpcs_level1"]
+
@@ -21,12 +21,34 @@ from sentence_transformers import SentenceTransformer
 # Load model
 model = SentenceTransformer("all-roberta-large-v1")
 def create_faiss_index(choices):
+    """
+    Creates a FAISS index for a list of choices using cosine similarity (inner product).
+
+    Args:
+        choices (list): A list of text items to index.
+
+    Returns:
+        faiss.IndexFlatIP: A FAISS index for the given embeddings.
+    """
    embeddings = model.encode(choices, normalize_embeddings=True).astype("float32")
    index = faiss.IndexFlatIP(embeddings.shape[1])  # Inner product for cosine similarity
    index.add(embeddings)
    return index

 def get_matches_above_threshold(target, choices, index, threshold=0.7, top_k=5):
+    """
+    Retrieves the top-k matches from a FAISS index with similarity scores above a given threshold.
+
+    Args:
+        target (str): The target text to compare against the choices.
+        choices (list): A list of choices to match the target with.
+        index (faiss.IndexFlatIP): The FAISS index for fast nearest-neighbor search.
+        threshold (float, optional): The minimum similarity score to consider a match. Defaults to 0.7.
+        top_k (int, optional): The number of top matches to retrieve. Defaults to 5.
+
+    Returns:
+        list: A list of tuples containing matched choice and score for each match above the threshold.
+    """
    target_vec = model.encode([target], normalize_embeddings=True).astype("float32")
    similarity_scores, match_indices = index.search(target_vec, top_k)
    return [
@@ -35,8 +57,39 @@ def get_matches_above_threshold(target, choices, index, threshold=0.7, top_k=5):
        if score >= threshold
    ]

+def crosswalk_levels():
+    """
+    Loads and processes crosswalk mappings for different levels (e.g., CPT, HCPCS) from CSV files, 
+    and returns a dictionary with clean-to-description and description-to-code mappings for each level.
+
+    Returns:
+        dict: A dictionary where each key is a level, and the value is a tuple containing:
+              - clean-to-description mapping (str -> str)
+              - description-to-code mapping (str -> str)
+    """
+    # Get necessary mappings
+    crosswalk_dict = {}
+    for level in investment_values.levels:
+        mapping_df = pd.read_csv(f"crosswalk/mapping_csvs/proc_cd/{level}.csv", dtype=str)
+        level_crosswalk = CrosswalkBuilder().from_df(mapping_df, from_col="Code", to_col="Description")
+        clean_to_description = {get_clean_value(x) : x for x in level_crosswalk.mapping.values()}
+        description_to_code = level_crosswalk.create_reverse_mapping()
+        crosswalk_dict[level] = (clean_to_description, description_to_code)
+
+    return crosswalk_dict  

 def code_primary(answer_dicts, filename):
+    """
+    Breaks down the direct code information from CONTRACT_SERVICE_CD_OR_DESC and CONTRACT_REIMBURSEMENT_METHOD fields
+    using an LLM prompt and adds the results to the provided answer dictionaries.
+
+    Args:
+        answer_dicts (list of dict): A list of dictionaries containing service and reimbursement data.
+        filename (str): The name of the file being processed, used for LLM processing.
+
+    Returns:
+        list of dict: The input answer dictionaries with additional primary code breakout information. These are only the codes that are explicitly written in the Service/Methodology
+    """
    code_primary_questions = FieldSet(file_path=config.FIELD_JSON_PATH, field_type="code_primary_breakout").get_prompt_dict()
    code_breakout_answers = []
    for answer_dict in answer_dicts:
@@ -55,6 +108,12 @@ def code_primary(answer_dicts, filename):
    return code_breakout_answers

 def get_proc_crosswalk():
+    """
+    Loads and concatenates all CSV files in the 'proc_cd' directory, and creates a crosswalk mapping from the 'Code' to 'Description' columns.
+
+    Returns:
+        CrosswalkBuilder: A CrosswalkBuilder object containing the mapping from procedure codes to descriptions.
+    """
    all_mappings = []
    mapping_dir = "crosswalk/mapping_csvs/proc_cd"
    for filename in os.listdir(mapping_dir):
@@ -66,6 +125,15 @@ def get_proc_crosswalk():


 def get_clean_value(service_description):
+    """
+    Cleans a service description by removing stop words, replacing certain terms, and standardizing acronyms.
+
+    Args:
+        service_description (str): The service description string to clean.
+
+    Returns:
+        str: The cleaned service description with stop words removed, specific terms replaced, and acronyms expanded.
+    """
    stop_words = {"the", "is", "in", "and", "to", "of", "a", "an", "that", "this", "it", "for", "on", "with", "as", "was", "were", "at", "by", "but", "be"}  # Experiment with not removing 'and'
    def remove_stop_words(text):
        words = text.split()
@@ -83,23 +151,27 @@ def get_clean_value(service_description):
    return service_clean


-def get_best_match(target, model, thresholds={
+def get_best_match(crosswalk_dict, target, model, thresholds={
    "cpt": 0.85, "cpt_level3": 0.8, "cpt_level2": 0.75, "cpt_level1": 0.7,
    "hcpcs": 0.85, "hcpcs_level3": 0.8, "hcpcs_level2": 0.75, "hcpcs_level1": 0.7
 }):
-    levels = ["cpt", "hcpcs", "cpt_level3", "cpt_level2", "hcpcs_level2", "cpt_level1", "hcpcs_level1"]
+    """
+    Finds the best matching code from a crosswalk dictionary by comparing a target string to precomputed embeddings
+    using a model and similarity threshold values for different levels.

-    # Get necessary mappings
-    crosswalk_dict = {}
-    for level in levels:
-        mapping_df = pd.read_csv(f"crosswalk/mapping_csvs/proc_cd/{level}.csv", dtype=str)
-        level_crosswalk = CrosswalkBuilder().from_df(mapping_df, from_col="Code", to_col="Description")
-        clean_to_description = {get_clean_value(x) : x for x in level_crosswalk.mapping.values()}
-        description_to_code = level_crosswalk.create_reverse_mapping()
-        crosswalk_dict[level] = (clean_to_description, description_to_code)
+    Args:
+        crosswalk_dict (dict): Dictionary containing the crosswalk data for different levels.
+        target (str): The target string to match.
+        model: The model used to encode the target string into an embedding.
+        thresholds (dict, optional): A dictionary of similarity thresholds for each level (default values provided).
+
+    Returns:
+        tuple: A tuple containing the best matching code, its description, and the similarity score.
+               If no match is found above the threshold, returns (None, None, 0).
+    """
    
    target_vec = model.encode([target], normalize_embeddings=True).astype("float32")
-    for level in levels:
+    for level in investment_values.levels:
        index, _, choices = embedding_utils.load_faiss_index(index_path=os.path.join("embeddings", level, "faiss_index.bin"),
                                                             embedding_path=os.path.join("embeddings", level, "embeddings.npy"),
                                                             choices_path=os.path.join("embeddings", level, "choices.pkl"),)
@@ -113,7 +185,7 @@ def get_best_match(target, model, thresholds={
    
    return None, None, 0 # No valid match found

-def code_indirect(answer_dicts):
+def code_indirect(answer_dicts, crosswalk):
    """
    Maps free-form service descriptions to the best-matching code and description. 
    Caches results for efficiency and skips entries with existing codes or irrelevant text.
@@ -135,7 +207,7 @@ def code_indirect(answer_dicts):
        if (not service                                     # Service DNE
            or "covered service" in service.lower()         # Is generic
            or isinstance(proc_code, list)                  # Is a list
-            or string_utils.is_empty(proc_code)             # Is empty
+            or not string_utils.is_empty(proc_code)         # Is not empty
            ):
            pass # pass for now - we may need to add functionality here later
        else:
@@ -143,7 +215,7 @@ def code_indirect(answer_dicts):
                code = already_seen[service]['code']
                description = already_seen[service]['description']
            else:               
-                code, description, score = get_best_match(service, model)
+                code, description, score = get_best_match(crosswalk, service, model)
                # print(service, " | ", description, " | ", code, " | ", score)

            answer_dict["CPT4_PROC_CD"] = code
@@ -152,6 +224,17 @@ def code_indirect(answer_dicts):
    return results

 def get_code_description(code, code_mapping):
+    """
+    Retrieves descriptions for a given code or list of codes using the provided code mapping.
+    If no description is found, attempts to resolve by modifying the code (e.g., prepending "0" or handling code patterns).
+
+    Args:
+        code (str, list): A single code or a list of codes to look up.
+        code_mapping (Crosswalk): The mapping object containing code-to-description mappings.
+
+    Returns:
+        str: A comma-separated string of descriptions for the provided codes.
+    """
    if isinstance(code, str):
        if string_utils.is_empty(code):
            return "N/A"
@@ -181,28 +264,57 @@ def get_code_description(code, code_mapping):

    return ",".join(code_description)

-def code_description_mapping(answer_dicts):
-    # Get mappings
+def get_mappings():
+    """
+    Loads and processes multiple CSV and Excel files containing code mappings for various categories
+    (e.g., PROC, DIAG, Grouper, Rev, Admit Type), and returns a dictionary of crosswalks for each code type.
+
+    Returns:
+        dict: A dictionary containing mappings for PROC, DIAG, Grouper, Rev, and Admit Type codes.
+    """
+
+    all_mappings = {}
+
    # PROC codes
    mapping_path = "crosswalk/mapping_csvs/"
    all_proc_mappings = pd.concat([pd.read_csv(os.path.join(mapping_path, "proc_cd", file), dtype=str) for file in os.listdir(os.path.join(mapping_path, "proc_cd")) if file.endswith(".csv")])
    proc_mapping = CrosswalkBuilder().from_df(all_proc_mappings, from_col="Code", to_col="Description")
+    all_mappings['proc_mapping'] = proc_mapping

    # Diag codes
    all_diag_mappings = pd.concat([pd.read_csv(os.path.join(mapping_path, "diag_cd", file), dtype=str) for file in os.listdir(os.path.join(mapping_path, "diag_cd")) if file.endswith(".csv")])
    diag_mapping = CrosswalkBuilder().from_df(all_diag_mappings, from_col="Code", to_col="Description")
+    all_mappings['diag_mapping'] = diag_mapping
    
    # Grouper codes
    all_grouper_mappings = pd.concat([pd.read_csv(os.path.join(mapping_path, "grouper_cd", file), dtype=str) for file in os.listdir(os.path.join(mapping_path, "grouper_cd")) if file.endswith(".csv")])
    grouper_mapping = CrosswalkBuilder().from_df(all_grouper_mappings, from_col="Code", to_col="Description")
    grouper_version_mapping = CrosswalkBuilder().from_df(all_grouper_mappings, from_col="Code", to_col="Version")
+    all_mappings['grouper_mapping'] = grouper_mapping
+    all_mappings['grouper_version_mapping'] = grouper_version_mapping

    # Rev codes
    rev_mapping = CrosswalkBuilder().from_excel(path=os.path.join(mapping_path, "rev_cd", "rev_mapping.csv"), from_col="Code", to_col="Description")
+    all_mappings['rev_mapping'] = rev_mapping

    # Admit Type codes
    admit_mapping = CrosswalkBuilder().from_excel(path=os.path.join(mapping_path, "admit_type_cd", "admit_mapping.csv"), from_col="Code", to_col="Description")
+    all_mappings['admit_mapping'] = admit_mapping

+    return all_mappings
+
+
+def code_description_mapping(answer_dicts, all_mappings):
+    """
+    Maps code descriptions to relevant fields in the answer dictionaries using provided mappings.
+
+    Args:
+        answer_dicts (list): List of dictionaries containing codes to be mapped.
+        all_mappings (dict): Dictionary containing the mappings for different code types.
+
+    Returns:
+        list: Updated list of dictionaries with code descriptions added.
+    """
    for answer_dict in answer_dicts:
        proc_code = answer_dict.get("CPT4_PROC_CD")
        proc_mod = answer_dict.get("CPT4_PROC_MOD")
@@ -211,25 +323,36 @@ def code_description_mapping(answer_dicts):
        grouper_code = answer_dict.get("FACILITY_GROUPER_CD")
        admit_type_code = answer_dict.get("CLAIM_ADMIT_TYPE_CD")
        
-        answer_dict["CPT4_PROC_DESC"] = get_code_description(proc_code, proc_mapping) 
-        answer_dict["CPT4_PROC_MOD_DESC"] = get_code_description(proc_mod, proc_mapping)
-        answer_dict["DIAG_CD_DESC"] = get_code_description(diag_code, diag_mapping)
-        answer_dict["REVENUE_CD_DESC"] = get_code_description(rev_code, rev_mapping)
-        answer_dict["FACILITY_GROUPER_DESC"] = get_code_description(grouper_code, grouper_mapping)
-        answer_dict["GROUPER_TYPE"] = get_code_description(grouper_code, grouper_version_mapping)
-        answer_dict["AUTH_ADMIT_TYPE_DESC"] = get_code_description(admit_type_code, admit_mapping)
+        answer_dict["CPT4_PROC_DESC"] = get_code_description(proc_code, all_mappings['proc_mapping'])
+        answer_dict["CPT4_PROC_MOD_DESC"] = get_code_description(proc_mod, all_mappings['proc_mapping'])
+        answer_dict["DIAG_CD_DESC"] = get_code_description(diag_code, all_mappings['diag_mapping'])
+        answer_dict["REVENUE_CD_DESC"] = get_code_description(rev_code, all_mappings['rev_mapping'])
+        answer_dict["FACILITY_GROUPER_DESC"] = get_code_description(grouper_code, all_mappings['grouper_mapping'])
+        answer_dict["GROUPER_TYPE"] = get_code_description(grouper_code, all_mappings['grouper_version_mapping'])
+        answer_dict["AUTH_ADMIT_TYPE_DESC"] = get_code_description(admit_type_code, all_mappings['admit_mapping'])
        
    return answer_dicts

-def get_code_breakout(answer_dicts: list[dict], filename: str):
+def get_code_breakout(answer_dicts: list[dict], filename: str, all_dataset: dict):
+    """
+    Extracts and maps codes from contract data, including primary, indirect, and description mappings.
+
+    Args:
+        answer_dicts (list[dict]): List of dictionaries containing contract-related answers.
+        filename (str): The name of the file being processed.
+        all_dataset (dict): Dictionary containing crosswalk and mappings for code translation.
+
+    Returns:
+        dict: A dictionary with primary, indirect/derived, and mapped code descriptions.
+    """

    # Code Primary (Pull codes directly from CONTRACT_SERVICE_CD_OR_DESC and CONTRACT_REIMBURSEMENT_METHOD)
    code_primary_answers = code_primary(answer_dicts, filename)

    # Only used for Implicit. Code Indirect (Derive codes indirectly from CONTRACT_SERVICE_CD_OR_DESC and CONTRACT_REIMBURSEMENT_METHOD)
-    code_indirect_answers = code_indirect(code_primary_answers)
+    code_indirect_answers = code_indirect(code_primary_answers, all_dataset['codes_crosswalk'])

    # Code Descriptions (Pull standardized descriptions from codes)
-    code_description_answers = code_description_mapping(code_indirect_answers)
+    code_description_answers = code_description_mapping(code_indirect_answers, all_dataset['codes_mappings'])
    
    return code_description_answers
@@ -20,7 +20,8 @@ from datetime import datetime
 def datetime_str():
    return datetime.now().strftime("[%Y-%m-%d %H:%M:%S]")

-def process_file(file_object, run_timestamp):
+
+def process_file(file_object, all_dataset, run_timestamp):
    filename, contract_text = file_object
    print(f"{datetime_str()} Processing {filename}...")

@@ -38,7 +39,7 @@ def process_file(file_object, run_timestamp):

    ################## ONE TO N ##################
    if string_utils.contains_reimbursement(contract_text):
-        one_to_n_results = run_one_to_n_prompts(filename, text_dict, exhibit_pages, exhibit_chunk_mapping) # Return df
+        one_to_n_results = run_one_to_n_prompts(filename, text_dict, exhibit_pages, exhibit_chunk_mapping, all_dataset) # Return df
        one_to_n_results['CONTRACT_FILE_NAME'] = filename
        contract_results = pd.merge(one_to_one_results, one_to_n_results, how='outer', on='CONTRACT_FILE_NAME')
    else:
@@ -90,7 +91,7 @@ def run_one_to_one_prompts(filename, contract_text, text_dict, top_sheet_dict):
    return final_df


-def run_one_to_n_prompts(filename, text_dict, exhibit_pages, exhibit_chunk_mapping):
+def run_one_to_n_prompts(filename, text_dict, exhibit_pages, exhibit_chunk_mapping, all_dataset):

    ################## RUN PROMPTS ##################
    one_to_n_results = []
@@ -111,7 +112,7 @@ def run_one_to_n_prompts(filename, text_dict, exhibit_pages, exhibit_chunk_mappi
            exhibit_level_answers['EXHIBIT_PAGE'] = exhibit_page

            ################## GET REIMBURSEMENT-LEVEL ANSWERS (INCLUDING DYNAMIC) ##################
-            reimbursement_level_answers = reimbursement_level(exhibit_chunk, filename, reimbursement_level_fields) # Return list of dictionaries
+            reimbursement_level_answers = reimbursement_level(exhibit_chunk, filename, reimbursement_level_fields, all_dataset) # Return list of dictionaries
            
            ################## COMBINE ANSWERS ##################
            full_answer_dict = combine_one_to_n_answers(exhibit_level_answers, reimbursement_level_answers, tin_npi_answers)
@@ -121,3 +122,6 @@ def run_one_to_n_prompts(filename, text_dict, exhibit_pages, exhibit_chunk_mappi
    ################## CONVERT TO DF ##################
    one_to_n_df = pd.DataFrame(one_to_n_results)
    return one_to_n_df
+
+
+
@@ -30,11 +30,14 @@ def main(testing=False, test_params = {}):
            input_dict = {k: v for k, v in input_dict.items() if k in test_params['input_files']}
    
    print(f"Total Input Files : {len(input_dict)}")
+
+    # load all crosswalk files and mappings
+    all_dataset = io_utils.load_all_dataset()
    
    # Process files with run_timestamp
    all_results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
-        results = executor.map(lambda item: file_processing.process_file(item, run_timestamp), input_dict.items())
+        results = executor.map(lambda item: file_processing.process_file(item, all_dataset, run_timestamp), input_dict.items())
    all_results.extend(results)
    
    FINAL_RESULT_DF = pd.concat(all_results)
@@ -263,9 +263,9 @@ def get_service_breakout(answer_dicts, filename):
                    filename
                )
            )
-            already_seen[service_methodology] = service_breakout_answer
+            already_seen[service_methodology] = service_breakout_dict
        else:
-            service_breakout_answer = already_seen[service_methodology]
+            service_breakout_dict = already_seen[service_methodology]
        
        service_breakout_answers.append({**answer_dict, **service_breakout_dict})
    
@@ -330,7 +330,7 @@ def combine_one_to_n_answers(exhibit_level_answers: dict, reimbursement_level_an
    return combined_answers


-def reimbursement_level(exhibit_text, filename, reimbursement_level_fields):
+def reimbursement_level(exhibit_text, filename, reimbursement_level_fields, all_dataset):
    """
    Processes reimbursement-level fields.
    Currently, runs REIMBURSEMENT_LEVEL_PRIMARY prompt only. Future secondary prompts will go in this function
@@ -349,6 +349,6 @@ def reimbursement_level(exhibit_text, filename, reimbursement_level_fields):
    # Later, we will only run methodology breakout for those without certain carveout indicators. We will run carveout-specific breakouts for the others
    methodology_breakout_answers = get_methodology_breakout(carveout_answers, filename)
    service_breakout_answers = get_service_breakout(methodology_breakout_answers, filename)
-    code_breakout_answers = code_funcs.get_code_breakout(service_breakout_answers, filename)
+    code_breakout_answers = code_funcs.get_code_breakout(service_breakout_answers, filename, all_dataset)

    return code_breakout_answers
@@ -7,6 +7,7 @@ from pyxlsb import open_workbook

 import src.tracking.tracking_utils as tracking_utils
 from src import config
+from src.investment.code_funcs import crosswalk_levels, get_mappings


 def read_local(file_path): # io_utils.py
@@ -258,4 +259,15 @@ def write_s3(df, filename, run_timestamp, output_type):
    if output_type == "final":
        print(f"Saved batch output file: {output_path}")
    elif output_type == "individual":
-        print(f"Saved individual output file: {output_path}")
+        print(f"Saved individual output file: {output_path}")
+
+
+def load_all_dataset():
+    """_summary_: loads all csv files required for processing contracts including crosswalk files
+    and mappings (embeddings, past results may be added later)
+    """
+    all_dataset = {}
+    all_dataset['codes_crosswalk'] = crosswalk_levels()
+    all_dataset['codes_mappings'] = get_mappings()
+
+    return all_dataset
@@ -289,6 +289,9 @@ def is_empty(value):
    Returns:
        bool: True if the value is empty or invalid, False otherwise.
    """
+    if isinstance(value, list):  # Handle list inputs
+        return not value or all(is_empty(v) for v in value)
+        
    if pd.isna(value):
        return True
    else: