import streamlit as st import os import pathlib import process_miner as pm st.session_state.update(st.session_state) if 'is_fuzzy_generated' not in st.session_state: st.session_state.is_fuzzy_generated = False st.set_page_config(page_title='Fuzzy Miner', page_icon=':chart_with_upwards_trend:') st.title('Fuzzy Miner ⛏️') st.sidebar.title("Menu") selected_section = st.sidebar.selectbox("Select:", ["Import", "Process Map"], key='fuzzy_menu', label_visibility='collapsed') if selected_section == "Import": uploaded_file = st.file_uploader("Import a logs file .XES, .CSV") def upload(): if uploaded_file is None: st.warning('Upload a file first!', icon="⚠️") else: data = uploaded_file.getvalue() parent_path = pathlib.Path(__file__).parent.parent.resolve() save_path = os.path.join(parent_path, "data") complete_name = os.path.join(save_path, uploaded_file.name) destination_file = open(complete_name, "wb") destination_file.write(data) destination_file.close() log = pm.read_log(os.path.join("webapp/data", uploaded_file.name)) param = [[fs_include, fs_invert, frequency_significance], [rs_include, rs_invert, routing_significance], [bfs_include, bfs_invert, binary_frequency_significance], [ds_include, ds_invert, distance_significance], [pc_include, pc_invert, proximity_correlation], [ec_include, ec_invert, endpoint_correlation], [oc_include, oc_invert, originator_correlation], [dtc_include, dtc_invert, data_type_correlation], [dvc_include, dvc_invert, data_value_correlation]] with st.spinner(text="In progress..."): st.session_state.fuzzy_obj = pm.discover_fuzzy_miner(log, param, attenuation, max_distance, num_of_echelons) st.success('Process map generated successfully!', icon="🤖") st.session_state.is_fuzzy_generated = True st.session_state.fuzzy_menu = "Process Map" expander = st.expander("Fuzzy Configuration", expanded=False) with expander: st.header("Metrics") st.subheader("Unary Metrics") unary1 = st.columns(2) with unary1[0]: fs_include = st.checkbox('Include', key=1, value=True) with unary1[1]: fs_invert = st.checkbox('Invert', key=2) frequency_significance = st.slider('Frequency Significance', 0.000, 1.000, 1.000, step=0.001, format="%f") unary2 = st.columns(2) with unary2[0]: rs_include = st.checkbox('Include', key=3, value=True) with unary2[1]: rs_invert = st.checkbox('Invert', key=4) routing_significance = st.slider('Routing Significance', 0.000, 1.000, 1.000, step=0.001, format="%f") st.subheader("Binary Significance") unary3 = st.columns(2) with unary3[0]: bfs_include = st.checkbox('Include', key=5, value=True) with unary3[1]: bfs_invert = st.checkbox('Invert', key=6) binary_frequency_significance = st.slider('Binary Frequency Significance', 0.000, 1.000, 1.000, step=0.001, format="%f") unary4 = st.columns(2) with unary4[0]: ds_include = st.checkbox('Include', key=7, value=True) with unary4[1]: ds_invert = st.checkbox('Invert', key=8) distance_significance = st.slider('Distance Significance', 0.000, 1.000, 1.000, step=0.001, format="%f") st.subheader("Binary Correlation") unary5 = st.columns(2) with unary5[0]: pc_include = st.checkbox('Include', key=9, value=True) with unary5[1]: pc_invert = st.checkbox('Invert', key=10) proximity_correlation = st.slider('Proximity Correlation', 0.000, 1.000, 1.000, step=0.001, format="%f") unary6 = st.columns(2) with unary6[0]: ec_include = st.checkbox('Include', key=11, value=True) with unary6[1]: ec_invert = st.checkbox('Invert', key=12) endpoint_correlation = st.slider('Endpoint Correlation', 0.000, 1.000, 1.000, step=0.001, format="%f") unary7 = st.columns(2) with unary7[0]: oc_include = st.checkbox('Include', key=13, value=True) with unary7[1]: oc_invert = st.checkbox('Invert', key=14) originator_correlation = st.slider('Originator Correlation', 0.000, 1.000, 1.000, step=0.001, format="%f") unary8 = st.columns(2) with unary8[0]: dtc_include = st.checkbox('Include', key=15, value=True) with unary8[1]: dtc_invert = st.checkbox('Invert', key=16) data_type_correlation = st.slider('Data Type Correlation', 0.000, 1.000, 1.000, step=0.001, format="%f") unary9 = st.columns(2) with unary9[0]: dvc_include = st.checkbox('Include', key=17, value=True) with unary9[1]: dvc_invert = st.checkbox('Invert', key=18) data_value_correlation = st.slider('Data Value Correlation', 0.000, 1.000, 1.000, step=0.001, format="%f") # else: st.header("Attenuation") max_distance = st.slider('Maximal event distance', 1, 20, 5) num_of_echelons = max_distance attenuation = st.radio("Select Attenuation", ["N(th) root with radical", "Linear Attenuation"], horizontal=True) if attenuation == "N(th) root with radical": num_of_echelons = st.slider('N(th) root', 1.00, 4.00, 2.70, label_visibility='collapsed') st.button("Generate process map", on_click=upload) else: path = pathlib.Path(__file__).parent.parent.resolve() try: f = open(os.path.join(path, "media/graphs/fuzzy.gv"), "r") lines = f.readlines() svg = ''.join(lines) st.graphviz_chart(svg) file = open(os.path.join(path, "media/graphs/fuzzy.gv.svg"), "r") btn = st.download_button( label="Download .svg", data=file, file_name="fuzzy.svg", mime="image/svg+xml" ) if st.session_state.is_fuzzy_generated: def update_node(): pm.update_node_filter(st.session_state.fuzzy_obj, st.session_state.sign_cutoff_slider_f) node_expander = st.sidebar.expander("Node", expanded=False) node_expander.slider('Significance CutOff', 0.000, 1.000, 0.000, step=0.001, format="%f", on_change=update_node, key='sign_cutoff_slider_f') def update_edge(): pm.update_edge_filter(st.session_state.fuzzy_obj, int(st.session_state.edge_transform_f == 'Fuzzy Edges'), st.session_state.sc_ratio_slider_f, st.session_state.preserve_edge_slider_f, st.session_state.interpret_abs_f, st.session_state.ignore_self_loops_f) edge_expander = st.sidebar.expander("Edge", expanded=False) edge_expander.radio("edge_transform", ["Fuzzy Edges", "Best Edges"], horizontal=True, on_change=update_edge, label_visibility='collapsed', key='edge_transform_f') if st.session_state.edge_transform_f == 'Fuzzy Edges': edge_expander.checkbox('Interpret Absolute', value=False, on_change=update_edge, key='interpret_abs_f') edge_expander.slider('Preserve Edge', 0.001, 1.000, 0.200, step=0.001, format="%f", on_change=update_edge, key='preserve_edge_slider_f') edge_expander.slider('S/C Ratio', 0.000, 1.000, 0.750, step=0.001, format="%f", on_change=update_edge, key='sc_ratio_slider_f') edge_expander.checkbox('Ignore Self-Loops', value=True, on_change=update_edge, key='ignore_self_loops_f') def update_concurrency(): pm.update_concurrency_filter(st.session_state.fuzzy_obj, st.session_state.filter_concurrency_f, st.session_state.preserve_slider_f, st.session_state.offset_slider_f) concur_expander = st.sidebar.expander("Concurrency", expanded=False) concur_expander.checkbox('Filter Concurrency', value=True, on_change=update_concurrency, key='filter_concurrency_f') if st.session_state.filter_concurrency_f: concur_expander.slider('Preserve', 0.000, 1.000, 0.600, step=0.001, format="%f", on_change=update_concurrency, key='preserve_slider_f') concur_expander.slider('Balance', 0.000, 1.000, 0.700, step=0.001, format="%f", on_change=update_concurrency, key='offset_slider_f') except: st.warning('First generate the process model!', icon="⚠️")