jacob-mathison/doczyai-pipelines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Error handling, streamlined IO

Browse Source
This commit is contained in:
Katon Minhas
2024-06-07 09:36:07 -07:00
committed by Michael McGuinness
parent 4710c55db2
commit a944dee2f8
5 changed files with 130 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/config.py
+13 -6
View File
@@ -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',
src/file_processing.py
+13 -17
View File
@@ -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} ")
src/main.py
+3 -8
View File
@@ -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__":
src/test.py
+73 -67
View File
@@ -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)
src/utils.py
+28
View File
@@ -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}")