Conformance Checking

Prepare your data with the DataFrame import examples and obtain a model through process discovery. For aggregate scores, see model evaluation; for larger workloads, see approximate alignments.

Conformance checking is a technique used to compare a process model with an event log of the same process. The goal is to determine whether the event log conforms to the model, and vice versa. In PM4Py, two fundamental techniques are implemented: token-based replay and alignments.

Token-based Replay

Token-based replay matches a trace with a Petri net model, starting from the initial place. It discovers which transitions are executed and identifies places with remaining or missing tokens for the given process instance. Token-based replay is particularly useful for conformance checking. A trace is considered to fit the model if, during its execution, transitions can be fired without the need to insert any missing tokens. If reaching the final marking is required, the trace fits if it reaches the final marking without any missing or remaining tokens. For further explanation, refer to the related section.

For each trace, four values must be determined: produced tokens, remaining tokens, missing tokens, and consumed tokens. Based on these values, the following formula can be derived, where a Petri net (n) and a trace (t) are provided as input:

fitness(n, t) = ½(1 − r_p) + ½(1 − m_c)

To apply this formula to the entire event log, p, r, m, and c are calculated for each trace, summed up, and then inserted into the formula above.

In PM4Py, a token replayer is implemented that can traverse hidden transitions (calculating the shortest paths between places) and works with any Petri net model containing both unique visible transitions and hidden transitions. When a visible transition needs to be fired but not all places in the preset have the correct number of tokens, starting from the current marking, the system checks if a sequence of hidden transitions can be fired to enable the visible transition. The hidden transitions are fired, and a marking that allows the visible transition to be enabled is reached.

The example on the right demonstrates how to apply token-based replay on a log and a Petri net. First, the log is loaded. Then, the Alpha Miner is applied to discover a Petri net. Finally, the token-based replay is applied. The output of the token-based replay, stored in the variable replayed_traces, contains the following information for each trace in the log:

• trace_is_fit: A Boolean value (True/False) that is true when the trace conforms to the model.
• activated_transitions: A list of transitions activated in the model by the token-based replay.
• reached_marking: The marking reached at the end of the replay.
• missing_tokens: The number of missing tokens.
• consumed_tokens: The number of consumed tokens.
• remaining_tokens: The number of remaining tokens.
• produced_tokens: The number of produced tokens.

import os
import pm4py
import pandas
from pm4py.objects.petri_net.obj import Marking, PetriNet
from typing import Any

if __name__ == "__main__":
    log: pandas.DataFrame = pm4py.read_xes(os.path.join("tests", "input_data", "running-example.xes"))

    net: PetriNet
    initial_marking: Marking
    final_marking: Marking
    net, initial_marking, final_marking = pm4py.discover_petri_net_alpha(log)

    replayed_traces: list[dict[str, Any]] = pm4py.conformance_diagnostics_token_based_replay(log, net, initial_marking, final_marking)

Diagnostics (TBR)

Executing token-based replay in PM4Py provides detailed information about transitions that did not execute correctly or activities that are present in the log but not in the model. In particular, executions that do not match the model are expected to take longer throughput times.

The diagnostics provided by PM4Py include the following:

• Throughput analysis on transitions that are executed incorrectly according to the process model (the Petri net).
• Throughput analysis on activities that are not contained in the model.
• Root cause analysis of the factors leading to an unfit execution of transitions.
• Root cause analysis of the factors leading to the execution of activities not present in the process model.

To provide execution contexts for the examples, the log must be loaded, and a model that is not perfectly fitting is required. The following instructions can be used to load the log:

import os
import pm4py
import pandas
if __name__ == "__main__":
    log: pandas.DataFrame = pm4py.read_xes(os.path.join("tests", "input_data", "receipt.xes"))

To create an unfit model, a filtering operation can be executed that produces a log with only part of the behavior:

import pm4py
import pandas
if __name__ == "__main__":
    filtered_log: pandas.DataFrame = pm4py.filter_variants_top_k(log, 3)

Then, apply the Inductive Miner algorithm:

import pm4py
from pm4py.objects.petri_net.obj import Marking, PetriNet
if __name__ == "__main__":
    net: PetriNet
    initial_marking: Marking
    final_marking: Marking
    net, initial_marking, final_marking = pm4py.discover_petri_net_inductive(filtered_log)

Next, apply the token-based replay with special settings. Specifically, set disable_variants to True to avoid replaying only a case with a specific variant, and set enable_pltr_fitness to True to instruct the algorithm to return localized conformance checking applications.

