app.py

# type: ignore -- ignores linting import issues when using multiple virtual environments
import streamlit.components.v1 as components
import streamlit as st
import pandas as pd
import logging
from deeploy import Client, CreateEvaluation
from constants import (
    relationship_dict,
    occupation_dict,
    education_dict,
    type_of_work_dict,
    countries_dict,
    marital_status_dict,
)

# reset Plotly theme after streamlit import
import plotly.io as pio


pio.templates.default = "plotly"

logging.basicConfig(level=logging.INFO)

st.set_page_config(layout="wide")

st.title("Loan application model example")


def get_model_url():
    model_url = st.text_area(
        "Model URL (without the /explain endpoint, default is the demo deployment)",
        "https://api.app.deeploy.ml/workspaces/708b5808-27af-461a-8ee5-80add68384c7/deployments/dc8c359d-5f61-4107-8b0f-de97ec120289/",
        height=125,
    )
    elems = model_url.split("/")
    try:
        workspace_id = elems[4]
        deployment_id = elems[6]
    except IndexError:
        workspace_id = ""
        deployment_id = ""
    return model_url, workspace_id, deployment_id


def ChangeButtonColour(widget_label, font_color, background_color="transparent"):
    # func to change button colors
    htmlstr = f"""
        <script>
            var elements = window.parent.document.querySelectorAll('button');
            for (var i = 0; i < elements.length; ++i) {{ 
                if (elements[i].innerText == '{widget_label}') {{ 
                    elements[i].style.color ='{font_color}';
                    elements[i].style.background = '{background_color}'
                }}
            }}
        </script>
        """
    components.html(f"{htmlstr}", height=0, width=0)


with st.sidebar:
    st.image("deeploy_logo_wide.png", width=250)

    # Ask for model URL and token
    host = st.text_input("Host (Changing is optional)", "app.deeploy.ml")
    model_url, workspace_id, deployment_id = get_model_url()
    st.session_state.deployment_id = deployment_id
    deployment_token = st.text_input("Deeploy API token", "my-secret-token")
    if deployment_token == "my-secret-token":
        st.warning("Please enter Deeploy API token.")
    # Split model URL into workspace and deployment ID

    # st.write("Values below are for debug only:")
    # st.write("Workspace ID: ", workspace_id)
    # st.write("Deployment ID: ", deployment_id)

    client_options = {
        "host": host,
        "deployment_token": deployment_token,
        "workspace_id": workspace_id,
    }
    client = Client(**client_options)


if "expander_toggle" not in st.session_state:
    st.session_state.expander_toggle = True

if "evaluation_submitted" not in st.session_state:
    st.session_state.evaluation_submitted = False

if "predict_button_clicked" not in st.session_state:
    st.session_state.predict_button_clicked = False

if "request_body" not in st.session_state:
    st.session_state.request_body = None

if "deployment_id" not in st.session_state:
    st.session_state.deployment_id = None

if "exp" not in st.session_state:
    st.session_state.exp = None

def form_request_body():
    """Create the request body for the prediction endpoint"""
    marital_status_id = marital_status_dict[st.session_state.marital_status]
    native_country_id = countries_dict[st.session_state.native_country]
    relationship_id = relationship_dict[st.session_state.relationship]
    occupation_id = occupation_dict[st.session_state.occupation]
    education_id = education_dict[st.session_state.education]
    type_of_work_id = type_of_work_dict[st.session_state.type_of_work]
    return {
        "instances": [
            [
                st.session_state.age,
                type_of_work_id,
                education_id,
                marital_status_id,
                occupation_id,
                relationship_id,
                st.session_state.capital_gain,
                st.session_state.capital_loss,
                st.session_state.hours_per_week,
                native_country_id,
            ]
        ]
    }


def predict_callback():
    """Callback function to call the prediction endpoint"""
    request_body = form_request_body() # Make sure we have the latest values after user input
    st.session_state.exp = None
    with st.spinner("Loading prediction and explanation..."):
        # Call the explain endpoint as it also includes the prediction
        exp = client.explain(
            request_body=request_body, deployment_id=st.session_state.deployment_id
        )
        st.session_state.exp = exp

    st.session_state.predict_button_clicked = True
    st.session_state.evaluation_submitted = False    

def hide_expander():
    st.session_state.expander_toggle = False

def show_expander():
    st.session_state.expander_toggle = True

