IEEE XES is a standard format describing how event logs are stored. For more information about the format, please study the IEEE XES Website. A simple synthetic event log can be downloaded from here. Note that several real event logs have been made available, over the past few years. You can find them here.

The example code on the right shows how to import an event log, stored in the IEEE XES format, given a file path to the log file. The code fragment uses the standard importer (iterparse, described in a later paragraph). Note that IEEE XES Event Logs are imported into an Event Log object, i.e., as described earlier.

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

Event logs are stored as an extension of the Python list data structure. To access a trace in the log, it is enough to provide its index in the event log. Consider the example on the right on how to access the different objects stored in the imported log.

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

The apply() method of the xes_importer, i.e. located in pm4py.objects.log.importer.xes.importer.py, contains two optional parameters: variant and parameters. The variant parameter indicates which variant of the importer to use. The parameters parameter is a Python dictionary, specifying specific parameters of choice.

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)

This method invocation style is used throughout PM4Py in the various different algorithms implemented, i.e., by wrapping around the different implementations, new variants of algorithms are easily called, using previously written PM4Py code. W.r.t. XES importers, two variants are provided. One implementation is based on the iterparse() function of xml.etree. The other variant is a line-by-line, custom parser (for improved performance). It does not follow the standard and is able to import traces, simple trace attributes, events, and simple event attributes. To specify a variant, we add the following argument to the call to the importer: variant=xes_importer.Variants.ITERPARSE (note that, in the example code, this is encapsulated in local varaible variant). The xes_importer.Variants.ITERPARSE-value, actually maps on the underlying Python module, implementing the iterparse-based importer. We are able to access that reference, by accessing the value property, e.g., xes_importer.Variants.ITERPARSE.value. That module, contains a parameter definition, i.e., Parameters, containing all possible parameters for the iterparse-variant. As an example, parameter TIMESTAMP_SORT is one of those, accessed by xes_importer.Variants.ITERPARSE.value.Parameters.TIMESTAMP_SORT. Click the button below, to reveal all variants and corresponding parameters defined for importing IEEE XES files.

Apart from the IEEE XES standard, a lot of event logs are actually stored in a CSV file. In general, there is two ways to deal with CSV files in PM4Py:

• Import the CSV into a pandas DataFrame; In general, most existing algorithms in PM4Py are coded to be flexible in terms of their input, i.e., if a certain event log object is provided that is not in the right form, we translate it to the appropriate form for you. Hence, after importing a dataframe, most algorithms are directly able to work with the data frame.
• Convert the CSV into an event log object (similar to the result of the IEEE XES importer presented in the previous section); In this case, the first step is to import the CSV file using pandas (similar to the previous bullet) and subsequently converting it to the event log object. In the remainder of this section, we briefly highlight how to convert a pandas DataFrame to an event log. Note that most algorithms use the same type of conversion, in case a given event data object is not of the right type.

To convert objects in PM4Py, there is a dedicated package, i.e., objects.conversion. The conversion package allows one to convert an object of a certain type to a new object of a different type (if such a conversion is applicable). Within the conversion package, a standard naming convention is applied, i.e., the type of the input object defines the package in which the code resides. Thus, since we assume that the imported DataFrame represents an event log, we find the appropriate conversion in the objects.conversion.log package.

The example code on the right shows how to convert a CSV file into the PM4Py internal event data object types. By default, the converter converts the dataframe to an Event Log object (i.e., not an Event Stream).

Actually, we suggest to sort the dataframe by its timestamp column. In the example on the right, it is assumed that the timestamp column is timestamp. This ensures that events are sorted by their timestamp.

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)

In this small example table, we observe four columns, i.e., case, activity, timestamp and clientID. Clearly, when importing the data and converting it to an Event Log object, we aim to combine all rows (events) with the same value for the case column together. Hence, the default value of the CASE_ID_KEY parameter is not set to the right value. Another interesting phenomenon in the example data is the fourth column, i.e., clientID. In fact, the client ID is an attribute that will not change over the course of execution a process instance, i.e., it is a case-level attribute. PM4Py allows us to specify that a column actually describes a case-level attribute (under the assumption that the attribute does not change during the execution of a process). However, for this, we need to specify an additional parameter, i.e., the CASE_ATTRIBUTE_PREFIX parameter, with default value 'case:'.

