utils/decisions_users.py

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

# 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((100,100))
    return img


def plot_n_similar(seed_id,similar_ids, test_path,n=10, scale=5):
    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)):
        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.5)
    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.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 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))
    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(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=(5, 4))
    sns.barplot(x='probability', y='class', data=tmp_df)
    middle_column.pyplot(fig)
    # middle_column.write("Probabilities")

    # grad img 
    right_column.image(roi_img, caption = "Decision 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 Set')
    st.pyplot(plot_n_similar(seed_id,closest_fns_tmp, img_path,n=10,scale=3))

def main():

    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 or Borderline Samples', ["Representative Samples","Borderline Cases"])
        if option:
            clss = st.selectbox('Select a category(class)', ["CNV","DME", "NORMAL", "DRUSEN"])
            side_1, side_2 = st.columns(2)

        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, then click Explain Me button', IMG_PATH_LISTS)
        with row2_col2:
            st.info("Press the button")
            pressed = st.button('Explain ME')
        
        if pressed:
            st.empty()
            st.sidebar.write('Please wait for the magic to happen! This may take up to a minute.')
            run_instance_exp_keras_model(option, new_model,feature_extractor_model)


        # # 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():
    return main()