from transformers import pipeline import gradio as gr import os import pandas as pd import tempfile import os #Get api key to record changes. api_key = os.environ.get("HF_API_KEY_INFERENCE") ######################## ## Loading the model: ## ####################### #Changed API endpoint calls to pipeline usage. classifier1 = pipeline( "text-classification", model="chernandezc/EEMM_7_categories_WB", token=os.environ.get("HF_API_KEY_INFERENCE") ) classifier2 = pipeline( "text-classification", model="chernandezc/EEMM_3_dimensions_1201", token=os.environ.get("HF_API_KEY_INFERENCE") ) # ########################################################## # Function to process the output and print classifications # ############################################################ def classify_output(item): # Dictionary for dimensions. label_dict1 = { 'LABEL_0': 'Cognition', 'LABEL_1': 'Affect', 'LABEL_2': 'Self', 'LABEL_3': 'Motivation', 'LABEL_4': 'Attention', 'LABEL_5': 'Overt Behavior', 'LABEL_6': 'Context' } # Dictionary for levels. label_dict2 = { 'LABEL_0': 'Social', 'LABEL_1': 'Psychological', 'LABEL_2': 'Physical' } # Call the local pipelines to get the list of dictionaries output1 = classifier1(item, top_k=None) output2 = classifier2(item, top_k=None) # Store classifications in a list classifications1 = [] classifications2 = [] # Find the item with the highest score for fallback cases highest_score_item1 = max(output1, key=lambda x: x['score']) highest_score_item2 = max(output2, key=lambda x: x['score']) # Process results for dimensions for result_item in output1: if result_item['score'] >= 0.5: classifications1.append((label_dict1[result_item['label']], result_item['score'])) # Process results for levels for result_item in output2: if result_item['score'] >= 0.5: classifications2.append((label_dict2[result_item['label']], result_item['score'])) # Construct and return the final classification message if classifications1 and classifications2: classification_str1 = ', '.join([f"{label} ({score:.2f})" for label, score in classifications1]) classification_str2 = ', '.join([f"{label} ({score:.2f})" for label, score in classifications2]) return f"For dimensions: {classification_str1}\nFor levels: {classification_str2}" elif classifications1 and not classifications2: classification_str1 = ', '.join([f"{label} ({score:.2f})" for label, score in classifications1]) return f"For dimensions: {classification_str1}\nFor levels: No classifications with a score of 0.5 or higher were found.\nHowever, the highest probability was for: '{label_dict2[highest_score_item2['label']]}' ({highest_score_item2['score']:.2f})\n Use this classification with caution due to uncertainty" elif classifications2 and not classifications1: classification_str2 = ', '.join([f"{label} ({score:.2f})" for label, score in classifications2]) return f"For levels: {classification_str2}\nFor dimensions: No classifications with a score of 0.5 or higher were found.\nHowever, the highest probability was for: '{label_dict1[highest_score_item1['label']]}' ({highest_score_item1['score']:.2f}) \n Use this classification with caution due to uncertainty" else: return (f"No classification with a score of 0.5 or higher were found for both levels and dimensions.\n" f"The highest probability for dimensions was: '{label_dict1[highest_score_item1['label']]}' ({highest_score_item1['score']:.2f}).\n" f"The highest probability for level was: '{label_dict2[highest_score_item2['label']]}' ({highest_score_item2['score']:.2f}). \n" f" Use this classification with caution due to uncertainty") ######################################### ######## RUN GRADIO APP ################# ######################################### #Article text. article_text = "### Have fun!" def extract_eemm_binary_columns(classification_text): labels = [ "Cognition", "Affect", "Self", "Motivation", "Attention", "Overt Behavior", "Context", "Social", "Physical", "Psychological" ] result = {label: 0 for label in labels} if classification_text is None: return result text = str(classification_text) # Only use the actual "For dimensions" and "For levels" classification lines. # This avoids accidentally coding the "highest probability was..." caution text as 1. valid_lines = [] for line in text.splitlines(): line = line.strip() if line.startswith("For dimensions:") or line.startswith("For levels:"): if "No classifications with a score" not in line: valid_lines.append(line) valid_text = " ".join(valid_lines) for label in labels: # Looks for labels followed by a probability, e.g. Context (1.00) if f"{label} (" in valid_text: result[label] = 1 return result def classify_bulk_csv(file): if file is None: raise ValueError("Please upload a CSV file.") file_path = file if isinstance(file, str) else file.name df = pd.read_csv(file_path) if "item" not in df.columns: raise ValueError("Your CSV must contain a column named exactly 'item'.") raw_outputs = [] binary_rows = [] for item in df["item"]: if pd.isna(item) or str(item).strip() == "": raw_output = "" else: raw_output = classify_output(str(item)) raw_outputs.append(raw_output) binary_rows.append(extract_eemm_binary_columns(raw_output)) binary_df = pd.DataFrame(binary_rows) # Keep original item, raw model output, and then binary EEMM columns df["raw_eemm_classification"] = raw_outputs ordered_columns = [ "Cognition", "Affect", "Self", "Motivation", "Attention", "Overt Behavior", "Context", "Social", "Physical", "Psychological" ] for col in ordered_columns: df[col] = binary_df[col] output_path = os.path.join( tempfile.gettempdir(), "eemm_bulk_classification_results_binary.csv" ) df.to_csv(output_path, index=False) return output_path # The app. with gr.Blocks(theme=gr.themes.Soft(primary_hue="orange")) as demo: gr.Markdown("# EEMM Item Classification Machine V 0.5") gr.Markdown(""" This app showcases two machine learning models to classify items and phrases into the EEMM grid. It will output a classification for dimensions such as Cognition, Self, Affect, Motivation, Attention, Overt Behavior, or Context. It will also output levels such as Social, Psychological, or Physical. Please note that the machine goes idle after a period of inactivity. If this occurs, waking it up may take a short while. """) with gr.Tab("Single item"): txtbx = gr.Textbox( value="I would like to feel better", label="Please enter your item:", container=True ) txtbxopt = gr.Textbox( label="The item you provided was classified as:", container=True ) single_button = gr.Button("Classify item") single_button.click( fn=classify_output, inputs=txtbx, outputs=txtbxopt ) with gr.Tab("Bulk CSV"): gr.Markdown(""" Upload a CSV file with one item per row. The CSV must contain a column named exactly: `item`. For large batches, such as 200 items, the app will return a downloadable CSV file only. """) csv_input = gr.File( label="Upload CSV file", file_types=[".csv"] ) bulk_button = gr.Button("Classify CSV") bulk_output_file = gr.File( label="Download classified CSV" ) bulk_button.click( fn=classify_bulk_csv, inputs=csv_input, outputs=bulk_output_file ) gr.Markdown(article_text) demo.launch()