The example code on the right shows how to convert the previously examplified csv data file. After loading the csv file of the example table, we rename the clientID column to case:clientID (this is a specific operation provided by pandas!). Then, we specify that the column identifying the case identifier attribute is the column with name 'case'. Note that the full parameter path is log_converter.Variants.TO_EVENT_LOG.value.Parameters.CASE_ID_KEY

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)

In case we would like to use a different prefix for the case-level attributes, e.g., 'caseAttr', we can do so by mapping the CASE_ATTRIBUTE_PREFIX (full path: log_converter.Variants.TO_EVENT_LOG.value.Parameters.CASE_ATTRIBUTE_PREFIX) to the value 'caseAttr'. Note that in the call to the converter, in this case, we explicitly set the variant to be used, e.g., log_converter.Variants.TO_EVENT_LOG. Finally, note that any type of data format that can be parsed to a Pandas DataFrame, is supported by PM4Py.

In this section, we describe how to convert event log objects from one object type to another object type. As mentioned in the previous section, the conversion functionality of event logs is located in pm4py.objects.conversion.log.converter. There are three objects, which we are able to 'switch' between, i.e., Event Log, Event Stream and Data Frame objects. Please refer to the previous code snippet for an example of applying log conversion (applied when importing a CSV object). Finally, note that most algorithms internally use the converters, in order to be able to handle an input event data object of any form. In such a case, the default parameters are used.

Exporting an Event Log object to an IEEE Xes file is fairly straightforward in PM4Py. Consider the example code fragment on the right, which depicts this functionality.

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

In the example, the log object is assumed to be an Event Log object. The exporter also accepts an Event Stream or DataFrame object as an input. However, the exporter will first convert the given input object into an Event Log. Hence, in this case, standard parameters for the conversion are used. Thus, if the user wants more control, it is advisable to apply the conversion to Event Log, prior to exporting.

To export an event log to a csv-file, PM4Py uses Pandas. Hence, an event log is first converted to a Pandas Data Frame, after which it is written to disk.

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>')
                                

In case an event log object is provided that is not a dataframe, i.e., an Event Log or Event Stream, the conversion is applied, using the default parameter values, i.e., as presented in the Converting Event Data section. Note that exporting event data to as csv file has no parameters. In case more control over the conversion is needed, please apply a conversion to dataframe first, prior to exporting to csv.

Since Event Logs and Event Streams are iterables (note: this does not apply for dataframes), they are applicable to be used in combination with lambda functions. However, as they contain more information (such as log-level attributes), directly appying, e.g., a filter, does not work. Therefore, a utility package is available that wraps around filtering/maps/sorting in order to combine this functionality with Event Logs. The code is located in pm4py.objects.log.util.func

Consider the code fragment on the right, which first imports an event log and then filters out each trace with a length shorter than three. The func.filter_ function mimics the built-in Python function filter(). However, it returns the filtered list of traces, included in an Event Log (or Event Stream) object.

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

log = xes_importer.apply('<path_to_imported_log.xes>')
log = func.filter_(lambda t: len(t) > 2, log)

Apart from the filter_-function, the pm4py.objects.log.util.func package provides a map_ and a sort_ function.

One might be interested in only keeping the traces that are contained in a specific interval, e.g. 09 March 2011 and 18 January 2012. The first code snippet works for a log object, the second one for a dataframe object.

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"})

However, it is also possible to keep the traces that are intersecting with a time interval. The first example is again for log objects, the second one for dataframe objects.

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"})

Until now, only trace based techniques have been discussed. However, there is a method to keep the events that are contained in specific timeframe. As previously mentioned, the first code snippet provides information about how to apply this technique on log objects, whereby the second snippets provides information about how to apply this on dataframe objects.

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"})

This filter permits to keep only traces with duration that is inside a specified interval. In the examples, traces between 1 and 10 days are kept. Note that the time parameters are given in seconds. The first code snippet applies this technique on log object, the second one on a dataframe object.

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"})

First of all, it might be necessary to know the starting activities. Therefore, code snippets are provided. Subsequently, an example of filtering is provided. The first snippet is working with log object, the second one is working on a dataframe. log_start is a dictionary that contains as key the activity and as value the number of occurrence.

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
						

