webapp/pages/2_Fuzzy_Miner.py

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="⚠️")