Importing IEEE XES files

from pm4py.objects.log.importer.xes import importer as xes_importer
log = xes_importer.apply('<path_to_xes_file.xes>')

print(log[0]) #prints the first trace of the log
print(log[0][0]) #prints the first event of the first trace

from pm4py.objects.log.importer.xes import importer as xes_importer
variant = xes_importer.Variants.ITERPARSE
parameters = {variant.value.Parameters.TIMESTAMP_SORT: True}
log = xes_importer.apply('<path_to_xes_file>',
                         variant=variant, parameters=parameters)

Importing CSV files

import pandas as pd
from pm4py.objects.log.util import dataframe_utils
from pm4py.objects.conversion.log import converter as log_converter

log_csv = pd.read_csv('<path_to_csv_file.csv>', sep=',')
log_csv = dataframe_utils.convert_timestamp_columns_in_df(log_csv)
log_csv = log_csv.sort_values('<timestamp_column>')
event_log = log_converter.apply(log_csv)

import pandas as pd
from pm4py.objects.conversion.log import converter as log_converter

log_csv = pd.read_csv('<path_to_csv_file.csv>', sep=',')
log_csv.rename(columns={'clientID': 'case:clientID'}, inplace=True)
parameters = {log_converter.Variants.TO_EVENT_LOG.value.Parameters.CASE_ID_KEY: 'case'}
event_log = log_converter.apply(log_csv, parameters=parameters, variant=log_converter.Variants.TO_EVENT_LOG)

Converting Event Data

Exporting IEEE XES files

from pm4py.objects.log.exporter.xes import exporter as xes_exporter
xes_exporter.apply(log, '<path_to_exported_log.xes>')

Exporting logs to CSV

import pandas as pd
from pm4py.objects.conversion.log import converter as log_converter
dataframe = log_converter.apply(log, variant=log_converter.Variants.TO_DATA_FRAME)
dataframe.to_csv('<path_to_csv_file.csv>')
                                

I/O with Other File Types

Filtering on timeframe

In the following paragraph, various methods regarding filtering with time frames are present. For each of the methods, the log and Pandas Dataframe methods are revealed.

from pm4py.algo.filtering.log.timestamp import timestamp_filter
filtered_log = timestamp_filter.filter_traces_contained
               (log, "2011-03-09 00:00:00", "2012-01-18 23:59:59")

from pm4py.algo.filtering.pandas.timestamp import timestamp_filter
df_timest_intersecting = timestamp_filter.filter_traces_intersecting
               (dataframe, "2011-03-09 00:00:00", "2012-01-18 23:59:59",
                                          parameters={timestamp_filter.Parameters.CASE_ID_KEY: "case:concept:name",
                                                      timestamp_filter.Parameters.TIMESTAMP_KEY: "time:timestamp"})

from pm4py.algo.filtering.log.timestamp import timestamp_filter
filtered_log = timestamp_filter.filter_traces_intersecting
               (log, "2011-03-09 00:00:00", "2012-01-18 23:59:59")

from pm4py.algo.filtering.pandas.timestamp import timestamp_filter
df_timest_intersecting = timestamp_filter.filter_traces_intersecting
               (dataframe, "2011-03-09 00:00:00", "2012-01-18 23:59:59",
                                          parameters={timestamp_filter.Parameters.CASE_ID_KEY: "case:concept:name",
                                                      timestamp_filter.Parameters.TIMESTAMP_KEY: "time:timestamp"})

from pm4py.algo.filtering.log.timestamp import timestamp_filter
filtered_log_events = timestamp_filter.apply_events
               (log, "2011-03-09 00:00:00", "2012-01-18 23:59:59")

from pm4py.algo.filtering.pandas.timestamp import timestamp_filter
df_timest_events = timestamp_filter.apply_events
               (dataframe, "2011-03-09 00:00:00", "2012-01-18 23:59:59",
                                          parameters={timestamp_filter.Parameters.CASE_ID_KEY: "case:concept:name",
                                                      timestamp_filter.Parameters.TIMESTAMP_KEY: "time:timestamp"})

Filter on case performance

from pm4py.algo.filtering.log.cases import case_filter
filtered_log = case_filter.filter_case_performance(log, 86400, 864000)

from pm4py.algo.filtering.pandas.cases import case_filter
df_cases = case_filter.filter_case_performance
               (dataframe, min_case_performance=86400, max_case_performance=864000,
                                          parameters={case_filter.Parameters.CASE_ID_KEY: "case:concept:name",
                                                      case_filter.Parameters.TIMESTAMP_KEY: "time:timestamp"})

Filter on start activities

In general, PM4Py offers two methods to filter a log or a dataframe on start activities. In the first method, a list of start activities has to be specified. On the activities that are contained in the list, the filter is applied on. In the second method, a decreasing factor is used. An explanation can be inspected by clicking on the button below.

from pm4py.algo.filtering.log.start_activities import start_activities_filter
log_start = start_activities_filter.get_start_activities(log)
filtered_log = start_activities_filter.apply(log, ["S1"]) #suppose "S1" is the start activity you want to filter on

from pm4py.algo.filtering.pandas.start_activities import start_activities_filter
log_start = start_activities_filter.get_start_activities(dataframe)
df_start_activities = start_activities_filter.apply(dataframe, ["S1"],
                                          parameters={start_activities_filter.Parameters.CASE_ID_KEY: "case:concept:name",
                                                      start_activities_filter.Parameters.ACTIVITY_KEY: "concept:name"}) #suppose "S1" is the start activity you want to filter on
						

from pm4py.algo.filtering.log.start_activities import start_activities_filter
log_af_sa = start_activities_filter.apply_auto_filter
               (log, parameters={start_activities_filter.Parameters.DECREASING_FACTOR: 0.6})

from pm4py.algo.filtering.pandas.start_activities import start_activities_filter
df_auto_sa = start_activities_filter.apply_auto_filter
               (dataframe, parameters={start_activities_filter.Parameters.DECREASING_FACTOR: 0.6})

Filter on end activities

In general, PM4Py offers two methods to filter a log or a dataframe on end activities. In the first method, a list of end activities has to be specified. On the activities that are contained in the list, the filter is applied on. In the second method, a decreasing factor is used. An explanation can be inspected by clicking on the button in the start activity section.

from pm4py.algo.filtering.log.end_activities import end_activities_filter
end_activities = end_activities_filter.get_end_activities(log)
filtered_log = end_activities_filter.apply(log, ["pay compensation"])

from pm4py.algo.filtering.pandas.end_activities import end_activities_filter
end_activities = end_activities_filter.get_end_activities(df)
filtered_df = end_activities_filter.apply(df, ["pay compensation"],
                                          parameters={end_activities_filter.Parameters.CASE_ID_KEY: "case:concept:name",
                                                      end_activities_filter.Parameters.ACTIVITY_KEY: "concept:name"})

Filter on variants

A variant is a set of cases that share the same control-flow perspective, so a set of cases that share the same classified events (activities) in the same order. In this section, we will focus for all methods first on log objects, then we will continue with the dataframe.

from pm4py.algo.filtering.log.variants import variants_filter
variants = variants_filter.get_variants(log)

from pm4py.statistics.traces.generic.pandas import case_statistics
variants = case_statistics.get_variants_df(df,
                                          parameters={case_statistics.Parameters.CASE_ID_KEY: "case:concept:name",
                                                      case_statistics.Parameters.ACTIVITY_KEY: "concept:name"})

from pm4py.statistics.traces.generic.log import case_statistics
variants_count = case_statistics.get_variant_statistics(log)
variants_count = sorted(variants_count, key=lambda x: x['count'], reverse=True)

from pm4py.statistics.traces.generic.pandas import case_statistics
variants_count = case_statistics.get_variant_statistics(df,
                                          parameters={case_statistics.Parameters.CASE_ID_KEY: "case:concept:name",
                                                      case_statistics.Parameters.ACTIVITY_KEY: "concept:name",
                                                      case_statistics.Parameters.TIMESTAMP_KEY: "time:timestamp"})
variants_count = sorted(variants_count, key=lambda x: x['case:concept:name'], reverse=True)

from pm4py.algo.filtering.log.variants import variants_filter
             filtered_log1 = variants_filter.apply(log, variants)

from pm4py.algo.filtering.pandas.variants import variants_filter
             filtered_df1 = variants_filter.apply(df, variants,
                                          parameters={variants_filter.Parameters.CASE_ID_KEY: "case:concept:name",
                                                      variants_filter.Parameters.ACTIVITY_KEY: "concept:name"})

filtered_log2 = variants_filter.apply(log, variants, parameters={variants_filter.Parameters.POSITIVE: False})

filtered_df2 = variants_filter.apply(df, variants,
                                          parameters={variants_filter.Parameters.POSITIVE: False, variants_filter.Parameters.CASE_ID_KEY: "case:concept:name",
                                                      variants_filter.Parameters.ACTIVITY_KEY: "concept:name"})

auto_filtered_log = variants_filter.apply_auto_filter(log)

auto_filtered_df = variants_filter.apply_auto_filter(df,
                                          parameters={variants_filter.Parameters.POSITIVE: False, variants_filter.Parameters.CASE_ID_KEY: "case:concept:name",
                                                      variants_filter.Parameters.ACTIVITY_KEY: "concept:name"})

from pm4py.algo.filtering.log.variants import variants_filter

filtered_log = variants_filter.filter_log_variants_percentage(log, percentage=0.5)
                            

• The variants of the log are found along with their number of occurrences
• A number N is chosen such that if we take all the variants with at least N occurrences, we include a percentage of cases that is at least P, while if we choose N+1 we would include a percentage of cases that is below P.

• ABC (2 occurrences)
• A,B,C,AB,AC,BC,CB,BA,CA,ABCD,ABCE,ABCF,ABCG,ABCH,ABCI,ABCL,ABCM,ABCN (1 occurrence each)

• with percentage=1, all the 20 cases would have been kept.
• If we choose percentage=0.1, and N=1, then we include all the cases, while choosing N=2 we include only the cases of the first variant (that are the 5% of the log, hence N=2 is not valid according to the above principle).
• If we choose percentage=0.05, and N=2, then we include exactly 5% of the cases of the log, that is the minimum requirement.

Filter on attributes values

Filtering on attributes values permits alternatively to:

• Keep cases that contains at least an event with one of the given attribute values
• Remove cases that contains an event with one of the the given attribute values
• Keep events (trimming traces) that have one of the given attribute values
• Remove events (trimming traces) that have one of the given attribute values

Example of attributes are the resource (generally contained in org:resource attribute) and the activity (generally contained in concept:name attribute). As noted before, the first method can be applied on log objects, the second on dataframe objects.

from pm4py.algo.filtering.log.attributes import attributes_filter
activities = attributes_filter.get_attribute_values(log, "concept:name")
resources = attributes_filter.get_attribute_values(log, "org:resource")

from pm4py.algo.filtering.pandas.attributes import attributes_filter
activities = attributes_filter.get_attribute_values(df, attribute_key="concept:name")
resources = attributes_filter.get_attribute_values(df, attribute_key="org:resource")

tracefilter_log_pos = attributes_filter.apply(log, ["Resource10"],
                                          parameters={attributes_filter.Parameters.ATTRIBUTE_KEY: "org:resource", attributes_filter.Parameters.POSITIVE: True})
tracefilter_log_neg = attributes_filter.apply(log, ["Resource10"],
                                          parameters={attributes_filter.Parameters.ATTRIBUTE_KEY: "org:resource", attributes_filter.Parameters.POSITIVE: False})

df_traces_pos = attributes_filter.apply(df, ["Resource10"],
                                          parameters={attributes_filter.Parameters.CASE_ID_KEY: "case:concept:name", attributes_filter.Parameters.ATTRIBUTE_KEY: "org:resource", attributes_filter.Parameters.POSITIVE: True})
df_traces_neg = attributes_filter.apply(df, ["Resource10"],
                                          parameters={attributes_filter.Parameters.CASE_ID_KEY: "case:concept:name", attributes_filter.Parameters.ATTRIBUTE_KEY: "org:resource", attributes_filter.Parameters.POSITIVE: False})

tracefilter_log_pos = attributes_filter.apply_events(log, ["Resource10"],
                                          parameters={attributes_filter.Parameters.ATTRIBUTE_KEY: "org:resource", attributes_filter.Parameters.POSITIVE: True})
tracefilter_log_neg = attributes_filter.apply_events(log, ["Resource10"],
                                          parameters={attributes_filter.Parameters.ATTRIBUTE_KEY: "org:resource", attributes_filter.Parameters.POSITIVE: False})

df_traces_pos = attributes_filter.apply_events(df, ["Resource10"],
                                          parameters={attributes_filter.Parameters.CASE_ID_KEY: "case:concept:name", attributes_filter.Parameters.ATTRIBUTE_KEY: "org:resource", attributes_filter.Parameters.POSITIVE: True})
df_traces_neg = attributes_filter.apply_events(df, ["Resource10"],
                                          parameters={attributes_filter.Parameters.CASE_ID_KEY: "case:concept:name", attributes_filter.Parameters.ATTRIBUTE_KEY: "org:resource", attributes_filter.Parameters.POSITIVE: False})

from pm4py.algo.filtering.log.attributes import attributes_filter
filtered_log = attributes_filter.apply_auto_filter(log, parameters={
    attributes_filter.Parameters.ATTRIBUTE_KEY: "concept:name", attributes_filter.Parameters.DECREASING_FACTOR: 0.6})

from pm4py.algo.filtering.pandas.attributes import attributes_filter
filtered_df = attributes_filter.apply_auto_filter(df, parameters={
    attributes_filter.Parameters.CASE_ID_KEY: "case:concept:name", attributes_filter.Parameters.ATTRIBUTE_KEY: "concept:name", attributes_filter.Parameters.DECREASING_FACTOR: 0.6})

Filter on numeric attribute values

Filtering on numeric attribute values provide options that are similar to filtering on string attribute values (that we already considered).

import os
from pm4py.objects.log.importer.xes import importer as xes_importer
log = xes_importer.apply(os.path.join("tests", "input_data", "roadtraffic100traces.xes"))

from pm4py.algo.filtering.log.attributes import attributes_filter
filtered_log_events = attributes_filter.apply_numeric_events(log, 34, 36,
                                             parameters={attributes_filter.Parameters.ATTRIBUTE_KEY: "amount"})

