webapp/pages/3_Stigmergic_Miner.py

import os
import pathlib
import streamlit as st
import pandas as pd
import numpy as np
import process_miner as pm

st.session_state.update(st.session_state)

# Defining the default archetypes used for analyzing transitions in standard process events,
# the ones present in the archetypes file of the stigmergic perceptron.
default_archetypes = np.array([[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                               [0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0],
                               [1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                               [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 1, 1, 1],
                               [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]])
default_labels = ['Dead Transition', 'Cold Transition', 'Falling Transition', 'Rising Transition',
                  'Hot Transition']

if "columns" not in st.session_state:
    st.session_state.columns = default_labels

if "data" not in st.session_state:
    st.session_state.data = default_archetypes

# Generate the archetype dataframe
if "chart_data" not in st.session_state:
    st.session_state.chart_data = pd.DataFrame(
        st.session_state.data.transpose(),
        columns=st.session_state.columns)

# Session state to prevent the generation of a new archetype if it has already been generated by the user.
if "disabled_generation" not in st.session_state:
    st.session_state.disabled_generation = False

if "radio_index" not in st.session_state:
    st.session_state.radio_index = 0

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


def init_archetypes():
    """
    Function that reinitializes the archetypes selection menu when a user loads a new dataset.
    """
    st.session_state.columns = default_labels
    st.session_state.data = default_archetypes
    st.session_state.chart_data = pd.DataFrame(
        st.session_state.data.transpose(),
        columns=st.session_state.columns)
    st.session_state.disabled_generation = False
    st.session_state.export_timeseries = True


# Set up the web page and the selection box to enable the user to initialize a new dataset import or explore
# the currently generated dataset.
st.set_page_config(page_title='Stigmergic Miner', page_icon=':chart_with_upwards_trend:')
st.title('Stigmergic Miner ⛏️')
st.sidebar.title("Menu")
selected_section = st.sidebar.selectbox("Select:", ["Import", "Stigmergic Map", "Archetypal Map"], key='stigmergic_menu',
                                        label_visibility='collapsed')

# Initialize the page with the 'import' option selected by default.
if selected_section == "Import":
    st.session_state.uploaded_file = st.file_uploader("Import a logs file .XES, .CSV", on_change=init_archetypes)

    def generate_process_map():
        """
        Function that generates the process map when a new dataset is loaded into the application.
        """
        if st.session_state.uploaded_file is None:
            st.warning('Upload a file first!', icon="⚠️")
        else:
            parent_path = pathlib.Path(__file__).parent.parent.resolve()
            save_path = os.path.join(parent_path, "data")
            complete_name = os.path.join(save_path, st.session_state.uploaded_file.name)
            destination_file = open(complete_name, "wb")

            loaded_data = st.session_state.uploaded_file.getvalue()
            destination_file.write(loaded_data)
            destination_file.close()

            log = pm.read_log(os.path.join("webapp/data", st.session_state.uploaded_file.name))
            params = [significance_norm, evaporation_rate, st.session_state.export_timeseries]

            with st.spinner(text="In progress..."):
                st.session_state.stigmergic_obj = pm.discover_stigmergic_miner(
                    log, st.session_state.columns.index(st.session_state.archetype), params)
            st.success('Process map generated successfully!', icon="🤖")
            st.session_state.is_stigmergic_generated = True
            st.session_state.stigmergic_menu = "Stigmergic Map"

    def generate_archetypal_map():
        """
        Function that generates the archetypal map when a new dataset is loaded into the application.
        """
        if st.session_state.uploaded_file is None:
            st.warning('Upload a file first!', icon="⚠️")
        else:
            parent_path = pathlib.Path(__file__).parent.parent.resolve()
            save_path = os.path.join(parent_path, "data")
            complete_name = os.path.join(save_path, st.session_state.uploaded_file.name)
            destination_file = open(complete_name, "wb")

            loaded_data = st.session_state.uploaded_file.getvalue()
            destination_file.write(loaded_data)
            destination_file.close()

            log = pm.read_log(os.path.join("webapp/data", st.session_state.uploaded_file.name))
            params = [significance_norm, evaporation_rate, st.session_state.export_timeseries]

            with st.spinner(text="In progress..."):
                st.session_state.stigmergic_obj = pm.discover_overall_stigmergic_miner(
                    log, np.arange(st.session_state.data.shape[0]), params)

            st.success('Process map generated successfully!', icon="🤖")
            st.session_state.is_stigmergic_generated = True
            st.session_state.stigmergic_menu = "Archetypal Map"

    st.header("Metrics")
    evaporation_rate = st.slider('Evaporation rate', 0.01, 1.00, 1.00, format="%f",
                                 disabled=st.session_state.disabled_generation)
    significance_norm = st.slider('Signals normalization', 0.1, 5.0, 0.5, format="%f",
                                  disabled=st.session_state.disabled_generation)

    def generate_archetype():
        """
        Function that generates a new archetype based on the dataset.
        """
        if st.session_state.uploaded_file is None:
            st.warning('Upload a file first!', icon="⚠️")
        elif len(st.session_state.columns) == 5:
            parent_path = pathlib.Path(__file__).parent.parent.resolve()
            save_path = os.path.join(parent_path, "data")
            complete_name = os.path.join(save_path, st.session_state.uploaded_file.name)
            destination_file = open(complete_name, "wb")

            loaded_data = st.session_state.uploaded_file.getvalue()
            destination_file.write(loaded_data)
            destination_file.close()

            log = pm.read_log(os.path.join("webapp/data", st.session_state.uploaded_file.name))
            st.session_state.columns.append("Dataset Transition")

            new_archetype = pm.export_time_series(log, significance_norm, True)

            st.session_state.data = np.vstack((st.session_state.data, new_archetype))
            st.session_state.chart_data = pd.DataFrame(
                st.session_state.data.transpose(),
                columns=st.session_state.columns)
            st.session_state.export_timeseries = False
            st.session_state.disabled_generation = True

    def insert_archetype():
        """
        Function that insert a new archetype defined by the user.
        """
        new_archetype = transposed_data.get(0)
        pm.update_archetype_file(new_archetype)
        if len(st.session_state.columns) == 5:
            st.session_state.columns.append("User Transition")
            st.session_state.data = np.vstack((st.session_state.data, new_archetype))
        else:
            st.session_state.data[-1] = new_archetype
        st.session_state.chart_data = pd.DataFrame(
            st.session_state.data.transpose(),
            columns=st.session_state.columns)


    expander = st.expander("Custom Archetype", expanded=False)
    with expander:
        st.subheader("Generate from dataset")
        st.button("Generete archetype", on_click=generate_archetype, disabled=st.session_state.disabled_generation)

        st.subheader("Custom archetype shape")

        def get_data() -> pd.DataFrame:
            df = pd.DataFrame(
                [
                    {"1": False, "2": True, "3": True, "4": True, "5": True, "6": True, "7": True, "8": True, "9": True,
                     "10": True, "11": True, "12": True, "13": True, "14": True, "15": True, "16": True, "17": True,
                     "18": True,
                     "19": True, "20": True},
                ]
            )
            return df

        def get_active_hist(df: pd.DataFrame) -> st.line_chart:
            return st.line_chart(df)

        df = get_data()

        edited_df = st.data_editor(
            df,
            use_container_width=True,
            hide_index=True,
            column_config=None,
            height=78
        )

        # Conversion of the data to show on screen
        edited_df = edited_df.astype(int)
        transposed_data = edited_df.transpose().to_dict(orient='list')

        st.line_chart(transposed_data, use_container_width=True, height=150)

        st.button("Custom archetype", on_click=insert_archetype, disabled=st.session_state.disabled_generation)

    def write_archetype():
        """
        Function that updates the archetypes menu with the newly generated archetype.
        """
        st.write("")
        st.session_state.archetype = st.radio("Select archetype", st.session_state.columns, horizontal=False,
                                              label_visibility='collapsed', index=st.session_state.radio_index)
        st.line_chart(st.session_state.chart_data[st.session_state.archetype], height=150, width=400)

    # Update archetypes menu
    write_archetype()

    cl1, cl2 = st.columns(2)
    with cl1:
        st.button("Generate process map", on_click=generate_process_map)
    with cl2:
        st.button("Generate archetypal map", on_click=generate_archetypal_map)

# Initialize the page with the 'Process Map' option selected.
elif selected_section == "Stigmergic Map":
    init_archetypes()
    path = pathlib.Path(__file__).parent.parent.resolve()
    try:
        # Plot the generated process map.
        f = open(os.path.join(path, "media/graphs/stigmergic.gv"), "r")
        lines = f.readlines()
        svg = ''.join(lines)
        st.graphviz_chart(svg)

        # Link the SVG file to the download button,
        file = open(os.path.join(path, "media/graphs/stigmergic.gv.svg"), "r")
        btn = st.download_button(
            label="Download .svg",
            data=file,
            file_name="stigmergic.svg",
            mime="image/svg+xml"
        )

        # Verify whether the process map has been generated during this session. If this condition is met, the process
        # map object is saved in the session state and allow updates using the provided parameters.
        if st.session_state.is_stigmergic_generated:
            def update_node():
                """
                Function that updates the graph nodes based on the parameter selected by the user.
                """
                pm.update_node_filter_stigmergic(st.session_state.stigmergic_obj, st.session_state.sign_cutoff_slider_s)

            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_s')

            def update_edge():
                """
                Function that updates the graph edges based on the parameters selected by the user.
                """
                pm.update_edge_filter_stigmergic(st.session_state.stigmergic_obj,
                                                 int(st.session_state.edge_transform_s == 'Fuzzy Edges'),
                                                 st.session_state.preserve_edge_slider_s,
                                                 st.session_state.interpret_abs_s, st.session_state.ignore_self_loops_s)

            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_s')
            if st.session_state.edge_transform_s == 'Fuzzy Edges':
                edge_expander.checkbox('Interpret Absolute', value=False, on_change=update_edge, key='interpret_abs_s')
                edge_expander.slider('Preserve Edge', 0.001, 1.000, 0.200, step=0.001, format="%f",
                                     on_change=update_edge, key='preserve_edge_slider_s')
            edge_expander.checkbox('Ignore Self-Loops', value=True, on_change=update_edge, key='ignore_self_loops_s')

            def update_concurrency():
                """
                Function that updates the concurrency based on the parameters selected by the user.
                """
                pm.update_concurrency_filter_stigmergic(st.session_state.stigmergic_obj,
                                                        st.session_state.filter_concurrency_s,
                                                        st.session_state.preserve_slider_s,
                                                        st.session_state.offset_slider_s)

            concur_expander = st.sidebar.expander("Concurrency", expanded=False)
            concur_expander.checkbox('Filter Concurrency', value=True, on_change=update_concurrency,
                                     key='filter_concurrency_s')
            if st.session_state.filter_concurrency_s:
                concur_expander.slider('Preserve', 0.000, 1.000, 0.600, step=0.001, format="%f",
                                       on_change=update_concurrency, key='preserve_slider_s')
                concur_expander.slider('Balance', 0.000, 1.000, 0.700, step=0.001, format="%f",
                                       on_change=update_concurrency, key='offset_slider_s')
    except:
        st.warning('First generate the process model!', icon="⚠️")

elif selected_section == "Archetypal Map":
    init_archetypes()
    path = pathlib.Path(__file__).parent.parent.resolve()
    try:
        # Plot the generated process map.
        f = open(os.path.join(path, "media/graphs/archetypal.gv"), "r")
        lines = f.readlines()
        svg = ''.join(lines)
        st.graphviz_chart(svg)

        # Link the SVG file to the download button,
        file = open(os.path.join(path, "media/graphs/archetypal.gv.svg"), "r")
        btn = st.download_button(
            label="Download .svg",
            data=file,
            file_name="archetypal.svg",
            mime="image/svg+xml"
        )
    except:
        st.warning('First generate the process model!', icon="⚠️")