import os import sys import json import shutil import urllib.request from pathlib import Path import pathlib import time import urllib from ast import literal_eval # import albumentations as A import tensorflow as tf import cv2 import numpy as np import pandas as pd import plotly.express as px import matplotlib.pyplot as plt from PIL import Image import streamlit as st import seaborn as sns sys.path.append(f'{os.getcwd()}/utils') from utils.eval_users import get_product_dev_page_layout # print(os.getcwd()) # Hide GPU from visible devices tf.config.set_visible_devices([], 'GPU') # Enable GPU memory growth - avoid allocating all memory at start # gpus = tf.config.experimental.list_physical_devices(device_type='GPU') # for gpu in gpus: # tf.config.experimental.set_memory_growth(device=gpu, enable=True) from utils.control import show_tsne_vis,show_random_samples from utils.annoy_sampling import load_annoy_tree from utils.model_utils import load_model from utils.model_utils import get_feature_vector, get_feature_extractor_model, get_predictions_and_roi sns.set_style('darkgrid') plt.rcParams['axes.grid'] = False # import tensorflow as tf os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2' # Suppress TensorFlow logging (1) tf.get_logger().setLevel('ERROR') # Suppress TensorFlow logging (2) # st.set_page_config(layout="wide") #https://github.com/IliaLarchenko/albumentations-demo/blob/3cb6528a513fe3b35dbb2c2a63cdcfbb9bb2a932/src/utils.py#L149 GRAD_CAM_IMAGE_DIR = f'{os.getcwd()}/data/gradcam_vis_data/' TEST_CSV_FILE = f'{os.getcwd()}/data/test_set_pred_prop.csv' annoy_tree_save_path =f'{os.getcwd()}/data/filtered_train_embedding/representative_samples_emb.annoy' test_emb_path = f'{os.getcwd()}/data/filtered_train_embedding/test_embeddings.npy' test_emb_id_path =f'{os.getcwd()}/data/filtered_train_embedding/test_ids.npy' train_emb_id_path =f'{os.getcwd()}/data/filtered_train_embedding/representative_train_ids.npy' repr_id_path =f'{os.getcwd()}/data/filtered_train_embedding/representative_train_ids.npy' borderline_id_path =f'{os.getcwd()}/data/filtered_train_embedding/borderline_train_ids.npy' MODEL_PATH = f'{os.getcwd()}/model/keras_model_0422/' ROOT_FIG_DIR = f'{os.getcwd()}/figures/' test_emb = np.load(test_emb_path) test_ids = np.load(test_emb_id_path) test_id_list = list(test_ids) test_labels = [_id.split("\\")[1] for _id in test_id_list] print(" NUmber of test samples: ",len(test_labels)) test_features = test_emb.reshape(-1,1792) # train embedding list train_ids = np.load(train_emb_id_path) train_id_list = list(train_ids) train_labels = [_id.split("\\")[1] for _id in train_id_list] print(" NUmber of training samples: ",len(train_labels)) annoy_tree = load_annoy_tree(test_features.shape[1],annoy_tree_save_path) def annoy_matching(annoy_f,query_item, query_index, n=10): return annoy_f.get_nns_by_vector(query_item, n) def get_img(fn ,thumbnail=False): img = Image.open(fn) if thumbnail: img.thumbnail((150,150)) return img def open_gray(fn): img = cv2.cvtColor(cv2.imread(fn), cv2.COLOR_BGR2GRAY) img = cv2.cvtColor(img, cv2.COLOR_GRAY2RGB) img = cv2.resize(img,(224,224)) return img def show_values(axs, orient="v", space=.01): def _single(ax): if orient == "v": for p in ax.patches: _x = p.get_x() + p.get_width() / 2 _y = p.get_y() + p.get_height() + (p.get_height()*0.01) value = '{:.1f}'.format(p.get_height()) ax.text(_x, _y, value, ha="center") elif orient == "h": for p in ax.patches: _x = p.get_x() + p.get_width() + float(space) _y = p.get_y() + p.get_height() - (p.get_height()*0.5) value = '{:.2f}'.format(p.get_width()) ax.text(_x, _y, value, ha="left") if isinstance(axs, np.ndarray): for idx, ax in np.ndenumerate(axs): _single(ax) else: _single(axs) def plot_n_similar(seed_id,similar_ids, test_path,n=10, scale=5): # img_list = [] # title_list = ["SEED ID:{0}
Label:{1}".format(seed_id,os.path.basename(test_labels[seed_id]))] # # for indx,row in cnv_cls_df.iterrows(): # img_list.append(open_gray(test_path.replace("F:/","E:/"))) # for i in range(len(similar_ids)): # print("PATH:",similar_ids[i].replace("F:/","E:/")) # img_list.append(open_gray(similar_ids[i].replace("F:/","E:/"))) # # ax[i+1].imshow(get_img(similar_ids[i].replace("F:/","E:/")),cmap='gray') # title = "ID:{0}
Distance: {1:.3f}
Label:{2}".format(i,0.1223,os.path.basename(similar_ids[i])[:-4]) # title_list.append(title) # fig = px.imshow(np.array(img_list), facet_col=0, binary_string=True,facet_row_spacing=0.002,facet_col_spacing=0.002) # # Set facet titles # for i, sigma in enumerate(title_list): # fig.layout.annotations[i]['text'] = sigma # fig.layout.annotations[i]['yshift'] = -40 # # fig.layout.tex # fig.update_layout( # margin=dict( # l=10, # r=10, # b=10, # t=40, # pad=1 # ), # ) # fig.update_xaxes(showticklabels=False) # fig.update_yaxes(showticklabels=False) # fig.show() f,ax = plt.subplots(1,n+1,figsize=((n+1)*scale,scale)) # print(os.path.basename(test_labels[seed_id])[:-4]) title = "SEED ID:{0}\nLabel:{1}".format(seed_id,os.path.basename(test_labels[seed_id])) # print("path:", test_labels[seed_id].replace("F:/","E:/")) ax[0].imshow(get_img(test_path.replace("F:/","E:/")),cmap='gray') ax[0].set_title(title,fontsize=12) for i in range(len(similar_ids)): # print("PATH:", similar_ids[i]) ax[i+1].imshow(get_img(similar_ids[i].replace("F:/","E:/")),cmap='gray') title = "ID:{0}\nDistance: {1:.3f}\nLabel:{2}".format(i,0.1223,os.path.basename(similar_ids[i])[:-4]) ax[i+1].set_title(title,fontsize=10) f.suptitle("Images similar to seed_id {0}".format(seed_id),fontsize=18) plt.subplots_adjust(top=0.4) plt.tight_layout() return f def load_image(filename,change_url=True): # if change_url: # print(filename) # print(os.path.exists(filename)) img = cv2.imread(filename) return img @st.cache(allow_output_mutation=True) def get_model(model_path): new_model = tf.keras.models.load_model(model_path) # keras_model = load_model(model_path) return new_model @st.cache(allow_output_mutation=True) def get_feature_vector_model(model_path): keras_model = tf.keras.models.load_model(model_path) feature_extractor = tf.keras.Model(keras_model.inputs,keras_model.layers[-3].output) return feature_extractor def load_pd_data_frame(df_csv_path): return pd.read_csv(df_csv_path) def get_path_list_from_df(df_data): return list(df_data['path']) def get_class_probs_from_df(df_data): return list(df_data['class_probs']) def visualize_bar_plot(df_data): fig = px.bar(df_data, x="probability", y="class", orientation='h') return fig # def run_instance_exp(img_path, img_path_list,prob_list,grad_vis_path_list): # st.subheader('Instance Exploration') # # st.columns((1,1,1)) with row4_2: # LABELS = ['CNV', 'DRUSEN', 'DME', 'NORMAL'] # left_column, middle_column, right_column = st.columns((1,1,1)) # display_image = load_image(img_path) # # fig = px.imshow(display_image) # # left_column.plotly_chart(fig, use_container_width=True) # left_column.image(cv2.resize(display_image, (180,180)),caption = "Selected Input") # # get class probabilities # indx = img_path_list.index(img_path) # print(img_path) # prb_tmp = prob_list[indx] # print(f"{prb_tmp[1:-1]}") # clss_probs = literal_eval('"'+prb_tmp[1:-1]+'"') # print(clss_probs[1:-1].split(' ')) # prob_cls = [float(p) for p in clss_probs[1:-1].split(' ')] # tmp_df = pd.DataFrame.from_dict({'class':LABELS,'probability':prob_cls}) # print(tmp_df.head()) # fig = plt.figure(figsize=(15, 13)) # sns.barplot(x='probability', y='class', data=tmp_df) # middle_column.pyplot(fig) # # st.caption('Predictions') # tmp_grad_img = GRAD_CAM_IMAGE_DIR + img_path.split("\\")[-2] +'/'+img_path.split("\\") [-1] # display_image = load_image(tmp_grad_img,replace=False) # # left_column.plotly_chart(fig, use_container_width=True) # right_column.image(display_image,caption = "ROI") # # seed_id = 900 # seed_id = test_id_list.index(img_path) # query_item = test_features[seed_id] # print(query_item.shape) # closest_idxs = annoy_matching(annoy_tree,query_item, seed_id, 10) # closest_fns = [train_ids[close_i] for close_i in closest_idxs] # st.subheader('Top-10 Similar Samples from Gallery Set') # st.plotly_chart(plot_n_similar(seed_id,closest_fns, img_path,n=10, scale=4), use_container_width=True) # # st.pyplot(plot_n_similar(seed_id,closest_fns, img_path,n=10, scale=4)) def run_instance_exp_keras_model(img_path, new_model, feature_extractor_model): # st.subheader('Instance Predictions') # st.columns((1,1,1)) with row4_2: LABELS = ['CNV', 'DRUSEN', 'DME', 'NORMAL'] left_column, middle_column, right_column = st.columns((1,1,1)) print(img_path) org_img_path = img_path img_path = f'{os.getcwd()}/data/oct2017/test/' + img_path.split("\\")[-2] +'/'+img_path.split("\\") [-1] # img_path.replace("F:/XAI/data/OCT2017/","/home/hodor/dev/Learning/XAI/streamlit_demo/multipage-app/data/xai_framework_data/") display_image = load_image(img_path) # fig = px.imshow(display_image) # left_column.plotly_chart(fig, use_container_width=True) left_column.image(display_image) left_column.write("Input Image") # left_column.image(cv2.resize(display_image, (180,180)),caption = "Selected Input") roi_img, probs = get_predictions_and_roi(img_path, new_model) ## probs # print(np.asarray(probs)) # print(probs.shape) prob_cls =np.asarray(probs)[0] # print(prob_cls) tmp_df = pd.DataFrame.from_dict({'class':LABELS,'probability':prob_cls}) fig = plt.figure(figsize=(8, 8.8)) p =sns.barplot(x='probability', y='class', data=tmp_df) show_values(p, "h", space=0.05) middle_column.pyplot(fig) middle_column.write("Predicted Class Probabilities") # fig = px.bar(LABELS, prob_cls) # fig = px.bar(tmp_df, x="class", y="probability", orientation='h') # middle_column.plotly_chart(fig, use_container_width=False) # grad img print("roi image stats", roi_img.shape) # right_column.image(roi_img, caption = "Decision ROI") print(display_image.shape) tmp_shape = display_image.shape[:2] right_column.image(cv2.resize(roi_img, (tmp_shape[1],tmp_shape[0]))) right_column.write("GradCAM RoI") # seed_id = 900 seed_id = test_id_list.index(org_img_path) query_item = get_feature_vector(img_path,feature_extractor_model) query_item = query_item.reshape(-1,1792) # print(query_item.shape) closest_idxs = annoy_matching(annoy_tree,query_item[0,:], seed_id, 10) closest_fns = [train_ids[close_i] for close_i in closest_idxs] closest_fns_tmp = [f'{os.getcwd()}/data/oct2017/train/' + each_fn.split("\\")[-2] +'/'+each_fn.split("\\") [-1] for each_fn in closest_fns] # print(closest_fns) st.subheader('Top-10 Similar Samples from Gallery(representative) Set') # st.plotly_chart(plot_n_similar(seed_id,closest_fns, img_path,n=10, scale=4), use_container_width=True) st.pyplot(plot_n_similar(seed_id,closest_fns_tmp, img_path,n=10,scale=4)) if "load_state" not in st.session_state: st.session_state.load_state = True # if st.session_state.load_state: # st.session_state.load_state = False # with st.expander('Correct Decision'): # st.info("Correct th decision if it is not valid. This sample will be added to next training bucket.") # tmp_col1, tmp_col2 = st.columns(2) # with tmp_col1: # label_correect = st.radio( # "Choose label visibility 👇", # ["CNV", "DME", "NORMAL","DRUSEN"], # disabled=False, # horizontal=True) # with tmp_col2: # tmp_btn = st.button('ADD TO TRAINING BUCKET') # if tmp_btn: # st.warning("Sample added to training set..") def main(user_type='Developer'): new_model = get_model(MODEL_PATH) feature_extractor_model = get_feature_vector_model(MODEL_PATH) row4_1, row4_2 = st.tabs(["Global Level Explanations", "Instance Level Explanations"]) with row4_1: borderline_cases = np.load(borderline_id_path) representative_cases = np.load(repr_id_path) borderline_id_list = list(borderline_cases) # print(borderline_id_list) borderline_id_list = [f'{os.getcwd()}/data/oct2017/train/' + each_fn.split("\\")[-2] +'/'+each_fn.split("\\") [-1] for each_fn in borderline_id_list] representative_id_list = list(representative_cases) representative_id_list = [f'{os.getcwd()}/data/oct2017/train/' + each_fn.split("\\")[-2] +'/'+each_fn.split("\\") [-1] for each_fn in representative_id_list] # st.info('GLOABAL EXPLANATION!! ') option = st.selectbox('Please select to explore Representative/Borderline Samples', ["Choose here","Representative Samples","Borderline Cases"],index=0) if not option.startswith("Choose"): if user_type!='Manager': if option.startswith('Rep'): with st.expander('Click to see Representative Sampling Algorithm'): algo_path = f'{ROOT_FIG_DIR}/representativesampling.png' st.image(algo_path) with st.expander('Click to see Manifold(t-SNE) Visualization of Representative Samples'): show_tsne_vis(f"{ROOT_FIG_DIR}/tsne_representative.png", title="Representative") else: with st.expander('Click to see Borderline Sampling Algorithm'): algo_path = f'{ROOT_FIG_DIR}/borderlinesampling.png' st.image(algo_path) with st.expander('Click to see Manifold(t-SNE) Visualization of Broderline Samples'): show_tsne_vis(f"{ROOT_FIG_DIR}/tsne_borderline.png", title="Borderline") clss = st.selectbox('Select a category(class)', ["CNV","DME", "NORMAL", "DRUSEN"]) # side_1, side_2 = st.columns(2) if option.startswith("Rep"): filter_lst = list(filter(lambda k: clss in k, representative_id_list)) show_random_samples(filter_lst,clss) else: filter_lst = list(filter(lambda k: clss in k, borderline_id_list)) show_random_samples(filter_lst,clss) # with side_1: # check_emb = st.checkbox('Embdedding Space Visuzalization') # with side_2: # check_samp = st.checkbox('Random Sample Visuzalization') # if check_emb and check_samp: # st.write("Emb and vis") # if option.startswith("Rep"): # filter_lst = list(filter(lambda k: clss in k, representative_id_list)) # show_random_samples(filter_lst,clss) # show_tsne_vis(f"{ROOT_FIG_DIR}/tsne_representative.png", title="Representative") # else: # filter_lst = list(filter(lambda k: clss in k, borderline_id_list)) # show_random_samples(filter_lst,clss) # # show_tsne_vis(f"{ROOT_FIG_DIR}/tsne_borderline.png", title="Borderline") # elif check_emb: # st.write("embedding vis") # if option.startswith("Rep"): # show_tsne_vis(f"{ROOT_FIG_DIR}/tsne_representative.png", title="Representative") # else: # show_tsne_vis(f"{ROOT_FIG_DIR}/tsne_borderline.png", title="Borderline") # elif check_samp: # st.write("rand vis") # if option.startswith("Rep"): # filter_lst = list(filter(lambda k: clss in k, representative_id_list)) # show_random_samples(filter_lst,clss) # else: # filter_lst = list(filter(lambda k: clss in k, borderline_id_list)) # show_random_samples(filter_lst,clss) with row4_2: DF_TEST_PROP = load_pd_data_frame(TEST_CSV_FILE) IMG_PATH_LISTS = get_path_list_from_df(DF_TEST_PROP) IMG_CLSS_PROBS_LIST = get_class_probs_from_df(DF_TEST_PROP) grad_vis_path_list = None row2_col1, row2_col2 = st.columns(2) with row2_col1: option = st.selectbox('Please select a sample image👇', IMG_PATH_LISTS) with row2_col2: st.write("Click button") pressed = st.button('Explain ME') if pressed: st.empty() st.write('Please wait for a while! This may take up to a minute.') run_instance_exp_keras_model(option, new_model,feature_extractor_model) def form_callback(): st.write("Training set updated") # st.write("test2") if user_type!='Manager': with st.expander('Correct the Decision'): with st.form("my_form"): st.info("Correct the decision if it is not valid. The sample will be added to next training bucket.") tmp_col1, tmp_col2 = st.columns(2) with tmp_col1: label_correect = st.radio( "Choose label 👇", ["NONE","CNV", "DME", "NORMAL","DRUSEN"], key="visibility", horizontal=True) # st.stop() # Every form must have a submit button. submit_button = st.form_submit_button(label='ADD TO TRAINING BUCKET', on_click=form_callback) # st.session_state.load_state = False # if submitted and not st.session_state.load_state: # st.warning("Sample added to training set..") # st.write("Outside the form") # st.write("slider", slider_val, "checkbox", checkbox_val) # with tmp_col2: # tmp_btn = st.button('ADD TO TRAINING BUCKET') # if tmp_btn: # st.warning("Sample added to training set..") # # new_model = load_model(MODEL_PATH) # option = st.sidebar.selectbox('Please select a sample image, then click Explain Me button', IMG_PATH_LISTS) # pressed = st.sidebar.button('Explain ME') # main() # expander_faq = st.expander("More About Our Project") # expander_faq.write("Hi there! If you have any questions about our project, or simply want to check out the source code, please visit our github repo: https://github.com/kaplansinan/MLOPS") def get_product_dev_page_layout(user_type ="Developer"): return main(user_type=user_type) # def get_product_dev_page_layout(): # return main()