As mentioned earlier, there is also a method that aims to keep the frequent start activities. Again, the first snippet is about a log object, the second is about a dataframe object. The default value for DECREASING_FACTOR is 0.6.

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})

This filter permits to keep only traces with an end activity among a set of specified activities. First of all, it might be necessary to know the end activities. Therefore, code snippets are provided. Subsequently, an example of filtering is provided. Here, for the dataframe filtering, a further attribute specification is possible: case:concept:name is in this case the column of the dataframe that is the Case ID, concept:name is the column of the dataframe that is the activity.

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_acitivites = 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"})

To get the list of variants contained in a given log, the following code could be used. The first code is for an log object, the second for a dataframe. The result is expressed as a dictionary having as key the variant and as value the list of cases that share the variant.

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

from pm4py.statistics.traces.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"})

If the number of occurrences of the variants is of interest, the following code retrieves a list of variants along with their count (so, a dictionary which key is the variant and the value is the number of occurrences).

from pm4py.statistics.traces.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.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)

To filter based on variants, assume that variants is a list, whereby each element is a variant (expressed in an equal way as in the variants retrieval method). The first method can be applied on log objects, the second can be applied on dataframe objects. Note that the variants given in variants are kept.

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"})

Contrary to the previous example, suppose you want to filter the given variants out. Again, let variants be a list, whereby each element is a variant.

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"})

A filter to keep automatically the most common variants could be applied through the apply_auto_filter method. This method accepts a parameter called DECREASING_FACTOR (default value is 0.6; further details are provided in the start activities filter).

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)
                            

To get the list of resources and activities contained in the log, the following code could be used.

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

To filter traces containing/not containing a given list of resources, the following code could be used.

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})

To apply automatically a filter on events attributes (trimming traces and keeping only events containing the attribute with a frequent value), the apply_auto_filter method is provided. The method accepts as parameters the attribute name and the DECREASING_FACTOR (default 0.6; an explanation could be found on the start activities filter).

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})

First, we import, the log. Subsequently, we want to keep only the events satisfying an amount comprised between 34 and 36. An additional filter aims to to keep only cases with at least an event satisfying the specified amount. The filter on cases provide the option to specify up to two attributes that are checked on the events that shall satisfy the numeric range. For example, if we are interested in cases having an event with activity Add penalty that has an amount between 34 and 500, a code snippet is also provided.

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"})

The former method can also be applied on dataframes.

import os
from pm4py.objects.log.adapters.pandas import csv_import_adapter
df = csv_import_adapter.import_dataframe_from_path(os.path.join("tests", "input_data", "roadtraffic100traces.csv"))

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"})

First, the log has to be imported.

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

Subsequently, the Alpha Miner is applied.

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

To mine a Petri Net, we provide an example. A log is read, the inductive miner 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. First, the log is read, then the inductive miner algorithm is applied.

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)

To obtain a process tree, the provided code snippet can be used. The last two lines of code are responsible for the visualization of the process tree.

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)

It is also possible to convert a process tree into a petri net.

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)

To apply the Heuristics Miner to discover an Heuristics Net, it is necessary to import a log. Then, a Heuristic Net can be found. There are also numerous possible parameters that can be inspected by clicking on the following button.

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})

To visualize the Heuristic Net, code is also provided on the right-hand side.

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

To obtain a Petri Net that is based on the Heuristics Miner, the code on the right hand side can be used. Also this Petri Net can be visualized.

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.petrinet import visualizer as pn_visualizer
gviz = pn_visualizer.apply(net, im, fm)
pn_visualizer.view(gviz)

First, we have to import the log. Subsequently, we can extract the Directly-Follows Graph. In addition, code is provided to visualize the Directly-Follows Graph. This visualization is a colored visualization of the Directly-Follows graph that is decorated with the frequency of 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)

To get a Directly-Follows graph decorated with the performance between the edges, two parameters of the previous code have to be replaced.

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)

To save the obtained DFG, for instance in the SVG format, code is also provided on the right-hand side.

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

First, we have to import the log. Subsequently, we have to mine the Directly-Follow graph. This DFG can then be converted to a workflow net.

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)

