From 3fee9fca4c631fed975cfeacb118c279a77b60a4 Mon Sep 17 00:00:00 2001 From: Katon Minhas Date: Fri, 7 Jun 2024 09:36:07 -0700 Subject: [PATCH] Error handling, streamlined IO --- src/config.py | 19 ++++-- src/file_processing.py | 30 ++++----- src/main.py | 11 +--- src/test.py | 140 +++++++++++++++++++++-------------------- src/utils.py | 28 +++++++++ 5 files changed, 130 insertions(+), 98 deletions(-) diff --git a/src/config.py b/src/config.py index ab922f1..9652321 100644 --- a/src/config.py +++ b/src/config.py @@ -8,7 +8,7 @@ from llama_index.llms.bedrock import Bedrock # General Settings TEST = False # True to run test prompt - just for testing model connection VERBOSE = True -CLIENT_NAME = 'Payer' +CLIENT_NAME = 'Priority Health' TODAY = datetime.now().strftime("%Y%m%d") # Valid values @@ -16,11 +16,18 @@ VALID_NETWORKS = ['HMO', 'PPO', 'EPO', 'POS', 'FFS'] VALID_PROGRAMS = [] VALID_LOB = [] -# I/O Options -WRITE_OUTPUT = True # True writes csvs, False prints result in console but no output written -OUTPUT_FOLDER = 'output' +# Input Settings READ_MODE = '_LOCAL_' # OR '_S3_' -OUTPUT_MODE = '_INDIVIDUAL_' # or '_CONSOLIDATED_' + +# Output Settings +WRITE_OUTPUT = True # True writes csvs, False prints result in console but no output written +OUTPUT_DIRECTORY = 'output' +CONSOLIDATED_OUTPUT_DIRECTORY = 'consolidated_output' +OUTPUT_CSV_PATH = f'consolidated_output_{TODAY}.csv' +TD_RESULTS_NAME = 'td_results.csv' +BU_RESULTS_NAME = 'bu_results.csv' +UNPROCESSED_RESULTS_NAME = 'combined_results_unprocessed.csv' +PROCESSED_RESULTS_NAME = 'combined_results_post_processed.csv' # Multithread Settings MAX_WORKERS = 20 @@ -43,7 +50,7 @@ AWS_SECRET_ACCESS_KEY="fPm3Rn3cde8AOVj0hP/n3u1MwSrQU17qxDPRK7ar" AWS_SESSION_TOKEN="IQoJb3JpZ2luX2VjEGYaCXVzLWVhc3QtMiJHMEUCIEFjYLb1fysSczLo2Ae7zUOD61H0YhfYPtrzlwXGw6TxAiEAnPVt3UX0UGGxJrAS3T14mVud26zn8i2qIbgjUZAkQmIqjAMI7///////////ARAAGgw2NjAxMzEwNjg3ODIiDH2L5zRJcBcGAXDxMSrgAhoB9QUZFydQiXFS7YUNrYBwSRX3fxPM6dml/WETf9qesJ7tkuOlXJIa7ZZLjJreuMCRfS4eUis0mMDZg+8pb1lLYQAAwuCe6JHN757NH7lwOXSdLY/pzqHc+w4atcGfKNsexj6BIGCyOKbqjbA0wcMgz1y5DqEY8lBeU+E/FTMBa6r9SKYH2hyD41KZQrtUAbKeOWrw2lMjxsr4TzlkdHXw8uQJC4WY7uC6iAzZoR+WDQvfOFu1ltzeJlMiJZA74V1pStK0s/f+QtlwmT6KLu8gVafepohAFs4tuWlz5Ohmn0T/hGQURD8Em1dU22Ljj0qfcGejEUFw8HQDI5kYtwFBDU4XLnDh5Z01Dyxidted8xV3DNI6Ok79cAT950jXXSipBuHKxNv9qMAbrgrAir5GUPhs04XnocjIw7FpOoYzglURU+Y/R+wHeOtXRou4VcDLdoOtlrnrKX9/WR5eOI8woZ+MswY6pgG+tl5kEnjwJJDMCErNLFBnAmBHaC+0oBFRPOaX7ZYHY7ryIfo21kqLEkgXD54XY6xRIyfpEMjBIled+AUOJC17BEjZUq3tUUMi1FEo4KFwANQntvQVDHQ+nwmtNQ4NWeCBfBOqqEArsXjG3SROTQOBKr75NHlPFYtsTr2vK03xCe6dVF/Uq0giyXvoEMyT9eCJTFW1Bzs9HUpUhG9h0lgO2q4LKaLd" # File Paths -LOCAL_PATH = 'data/test/' # Replace with local +LOCAL_PATH = 'data/priority_health/' # Replace with local # S3 Settings S3_CLIENT = boto3.client('s3', diff --git a/src/file_processing.py b/src/file_processing.py index c0afa8e..cc3caa6 100644 --- a/src/file_processing.py +++ b/src/file_processing.py @@ -94,47 +94,43 @@ def merge_results(td_results, bu_results, text_dict): def process_file(file_object): - # Start Processing + ################## INITIATE PROCESSING ################## filename, contract_text = file_object if config.VERBOSE: print(f"Processing {filename}...") - # Create directories + ################## CREATE OUTPUT DIRECTORIES ################## base_filename = os.path.splitext(filename)[0].strip() - output_dir = os.path.join(config.OUTPUT_FOLDER, base_filename) + output_dir = os.path.join(config.OUTPUT_DIRECTORY, base_filename) os.makedirs(output_dir, exist_ok=True) - - #### PREPROCESS #### + ################## PREPROCESS ################## contract_text = preprocess.clean_newlines(contract_text) text_dict = preprocess.split_text(contract_text) text_dict = table_funcs.align_and_format_tables(text_dict) text_dict = preprocess.highlight_rates(text_dict) - #### Run Top Down #### + ################## RUN TOP DOWN ################## td_results = prompt_funcs.run_top_down(filename, text_dict) # Returns list of dictionaries for each page - # Write TD results - pd.DataFrame(td_results).to_csv(os.path.join(output_dir, 'td_results.csv'), index=False) + pd.DataFrame(td_results).to_csv(os.path.join(output_dir, config.TD_RESULTS_NAME), index=False) print(f"Top Down Complete - {filename}") - #### Run Bottom Up #### + ################## RUN BOTTOM UP ################## bu_results = prompt_funcs.run_bottom_up(filename, text_dict) # Returns list of dictionaries - # Write BU results - pd.DataFrame(bu_results).to_csv(os.path.join(output_dir, 'bu_results.csv'), index=False) + pd.DataFrame(bu_results).to_csv(os.path.join(output_dir, config.BU_RESULTS_NAME), index=False) print(f"Bottom Up Complete - {filename}") - #### Combine #### + ################## COMBINE TOP DOWN AND BOTTOM UP ################## combined_results = merge_results(clean_td(td_results), bu_results, text_dict) print(f"TD/BU Merge Complete - {filename} ") - #### Write unprocessed output #### - # Convert to DataFrame + ################## WRITE UNPROCESSED OUTPUT ################## combined_df = pd.DataFrame(combined_results) - combined_df.to_csv(os.path.join(output_dir, 'combined_results_unprocessed.csv'), index=False) + combined_df.to_csv(os.path.join(output_dir, config.UNPROCESSED_RESULTS_NAME), index=False) - #### POSTPROCESSING #### + ################## RUN POSTPROCESSING ################## combined_df = combined_df.applymap(postprocessingfuncs.sanitize_value) post_processed_combined_df = postprocess.postprocess_results(combined_df) - post_processed_combined_df.to_csv(os.path.join(output_dir, 'combined_results_post_processed.csv'), index=False) + post_processed_combined_df.to_csv(os.path.join(output_dir, config.PROCESSED_RESULTS_NAME), index=False) print(f"Postprocessing Complete - {filename} ") print(f"Output Complete - {filename} ") diff --git a/src/main.py b/src/main.py index 18dc9e3..4912bfa 100644 --- a/src/main.py +++ b/src/main.py @@ -18,10 +18,6 @@ def main(): input_dict = utils.read_input() # already_processed = [s.split('.txt_results')[0]+'.txt' for s in os.listdir('results/')] # input_dict = {key : input_dict[key] for key in input_dict.keys() if key not in already_processed} - - # Set up results folder - if not os.path.exists('results'): - os.makedirs('results') def process_item(item): key, value = item @@ -45,10 +41,9 @@ def main(): except Exception as e: print(f"Error: {e}") - - # Write Output - if config.WRITE_OUTPUT and config.OUTPUT_MODE == '_CONSOLIDATED_': - consolidated_df = utils.consolidate_individual(input_folder='temp') # Consolidate temp results to one file, then write output + # Write Consolidated Output + if config.WRITE_OUTPUT: + utils.consolidate_csvs(config.OUTPUT_DIRECTORY, config.OUTPUT_CSV_PATH) if __name__ == "__main__": diff --git a/src/test.py b/src/test.py index 2f2b869..f5b7790 100644 --- a/src/test.py +++ b/src/test.py @@ -3,6 +3,7 @@ import re import json import csv from io import StringIO +import config import utils import preprocess @@ -12,92 +13,97 @@ import prompt_funcs import claude_funcs -input_dict = utils.read_input() +if config.WRITE_OUTPUT: + utils.consolidate_csvs(config.OUTPUT_DIRECTORY, config.OUTPUT_CSV_PATH) -filename = list(input_dict.keys())[0] -contract_text = input_dict[filename] -def convert_to_dict(table_text): - table_text = table_text.strip('{}') - table_text_list = table_text.split(']') - table_text_list = [item for item in table_text_list if len(item) > 0] - final_dict = {} - num_elements = 0 - for key_value_text in table_text_list: - key = key_value_text.split(':')[0].strip(', \'') - value = ':'.join(key_value_text.split(':')[1:]).strip(' [] \'') +# input_dict = utils.read_input() + +# filename = list(input_dict.keys())[0] +# contract_text = input_dict[filename] + +# def convert_to_dict(table_text): +# table_text = table_text.strip('{}') +# table_text_list = table_text.split(']') +# table_text_list = [item for item in table_text_list if len(item) > 0] + +# final_dict = {} +# num_elements = 0 +# for key_value_text in table_text_list: +# key = key_value_text.split(':')[0].strip(', \'') +# value = ':'.join(key_value_text.split(':')[1:]).strip(' [] \'') - f = StringIO(value) - reader = csv.reader(f, delimiter=',') +# f = StringIO(value) +# reader = csv.reader(f, delimiter=',') - value_list = [] - for row in reader: - for r in row: - value_list.append(r) +# value_list = [] +# for row in reader: +# for r in row: +# value_list.append(r) - final_dict[key] = value_list - num_elements = len(value_list) - return final_dict, num_elements +# final_dict[key] = value_list +# num_elements = len(value_list) +# return final_dict, num_elements -def format_table(table_json, table_size): - table_text = "" - for i in range(table_size): - for key in table_json.keys(): - if table_json[key][i]: - #table_text += key + ': ' + table_json[key][i] + ', ' - table_text += table_json[key][i] + ' : ' - table_text += r'\n' - return table_text +# def format_table(table_json, table_size): +# table_text = "" +# for i in range(table_size): +# for key in table_json.keys(): +# if table_json[key][i]: +# #table_text += key + ': ' + table_json[key][i] + ', ' +# table_text += table_json[key][i] + ' : ' +# table_text += r'\n' +# return table_text -def align_and_format_tables(text_dict): - aligned_text_dict = {} - for key, text in text_dict.items(): - if 'Table Start' in text: - table_texts = re.findall(r'-------Table Start--------(.*?)-------Table End--------', text, re.DOTALL) - for table_text in table_texts: - try: - # Extract pretable text - pretable = table_text.split('{')[0].strip() +# def align_and_format_tables(text_dict): +# aligned_text_dict = {} +# for key, text in text_dict.items(): +# if 'Table Start' in text: +# table_texts = re.findall(r'-------Table Start--------(.*?)-------Table End--------', text, re.DOTALL) +# for table_text in table_texts: +# try: +# # Extract pretable text +# pretable = table_text.split('{')[0].strip() - # Extract and format table text - #table_only = "{" + table_text.split('{', 1)[1].rsplit('}', 1)[0].replace("'", '"') + "}" - table_only = "{" + table_text.split('{', 1)[1].rsplit('}', 1)[0] + "}" +# # Extract and format table text +# #table_only = "{" + table_text.split('{', 1)[1].rsplit('}', 1)[0].replace("'", '"') + "}" +# table_only = "{" + table_text.split('{', 1)[1].rsplit('}', 1)[0] + "}" - table, table_size = convert_to_dict(table_only) - table_formatted = format_table(table, table_size) +# table, table_size = convert_to_dict(table_only) +# table_formatted = format_table(table, table_size) - # Align - if text.count(pretable) == 2: - # Remove original table - aligned_text = text.replace(table_text, "") - # Align new table - aligned_text = aligned_text.replace(pretable, pretable + table_formatted) - else: - aligned_text = text.replace(table_text, pretable + table_formatted) - except: - aligned_text = text - aligned_text_dict[key] = aligned_text.replace("-------Table Start--------", "").replace("-------Table End--------", "") - else: - aligned_text_dict[key] = text +# # Align +# if text.count(pretable) == 2: +# # Remove original table +# aligned_text = text.replace(table_text, "") +# # Align new table +# aligned_text = aligned_text.replace(pretable, pretable + table_formatted) +# else: +# aligned_text = text.replace(table_text, pretable + table_formatted) +# except: +# aligned_text = text +# aligned_text_dict[key] = aligned_text.replace("-------Table Start--------", "").replace("-------Table End--------", "") +# else: +# aligned_text_dict[key] = text - return aligned_text_dict +# return aligned_text_dict -contract_text = preprocess.clean_newlines(contract_text) -text_dict = preprocess.split_text(contract_text) -text_dict = align_and_format_tables(text_dict) -text_dict = preprocess.highlight_rates(text_dict) +# contract_text = preprocess.clean_newlines(contract_text) +# text_dict = preprocess.split_text(contract_text) +# text_dict = align_and_format_tables(text_dict) +# text_dict = preprocess.highlight_rates(text_dict) -#print(text_dict['13']) +# #print(text_dict['13']) -# d = {'SERVICE' : 'Personal Care Services', 'FULL_METHODOLOGY' : 'Per 15 min (group), Health Plan fee schedule: $3.00 per unit (1 unit = 15 minutes)'} -# prompt = BOTTOM_UP_CODES(d, text_dict['3']) +# # d = {'SERVICE' : 'Personal Care Services', 'FULL_METHODOLOGY' : 'Per 15 min (group), Health Plan fee schedule: $3.00 per unit (1 unit = 15 minutes)'} +# # prompt = BOTTOM_UP_CODES(d, text_dict['3']) -# answer = claude_funcs.invoke_claude_3(prompt, max_tokens=4000) +# # answer = claude_funcs.invoke_claude_3(prompt, max_tokens=4000) -# print(answer) +# # print(answer) diff --git a/src/utils.py b/src/utils.py index e05e140..43a16b0 100644 --- a/src/utils.py +++ b/src/utils.py @@ -107,3 +107,31 @@ def format_td_check(td_dicts, dont_include_list): final_str += '\n' dict_count += 1 return final_str + + +def consolidate_csvs(output_dir, output_file): + os.makedirs(config.CONSOLIDATED_OUTPUT_DIRECTORY, exist_ok=True) + + df_list = [] + # Walk through each folder in the output directory + for root, dirs, files in os.walk(output_dir): + for dir_name in dirs: + dir_path = os.path.join(root, dir_name) + file_path = os.path.join(dir_path, config.PROCESSED_RESULTS_NAME) + if os.path.isfile(file_path): + try: + df = pd.read_csv(file_path) + df_list.append(df) + except: + pass + + concatenated_df = pd.concat(df_list, ignore_index=True) + concatenated_df.to_csv(os.path.join(config.CONSOLIDATED_OUTPUT_DIRECTORY, output_file), index=False) + print(f"All CSV files have been consolidated into {output_file}") + + + + + + +