diff --git a/src/codes/code_funcs.py b/src/codes/code_funcs.py index e4da13d..a7a1162 100644 --- a/src/codes/code_funcs.py +++ b/src/codes/code_funcs.py @@ -235,7 +235,7 @@ def code_implicit_special(service: str, filename: str) -> dict: def get_embedding_levels( level_suffix: int, implicit_run_dict: dict, constants: Constants -) -> tuple[list, dict, dict, dict]: +) -> tuple[list, dict, dict]: """ Retrieves the embedding levels and their corresponding mappings for a given level suffix. @@ -246,7 +246,7 @@ def get_embedding_levels( constants: Application constants with mapping data. Returns: - (embedding_levels, cpt_mapping, hcpcs_mapping, rev_mapping). + (embedding_levels, cpt_mapping, hcpcs_mapping). Raises: ValueError: If level_suffix is not 1 or 2. @@ -254,20 +254,16 @@ def get_embedding_levels( if level_suffix == 1: cpt_mapping = constants.CPT_LEVEL1_MAPPING hcpcs_mapping = constants.HCPCS_LEVEL1_MAPPING - rev_mapping = constants.REV_LEVEL1_MAPPING embedding_levels = [] if implicit_run_dict["PROCEDURE_CD"]: embedding_levels += [ f"cpt_level{level_suffix}", f"hcpcs_level{level_suffix}", ] - if implicit_run_dict["REVENUE_CD"]: - embedding_levels += [f"rev_level{level_suffix}"] elif level_suffix == 2: cpt_mapping = constants.CPT_LEVEL2_MAPPING hcpcs_mapping = constants.HCPCS_LEVEL2_MAPPING - rev_mapping = constants.REV_MAPPING embedding_levels = [] if implicit_run_dict["PROCEDURE_CD"]: @@ -275,13 +271,11 @@ def get_embedding_levels( f"cpt_level{level_suffix}", f"hcpcs_level{level_suffix}", ] - if implicit_run_dict["REVENUE_CD"]: - embedding_levels += [f"rev"] else: raise ValueError(f"Unsupported level_suffix: {level_suffix}") - return embedding_levels, cpt_mapping, hcpcs_mapping, rev_mapping + return embedding_levels, cpt_mapping, hcpcs_mapping def get_match_list( @@ -341,13 +335,13 @@ def code_implicit_rag( Returns: List of {"source": "Level 1"|"Level 2", "code_answer_dict": dict}. - Only includes entries that have PROCEDURE_CD or REVENUE_CD (no placeholder-only). + Only includes entries that have PROCEDURE_CD (no placeholder-only). """ # Get embeddings, mappings, and best matches for Levels 1 and 2. level_dicts: list[dict] = [] for level_suffix in [1, 2]: - embedding_levels, cpt_mapping, hcpcs_mapping, rev_mapping = ( - get_embedding_levels(level_suffix, implicit_run_dict, constants) + embedding_levels, cpt_mapping, hcpcs_mapping = get_embedding_levels( + level_suffix, implicit_run_dict, constants ) if not embedding_levels: return [] @@ -362,7 +356,6 @@ def code_implicit_rag( "highest_similarity": highest_similarity, "cpt_mapping": cpt_mapping, "hcpcs_mapping": hcpcs_mapping, - "rev_mapping": rev_mapping, } ) @@ -372,7 +365,6 @@ def code_implicit_rag( for level_dict in level_dicts: cpt_mapping = level_dict["cpt_mapping"] hcpcs_mapping = level_dict["hcpcs_mapping"] - rev_mapping = level_dict["rev_mapping"] prompt, _parser = prompt_templates.CODE_IMPLICIT( service, level_dict["match_list"] @@ -401,9 +393,7 @@ def code_implicit_rag( code_answer_dict: dict = {} proc_codes: list[str] = [] - rev_codes: list[str] = [] proc_descs: list[str] = [] - rev_descs: list[str] = [] for description in llm_answer_final: if description in cpt_mapping.values(): matching_codes = [ @@ -419,18 +409,6 @@ def code_implicit_rag( if matching_codes: proc_codes.extend(matching_codes) proc_descs.append(description) - if description in rev_mapping.values(): - matching_codes = [ - str(key) for key, val in rev_mapping.items() if val == description - ] - if matching_codes: - # REV_LEVEL1_MAPPING keys are pipe-delimited (e.g. "0810|0811|0812"). - for code in matching_codes: - if "|" in code: - rev_codes.extend(code.split("|")) - else: - rev_codes.append(code) - rev_descs.append(description) if proc_codes: code_answer_dict["PROCEDURE_CD"] = proc_codes @@ -438,12 +416,6 @@ def code_implicit_rag( code_answer_dict["CODE_METHODOLOGY"] = ( f"Implicit - Level {level_dict['level_suffix']}" ) - if rev_codes: - code_answer_dict["REVENUE_CD"] = rev_codes - code_answer_dict["REVENUE_CD_DESC"] = rev_descs - code_answer_dict["CODE_METHODOLOGY"] = ( - f"Implicit - Level {level_dict['level_suffix']}" - ) if code_answer_dict: source_label = f"Level {level_dict['level_suffix']}" @@ -459,7 +431,7 @@ def _format_implicit_candidate_for_prompt(candidate: dict, index: int) -> str: source = candidate.get("source", "Unknown") code_dict = candidate.get("code_answer_dict", {}) parts = [f"Candidate {index} (Source: {source})"] - for key in ["PROCEDURE_CD", "PROCEDURE_CD_DESC", "REVENUE_CD", "REVENUE_CD_DESC"]: + for key in ["PROCEDURE_CD", "PROCEDURE_CD_DESC"]: if key in code_dict and code_dict[key]: val = code_dict[key] if isinstance(val, list): @@ -768,15 +740,8 @@ def get_implicit_runs(answer_dict: dict) -> dict: claim_type = answer_dict.get("AARETE_DERIVED_CLAIM_TYPE_CD") bill_type = answer_dict.get("BILL_TYPE_CD_DESC") - if claim_type == "H": - run_dict["REVENUE_CD"] = True - elif claim_type == "M": - if not string_utils.is_empty(bill_type): - run_dict["REVENUE_CD"] = True - else: - run_dict["REVENUE_CD"] = False - else: - run_dict["REVENUE_CD"] = False + # REVENUE_CD implicit extraction is disabled + run_dict["REVENUE_CD"] = False # Run proc code implicit for all bill types except 2 if bill_type in ["Inpatient Hospital", "Skilled Nursing Facility"]: @@ -1149,7 +1114,6 @@ def extract_codes_from_service(answer_dict: dict, constants: Constants) -> dict: - NDC_CD: National Drug Codes (NDC) Implicit Codes (Up to Level 2): - PROCEDURE_CD: CPT, HCPCS - - REVENUE_CD: Revenue codes """ service, bill_type = answer_dict.get("SERVICE_TERM"), answer_dict.get( diff --git a/src/constants/constants.py b/src/constants/constants.py index 26f6cf3..f598d92 100644 --- a/src/constants/constants.py +++ b/src/constants/constants.py @@ -125,6 +125,26 @@ class Constants: ) .mapping ) + # Revenue crosswalks (explicit extraction, validation, description fill). + # Implicit RAG for revenue is disabled; embeddings for rev are not loaded. + self.REV_LEVEL1_MAPPING = ( + CrosswalkBuilder() + .from_excel( + "src/constants/mapping_csvs/rev_cd/rev_level1.csv", + "Code", + "Description", + ) + .mapping + ) + self.REV_MAPPING = ( + CrosswalkBuilder() + .from_excel( + _path("constants/mapping_csvs/rev_cd/rev.csv"), + "Code", + "Description", + ) + .mapping + ) # Full procedure code mappings (exact code to description) for validation and # description fill. Used by code extraction validation and fill_code_descriptions. self.CPT_FULL_MAPPING = ( @@ -146,22 +166,6 @@ class Constants: .mapping ) - self.REV_LEVEL1_MAPPING = ( - CrosswalkBuilder() - .from_excel( - "src/constants/mapping_csvs/rev_cd/rev_level1.csv", - "Code", - "Description", - ) - .mapping - ) - self.REV_MAPPING = ( - CrosswalkBuilder() - .from_excel( - _path("constants/mapping_csvs/rev_cd/rev.csv"), "Code", "Description" - ) - .mapping - ) self.GROUPER_APR_DRG_MAPPING = ( CrosswalkBuilder() .from_excel( @@ -203,9 +207,6 @@ class Constants: self.HCPCS_LEVEL1_EMBEDDING = CodeEmbedding("hcpcs_level1") self.HCPCS_LEVEL2_EMBEDDING = CodeEmbedding("hcpcs_level2") - self.REV_LEVEL1_EMBEDDING = CodeEmbedding("rev_level1") - self.REV_EMBEDDING = CodeEmbedding("rev") - #################################### Methodology Breakout #################################### self.VALID_AARETE_DERIVED_FEE_SCHEDULE = ( ListBuilder() @@ -312,15 +313,12 @@ class Constants: Returns the CodeEmbedding attribute corresponding to the given level. Args: - level (str): The embedding level identifier (e.g., "rev", "cpt_level1"). + level (str): The embedding level identifier (e.g., "cpt_level1"). Returns: CodeEmbedding: The matching embedding object, or None if not found. """ level_map = { - # Revenue code embeddings - "rev": self.REV_EMBEDDING, - "rev_level1": self.REV_LEVEL1_EMBEDDING, # CPT code embeddings "cpt_level1": self.CPT_LEVEL1_EMBEDDING, "cpt_level2": self.CPT_LEVEL2_EMBEDDING, diff --git a/src/embeddings/rev/choices.pkl b/src/embeddings/rev/choices.pkl deleted file mode 100644 index 0e52152..0000000 Binary files a/src/embeddings/rev/choices.pkl and /dev/null differ diff --git a/src/embeddings/rev/embeddings.npy b/src/embeddings/rev/embeddings.npy deleted file mode 100644 index 6459a50..0000000 Binary files a/src/embeddings/rev/embeddings.npy and /dev/null differ diff --git a/src/embeddings/rev/faiss_index.bin b/src/embeddings/rev/faiss_index.bin deleted file mode 100644 index 086c0a2..0000000 Binary files a/src/embeddings/rev/faiss_index.bin and /dev/null differ diff --git a/src/embeddings/rev_level1/choices.pkl b/src/embeddings/rev_level1/choices.pkl deleted file mode 100644 index 3ac0f7a..0000000 Binary files a/src/embeddings/rev_level1/choices.pkl and /dev/null differ diff --git a/src/embeddings/rev_level1/embeddings.npy b/src/embeddings/rev_level1/embeddings.npy deleted file mode 100644 index 3dc9e97..0000000 Binary files a/src/embeddings/rev_level1/embeddings.npy and /dev/null differ diff --git a/src/embeddings/rev_level1/faiss_index.bin b/src/embeddings/rev_level1/faiss_index.bin deleted file mode 100644 index a6a971e..0000000 Binary files a/src/embeddings/rev_level1/faiss_index.bin and /dev/null differ diff --git a/src/tests/test_code_funcs.py b/src/tests/test_code_funcs.py index 561bf3a..43906b0 100644 --- a/src/tests/test_code_funcs.py +++ b/src/tests/test_code_funcs.py @@ -162,26 +162,24 @@ class TestCodeFuncs(unittest.TestCase): def test_get_embedding_levels(self): """Test the get_embedding_levels function for determining embedding levels.""" - # Test level 1 with both codes enabled - implicit_run_dict = {"PROCEDURE_CD": True, "REVENUE_CD": True} - embedding_levels, cpt_mapping, hcpcs_mapping, rev_mapping = ( - code_funcs.get_embedding_levels(1, implicit_run_dict, self.constants) + # Test level 1 with procedure code enabled + implicit_run_dict = {"PROCEDURE_CD": True} + embedding_levels, cpt_mapping, hcpcs_mapping = code_funcs.get_embedding_levels( + 1, implicit_run_dict, self.constants ) self.assertIn("cpt_level1", embedding_levels) self.assertIn("hcpcs_level1", embedding_levels) - self.assertIn("rev_level1", embedding_levels) self.assertEqual(cpt_mapping, self.constants.CPT_LEVEL1_MAPPING) # Test level 2 with only procedure code enabled - implicit_run_dict = {"PROCEDURE_CD": True, "REVENUE_CD": False} - embedding_levels, cpt_mapping, hcpcs_mapping, rev_mapping = ( - code_funcs.get_embedding_levels(2, implicit_run_dict, self.constants) + implicit_run_dict = {"PROCEDURE_CD": True} + embedding_levels, cpt_mapping, hcpcs_mapping = code_funcs.get_embedding_levels( + 2, implicit_run_dict, self.constants ) self.assertIn("cpt_level2", embedding_levels) self.assertIn("hcpcs_level2", embedding_levels) - self.assertNotIn("rev", embedding_levels) self.assertEqual(cpt_mapping, self.constants.CPT_LEVEL2_MAPPING) def test_get_match_list(self): @@ -228,7 +226,7 @@ class TestCodeFuncs(unittest.TestCase): mock_json_load.return_value = ["Test Procedure"] # Test with successful match - implicit_run_dict = {"PROCEDURE_CD": True, "REVENUE_CD": True} + implicit_run_dict = {"PROCEDURE_CD": True} # Patch the get_embedding_levels function to return predictable values with patch( @@ -239,7 +237,6 @@ class TestCodeFuncs(unittest.TestCase): ["cpt_level1"], {"12345": "Test Procedure"}, {}, - {}, ) # Call the function @@ -280,7 +277,7 @@ class TestCodeFuncs(unittest.TestCase): mock_get_match_list.return_value = (["Test Procedure"], 0.9) mock_invoke_claude.return_value = '["INVALID_SERVICE", "Test Procedure"]' mock_json_load.return_value = ["INVALID_SERVICE", "Test Procedure"] - implicit_run_dict = {"PROCEDURE_CD": True, "REVENUE_CD": True} + implicit_run_dict = {"PROCEDURE_CD": True} with patch( "src.codes.code_funcs.get_embedding_levels" @@ -289,7 +286,6 @@ class TestCodeFuncs(unittest.TestCase): ["cpt_level1"], {"12345": "Test Procedure"}, {}, - {}, ) result = code_funcs.code_implicit_rag( "TEST SERVICE", implicit_run_dict, "test.pdf", self.constants @@ -351,7 +347,7 @@ class TestCodeFuncs(unittest.TestCase): mock_category.return_value = {} mock_special.return_value = {} mock_rag.return_value = [] # RAG returns list of candidates - implicit_run_dict = {"PROCEDURE_CD": True, "REVENUE_CD": True} + implicit_run_dict = {"PROCEDURE_CD": True} result = code_funcs.build_implicit_candidates( "SERVICE", {}, @@ -374,7 +370,7 @@ class TestCodeFuncs(unittest.TestCase): } mock_special.return_value = {} mock_rag.return_value = [] # RAG returns list of candidates - implicit_run_dict = {"PROCEDURE_CD": True, "REVENUE_CD": True} + implicit_run_dict = {"PROCEDURE_CD": True} code_after_explicit = {"PROCEDURE_CD": ["Category: J"]} result = code_funcs.build_implicit_candidates( "J CODES", @@ -400,7 +396,7 @@ class TestCodeFuncs(unittest.TestCase): "PROCEDURE_CD_DESC": ["Drugs"], } mock_rag.return_value = [] # RAG returns list of candidates - implicit_run_dict = {"PROCEDURE_CD": True, "REVENUE_CD": True} + implicit_run_dict = {"PROCEDURE_CD": True} result = code_funcs.build_implicit_candidates( "DRUG SERVICE", {}, @@ -437,7 +433,7 @@ class TestCodeFuncs(unittest.TestCase): }, }, ] - implicit_run_dict = {"PROCEDURE_CD": True, "REVENUE_CD": True} + implicit_run_dict = {"PROCEDURE_CD": True} code_after_explicit = {"PROCEDURE_CD": ["Category: A"]} result = code_funcs.build_implicit_candidates( "SERVICE", @@ -571,13 +567,13 @@ class TestCodeFuncs(unittest.TestCase): """Test the get_implicit_runs function for determining which code runs to execute.""" # Test hospital claim type result = code_funcs.get_implicit_runs({"AARETE_DERIVED_CLAIM_TYPE_CD": "H"}) - self.assertTrue(result["REVENUE_CD"]) + self.assertFalse(result["REVENUE_CD"]) # Test medical claim type with bill type result = code_funcs.get_implicit_runs( {"AARETE_DERIVED_CLAIM_TYPE_CD": "M", "BILL_TYPE_CD_DESC": "Outpatient"} ) - self.assertTrue(result["REVENUE_CD"]) + self.assertFalse(result["REVENUE_CD"]) self.assertTrue(result["PROCEDURE_CD"]) # Test inpatient bill type @@ -587,7 +583,7 @@ class TestCodeFuncs(unittest.TestCase): "BILL_TYPE_CD_DESC": "Inpatient Hospital", } ) - self.assertTrue(result["REVENUE_CD"]) + self.assertFalse(result["REVENUE_CD"]) self.assertFalse(result["PROCEDURE_CD"]) @patch("src.codes.code_funcs.clean_service") @@ -616,7 +612,7 @@ class TestCodeFuncs(unittest.TestCase): "SERVICE_TERM": "TEST SERVICE", "BILL_TYPE_CD_DESC": "Outpatient", } - mock_implicit_runs.return_value = {"PROCEDURE_CD": True, "REVENUE_CD": True} + mock_implicit_runs.return_value = {"PROCEDURE_CD": True} # Test explicit code match - extract_codes_from_service normalizes to JSON list format mock_explicit.return_value = {"PROCEDURE_CD": "12345"} @@ -855,7 +851,7 @@ class TestCodeFuncs(unittest.TestCase): with patch("src.codes.code_funcs.fill_bill_type", return_value={}): with patch( "src.codes.code_funcs.get_implicit_runs", - return_value={"PROCEDURE_CD": True, "REVENUE_CD": True}, + return_value={"PROCEDURE_CD": True}, ): result = code_funcs.extract_codes_from_service( {"SERVICE_TERM": "TEST"}, self.constants @@ -876,7 +872,7 @@ class TestCodeFuncs(unittest.TestCase): with patch("src.codes.code_funcs.fill_bill_type", return_value={}): with patch( "src.codes.code_funcs.get_implicit_runs", - return_value={"PROCEDURE_CD": True, "REVENUE_CD": True}, + return_value={"PROCEDURE_CD": True}, ): result = code_funcs.extract_codes_from_service( {"SERVICE_TERM": "Office visit"}, self.constants @@ -916,7 +912,7 @@ class TestCodeFuncs(unittest.TestCase): with patch("src.codes.code_funcs.fill_bill_type", return_value={}): with patch( "src.codes.code_funcs.get_implicit_runs", - return_value={"PROCEDURE_CD": True, "REVENUE_CD": True}, + return_value={"PROCEDURE_CD": True}, ): result = code_funcs.extract_codes_from_service( {"SERVICE_TERM": "J CODES"}, self.constants @@ -956,7 +952,7 @@ class TestCodeFuncs(unittest.TestCase): with patch("src.codes.code_funcs.fill_bill_type", return_value={}): with patch( "src.codes.code_funcs.get_implicit_runs", - return_value={"PROCEDURE_CD": True, "REVENUE_CD": True}, + return_value={"PROCEDURE_CD": True}, ): result = code_funcs.extract_codes_from_service( {"SERVICE_TERM": "J CODES"}, self.constants