filtered_log_cases = attributes_filter.apply_numeric(log, 34, 36,
                                             parameters={attributes_filter.Parameters.ATTRIBUTE_KEY: "amount"})

filtered_log_cases = attributes_filter.apply_numeric(log, 34, 500,
                                             parameters={attributes_filter.Parameters.ATTRIBUTE_KEY: "amount",
                                                         attributes_filter.Parameters.STREAM_FILTER_KEY1: "concept:name",
                                                         attributes_filter.Parameters.STREAM_FILTER_VALUE1: "Add penalty"})

import os
import pandas as pd
from pm4py.objects.log.util import dataframe_utils
df = pd.read_csv(os.path.join("tests", "input_data", "roadtraffic100traces.csv"))
df = dataframe_utils.convert_timestamp_columns_in_df(df)

from pm4py.algo.filtering.pandas.attributes import attributes_filter
filtered_df_events = attributes_filter.apply_numeric_events(df, 34, 36,
                                             parameters={attributes_filter.Parameters.CASE_ID_KEY: "case:concept:name", attributes_filter.Parameters.ATTRIBUTE_KEY: "amount"})

filtered_df_cases = attributes_filter.apply_numeric(df, 34, 36,
                                             parameters={attributes_filter.Parameters.CASE_ID_KEY: "case:concept:name", attributes_filter.Parameters.ATTRIBUTE_KEY: "amount"})

filtered_df_cases = attributes_filter.apply_numeric(df, 34, 500,
                                             parameters={attributes_filter.Parameters.CASE_ID_KEY: "case:concept:name", attributes_filter.Parameters.ATTRIBUTE_KEY: "amount",
                                                         attributes_filter.Parameters.STREAM_FILTER_KEY1: "concept:name",
                                                         attributes_filter.Parameters.STREAM_FILTER_VALUE1: "Add penalty"})

Alpha Miner

The alpha miner is one of the most known Process Discovery algorithm and is able to find:

• A Petri net model where all the transitions are visible and unique and correspond to classified events (for example, to activities).
• An initial marking that describes the status of the Petri net model when a execution starts.
• A final marking that describes the status of the Petri net model when a execution ends.

We provide an example where a log is read, the Alpha algorithm is applied and the Petri net along with the initial and the final marking are found. The log we take as input is the running-example.xes.

import os
from pm4py.objects.log.importer.xes import importer as xes_importer
log = xes_importer.apply(os.path.join("tests","input_data","running-example.xes"))

from pm4py.algo.discovery.alpha import algorithm as alpha_miner
net, initial_marking, final_marking = alpha_miner.apply(log)

Inductive Miner

In PM4Py, we offer an implementation of the inductive miner (IM), of the inductive miner infrequent (IMf), and of the inductive miner directly-follows (IMd) algorithm. The papers describing the approaches are the following:

• Inductive Miner: Discovering block-structured process models from event logs-a constructive approach
• Inductive Miner infrequent: Discovering block-structured process models from event logs containing infrequent behaviour
• Inductive Miner directly-follows Scalable process discovery with guarantees

The basic idea of Inductive Miner is about detecting a 'cut' in the log (e.g. sequential cut, parallel cut, concurrent cut and loop cut) and then recur on sublogs, which were found applying the cut, until a base case is found. The Directly-Follows variant avoids the recursion on the sublogs but uses the Directly Follows graph.

Inductive miner models usually make extensive use of hidden transitions, especially for skipping/looping on a portion on the model. Furthermore, each visible transition has a unique label (there are no transitions in the model that share the same label).

Two process models can be derived: Petri Net and Process Tree.

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

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

from pm4py.algo.discovery.inductive import algorithm as inductive_miner
from pm4py.visualization.process_tree import visualizer as pt_visualizer

tree = inductive_miner.apply_tree(log)

gviz = pt_visualizer.apply(tree)
pt_visualizer.view(gviz)

from pm4py.objects.conversion.process_tree import converter as pt_converter
net, initial_marking, final_marking = pt_converter.apply(tree, variant=pt_converter.Variants.TO_PETRI_NET)

As we said in the introduction of the approach, the available variants inside PM4Py are three:

The list of parameters for such variants are:
Inspect parameters

Heuristic Miner

Heuristics Miner is an algorithm that acts on the Directly-Follows Graph, providing way to handle with noise and to find common constructs (dependency between two activities, AND). The output of the Heuristics Miner is an Heuristics Net, so an object that contains the activities and the relationships between them. The Heuristics Net can be then converted into a Petri net. The paper can be visited by clicking on the upcoming link: this link).

It is possible to obtain a Heuristic Net and a Petri Net.

from pm4py.objects.log.importer.xes import importer as xes_importer
import os
log_path = os.path.join("tests", "compressed_input_data", "09_a32f0n00.xes.gz")
log = xes_importer.apply(log_path)

from pm4py.algo.discovery.heuristics import algorithm as heuristics_miner
heu_net = heuristics_miner.apply_heu(log, parameters={heuristics_miner.Variants.CLASSIC.value.Parameters.DEPENDENCY_THRESH: 0.99})

from pm4py.visualization.heuristics_net import visualizer as hn_visualizer
gviz = hn_visualizer.apply(heu_net)
hn_visualizer.view(gviz)

from pm4py.algo.discovery.heuristics import algorithm as heuristics_miner
net, im, fm = heuristics_miner.apply(log, parameters={heuristics_miner.Variants.CLASSIC.value.Parameters.DEPENDENCY_THRESH: 0.99})

from pm4py.visualization.petri_net import visualizer as pn_visualizer
gviz = pn_visualizer.apply(net, im, fm)
pn_visualizer.view(gviz)

Directly-Follows Graph

Process models modeled using Petri nets have a well-defined semantic: a process execution starts from the places included in the initial marking and finishes at the places included in the final marking. In this section, another class of process models, Directly-Follows Graphs, are introduced. Directly-Follows graphs are graphs where the nodes represent the events/activities in the log and directed edges are present between nodes if there is at least a trace in the log where the source event/activity is followed by the target event/activity. On top of these directed edges, it is easy to represent metrics like frequency (counting the number of times the source event/activity is followed by the target event/activity) and performance (some aggregation, for example, the mean, of time inter-lapsed between the two events/activities).

import os
from pm4py.objects.log.importer.xes import importer as xes_importer
log = xes_importer.apply(os.path.join("tests","input_data","running-example.xes"))

from pm4py.algo.discovery.dfg import algorithm as dfg_discovery
dfg = dfg_discovery.apply(log)

from pm4py.visualization.dfg import visualizer as dfg_visualization
gviz = dfg_visualization.apply(dfg, log=log, variant=dfg_visualization.Variants.FREQUENCY)
dfg_visualization.view(gviz)

from pm4py.algo.discovery.dfg import algorithm as dfg_discovery
from pm4py.visualization.dfg import visualizer as dfg_visualization

dfg = dfg_discovery.apply(log, variant=dfg_discovery.Variants.PERFORMANCE)
gviz = dfg_visualization.apply(dfg, log=log, variant=dfg_visualization.Variants.PERFORMANCE)
dfg_visualization.view(gviz)

from pm4py.algo.discovery.dfg import algorithm as dfg_discovery
from pm4py.visualization.dfg import visualizer as dfg_visualization

dfg = dfg_discovery.apply(log, variant=dfg_discovery.Variants.PERFORMANCE)
parameters = {dfg_visualization.Variants.PERFORMANCE.value.Parameters.FORMAT: "svg"}
gviz = dfg_visualization.apply(dfg, log=log, variant=dfg_visualization.Variants.PERFORMANCE, parameters=parameters)
dfg_visualization.save(gviz, "dfg.svg")

Adding information about Frequency/Performance

Similar to the Directly-Follows graph, it is also possible to decorate the Petri net with frequency or performance information. This is done by using a replay technique on the model and then assigning frequency/performance to the paths. The variant parameter of the visualizer specifies which annotation should be used. The values for the variant parameter are the following:

• pn_visualizer.Variants.WO_DECORATION: This is the default value and indicates that the Petri net is not decorated.
• pn_visualizer.Variants.FREQUENCY: This indicates that the model should be decorated according to frequency information obtained by applying replay.
• pn_visualizer.Variants.PERFORMANCE: This indicates that the model should be decorated according to performance (aggregated by mean) information obtained by applying replay.

In the case the frequency and performance decoration are chosen, it is required to pass the log as a parameter of the visualization (it needs to be replayed).

from pm4py.visualization.petri_net import visualizer as pn_visualizer
parameters = {pn_visualizer.Variants.FREQUENCY.value.Parameters.FORMAT: "png"}
gviz = pn_visualizer.apply(net, initial_marking, final_marking, parameters=parameters, variant=pn_visualizer.Variants.FREQUENCY, log=log)
pn_visualizer.save(gviz, "inductive_frequency.png")

Classifier

Algorithms implemented in pm4py assume to classify events based on their activity name, which is usually reported inside the concept:name event attribute. In some contexts, it is useful to use another event attribute as activity:

• Importing an event log from a CSV does not assure to have a concept:name event attribute
• Multiple events in a case may refer to different lifecycles of the same activity

import os
from pm4py.objects.log.importer.xes import importer as xes_importer
from pm4py.algo.discovery.alpha import algorithm as alpha_miner
log = xes_importer.apply(os.path.join("tests","input_data","running-example.xes"))
parameters = {alpha_miner.Variants.ALPHA_CLASSIC.value.Parameters.ACTIVITY_KEY: "concept:name"}
net, initial_marking, final_marking = alpha_miner.apply(log, parameters=parameters)

For logs imported from XES format, a list of fields that could be used in order to classify events and apply Process Mining algorithms is usually reported in the classifiers section. The Standard classifier usually includes the activity name (the concept:name attribute) and the lifecycle (the lifecycle:transition attribute); the Event name classifier includes only the activity name.

In PM4Py, it is assumed that algorithms work on a single activity key. In order to use multiple fields, a new attribute should be inserted for each event as the concatenation of the two.

In the following, retrieval and insertion of a corresponding attribute regarding classifiers are discussed.

import os
from pm4py.objects.log.importer.xes import importer as xes_importer

log = xes_importer.apply(os.path.join("tests","input_data","receipt.xes"))
print(log.classifiers)

from pm4py.objects.log.util import insert_classifier
log, activity_key = insert_classifier.insert_activity_classifier_attribute(log, "Activity classifier")

from pm4py.algo.discovery.alpha import algorithm as alpha_miner
parameters = {alpha_miner.Variants.ALPHA_CLASSIC.value.Parameters.ACTIVITY_KEY: activity_key}
net, initial_marking, final_marking = alpha_miner.apply(log, parameters=parameters)

In the following, a technique is shown to insert a new attribute manually.

import os
from pm4py.objects.log.importer.xes import importer as xes_importer

log = xes_importer.apply(os.path.join("tests","input_data","receipt.xes"))
for trace in log:
 for event in trace:
  event["customClassifier"] = event["concept:name"] + event["lifecycle:transition"]

from pm4py.algo.discovery.alpha import algorithm as alpha_miner
parameters = {alpha_miner.Variants.ALPHA_CLASSIC.value.Parameters.ACTIVITY_KEY: "customClassifier"}
net, initial_marking, final_marking = alpha_miner.apply(log, parameters=parameters)

Correlation Miner

In Process Mining, we are used to have logs containing at least:

• A case identifier
• An activity
• A timestamp

The case identifier associates an event, happening to a system, to a particular execution of the process. This permits to apply algorithms such as process discovery, conformance checking, …

However, in some systems (for example, the data collected from IoT systems), it may be difficult to associate a case identifier. On top of these logs, performing classic process mining is impossible. Correlation mining borns as a response to the challenge to extract a process model from such event logs, that permits to read useful information that is contained in the logs without a case identifier, that contains only:

• An activity column
• A timestamp column

In this description, we assume there is a total order on events (that means that no events happen in the same timestamp). Situations where a total order is not defined are more complicated.

The Correlation Miner is an approach proposed in:

Pourmirza, Shaya, Remco Dijkman, and Paul Grefen. “Correlation miner: mining business process models and event correlations without case identifiers.” International Journal of Cooperative Information Systems 26.02 (2017): 1742002.

That aims to resolve this problem by resolving an (integer) linear problem defined on top of:

• The P/S matrix: expressing the relationship of order between the activities as recorded in the log.
• The Duration matrix: expressing an aggregation of the duration between two activities, obtained by solving an optimization problem

The solution of this problem provides a set of couples of activities that are, according to the approach, in directly-follows relationship, along with the strength of the relationship. This is the “frequency” DFG.

A “performance” DFG can be obtained by the duration matrix, keeping only the entries that appear in the solution of the problem (i.e., the couples of activities that appear in the “frequency” DFG).

This can be then visualized (using for example the PM4Py DFG visualization).

To have a “realistic” example (for which we know the “real” DFG), we can take an existing log and simply remove the case ID column .. Trying then to reconstruct the DFG without having that.

import pandas as pd
from pm4py.objects.log.util import dataframe_utils
df = pd.read_csv(os.path.join("tests", "input_data", "receipt.csv"))
df = dataframe_utils.convert_timestamp_columns_in_df(df)
df = df[["concept:name", "time:timestamp"]]
                                

from pm4py.algo.discovery.correlation_mining import algorithm as correlation_miner

frequency_dfg, performance_dfg = correlation_miner.apply(df, parameters={correlation_miner.Variants.CLASSIC.value.Parameters.ACTIVITY_KEY: "concept:name",
                                correlation_miner.Variants.CLASSIC.value.Parameters.TIMESTAMP_KEY: "time:timestamp"})
                                

activities_freq = dict(df["concept:name"].value_counts())
                                

from pm4py.visualization.dfg import visualizer as dfg_visualizer
gviz_freq = dfg_visualizer.apply(frequency_dfg, variant=dfg_visualizer.Variants.FREQUENCY, activities_count=activities_freq, parameters={"format": "svg"})
gviz_perf = dfg_visualizer.apply(performance_dfg, variant=dfg_visualizer.Variants.PERFORMANCE, activities_count=activities_freq, parameters={"format": "svg"})
dfg_visualizer.view(gviz_freq)
dfg_visualizer.view(gviz_perf)
                                

Visualizing the DFGs, we can say that the correlation miner was able to discover a visualization where the main path is clear. Different variants of the correlation miner are available:

Temporal Profile