from pm4py.algo.conformance.tokenreplay import algorithm as token_based_replay
from typing import Any
if __name__ == "__main__":
    parameters_tbr: dict = {
        token_based_replay.Variants.TOKEN_REPLAY.value.Parameters.DISABLE_VARIANTS: True,
        token_based_replay.Variants.TOKEN_REPLAY.value.Parameters.ENABLE_PLTR_FITNESS: True
    }
    replayed_traces: list[dict[str, Any]]
    place_fitness: dict
    trans_fitness: dict
    unwanted_activities: dict
    replayed_traces, place_fitness, trans_fitness, unwanted_activities = token_based_replay.apply(
        log, net, initial_marking, final_marking, parameters=parameters_tbr
    )

Afterward, pass the diagnostic information.

Throughput Analysis (Unfit Execution)

To perform throughput analysis on transitions that were executed incorrectly and print the result to the console, use the following code:

from pm4py.algo.conformance.tokenreplay.diagnostics import duration_diagnostics
if __name__ == "__main__":
    trans_diagnostics: dict = duration_diagnostics.diagnose_from_trans_fitness(log, trans_fitness)
    for trans in trans_diagnostics:
        print(trans, trans_diagnostics[trans])

This will show whether unfit executions result in significantly higher throughput times (e.g., from 126 to 146 times higher).

Throughput Analysis (Activities)

To perform throughput analysis on activities not contained in the model and print the result to the console, use the following code:

from pm4py.algo.conformance.tokenreplay.diagnostics import duration_diagnostics
if __name__ == "__main__":
    act_diagnostics: dict = duration_diagnostics.diagnose_from_notexisting_activities(log, unwanted_activities)
    for act in act_diagnostics:
        print(act, act_diagnostics[act])

Root Cause Analysis

The output of root cause analysis in the diagnostic context is a decision tree that helps identify the causes of deviations. For each deviation, a distinct decision tree is built and visualized. In the following examples, the decision trees are constructed using the org:group attribute, based on the Receipt log.

if __name__ == "__main__":
    # Build decision trees
    string_attributes = ["org:group"]
    numeric_attributes = []
    parameters: dict = {"string_attributes": string_attributes, "numeric_attributes": numeric_attributes}

Root Cause Analysis (Unfit Execution)

To perform root cause analysis on transitions that were executed incorrectly, use the following code:

from pm4py.algo.conformance.tokenreplay.diagnostics import root_cause_analysis
if __name__ == "__main__":
    trans_root_cause: dict = root_cause_analysis.diagnose_from_trans_fitness(log, trans_fitness, parameters=parameters)

To visualize the decision trees obtained from root cause analysis, use the following code:

from pm4py.visualization.decisiontree import visualizer as dt_vis

if __name__ == "__main__":
    for trans in trans_root_cause:
        clf = trans_root_cause[trans]["clf"]
        feature_names = trans_root_cause[trans]["feature_names"]
        classes = trans_root_cause[trans]["classes"]
        # Visualization can be called
        gviz = dt_vis.apply(clf, feature_names, classes)
        dt_vis.view(gviz)

Root Cause Analysis (Activities not in the Model)

To perform root cause analysis on activities that were executed but are not part of the process model, use the following code:

from pm4py.algo.conformance.tokenreplay.diagnostics import root_cause_analysis
if __name__ == "__main__":
    act_root_cause: dict = root_cause_analysis.diagnose_from_notexisting_activities(log, unwanted_activities, parameters=parameters)

To visualize the decision trees obtained from root cause analysis, use the following code:

from pm4py.visualization.decisiontree import visualizer as dt_vis
if __name__ == "__main__":
    for act in act_root_cause:
        clf = act_root_cause[act]["clf"]
        feature_names = act_root_cause[act]["feature_names"]
        classes = act_root_cause[act]["classes"]
        # Visualization can be called
        gviz = dt_vis.apply(clf, feature_names, classes)
        dt_vis.view(gviz)

Alignments

PM4Py includes several linear solvers: Scipy (available for all platforms), CVXOPT (available for most widely used platforms, including Windows and Linux), and ORTools, which can also be installed via PIP.

Alignment-based replay aims to find one of the best alignments between the trace and the model. For each trace, the output of an alignment consists of a list of pairs, where the first element is an event (from the trace) or » and the second element is a transition (from the model) or ». For each pair, the following classifications can be made:

• Sync Move: The event corresponds to the transition label; both the trace and the model progress similarly during the replay.
• Move on Log: Pairs where the second element is » correspond to a move in the trace that is not mimicked in the model. These moves are unfit and indicate a deviation between the trace and the model.
• Move on Model: Pairs where the first element is » correspond to a move in the model that is not mimicked in the trace. These moves can be further classified as:
  • Moves on Model Involving Hidden Transitions: These are valid even if not sync moves.
  • Moves on Model Not Involving Hidden Transitions: These moves are unfit and indicate a deviation between the trace and the model.

First, import the log. Then, apply the Inductive Miner on the imported log and compute the alignments.

