Spaces:
Runtime error
Runtime error
| import json | |
| import os | |
| import pickle | |
| import random | |
| import time | |
| from collections import Counter | |
| from datetime import datetime | |
| from glob import glob | |
| import gdown | |
| import matplotlib.pyplot as plt | |
| import numpy as np | |
| import pandas as pd | |
| import seaborn as sns | |
| import streamlit as st | |
| from PIL import Image | |
| import SessionState | |
| from download_utils import * | |
| from image_utils import * | |
| random.seed(datetime.now()) | |
| np.random.seed(int(time.time())) | |
| NUMBER_OF_TRIALS = 20 | |
| CLASSIFIER_TAG = "" | |
| explaination_functions = [load_chm_nns, load_knn_nns] | |
| selected_xai_tool = None | |
| # Config | |
| folder_to_name = {} | |
| class_descriptions = {} | |
| classifier_predictions = {} | |
| selected_dataset = "Final" | |
| root_visualization_dir = "./visualizations/" | |
| viz_url = "https://drive.google.com/uc?id=1LpmOc_nFBzApYWAokO2J-s9RRXsk3pBN" | |
| viz_archivefile = "Final.zip" | |
| demonstration_url = "https://drive.google.com/uc?id=1C92llG5VrlABrsIEvxfNlSDc_gIeLlls" | |
| demonst_zipfile = "demonstrations.zip" | |
| picklefile_url = "https://drive.google.com/uc?id=1Yx4abA4VLZGO5JkzhXVGdy6mbPltMd68" | |
| prediction_root = "./predictions/" | |
| prediction_pickle = f"{prediction_root}predictions.pickle" | |
| # Get the Data | |
| download_files( | |
| root_visualization_dir, | |
| viz_url, | |
| viz_archivefile, | |
| demonstration_url, | |
| demonst_zipfile, | |
| picklefile_url, | |
| prediction_root, | |
| prediction_pickle, | |
| ) | |
| ################################################ | |
| # GLOBAL VARIABLES | |
| app_mode = "" | |
| ## Shared/Global Information | |
| with open("imagenet-labels.json", "rb") as f: | |
| folder_to_name = json.load(f) | |
| with open("gloss.txt", "r") as f: | |
| description_file = f.readlines() | |
| class_descriptions = {l.split("\t")[0]: l.split("\t")[1] for l in description_file} | |
| ################################################ | |
| with open(prediction_pickle, "rb") as f: | |
| classifier_predictions = pickle.load(f) | |
| # SESSION STATE | |
| session_state = SessionState.get( | |
| page=1, | |
| first_run=1, | |
| user_feedback={}, | |
| queries=[], | |
| is_classifier_correct={}, | |
| XAI_tool="Unselected", | |
| ) | |
| ################################################ | |
| def resmaple_queries(): | |
| if session_state.first_run == 1: | |
| both_correct = glob( | |
| root_visualization_dir + selected_dataset + "/Both_correct/*.JPEG" | |
| ) | |
| both_wrong = glob( | |
| root_visualization_dir + selected_dataset + "/Both_wrong/*.JPEG" | |
| ) | |
| correct_samples = list( | |
| np.random.choice(a=both_correct, size=NUMBER_OF_TRIALS // 2, replace=False) | |
| ) | |
| wrong_samples = list( | |
| np.random.choice(a=both_wrong, size=NUMBER_OF_TRIALS // 2, replace=False) | |
| ) | |
| all_images = correct_samples + wrong_samples | |
| random.shuffle(all_images) | |
| session_state.queries = all_images | |
| session_state.first_run = -1 | |
| # RESET INTERACTIONS | |
| session_state.user_feedback = {} | |
| session_state.is_classifier_correct = {} | |
| def render_experiment(query): | |
| current_query = session_state.queries[query] | |
| query_id = os.path.basename(current_query) | |
| predicted_wnid = classifier_predictions[query_id][f"{CLASSIFIER_TAG}-predictions"] | |
| prediction_confidence = classifier_predictions[query_id][ | |
| f"{CLASSIFIER_TAG}-confidence" | |
| ] | |
| prediction_label = folder_to_name[predicted_wnid] | |
| class_def = class_descriptions[predicted_wnid] | |
| session_state.is_classifier_correct[query_id] = classifier_predictions[query_id][ | |
| f"{CLASSIFIER_TAG.upper()}-Output" | |
| ] | |
| ################################### SHOW QUERY and PREDICTION | |
| col1, col2 = st.columns(2) | |
| with col1: | |
| st.image(load_query(current_query), caption=f"Query ID: {query_id}") | |
| with col2: | |
| ################################### SHOW DESCRIPTION OF CLASS | |
| with st.expander("Show Class Description"): | |
| st.write(f"**Name**: {prediction_label}") | |
| st.write("**Class Definition**:") | |
| st.markdown("`" + class_def + "`") | |
| st.image( | |
| Image.open(f"demonstrations/{predicted_wnid}.jpeg"), | |
| caption=f"Class Explanation", | |
| use_column_width=True, | |
| ) | |
| default_value = 0 | |
| if query_id in session_state.user_feedback.keys(): | |
| if session_state.user_feedback[query_id] == "Correct": | |
| default_value = 1 | |
| elif session_state.user_feedback[query_id] == "Wrong": | |
| default_value = 2 | |
| session_state.user_feedback[query_id] = st.radio( | |
| "What do you think about model's prediction?", | |
| ("-", "Correct", "Wrong"), | |
| key=query_id, | |
| index=default_value, | |
| ) | |
| st.write(f"**Model Prediction**: {prediction_label}") | |
| st.write(f"**Model Confidence**: {prediction_confidence}") | |
| ################################### SHOW Model Explanation | |
| if selected_xai_tool is not None: | |
| st.image( | |
| selected_xai_tool(current_query), | |
| caption=f"Explaination", | |
| use_column_width=True, | |
| ) | |
| ################################### SHOW DEBUG INFO | |
| if st.button("Debug: Show Everything"): | |
| st.image(Image.open(current_query)) | |
| def render_results(): | |
| user_correct_guess = 0 | |
| for q in session_state.user_feedback.keys(): | |
| uf = True if session_state.user_feedback[q] == "Correct" else False | |
| if session_state.is_classifier_correct[q] == uf: | |
| user_correct_guess += 1 | |
| st.write( | |
| f"User performance on {CLASSIFIER_TAG}: {user_correct_guess} out of {len( session_state.user_feedback)} Correct" | |
| ) | |
| st.markdown("## User Performance Breakdown") | |
| categories = [ | |
| "Correct", | |
| "Wrong", | |
| ] # set(session_state.is_classifier_correct.values()) | |
| breakdown_stats_correct = {c: 0 for c in categories} | |
| breakdown_stats_wrong = {c: 0 for c in categories} | |
| experiment_summary = [] | |
| for q in session_state.user_feedback.keys(): | |
| category = "Correct" if session_state.is_classifier_correct[q] else "Wrong" | |
| is_user_correct = category == session_state.user_feedback[q] | |
| if is_user_correct: | |
| breakdown_stats_correct[category] += 1 | |
| else: | |
| breakdown_stats_wrong[category] += 1 | |
| experiment_summary.append( | |
| [ | |
| q, | |
| classifier_predictions[q]["real-gts"], | |
| folder_to_name[ | |
| classifier_predictions[q][f"{CLASSIFIER_TAG}-predictions"] | |
| ], | |
| category, | |
| session_state.user_feedback[q], | |
| is_user_correct, | |
| ] | |
| ) | |
| ################################### Summary Table | |
| experiment_summary_df = pd.DataFrame.from_records( | |
| experiment_summary, | |
| columns=[ | |
| "Query", | |
| "GT Labels", | |
| f"{CLASSIFIER_TAG} Prediction", | |
| "Category", | |
| "User Prediction", | |
| "Is User Prediction Correct", | |
| ], | |
| ) | |
| st.write("Summary", experiment_summary_df) | |
| csv = convert_df(experiment_summary_df) | |
| st.download_button( | |
| "Press to Download", csv, "summary.csv", "text/csv", key="download-records" | |
| ) | |
| ################################### SHOW BREAKDOWN | |
| user_pf_by_model_pred = experiment_summary_df.groupby("Category").agg( | |
| {"Is User Prediction Correct": ["count", "sum", "mean"]} | |
| ) | |
| # rename columns | |
| user_pf_by_model_pred.columns = user_pf_by_model_pred.columns.droplevel(0) | |
| user_pf_by_model_pred.columns = [ | |
| "Count", | |
| "Correct User Guess", | |
| "Mean User Performance", | |
| ] | |
| user_pf_by_model_pred.index.name = "Model Prediction" | |
| st.write("User performance break down by Model prediction:", user_pf_by_model_pred) | |
| csv = convert_df(user_pf_by_model_pred) | |
| st.download_button( | |
| "Press to Download", | |
| csv, | |
| "user-performance-by-model-prediction.csv", | |
| "text/csv", | |
| key="download-performance-by-model-prediction", | |
| ) | |
| ################################### CONFUSION MATRIX | |
| confusion_matrix = pd.crosstab( | |
| experiment_summary_df["Category"], | |
| experiment_summary_df["User Prediction"], | |
| rownames=["Actual"], | |
| colnames=["Predicted"], | |
| ) | |
| st.write("Confusion Matrix", confusion_matrix) | |
| csv = convert_df(confusion_matrix) | |
| st.download_button( | |
| "Press to Download", | |
| csv, | |
| "confusion-matrix.csv", | |
| "text/csv", | |
| key="download-confusiion-matrix", | |
| ) | |
| def render_menu(): | |
| # Render the readme as markdown using st.markdown. | |
| readme_text = st.markdown( | |
| """ | |
| # Instructions | |
| ``` | |
| When testing this study, you should first see the class definition, then hide the expander and see the query. | |
| ``` | |
| """ | |
| ) | |
| app_mode = st.selectbox( | |
| "Choose the page to show:", | |
| ["Experiment Instruction", "Start Experiment", "See the Results"], | |
| ) | |
| if app_mode == "Experiment Instruction": | |
| st.success("To continue select an option in the dropdown menu.") | |
| elif app_mode == "Start Experiment": | |
| # Clear Canvas | |
| readme_text.empty() | |
| page_id = session_state.page | |
| col1, col4, col2, col3 = st.columns(4) | |
| prev_page = col1.button("Previous Image") | |
| if prev_page: | |
| page_id -= 1 | |
| if page_id < 1: | |
| page_id = 1 | |
| next_page = col2.button("Next Image") | |
| if next_page: | |
| page_id += 1 | |
| if page_id > NUMBER_OF_TRIALS: | |
| page_id = NUMBER_OF_TRIALS | |
| if page_id == NUMBER_OF_TRIALS: | |
| st.success( | |
| 'You have reached the last image. Please go to the "Results" page to see your performance.' | |
| ) | |
| if st.button("View"): | |
| app_mode = "See the Results" | |
| if col3.button("Resample"): | |
| st.write("Restarting ...") | |
| page_id = 1 | |
| session_state.first_run = 1 | |
| resmaple_queries() | |
| session_state.page = page_id | |
| st.write(f"Render Experiment: {session_state.page}") | |
| render_experiment(session_state.page - 1) | |
| elif app_mode == "See the Results": | |
| readme_text.empty() | |
| st.write("Results Summary") | |
| render_results() | |
| def main(): | |
| global app_mode | |
| global session_state | |
| global selected_xai_tool | |
| global CLASSIFIER_TAG | |
| # Set the session state | |
| # State Management and General Setup | |
| st.set_page_config(layout="wide") | |
| st.title("Visual CorrespondenceHuman Study - ImageNet") | |
| options = [ | |
| "Unselected", | |
| "NOXAI", | |
| "KNN", | |
| "EMD-Corr Nearest Neighbors", | |
| "EMD-Corr Correspondence", | |
| "CHM-Corr Nearest Neighbors", | |
| "CHM-Corr Correspondence", | |
| ] | |
| st.markdown( | |
| """ <style> | |
| div[role="radiogroup"] > :first-child{ | |
| display: none !important; | |
| } | |
| </style> | |
| """, | |
| unsafe_allow_html=True, | |
| ) | |
| if session_state.XAI_tool == "Unselected": | |
| default = options.index(session_state.XAI_tool) | |
| session_state.XAI_tool = st.radio( | |
| "What explaination tool do you want to evaluate?", | |
| options, | |
| key="which_xai", | |
| index=default, | |
| ) | |
| # print(session_state.XAI_tool) | |
| if session_state.XAI_tool != "Unselected": | |
| st.markdown(f"## SELECTED METHOD ``{session_state.XAI_tool}``") | |
| if session_state.XAI_tool == "NOXAI": | |
| CLASSIFIER_TAG = "knn" | |
| selected_xai_tool = None | |
| elif session_state.XAI_tool == "KNN": | |
| selected_xai_tool = load_knn_nns | |
| CLASSIFIER_TAG = "knn" | |
| elif session_state.XAI_tool == "CHM-Corr Nearest Neighbors": | |
| selected_xai_tool = load_chm_nns | |
| CLASSIFIER_TAG = "CHM" | |
| elif session_state.XAI_tool == "CHM-Corr Correspondence": | |
| selected_xai_tool = load_chm_corrs | |
| CLASSIFIER_TAG = "CHM" | |
| elif session_state.XAI_tool == "EMD-Corr Nearest Neighbors": | |
| selected_xai_tool = load_emd_nns | |
| CLASSIFIER_TAG = "EMD" | |
| elif session_state.XAI_tool == "EMD-Corr Correspondence": | |
| selected_xai_tool = load_emd_corrs | |
| CLASSIFIER_TAG = "EMD" | |
| resmaple_queries() | |
| render_menu() | |
| if __name__ == "__main__": | |
| main() | |