We propose in PM4Py an implementation of the temporal profile model. This has been described in:

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 couple of activities in the log the average time and the standard deviation between events having the provided activities. The time is measured between the completion of the first event and the start of the second event. Hence, it is assumed to work with an interval log where the events have two timestamps. The output of the temporal profile discovery is a dictionary where each couple of activities (expressed as a tuple) is associated to a couple of numbers, the first is the average and the second is the average standard deviation.

We provide an example of discovery for the temporal profile.

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

log = pm4py.read_xes("tests/input_data/running-example.xes")
parameters = {}
temporal_profile = temporal_profile_discovery.apply(log, parameters=parameters)
                                

Some parameters can be used in order to customize the execution of the temporal profile:

Importing and exporting

Petri nets, along with their initial and final marking, can be imported/exported from the PNML file format. The code on the right-hand side can be used to import a Petri net along with the initial and final marking.

import os
from pm4py.objects.petri_net.importer import importer as pnml_importer
net, initial_marking, final_marking = pnml_importer.apply(os.path.join("tests","input_data","running-example.pnml"))

from pm4py.visualization.petri_net import visualizer as pn_visualizer
gviz = pn_visualizer.apply(net, initial_marking, final_marking)
pn_visualizer.view(gviz)

from pm4py.objects.petri_net.exporter import exporter as pnml_exporter
pnml_exporter.apply(net, initial_marking, "petri.pnml")

pnml_exporter.apply(net, initial_marking, "petri_final.pnml", final_marking=final_marking)

Petri Net properties

This section is about how to get the properties of a Petri Net. A property of the pet is, for example, a the enabled transition in a particular marking. However, also a list of places, transitions or arcs can be inspected.

from pm4py.objects.petri_net import semantics
transitions = semantics.enabled_transitions(net, initial_marking)

places = net.places
transitions = net.transitions
arcs = net.arcs

for place in places:
 print("\nPLACE: "+place.name)
 for arc in place.in_arcs:
  print(arc.source.name, arc.source.label)

Creating a new Petri Net

In this section, an overview of the code necessary to create a new Petri net with places, transitions, and arcs is provided. A Petri net object in pm4py should be created with a name.

# creating an empty Petri net
from pm4py.objects.petri_net.obj import PetriNet, Marking
net = PetriNet("new_petri_net")

# creating source, p_1 and sink place
source = PetriNet.Place("source")
sink = PetriNet.Place("sink")
p_1 = PetriNet.Place("p_1")
# add the places to the Petri Net
net.places.add(source)
net.places.add(sink)
net.places.add(p_1)

# Create transitions
t_1 = PetriNet.Transition("name_1", "label_1")
t_2 = PetriNet.Transition("name_2", "label_2")
# Add the transitions to the Petri Net
net.transitions.add(t_1)
net.transitions.add(t_2)

# Add arcs
from pm4py.objects.petri_net.utils import petri_utils
petri_utils.add_arc_from_to(source, t_1, net)
petri_utils.add_arc_from_to(t_1, p_1, net)
petri_utils.add_arc_from_to(p_1, t_2, net)
petri_utils.add_arc_from_to(t_2, sink, net)

# Adding tokens
initial_marking = Marking()
initial_marking[source] = 1
final_marking = Marking()
final_marking[sink] = 1

from pm4py.objects.petri_net.exporter import exporter as pnml_exporter
pnml_exporter.apply(net, initial_marking, "createdPetriNet1.pnml", final_marking=final_marking)

from pm4py.visualization.petri_net import visualizer as pn_visualizer
gviz = pn_visualizer.apply(net, initial_marking, final_marking)
pn_visualizer.view(gviz)

from pm4py.visualization.petri_net import visualizer as pn_visualizer
parameters = {pn_visualizer.Variants.WO_DECORATION.value.Parameters.FORMAT:"svg"}
gviz = pn_visualizer.apply(net, initial_marking, final_marking, parameters=parameters)
pn_visualizer.view(gviz)

from pm4py.visualization.petri_net import visualizer as pn_visualizer
parameters = {pn_visualizer.Variants.WO_DECORATION.value.Parameters.FORMAT: "svg"}
gviz = pn_visualizer.apply(net, initial_marking, final_marking, parameters=parameters)
pn_visualizer.save(gviz, "alpha.svg")

Maximal Decomposition

The decomposition technique proposed in this section is useful for conformance checking purpose. Indeed, splitting the overall model in smaller models can reduce the size of the state space, hence increasing the performance of the conformance checking operation. We propose to use the decomposition technique (maximal decomposition of a Petri net) described in:

Van der Aalst, Wil MP. “Decomposing Petri nets for process mining: A generic approach.” Distributed and Parallel Databases 31.4 (2013): 471-507.

We can see an example of maximal decomposition on top of the Petri net extracted by the Alpha Miner on top of the Running Example log.

import os
from pm4py.objects.log.importer.xes import importer as xes_importer
from pm4py.algo.discovery.alpha import algorithm as alpha_miner

log = xes_importer.apply(os.path.join("tests", "input_data", "running-example.xes"))
net, im, fm = alpha_miner.apply(log)
                                

from pm4py.objects.petri_net.utils.decomposition import decompose

list_nets = decompose(net, im, fm)
                                

from pm4py.visualization.petri_net import visualizer
gviz = []
for index, model in enumerate(list_nets):
    subnet, s_im, s_fm = model

    gviz.append(visualizer.apply(subnet, s_im, s_fm, parameters={visualizer.Variants.WO_DECORATION.value.Parameters.FORMAT: "png"}))
    visualizer.save(gviz[-1], str(index)+".png")
                                

A log that is fit according to the original model is also fit (projecting on the activities of the net) for these nets. Conversely, any deviation on top of these models represent a deviation also on the original model.

Reachability Graph

A reachability graph is a transition system that can constructed on any Petri net along with an initial marking, and is the graph of all the markings of the Petri net. These markings are connected by as many edges as many transitions connect the two different markings.

The main goal of the reachability graph is to provide an understanding of the state space of the Petri net. Usually, Petri nets containing a lot of concurrency have an incredibly big reachability graph. The same computation of the reachability graph may be unfeasible for such models.

from pm4py.objects.petri_net.utils import reachability_graph

ts = reachability_graph.construct_reachability_graph(net, im)
                                

from pm4py.visualization.transition_system import visualizer as ts_visualizer

gviz = ts_visualizer.apply(ts, parameters={ts_visualizer.Variants.VIEW_BASED.value.Parameters.FORMAT: "svg"})
ts_visualizer.view(gviz)
                                

Token-based replay

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.

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.

import os
from pm4py.objects.log.importer.xes import importer as xes_importer
from pm4py.algo.discovery.alpha import algorithm as alpha_miner

log = xes_importer.apply(os.path.join("tests", "input_data", "running-example.xes"))

net, initial_marking, final_marking = alpha_miner.apply(log)

from pm4py.algo.conformance.tokenreplay import algorithm as token_replay
replayed_traces = token_replay.apply(log, net, initial_marking, final_marking)

Diagnostics (TBR)

The execution of token-based replay in PM4Py permits to obtain detailed information about transitions that did not execute correctly, or activities that are in the log and not in the model. In particular, executions that do not match the model are expected to take longer throughput time.

The diagnostics that are provided by PM4Py are the following:

• Throughput analysis on the transitions that are executed in an unfit way according to the process model (the Petri net).
• Throughput analysis on the activities that are not contained in the model.
• Root Cause Analysis on the causes that lead to an unfit execution of the transitions.
• Root Cause Analysis on the causes that lead to executing activities that are not contained in the process model.

                                    import os
from pm4py.objects.log.importer.xes import importer as xes_importer
log = xes_importer.apply(os.path.join("tests", "input_data", "receipt.xes"))
                                

from pm4py.algo.filtering.log.auto_filter.auto_filter import apply_auto_filter
filtered_log = apply_auto_filter(log)
                                

from pm4py.algo.discovery.inductive import algorithm as inductive_miner
net, initial_marking, final_marking = inductive_miner.apply(filtered_log)
                                