def submit_and_clear(agree: str, comment: str = None):
    if agree == "yes":
        evaluation_input: CreateEvaluation = {
            "agree": True,
            "comment": comment,
        }
    else:
        desired_output = not predictions[0]
        evaluation_input: CreateEvaluation = {
            "agree": False,
            "desired_output": { "predictions": [desired_output] },
            "comment": comment,
        }

    try:
        client.evaluate(st.session_state.deployment_id, prediction_log_id, evaluation_input)
        st.session_state.evaluation_submitted = True
        st.session_state.predict_button_clicked = False
        st.session_state.exp = None
        show_expander()
    except Exception as e:
        logging.error(e)
        st.error(
            "Failed to submit feedback."
            + "Check whether you are using the right model URL and token for evaluations. "
            + "Contact Deeploy if the problem persists."
        )

# with st.expander("Debug session state", expanded=False):
#     st.write(st.session_state)

# Attributes
with st.expander("**Loan application form**", expanded=st.session_state.expander_toggle):
    # Split view in 2 columns
    col1, col2 = st.columns(2)
    with col1:
        # Create input fields for attributes from constant dicts
        age = st.number_input("Age", min_value=10, max_value=100, value=30, key="age", on_change=predict_callback)
        marital_status = st.selectbox("Marital Status", marital_status_dict.keys(), key="marital_status", on_change=predict_callback,)
        native_country = st.selectbox(
            "Native Country", countries_dict.keys(), index=len(countries_dict) - 1, key="native_country",on_change=predict_callback
        )
        relationship = st.selectbox("Family situation", relationship_dict.keys(), key="relationship", on_change=predict_callback)
        occupation = st.selectbox("Occupation", occupation_dict.keys(), index=1, key="occupation", on_change=predict_callback)

    with col2:
        education = st.selectbox("Highest education level", education_dict.keys(), key="education", index=4, on_change=predict_callback)
        type_of_work = st.selectbox("Type of work", type_of_work_dict.keys(), key="type_of_work", on_change=predict_callback)
        hours_per_week = st.number_input(
            "Working hours per week", min_value=0, max_value=100, value=40, key="hours_per_week", on_change=predict_callback,
        )
        capital_gain = st.number_input(
            "Yearly income [€]", min_value=0, max_value=10000000, value=70000, key="capital_gain", on_change=predict_callback,
        )
        capital_loss = st.number_input(
            "Yearly expenditures [€]", min_value=0, max_value=10000000, value=60000, key="capital_loss", on_change=predict_callback,
        )
    data_df = pd.DataFrame(
        [
            [
                st.session_state.age,
                st.session_state.type_of_work,
                st.session_state.education,
                st.session_state.marital_status,
                st.session_state.occupation,
                st.session_state.relationship,
                st.session_state.capital_gain,
                st.session_state.capital_loss,
                st.session_state.hours_per_week,
                st.session_state.native_country,
            ]
        ],
        columns=[
            "Age",
            "Type of work",
            "Highest education level",
            "Marital Status",
            "Occupation",
            "Family situation",
            "Yearly Income [€]",
            "Yearly expenditures [€]",
            "Working hours per week",
            "Native Country",
        ],
    )
    data_df_t = data_df.T

# Show predict button if token is set
if deployment_token != "my-secret-token" and st.session_state.exp is None:
    predict_button = st.button(
        "Send loan application", key="predict_button", help="Click to get the AI prediction.", on_click=predict_callback,
    )        

if st.session_state.evaluation_submitted:
    st.success("Evaluation submitted successfully!")

