Spaces:
Runtime error
Runtime error
import datetime as datetime | |
import hashlib | |
import json | |
import os | |
import sys | |
import pandas as pd | |
import streamlit as st | |
current = os.path.dirname(os.path.realpath(__file__)) | |
parent = os.path.dirname(current) | |
sys.path.append(parent) | |
from helpers import ( | |
apply_style, | |
callback_add_to_multiselect, | |
choose_text_menu, | |
do_prent, | |
find_event_types, | |
get_additional_words, | |
get_idx_column, | |
get_nli_limit, | |
get_num_sentences_in_list_text, | |
get_top_k, | |
initiate_widget_st_state, | |
run_prent, | |
) | |
# Set constant values | |
TOP_K = get_top_k() | |
NLI_LIMIT = get_nli_limit() | |
### Styling | |
# Needs to be done first | |
apply_style() | |
# Avoid having ellipsis in the multi select options | |
styl = """ | |
<style> | |
.stMultiSelect span{ | |
max-width: none; | |
} | |
</style> | |
""" | |
st.markdown(styl, unsafe_allow_html=True) | |
# Set color of multiselect to red | |
st.markdown( | |
""" | |
<style> | |
span[data-baseweb="tag"] { | |
background-color: red !important; | |
} | |
</style> | |
""", | |
unsafe_allow_html=True, | |
) | |
def validated_metric_per_event_types(validated_dataset): | |
"""Compute the accuracy metrics of the validated dataset | |
for each event type. Compute True Positive, False Negative, | |
True Negative, False Positive. | |
:param validated_dataset: Dictionary containing results of PRENT validated by the user | |
:type validated_dataset: dict | |
:return: Dictionnary containing accuracy metric for all event types | |
:rtype: dict | |
""" | |
dict_acc = {} | |
for key, val in validated_dataset.items(): | |
# Compute the event types based on the computed templates of PRENT | |
pred_event_types = find_event_types( | |
st.session_state.codebook, val["filled_templates"] | |
) | |
true_event_types = val["event_types"] | |
# Compute only accuracy for accepted samples | |
if val["decision"] == "Accept": | |
# Iterate over all possible event types | |
for event_type in st.session_state.codebook["events"].keys(): | |
dict_acc.setdefault(event_type, {}) | |
dict_acc[event_type].setdefault("TP", 0) | |
dict_acc[event_type].setdefault("FN", 0) | |
dict_acc[event_type].setdefault("FP", 0) | |
dict_acc[event_type].setdefault("TN", 0) | |
if (event_type in true_event_types) and ( | |
event_type in pred_event_types | |
): | |
dict_acc[event_type]["TP"] += 1 | |
elif (event_type in true_event_types) and not ( | |
event_type in pred_event_types | |
): | |
dict_acc[event_type]["FN"] += 1 | |
elif not (event_type in true_event_types) and ( | |
event_type in pred_event_types | |
): | |
dict_acc[event_type]["FP"] += 1 | |
else: | |
dict_acc[event_type]["TN"] += 1 | |
# Normalize metrics | |
if dict_acc: | |
for event_type in st.session_state.codebook["events"].keys(): | |
dict_acc[event_type]["Accuracy"] = ( | |
dict_acc[event_type]["TP"] + dict_acc[event_type]["TN"] | |
) / ( | |
dict_acc[event_type]["TP"] | |
+ dict_acc[event_type]["TN"] | |
+ dict_acc[event_type]["FP"] | |
+ dict_acc[event_type]["FN"] | |
) | |
return dict_acc | |
def store_validated_data( | |
text, | |
decision, | |
text_idx, | |
templates, | |
additional_words, | |
list_event_type, | |
prent_params=(TOP_K, NLI_LIMIT), | |
): | |
"""Function used to store the results of PRENT in a DataFrame and in the | |
session state of Streamlit. | |
:param text: Event description | |
:type text: string | |
:param decision: Decision of the user (Accept/Reject/Ignore) | |
:type decision: string | |
:param text_idx: Index of the event | |
:type text_idx: int | |
:param templates: List of template used | |
:type templates: list | |
:param additional_words: List of additional words used | |
:type additional_words: list | |
:param list_event_type: List of event type found by PRENT and Codebook | |
:type list_event_type: list | |
:param prent_params: Parameters of PRENT, defaults to (TOP_K, NLI_LIMIT) | |
:type prent_params: tuple, optional | |
""" | |
if "validated_data" not in st.session_state: | |
st.session_state["validated_data"] = {} | |
# Generate an index if the text is not coming from a csv | |
if not text_idx: | |
# Create a hash of 8 digits of the text to put as index | |
data_idx = str( | |
"manual_{}".format( | |
int( | |
hashlib.sha256(text.encode("utf-8")).hexdigest(), | |
16, | |
) | |
% 10**8 | |
) | |
) | |
else: | |
data_idx = str(text_idx) | |
if data_idx not in st.session_state["validated_data"]: | |
st.session_state["validated_data"][data_idx] = {} | |
st.session_state["validated_data"][data_idx]["text"] = text | |
st.session_state["validated_data"][data_idx]["templates"] = [ | |
template.replace("{}", "[Z]") for template in templates | |
] | |
st.session_state["validated_data"][data_idx]["additional_words"] = additional_words | |
st.session_state["validated_data"][data_idx]["event_types"] = list_event_type | |
st.session_state["validated_data"][data_idx][ | |
"filled_templates" | |
] = list_filled_templates | |
st.session_state["validated_data"][data_idx]["decision"] = decision | |
st.session_state["validated_data"][data_idx]["prent_params"] = prent_params | |
### Initialize session state variables | |
if "codebook" not in st.session_state: | |
st.session_state.codebook = {} | |
st.session_state.codebook.setdefault("events", {}) | |
st.session_state.codebook["templates"] = [] | |
if "text" not in st.session_state: | |
st.session_state.text = "" | |
if "res" not in st.session_state: | |
st.session_state.res = None | |
if "accept_reject_text_perm" not in st.session_state: | |
st.session_state.accept_reject_text_perm = None | |
if "validated_data" not in st.session_state: | |
st.session_state["validated_data"] = {} | |
if "time_comput" not in st.session_state: | |
st.session_state.time_comput = 20 | |
if "rerun" not in st.session_state: | |
st.session_state.rerun = False | |
if "recompute_all_templates" not in st.session_state: | |
st.session_state.recompute_all_templates = False | |
def reset_computation_results(): | |
"""Reset cached values in session state related to computations""" | |
st.session_state.res = {} | |
st.session_state.recompute_all_templates = True | |
st.session_state["accept_reject_text_perm"] = "Ignore" | |
st.session_state.rerun = True | |
def get_all_filled_templates(results): | |
"""Create the filled templates from PRENT results. Merging template with mask | |
with the entailed tokens. | |
:param results: Dictionary containing PRENT results | |
:type results: dict | |
:return: List of all entailed templates | |
:rtype: list | |
""" | |
filled_templates = [] | |
templates_used = [x.replace("[Z]", "{}") for x in results["templates_used"]] | |
for template in templates_used: | |
filled_template = [template.format(x) for x in results[template]] | |
filled_templates.extend(filled_template) | |
return filled_templates | |
# Split streamlit dashboard | |
col_intro_left, col_intro_righter = st.columns([8, 8]) | |
with col_intro_left: | |
st.markdown( | |
""" # Codebook Design | |
""" | |
) | |
def load_demo( | |
codebook_path="codebook_demo.json", | |
validated_data_path="validated_data_demo.json", | |
csv_data_path="data_demo.csv", | |
): | |
"""Load demonstration files from disk | |
:param codebook_path: path to codebook, defaults to "codebook_demo.json" | |
:type codebook_path: str, optional | |
:param validated_data_path: path to validated dataset, defaults to "validated_data_demo.json" | |
:type validated_data_path: str, optional | |
:param csv_data_path: path to raw data, defaults to "data_demo.csv" | |
:type csv_data_path: str, optional | |
""" | |
st.session_state.codebook = json.load(open(codebook_path)) | |
st.session_state.validated_data = json.load(open(validated_data_path)) | |
st.session_state.data = pd.read_csv(csv_data_path, delimiter=";") | |
st.session_state.filtered_df = st.session_state.data | |
st.session_state.text_column_design_perm = "Event Descriptions" | |
st.session_state["multiselect_classes"] = list( | |
st.session_state.codebook["events"].keys() | |
) | |
st.session_state.text_idx = 0 | |
st.session_state.text = ( | |
"On 23 August, a group attacked a village, abducting 6 people." | |
) | |
st.session_state.text_display = ( | |
"On 23 August, a group attacked a village, abducting 6 people." | |
) | |
st.session_state["text_options_valid_perm"] = "From CSV" | |
st.session_state["text_options_valid"] = "From CSV" | |
def clear_all(): | |
"""Cleare session state""" | |
for each in st.session_state: | |
del st.session_state[each] | |
st.experimental_rerun() | |
# Add two buttons in the sidebar to load and clear the demo | |
with st.sidebar: | |
if st.button("Load Demo"): | |
load_demo() | |
if st.button("Clear Demo"): | |
clear_all() | |
st.write("********") | |
with st.sidebar: | |
# Next function used for callback when download | |
def update_codebook_save_time(): | |
st.session_state.save_codebook_time = ( | |
datetime.datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S %z") | |
) | |
if st.download_button( | |
label="Download codebook as JSON", | |
data=json.dumps(st.session_state.codebook, indent=3).encode("ASCII"), | |
file_name="codebook.json", | |
mime="application/json", | |
): | |
update_codebook_save_time() | |
if "save_codebook_time" in st.session_state: | |
st.write("Saved on: " + st.session_state.save_codebook_time) | |
with st.sidebar: | |
# Next function used for callback when download | |
def update_validated_save_time(): | |
st.session_state.save_validated_time = ( | |
datetime.datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S %z") | |
) | |
if st.download_button( | |
label="Download labeled data", | |
data=json.dumps(st.session_state["validated_data"], indent=3).encode("ASCII"), | |
file_name="validated_data.json", | |
mime="application/json", | |
): | |
update_validated_save_time() | |
if "save_validated_time" in st.session_state: | |
st.write("Saved on: " + st.session_state.save_validated_time) | |
# Add text to sidebar | |
with st.sidebar: | |
st.write("********") | |
st.markdown( | |
""" | |
#### Manual: | |
1. Set the list of possible event types | |
2. Select the input mode of the data (Manual or CSV) | |
3. If the codebook is empty, write a default template | |
- `This event involves [Z].` is a good starting point | |
4. Write/Select an event description | |
5. Run PR-ENT | |
6. Check the event type classification | |
- If it is correct then select Accept and return to step 4. | |
- If it is wrong then select Reject and populate the codebook with the appropriate filled templates. The classification is updated for each change, when it is correct, click Accept. | |
7. Return to step 4 | |
#### Tips & Tricks: | |
- If you start a codebook from scratch, it may be easier to pass a manual text example for each event type to get a first codebook draft | |
- Current codebook accuracy based on labeled data can be found in the top right | |
- The approach does not aim for perfect accuracy and some failures can happen, e.g. some event descriptions can produce filled templates that are not satisfactory. | |
""" | |
) | |
# Add accuracy table | |
with col_intro_righter: | |
accuracy = st.empty() | |
# We fill the table with the last acc to avoid having it disappearing each time | |
if "acc_df" in st.session_state: | |
accuracy.table( | |
st.session_state.acc_df.loc["Accuracy":"Accuracy"].style.format("{:.2}") | |
) | |
performance_container = st.expander("Detailed Performances") | |
st.write("*********") | |
col_left, col_right = st.columns(2) | |
# Add widgets to add event type and choose text input | |
with col_intro_left: | |
with st.expander("Event Types List"): | |
st.markdown( | |
""" | |
## Select Event Types. | |
""" | |
) | |
if "class_list_perm" not in st.session_state: | |
st.session_state["class_list_perm"] = [] | |
# Text field + button to add new event types to multiselect | |
new_class = st.text_input( | |
"Add a new event type", "", key="new_class_text_input" | |
) | |
st.button( | |
"Add Class", | |
on_click=callback_add_to_multiselect, | |
args=( | |
new_class, | |
"multiselect_classes", | |
"new_class_text_input", | |
"class_list_perm", | |
), | |
) | |
# Multiselect to choose event types | |
if "multiselect_classes" not in st.session_state: | |
st.session_state["multiselect_classes"] = list( | |
st.session_state.codebook["events"].keys() | |
) | |
class_list = st.multiselect( | |
"Event Type List", | |
set( | |
st.session_state["class_list_perm"] | |
+ list(st.session_state.codebook["events"].keys()) | |
), | |
st.session_state["multiselect_classes"], | |
key="multiselect_classes", | |
) | |
st.session_state["class_list_perm"] = class_list | |
with st.expander("Select Text Input Mode (Manual, CSV)"): | |
st.write( | |
""" | |
Choose the text input of the event descriptions. Three choices: | |
- Manual: One event description can be manually input | |
- From CSV: If a CSV of event descriptions was provided | |
""" | |
) | |
def callback_radio_text_choice(): | |
st.session_state.text = "" | |
st.session_state.text_display = "" | |
initiate_widget_st_state( | |
"text_options_valid", "text_options_valid_perm", "Manual" | |
) | |
st.session_state["text_options_valid_perm"] = st.radio( | |
"Choose text input", | |
["Manual", "From CSV"], | |
index=get_idx_column( | |
st.session_state["text_options_valid"], ["Manual", "From CSV"] | |
), | |
key="text_options_valid", | |
on_change=callback_radio_text_choice, | |
horizontal=True, | |
) | |
with col_left: | |
if st.session_state["text_options_valid_perm"] == "Manual": | |
text = choose_text_menu("") | |
# Reset all computations if text has changed | |
if text != st.session_state.text: | |
reset_computation_results() | |
st.session_state.text_idx = None | |
st.session_state.text = text | |
st.session_state.text_display = text | |
elif st.session_state["text_options_valid_perm"] == "From CSV": | |
if st.button("Select Random Text"): | |
sample = st.session_state.filtered_df.sample(n=1).iloc[0] | |
text = sample[st.session_state["text_column_design_perm"]] | |
idx = sample.name | |
if text != st.session_state.text: | |
reset_computation_results() | |
st.session_state.text = text | |
st.session_state.text_idx = idx | |
st.session_state.text_display = st.session_state.text | |
expected_time = st.session_state.time_comput * get_num_sentences_in_list_text( | |
[st.session_state.text] | |
) | |
if st.button("Run PR-ENT / Expected time: {}sec".format(expected_time)): | |
if "templates" in st.session_state.codebook: | |
templates = st.session_state.codebook["templates"] | |
else: | |
templates = [] | |
st.warning("No template in codebook. Please add one.") | |
additional_words = get_additional_words() | |
st.session_state.res = {} | |
res, time_comput = run_prent(st.session_state.text, templates, additional_words) | |
st.session_state.res = res | |
st.write("**Event Descriptions:** {}".format(st.session_state.text_display)) | |
ev_desc = st.empty() | |
radio_empty = st.empty() | |
if st.session_state.res: | |
list_filled_templates = get_all_filled_templates(st.session_state.res) | |
list_event_type = find_event_types( | |
st.session_state.codebook, list_filled_templates | |
) | |
event_type_text = ev_desc.markdown( | |
"**Current Event Types Classification**: {}".format( | |
"; ".join(list_event_type) | |
) | |
) | |
if "accept_reject_text_perm" not in st.session_state: | |
st.session_state["accept_reject_text_perm"] = "Ignore" | |
def callback_function(mod, key): | |
st.session_state[mod] = st.session_state[key] | |
radio_empty.radio( | |
"Accept or Reject Coding", | |
["Ignore", "Accept", "Reject"], | |
key="accept_reject_text", | |
on_change=callback_function, | |
args=( | |
"accept_reject_text_perm", | |
"accept_reject_text", | |
), | |
index=get_idx_column( | |
st.session_state["accept_reject_text_perm"], | |
["Ignore", "Accept", "Reject"], | |
), | |
horizontal=True, | |
) | |
decision = st.session_state["accept_reject_text_perm"] | |
text_idx = st.session_state.text_idx | |
text = st.session_state.text | |
store_validated_data( | |
text, | |
decision, | |
text_idx, | |
st.session_state.res["templates_used"], | |
st.session_state.res["additional_words_used"], | |
list_event_type, | |
prent_params=(TOP_K, NLI_LIMIT), | |
) | |
with col_right: | |
if ( | |
st.session_state["accept_reject_text_perm"] == "Reject" | |
) or not st.session_state.codebook["templates"]: | |
with st.expander("Add Templates + Explanation"): | |
st.markdown( | |
""" | |
## Add Templates | |
""" | |
) | |
st.markdown( | |
""" | |
For each template added. PR-ENT will be run on the selected text. | |
""" | |
) | |
if "templates" not in st.session_state.codebook: | |
st.session_state.codebook["templates"] = [] | |
template = st.text_input( | |
"Template with a mask [Z].", "This event involves [Z]." | |
) | |
if st.button("Add template"): | |
if template not in st.session_state.codebook["templates"]: | |
## Add template to codebook | |
st.session_state.codebook["templates"].append(template) | |
additional_words = get_additional_words() | |
prompt = template.replace("[Z]", "{}") | |
results_nli, _ = do_prent( | |
st.session_state.text, | |
prompt, | |
TOP_K, | |
NLI_LIMIT, | |
additional_words, | |
) | |
tokens_nli = [x[0] for x in results_nli] | |
# Update result table with new template | |
if not st.session_state["res"]: | |
st.session_state.res = {} | |
st.session_state.res["additional_words_used"] = additional_words | |
st.session_state.res["templates_used"] = [] | |
st.session_state.res[prompt] = tokens_nli | |
st.session_state.res["templates_used"].append(template) | |
st.write("Template '{}' added.".format(template)) | |
else: | |
st.write("Template '{}' already added.".format(template)) | |
if st.session_state.codebook["templates"]: | |
with st.expander("Populate Codebook Explanation"): | |
st.markdown( | |
""" | |
## Set the filled template to each class. | |
For each class you can select one or more filled templates. When the evaluation will | |
be made, these templates will be compared with the results of PR-ENT. There are 4 options: | |
- ALL: If **ALL** of these filled templates are present in the results of PR-ENT then this event type is correct | |
- ANY: If **ANY** of these filled templates is present in the results of PR-ENT then this event type is correct | |
- NOT ALL: If **ALL** of these filled templates are present in the results of PR-ENT, then this event type is **not** correct | |
- e.g. You may want to remove all *explosions* events from a class *Killings*. | |
- NOT ANY: If **ANY** of these filled templates is present in the results of PR-ENT, then this event type is **not** correct | |
Moreover, **ANY/ALL** and **NOT ANY/ NOT ALL** can be made in relation by a **AND / OR** condition. | |
""" | |
) | |
st.write("***************") | |
st.write("### Populate Codebook") | |
if not class_list: | |
st.warning("No event type in codebook.") | |
tokens_list = get_all_filled_templates(st.session_state.res) | |
for event_type in class_list: | |
st.session_state.codebook["events"].setdefault(event_type, {}) | |
event_type_chosen = event_type | |
with st.expander(event_type): | |
def declare_ms_event_templates( | |
widget_key, widget_display, codebook_key | |
): | |
if widget_key not in st.session_state: | |
st.session_state[widget_key] = st.session_state.codebook[ | |
"events" | |
][event_type_chosen].setdefault(codebook_key, []) | |
tokens_all = st.multiselect( | |
widget_display, | |
set( | |
list( | |
tokens_list | |
+ st.session_state.codebook["events"][ | |
event_type_chosen | |
].setdefault(codebook_key, []) | |
) | |
), | |
st.session_state[widget_key], | |
key=widget_key, | |
) | |
st.session_state.codebook["events"][event_type_chosen][ | |
codebook_key | |
] = tokens_all | |
declare_ms_event_templates( | |
"ms_all_{}".format(event_type_chosen), "ALL", "all" | |
) | |
st.session_state.codebook["events"][event_type_chosen][ | |
"all_any_rel" | |
] = st.selectbox( | |
"Relation", | |
["AND", "OR"], | |
index=get_idx_column( | |
st.session_state.codebook["events"][ | |
event_type_chosen | |
].setdefault("all_any_rel", "OR"), | |
["AND", "OR"], | |
), | |
key="select_relation_any_all_{}".format(event_type_chosen), | |
) | |
declare_ms_event_templates( | |
"ms_any_{}".format(event_type_chosen), "ANY", "any" | |
) | |
declare_ms_event_templates( | |
"ms_not_all_{}".format(event_type_chosen), "NOT ALL", "not_all" | |
) | |
st.session_state.codebook["events"][event_type_chosen][ | |
"not_all_any_rel" | |
] = st.selectbox( | |
"Relation", | |
["AND", "OR"], | |
index=get_idx_column( | |
st.session_state.codebook["events"][ | |
event_type_chosen | |
].setdefault("not_all_any_rel", "OR"), | |
["AND", "OR"], | |
), | |
key="select_relation_not_any_all_{}".format(event_type_chosen), | |
) | |
declare_ms_event_templates( | |
"ms_not_any_{}".format(event_type_chosen), "NOT ANY", "not_any" | |
) | |
# Workaround to avoid the expanders closing after first modification | |
# I have no explanation for the bug | |
if st.session_state.rerun: | |
st.session_state.rerun = False | |
st.experimental_rerun() | |
if "validated_data" in st.session_state: | |
recompute = False | |
performance_container.markdown( | |
"If a new template is added, the previous labeled samples needs to be recomputed with it. The next button allows that, however it can take some time depending on the number of samples." | |
) | |
if performance_container.button( | |
"Recompute Missing Templates", key="recompute_temp" | |
): | |
prog_bar = performance_container.progress(0) | |
for i, datapoint in enumerate(st.session_state["validated_data"].values()): | |
if not set(st.session_state.codebook["templates"]).issubset( | |
set(datapoint["templates"]) | |
): | |
# Get templates that are missing from results but present in codebook | |
# These happens if templates are added a posteriori | |
missing_templates = list( | |
set(st.session_state.codebook["templates"]) | |
- set(set(datapoint["templates"])) | |
) | |
recompute = True | |
# For now additional words are not recomputed | |
if not set(st.session_state.codebook["add_words"]).issubset( | |
set(datapoint["additional_words"]) | |
): | |
missing_add_words = list( | |
set(st.session_state.codebook["add_words"]) | |
- set(set(datapoint["additional_words"])) | |
) | |
recompute = True | |
else: | |
missing_add_words = None | |
if recompute: | |
res, _ = run_prent( | |
datapoint["text"], | |
missing_templates, | |
missing_add_words, | |
progress=False, | |
) | |
datapoint["filled_templates"].extend(get_all_filled_templates(res)) | |
datapoint["templates"].extend(missing_templates) | |
prog_bar.progress( | |
(1 / len(st.session_state["validated_data"].values())) * (i + 1) | |
) | |
st.session_state.acc_df = pd.DataFrame( | |
validated_metric_per_event_types(st.session_state["validated_data"]) | |
) | |
accuracy.table( | |
st.session_state.acc_df.loc["Accuracy":"Accuracy"].style.format("{:.2}") | |
) | |
performance_container.markdown("### Performances on labeled dataset") | |
performance_container.dataframe(st.session_state.acc_df.style.format("{:.3}")) | |
if st.session_state.res: | |
list_filled_templates = get_all_filled_templates(st.session_state.res) | |
list_event_type = find_event_types(st.session_state.codebook, list_filled_templates) | |
ev_desc.markdown( | |
"**Current Event Types Classification**: {}".format("; ".join(list_event_type)) | |
) | |