from pm4py.algo.conformance.tokenreplay import algorithm as token_based_replay
parameters_tbr = {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, place_fitness, trans_fitness, unwanted_activities = token_based_replay.apply(log, net,
                                                                                              initial_marking,
                                                                                              final_marking,
                                                                                              parameters=parameters_tbr)
                                

Then, we pass to diagnostics information.

Throughput analysis (unfit execution)

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

Obtaining an output where is clear that unfit executions lead to much higher throughput times (from 126 to 146 times higher throughput time).

Throughput analysis (activities)

from pm4py.algo.conformance.tokenreplay.diagnostics import duration_diagnostics
act_diagnostics = 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 diagnostics context is a decision tree that permits to understand the causes of a deviation. In the following examples, for each deviation, a different decision tree is built and visualized.

# build decision trees
string_attributes = ["org:group"]
numeric_attributes = []
parameters = {"string_attributes": string_attributes, "numeric_attributes": numeric_attributes}
                                

Root Cause Analysis (unfit execution)

from pm4py.algo.conformance.tokenreplay.diagnostics import root_cause_analysis
trans_root_cause = root_cause_analysis.diagnose_from_trans_fitness(log, trans_fitness, parameters=parameters)
                                

from pm4py.visualization.decisiontree import visualizer as dt_vis
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 could be called
    gviz = dt_vis.apply(clf, feature_names, classes)
    dt_vis.view(gviz)
                                

Root Cause Analysis (activities that are not in the model)

from pm4py.algo.conformance.tokenreplay.diagnostics import root_cause_analysis
act_root_cause = root_cause_analysis.diagnose_from_notexisting_activities(log, unwanted_activities,
                                                                          parameters=parameters)
                                

from pm4py.visualization.decisiontree import visualizer as dt_vis
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 could be called
    gviz = dt_vis.apply(clf, feature_names, classes)
    dt_vis.view(gviz)
                                

Alignments

PM4Py comes with the following set of linear solvers: PuLP (available for any platform), CVXOPT (available for the most widely used platforms including Windows/Linux for Python 3.6/3.7). Alternatively, ORTools can also be used and installed from PIP.

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.

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

log = xes_importer.apply(os.path.join("tests", "input_data", "running-example.xes"))

net, initial_marking, final_marking = inductive_miner.apply(log)

from pm4py.algo.conformance.alignments.petri_net import algorithm as alignments
aligned_traces = alignments.apply_log(log, net, initial_marking, final_marking)

print(alignments)

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

# define the activity key in the parameters
parameters = {inductive_miner.Variants.DFG_BASED.value.Parameters.ACTIVITY_KEY: "customClassifier", alignments.Variants.VERSION_STATE_EQUATION_A_STAR.value.Parameters.ACTIVITY_KEY: "customClassifier"}

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

from pm4py.algo.evaluation.replay_fitness import algorithm as replay_fitness
log_fitness = replay_fitness.evaluate(alignments, variant=replay_fitness.Variants.ALIGNMENT_BASED)

print(log_fitness) 

It is also possible to select other parameters for 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.

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

parameters = {}
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 = alignments.apply_log(log, net, initial_marking, final_marking, parameters=parameters)

Different variants of the alignments are available:

Among the common parameters of these variants, we have:
Inspect parameters

Decomposition of Alignments

Alignments represent a computationally expensive problem on models that contain a lot of concurrency. Yet, they are the conformance checking technique that provides the best results in term of finding a match between the process execution(s) and the model. To overcome the difficulties related to the size of the state space, various attempts to decompose the model into “smaller” pieces, into which the alignment is easier and still permit to diagnose problems, have been done.

We have seen how to obtain a maximal decomposition of the Petri net model. Now we can see how to perform the decomposition of alignments (that is based on a maximal decomposition of the Petri net model). The approach described here has been 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.

The recomposition permits to understand whether each step of the process has been executed in a sync way or some deviations happened. First, an alignment is performed on top of the decomposed Petri nets.

Then, the agreement between the activities at the border is checked. If a disagreement is found, the two components that are disagreeing are merged and the alignment is repeated on them.

When the steps are agreeing between the different alignments of the components, these can be merged in a single alignment. The order of recomposition is based on the Petri net graph. Despite that, in the case of concurrency, the “recomposed” alignment contains a valid list of moves that may not be in the correct order.

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

conf = decomp_alignments.apply(log, net, im, fm, parameters={decomp_alignments.Variants.RECOMPOS_MAXIMAL.value.Parameters.PARAM_THRESHOLD_BORDER_AGREEMENT: 2})

Since decomposed models are expected to have less concurrency, the components are aligned using a Dijkstra approach. In the case of border disagreements, this can degrade the performance of the algorithm.

It should be noted that this is not an approximation technique; according to the authors, it should provide the same fitness as the original alignments.

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

fitness = rp_fitness_evaluator.evaluate(conf, variant=rp_fitness_evaluator.Variants.ALIGNMENT_BASED)
                                

Footprints

Footprints are a very basic (but scalable) conformance checking technique to compare entities (such that event logs, DFGs, Petri nets, process trees, any other kind of model). Essentially, a relationship between any couple of activities of the log/model is inferred. This can include:

• Directly-Follows Relationships: in the log/model, it is possible that the activity A is directly followed by B.
• Directly-Before Relationships: in the log/model, it is possible that the 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, that describes for each couple of activities the footprint relationship. It is possible to calculate that for different types of models and for the entire event log, but also trace-by-trace (if the local behavior is important).

from pm4py.algo.simulation.tree_generator import simulator as tree_gen
from pm4py.objects.log.importer.xes import importer as xes_importer
import os
log = xes_importer.apply(os.path.join("tests", "input_data", "running-example.xes"))

from pm4py.algo.discovery.inductive import algorithm as inductive_miner

net, im, fm = inductive_miner.apply(log)

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

fp_log = footprints_discovery.apply(log, variant=footprints_discovery.Variants.ENTIRE_EVENT_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, as keys, sequence (expressing directly-follows relationships) and parallel (expressing the parallel behavior that can happen in either way).

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

fp_trace_by_trace = footprints_discovery.apply(log, variant=footprints_discovery.Variants.TRACE_BY_TRACE)

fp_net = footprints_discovery.apply(net, im, fm)

And are the following:

{‘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, as keys, sequence (expressing directly-follows relationships) and parallel (expressing the parallel behavior that can happen in either way).

It is possible to visualize a comparison between the footprints of the (entire) log and the footprints of the (entire) model.

from pm4py.visualization.footprints import visualizer as fp_visualizer

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

from pm4py.visualization.footprints import visualizer as fp_visualizer

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

from pm4py.objects.log.importer.xes import importer as xes_importer
import os
from copy import deepcopy
from pm4py.algo.filtering.log.variants import variants_filter

log = xes_importer.apply(os.path.join("tests", "input_data", "receipt.xes"))
filtered_log = variants_filter.apply_auto_filter(deepcopy(log))

from pm4py.algo.discovery.inductive import algorithm as inductive_miner
net, im, fm = inductive_miner.apply(filtered_log)

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

conf_fp = footprints_conformance.apply(fp_trace_by_trace, fp_net)

And will contain, for each trace of the log, a set with the deviations. 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 that contains deviations, there are two deviations, the first related to T06 Determine necessity of stop advice being executed before T04 Determine confirmation of receipt; the second related to T02 Check confirmation of receipt being followed by T06 Determine necessity of stop advice.

The traces for which the conformance returns nothing are fit (at least according to the footprints).

Footprints conformance checking is a way to identify obvious deviations, behavior of the log that is not allowed by the model.

On the log side, their scalability is wonderful! The calculation of footprints for a Petri net model may be instead more expensive.

If we change the underlying model, from Petri nets to process tree, it is possible to exploit its bottomup structure in order to calculate the footprints almost instantaneously.

from pm4py.objects.log.importer.xes import importer as xes_importer

log = xes_importer.apply("tests/input_data/running-example.xes")

from pm4py.algo.discovery.inductive import algorithm as inductive_miner

tree = inductive_miner.apply_tree(log)

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

fp_log = fp_discovery.apply(log, variant=fp_discovery.Variants.ENTIRE_EVENT_LOG)
fp_trace_trace = fp_discovery.apply(log, variant=fp_discovery.Variants.TRACE_BY_TRACE)
fp_tree = fp_discovery.apply(tree, variant=fp_discovery.Variants.PROCESS_TREE)

Each one 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

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

conf_result = fp_conformance.apply(fp_log, fp_tree, variant=fp_conformance.Variants.LOG_EXTENSIVE)

The result contains, for each item of the previous list, the violations.

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

fitness = evaluation.fp_fitness(fp_log, fp_tree, conf_result)
precision = evaluation.fp_precision(fp_log, fp_tree)

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

Log Skeleton

The concept of log skeleton has been described in the contribution

Verbeek, H. M. W., and R. Medeiros de Carvalho. “Log skeletons: A classification approach to process discovery.” arXiv preprint arXiv:1806.08247 (2018).

And is claimingly the most accurate classification approach to decide whether a trace belongs to (the language) of a log or not.

For a log, an object containing a list of relations is calculated.
Inspect relations

It is also possible to provide a noise threshold. In that case, more relations are found since the conditions are relaxed.

from pm4py.objects.log.importer.xes import importer as xes_importer
import os
log = xes_importer.apply(os.path.join("tests", "input_data", "running-example.xes"))

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

We can also print that:

{‘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}}}

We can see the relations (equivalence, always_after, always_before, never_together, directly_follows, activ_freq) as key of the object, and the values are the activities/couples of activities that follow such pattern.

from pm4py.objects.log.importer.xes import importer as xes_importer
import os
log = xes_importer.apply(os.path.join("tests", "input_data", "receipt.xes"))
from copy import deepcopy
from pm4py.algo.filtering.log.variants import variants_filter
filtered_log = variants_filter.apply_auto_filter(deepcopy(log))

from pm4py.algo.conformance.log_skeleton import algorithm as lsk_conformance
conf_result = lsk_conformance.apply(log, skeleton)
for trace in conf_result:
    print(conf_result)

In such way, we can get for each trace whether it has been classified as belonging to the filtered log, or not. When deviations are found, the trace does not belong to the language of the original log.

from pm4py.algo.discovery.log_skeleton import algorithm as lsk_discovery
skeleton = lsk_discovery.apply(log, parameters={lsk_discovery.Variants.CLASSIC.value.Parameters.NOISE_THRESHOLD: 0.03})

from pm4py.algo.conformance.log_skeleton import algorithm as lsk_conformance
conf_result = lsk_conformance.apply(log, skeleton)
for trace in conf_result:
    print(conf_result)

We can see that some traces are classified as uncorrect also calculating the log skeleton on the original log, if a noise threshold is provided.

Alignments between Logs

In some situations, performing an optimal alignment between an event log and a process model might be unfeasible. Hence, getting an approximated alignment that highlights the main points of deviation is an option. In PM4Py, we offer support for alignments between two event logs. Such alignment operation is based on the edit distance, i.e., for a trace of the first log, the trace of the second log which has the least edit distance is found. In the following example, we see how to perform alignments between an event log and the simulated log obtained by performing a playout operation on the process model.

import pm4py
log = pm4py.read_xes("tests/input_data/running-example.xes")
net, im, fm = pm4py.discover_petri_net_inductive(log)

from pm4py.algo.simulation.playout.petri_net import algorithm as simulator
simulated_log = simulator.apply(net, im, fm)

from pm4py.algo.conformance.alignments.edit_distance import algorithm as logs_alignments
parameters = {}
alignments = logs_alignments.apply(log, simulated_log, parameters=parameters)

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

With this utility, it's also possible to perform anti-alignments. In this case, an anti-alignment is corresponding to a trace of the second log that has the biggest edit distance against the given trace of the first log.

from pm4py.algo.conformance.alignments.edit_distance import algorithm as logs_alignments
parameters = {logs_alignments.Variants.EDIT_DISTANCE.value.Parameters.PERFORM_ANTI_ALIGNMENT: True}
alignments = logs_alignments.apply(log, simulated_log, parameters=parameters)

Temporal Profile

We propose in PM4Py an implementation of the temporal profile model. This has been described in:

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 couple of activities in the log the average time and the standard deviation between events having the provided activities. The time is measured between the completion of the first event and the start of the second event. Hence, it is assumed to work with an interval log where the events have two timestamps. The output of the temporal profile discovery is a dictionary where each couple of activities (expressed as a tuple) is associated to a couple of numbers, the first is the average and the second is the average standard deviation.

It is possible to use a temporal profile to perform conformance checking on an event log. The times between the couple of activities in the log are assessed against the numbers stored in the temporal profile. Specifically, a value is calculated that shows how many standard deviations the value is different from the average. If that value exceeds a threshold (by default set to 6, according to the six-sigma principles), then the couple of activities is signaled.

The output of conformance checking based on a temporal profile is a list containing the deviations for each case in the log. Each deviation is expressed as a couple of activities, along with the calculated value and the distance (based on number of standard deviations) from the average.

We provide an example of conformance checking based on a temporal profile.

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

log = pm4py.read_xes("tests/input_data/receipt.xes")
parameters = {}
temporal_profile = temporal_profile_discovery.apply(log, parameters=parameters)
                                

from pm4py.algo.conformance.temporal_profile import algorithm as temporal_profile_conformance
parameters = {}
results = temporal_profile_conformance.apply(log, temporal_profile, parameters=parameters)
                                

Some parameters can be used in order to customize the conformance checking of the temporal profile:

Extended Marking Equation

The extended marking equation, given a trace and a Petri net process model, provides an underestimation of the cost of an alignment going from the initial state of the corresponding synchronous product net to the final state of the corresponding synchronous product net.

It improves the quality of the underestimation in comparison to the better known marking equation because of the specification of some split points in which the state of the synchronous net is verified.

In PM4Py, we provide an implementation of the extended marking equation. The approach is described in the following research paper:

van Dongen, Boudewijn F. 'Efficiently computing alignments: using the extended marking equation.' 16th International Conference on Business Process Management, BPM 2018. Springer, 2018.

import pm4py, os

log = pm4py.read_xes(os.path.join("tests", "input_data", "running-example.xes"))
net, im, fm = pm4py.discover_petri_net_inductive(log)
                                

from pm4py.algo.analysis.extended_marking_equation import algorithm as extended_marking_equation_builder

ext_solver = extended_marking_equation_builder.build(log[0], sync_net, sync_im, sync_fm,
                                                     parameters={"split_points": [1, 3]})
                                

h, x = ext_solver.solve()
                                

firing_sequence, reach_fm, explained_events = ext_solver.get_firing_sequence(x)
                                

LTL Checking

LTL Checking is a form of filtering/conformance checking in which some rules are verified against the process executions contained in the log.

This permits to check more complex patterns such as:

• Four eyes principle: two given activities should be executed by two different people. For example, the approval of an expense refund should be generally done by a different person rather than the insertion of the expense refund.
• Activity repeated by different people: the same activity in a process execution is repeated (that means rework) from different people.

The verification of LTL rules requires the insertion of the required parameters (of the specific rule). Hence, this form of conformance checking is not automatic.

The LTL rules that are implemented in PM4Py are found in the following table:

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

Importing/Exporting Process Trees

In PM4Py, we offer support for importing/exporting process trees in the PTML format.

from pm4py.objects.process_tree.importer import importer as ptml_importer

tree = ptml_importer.apply("tests/input_data/running-example.ptml")
                                

from pm4py.objects.process_tree.exporter import exporter as ptml_exporter

ptml_exporter.apply(tree, "running-example.ptml")
                                

Generation of process trees

The approach 'PTAndLogGenerator', described by the scientific paper 'PTandLogGenerator: A Generator for Artificial Event Data', has been implemented in the PM4Py library.

from pm4py.algo.simulation.tree_generator import simulator as tree_gen
parameters = {}
tree = tree_gen.apply(parameters=parameters)

Generation of a log out of a process tree

from pm4py.objects.process_tree import semantics
log = semantics.generate_log(tree, no_traces=100)

Conversion into Petri net

from pm4py.objects.conversion.process_tree import converter as pt_converter
net, im, fm = pt_converter.apply(tree)

Visualize a Process Tree

print(tree)

from pm4py.visualization.process_tree import visualizer as pt_visualizer
gviz = pt_visualizer.apply(tree, parameters={pt_visualizer.Variants.WO_DECORATION.value.Parameters.FORMAT: "png"})
pt_visualizer.view(gviz)

Converting a Petri net to a Process Tree

We propose an approach to convert a block-structured accepting Petri net to a process tree. The implement approach is:

van Zelst, Sebastiaan J. "Translating Workflow Nets to Process Trees: An Algorithmic Approach." arXiv preprint arXiv:2004.08213 (2020).

The approach, given an accepting Petri net, returns a process tree if the Petri net is block-structured, while it raises an exception if the Petri net is not block-structured.

import pm4py
import os

log = pm4py.read_xes(os.path.join("tests", "input_data", "running-example.xes"))
net, im, fm = pm4py.discover_petri_net_alpha(log)

from pm4py.objects.conversion.wf_net import converter as wf_net_converter

tree = wf_net_converter.apply(net, im, fm)
print(tree)
                                

The method succeeds, since the accepting Petri net is block-structured, and discovers a process tree (incidentally, the same process tree as if the inductive miner was applied).

Automatic Feature Selection

In PM4Py, we offer ways to perform an automatic feature selection. As example, let us import the receipt log and perform an automatic feature selection on top of it.

from pm4py.objects.log.importer.xes import importer as xes_importer

log = xes_importer.apply("tests/input_data/receipt.xes")
                                

from pm4py.objects.log.util import get_log_representation

data, feature_names = get_log_representation.get_default_representation(log)
print(feature_names)
                                

Printing the value feature_names, we see that the following attributes were selected:

• The attribute channel at the trace level (this assumes values Desk, Intern, Internet, Post, e-mail)
• The attribute department at the trace level (this assumes values Customer contact, Experts, General)
• The attribute group at the event level (this assumes values EMPTY, Group 1, Group 12, Group 13, Group 14, Group 15, Group 2, Group 3, Group 4, Group 7).

No numeric attribute value is selected. If we print feature_names, we get the following representation:

[‘trace:channel@Desk’, ‘trace:channel@Intern’, ‘trace:channel@Internet’, ‘trace:channel@Post’, ‘trace:channel@e-mail’, ‘trace:department@Customer contact’, ‘trace:department@Experts’, ‘trace:department@General’, ‘event:org:group@EMPTY’, ‘event:org:group@Group 1’, ‘event:org:group@Group 12’, ‘event:org:group@Group 13’, ‘event:org:group@Group 14’, ‘event:org:group@Group 15’, ‘event:org:group@Group 2’, ‘event:org:group@Group 3’, ‘event:org:group@Group 4’, ‘event:org:group@Group 7’]

So, we see that we have different features for different values of the attribute. This is called one-hot encoding. Actually, a case is assigned to 0 if it does not contain an event with the given value for the attribute; a case is assigned to 1 if it contains at least one event with the attribute.

import pandas as pd
df = pd.DataFrame(data, columns=feature_names)
print(df)
                                

We can see the features assigned to each different case.

Manual feature selection

The manual feature selection permits to specify which attributes should be included in the feature selection. These may include for example:

• The activities performed in the process execution (contained usually in the event attribute concept:name ).
• The resources that perform the process execution (contained usually in the event attribute org:resource ).
• Some numeric attributes, at discretion of the user.

To do so, we have to call the method get_log_representation. The types of features that can be considered by a manual feature selection are:

Let’s consider for example a feature selection where we are interested to:

• If a process execution contains, or not, an activity.
• If a process execution contains, or not, a resource.
• If a process execution contains, or not, a directly-follows path between different activities.
• If a process execution contains, or not, a directly-follows path between different resources.

from pm4py.objects.log.util import get_log_representation

data, feature_names = get_log_representation.get_representation(log, str_ev_attr=["concept:name", "org:resource"],
                                                                str_tr_attr=[], num_ev_attr=[], num_tr_attr=[],
                                                                str_evsucc_attr=["concept:name", "org:resource"])
print(len(feature_names))
                                

Calculating useful features

Other features are for example the cycle and the lead time associated to a case.

Here, we may suppose to have:

• A log with lifecycles, where each event is instantaneous
• OR an interval log, where events may be associated to two timestamps (start and end timestamp).

from pm4py.objects.log.util import interval_lifecycle
log = interval_lifecycle.to_interval(log)
                                

from pm4py.objects.log.util import interval_lifecycle
from pm4py.util import constants

log = interval_lifecycle.assign_lead_cycle_time(log, parameters={
    constants.PARAMETER_CONSTANT_START_TIMESTAMP_KEY: "start_timestamp",
    constants.PARAMETER_CONSTANT_TIMESTAMP_KEY: "time:timestamp"})
                                

After the provision of the start timestamp attribute (in this case, start_timestamp) and of the timestamp attribute (in this case, time:timestamp), the following features are returned by the method:

• @@approx_bh_partial_cycle_time => incremental cycle time associated to the event (the cycle time of the last event is the cycle time of the instance)
• @@approx_bh_partial_lead_time => incremental lead time associated to the event
• @@approx_bh_overall_wasted_time => difference between the partial lead time and the partial cycle time values
• @@approx_bh_this_wasted_time => wasted time ONLY with regards to the activity described by the ‘interval’ event
• @@approx_bh_ratio_cycle_lead_time => measures the incremental Flow Rate (between 0 and 1).

from pm4py.objects.log.util import get_log_representation

data, feature_names = get_log_representation.get_representation(log, str_ev_attr=["concept:name", "org:resource"],
                                                                str_tr_attr=[],
                                                                num_ev_attr=["@@approx_bh_partial_cycle_time",
                                                                             "@@approx_bh_partial_lead_time",
                                                                             "@@approx_bh_overall_wasted_time",
                                                                             "@@approx_bh_this_wasted_time",
                                                                             "@approx_bh_ratio_cycle_lead_time"],
                                                                num_tr_attr=[],
                                                                str_evsucc_attr=["concept:name", "org:resource"])
                                

PCA – Reducing the number of features

Some techniques (such as the clustering, prediction, anomaly detection) suffer if the dimensionality of the dataset is too high. Hence, a dimensionality reduction technique (as PCA) helps to cope with the complexity of the data.

import pandas as pd

df = pd.DataFrame(data, columns=feature_names)
                                

It is possible to reduce the number of features using a techniques like PCA.

from sklearn.decomposition import PCA

pca = PCA(n_components=5)
df2 = pd.DataFrame(pca.fit_transform(df))
                                

So, from more than 400 columns, we pass to 5 columns that contains most of the variance.

Anomaly Detection

In this section, we consider the calculation of an anomaly score for the different cases. This is based on the features extracted; and to work better requires the application of a dimensionality reduction technique (such as the PCA in the previous section).

from sklearn.ensemble import IsolationForest
model=IsolationForest()
model.fit(df2)
df2["scores"] = model.decision_function(df2)
                                

df2["@@index"] = df2.index
df2 = df2[["scores", "@@index"]]
df2 = df2.sort_values("scores")
print(df2)
                                

Decision tree about the ending activity of a process

Decision trees are objects that help the understandement of the conditions leading to a particular outcome. In this section, several examples related to the construction of the decision trees are provided.

Ideas behind the building of decision trees are provided in scientific paper: de Leoni, Massimiliano, Wil MP van der Aalst, and Marcus Dees. 'A general process mining framework for correlating, predicting and clustering dynamic behavior based on event logs.'

The general scheme is the following:

• A representation of the log, on a given set of features, is obtained (for example, using one-hot encoding on string attributes and keeping numeric attributes as-they-are)
• A representation of the target classes is constructed
• The decision tree is calculated
• The decision tree is represented in some ways

A process instance may potentially finish with different activities, signaling different outcomes of the process instance. A decision tree may help to understand the reasons behind each outcome.

import os
from pm4py.objects.log.importer.xes import importer as xes_importer
log = xes_importer.apply(os.path.join("tests", "input_data", "roadtraffic50traces.xes"))

from pm4py.objects.log.util import get_log_representation
str_trace_attributes = []
str_event_attributes = ["concept:name"]
num_trace_attributes = []
num_event_attributes = ["amount"]
data, feature_names = get_log_representation.get_representation(
                           log, str_trace_attributes, str_event_attributes,
                           num_trace_attributes, num_event_attributes)

data, feature_names = get_log_representation.get_default_representation(log)

import pandas as pd
dataframe = pd.DataFrame(data, columns=feature_names)
                                

dataframe.to_csv("features.csv", index=False)
                                

from pm4py.objects.log.util import get_class_representation
target, classes = get_class_representation.get_class_representation_by_str_ev_attr_value_value(log, "concept:name")

from sklearn import tree
clf = tree.DecisionTreeClassifier()
clf.fit(data, target)

from pm4py.visualization.decisiontree import visualizer as dectree_visualizer
gviz = dectree_visualizer.apply(clf, feature_names, classes)

Decision tree about the duration of a case (Root Cause Analysis)

A decision tree about the duration of a case helps to understand the reasons behind an high case duration (or, at least, a case duration that is above the threshold).

import os
from pm4py.objects.log.importer.xes import importer as xes_importer
log = xes_importer.apply(os.path.join("tests", "input_data", "roadtraffic50traces.xes"))

from pm4py.objects.log.util import get_log_representation
str_trace_attributes = []
str_event_attributes = ["concept:name"]
num_trace_attributes = []
num_event_attributes = ["amount"]

data, feature_names = get_log_representation.get_representation(log, str_trace_attributes, str_event_attributes,
                                                             num_trace_attributes, num_event_attributes)

data, feature_names = get_log_representation.get_default_representation(log)

from pm4py.objects.log.util import get_class_representation
target, classes = get_class_representation.get_class_representation_by_trace_duration(log, 2 * 8640000)

from sklearn import tree
clf = tree.DecisionTreeClassifier()
clf.fit(data, target)

from pm4py.visualization.decisiontree import visualizer as dectree_visualizer
gviz = dectree_visualizer.apply(clf, feature_names, classes)

Decision Mining

Decision mining permits, provided:

• An event log
• A process model (an accepting Petri net)
• A decision point

To retrieve the features of the cases that go in the different directions. This permits, for example, to calculate a decision tree that explains the decisions.

from pm4py.objects.log.importer.xes import importer as xes_importer

log = xes_importer.apply("tests/input_data/running-example.xes")
                                

from pm4py.algo.discovery.inductive import algorithm as inductive_miner

net, im, fm = inductive_miner.apply(log)
                                

from pm4py.visualization.petri_net import visualizer

gviz = visualizer.apply(net, im, fm, parameters={visualizer.Variants.WO_DECORATION.value.Parameters.DEBUG: True})
visualizer.view(gviz)
                                

For this example, we choose the decision point p_10. There, a decision, is done between the activities examine casually and examine throughly.

from pm4py.algo.decision_mining import algorithm as decision_mining

X, y, class_names = decision_mining.apply(log, net, im, fm, decision_point="p_10")
                                

As we see, the outputs of the apply method are the following:

• X: a Pandas dataframe containing the features associated to the cases leading to a decision.
• y: a Pandas dataframe, that is a single column, containing the number of the class that is the output of the decision (in this case, the values possible are 0 and 1, since we have two target classes)
• class_names: the names of the output classes of the decision (in this case, examine casually and examine thoroughly).

These outputs can be used in a generic way with any classification or comparison technique.

In particular, decision trees can be useful. We provide a function to automate the discovery of decision trees out of the decision mining technique.

from pm4py.algo.decision_mining import algorithm as decision_mining

clf, feature_names, classes = decision_mining.get_decision_tree(log, net, im, fm, decision_point="p_10")
                                

from pm4py.visualization.decisiontree import visualizer as tree_visualizer

gviz = tree_visualizer.apply(clf, feature_names, classes)
                                

Throughput Time

from pm4py.statistics.traces.generic.log import case_statistics
all_case_durations = case_statistics.get_all_casedurations(log, parameters={
    case_statistics.Parameters.TIMESTAMP_KEY: "time:timestamp"})
                                    

from pm4py.statistics.traces.generic.log import case_statistics
median_case_duration = case_statistics.get_median_caseduration(log, parameters={
    case_statistics.Parameters.TIMESTAMP_KEY: "time:timestamp"
})
                                    

Case Arrival/Dispersion Ratio

from pm4py.statistics.traces.generic.log import case_arrival
case_arrival_ratio = case_arrival.get_case_arrival_avg(log, parameters={
    case_arrival.Parameters.TIMESTAMP_KEY: "time:timestamp"})
                                    

from pm4py.statistics.traces.generic.log import case_arrival
case_dispersion_ratio = case_arrival.get_case_dispersion_avg(log, parameters={
    case_arrival.Parameters.TIMESTAMP_KEY: "time:timestamp"})
                                    

Performance Spectrum

The performance spectrum is a novel visualization of the performance of the process of the time elapsed between different activities in the process executions. The performance spectrum has initially been described in:

Denisov, Vadim, et al. "The Performance Spectrum Miner: Visual Analytics for Fine-Grained Performance Analysis of Processes." BPM (Dissertation/Demos/Industry). 2018.

The performance spectrum assumes to work with an event log and a list of activities that are considered to build the spectrum. In the following example, the performance spectrum is built on the receipt event log including the Confirmation of receipt, T04 Determine confirmation of receipt and T10 Determine necessity to stop indication activities.

import pm4py
import os

log = pm4py.read_xes(os.path.join("tests", "input_data", "receipt.xes"))
pm4py.view_performance_spectrum(log, ["Confirmation of receipt", "T04 Determine confirmation of receipt",
                                     "T10 Determine necessity to stop indication"], format="svg")
                                    

In the aforementioned example, we see three horizontal lines, corresponding to the activities included in the spectrum, and many oblique lines that represent the elapsed times between two activities. The more obliquous lines are highlighted by a different color.

This permits to identify the timestamps in which the execution was more bottlenecked, and possible patterns (FIFO, LIFO, ...).

Business Hours

Given an interval event log (an EventLog object where each event is characterised by two timestamps, a start timestamp usually contained in the start_timestamp attribute and a completion timestamp usually contained in the time:timestamp attribute), the duration of the event is the difference between the completion timestamp and the start timestamp.

This may be inficiated by nights (where an activity is not actively worked), weekends (where the workers may not be at the workplace) and other kind of pauses. In PM4Py, a way to consider only the time in which the activity could actually be worked (so, excluding time outside of the working hours and weekends) is provided.

from pm4py.util.business_hours import BusinessHours
from datetime import datetime

st = datetime.fromtimestamp(100000000)
et = datetime.fromtimestamp(200000000)
bh_object = BusinessHours(st, et)
worked_time = bh_object.getseconds()
print(worked_time)
                                    

bh_object = BusinessHours(st, et, worktiming=[10, 16], weekends=[5, 6, 7])
worked_time = bh_object.getseconds()
print(worked_time)
                                    

Cycle Time and Waiting Time

Two important KPI for a process executions are:

• The Lead Time: the overall time in which the instance was worked, from the start to the end, without considering if it was actively worked or not.
• The Cycle Time: the overall time in which the instance was worked, from the start to the end, considering only the times where it was actively worked.

For these concepts, it is important to consider only business hours (so, excluding nights and weekends). Indeed, in that period the machinery and the workforce is at home, so could not proceed in working the instance, so the time “wasted” there is not recoverable.

Within ‘interval’ event logs (that have a start and an end timestamp), it is possible to calculate incrementally the lead time and the cycle time (event per event). The lead time and the cycle time that are reported on the last event of the case are the ones related to the process execution. With this, it is easy to understand which activities of the process have caused a bottleneck (e.g. the lead time increases significantly more than the cycle time).

The algorithm implemented in PM4Py start sorting each case by the start timestamp (so, activities started earlier are reported earlier in the log), and is able to calculate the lead and cycle time in all the situations, also the complex ones reported in the following picture:

In the following, we aim to insert the following attributes to events inside a log:
Attributes

The method that calculates the lead and the cycle time could accept the following optional parameters:
Inspect Parameters

from pm4py.objects.log.util import interval_lifecycle
enriched_log = interval_lifecycle.assign_lead_cycle_time(log)
                                    

With this, an enriched log that contains for each event the corresponding attributes for lead/cycle time is obtained.

Sojourn Time

This statistic work only with interval event logs, i.e., event logs where each event has a start timestamp and a completion timestamp.

The average sojourn time statistic permits to know, for each activity, how much time was spent executing the activity. This is calculated as the average of time passed between the start timestamp and the completion timestamp for the activity's events.

import pm4py
import os

log = pm4py.read_xes(os.path.join("tests", "input_data", "interval_event_log.xes"))
                                    

from pm4py.statistics.sojourn_time.log import get as soj_time_get

soj_time = soj_time_get.apply(log, parameters={soj_time_get.Parameters.TIMESTAMP_KEY: "time:timestamp", soj_time_get.Parameters.START_TIMESTAMP_KEY: "start_timestamp"})
print(soj_time)
                                    

The same statistic can be applied seamlessy on Pandas dataframes. We provide an alternative class for doing so: pm4py.statistics.sojourn_time.pandas.get

Concurrent Activities

This statistic work only with interval event logs, i.e., event logs where each event has a start timestamp and a completion timestamp.

In an interval event log, the definition of an order between the events is weaker. Different intersections between a couple of events in a case can happen:

• An event where the start timestamp is greater or equal than the completion timestamp of the other.
• An event where the start timestamp is greater or equal than the start timestamp of the other event, but is lower than the completion timestamp of the other event.

In particular, the latter case define an event-based concurrency, where several events are actively executed at the same time.

We might be interested in retrieving the set of activities for which such concurrent execution happens, and the frequency of such occurrence. We offer this type of calculation in PM4Py.

import pm4py
import os

log = pm4py.read_xes(os.path.join("tests", "input_data", "interval_event_log.xes"))
                                    

from pm4py.statistics.concurrent_activities.log import get as conc_act_get

conc_act = conc_act_get.apply(log, parameters={conc_act_get.Parameters.TIMESTAMP_KEY: "time:timestamp", conc_act_get.Parameters.START_TIMESTAMP_KEY: "start_timestamp"})
print(conc_act)
                                    

The same statistic can be applied seamlessy on Pandas dataframes. We provide an alternative class for doing so: pm4py.statistics.concurrent_activities.pandas.get

Eventually-Follows Graph

We provide an approach for the calculation of the eventually-follows graph. The eventually-follows graph (EFG) is a graph that represents the partial order of the events inside the process executions of the log.

Our implementation can be applied to both lifecycle logs, so logs where each event has only one timestamp, both to interval logs, where each event has a start and a completion timestamp. In the later, the start timestamp is actively considered for the definition of the EFG / partial order

In particular, the method assumes to work with lifecycle logs when a start timestamp is NOT passed in the parameters, while it assumes to work with interval logs when a start timestamp is passed in the parameters.

import pm4py
import os

log = pm4py.read_xes(os.path.join("tests", "input_data", "interval_event_log.xes"))
                                    

from pm4py.statistics.eventually_follows.log import get as efg_get

efg_graph = efg_get.apply(log)
print(efg_graph)
                                    

The same statistic can be applied seamlessy on Pandas dataframes. We provide an alternative class for doing so: pm4py.statistics.eventually_follows.pandas.get

Displaying Graphs

Graphs permits to understand several aspects of the current log (for example, the distribution of a numeric attribute, or the distribution of case duration, or the events over time).

Distribution of case duration

import os
from pm4py.objects.log.importer.xes import importer as xes_importer
log_path = os.path.join("tests","input_data","receipt.xes")
log = xes_importer.apply(log_path)

from pm4py.util import constants
from pm4py.statistics.traces.generic.log import case_statistics
x, y = case_statistics.get_kde_caseduration(log, parameters={constants.PARAMETER_CONSTANT_TIMESTAMP_KEY: "time:timestamp"})

from pm4py.visualization.graphs import visualizer as graphs_visualizer

gviz = graphs_visualizer.apply_plot(x, y, variant=graphs_visualizer.Variants.CASES)
graphs_visualizer.view(gviz)

gviz = graphs_visualizer.apply_semilogx(x, y, variant=graphs_visualizer.Variants.CASES)
graphs_visualizer.view(gviz)
                                    

Distribution of events over time

from pm4py.algo.filtering.log.attributes import attributes_filter

x, y = attributes_filter.get_kde_date_attribute(log, attribute="time:timestamp")

from pm4py.visualization.graphs import visualizer as graphs_visualizer

gviz = graphs_visualizer.apply_plot(x, y, variant=graphs_visualizer.Variants.DATES)
graphs_visualizer.view(gviz)
                                    

Distribution of a numeric attribute

import os
from pm4py.objects.log.importer.xes import importer as xes_importer

log_path = os.path.join("tests", "input_data", "roadtraffic100traces.xes")
log = xes_importer.apply(log_path)

from pm4py.algo.filtering.log.attributes import attributes_filter

x, y = attributes_filter.get_kde_numeric_attribute(log, "amount")

from pm4py.visualization.graphs import visualizer as graphs_visualizer

gviz = graphs_visualizer.apply_plot(x, y, variant=graphs_visualizer.Variants.ATTRIBUTES)
graphs_visualizer.view(gviz)

from pm4py.visualization.graphs import visualizer as graphs_visualizer

gviz = graphs_visualizer.apply_semilogx(x, y, variant=graphs_visualizer.Variants.ATTRIBUTES)
graphs_visualizer.view(gviz)
                                    

Dotted Chart

The dotted chart is a classic visualization of the events inside an event log across different dimensions. Each event of the event log is corresponding to a point. The dimensions are projected on a graph having:

• X axis: the values of the first dimension are represented there.
• Y-axis: the values of the second dimension are represented there.
• Color: the values of the third dimension are represented as different colors for the points of the dotted chart.

The values can be either string, numeric or date values, and are managed accordingly by the dotted chart.

The dotted chart can be built on different attributes. A convenient choice for the dotted chart is to visualize the distribution of cases and events over the time, with the following choices:

• X-axis: the timestamp of the event.
• Y-axis: the index of the case inside the event log.
• Color: the activity of the event.

The aforementioned choice permits to identify visually patterns such as:

• Batches.
• Variations in the case arrival rate.
• Variations in the case finishing rate.

In the following examples, we will build and visualize the dotted chart based on different selections of the attributes (default and custom).

import pm4py
import os

log = pm4py.read_xes(os.path.join("tests", "input_data", "receipt.xes"))
pm4py.view_dotted_chart(log, format="svg")
                                    

import pm4py
import os

log = pm4py.read_xes(os.path.join("tests", "input_data", "receipt.xes"))
pm4py.view_dotted_chart(log, format="svg", attributes=["concept:name", "org:resource", "org:group"])
                                    

Events Distribution

Observing the distribution of events over time permits to infer useful information about the work shifts, the working days, and the period of the year that are more or less busy.

import pm4py
import os

log = pm4py.read_xes(os.path.join("tests", "input_data", "receipt.xes"))
pm4py.view_events_distribution_graph(log, distr_type="days_week", format="svg")
                                    

The possible values for the parameter distr_type are:

• hours: plots the distribution over the hours of a day.
• days_week: plots the distribution over the days of a week.
• days_month: plots the distribution over the days of a month.
• months: plots the distribution over the months of a year.
• yearsr: plots the distribution over the different years of the log.

Rework

The rework statistic permits to identify the activities which have been repeated during the same process execution. This shows the underlying inefficiencies in the process.

In our implementation, the rework takes into account an event log / Pandas dataframe and returns a dictionary associating to each activity the number of cases containing the rework for the given activity.

import pm4py
import os

log = pm4py.read_xes(os.path.join("tests", "input_data", "receipt.xes"))
rework = pm4py.get_rework_cases_per_activity(log)
                                    

Detection of Batches

We say that an activity is executed in batches by a given resource when the resource executes several times the same activity in a short period of time.

Identifying such activities may identify points of the process that can be automated, since the activity of the person may be repetitive.

for act_res in batches:
    print("")
    print("activity: "+act_res[0][0]+" resource: "+act_res[0][1])
    print("number of distinct batches: "+str(act_res[1]))
    for batch_type in act_res[2]:
        print(batch_type, len(act_res[2][batch_type]))
                                    

There are indeed different types of batches that are detected by our method:

• Simultaneous: all the events in the batch have identical start and end timestamps.
• Batching at start: all the events in the batch have identical start timestamp.
• Batching at end: all the events in the batch have identical end timestamp.
• Sequential batching: for all the consecutive events, the end of the first is equal to the start of the second.
• Concurrent batching: for all the consecutive events that are not sequentially matched.

Playout of a Petri Net

A playout of a Petri net takes as input a Petri net along with an initial marking, and returns a list of process executions that are allowed from the process model.

We offer different types of playouts:

The list of parameters for such variants are:
Inspect parameters

from pm4py.algo.simulation.playout.petri_net import algorithm as simulator

simulated_log = simulator.apply(net, im, variant=simulator.Variants.BASIC_PLAYOUT, parameters={simulator.Variants.BASIC_PLAYOUT.value.Parameters.NO_TRACES: 50})
                                    

from pm4py.algo.simulation.playout.petri_net import algorithm as simulator

simulated_log = simulator.apply(net, im, variant=simulator.Variants.EXTENSIVE, parameters={simulator.Variants.EXTENSIVE.value.Parameters.MAX_TRACE_LENGTH: 7})
                                    

Monte Carlo Simulation

import os
from pm4py.objects.log.importer.xes import importer as xes_importer
log = xes_importer.apply(os.path.join("tests", "input_data", "running-example.xes"))
from pm4py.algo.discovery.dfg import algorithm as dfg_discovery
dfg_perf = dfg_discovery.apply(log, variant=dfg_discovery.Variants.PERFORMANCE)
from pm4py.statistics.start_activities.log import get as start_activities
from pm4py.statistics.end_activities.log import get as end_activities
sa = start_activities.get_start_activities(log)
ea = end_activities.get_end_activities(log)
                                    

from pm4py.statistics.traces.generic.log import case_arrival
ratio = case_arrival.get_case_arrival_avg(log)
print(ratio)
                                    

Using the DFG mining approach, it is possible to retrieve a Petri net model from the DFG. This kind of models is the “default” one for Monte Carlo simulation (because its execution semantics is very clear). Moreover, the Petri net extracted by the DFG mining approach is a sound workflow net (that gives other good properties to the model).

from pm4py.objects.conversion.dfg import converter
net, im, fm = converter.apply(dfg_perf, variant=converter.Variants.VERSION_TO_PETRI_NET_ACTIVITY_DEFINES_PLACE,
                              parameters={converter.Variants.VERSION_TO_PETRI_NET_ACTIVITY_DEFINES_PLACE.value.Parameters.START_ACTIVITIES: sa,
                                          converter.Variants.VERSION_TO_PETRI_NET_ACTIVITY_DEFINES_PLACE.value.Parameters.END_ACTIVITIES: ea})
                                    

from pm4py.algo.simulation.montecarlo import simulator as montecarlo_simulation
from pm4py.algo.conformance.tokenreplay.algorithm import Variants
parameters = {}
parameters[
    montecarlo_simulation.Variants.PETRI_SEMAPH_FIFO.value.Parameters.TOKEN_REPLAY_VARIANT] = Variants.BACKWARDS
parameters[montecarlo_simulation.Variants.PETRI_SEMAPH_FIFO.value.Parameters.PARAM_CASE_ARRIVAL_RATIO] = 10800
simulated_log, res = montecarlo_simulation.apply(log, net, im, fm, parameters=parameters)
                                    

During the replay operation, some debug messages are written to the screen. The main outputs of the simulation process are:

Among res, that is the result of the simulation, we have the following keys:
Inspect outputs

import random
last_timestamp = max(event["time:timestamp"] for trace in log for event in trace).timestamp()
first_timestamp = min(event["time:timestamp"] for trace in log for event in trace).timestamp()
pick_trans = random.choice(list(res["transitions_interval_trees"]))
print(pick_trans)
n_div = 10
i = 0
while i < n_div:
    timestamp = first_timestamp + (last_timestamp - first_timestamp)/n_div * i
    print("\t", timestamp, len(res["transitions_interval_trees"][pick_trans][timestamp]))
    i = i + 1
                                    

import random
last_timestamp = max(event["time:timestamp"] for trace in log for event in trace).timestamp()
first_timestamp = min(event["time:timestamp"] for trace in log for event in trace).timestamp()
pick_place = random.choice(list(res["places_interval_trees"]))
print(pick_place)
n_div = 10
i = 0
while i < n_div:
    timestamp = first_timestamp + (last_timestamp - first_timestamp)/n_div * i
    print("\t", timestamp, len(res["places_interval_trees"][pick_place][timestamp]))
    i = i + 1
                                    

The information can be used to build some graphs like these (using external programs such as Microsoft Excel).

The simulation process can be resumed as follows:

• An event log and a model (DFG) is considered.
• Internally in the simulation, a replay operation is done between the log and the model.
• The replay operation leads to the construction of a stochastic map that associates to each transition a probability distribution (for example, a normal distribution, an exponential distribution …). The probability distribution that maximizes the likelihood of the observed values during the replay is chosen. The user can force a specific transition (like exponential) if he wants.
• Moreover, during the replay operation, the frequency of each transition is found. That helps in picking in a “weighted” way one of the transitions enabled in a marking, when the simulation occurs.
• The simulation process occurs. For each one of the trace that are generated (the distance between the start of them is fixed) a thread is spawned, stochastic choices are made. The possibility to use a given place (depending on the maximum number of resources that is possible to use) is given by a semaphore object in Python.
• A maximum amount of time is specified for the simulation. If one or more threads exceed that amount of time, the threads are killed and the corresponding trace is not added to the simulation log.

Hence, several parameters are important in order to perform a Monte Carlo simulation. These parameters, that are inside the petri_semaph_fifo class, are (ordered by importance).
Inspect parameters

Extensive Playout of a Process Tree

An extensive playout operation permits to obtain (up to the provided limits) the entire language of the process model. Doing an extensive playout operation on a Petri net can be incredibly expensive (the reachability graph needs to be explored). Process trees, with their bottom-up structure, permit to obtain the entire language of an event log in a much easier way, starting from the language of the leafs (that is obvious) and then following specific merge rules for the operators.

However, since the language of a process tree can be incredibly vast (when parallel operators are involved) or also infinite (when loops are involved), the extensive playouts is possible up to some limits:

• A specification of the maximum number of occurrences for a loop must be done, if a loop is there. This stops an extensive playout operation at the given number of occurences.
• Since the number of different executions, when loops are involved, is still incredibly big, it is possible to specify the maximum length of a trace to be returned. So, traces that are above the maximum length are automatically discarded.
• For further limiting the number of different executions, the maximum number of traces returned by the algorithm might be provided.

Moreover, from the structure of the process tree, it is easy to infer the minimum length of a trace allowed by the process model (always following the bottom-up approach).

Some reasonable settings for the extensive playout are the following:

• Overall, the maximum number of traces returned by the algorithm is set to 100000.
• The maximum length of a trace that is an output of the playout is, by default, set to the minimum length of a trace accepted by a process tree.
• The maximum number of loops is set to be the minimum length of a trace divided by two.

The list of parameters are:
Inspect parameters

from pm4py.objects.log.importer.xes import importer as xes_importer
import os

log = xes_importer.apply(os.path.join("tests", "input_data", "receipt.xes"))
                                    

from pm4py.algo.discovery.inductive import algorithm as inductive_miner

tree = inductive_miner.apply_tree(log)
                                    

from pm4py.algo.simulation.playout.process_tree import algorithm as tree_playout

playout_variant = tree_playout.Variants.EXTENSIVE
param = tree_playout.Variants.EXTENSIVE.value.Parameters

simulated_log = tree_playout.apply(tree, variant=playout_variant,
                                   parameters={param.MAX_TRACE_LENGTH: 3, param.MAX_LIMIT_NUM_TRACES: 100000})
print(len(simulated_log))
                                    

At this point, the extensive playout operation is done.

Handover of Work

from pm4py.algo.organizational_mining.sna import algorithm as sna
hw_values = sna.apply(log, variant=sna.Variants.HANDOVER_LOG)
                                    

from pm4py.visualization.sna import visualizer as sna_visualizer
gviz_hw_py = sna_visualizer.apply(hw_values, variant=sna_visualizer.Variants.PYVIS)
sna_visualizer.view(gviz_hw_py, variant=sna_visualizer.Variants.PYVIS)
                                    

Subcontracting

from pm4py.algo.organizational_mining.sna import algorithm as sna
sub_values = sna.apply(log, variant=sna.Variants.SUBCONTRACTING_LOG)
                                    

from pm4py.visualization.sna import visualizer as sna_visualizer
gviz_sub_py = sna_visualizer.apply(sub_values, variant=sna_visualizer.Variants.PYVIS)
sna_visualizer.view(gviz_sub_py, variant=sna_visualizer.Variants.PYVIS)
                                    

Working Together

from pm4py.algo.organizational_mining.sna import algorithm as sna
wt_values = sna.apply(log, variant=sna.Variants.WORKING_TOGETHER_LOG)
                                    

from pm4py.visualization.sna import visualizer as sna_visualizer
gviz_wt_py = sna_visualizer.apply(wt_values, variant=sna_visualizer.Variants.PYVIS)
sna_visualizer.view(gviz_wt_py, variant=sna_visualizer.Variants.PYVIS)
                                    

Similar Activities

from pm4py.algo.organizational_mining.sna import algorithm as sna
ja_values = sna.apply(log, variant=sna.Variants.JOINTACTIVITIES_LOG)
                                    

from pm4py.visualization.sna import visualizer as sna_visualizer
gviz_ja_py = sna_visualizer.apply(ja_values, variant=sna_visualizer.Variants.PYVIS)
sna_visualizer.view(gviz_ja_py, variant=sna_visualizer.Variants.PYVIS)
                                    

Roles Discovery

A role is a set of activities in the log that are executed by a similar (multi)set of resources. Hence, it is a specific function into organization. Grouping the activities in roles can help:

An article on roles detection, that has inspired the technique implemented in PM4Py, is:

Burattin, Andrea, Alessandro Sperduti, and Marco Veluscek. “Business models enhancement through discovery of roles.” 2013 IEEE Symposium on Computational Intelligence and Data Mining (CIDM). IEEE, 2013.

• In understanding which activities are executed by which roles.
• By understanding roles itself (numerosity of resources for a single activity may not provide enough explanation)

Initially, each activity corresponds to a different role, and is associated to the multiset of his originators. After that, roles are merged according to their similarity, until no more merges are possible.

from pm4py.objects.log.importer.xes import importer as xes_importer
import os
log = xes_importer.apply(os.path.join("tests", "input_data", "receipt.xes"))
                                    

from pm4py.algo.organizational_mining.roles import algorithm as roles_discovery
roles = roles_discovery.apply(log)
                                    

We can print the sets of activities that are grouped in roles by doing print([x[0] for x in roles]).

Clustering (SNA results)

Given the results of applying a SNA metric, a clustering operation permits to group the resources that are connected by a meaningful connection in the given metric. For example:

• Clustering the results of the working together metric, individuals that work often together would be inserted in the same group.
• Clustering the results of the similar activities metric, individuals that work on the same tasks would be inserted in the same group.

We provide a baseline method to get a list of groups (where each group is a list of resources) from the specification of the values of a SNA metric. This can be applied as follows on the running-example log and the results of the similar activities metric:

import pm4py
import os
log = pm4py.read_xes(os.path.join("tests", "input_data", "running-example.xes"))

from pm4py.algo.organizational_mining.sna import algorithm as sna
sa_metric = sna.apply(log, variant=sna.Variants.JOINTACTIVITIES_LOG)

from pm4py.algo.organizational_mining.sna import util
clustering = util.cluster_affinity_propagation(sa_metric)

Resource Profiles

The profilation of resources from event logs is also possible. We implement the approach described in:

Pika, Anastasiia, et al. "Mining resource profiles from event logs." ACM Transactions on Management Information Systems (TMIS) 8.1 (2017): 1-30.

Basically, the behavior of a resource can be measured over a period of time with different metrics presented in the paper:

• RBI 1.1 (number of distinct activities): Number of distinct activities done by a resource in a given time interval [t1, t2)
• RBI 1.3 (activity frequency): Fraction of completions of a given activity a, by a given resource r, during a given time slot, [t1, t2), with respect to the total number of activity completions by resource r during [t1, t2)
• RBI 2.1 (activity completions): The number of activity instances completed by a given resource during a given time slot.
• RBI 2.2 (case completions): The number of cases completed during a given time slot in which a given resource was involved.
• RBI 2.3 (fraction case completion): The fraction of cases completed during a given time slot in which a given resource was involved with respect to the total number of cases completed during the time slot.
• RBI 2.4 (average workload): The average number of activities started by a given resource but not completed at a moment in time.
• RBI 3.1 (multitasking): The fraction of active time during which a given resource is involved in more than one activity with respect to the resource's active time.
• RBI 4.3 (average duration activity): The average duration of instances of a given activity completed during a given time slot by a given resource.
• RBI 4.4 (average case duration): The average duration of cases completed during a given time slot in which a given resource was involved.
• RBI 5.1 (interaction two resources): The number of cases completed during a given time slot in which two given resources were involved.
• RBI 5.2 (social position): The fraction of resources involved in the same cases with a given resource during a given time slot with respect to the total number of resources active during the time slot.

The following example calculates these metrics starting from the running-example XES event log:

import os
from pm4py.algo.organizational_mining.resource_profiles import algorithm
import pm4py
log = pm4py.read_xes(os.path.join("tests", "input_data", "running-example.xes"))
# Metric RBI 1.1: Number of distinct activities done by a resource in a given time interval [t1, t2)
print(algorithm.distinct_activities(log, "2010-12-30 00:00:00", "2011-01-25 00:00:00", "Sara"))
# Metric RBI 1.3: Fraction of completions of a given activity a, by a given resource r,
# during a given time slot, [t1, t2), with respect to the total number of activity completions by resource r
# during [t1, t2)
print(algorithm.activity_frequency(log, "2010-12-30 00:00:00", "2011-01-25 00:00:00", "Sara", "decide"))
# Metric RBI 2.1: The number of activity instances completed by a given resource during a given time slot.
print(algorithm.activity_completions(log, "2010-12-30 00:00:00", "2011-01-25 00:00:00", "Sara"))
# Metric RBI 2.2: The number of cases completed during a given time slot in which a given resource was involved.
print(algorithm.case_completions(log, "2010-12-30 00:00:00", "2011-01-25 00:00:00", "Pete"))
# Metric RBI 2.3: The fraction of cases completed during a given time slot in which a given resource was involved
# with respect to the total number of cases completed during the time slot.
print(algorithm.fraction_case_completions(log, "2010-12-30 00:00:00", "2011-01-25 00:00:00", "Pete"))
# Metric RBI 2.4: The average number of activities started by a given resource but not completed at a moment in time.
print(algorithm.average_workload(log, "2010-12-30 00:00:00", "2011-01-15 00:00:00", "Mike"))
# Metric RBI 3.1: The fraction of active time during which a given resource is involved in more than one activity
# with respect to the resource's active time.
print(algorithm.multitasking(log, "2010-12-30 00:00:00", "2011-01-25 00:00:00", "Mike"))
# Metric RBI 4.3: The average duration of instances of a given activity completed during a given time slot by
# a given resource.
print(algorithm.average_duration_activity(log, "2010-12-30 00:00:00", "2011-01-25 00:00:00", "Sue", "examine thoroughly"))
# Metric RBI 4.4: The average duration of cases completed during a given time slot in which a given resource was involved.
print(algorithm.average_case_duration(log, "2010-12-30 00:00:00", "2011-01-25 00:00:00", "Sue"))
# Metric RBI 5.1: The number of cases completed during a given time slot in which two given resources were involved.
print(algorithm.interaction_two_resources(log, "2010-12-30 00:00:00", "2011-01-25 00:00:00", "Mike", "Pete"))
# Metric RBI 5.2: The fraction of resources involved in the same cases with a given resource during a given time slot
# with respect to the total number of resources active during the time slot.
print(algorithm.social_position(log, "2010-12-30 00:00:00", "2011-01-25 00:00:00", "Sue"))
                                    

Organizational Mining

With event logs, we are able to identify groups of resources doing similar activities. As we have seen in the previous sections, we have different ways to detect automatically these groups from event logs:

• Discovering the Similar Activities metric and applying a clustering algorithm to find the groups.
• Applying the roles discovery algorithm (Burattin et al.)

As a third option, an attribute might be there in the events, describing the group that performed the event.

With the term organizational mining, we mean the discovery of behavior-related information specific to an organizational group (e.g. which activities are done by the group?).

We provide an implementation of the approach described in:

Yang, Jing, et al. 'OrgMining 2.0: A Novel Framework for Organizational Model Mining from Event Logs.' arXiv preprint arXiv:2011.12445 (2020).

The approach provides the description of some group-related metrics (local diagnostics). Among these, we have:

• Group Relative Focus: (on a given type of work) specifies how much a resource group performed this type of work compared to the overall workload of the group. It can be used to measure how the workload of a resource group is distributed over different types of work, i.e., work diversification of the group.
• Group Relative Stake: (in a given type of work) specifies how much this type of work was performed by a certain resource group among all groups. It can be used to measure how the workload devoted to a certain type of work is distributed over resource groups in an organizational model, i.e., work participation by different groups.
• Group Coverage: with respect to a given type of work specifies the proportion of members of a resource group that performed this type of work.
• Group Member Contribution: of a member of a resource group with respect to the given type of work specifies how much of this type of work by the group was performed by the member. It can be used to measure how the workload of the entire group devoted to a certain type of work is distributed over the group members.

The following example calculates these metrics starting from the receipt XES event log, and how the information can be exploited, from an attribute that specifies which is the group doing the task:

import pm4py
import os
from pm4py.algo.organizational_mining.local_diagnostics import algorithm as local_diagnostics
log = pm4py.read_xes(os.path.join("tests", "input_data", "receipt.xes"))
# this applies the organizational mining from an attribute that is in each event, describing the group that is performing the task.
ld = local_diagnostics.apply_from_group_attribute(log, parameters={local_diagnostics.Parameters.GROUP_KEY: "org:group"})
# GROUP RELATIVE FOCUS (on a given type of work) specifies how much a resource group performed this type of work
# compared to the overall workload of the group. It can be used to measure how the workload of a resource group
# is distributed over different types of work, i.e., work diversification of the group.
print("\ngroup_relative_focus")
print(ld["group_relative_focus"])
# GROUP RELATIVE STAKE (in a given type of work) specifies how much this type of work was performed by a certain
# resource group among all groups. It can be used to measure how the workload devoted to a certain type of work is
# distributed over resource groups in an organizational model, i.e., work participation by different groups.
print("\ngroup_relative_stake")
print(ld["group_relative_stake"])
# GROUP COVERAGE with respect to a given type of work specifies the proportion of members of a resource group that
# performed this type of work.
print("\ngroup_coverage")
print(ld["group_coverage"])
# GROUP MEMBER CONTRIBUTION of a member of a resource group with respect to the given type of work specifies how
# much of this type of work by the group was performed by the member. It can be used to measure how the workload
# of the entire group devoted to a certain type of work is distributed over the group members.
print("\ngroup_member_contribution")
print(ld["group_member_contribution"])
                    

Alternatively, the apply_from_clustering_or_roles method of the same class can be used, providing the log as first argument, and the results of the clustering as second argument.

BPMN 2.0 – Importing

import pm4py
import os

bpmn_graph = pm4py.read_bpmn(os.path.join("tests", "input_data", "running-example.bpmn"))
                                    

BPMN 2.0 – Exporting

import pm4py
import os

pm4py.write_bpmn(bpmn_graph, "ru.bpmn", enable_layout=True)
                                    

The enable_layout, set to True, enables the automatic layouting of the BPMN model before the export.

BPMN 2.0 – Layouting

A layouting operation tries to give a good position to the nodes and the edges of the BPMN diagram. For our purposes, we chose an octilinear edges layout.

from pm4py.objects.bpmn.layout import layouter

bpmn_graph = layouter.apply(bpmn_graph)
                                    

BPMN 2.0 – Conversion to Petri net

A conversion of a BPMN model into a Petri net model enables different PM4Py algorithms (such as conformance checking and simulation algorithms), hence is a particularly important operation.

from pm4py.objects.conversion.bpmn import converter as bpmn_converter

net, im, fm = bpmn_converter.apply(bpmn_graph)
                                    

BPMN 2.0 – Conversion from a process tree

Process trees are important classes of block-structured processes (and the output of the inductive miner algorithm). These models can be easily converted to BPMN models.

import pm4py
import os

log = pm4py.read_xes(os.path.join("tests", "input_data", "running-example.xes"))
tree = pm4py.discover_process_tree_inductive(log)
                                    

from pm4py.objects.conversion.process_tree import converter

bpmn_graph = converter.apply(tree, variant=converter.Variants.TO_BPMN)
                                    

Filtering activities/paths

Directly-follows graphs can contain a huge number of activities and paths, with some of them being outliers. In this section, we will see how to filter on the activities and paths of the graph, keeping a subset of its behavior.

import pm4py
log = pm4py.read_xes("tests/input_data/running-example.xes")
dfg, sa, ea = pm4py.discover_directly_follows_graph(log)
activities_count = pm4py.get_attribute_values(log, "concept:name")
                                    

from pm4py.algo.filtering.dfg import dfg_filtering
dfg, sa, ea, activities_count = dfg_filtering.filter_dfg_on_activities_percentage(dfg, sa, ea, activities_count, 0.2)
                                    

from pm4py.algo.filtering.dfg import dfg_filtering
dfg, sa, ea, activities_count = dfg_filtering.filter_dfg_on_paths_percentage(dfg, sa, ea, activities_count, 0.2)
                                    

Playout of a DFG

A playout operation on a directly-follows graph is useful to retrieve the traces that are allowed from the directly-follows graph. In this case, a trace is a set of activities visited in the DFG from the start node to the end node. We can assign a probability to each trace (assuming that the DFG represents a Markov chain). In particular, we are interested in getting the most likely traces. In this section, we will see how to perform the playout of a directly-follows graph.

import pm4py
log = pm4py.read_xes("tests/input_data/running-example.xes")
dfg, sa, ea = pm4py.discover_directly_follows_graph(log)
activities_count = pm4py.get_attribute_values(log, "concept:name")
                                    

from pm4py.algo.simulation.playout.dfg import algorithm as dfg_playout
parameters = {}
simulated_log = dfg_playout.apply(dfg, sa, ea, parameters=parameters)
                                    

The playout operation can be customized using some parameters.

Alignments on a DFG

A popular conformance checking technique is the one of alignments. Alignments are usually performed on Petri nets; however, this could take time, since the state space of Petri nets can be huge. It is also possible to perform alignments on a directly-follows graph. Since the state space of a directly-follows graph is small, the result is a very efficient computation of alignments. This permits to get quick diagnostics on the activities and paths that are executed in a wrong way. In this section, we will show an example on how to perform alignments between a process execution and a DFG.

import pm4py
log = pm4py.read_xes("tests/input_data/running-example.xes")
dfg, sa, ea = pm4py.discover_directly_follows_graph(log)
activities_count = pm4py.get_attribute_values(log, "concept:name")
                                    

from pm4py.algo.conformance.alignments.dfg import algorithm as dfg_alignment
alignments = dfg_alignment.apply(simulated_log, dfg, sa, ea)
                                    

The output of the alignments is equivalent to the one obtained against Petri nets. In particular, the output is a list containing for each trace the result of the alignment. Each alignment consists in some moves from the start to the end of both the trace and the DFG. We can have sync moves, moves on log (whether a move in the process execution is not mimicked by the DFG) and moves on model (whether a move is needed in the model that is not supported by the process execution).

Convert Directly-Follows Graph to a Workflow Net

The Directly-Follows Graph is the representation of a process provided by many commercial tools. An idea of Sander Leemans is about converting the DFG into a workflow net that perfectly mimic the DFG. This is called DFG mining. The following steps are useful to load the log, calculate the DFG, convert it into a workflow net and perform alignments.

from pm4py.objects.log.importer.xes import importer as xes_importer
import os
log = xes_importer.apply(os.path.join("tests", "input_data", "running-example.xes"))

from pm4py.algo.discovery.dfg import algorithm as dfg_discovery
dfg = dfg_discovery.apply(log)

from pm4py.objects.conversion.dfg import converter as dfg_mining
net, im, fm = dfg_mining.apply(dfg)

CTMC Simulation (DFG)

A time-related simulation permits to know how probable is that a process execution is terminated after a given amount of time. This leads to a better estimation of Service Level Agreements, or a better identification of the process instances that are most likely to have an high throughput time.

import os
from pm4py.objects.log.importer.xes import importer as xes_importer
log = xes_importer.apply(os.path.join("tests", "input_data", "running-example.xes"))
from pm4py.algo.discovery.dfg import algorithm as dfg_discovery
dfg_perf = dfg_discovery.apply(log, variant=dfg_discovery.Variants.PERFORMANCE)
from pm4py.statistics.start_activities.log import get as start_activities
from pm4py.statistics.end_activities.log import get as end_activities
sa = start_activities.get_start_activities(log)
ea = end_activities.get_end_activities(log)
                                    

For the simulation model, a CTMC (Continuous Time Markov Chain) is built from the DFG. This model is very powerful, assuming that the frequency of the edges outgoing an activity is similar. If that is not the case (for example, when an outgoing edge has frequency 1 and the other 10000) the model works less well.

from pm4py.algo.filtering.log.variants import variants_filter
log = variants_filter.apply_auto_filter(log)
                                    

Given that the edge contains the average of time between the states, it is assumed by the CTMC that the distribution of times follows an exponential distribution with the given average.

from pm4py.objects.stochastic_petri import ctmc
reach_graph, tang_reach_graph, stochastic_map, q_matrix = ctmc.get_tangible_reachability_and_q_matrix_from_dfg_performance(dfg_perf, parameters={"start_activities": sa, "end_activities": ea})
print(tang_reach_graph.states)
                                    

The last line prints the states of the model, that are:

{reinitiaterequest1, sink1, source1, paycompensation1, registerrequest1, decide1, rejectrequest1, examinethoroughly1, checkticket1, examinecasually1}

“source1” is the source state of the model (that is implicitly connected to the start activity “register request”). “sink1” is the terminal state of the model (that is implicitly connected to the end activities “pay compensation” and “reject request”). The other states of the model are the ones in which you go after executing the corresponding activity (for example, “decide1” is the state in which you sink after a “Decide” activity).

# pick the source state
state = [x for x in tang_reach_graph.states if x.name == "source1"][0]
# analyse the distribution over the states of the system starting from the source after 172800.0 seconds (2 days)
transient_result = ctmc.transient_analysis_from_tangible_q_matrix_and_single_state(tang_reach_graph, q_matrix, state,
                                                                                 172800.0)
                                    

We get the following dictionary as output:

Counter({examinethoroughly1: 0.2962452842059696, sink1: 0.22722259091795302, decide1: 0.21581166958939804, checkticket1: 0.14875795862276098, examinecasually1: 0.10303611911547725, reinitiaterequest1: 0.008919293375341046, registerrequest1: 7.082354169889813e-06, rejectrequest1: 1.399193916663259e-09, paycompensation1: 4.1973640443858166e-10, source1: 9.448511222571739e-23})

That means that we have the 22,72% of probability to have already finished the process execution (being in the “sink” state) after 2 days. Let’s calculate that for 100 days (8640000 seconds):

Counter({sink1: 0.9999999999459624, examinethoroughly1: 2.439939596108015e-11, decide1: 1.773740724895241e-11, checkticket1: 7.016711086804287e-12, examinecasually1: 4.146293819445992e-12, reinitiaterequest1: 7.377366868580615e-13, rejectrequest1: 1.1499875063552587e-19, paycompensation1: 3.449783588381732e-20, registerrequest1: 3.227610398476988e-258, source1: 1.3702507295618392e-274})

According to the model, we have probability 99,999999995 % to have finished the process after 100 days! That is practically 100%.

state = [x for x in tang_reach_graph.states if x.name == "decide1"][0]
transient_result = ctmc.transient_analysis_from_tangible_q_matrix_and_single_state(tang_reach_graph, q_matrix, state,
                                                                                   864000.0)
print(transient_result)
                                    

The result is:

Counter({sink1: 0.9293417034466963, examinethoroughly1: 0.03190639167001194, decide1: 0.02319465799554023, checkticket1: 0.009172805258533999, examinecasually1: 0.005419723629171662, reinitiaterequest1: 0.0009647178045538061, rejectrequest1: 1.5038030381048534e-10, paycompensation1: 4.511175132201897e-11, registerrequest1: 0.0, source1: 0.0})

so we have probability 92,9% to “sink” in 10 days after a decision.

Streaming Package General Structure

In PM4Py, we offer support for streaming process mining functionalities, including:

• Streaming process discovery (DFG)
• Streaming conformance checking (footprints and TBR)
• Streaming importing of XES/CSV files

• register(algo): registers a new algorithm to the live event stream (that will be notified when an event is added to the stream.
• append(event): adds an event to the live event stream.

The LiveEventStream processes the incoming events using a thread pool. This helps to manage a “flood” of events using a given number of different threads.

For the streaming algorithms, that are registered to the LiveEventStream, we provide an interface that should be implemented. The following methods should be implemented inside each streaming algorithm:

• _process(event): a method that accepts and process an incoming event.
• _current_result(): a method that returns the current state of the streaming algorithm.

Streaming Process Discovery (Directly-Follows Graph)

The following example will show how to discover a DFG from a stream of events.

from pm4py.streaming.stream.live_event_stream import LiveEventStream

live_event_stream = LiveEventStream()
                                    

from pm4py.streaming.algo.discovery.dfg import algorithm as dfg_discovery

streaming_dfg = dfg_discovery.apply()
                                    

live_event_stream.register(streaming_dfg)
                                    

live_event_stream.start()
                                    

import os
from pm4py.objects.log.importer.xes import importer as xes_importer

log = xes_importer.apply(os.path.join("tests", "input_data", "running-example.xes"))
                                    

from pm4py.objects.conversion.log import converter as stream_converter

static_event_stream = stream_converter.apply(log, variant=stream_converter.Variants.TO_EVENT_STREAM)
                                    

for ev in static_event_stream:
    live_event_stream.append(ev)
                                    

live_event_stream.stop()
                                    

dfg, activities, sa, ea = streaming_dfg.get()
                                    

If we do print(dfg) on the running-example.xes log we obtain:

{(‘register request’, ‘examine casually’): 3, (‘examine casually’, ‘check ticket’): 4, (‘check ticket’, ‘decide’): 6, (‘decide’, ‘reinitiate request’): 3, (‘reinitiate request’, ‘examine thoroughly’): 1, (‘examine thoroughly’, ‘check ticket’): 2, (‘decide’, ‘pay compensation’): 3, (‘register request’, ‘check ticket’): 2, (‘check ticket’, ‘examine casually’): 2, (‘examine casually’, ‘decide’): 2, (‘register request’, ‘examine thoroughly’): 1, (‘decide’, ‘reject request’): 3, (‘reinitiate request’, ‘check ticket’): 1, (‘reinitiate request’, ‘examine casually’): 1, (‘check ticket’, ‘examine thoroughly’): 1, (‘examine thoroughly’, ‘decide’): 1}

Streaming Conformance Checking (TBR)

The following examples will show how to check conformance against a stream of events with the footprints and token-based replay algorithms. For both the examples that follow, we assume to work with the running-example.xes log and with a log discovered using inductive miner infrequent with the default noise threshold (0.2).

import os
from pm4py.objects.log.importer.xes import importer as xes_importer
log = xes_importer.apply(os.path.join("tests", "input_data", "receipt.xes"))
                                    

from pm4py.objects.conversion.log import converter as log_converter
static_event_stream = log_converter.apply(log, variant=log_converter.Variants.TO_EVENT_STREAM)
                                    

from pm4py.algo.discovery.inductive import algorithm as inductive_miner
tree = inductive_miner.apply_tree(log, variant=inductive_miner.Variants.IMf)
                                    

from pm4py.objects.conversion.process_tree import converter as pt_converter
net, im, fm = pt_converter.apply(tree)
                                    

Now, we can apply the streaming TBR.

from pm4py.streaming.stream.live_event_stream import LiveEventStream
live_event_stream = LiveEventStream()
                                    

from pm4py.streaming.algo.conformance.tbr import algorithm as tbr_algorithm
streaming_tbr = tbr_algorithm.apply(net, im, fm)
                                    

live_event_stream.register(streaming_tbr)
                                    

live_event_stream.start()
                                    

for ev in static_event_stream:
    live_event_stream.append(ev)
                                    

live_event_stream.stop()
                                    

conf_stats = streaming_tbr.get()
print(conf_stats)
                                    

In addition to this, the following methods are available inside the streaming TBR that print some warning during the replay. The methods can be overriden easily (for example, to send the message with mail):

• message_case_or_activity_not_in_event
• message_activity_not_possible
• message_missing_tokens
• message_case_not_in_dictionary
• message_final_marking_not_reached

Streaming Conformance Checking (footprints)

Footprints is another conformance checking method offered in PM4Py, which can be implemented in the context of streaming events. In the following, we see an application of the streaming footprints.

from pm4py.algo.discovery.footprints import algorithm as fp_discovery
footprints = fp_discovery.apply(tree)
                                    

from pm4py.streaming.stream.live_event_stream import LiveEventStream
live_event_stream = LiveEventStream()
                                    

from pm4py.streaming.algo.conformance.footprints import algorithm as fp_conformance
streaming_footprints = fp_conformance.apply(footprints)
                                    

live_event_stream.register(streaming_footprints)
                                    

live_event_stream.start()
                                    

for ev in static_event_stream:
    live_event_stream.append(ev)
                                    

live_event_stream.stop()
                                    

conf_stats = streaming_footprints.get()
print(conf_stats)
                                    

In addition to this, the following methods are available inside the streaming footprints that print some warning during the replay. The methods can be overriden easily (for example, to send the message with mail):

• message_case_or_activity_not_in_event
• message_activity_not_possible
• message_footprints_not_possible
• message_start_activity_not_possible
• message_end_activity_not_possible
• message_case_not_in_dictionary

Streaming Conformance Checking (Temporal Profile)

We propose in PM4Py an implementation of the temporal profile model. This has been described in:

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 couple of activities in the log the average time and the standard deviation between events having the provided activities. The time is measured between the completion of the first event and the start of the second event. Hence, it is assumed to work with an interval log where the events have two timestamps. The output of the temporal profile discovery is a dictionary where each couple of activities (expressed as a tuple) is associated to a couple of numbers, the first is the average and the second is the average standard deviation.

It is possible to use a temporal profile to perform conformance checking on an event log. The times between the couple of activities in the log are assessed against the numbers stored in the temporal profile. Specifically, a value is calculated that shows how many standard deviations the value is different from the average. If that value exceeds a threshold (by default set to 6, according to the six-sigma principles), then the couple of activities is signaled.

In PM4Py, we provide a streaming conformance checking algorithm based on the temporal profile. The algorithm checks an incoming event against every event that happened previously in the case, identifying deviations according to the temporal profile. This section provides an example where a temporal profile is discovered, the streaming conformance checking is set-up and actually a log is replayed on the stream.

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

log = pm4py.read_xes("tests/input_data/running-example.xes")
parameters = {}
temporal_profile = temporal_profile_discovery.apply(log, parameters=parameters)
                                

from pm4py.streaming.stream.live_event_stream import LiveEventStream
from pm4py.streaming.algo.conformance.temporal import algorithm as temporal_conformance_checker

stream = LiveEventStream()
parameters = {}
temp_cc = temporal_conformance_checker.apply(temporal_profile, parameters={})
stream.register(temp_cc)
stream.start()
                                

static_stream = pm4py.convert_to_event_stream(log)
for event in static_stream:
    stream.append(event)
                                

During the execution of the streaming temporal profile conformance checker, some warnings are printed if a couple of events violate the temporal profile. Moreover, it is also possible to get a dictionary containing the cases with deviations associated with all their deviations.

stream.stop()
res = temp_cc.get()
                                

Streaming Importer (XES trace-by-trace)

In order to be able to process the traces of a XES event log that might not fit in the memory, or when a sample of a big log is needed, the usage of the XES trace-by-trace streaming importer helps to cope with the situation.

from pm4py.streaming.importer.xes import importer as xes_importer

streaming_ev_object = xes_importer.apply(os.path.join("tests", "input_data", "running-example.xes"), variant=xes_importer.Variants.XES_EVENT_STREAM)
                                    

for trace in streaming_log_object:
    print(trace)
                                    

Streaming Importer (XES event-by-event)

In order to be able to process the events of a XES event log that might not fit in the memory, or when the sample of a big log is needed, the usage of the XES event-by-event streaming importer helps to cope with the situation. In this case, the single events inside the traces are picked during the iteration.

from pm4py.streaming.importer.xes import importer as xes_importer

streaming_ev_object = xes_importer.apply(os.path.join("tests", "input_data", "running-example.xes"), variant=xes_importer.Variants.XES_EVENT_STREAM)
                                    

for event in streaming_ev_object:
    print(event)
                                    

Streaming Importer (CSV event-by-event)

In order to be able to process the events of a CSV event log that might not fit in the memory, or when the sample of a big log is needed, Pandas might not be feasible. In this case, the single rows of the CSV file are parsed during the iteration.

from pm4py.streaming.importer.csv import importer as csv_importer
log_object = csv_importer.apply(os.path.join("tests", "input_data", "running-example.csv"))
                                    

for ev in log_object:
    print(ev)