Token-based replayer

Token-based replayer

Token-based replay matches a trace and a Petri net model, starting from the initial place, in order to discover which transitions are executed and in which places we have remaining or missing tokens for the given process instance. Token-based replay is useful for Conformance Checking: indeed, a trace is fitting according to the model if, during its execution, the transitions can be fired without the need to insert any missing token. If the reaching of the final marking is imposed, then a trace is fitting if it reaches the final marking without any missing or remaining tokens.

Token-based replay permits both global and local Conformance Checking. For each trace, we can assign a fitness value that is between 0 and 1 and is defined as:

In pm4py there is an implementation of a token replayer that is able to go across hidden transitions (calculating shortest paths between places) and can be used with any Petri net model with unique visible transitions and hidden transitions. When a visible transition needs to be fired and not all places in the preset are provided with the correct number of tokens, starting from the current marking it is checked if for some place there is a sequence of hidden transitions that could be fired in order to enable the visible transition. The hidden transitions are then fired and a marking that permits to enable the visible transition is reached. In the following picture, an example representing the algorithm is provided:

Visible transitions TRANS could be enabled from the current marking by firring hidden transitions ht1 and ht2.

Aside from the fitness value, the replay algorithm can be configured in order to consider a trace completely fitting even if there are remaining tokens, as long as all visible transitions corresponding to events in the trace can be fired. Moreover, it can be configured to reach the final marking through hidden transitions. This is useful when after the last activity, the final marking is not reached but could be reached with the execution of hidden transitions.

We provide the following example showing the application of token-based replay. The example starts as usual with the import of the running-example.xes log and the application of the Inductive Miner.

import os
from pm4py.objects.log.importer.xes import factory as xes_importer
from pm4py.algo.discovery.inductive import factory as inductive_miner

log = xes_importer.import_log(os.path.join("tests", "input_data", "running-example.xes"))
net, initial_marking, final_marking = inductive_miner.apply(log)

To apply token-based replay, the following code is applied:

from pm4py.algo.conformance.tokenreplay import factory as token_replay

replay_result = token_replay.apply(log, net, initial_marking, final_marking)

The print(replay_result) command prints the results:

[{'trace_is_fit': True, 'trace_fitness': 1.0, 'activated_transitions': [register request, examine casually, check ticket, decide, reinitiate request, loop_2, examine thoroughly, check ticket, decide, pay compensation, tau_1], 'reached_marking': ['sink:1'], 'enabled_transitions_in_marking': set(), 'transitions_with_problems': []}, {'trace_is_fit': True, 'trace_fitness': 1.0, 'activated_transitions': [register request, skip_5, check ticket, loop_4, examine casually, skip_6, decide, pay compensation, tau_1], 'reached_marking': ['sink:0'], 'enabled_transitions_in_marking': set(), 'transitions_with_problems': []}, {'trace_is_fit': True, 'trace_fitness': 1.0, 'activated_transitions': [register request, examine thoroughly, check ticket, decide, reject request, tau_1], 'reached_marking': ['sink:1'], 'enabled_transitions_in_marking': set(), 'transitions_with_problems': []}, {'trace_is_fit': True, 'trace_fitness': 1.0, 'activated_transitions': [register request, examine casually, check ticket, decide, pay compensation, tau_1], 'reached_marking': ['sink:0'], 'enabled_transitions_in_marking': set(), 'transitions_with_problems': []}, {'trace_is_fit': True, 'trace_fitness': 1.0, 'activated_transitions': [register request, examine casually, check ticket, decide, reinitiate request, loop_2, skip_5, check ticket, loop_4, examine casually, skip_6, decide, reinitiate request, loop_2, examine casually, check ticket, decide, reject request, tau_1], 'reached_marking': ['sink:0'], 'enabled_transitions_in_marking': set(), 'transitions_with_problems': []}, {'trace_is_fit': True, 'trace_fitness': 1.0, 'activated_transitions': [register request, skip_5, check ticket, loop_4, examine thoroughly, skip_6, decide, reject request, tau_1], 'reached_marking': ['sink:0'], 'enabled_transitions_in_marking': set(), 'transitions_with_problems': []}]

