Spaces:
Sleeping
Sleeping
| 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() | |