import os
import pm4py
import pandas
from pm4py.objects.petri_net.obj import Marking, PetriNet
from typing import Any

if __name__ == "__main__":
    log: pandas.DataFrame = pm4py.read_xes(os.path.join("tests", "input_data", "running-example.xes"))

    net: PetriNet
    initial_marking: Marking
    final_marking: Marking
    net, initial_marking, final_marking = pm4py.discover_petri_net_inductive(log)

    import pm4py
    aligned_traces: list[dict[str, Any]] = pm4py.conformance_diagnostics_alignments(log, net, initial_marking, final_marking)

To inspect the alignments, use the following code. The output (a list) reports, for each trace, the corresponding alignment along with its statistics. Each trace is associated with a dictionary containing the following information, among others:

• alignment: Contains the alignment of the trace (sync moves, moves on log, moves on model).
• cost: The cost of the alignment based on the provided cost function.
• fitness: Equals 1 if the trace is perfectly fitting.

print(aligned_traces)

The following code combines activity and resource values into a custom classifier column in the DataFrame.

# Combine activity and resource columns without iterating over legacy traces.
log = log.copy()
log["customClassifier"] = log["concept:name"].astype(str) + "+" + log["org:resource"].astype(str)

A parameters dictionary containing the activity key can be formed.

# Define the activity key in the parameters
from pm4py.algo.discovery.inductive import algorithm as inductive_miner
from pm4py.algo.conformance.alignments.petri_net import algorithm as alignments
from pm4py.objects.conversion.process_tree import converter as process_tree_converter
parameters: dict[str, str] = {"pm4py:param:activity_key": "customClassifier"}

Then, a process model is computed, and alignments are applied with the custom classifier.

from pm4py.objects.petri_net.obj import Marking, PetriNet
from pm4py.objects.process_tree.obj import ProcessTree
from typing import Any
# Calculate process model using the given classifier
if __name__ == "__main__":
    process_tree: ProcessTree = inductive_miner.apply(log, parameters=parameters)
    net: PetriNet
    initial_marking: Marking
    final_marking: Marking
    net, initial_marking, final_marking = process_tree_converter.apply(process_tree)
    aligned_traces: list[dict[str, Any]] = alignments.apply_log(log, net, initial_marking, final_marking, parameters=parameters)

    from pm4py.algo.evaluation.replay_fitness import algorithm as replay_fitness
    log_fitness: dict[str, float] = replay_fitness.evaluate(aligned_traces, variant=replay_fitness.Variants.ALIGNMENT_BASED)

    print(log_fitness)

It is also possible to select other parameters for the alignments, such as:

• Model Cost Function: Assigning a cost to each transition in the Petri net for a move-on-model.
• Sync Cost Function: Assigning a cost to each visible transition in the Petri net for a sync move.

The following code defines custom model and sync cost functions. The model and sync cost functions must be included in the parameters. After that, the replay is performed.

from typing import Any
if __name__ == "__main__":
    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 with visible transitions
            model_cost_function[t] = 1000
            # Associate cost 0 to each sync move
            sync_cost_function[t] = 0
        else:
            # Associate cost 1 to each move-on-model associated with hidden transitions
            model_cost_function[t] = 1

    parameters: dict = {"pm4py:param:activity_key": "customClassifier"}
    parameters[alignments.Variants.VERSION_STATE_EQUATION_A_STAR.value.Parameters.PARAM_MODEL_COST_FUNCTION] = model_cost_function
    parameters[alignments.Variants.VERSION_STATE_EQUATION_A_STAR.value.Parameters.PARAM_SYNC_COST_FUNCTION] = sync_cost_function

    aligned_traces: list[dict[str, Any]] = alignments.apply_log(log, net, initial_marking, final_marking, parameters=parameters)

Anti-alignments

The low-level anti-alignment and multi-alignment variants below require legacy traces. These examples explicitly request an EventLog with return_legacy_log_object=True.

PM4Py also supports anti-alignments on sound Petri nets. Unlike ordinary alignments, which search for the best explanation of the observed traces, anti-alignments search for an executable model run that is as far away as possible from the observed behavior in the log. This makes them useful for precision analysis because they highlight behavior that the model allows but the log does not support well.

This is different from the edit-distance anti-alignments between two event logs described later on this page. For Petri-net anti-alignments, the most relevant parameters are Parameters.EXPONENT, Parameters.EPSILON, and Parameters.MARKING_LIMIT.

The following example loads a log and a sound Petri net from disk, computes an anti-alignment using the discounted A* variant, and prints the returned anti-alignment together with the derived precision estimate:

