From 001c2e45b5402c6156f91886b6e042a94c455607 Mon Sep 17 00:00:00 2001 From: Alex Galarce Date: Wed, 13 Nov 2024 20:12:40 +0000 Subject: [PATCH] Merged in hotfix/plan_state (pull request #267) plan state hotfix * updated state_check function * added a wrapper function apply_state_check_to_df function to enable state_check function to work with dataframe and also wrote unit tests for both the functions. * Merged main into hotfix/plan_state * append extra column with new health plan state corrected * Merge remote-tracking branch 'refs/remotes/origin/hotfix/plan_state' into hotfix/plan_state * fix tests * move apply_state_check_to_df() to cnc_hotfix.py * move all functionality out of utils and into cnc_hotfix.py * Merge remote-tracking branch 'origin/main' into hotfix/plan_state * remove duplicated pythonpath in pyproject.toml * format cnc_hotfix.py * changed to use valid.STATE_MAP to not duplicate functionality Approved-by: Katon Minhas --- fieldExtraction/pyproject.toml | 2 +- fieldExtraction/src/cnc_hotfix.py | 91 ++++++++++ fieldExtraction/tests/cnc_hotfix_test.py | 209 +++++++++++++++++++++++ 3 files changed, 301 insertions(+), 1 deletion(-) create mode 100644 fieldExtraction/src/cnc_hotfix.py create mode 100644 fieldExtraction/tests/cnc_hotfix_test.py diff --git a/fieldExtraction/pyproject.toml b/fieldExtraction/pyproject.toml index ef9964a..04f112b 100644 --- a/fieldExtraction/pyproject.toml +++ b/fieldExtraction/pyproject.toml @@ -32,7 +32,7 @@ target-version = ['py312'] disable_error_code = ["import-untyped","assignment","name-defined","call-arg","var-annotated","attr-defined","arg-type"] [tool.pytest.ini_options] +pythonpath=["src"] testpaths = [ "tests" ] -pythonpath = ["src"] \ No newline at end of file diff --git a/fieldExtraction/src/cnc_hotfix.py b/fieldExtraction/src/cnc_hotfix.py new file mode 100644 index 0000000..28fa572 --- /dev/null +++ b/fieldExtraction/src/cnc_hotfix.py @@ -0,0 +1,91 @@ +import pandas as pd +from valid import STATE_MAP + + +def state_check(states: str | list[str], is_dataframe: bool = False) -> str | list[str]: + # TODO: genericize this function to take in and put out a dataframe and put it into + # postprocessing_funcs.py + + keywords_to_detect_invalid_state_name = [ + "suggest", + "The Contract Text", + "refers", + "For ", + "Answer", + "Multiple", + "Statewide", + ] + + if ( + isinstance(states, list) and all(isinstance(state, str) for state in states) + ) or isinstance(states, str): + + if isinstance(states, str): + if ( + not any( + kw.lower() in states.lower() + for kw in keywords_to_detect_invalid_state_name + ) + and "," in states + ): + states = states.split(",") + else: + states = [states] + + new_states = [] + + for state in states: + state = state.strip() + + # fix 1 + if state == "Hlorida": + state = "Florida" + + # fix 2 - LLM self-talking + if any( + kw.lower() in state.lower() + for kw in keywords_to_detect_invalid_state_name + ): + state = "N/A" + + if state in STATE_MAP.keys() or state.upper() in STATE_MAP.keys(): + new_states.append(STATE_MAP[state.upper()]) + + elif state.strip().title() in STATE_MAP.values(): + new_states.append(state.strip().title()) + + else: + new_states.append("N/A") + + if is_dataframe is False: + return new_states # list + + return ", ".join(new_states) # concatenated string + + raise TypeError( + f"Expected a string or list of strings, got {type(states).__name__}" + ) + + +def apply_state_check_to_df( + df: pd.DataFrame, is_dataframe: bool = True +) -> pd.DataFrame: + + if not isinstance(df, pd.DataFrame): + raise TypeError(f"Expected a dataframe, got {type(df).__name__}") + + if is_dataframe is False: + raise ValueError( + f"Expected 'is_dataframe' parameter as True, got {is_dataframe}" + ) + + if "Health Plan State" not in df.columns: + raise KeyError( + f"Column 'Health Plan State' not found in DataFrame. Available columns are: {list(df.columns)}" + ) + + df["Health Plan State_corrected"] = df["Health Plan State"].apply( + state_check, args=(is_dataframe,) + ) + + return df diff --git a/fieldExtraction/tests/cnc_hotfix_test.py b/fieldExtraction/tests/cnc_hotfix_test.py new file mode 100644 index 0000000..42eaae4 --- /dev/null +++ b/fieldExtraction/tests/cnc_hotfix_test.py @@ -0,0 +1,209 @@ +import pytest +import math +import pandas as pd +from pandas.testing import assert_frame_equal + +from cnc_hotfix import state_check, apply_state_check_to_df + + +class TestStateCheck: + + @pytest.mark.parametrize("input_state,expected_df_output,expected_non_df_output", [ + ("Multiple", "N/A",["N/A"]), + ("Peach State", "N/A",["N/A"]), + ("Statewide", "N/A",["N/A"]), + ("Sunshine State Health Plan Suggests This Is For Florida, But Since The State Is Not Explicitly Specified, The Most Appropriate Answer Is:\n\nN/A", "N/A", ["N/A"]), + ("The Contract Text Indicates This Health Plan Is For The Sunshine State, Which Is Florida. Therefore, The Answer Is:\n\nFlorida->N/A", "N/A", ["N/A"]), + ("Sunshine State", "N/A", ["N/A"]), + ("The Health Plan Is For The Sunshine State, Which Refers To Florida. However, Since The State Is Not Explicitly Specified, The Most Accurate Answer Based Solely On The Information Provided Is:\n\nN/A", "N/A", ["N/A"]), + ]) + + def test_invalid_state_names(self, input_state, expected_df_output, expected_non_df_output): + """Test invalid inputs""" + + assert state_check(input_state,is_dataframe=True) == expected_df_output + assert state_check(input_state,is_dataframe=False) == expected_non_df_output + + def test_hlorida_correction(self): + """Test fix - Hlorida -> Florida""" + assert state_check("Hlorida",is_dataframe=False) == ["Florida"] + + @pytest.mark.parametrize("input_state,expected_df_output,expected_non_df_output", [ + ("AL", "Alabama",["Alabama"]), + ("Florida", "Florida",["Florida"]), + ("FLORIDA", "Florida",["Florida"]), + ("florida", "Florida",["Florida"]), + ]) + + def test_valid_states(self, input_state, expected_df_output, expected_non_df_output): + """Test valid inputs""" + assert state_check(input_state,is_dataframe=True) == expected_df_output + assert state_check(input_state,is_dataframe=False) == expected_non_df_output + + def test_list_inputs(self): + """Test a list of input""" + input_states = ["AL", "Florida", "Hlorida", "Multiple"] + + expected_df_output = "Alabama, Florida, Florida, N/A" + expected_non_df_output = ["Alabama", "Florida", "Florida", "N/A"] + + assert state_check(input_states,is_dataframe=True) == expected_df_output + assert state_check(input_states,is_dataframe=False) == expected_non_df_output + + @pytest.mark.parametrize("invalid_input", [ + 123, + None, + {"state": "Florida"}, + True, + math.nan, + pd.NA + ]) + + def test_invalid_input_types(self, invalid_input): + """Test invalid input types""" + + with pytest.raises(TypeError): + state_check(invalid_input) + + @pytest.mark.parametrize("empty_input,expected_df_output,expected_non_df_output", [ + ("", "N/A",["N/A"]), + ([], "",[]), + ([" "], "N/A",["N/A"]) + ]) + + def test_empty_inputs(self, empty_input, expected_df_output, expected_non_df_output): + """Test empty inputs""" + assert state_check(empty_input,is_dataframe=True) == expected_df_output + assert state_check(empty_input,is_dataframe=False) == expected_non_df_output + + def test_multiple_invalid_states(self): + """Test multiple invalid inputs""" + + input_states = [ + "Multiple", + "Peach State", + "The Health Plan Is For The Sunshine State, Which Refers To Florida. So The Answer Is:\n\nFlorida->N/A", + "The Health Plan Is For The Florida State, Which Refers To Florida", + "Sunshine State Health Plan Suggests This Is For Florida, But Since The State Is Not Explicitly Specified, The Most Appropriate Answer Is:\n\nN/A->N/A", + "Sunshine State Health Plan->N/A", + "Sunshine State->N/A", + "The Contract Text Indicates This Health Plan Is For Florida, As It Mentions Sunshine State Health Plan, Inc. Which Operates In Florida. However, To Strictly Follow The Instructions, I Will Provide The Answer:\n\nN/A->N/A" + + ] + + expected_df_output = ", ".join(["N/A"]*8) + expected_non_df_output = ["N/A"]*8 + + assert state_check(input_states,is_dataframe=True) == expected_df_output + assert state_check(input_states,is_dataframe=False) == expected_non_df_output + + def test_mixed_states(self): + """Test mixed input""" + input_states = ["AL", "Multiple", "Florida", "Hlorida", "Florida State Health Plan"] + + expected_df_output = "Alabama, N/A, Florida, Florida, N/A" + expected_non_df_output = ["Alabama", "N/A", "Florida", "Florida", "N/A"] + + assert state_check(input_states,is_dataframe=True) == expected_df_output + assert state_check(input_states,is_dataframe=False) == expected_non_df_output + + def concatenated_string_states(self): + """Test concatentated input""" + input_states = ["Arizona, Louisiana","NY, NJ, Connecticut"] + + expected_df_output = "Arizona, Louisiana, New York, New Jersey" + expected_non_df_output = ["Arizona", "Louisiana", "New York", "New Jersey"] + + assert state_check(input_states,is_dataframe=True) == expected_df_output + assert state_check(input_states,is_dataframe=False) == expected_non_df_output + +class TestApplyStateCheckToDataFrame: + + def test_basic_state_processing(self): + """Test basic state code to full name conversion""" + df = pd.DataFrame({ + 'Health Plan State': ['NY', 'CA', 'TX', ['CO','CT'], 'NJ, AR'] + }) + + expected_df = pd.DataFrame({ + 'Health Plan State': ['NY', 'CA', 'TX', ['CO','CT'], 'NJ, AR'], + 'Health Plan State_corrected': ['New York', 'California', 'Texas', 'Colorado, Connecticut', 'New Jersey, Arkansas'] + }) + + result_df = apply_state_check_to_df(df,is_dataframe=True) + + assert_frame_equal(result_df, expected_df) + + def test_mixed_case_states(self): + """Test handling of mixed case input""" + df = pd.DataFrame({ + 'Health Plan State': ['ny', 'Ca', 'TeXaS'] + }) + + expected_df = pd.DataFrame({ + 'Health Plan State': ['ny', 'Ca', 'TeXaS'], + 'Health Plan State_corrected': ['New York', 'California', 'Texas'] + }) + + result_df = apply_state_check_to_df(df,is_dataframe=True) + + assert_frame_equal(result_df, expected_df) + + def test_invalid_states(self): + """Test handling of invalid state codes""" + + df = pd.DataFrame({ + 'Health Plan State': [['SH', 'IO'], 'LM', 'XY'] + }) + + expected_df = pd.DataFrame({ + 'Health Plan State': [['SH', 'IO'], 'LM', 'XY'], + 'Health Plan State_corrected': ['N/A, N/A', 'N/A', 'N/A'] + }) + + result_df = apply_state_check_to_df(df,is_dataframe=True) + + assert_frame_equal(result_df, expected_df) + + def test_already_full_state_names(self): + """Test handling of already expanded state names""" + df = pd.DataFrame({ + 'Health Plan State': ['Vermont', 'Tennessee', 'Mississippi'] + }) + + expected_df = pd.DataFrame({ + 'Health Plan State': ['Vermont', 'Tennessee', 'Mississippi'], + 'Health Plan State_corrected': ['Vermont', 'Tennessee', 'Mississippi'] + }) + + result_df = apply_state_check_to_df(df,is_dataframe=True) + + assert_frame_equal(result_df, expected_df) + + def test_empty_dataframe(self): + """Test handling of empty DataFrame""" + df = pd.DataFrame({'Health Plan State': []}) + expected_df = pd.DataFrame({ + 'Health Plan State': [], + 'Health Plan State_corrected': [] + }) + + result_df = apply_state_check_to_df(df,is_dataframe=True) + + assert_frame_equal(result_df, expected_df) + + def test_wrong_column_name(self): + """Test error handling for non-existent column""" + df = pd.DataFrame({ + 'Heaalth Plans state': ['NY', 'CA'], + }) + + with pytest.raises(KeyError) as exc_info: + apply_state_check_to_df(df,is_dataframe=True) + + def test_is_dataframe_parameter_false(self): + """Test 'is_dataframe' parameter as false""" + df = pd.DataFrame({'Health Plan State': ['NY', 'CA']}) + + with pytest.raises(ValueError): + apply_state_check_to_df(df,is_dataframe=False) \ No newline at end of file