app.py

#!/usr/bin/env python
# coding: utf-8

# In[1]:


import pandas as pd
import numpy as np
import gradio as gr
# gr.__version__
import os, tempfile
from scipy.sparse import csr_matrix
import gensim.downloader as api
from sentence_transformers import SentenceTransformer, util
from sklearn.feature_extraction.text import CountVectorizer
from gensim.utils import simple_preprocess
from concurrent.futures import ThreadPoolExecutor
import chardet
import torch
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
from UI_input_output import read_clean_model
import time

# a wrapper function, to show how long a function takes
def timeit(func):
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        duration = end - start
        
        # Convert seconds to minutes and seconds
        minutes = int(duration // 60)
        seconds = int(duration % 60)
        
        print(f"\nFunction '{func.__name__}' took {minutes} minutes and {seconds} seconds.")
        return result
    return wrapper

# Declare global placeholder for dataframes
input_df = pd.DataFrame()
mdr_df = pd.DataFrame()
result_df = pd.DataFrame()

# set the default value for slider
slider_default_v = 0
threshold_slider_input = 'No filter'

# to load the csv files as input and mdr file; return a list of columns for users to select
@timeit
def process_two_csvs(file_path_input, file_path_mdr, model_selected):
    
    ###########  upload files
    file_missing_msg =  """<p>  
        <span style="font-weight:bold; color:red; font-size:20px;">At least one of the files is missing from upload</span>.
        </p>"""
        

    if file_path_input is None or file_path_mdr is None:
        return None, file_missing_msg

    try:
        # Read both files into DataFrames with their own encoding
        with open(file_path_input, 'rb') as f:
            result = chardet.detect(f.read())
        encoding_input = result['encoding']
        
        with open(file_path_mdr, 'rb') as f:
            result = chardet.detect(f.read())
        encoding_mdr = result['encoding']
    
        input_df = pd.read_csv(file_path_input, encoding=encoding_input)
        mdr_df = pd.read_csv(file_path_mdr, encoding=encoding_mdr)
        
        # lower case all column names
        input_df.columns = input_df.columns.str.lower()
        mdr_df.columns = mdr_df.columns.str.lower()
        
        # # the key columns from input and mdr.  check if uploaded files contains these
        # input_key_columns: 'variable','description'
        # mdr_key_columns: 'name','definition'
        
        for col in ['variable','description']:
            if col not in input_df.columns:
                return None, f"""<p>Input file is missing 
                    <span style="font-weight:bold; color:red; font-size:20px;">{col}</span>
                    column
                    </p>"""
            
        for col in ['name','definition']:
            if col not in mdr_df.columns:
                return None, f"""<p>MDR file is missing 
                    <span style="font-weight:bold; color:red; font-size:20px;">{col}</span>
                    column
                    </p>"""
        
        print('\n')
        print(input_df.shape)
        print(mdr_df.shape)
    
        # clean and run through models
        print('\n model selected: ',model_selected)
        result_df = read_clean_model(input_df, mdr_df, model_selected)
    
        # # flag where one input variable matched multiple name from MDR
        result_df['Flag'] = (result_df['input_name'].duplicated(keep=False) | result_df['mdr_name'].duplicated(keep=False)) \
                              .map(lambda x: 'Duplicate' if x else 'Unique')
        
        # because per input_name matches to multiple mdr_name
        ## be able to select top # of mdr matched to input will reduce ananlyst manual labor
        result_df = result_df.sort_values(by=['input_name','similarity_score'], ascending=[True, False])
        result_df['mdr_per_inpnt_name_cnt'] = (result_df.groupby(['input_name']).cumcount() + 1)

        return result_df, None
        
    except Exception as e:
        return None, f"Error during processing: {e}"

# Save a filtered DataFrame to CSV and return path
def save_filtered_csv(df, model_selected):
    # Use the system temp directory to save result dataframe as a CSV temp file
    temp_dir = tempfile.gettempdir()
    custom_filename = f"{model_selected}.csv"
    file_path = os.path.join(temp_dir, custom_filename)
    df.to_csv(file_path, index=False)
    
    return file_path

# Triggered after the initial load after process file 
def on_files_uploaded(file_path_input, file_path_mdr, model_selected):
    process_two_csvs_results = process_two_csvs(file_path_input, file_path_mdr, model_selected)
    df = process_two_csvs_results[0]
    proc_err_msg = process_two_csvs_results[1]
    
    try:
        df # if df exist
        csv_path = save_filtered_csv(df, model_selected)
        return df, proc_err_msg, csv_path, df, slider_default_v, threshold_slider_input
    except:
        print('\n no df from on_files_upload() function')
        return None, 'no output table after processing', None, None, slider_default_v, threshold_slider_input


# Threshold filter on similarity scores, and number filter on the number of mdr matched per input
def apply_filters(score_threshold, num_mdr_per_input, df, model_selected):
    df["similarity_score"] = df["similarity_score"].astype(float)
    filtered_df = df.copy()
    
    # Apply (threshold on similarity score) filter (if not None)
    if score_threshold is not None:
        filtered_df = filtered_df[filtered_df["similarity_score"] >= score_threshold]
    
    # Apply (the number of MDR names matched per Input variable) filter (if not "All")
    if num_mdr_per_input != "No filter":
        filtered_df = filtered_df[filtered_df["mdr_per_inpnt_name_cnt"] <= num_mdr_per_input]
    
    csv_path = save_filtered_csv(filtered_df, model_selected)
    
    return filtered_df, csv_path


######################################## APP ###########################################
with gr.Blocks() as demo:
    
    ################## app layout ####################
    gr.Markdown("# Semantic Matching between Input and MDR")
    
    with gr.Row():
        
        # box to load the input file
        l1 = gr.Markdown(
        """
        Upload Input file
        (file must contains at least two columns (not case sensitive): **variable** and **description**)
        """)
        csv_input = gr.File(label="Upload Input File",
                    file_types=[".csv"])
        
        # box to load the MDR file
        l2 = gr.Markdown(
        """
        Upload MDR file
        (file must contains at least two columns (not case sensitive): **name** and **definition**)
        """)
        csv_mdr = gr.File(label="Upload MDR File", 
                    file_types=[".csv"])

    errmsg_proc = gr.HTML(label="Error message") # to display error message for process csvs
    
    # Store processed DataFrame in State for download button to work after load
    process_df_state = gr.State()

    # drop down to select models
    model_dropdown = gr.Dropdown(label="Select the model", choices=['Cosine','CosJaccard'], value='Cosine', multiselect=False)
    
    # button to click and initiate processing the files, i.e. clean the data sets, run through models
    process_btn = gr.Button("Process Files")
    
    # slider filters row
    with gr.Row():
        with gr.Column(scale=2):
            # slider, for filtering similarity score as a threshold --- position slider above the table due to table size change during sliding
            threshold_slider = gr.Slider(minimum=0, maximum=100, step=1, value=slider_default_v, label=">= Similarity Score") # to download all at initial load
        with gr.Column(scale=1):
            # slider, for filtering the number of mdr names matched per input name
            threshold_slider_input = gr.Dropdown(choices=['No filter', 1, 2, 3, 4, 5], label="<= Number of MDR name Per Input Name (or no filter)", value=threshold_slider_input)
                
            
    # table that displays the final result dataframe
    output_table = gr.Dataframe(
        interactive=True,   # True Enables sorting, and editing;  need False in order to do the filtering
        row_count=10,
        column_widths="auto"
    )

    # the link that enables download of this result table
    download_link = gr.File(label="⬇️ Download Result CSV")
    

    ################## app functions ####################

    # link process button to its function and input output
    process_btn.click(
        fn=on_files_uploaded,
        inputs=[csv_input, csv_mdr, model_dropdown],
        outputs=[output_table, errmsg_proc, download_link, process_df_state, threshold_slider, threshold_slider_input]
    )

    # Trigger filtering on slider or dropdown change
    threshold_slider.change(apply_filters, inputs=[threshold_slider, threshold_slider_input, process_df_state, model_dropdown], outputs=[output_table, download_link])
    threshold_slider_input.change(apply_filters, inputs=[threshold_slider, threshold_slider_input, process_df_state, model_dropdown], outputs=[output_table, download_link])



demo.launch()