import os
from pm4py.algo.conformance.antialignments.variants.discounted_a_star import apply as anti_alignments
from pm4py.algo.conformance.antialignments.algorithm import Parameters
import pm4py
from pm4py.objects.petri_net.importer import importer as petri_importer
from pm4py.objects.log.obj import EventLog
from pm4py.objects.petri_net.obj import Marking, PetriNet

if __name__ == "__main__":
    log_path: str = os.path.join("tests", "input_data", "running-example.xes")
    pnml_path = os.path.join("tests", "input_data", "running-example.pnml")

    log: EventLog = pm4py.read_xes(log_path, return_legacy_log_object=True)
    net: PetriNet
    initial_marking: Marking
    final_marking: Marking
    net, initial_marking, final_marking = petri_importer.apply(pnml_path)

    result: dict = anti_alignments(
        log,
        net,
        initial_marking,
        final_marking,
        parameters={
            Parameters.EXPONENT: 1.5,
            Parameters.EPSILON: 0.01,
            Parameters.MARKING_LIMIT: 20
        }
    )

    print(result["anti-alignment"])
    print("Precision:", result["precision"])

Multi-alignments

Multi-alignments generalize standard alignments from a single trace to the entire log. Instead of replaying each trace independently, the algorithm searches for one executable model run that minimizes the maximal distance to the traces in the log, providing a compact representative explanation of the observed behavior.

As for anti-alignments, the implementation expects a sound Petri net. The key parameters are Parameters.EXPONENT and Parameters.MARKING_LIMIT. The returned dictionary contains the computed multi-alignment and the max_distance_to_log, which reports the largest Levenshtein distance between the selected model run and the traces in the log.

import os
from pm4py.algo.conformance.multialignments.variants.discounted_a_star import apply as multi_alignments
from pm4py.algo.conformance.multialignments.algorithm import Parameters
import pm4py
from pm4py.objects.petri_net.importer import importer as petri_importer
from pm4py.objects.log.obj import EventLog
from pm4py.objects.petri_net.obj import Marking, PetriNet

if __name__ == "__main__":
    log_path: str = os.path.join("tests", "input_data", "running-example.xes")
    pnml_path = os.path.join("tests", "input_data", "running-example.pnml")

    log: EventLog = pm4py.read_xes(log_path, return_legacy_log_object=True)
    net: PetriNet
    initial_marking: Marking
    final_marking: Marking
    net, initial_marking, final_marking = petri_importer.apply(pnml_path)

    result: dict = multi_alignments(
        log,
        net,
        initial_marking,
        final_marking,
        parameters={
            Parameters.EXPONENT: 1.1,
            Parameters.MARKING_LIMIT: 20
        }
    )

    print("Multi-alignment:", result["multi-alignment"])
    print("Maximal Levenshtein Edit Distance to Log:", result["max_distance_to_log"])

Decomposition of Alignments

Alignments can be computationally expensive, especially for models with significant concurrency. However, they are the conformance checking technique that provides the best results when it comes to finding a match between process executions and the model. To overcome the challenges associated with the large state space, various methods have been proposed to decompose the model into smaller pieces. This decomposition makes the alignment process easier and still enables the diagnosis of problems.

We've already discussed how to achieve a maximal decomposition of the Petri net model. Now, let's look at how to decompose alignments based on the maximal decomposition of the Petri net model. This approach was published in:

Lee, Wai Lam Jonathan, et al. "Recomposing conformance: Closing the circle on decomposed alignment-based conformance checking in process mining." Information Sciences 466 (2018): 55-91.

Recomposition helps to determine whether each step of the process was executed synchronously or if any deviations occurred. The process starts by performing an alignment on the decomposed Petri nets. Then, the agreement between the activities at the borders is checked. If any disagreements are found, the two components with the disagreement are merged, and the alignment is repeated. When the steps agree across the different alignments of the components, they can be merged into a single alignment. The recomposition order follows the Petri net graph. However, in cases of concurrency, the recomposed alignment may contain valid moves that are not in the correct order.

To perform alignments through decomposition and recomposition, use the following code. A maximum number of border disagreements can be specified for the algorithm. If the number of border disagreements exceeds the provided limit, the alignment is interrupted, and None is returned for the specific trace's alignment.

from pm4py.algo.conformance.alignments.decomposed import algorithm as decomp_alignments
from typing import Any

if __name__ == "__main__":
    conf: list[dict[str, Any]] = decomp_alignments.apply(
        log, net, initial_marking, final_marking,
        parameters={
            decomp_alignments.Variants.RECOMPOS_MAXIMAL.value.Parameters.PARAM_THRESHOLD_BORDER_AGREEMENT: 2
        }
    )

Since decomposed models typically have less concurrency, the components are aligned using a Dijkstra approach. However, in the case of border disagreements, this can degrade the algorithm's performance. It's important to note that this is not an approximation technique; according to the authors, it should provide the same fitness as the original alignments. Since the alignment is recomposed, we can use the fitness evaluator to assess the fitness (which is distinct from the computation of fitness described in the paper).