The code on the right-hand side can be used to obtain the Petri net mined by the Inductive Miner decorated with frequency information.

from pm4py.visualization.petrinet 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")

The example on the right-hand side shows the specification of an activity key for the Alpha Miner algorithm.

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)

The example on the right-hand side demonstrates the retrieval of the classifiers inside a log file, using the receipt.xes log. The print command returns a dictionary, whereby the corresponding classifier attribute is revealed.

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)

To use the classifier Activity classifier and write a new attribute for each event in the log, the following code can be used.

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

Then, as before, the Alpha Miner can be applied on the log specifying the newly inserted activity key.

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 case, the XES specifies no classifiers, and a different field should be used as activity key, there is the option to specify it manually. For example, in this piece of code we read the receipt.xes log and create a new attribute called customClassifier that is the activity name plus the transition. Subsequently, the Alpha Miner can be applied on this new classifier.

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)

Let’s try an example of that. First, we load a CSV file into a Pandas dataframe, keeping only the concept:name and the time:timestamp columns:

from pm4py.objects.log.adapters.pandas import csv_import_adapter

df = csv_import_adapter.import_dataframe_from_path("tests/input_data/receipt.csv")
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)
                                

First, we have to import the log. Subsequently, the Petri net is visualized by using the Petri Net visualizer. In addition, the Petri net is exported with its initial marking or initial marking and final marking.

import os
from pm4py.objects.petri.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.petrinet import visualizer as pn_visualizer
gviz = pn_visualizer.apply(net, initial_marking, final_marking)
pn_visualizer.view(gviz)

from pm4py.objects.petri.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)

The list of transitions enabled in a particular marking can be obtained using the right-hand code.

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

The function print(transitions) reports that only the transition register request is enabled in the initial marking in the given Petri net. To obtain all places, transitions, and arcs of the Petri net, the code which can be obtained on the right-hand side can be used.

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

Each place has a name and a set of input/output arcs (connected at source/target to a transition). Each transition has a name and a label and a set of input/output arcs (connected at source/target to a place). The code on the right-hand side prints for each place the name, and for each input arc of the place the name and the label of the corresponding transition. However, there also exsits trans.name, trans.label, arc.target.name.

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

The code on the right-hand side creates a Petri Net with the name new_petri_net.

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

In addition, three places are created, namely source, sink, and p_1. These places are added to the previously created 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)

Similar to the places, transitions can be created. However, they need to be assigned a name and a label.

# 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)

Arcs that connect places with transitions or transitions with places might be necessary. To add arcs, code is provided. The first parameter specifies the starting point of the arc, the second parameter its target and the last parameter states the Petri net it belongs to.

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

To complete the Petri net, an initial and possibly a final marking need to be defined. To accomplish this, we define the initial marking to contain 1 token in the source place and the final marking to contain 1 token in the sink place.

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

The resulting Petri net along with the initial and final marking can be exported, or visualized.

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

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

To obtain a specific output format (e.g. svg or png) a format parameter should be provided to the algorithm. The code snippet explains how to obtain an SVG representation of the Petri net. The last lines provide an option to save the visualization of the model.

from pm4py.visualization.petrinet 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.petrinet 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")

Let’s first load the running example log and apply the Alpha Miner.

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)
                                

Then, the decomposition can be found using:

from pm4py.objects.petri.decomposition import decompose

list_nets = decompose(net, im, fm)
                                

If we want to represent each one of the Petri nets, we can use a FOR loop:

from pm4py.visualization.petrinet 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")
                                

from pm4py.objects.petri 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)
                                

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

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)

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

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

To create an unfit model, a filtering operation producing a log where only part of the behavior is kept can be executed:

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

Then, applying the Inductive Miner algorithm:

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

We then apply the token-based replay with special settings. In particular, with disable_variants set to True we avoid to replay only a case with variant; with enable_pltr_fitness set to True we tell the algorithm to return localized Conformance Checking application.

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)
                                

To perform throughput analysis on the transitions that were executed unfit, and then print on the console the result, the following code could be used:

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])
                                

To perform throughput analysis on the process executions containing activities that are not in the model, and then print the result on the screen, the following code could be used:

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])
                                