# Show prediction and explanation after predict button is clicked
elif st.session_state.predict_button_clicked and st.session_state.exp is not None:
    try:
        exp = st.session_state.exp
        # Read explanation to dataframe from json
        predictions = exp["predictions"]
        request_log_id = exp["requestLogId"]
        prediction_log_id = exp["predictionLogIds"][0]

        exp_df = pd.DataFrame(
            [exp["explanations"][0]["shap_values"]], columns=exp["featureLabels"]
        )

        exp_df.columns = data_df.columns

        exp_df_t = exp_df.T

        # Merge data and explanation
        exp_df_t = data_df_t.merge(exp_df_t, left_index=True, right_index=True)

        weight_feat = "Weight"
        feat_val_col = "Value"
        exp_df_t.columns = [feat_val_col, weight_feat]
        exp_df_t["Feature"] = exp_df_t.index
        exp_df_t = exp_df_t[["Feature", feat_val_col, weight_feat]]
        exp_df_t[feat_val_col] = exp_df_t[feat_val_col].astype(str)
        
        # Filter values below 0.01
        exp_df_t = exp_df_t[
            (exp_df_t[weight_feat] > 0.01) | (exp_df_t[weight_feat] < -0.01)
        ]
        exp_df_t[weight_feat] = exp_df_t[weight_feat].astype(float).round(2)

        pos_exp_df_t = exp_df_t[exp_df_t[weight_feat] > 0]
        pos_exp_df_t = pos_exp_df_t.sort_values(by=weight_feat, ascending=False)

        neg_exp_df_t = exp_df_t[exp_df_t[weight_feat] < 0]
        neg_exp_df_t = neg_exp_df_t.sort_values(by=weight_feat, ascending=True)
        neg_exp_df_t[weight_feat] = neg_exp_df_t[weight_feat].abs()

        # Get 3 features with highest positive relevance score
        pos_feats = pos_exp_df_t[weight_feat].nlargest(3).index.tolist()
        # For feature, get feature value and concatenate into a single string
        pos_feats = [
            f"{feat}: {pos_exp_df_t.loc[feat, feat_val_col]}"
            for feat in pos_feats
        ]
        # Get 3 features with highest negative relevance score
        neg_feats = neg_exp_df_t[weight_feat].nlargest(3).index.tolist()
        # For feature, get feature value and concatenate into a single string
        neg_feats = [
            f"{feat}: {neg_exp_df_t.loc[feat, feat_val_col]}"
            for feat in neg_feats
        ]
        if predictions[0]:
            # Show prediction
            st.subheader("Loan Decision: :green[Approve]", divider="green")
            # Format subheader to green
            st.markdown(
                "<style>.css-1v3fvcr{color: green;}</style>", unsafe_allow_html=True
            )
            col1, col2 = st.columns(2)
            with col1:
                # If prediction is positive, first show positive features, then negative features
                st.success(
                    "The most important characteristics in favor of loan approval are:  \n - "
                    + "  \n- ".join(pos_feats)
                )
            with col2:
                st.error(
                    "However, the following features weight against the loan applicant:  \n - "
                    + "  \n- ".join(neg_feats)
                    # + "  \n For more details, see full explanation of the credit assessment below.",
                )
        else:
            st.subheader("Loan Decision: :red[Reject]", divider="red")
            col1, col2 = st.columns(2)
            with col1:
                # If prediction is negative, first show negative features, then positive features
                st.error(
                    "The most important reasons for loan rejection are:  \n - "
                    + "  \n - ".join(neg_feats)
                )
            with col2:
                st.success(
                    "However, the following factors weigh in favor of the loan applicant:  \n - "
                    + "  \n - ".join(pos_feats)
                )
        try:
            # Show explanation
            if predictions[0]:
                col_pos, col_neg = st.columns(2)
            else:
                col_neg, col_pos = st.columns(2) # Swap columns if prediction is negative
            with col_pos:
                st.subheader("Factors :green[in favor] of loan approval")
                # st.success("**Factors in favor of loan approval**")
                st.dataframe(
                    pos_exp_df_t,
                    hide_index=True,
                    width=600,
                    column_config={
                        "Weight": st.column_config.ProgressColumn(
                            "Weight",
                            width="small",
                            format=" ",
                            min_value=0,
                            max_value=1,
                        )
                    },
                )

            with col_neg:
                st.subheader("Factors :red[against] loan approval")
                # st.error("**Factors against loan approval**")
                st.dataframe(
                    neg_exp_df_t,
                    hide_index=True,
                    width=600,
                    column_config={
                        "Weight": st.column_config.ProgressColumn(
                            "Weight",
                            width="small",
                            format=" ",
                            min_value=0,
                            max_value=1,
                        )
                    },
                )
        except Exception as e:
            logging.error(e)
            st.error(
                "Failed to show the explanation."
                + "Refresh the page to reset the application."
                + "Contact Deeploy if the problem persists."
            )

        st.divider()

        if not st.session_state.evaluation_submitted:
            # Add prediction evaluation
            st.subheader("Evaluation: Do you agree with the loan assessment?")
            st.write(
                "AI model predictions always come with a certain level of uncertainty. Evaluate the correctness of the assessment based on your expertise and experience."
            )
            st.session_state.evaluation_input = {}
            comment = st.text_input("Your assessment:", placeholder="For example: 'Income is too low, given applicant's background'")

            cols = st.columns(4)
            col_yes, col_no = cols[:2]
            with col_yes:
                yes_button = st.button(
                    "Yes, I agree",
                    key="yes_button",
                    use_container_width=True,
                    help="Click if you agree with the prediction",
                    on_click=submit_and_clear,
                    args=["yes", comment]
                )
                ChangeButtonColour("Yes, I agree", "white", "green")

            with col_no:
                no_button = st.button(
                    "No, I disagree",
                    key="no_button",
                    use_container_width=True,
                    help="Click if you disagree with the prediction",
                    type="primary",
                    on_click=submit_and_clear,
                    args=["no"]
                )
                ChangeButtonColour("No, I disagree", "white", "#DD360C") # Red color for disagree button

    except Exception as e:
        logging.error(e)
        st.error(
            "Failed to retrieve the prediction or explanation."
            + "Check whether you are using the right model URL and Token. "
            + "Contact Deeploy if the problem persists."
        )