Skip to content
Snippets Groups Projects
Select Git revision
  • 710548faceee0e83303361c5e31d6be19d6e210e
  • main default protected
  • ravaflow3g
  • docs
  • v0.1.2
  • v0.1.1
  • v0.1.0
7 results

metropolis_hastings.py

Blame
  • 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")