Alignments

Alignments

Alignment-based replay aims to find one of the best alignment between the trace and the model. For each trace, the output of an alignment is a list of couples where the first element is an event (of the trace) or » and the second element is a transition (of the model) or ». For each couple, the following classification could be provided:

  • Sync move: the classification of the event corresponds to the transition label; in this case, both the trace and the model advance in the same way during the replay.
  • Move on log: for couples where the second element is », it corresponds to a replay move in the trace that is not mimicked in the model. This kind of move is unfit and signal a deviation between the trace and the model.
  • Move on model: for couples where the first element is », it corresponds to a replay move in the model that is not mimicked in the trace. For moves on model, we can have the following distinction:
    • Moves on model involving hidden transitions: in this case, even if it is not a sync move, the move is fit.
    • Moves on model not involving hidden transitions: in this case, the move is unfit and signals a deviation between the trace and the model.

The following code implements an example for obtaining alignments. First, the running-example.xes log is loaded and the Inductive Miner is applied:

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)

And the alignments can be obtained by this piece of code:

import pm4py
from pm4py.algo.conformance.alignments import factory as align_factory

alignments = align_factory.apply_log(log, net, initial_marking, final_marking)

If we execute print(alignments) we get the following output:

[{'alignment': [('register request', 'register request'), ('examine casually', 'examine casually'), ('>>', None), ('check ticket', 'check ticket'), ('>>', None), ('decide', 'decide'), ('reinitiate request', 'reinitiate request'), ('>>', None), ('>>', None), ('examine thoroughly', 'examine thoroughly'), ('>>', None), ('check ticket', 'check ticket'), ('>>', None), ('decide', 'decide'), ('pay compensation', 'pay compensation'), ('>>', None)], 'cost': 7, 'visited_states': 18, 'queued_states': 50, 'traversed_arcs': 100, 'fitness': 1.0}, {'alignment': [('register request', 'register request'), ('check ticket', 'check ticket'), ('>>', None), ('examine casually', 'examine casually'), ('>>', None), ('decide', 'decide'), ('pay compensation', 'pay compensation'), ('>>', None)], 'cost': 3, 'visited_states': 9, 'queued_states': 26, 'traversed_arcs': 45, 'fitness': 1.0}, {'alignment': [('register request', 'register request'), ('examine thoroughly', 'examine thoroughly'), ('>>', None), ('check ticket', 'check ticket'), ('>>', None), ('decide', 'decide'), ('reject request', 'reject request'), ('>>', None)], 'cost': 3, 'visited_states': 9, 'queued_states': 26, 'traversed_arcs': 45, 'fitness': 1.0}, {'alignment': [('register request', 'register request'), ('examine casually', 'examine casually'), ('>>', None), ('check ticket', 'check ticket'), ('>>', None), ('decide', 'decide'), ('pay compensation', 'pay compensation'), ('>>', None)], 'cost': 3, 'visited_states': 9, 'queued_states': 26, 'traversed_arcs': 45, 'fitness': 1.0}, {'alignment': [('register request', 'register request'), ('examine casually', 'examine casually'), ('>>', None), ('check ticket', 'check ticket'), ('>>', None), ('decide', 'decide'), ('reinitiate request', 'reinitiate request'), ('>>', None), ('>>', None), ('check ticket', 'check ticket'), ('>>', None), ('examine casually', 'examine casually'), ('>>', None), ('decide', 'decide'), ('reinitiate request', 'reinitiate request'), ('>>', None), ('>>', None), ('examine casually', 'examine casually'), ('>>', None), ('check ticket', 'check ticket'), ('>>', None), ('decide', 'decide'), ('reject request', 'reject request'), ('>>', None)], 'cost': 11, 'visited_states': 29, 'queued_states': 75, 'traversed_arcs': 157, 'fitness': 1.0}, {'alignment': [('register request', 'register request'), ('check ticket', 'check ticket'), ('>>', None), ('examine thoroughly', 'examine thoroughly'), ('>>', None), ('decide', 'decide'), ('reject request', 'reject request'), ('>>', None)], 'cost': 3, 'visited_states': 9, 'queued_states': 26, 'traversed_arcs': 45, 'fitness': 1.0}]

This list reports for each trace the corresponding alignment along with its statistics. With each trace, a dictionary containing among the others the following information is associated:

  • alignment: contains the alignment (sync moves, moves on log, moves on model)
  • cost: contains the cost of the alignment according to the provided cost function
  • fitness: is equal to 1 if the trace is perfectly fitting

To use a different classifier, we recall the Classifiers section in documentation of Process Discovery. Indeed, the following code defines a custom classifier for each event of each trace in the log:

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

A parameters dictionary containing the activity key can be formed:

# 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 could be 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:

alignments = align_factory.apply_log(log, net, initial_marking, final_marking, parameters=parameters)

To get the overall log fitness value, the following code can be used:

from pm4py.evaluation.replay_fitness import factory as replay_fitness_factory

log_fitness = replay_fitness_factory.evaluate(alignments, variant="alignments")

Using print(log_fitness) the following result is obtained:

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

