Spaces:
Sleeping
Sleeping
File size: 9,311 Bytes
9844377 5cd9764 9844377 5cd9764 9844377 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 |
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="⚠️")
|