In the following examples, that consider the Receipt log, the decision trees will be built on the following choice of attributes (i.e. only org:group attribute will be considered).

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

To perform root cause analysis on the transitions that are executed in an unfit way, the following code could be used:

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)
                                

To visualize the decision trees obtained by root cause analysis, the following code could be used:

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)
                                

To perform root cause analysis on activities that are executed but are not in the process model, the following code could be used:

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)
                                

To visualize the decision trees obtained by root cause analysis, the following code could be used:

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)
                                

First, we have to import the log. Subsequently, we apply the Inductive Miner on the imported log. In addition, we compute the alignments.

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 import algorithm as alignments
alignments = alignments.apply_log(log, net, initial_marking, final_marking)

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

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

print(alignments)

To use a different classifier, we refer to the Classifier section. However, the following code defines a custom classifier for each event of each trace in the log.

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

A parameters dictionary containing the activity key can be formed.

# 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"}

Then, a process model is computed, and alignments are also calculated. Besides, the fitness value is calculated and the resulting values are printed.

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

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

print(log_fitness) 

On the right-hand side, an implementation of a custom model cost function, and sync cost function can be noted. Also, the model cost funtions and sync cost function has to be inserted later in the parameters. Subsequently, the replay is done.

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

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

To perform alignments through decomposition/recomposition, the following code can be used. A maximum number of border disagreements can be provided to the algorithm. If the number of border disagreements is reached, then the alignment is interrupted a None as alignment of the specific trace is returned.

from pm4py.algo.conformance.decomp_alignments 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})

from pm4py.evaluation.replay_fitness import evaluator as rp_fitness_evaluator

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

from pm4py.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)

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)

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)

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)

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)

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)

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})

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)

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)

The following code can be used to import a process tree from a PTML file.

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

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

The following code can be used to export a process tree into a PTML file.

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

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

The code snippet can be used to generate a process tree.

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

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

from pm4py.objects.conversion.process_tree import converter as pt_converter
net, im, fm = pt_converter.apply(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)

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)
                                

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

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))
                                

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"})
                                

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"])
                                

import pandas as pd

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

from sklearn.decomposition import PCA

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

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)
                                

First, a log could be loaded. Then, a representation of a log on a given set of features could be obtained.

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)

First, a log has to be loaded. A representation of a log on a given set of features could be obtained.

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)

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.petrinet import visualizer

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

from pm4py.algo.enhancement.decision import algorithm as decision_mining

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

from pm4py.algo.enhancement.decision 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)
                                

from pm4py.statistics.traces.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.log import case_statistics
median_case_duration = case_statistics.get_median_caseduration(log, parameters={
    case_statistics.Parameters.TIMESTAMP_KEY: "time:timestamp"
})
                                    

from pm4py.statistics.traces.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.log import case_arrival
case_dispersion_ratio = case_arrival.get_case_dispersion_avg(log, parameters={
    case_arrival.Parameters.TIMESTAMP_KEY: "time:timestamp"})
                                    

from pm4py.statistics.performance_spectrum import algorithm as performance_spectrum
ps = performance_spectrum.apply(log, ["register request", "decide"],
                                parameters={performance_spectrum.Parameters.ACTIVITY_KEY: "concept:name",
                                            performance_spectrum.Parameters.TIMESTAMP_KEY: "time:timestamp"})
                                    

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)
                                    

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

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.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)
                                    

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)
                                    

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)
                                    

from pm4py.simulation.playout import simulator

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

from pm4py.simulation.playout import simulator

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

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.log import case_arrival
ratio = case_arrival.get_case_arrival_avg(log)
print(ratio)
                                    

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.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)
                                    

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
                                    

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.algo.filtering.log.variants import variants_filter
log = variants_filter.apply_auto_filter(log)
                                    

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)
                                    

# 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)
                                    

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)
                                    

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.simulation.tree_playout 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))
                                    

from pm4py.algo.enhancement.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)
                                    

from pm4py.algo.enhancement.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)
                                    

from pm4py.algo.enhancement.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)
                                    

from pm4py.algo.enhancement.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)
                                    

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.enhancement.roles import algorithm as roles_discovery
roles = roles_discovery.apply(log)