The following parameters can also be provided to the alignments:

  • Model cost function: associating to each transition in the Petri net the corresponding cost of a move-on-model.
  • Sync cost function: associating to each visible transition in the Petri net the cost of a sync move.

Implementation of a custom model cost function, and sync cost function:

model_cost_function = dict()
sync_cost_function = dict()
for t in net.transitions:
	# if the label is not None, we have a visible transition
	if t.label is not None:
		# associate cost 1000 to each move-on-model associated to visible transitions
		model_cost_function[t] = 1000
		# associate cost 0 to each move-on-log
		sync_cost_function[t] = 0
	else:
		# associate cost 1 to each move-on-model associated to hidden transitions
		model_cost_function[t] = 1

Insertion of the model cost function and sync cost function in the parameters:

parameters[pm4py.algo.conformance.alignments.versions.state_equation_a_star.PARAM_MODEL_COST_FUNCTION] = model_cost_function
parameters[pm4py.algo.conformance.alignments.versions.state_equation_a_star.PARAM_SYNC_COST_FUNCTION] = sync_cost_function

And eventually the replay is done:

alignments = align_factory.apply_log(log, net, initial_marking, final_marking, parameters=parameters)

Visualization of the alignments (for Conformance purpose)

Visualization on top of the petri net model

!!! Beta feature, available in the ‘develop’ branch !!!

It is possible to visualize the result of the alignments on top of the Petri net model. This requires a slight modification of the executions, that means that the output of the alignment contains for each move both the corresponding transitions names and the corresponding transitions labels.

Therefore, the alignments shall be applied in this way in order to make the Petri net decoration possible:

alignments = alignments_factory.apply(red_log, net, im, fm, parameters={"ret_tuple_as_trans_desc": True})

Then, the decorations could be obtained using the following commands:

from pm4py.visualization.petrinet.util import alignments_decoration
decorations = alignments_decoration.get_alignments_decoration(net, im, fm, log=red_log)

And the visualization of the Petri net obtained using the following commands:

from pm4py.visualization.petrinet import factory as pn_vis_factory

gviz = pn_vis_factory.apply(net, im, fm, aggregated_statistics=decorations, variant="alignments")
pn_vis_factory.view(gviz)

If the alignments are still not performed, then the decoration of the Petri net could automatically calculate the alignments:

from pm4py.visualization.petrinet import factory as pn_vis_factory

gviz = pn_vis_factory.apply(net, im, fm, log=log, variant="alignments")
pn_vis_factory.view(gviz)

An example visualization (obtained on a filtered version of the Receipt log) is the following (along the label of the transition, the number of moves on model and moves on log is reported):

 

!!! Beta feature, available in the ‘bpmnIntegration2’ branch !!!

If working with BPMN is needed, then the same alignments decoration feature is available for BPMN, e.g.

Visualization using Pydotplus (a Petri net is discovered out of the log; then converted to BPMN visualization):

from pm4py.visualization.bpmn import factory as bpmn_vis_factory

gviz = bpmn_vis_factory.apply_petri(net, im, fm, log=log, variant="alignments")
bpmn_vis_factory.save(gviz, "alignments.png")

Embedding on the BPMN graph: executing the following code

from pm4py.visualization.bpmn import factory as bpmn_vis_factory
bpmn_graph = bpmn_vis_factory.apply_embedding(bpmn_graph, log=red_log, variant="alignments")
from pm4py.objects.bpmn.exporter import bpmn20 as bpmn_exporter
bpmn_exporter.export_bpmn(bpmn_graph, "result.bpmn")

You could get for each transition a decoration inside the XML 2.0 like:

<task id=”T05 Print and send confirmation of receipt” name=”T05 Print and send confirmation of receipt”>
<incoming>idabae7257-cd79-49b2-8576-bbba99797831</incoming>
<outgoing>id60a47339-74fe-4967-9b26-674ddaa5fd5d</outgoing>
<decorations>
<conformance_count_fit>3</conformance_count_fit>
<conformance_count_move_on_model>2</conformance_count_move_on_model>
<conformance_label>T05 Print and send confirmation of receipt (2,3)</conformance_label>
<conformance_color>#FF9898</conformance_color>
</decorations>
</task>

VISUALIZATION OF THE ALIGNMENTS TABLE

!!! Beta feature, available in the ‘develop’ branch !!!

Getting the alignments table permits to understand how the single variants (set of cases in the log sharing the same control flow perspective) were aligned according to the model.

The following example shows a visualization of the alignments table. Color GREEN is associated to sync moves, color GRAY is associated to moves on model, color VIOLET is associated to moves on log.

The alignments table could be obtained as follows. This requires a slight modification of the execution, that means that the output of the alignment contains for each move both the corresponding transitions names and the corresponding transitions labels.

Therefore, the alignments shall be applied in this way in order to make the alignments table visualization possible:

alignments = alignments_factory.apply(log, net, im, fm, parameters={"ret_tuple_as_trans_desc": True})

Then, the alignment table could be obtained using the following commands (format SVG in this case permits to get a browser window where an optimal size of the image could be set):

from pm4py.visualization.align_table import factory as align_table_factory
gviz = align_table_factory.apply(log, alignments, parameters={"format": "svg"})
align_table_factory.view(gviz)

.