There is one dictionary in the list for each trace, the keys provided in the dictionary for each trace are:

  • trace_is_fit -> Indicates if the trace is completely fit to the model
  • trace_fitness -> The fitness value calculated for the trace
  • activated_transitions -> List of transitions in the model that are activated during the replay
  • reached_marking -> Marking that is reached at the end of the replay
  • enabled_transitions_in_marking -> In the reached marking at the end of the replay, indicates if there are enabled transitions

The following code provides the overall log fitness value:

from pm4py.evaluation.replay_fitness import factory as replay_fitness_factory

log_fitness = replay_fitness_factory.evaluate(replay_result, variant="token_replay")

If we execute print(log_fitness) then the following result is obtained:

{'percFitTraces': 100.0, 'averageFitness': 1.0}

The token-based replayer can also be configured to return local conformance information about places/transitions. This is achieved through the enable_pltr_fitness parameter. The following code could be applied:

from pm4py.algo.conformance.tokenreplay import factory as token_replay

replay_result, place_fitness, trans_fitness, notexisting_activities_in_model = token_replay.apply(log, net, initial_marking, final_marking, parameters={"enable_pltr_fitness": True})

If we do print(place_fitness), the following result is obtained:

{({'check ticket'}, {'decide'}): {'underfedTraces': set(), 'overfedTraces': set()}, ({'examine thoroughly', 'examine casually'}, {'decide'}): {'underfedTraces': set(), 'overfedTraces': set()}, ({'reinitiate request', 'register request'}, {'examine thoroughly', 'examine casually'}): {'underfedTraces': set(), 'overfedTraces': set()}, ({'decide'}, {'reinitiate request', 'pay compensation', 'reject request'}): {'underfedTraces': set(), 'overfedTraces': set()}, start: {'underfedTraces': set(), 'overfedTraces': set()}, end: {'underfedTraces': set(), 'overfedTraces': set()}, ({'reinitiate request', 'register request'}, {'check ticket'}): {'underfedTraces': set(), 'overfedTraces': set()}}

The keys of this dictionary are places, the values are dictionaries containing the set of traces for which the place is underfed and the set of traces for which the place is overfed.

Additional parameters of the token-replay algorithm, that could be passed in the parameters dictionary, are:

  • consider_remaining_in_fitness -> Default True. In considering a trace fit according to the model, it is checked that no place is overfed (that means, the final marking is reached)
  • try_to_reach_final_marking_through_hidden -> Default True. When the visible transitions corresponding to events in the trace are fired, and the final marking is not still reached but could be reached by firing hidden transitions, then the hidden transitions are fired in order to reach the final marking
  • stop_immediately_unfit -> Default False. If True, the replay stops at the first deviation encountered
  • walk_through_hidden_trans -> Default True. Enable going across the hidden transitions in token-based replay
  • activity key (pm4py.util.constants.PARAMETER_CONSTANT_ACTIVITY_KEY) -> Must be specified if a different classifier from the activity name (concept:name) is needed.

To use a different classifier, we recall the Classifiers section in the documentation of Process Discovery:

for trace in log:
    for event in trace:
        event["customClassifier"] = event["concept:name"] + event["concept:name"]

A parameters dictionary containing the activity key is constructed:

# import constants
from pm4py.util import constants
# define the activity key in the parameters
parameters = {constants.PARAMETER_CONSTANT_ACTIVITY_KEY: "customClassifier"}

Then the process model is calculated:

# calculate process model using the given classifier
net, initial_marking, final_marking = inductive_miner.apply(log, parameters=parameters)

And eventually the replay is done:

# apply token-based replay
replay_result = token_replay.apply(log, net, initial_marking, final_marking, parameters=parameters)