from pm4py.algo.evaluation.replay_fitness import algorithm as rp_fitness_evaluator

if __name__ == "__main__":
    fitness: dict[str, float] = rp_fitness_evaluator.evaluate(conf, variant=rp_fitness_evaluator.Variants.ALIGNMENT_BASED)

Footprints

Whole-log footprints accept a DataFrame directly. The trace-by-trace variant requires legacy traces, so it uses an explicit conversion.

Footprints are a basic but scalable conformance checking technique used to compare entities such as event logs, DFGs, Petri nets, process trees, and other types of models. Essentially, a relationship is inferred between any pair of activities in the log or model. This includes:

• Directly-Follows Relationships: In the log/model, activity A is directly followed by B.
• Directly-Before Relationships: In the log/model, activity B is directly preceded by A.
• Parallel Behavior: It is possible that A is followed by B and B is followed by A.

A footprints matrix can be calculated to describe the footprint relationship for each pair of activities. This can be computed for different types of models and for the entire event log. It can also be done trace-by-trace if local behavior is important.

Let’s assume that the running-example.xes event log is loaded:

import pm4py
import os
import pandas
if __name__ == "__main__":
    log: pandas.DataFrame = pm4py.read_xes(os.path.join("tests", "input_data", "running-example.xes"))

Apply the Inductive Miner to such a log:

from pm4py.objects.petri_net.obj import Marking, PetriNet
if __name__ == "__main__":
    net: PetriNet
    im: Marking
    fm: Marking
    net, im, fm = pm4py.discover_petri_net_inductive(log)

To calculate footprints for the entire log, use the following code:

from pm4py.algo.discovery.footprints import algorithm as footprints_discovery

if __name__ == "__main__":
    fp_log: dict = footprints_discovery.apply(log)

The footprints of the entire log are:

{
    'sequence': {('examine casually', 'decide'), ('decide', 'pay compensation'),
                 ('register request', 'examine thoroughly'), ('reinitiate request', 'examine casually'),
                 ('check ticket', 'decide'), ('register request', 'examine casually'),
                 ('reinitiate request', 'examine thoroughly'), ('decide', 'reject request'),
                 ('examine thoroughly', 'decide'), ('reinitiate request', 'check ticket'),
                 ('register request', 'check ticket'), ('decide', 'reinitiate request')},
    'parallel': {('examine casually', 'check ticket'), ('check ticket', 'examine casually'),
                ('check ticket', 'examine thoroughly'), ('examine thoroughly', 'check ticket')},
    'start_activities': {'register request'},
    'end_activities': {'pay compensation', 'reject request'},
    'activities': {'reject request', 'register request', 'check ticket', 'decide',
                  'pay compensation', 'examine thoroughly', 'examine casually', 'reinitiate request'}
}

The data structure is a dictionary with keys like sequence (expressing directly-follows relationships) and parallel (expressing parallel behavior that can occur in either direction).

The footprints of the log, trace-by-trace, can be calculated as follows. This will result in a list of footprints for each trace:

from pm4py.algo.discovery.footprints import algorithm as footprints_discovery

if __name__ == "__main__":
    fp_trace_by_trace: list = footprints_discovery.apply(pm4py.convert_to_event_log(log), variant=footprints_discovery.Variants.TRACE_BY_TRACE)

The footprints of the Petri net model can be calculated as follows:

if __name__ == "__main__":
    fp_net: dict = footprints_discovery.apply(net, im, fm)

The footprints of the model are:

{
    'sequence': {('check ticket', 'decide'), ('reinitiate request', 'examine casually'),
                 ('register request', 'examine thoroughly'), ('decide', 'reject request'),
                 ('register request', 'check ticket'), ('register request', 'examine casually'),
                 ('decide', 'reinitiate request'), ('reinitiate request', 'examine thoroughly'),
                 ('decide', 'pay compensation'), ('reinitiate request', 'check ticket'),
                 ('examine casually', 'decide'), ('examine thoroughly', 'decide')},
    'parallel': {('check ticket', 'examine thoroughly'), ('examine thoroughly', 'check ticket'),
                ('check ticket', 'examine casually'), ('examine casually', 'check ticket')},
    'activities': {'decide', 'examine casually', 'reinitiate request', 'check ticket',
                  'examine thoroughly', 'register request', 'reject request', 'pay compensation'},
    'start_activities': {'register request'}
}

The data structure is a dictionary with keys like sequence and parallel. It is possible to visualize a comparison between the footprints of the entire log and those of the model.

Visualizing Footprints

To visualize a single footprints table, for example, the one for the model, use the following code:

from pm4py.visualization.footprints import visualizer as fp_visualizer

