Select Git revision
metropolis_hastings.py
petrinet.py 2.48 KiB
from pm4py.objects.petri_net.obj import PetriNet, Marking
from pm4py.objects.conversion.log import converter as log_converter
import pm4py
import pandas as pd
import eventlog
from collections import OrderedDict
def generate_petri_net():
dataframe = pd.read_csv(r'Frontend\export\eventlog.csv', sep=',')
dataframe['StartTimestamp'] = pd.to_datetime(dataframe['StartTimestamp'])
# Sort the event log by case ID and start timestamp
dataframe.sort_values(by=['CaseID', 'StartTimestamp'], inplace=True)
dataframe = dataframe.head(900)
dataframe = pm4py.format_dataframe(dataframe, case_id='CaseID', activity_key='Activity', timestamp_key='StartTimestamp')
event_log = pm4py.convert_to_event_log(dataframe)
net, initial_marking, final_marking = pm4py.discover_petri_net_inductive(event_log)
# print(net)
# pm4py.view_petri_net(net, initial_marking, final_marking)
for t in net.transitions:
print("\nTransition: "+t.name)
return net, initial_marking, final_marking
def decorate_petri_net(case, rec, name):
activity_mapping = {
1: 'place order',
2: 'arrange standard order',
3: 'arrange custom order',
4: 'pick from stock A',
5: 'pick from stock B',
6: 'pick from stock C',
7: 'manufacture A',
8: 'manufacture B',
9: 'pack A',
10: 'pack B',
11: 'pack C',
12: 'attempt delivery A',
13: 'attempt delivery B',
14: 'attempt delivery C',
15: 'order completed',
}
net, initial_marking, final_marking = generate_petri_net()
state = eventlog.get_state(case, name)
events = state['case']
event_names = []
for i in range(len(events)):
if events[i] == 1:
event_names.append(activity_mapping[i+1])
decoration = {}
rec_name = activity_mapping[rec]
for t in net.transitions:
print(str(t.label))
if str(t.label) in event_names:
decoration.update({t: {}})
decoration[t].update({"color":"blue"})
decoration[t].update({"label":t.label})
elif str(t.label) == rec_name:
decoration.update({t: {}})
decoration[t].update({"color":"green"})
decoration[t].update({"label":t.label})
pm4py.view_petri_net(net, initial_marking, final_marking, decorations = decoration)
pm4py.save_vis_petri_net(net, initial_marking, final_marking, r"Frontend\export\net.png")
# decorate_petri_net(646, 7, "eventlog.csv")