Interface_v1 / app.py
sehyun
Updated date format
d2affc1
from datetime import datetime
import gradio as gr
import nltk
import numpy as np
import pandas as pd
import torch
from flair.data import Sentence
from flair.embeddings import TransformerDocumentEmbeddings
from imblearn.over_sampling import SMOTE
from lime.lime_text import LimeTextExplainer
from nltk.corpus import stopwords
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.svm import SVC
# from importlib.metadata import version
# print(version('scikit-learn'))
smote = SMOTE(sampling_strategy='minority', random_state=42)
nltk.download('stopwords')
y = np.array([2, 1, 5, 3, 0, 3, 2, 1, 2, 2, 0, 2, 0, 2, 1, 4, 2, 0, 1, 4, 2, 2,
0, 1, 4, 2, 1, 2, 1, 2, 2, 0, 2, 0, 0, 0, 3, 2, 1, 0, 0, 0, 0, 0,
0, 3, 3, 1, 2, 2, 0, 2, 4, 0, 0, 0, 3, 0, 3, 0, 2, 2, 2, 2, 1, 1,
1, 1, 2, 2, 0, 1, 1, 3, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 3, 3, 1,
1, 1, 3, 3, 1, 0, 3, 1, 0, 2, 2, 2, 3, 1, 2, 2, 0, 2, 2, 3, 1, 0,
3, 1, 1, 1, 2, 0, 3, 0, 4, 0, 4, 2, 2, 3, 3, 0, 3, 0, 1, 1, 0, 6,
1, 3, 4, 0, 1, 1, 2, 2, 1, 1, 0, 3, 1, 1, 3, 3, 1, 1, 0, 0, 1, 3,
2, 0, 2, 3, 0, 4, 4, 4, 2, 2, 0, 4, 1, 3, 1, 2, 1, 2, 0, 1, 1, 1,
1, 2, 0, 0, 0, 2, 0, 0, 0, 4, 1, 5, 0, 6, 0, 1, 3, 1, 1, 1, 1, 1,
0, 0, 0, 1, 0, 2, 0, 0, 0, 2, 1, 1, 1, 1, 2, 1, 4, 4, 4, 0, 1, 0,
1, 1, 1, 1, 1, 3, 0, 3, 3, 1, 0, 1, 1, 0, 3, 4, 3, 3, 4, 3, 4, 4,
0, 4, 2, 1, 1, 0, 0, 1, 1, 1, 5, 0, 3, 4, 4, 4, 1, 0, 6, 4, 4, 2,
4, 0, 3, 1, 3, 0, 0, 0, 0, 2, 0, 0, 1, 4, 1, 0, 1, 0, 0, 0, 4, 0,
0, 3, 0, 4, 0, 4, 0, 0, 4, 3, 3, 0, 3, 5, 0, 0, 2, 1, 0, 0, 0, 0,
0, 0, 3, 0, 5, 5, 5, 3, 0, 3, 0, 2, 0, 0, 3, 0, 1, 2, 3, 0, 0, 5,
0, 0, 4, 0, 0, 0, 4, 5, 5, 0, 4, 0, 0, 4, 4, 2, 4, 0, 0, 0, 0, 0,
2, 3, 6, 0, 0, 1, 2, 3, 1, 1, 0, 0, 2, 0, 0, 0, 0, 0, 4, 2, 0, 0,
3, 3, 4, 0, 0, 0, 3, 0, 1, 5, 0, 5, 5, 5, 5, 0, 0, 0, 4, 5, 0, 3,
5, 0, 0, 1, 2, 0, 1, 0, 2, 2, 2, 1, 5, 0, 0, 0, 4, 0, 3, 0, 2, 4,
4, 0, 1, 0, 2, 3, 4, 1, 3, 5, 5, 2, 2, 1, 3, 1, 1, 2, 0, 0, 4, 3,
2, 0, 0, 0, 0, 4, 3, 2, 0, 2, 4, 0, 4, 0, 1, 1, 2, 1, 2, 2, 0, 3,
0, 0, 0, 0, 1, 2, 0, 4, 4, 1, 4, 2, 0, 0, 6, 0, 3, 2, 1, 1, 2, 4,
2, 3, 2, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 3, 2, 2, 0, 0, 2, 2, 0,
4, 0, 2, 0, 5, 1, 4, 0, 0, 0, 4, 1, 1, 1, 1, 1, 4, 1, 5, 2, 2, 0,
2, 0, 1, 0, 1, 0, 1, 4, 4, 3, 0, 6, 0, 0, 0, 1, 3, 1, 5, 2, 5, 1,
3, 3, 5, 5, 6, 6, 6, 5, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 5, 6, 6,
6, 6, 5, 6, 6, 6, 6, 5, 6, 5, 6, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5,
5, 5, 6, 5, 5, 5, 5, 5, 5, 5, 5, 6, 5, 5, 5, 5, 6, 6, 6, 6, 6, 6,
6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6])
s_4 = torch.load('s_4.pt')
embedding_np_7 = s_4.numpy() # converting from tensor to numpy
xtrain_rb, xvalid_rb, ytrain_rb, yvalid_rb = train_test_split(embedding_np_7, y,
stratify=y,
random_state=10, test_size=0.2, shuffle=True)
standard_transformer = StandardScaler()
xtrain_rb = standard_transformer.fit_transform(xtrain_rb)
xvalid_rb = standard_transformer.transform(xvalid_rb)
smote = SMOTE(sampling_strategy='minority', random_state=42)
xtrain_rb, ytrain_rb = smote.fit_resample(xtrain_rb, ytrain_rb)
xtrain_rb, ytrain_rb = smote.fit_resample(xtrain_rb, ytrain_rb)
xtrain_rb, ytrain_rb = smote.fit_resample(xtrain_rb, ytrain_rb)
xtrain_rb, ytrain_rb = smote.fit_resample(xtrain_rb, ytrain_rb)
xtrain_rb, ytrain_rb = smote.fit_resample(xtrain_rb, ytrain_rb)
xtrain_rb, ytrain_rb = smote.fit_resample(xtrain_rb, ytrain_rb)
pse_classifier = SVC(C=3.2, probability=True, class_weight='balanced', kernel='rbf',
gamma=0.0008) # since we need probabilities
pse_classifier.fit(xtrain_rb, ytrain_rb) # fit the classifier with model
predictions16 = pse_classifier.predict(xvalid_rb)
transformer_embedding = TransformerDocumentEmbeddings('roberta-base')
def classify(incident_description):
Event_types = [
"Care coordination / communication",
"Laboratory test",
"Medication related",
"Omission / errors in assessment, diagnosis, monitoring",
"Maternal",
"Equipment / devices",
"Supplies"]
def predictor(text_list):
proba_list = []
for text in text_list:
sentence = Sentence(text)
transformer_embedding.embed(sentence)
inpt = sentence.embedding.view(1, 768)
inpt = inpt.cpu()
inpt = standard_transformer.transform(inpt)
# Predicted probability for the input sentence
predicted_proba = pse_classifier.predict_proba(inpt)
proba_list.append(predicted_proba)
proba_array = np.array(proba_list)
proba_array = np.squeeze(proba_array, axis=1)
return proba_array
explainer = LimeTextExplainer(class_names=Event_types)
# explanations
exp = explainer.explain_instance(incident_description, predictor, num_features=8, num_samples=50, top_labels=1)
available_labels = exp.available_labels()
# Iterate over the desired labels
for label in range(7): # Assuming you want to try labels 0 through 6
if label in available_labels:
words = exp.as_list(label=label)
string_XAI = ""
words_num_xai = 5
for i in range(words_num_xai):
string_XAI += words[i][0] + " "
# Recommended event type
sentence = Sentence(incident_description)
transformer_embedding.embed(sentence)
inpt = sentence.embedding
inpt = inpt.cpu()
inpt = inpt.numpy()
inpt = inpt.reshape(1, -1)
inpt = standard_transformer.transform(inpt)
# Predicted probability for the input sentence
predicted_proba = pse_classifier.predict_proba(inpt)
# Using np.argsort to get the indices of all elements in sorted order and then taking the last two (top two)
top_two_indices = np.argsort(predicted_proba[0])[-2:]
# Reversing to ensure the highest probability is first
top_two_indices = top_two_indices[::-1]
# Getting the top two probability values
top_two_probabilities = predicted_proba[0][top_two_indices]
# top_two_indices, top_two_probabilities
top_two_event_types = [Event_types[i] for i in top_two_indices]
output = {
'Event type': [top_two_event_types[0], top_two_event_types[1]],
'Probability (%)': [round(top_two_probabilities[0] * 100, 2), round(top_two_probabilities[1] * 100, 2)]
}
df = pd.DataFrame(output)
return df, string_XAI
def classify(incident_description):
Event_types = [
"Care coordination / communication",
"Laboratory test",
"Medication related",
"Omission / errors in assessment, diagnosis, monitoring",
"Maternal",
"Equipment / devices",
"Supplies"]
def predictor(text_list):
proba_list = []
for text in text_list:
sentence = Sentence(text)
transformer_embedding.embed(sentence)
inpt = sentence.embedding.view(1, 768)
inpt = inpt.cpu()
inpt = standard_transformer.transform(inpt)
# Predicted probability for the input sentence
predicted_proba = pse_classifier.predict_proba(inpt)
proba_list.append(predicted_proba)
proba_array = np.array(proba_list)
proba_array = np.squeeze(proba_array, axis=1)
return proba_array
explainer = LimeTextExplainer(class_names=Event_types)
# explanations
exp = explainer.explain_instance(incident_description, predictor, num_features=50, num_samples=300, top_labels=1)
available_labels = exp.available_labels()
# Iterate over the desired labels
for label in range(7): # Assuming you want to try labels 0 through 6
if label in available_labels:
words = exp.as_list(label=label)
string_XAI = []
score_XAI = []
stop_words = set(stopwords.words('english'))
stop_words = list(stop_words)
words_num_xai = 5
for i in range(words_num_xai):
if words[i][0] not in stop_words:
string_XAI.append(words[i][0])
score_XAI.append(words[i][1])
def softmax(x):
e_x = np.exp(x - np.max(x))
return e_x / e_x.sum().tolist()
score_XAI = softmax(score_XAI)
explanation_highlighted = []
incident_description_splitted = incident_description.split()
for word in incident_description_splitted:
if word in string_XAI:
for i in range(len(score_XAI)):
if string_XAI[i] == word:
explanation_highlighted.append((word, score_XAI[i]))
else:
explanation_highlighted.append((word, None))
# Recommended event type
sentence = Sentence(incident_description)
transformer_embedding.embed(sentence)
inpt = sentence.embedding
inpt = inpt.cpu()
inpt = inpt.numpy()
inpt = inpt.reshape(1, -1)
inpt = standard_transformer.transform(inpt)
# Predicted probability for the input sentence
predicted_proba = pse_classifier.predict_proba(inpt)
# Using np.argsort to get the indices of all elements in sorted order and then taking the last two (top two)
top_two_indices = np.argsort(predicted_proba[0])[-2:]
# Reversing to ensure the highest probability is first
top_two_indices = top_two_indices[::-1]
# Getting the top two probability values
top_two_probabilities = predicted_proba[0][top_two_indices]
# top_two_indices, top_two_probabilities
top_two_event_types = [Event_types[i] for i in top_two_indices]
output = {
'Event type': [top_two_event_types[0], top_two_event_types[1]],
'Probability (%)': [round(top_two_probabilities[0] * 100, 2), round(top_two_probabilities[1] * 100, 2)]
}
df = pd.DataFrame(output)
return df, explanation_highlighted
incident_description_placeholder = "Please provide factual description of the incident."
Event_types = [
"Care coordination / communication",
"Laboratory test",
"Medication related",
"Omission / errors in assessment, diagnosis, monitoring",
"Maternal",
"Equipment / devices",
"Supplies"]
webpage_name = "PSE Report Classification Interface"
sites = ["Royal Health Hospital", "Queen's Medical Centre", "Princess Health Hospital"]
departments = ['Labour & Delivery', 'Dermatology', 'Emergency', 'Heart & Vascular',
'Pediatric', 'Surgery', 'Respirology', 'Neurology', 'Other--Please specify by typing', 'Not applicable']
harm_levels = ["Unsafe condition", "Near miss", "No harm evident", "emotional distress", "additional treatment",
"temporary harm", "permanent harm", "severe permanent harm", "Death"]
contributing_factors = ["Patient factors", "Provider factors", "Task factors", "Team factors", "Training and education",
'Information technology', "Institutional environment", 'Not Sure']
professional_roles = ['Physician', 'Nurse', 'Pharmacist', 'Radiologst', 'Legal',
'Technician', 'Social worker', 'Dietician', 'Other--Please specify by typing']
PSE_examples = [["Specimen collection time and collector NetID were not documented in Epic for the ABORh."],
["I needed to start Pitocin on my pt , no pump channels available to use. This is a daily occurrence"],
[
"azithromycin 1g po daily without a stop date paged MD since it should be x1 dose changed to x1 dose instead of never ending"],
[
"After doing my chart review this morning in preparation for my Lactation Consult I noted that it was documented that pt was set up to pump 0m 9/23/19 at 0022 by Tracey Rudolf, RN. However, when I went to discuss the patient with present nurse caring for the her, I was told that patient was not set up to pump. The pump and the pump kit were in the room but the kit was not open. Mom was in the OR at the time I went into her room so I was not able to set mom up to pump. Unfortunately mom developed pre-eclampsia while in the PACU and did not return to her room until 1630. At that time baby was being transferred to Level One nursery."]]
event_type_tooltips = {
"Care coordination / communication": "Miscommunication/mismanagement between healthcare professionals.",
"Laboratory test": "Errors in laboratory testing process.",
"Medication related": "Events related to error in the use, prescription, or administration of medications.",
"Omission / errors in assessment, diagnosis, monitoring": "Failures to properly assess, diagnose, or monitor a patient's condition.",
"Maternal": "Event that compromise the safety of mothers during pregnancy, childbirth, and postpartum.",
"Equipment / devices": "Problems with the use or malfunction of medical equipment and devices.",
"Supplies": "Concerns with the availability, quality, or use of medical supplies.",
}
import base64
def image_to_base64(path):
with open(path, "rb") as image_file:
return base64.b64encode(image_file.read()).decode('utf-8')
image_base64 = image_to_base64("SED_ICON.png")
title_html = f"""
<div style='display: flex; justify-content: space-between; align-items: center; width: 100%;'>
<span style='color: #2874A6; font-size: 44px; font-weight: 900; font-style: normal;'>
Patient Safety Event Reporting Interface
</span>
<img src="data:image/png;base64,{image_base64}" alt="Company Icon" style="width:250px;">
</div>
"""
css = """
.gradio-info{color:white}
.accordion_sec, .accordion_sec * {
color: black; /* Ensures text color is black */
font-size: 26px !important; /* Increase font size */
font-weight: bold;
}
#button_classify {color: #dfdee3; font-size: 24px !important}
#example {color: black; font-size: 24px !important}
.ID_box, .ID_box * {
color: black; /* Ensures text color is black */
font-size: 18px !important; /* Increase font size */
}
.ID_box_e, .ID_box_e * {
color: black; /* Ensures text color is black */
font-size: 18px !important; /* Increase font size */
}
.ID_box_3, .ID_box_3 * {
color: black; /* Ensures text color is black */
#font-style: italic !important; /* Bold font */
font-size: 18px !important; /* Increase font size */
}
"""
# The end bracket must be in a seperate line
# Useless CSS
"""
.ID_box_2, .ID_box_2 {
font-size: 18px !important; /* Increase font size */
color: #FF0000; /* Example: Change to your desired color */
}
"""
event_type_tooltips = {
"Care coordination / communication": "Miscommunication/mismanagement between healthcare professionals.",
"Laboratory test": "Errors in laboratory testing process.",
"Medication related": "Events related to error in the use, prescription, or administration of medications.",
"Omission / errors in assessment, diagnosis, monitoring": "Failures to properly assess, diagnose, or monitor a patient's condition.",
"Maternal": "Event that compromise the safety of mothers during pregnancy, childbirth, and postpartum.",
"Equipment / devices": "Problems with the use or malfunction of medical equipment and devices.",
"Supplies": "Concerns with the availability, quality, or use of medical supplies.",
}
# use set to adjutst certain attribute in the theme
theme = gr.themes.Monochrome(text_size=gr.themes.sizes.text_lg, radius_size='lg'
).set(button_primary_background_fill="dfdee3",
button_secondary_background_fill="#000000",
)
# body_text_color_subdued='#FF0000')
with gr.Blocks(css=css, theme=theme, title=webpage_name) as Interface:
# Title and lab icon
with gr.Row():
gr.HTML(title_html, elem_classes='image')
# Section 1: General Event Information
with gr.Accordion(label="Section 1: General Event Information", elem_classes="accordion_sec"):
with gr.Column():
gr.Textbox(label="1. Incident Date ✱", placeholder='YYYY/MM/DD', scale=1, elem_classes="ID_box")
gr.Textbox(label="2. Incident Time ✱", placeholder="00:00", scale=1, elem_classes="ID_box")
# with gr.Row():
gr.Dropdown(label="3. Site where incident occurred ✱", choices=sites, scale=1, elem_classes="ID_box")
gr.Dropdown(label="4. Department where incident occurred ✱", choices=departments, allow_custom_value=True,
scale=1, elem_classes="ID_box")
gr.Dropdown(label="5. Other department involved", choices=departments, scale=1,
allow_custom_value=True, elem_classes="ID_box")
# Section 2: Affected Patient Information
with gr.Accordion(label="Section 2: Affected Patient Information",
elem_classes="accordion_sec"):
with gr.Column():
gr.Textbox(label="1. First Name ✱", scale=1, elem_classes="ID_box") # ,info="Enter 'N/A' if unknown."
gr.Textbox(label="2. Last Name ✱", scale=1, elem_classes="ID_box")
# with gr.Row():
gr.Textbox(label="3. Date of Birth ✱", scale=1, placeholder='YYYY/MM/DD', elem_classes="ID_box")
gr.Textbox(label="4. Medical Record Number ✱", scale=1, elem_classes="ID_box")
#
# Section 3: Witness Information
with gr.Accordion(label="Section 3: Reporter Information (Optional)", elem_classes="accordion_sec"):
with gr.Column():
gr.Textbox(label="1. First Name", scale=1, elem_classes="ID_box")
gr.Textbox(label="2. Last Name", scale=1, elem_classes="ID_box")
# with gr.Row():
gr.Textbox(label="3. Employee ID", scale=1, elem_classes="ID_box")
gr.Dropdown(label="4. Professional Role", choices=professional_roles, scale=1, elem_classes="ID_box",
allow_custom_value=True)
# with gr.Row():
gr.Textbox(label="5. Email Address", scale=1, elem_classes="ID_box", type='email')
gr.Textbox(label="6. Phone Number", scale=1, elem_classes="ID_box")
# Section 4: Incident Description & Analysis
with gr.Accordion(label="Section 4: Incident Description & Analysis", elem_classes="accordion_sec"):
# Incident description
Incident_description = [gr.Textbox(lines=10, label="1. Incident Description ✱", elem_classes="ID_box",
placeholder=incident_description_placeholder)]
greet_btn = gr.Button(value="Recommend Event Type", size="lg", scale=0,
elem_id="button_classify", elem_classes="ID_box")
# AI output
with gr.Row(visible=True) as output_row:
output = gr.Dataframe(headers=['Event type', 'Probability (%)'], row_count=2, elem_classes="ID_box",
label="Top two recommended event types", scale=1)
explain = gr.HighlightedText(scale=1, label=
"""The recommendation is based on highlighted words in incident description.
If all words are irrelevant, the recommendation may not be accurate.
""", color_map="blue",
elem_classes="ID_box_3")
# Event type of incident
gr.Dropdown(choices=Event_types, label='2. Event Type (Select one or more) ✱',
info="Uncertain about the event type, click the button above to get recommendations from artificial intelligence (AI).",
elem_classes="ID_box_e", multiselect=True)
# Button for classifying PSE report
# What happens after the button has been clicked
greet_btn.click(fn=classify, inputs=Incident_description, outputs=[output, explain])
with gr.Column(): # equal_height=True
gr.Dropdown(choices=["Agree", "Disagree", "Not sure"], scale=1, elem_classes="ID_box",
label="3. Do you agree or disagree with recommended event types? ✱")
gr.Dropdown(label="4. Patient Harm Level ✱", choices=harm_levels, scale=1, elem_classes="ID_box")
# with gr.Row(equal_height=True):
aaa = gr.Dropdown(label="5. Contributing Factors (Select one or more) ✱", choices=contributing_factors,
scale=1,
multiselect=True, elem_classes="ID_box")
gr.Textbox(label="6. Recommended Solution", elem_classes="ID_box",
placeholder='Please recommend solutions to prevent further occurence.', scale=1)
# Feedback information
# portal for uploading supporting documents
file_output = gr.File(label='Please upload all relevant documents for this incident.',
interactive=True, elem_classes="ID_box")
# Submit, should be a saving function, currently not done.
# Modify this for saving files and other details.
# clear=gr.LogoutButton(value="Submit",icon=None,scale=4)
js = """
(x) => {
if (confirm('Submitted, thank you! To submit a new report, click OK')) {
window.location.reload(); // Reload the page only if the user confirmed
}
}
"""
### Fill the link with below box
clear_btn = gr.ClearButton(value="Submit", icon=None, scale=4) # , link='https://5f34c402bdaa663819.gradio.live')
checkbox = gr.Checkbox(visible=False)
clear_btn.click(None, None, checkbox, js=js)
# Need to give examples to facilitate testing examples = gr.Examples(examples=PSE_examples,
# inputs=Incident_description,label="Testing examples of PSE reports", elem_id="example")
Interface.launch(share=True)