if __name__ == "__main__":
    gviz = fp_visualizer.apply(
        fp_net,
        parameters={fp_visualizer.Variants.SINGLE.value.Parameters.FORMAT: "svg"}
    )
    fp_visualizer.view(gviz)

To compare the two footprints tables, use the following code. Note that the visualization will look the same if no deviations are found. If deviations are discovered, they will be colored red.

from pm4py.visualization.footprints import visualizer as fp_visualizer

if __name__ == "__main__":
    gviz = fp_visualizer.apply(
        fp_log, fp_net,
        parameters={fp_visualizer.Variants.COMPARISON.value.Parameters.FORMAT: "svg"}
    )
    fp_visualizer.view(gviz)

To find deviations, repeat the procedure on the receipt.xes log, applying a heavy filter to the log to discover a simpler model:

import pm4py
import os
from copy import deepcopy
import pandas
from pm4py.objects.petri_net.obj import Marking, PetriNet

if __name__ == "__main__":
    log: pandas.DataFrame = pm4py.read_xes(os.path.join("tests", "input_data", "receipt.xes"))
    filtered_log: pandas.DataFrame = pm4py.filter_variants_top_k(log, 3)

    net: PetriNet
    im: Marking
    fm: Marking
    net, im, fm = pm4py.discover_petri_net_inductive(filtered_log)

Using a conformance checking operation, compare the behavior of the log traces against the footprints of the model:

if __name__ == "__main__":
    conf_fp: dict = pm4py.conformance_diagnostics_footprints(fp_trace_by_trace, fp_net)

This will include, for each trace of the log, a set of deviations. Here’s an extract of the list for some traces:

{('T06 Determine necessity of stop advice', 'T04 Determine confirmation of receipt'),
 ('T02 Check confirmation of receipt', 'T06 Determine necessity of stop advice')}
set()
{('T19 Determine report Y to stop indication', 'T20 Print report Y to stop indication'),
 ('T10 Determine necessity to stop indication', 'T16 Report reasons to hold request'),
 ('T16 Report reasons to hold request', 'T17 Check report Y to stop indication'),
 ('T17 Check report Y to stop indication', 'T19 Determine report Y to stop indication')}
set()
set()
{('T02 Check confirmation of receipt', 'T06 Determine necessity of stop advice'),
 ('T10 Determine necessity to stop indication', 'T04 Determine confirmation of receipt'),
 ('T04 Determine confirmation of receipt', 'T03 Adjust confirmation of receipt'),
 ('T03 Adjust confirmation of receipt', 'T02 Check confirmation of receipt')}
set()

We can see that for the first trace containing deviations, there are two issues: one relates to T06 Determine necessity of stop advice being executed before T04 Determine confirmation of receipt, and the other to T02 Check confirmation of receipt being followed by T06 Determine necessity of stop advice. The traces for which conformance returns nothing are considered fit (at least according to the footprints).

Footprints conformance checking is a method for identifying obvious deviations—behaviors in the log that are not allowed by the model. On the log side, scalability is excellent! The calculation of footprints for a Petri net model may be more expensive.

If we change the underlying model from Petri nets to a process tree, it’s possible to exploit its bottom-up structure to calculate the footprints almost instantaneously. Let’s open a log, calculate a process tree, and then apply footprint discovery.

Open the running-example log:

import pm4py
import pandas

if __name__ == "__main__":
    log: pandas.DataFrame = pm4py.read_xes("tests/input_data/running-example.xes")

Apply the Inductive Miner to discover a process tree:

from pm4py.objects.process_tree.obj import ProcessTree
if __name__ == "__main__":
    tree: ProcessTree = pm4py.discover_process_tree_inductive(log)

Then, discover the footprints. Discover the footprints on the entire log, trace-by-trace in the log, and on the process tree:

from pm4py.algo.discovery.footprints import algorithm as fp_discovery

if __name__ == "__main__":
    fp_log: dict = fp_discovery.apply(log)
    fp_trace_trace: list = fp_discovery.apply(pm4py.convert_to_event_log(log), variant=fp_discovery.Variants.TRACE_BY_TRACE)
    fp_tree: dict = fp_discovery.apply(tree, variant=fp_discovery.Variants.PROCESS_TREE)

Each of these contains:

• A list of sequential footprints contained in the log/allowed by the model.
• A list of parallel footprints contained in the log/allowed by the model.
• A list of activities contained in the log/allowed by the model.
• A list of start activities contained in the log/allowed by the model.
• A list of end activities contained in the log/allowed by the model.

Execute an enhanced conformance check between the footprints of the entire log and the footprints of the model:

from pm4py.algo.conformance.footprints import algorithm as fp_conformance

if __name__ == "__main__":
    conf_result: dict = fp_conformance.apply(fp_log, fp_tree, variant=fp_conformance.Variants.LOG_EXTENSIVE)

The result contains, for each item in the previous list, the violations. Given the result of conformance checking, calculate the footprints-based fitness and precision of the process model:

from pm4py.algo.conformance.footprints.util import evaluation

if __name__ == "__main__":
    fitness: float = evaluation.fp_fitness(fp_log, fp_tree, conf_result)
    precision: float = evaluation.fp_precision(fp_log, fp_tree)

These values are both included in the interval [0, 1].

Log Skeleton

The concept of a log skeleton is introduced in the following paper:

Verbeek, H. M. W., and R. Medeiros de Carvalho. “Log Skeletons: A Classification Approach to Process Discovery.” arXiv preprint arXiv:1806.08247 (2018).

This method is considered the most accurate classification approach for determining whether a trace belongs to (the language of) a log. The log skeleton calculates an object containing a list of relations for a given log.

Inspect Relations:

• Equivalence: Pairs of activities that always occur with the same frequency within a trace.
• Always-After: Pairs of activities (A, B) such that whenever activity A occurs, it is always followed by activity B at some point in the future of the trace.
• Always-Before: Pairs of activities (B, A) such that whenever activity B occurs, it is always preceded by activity A at some point in the past of the trace.
• Never-Together: Pairs of activities (A, B) that never occur together in the history of the trace.
• Directly-Follows: A list of directly-follows relations for the log.
• Activity Frequency: The number of possible occurrences of each activity per trace.

A noise threshold can also be specified. In this case, more relations are discovered, as the conditions are relaxed.

Suppose we take the running-example.xes log:

import pm4py
import os
import pandas
if __name__ == "__main__":
    log: pandas.DataFrame = pm4py.read_xes(os.path.join("tests", "input_data", "running-example.xes"))

Next, calculate the log skeleton:

from pm4py.algo.discovery.log_skeleton import algorithm as lsk_discovery
if __name__ == "__main__":
    skeleton: dict = lsk_discovery.apply(
        log,
        parameters={lsk_discovery.Variants.CLASSIC.value.Parameters.NOISE_THRESHOLD: 0.0}
    )

The resulting skeleton might look like:

{
    'equivalence': {('pay compensation', 'register request'), ('examine thoroughly', 'register request'),
                    ('reject request', 'register request'), ('pay compensation', 'examine casually')},
    'always_after': {('register request', 'check ticket'), ('examine thoroughly', 'decide'),
                    ('register request', 'decide')},
    'always_before': {('pay compensation', 'register request'), ('pay compensation', 'decide'),
                      ('pay compensation', 'check ticket'), ('reject request', 'decide'),
                      ('pay compensation', 'examine casually'), ('reject request', 'check ticket'),
                      ('examine thoroughly', 'register request'), ('reject request', 'register request')},
    'never_together': {('pay compensation', 'reject request'), ('reject request', 'pay compensation')},
    'directly_follows': set(),
    'activ_freq': {
        'register request': {1},
        'examine casually': {0, 1, 3},
        'check ticket': {1, 2, 3},
        'decide': {1, 2, 3},
        'reinitiate request': {0, 1, 2},
        'examine thoroughly': {0, 1},
        'pay compensation': {0, 1},
        'reject request': {0, 1}
    }
}

To see how the log skeleton functions for classification and conformance purposes, change to another log (the receipt.xes log) and calculate a heavily filtered version to reduce the behavior:

import pm4py
import os
import pandas
if __name__ == "__main__":
    log: pandas.DataFrame = pm4py.read_xes(os.path.join("tests", "input_data", "receipt.xes"))
    from copy import deepcopy
    filtered_log: pandas.DataFrame = pm4py.filter_variants_top_k(log, 3)

Now, calculate the log skeleton for the filtered log and apply the classification as follows:

from pm4py.algo.conformance.log_skeleton import algorithm as lsk_conformance
if __name__ == "__main__":
    conf_result: dict = lsk_conformance.apply(log, skeleton)

This process will indicate whether each trace belongs to the filtered log or not. When deviations are found, it means the trace does not belong to the language of the original log.

You can also calculate a log skeleton on the original log by specifying a noise threshold (e.g., 0.03) to see its impact on classification:

from pm4py.algo.discovery.log_skeleton import algorithm as lsk_discovery
from pm4py.algo.conformance.log_skeleton import algorithm as lsk_conformance

if __name__ == "__main__":
    skeleton: dict = lsk_discovery.apply(
        log,
        parameters={lsk_discovery.Variants.CLASSIC.value.Parameters.NOISE_THRESHOLD: 0.03}
    )

    conf_result: dict = lsk_conformance.apply(log, skeleton)

Some traces may be classified as incorrect, even when using the original log, if a noise threshold is applied.

Alignments Between Logs

The playout API returns a legacy EventLog. The alignment API accepts it directly alongside the original DataFrame.

In some cases, performing an optimal alignment between an event log and a process model may be impractical. An alternative is to use an approximate alignment that highlights the main points of deviation. PM4Py provides support for alignments between two event logs. This alignment operation is based on the edit distance, where for each trace in the first log, the trace in the second log with the least edit distance is identified.

In the following example, you will see how to perform alignments between an event log and a simulated log obtained by performing a playout operation on the process model. First, load an example log and discover a process model using the inductive miner:

import pm4py
import pandas
from pm4py.objects.petri_net.obj import Marking, PetriNet
if __name__ == "__main__":
    log: pandas.DataFrame = pm4py.read_xes("tests/input_data/running-example.xes")
    net: PetriNet
    im: Marking
    fm: Marking
    net, im, fm = pm4py.discover_petri_net_inductive(log)

Then, perform a playout operation on the process model:

from pm4py.objects.log.obj import EventLog
if __name__ == "__main__":
    simulated_log: EventLog = pm4py.play_out(net, im, fm)

Next, perform alignments between the two logs:

from pm4py.algo.conformance.alignments.edit_distance import algorithm as logs_alignments
from typing import Any
if __name__ == "__main__":
    alignments: list[dict[str, Any]] = logs_alignments.apply(log, simulated_log)

The result is a list of alignments, each containing a list of moves (sync move, move on log n.1, move on log n.2).

It is also possible to perform anti-alignments. An anti-alignment corresponds to a trace from the second log that has the greatest edit distance from a given trace in the first log. To perform anti-alignments, use the following code:

from pm4py.algo.conformance.alignments.edit_distance import algorithm as logs_alignments
from typing import Any
if __name__ == "__main__":
    parameters: dict = {logs_alignments.Variants.EDIT_DISTANCE.value.Parameters.PERFORM_ANTI_ALIGNMENT: True}
    alignments: list[dict[str, Any]] = logs_alignments.apply(log, simulated_log, parameters=parameters)

Temporal Profile

We propose an implementation of the temporal profile model in PM4Py, as described in the following paper:

Stertz, Florian, Jürgen Mangler, and Stefanie Rinderle-Ma. "Temporal Conformance Checking at Runtime Based on Time-Infused Process Models." arXiv preprint arXiv:2008.07262 (2020).

A temporal profile measures, for every pair of activities in the log, the average time and the standard deviation between events with the specified activities. The time is measured between the completion of the first event and the start of the second event. This requires working with an interval log where events have two timestamps. The output of temporal profile discovery is a dictionary where each pair of activities (represented as a tuple) is associated with two values: the first is the average time, and the second is the average standard deviation.

A temporal profile can also be used for conformance checking on an event log. The time between pairs of activities in the log is compared to the values stored in the temporal profile. Specifically, a value is calculated to show how many standard deviations the time differs from the average. If this value exceeds a threshold (default set to 6, according to the six-sigma principles), the pair of activities is flagged. The result of conformance checking based on the temporal profile is a list containing deviations for each case in the log. Each deviation is represented by a pair of activities, along with the calculated value and the distance (in terms of standard deviations) from the average.

Example of Temporal Profile Conformance Checking

First, load an event log and apply the discovery algorithm:

import pm4py
from pm4py.algo.discovery.temporal_profile import algorithm as temporal_profile_discovery
import pandas

if __name__ == "__main__":
    log: pandas.DataFrame = pm4py.read_xes("tests/input_data/receipt.xes")
    temporal_profile: dict = temporal_profile_discovery.apply(log)

Then, apply conformance checking based on the temporal profile:

from pm4py.algo.conformance.temporal_profile import algorithm as temporal_profile_conformance
if __name__ == "__main__":
    results: list = temporal_profile_conformance.apply(log, temporal_profile)

Several parameters can be adjusted to customize the conformance checking process using the temporal profile.

LTL Checking

LTL (Linear Temporal Logic) Checking is a form of filtering and conformance checking where specific rules are verified against the process executions contained in the log. This allows the checking of more complex patterns, such as:

• Four-Eyes Principle: Two activities should be performed by different people. For example, the approval of an expense refund should be done by someone other than the person who entered the expense.
• Activity Repeated by Different People: The same activity in a process execution is repeated (indicating rework) by different people.

The verification of LTL rules requires providing the appropriate parameters for the specific rule. As such, this form of conformance checking is not fully automatic.

The LTL rules implemented in PM4Py are listed in the table below:

The rules can be applied to both traditional event logs (XES) and Pandas dataframes by accessing the packages pm4py.algo.filtering.log.ltl and pm4py.algo.filtering.pandas.ltl, respectively.