MicroPM4Py

Process Mining for Microcontrollers in Python

Documentation

This page contains the documentation of the MicroPM4Py Python process mining library for microcontrollers / embedded systems. Please use these links to reach the section of interest.

Importing/exporting XES files

All the features of import/export in MicroPM4Py works only with the case ID and activity attributes. Other attributes of the log are currently simply ignored.

There are several ways to import a XES file in MicroPM4Py:
  • Importing a XES file as a list of cases (assuming one tag per row in the XML file).
  • (Standard importer; works always albeit is slower) importing the XES as a list of cases.
  • Importing the XES as a DFG object.
  • Importing a XES as a dictionary of variants.
  • Importing XES piece-by-piece with an iterator.
In the following, an example of the format in which the log is stored in memory is reported. Within this data structure, no further compression of strings is done (while the DFG object is optimized to minimize the memory footprint). Here, a list of tuples is obtained. For each tuple, the first element is the case ID, while the second is the tuple of activities contained in the case (the so-called trace).
[('3', ('register request', 'examine casually', 'check ticket', 'decide', 'reinitiate request', 'examine thoroughly', 'check ticket', 'decide', 'pay compensation')), ('2', ('register request', 'check ticket', 'examine casually', 'decide', 'pay compensation')), ('1', ('register request', 'examine thoroughly', 'check ticket', 'decide', 'reject request')), ('6', ('register request', 'examine casually', 'check ticket', 'decide', 'pay compensation')), ('5', ('register request', 'examine casually', 'check ticket', 'decide', 'reinitiate request', 'check ticket', 'examine casually', 'decide', 'reinitiate request', 'examine casually', 'check ticket', 'decide', 'reject request')), ('4', ('register request', 'check ticket', 'examine thoroughly', 'decide', 'reject request'))]
The following code snippet helps to import a XES file into the previously described data structure (here, the faster but non-standard importer is used):
from micropm4py.log import xes_import_traces_file
log = xes_import_traces_file.imp_list_traces_from_file('micro_tests/running-example.xes')
Instead, here the standard importer is used (slower but always works)
from micropm4py.log import xes_import_traces_file_standard
log = xes_import_traces_file_standard.imp_list_traces_from_file('micro_tests/running-example.xes')
To export a log structure into a XES file (only with case ID and activity), the following code can be used:
from micropm4py.log import xes_export_traces_file
xes_export_traces_file.export_traces(log, 'ru.xes')
To import a DFG from the XES, without need to store all the event log, the following code can be used. In the following of the documentation, the description of the DFG structure is reported.
from micropm4py.log import xes_import_traces_file_standard
dfg = xes_import_traces_file_standard.imp_dfg_file('micro_tests/running-example.xes')
To import a dictionary of variants from the XES, without need to store all the event log, the following code can be used.
from micropm4py.log import xes_import_traces_file_standard
variants = xes_import_traces_file_standard.imp_variants_from_file('micro_tests/running-example.xes')
The dictionary of variants for the running-example log looks as follow:
{('register request', 'examine casually', 'check ticket', 'decide', 'pay compensation'): 1, ('register request', 'examine casually', 'check ticket', 'decide', 'reinitiate request', 'check ticket', 'examine casually', 'decide', 'reinitiate request', 'examine casually', 'check ticket', 'decide', 'reject request'): 1, ('register request', 'examine casually', 'check ticket', 'decide', 'reinitiate request', 'examine thoroughly', 'check ticket', 'decide', 'pay compensation'): 1, ('register request', 'check ticket', 'examine thoroughly', 'decide', 'reject request'): 1, ('register request', 'examine thoroughly', 'check ticket', 'decide', 'reject request'): 1, ('register request', 'check ticket', 'examine casually', 'decide', 'pay compensation'): 1}
It is possible to import the log trace-by-trace using an iterator pattern. There are two types of iterations that are possible in MicroPM4Py:
  • Iterate trace-by-trace the XES (without caring for uniqueness; this option is also the most memory efficient)
  • Iterate on the unique traces of the XES file

Two examples for such iterations have been proposed in the examples folder (xes_example_iterator and xes_example_it_unique).


Importing/exporting CSV files
For importing a CSV file (just the case ID and the activity), the following information is needed:
  • The separator character (, or ;) that separates the columns of the CSV file
  • The name of the column containing the case ID
  • The name of the column containing the activity
The following code snippet helps to import a CSV file into the log data structure. Here, the first argument of the method is the path to the CSV file, the second element is the separator, the third element is the case ID column, and the fourth element is the activity column. No support for the quotechar is there.
from micropm4py.log import csv_import_traces_file
log = csv_import_traces_file.import_traces_path('micro_tests/running-example.csv', ',', 'case:concept:name', 'concept:name')
The following code snippet helps to export a log into a CSV. Here, the first argument is the log structure, the second is the (target) file path, the third is the separator, the fourth is the name of the (exported file) case ID column, the fifth is the name of the (exported file) activity column.
from micropm4py.log import csv_export_traces_file
csv_export_traces_file.export_to_path(log, 'ru.csv', ',', 'case:concept:name', 'concept:name')
To import directly the DFG from the CSV, without the need to store the log in-memory (so, significant memory gains are achieved), the following code can be used:
from micropm4py.log import csv_import_traces_file
dfg = csv_import_traces_file.import_dfg_path('micro_tests/running-example.csv', ',', 'case:concept:name', 'concept:name')

Working with DFGs

A DFG is an object that is central for many process mining applications (e.g. Alpha Miner, Inductive Miner Directly-Follows, DFG mining techniques ...).

It can be described as a composite object containing:
  • The list of activities of the log
  • The start activities of the log
  • The end activities of the log
  • The tuples of activities directly-following each other in some cases of the log, along with the occurrences of the relationship

The DFG data strucure in MicroPM4Py is described by the following example:

[('>>', 'register request', 'examine casually', 'check ticket', 'decide', 'reinitiate request', 'examine thoroughly', 'pay compensation', '[]', 'reject request'), {0: 6, 1: 6, 2: 6, 3: 9, 4: 9, 5: 3, 6: 3, 7: 3, 8: 6, 9: 3}, (0,), (8,), {(0, 1): 6, (1, 2): 3, (3, 2): 2, (1, 3): 2, (4, 9): 3, (6, 4): 1, (5, 6): 1, (3, 4): 6, (9, 8): 3, (5, 2): 1, (4, 7): 3, (6, 3): 2, (4, 5): 3, (2, 3): 4, (3, 6): 1, (1, 6): 1, (7, 8): 3, (2, 4): 2, (5, 3): 1}]

Here, the first sublist is the tuple of activities in the log, the second dictionary associates the index of the activity (according to the first tuple) to the number of occurrences of the activity, the third set is the tuple of start activities of the log, the fourth tuple is the set of end activities of the log; the last dictionary contains the tuples of activities directly-following each other.

DFGs can be directly imported in MicroPM4Py without the need to 'pass into' the log structure, as seen in the previous sections of the documentation.

Alternatively, a log structure can be converted to a DFG with the following snippet:

from micropm4py.conversion.dfg import traces_to_dfg
dfg = traces_to_dfg.trs_to_dfg(log)
To produce a visualization of the DFG into the .dot format (Graphviz), the following code snippet can be used. Here, the first element is the (target) file path, while the second element is the DFG object:
from micropm4py.visualization.dfg_saver import save_dot
save_dot('graph.dot', dfg)
The DOT file can be then rendered using the Graphviz tools. For small graphs, to render the DOT into a PNG, the following shell command can be used
dot -Tpng graph.dot > graph.png
For bigger graphs, the following shell command can be used:
neato -Tpng graph.dot > graph.png

Petri Nets - Importing/Exporting/Visualization

Petri Nets are a widely used class of models in the process mining world. In MicroPM4Py, we offer support for importing/exporting/visualizing accepting Petri nets (that are Petri nets along with an initial and a final marking).

A limitation of the implementation is the lack of compatibility for invisible transitions. Duplicate visible transitions are instead supported.

Another limitation is the support for arc weights equal to 1 (this was originally a limitation only of the importing, but the resulting in-memory structure is way more efficient without the specification of 1 as arc weight).

The data structure hosting the Petri nets in MicroPM4Py is reported in the following example. The first tuple contains the names of the places (the example contains 3 places, the source place, the sink place and a place named p1). The second tuple contains the transitions. Each transition is a (sub)tuple where the first element is the label of the transition, the second element is the preset of the transition (a tuple containing the places consumed, with weight 1, from the preset), the third element is the postset of the transition (described in an analogous way as the preset).

[('source', 'p1', 'sink'), (('A', (0,), (1,)), ('B', (1,), (2,)))]

The structure of the initial/final marking is a dictionary (which structure is a dictionary associating a place with the number of tokens in the place). As example:

{0: 1}
The following code snippet imports a Petri net from a PNML file, along with its initial and final marking:
from micropm4py.petrinet import petrinet_import_file
net, im, fm = petrinet_import_file.imp_file('micro_tests/running-example.pnml')
The following code snippet exports a Petri net into a PNML file, along with its initial and final marking:
from micropm4py.petrinet import petrinet_export_file
petrinet_export_file.export(net, im, fm, "target.pnml")
To produce a visualization of the Petri net into the .dot format (Graphviz), the following code snippet can be used. Here, the first element is the (target) file path, while the second element is the Petri net object (visualized without specification of the initial and final marking):
from micropm4py.visualization.petri_saver import save_dot
save_dot('graph.dot', net)
The DOT file can be then rendered using the Graphviz tools. For small graphs, to render the DOT into a PNG, the following shell command can be used
dot -Tpng graph.dot > graph.png
For bigger graphs, the following shell command can be used:
neato -Tpng graph.dot > graph.png

Petri Nets - Execution semantics

Petri nets are a widely used formalism in process mining, and their importing/exporting/visualization has been described in the previous section. A marking is a set of tokens distributed among the places of the Petri net.

The execution semantics of a Petri net is clear: when a transition is executed, a number of tokens (the preset of the transition) is consumed from the current marking, and a number of tokens (the postset of the transition) is inserted into the marking.

In an accepting Petri net, the initial state of any execution is the initial marking. Ideally, the expectation is that every process execution finishes in the final marking.

Let's see an example. Before everything, import the execution semantics of the Petri net, and actually import a Petri net, for example the running-example one.

from micropm4py.petrinet import petrinet_exec
from micropm4py.petrinet import petrinet_import_file
net, im, fm = petrinet_import_file.imp_file('../micro_tests/running-example.pnml')
This Petri net can be represented (using MicroPM4Py) as follows:

The initial activity, as prescribed by the Petri net, is always register request. Then, check ticket is executed concurrently to one between examine casually and examine throughly. Then, a decision occurs (decide) on the outcome of the ticket. A reinitiation of the request can happen; if not, one between pay compensation and reject request is executed.

To execute the same Petri net several times, since the execution acts on the marking, we need to copy the initial marking when we start the execution. This can be done with the following code snippet:

from micropm4py.util import copy
m = copy.copy(im)
The Petri net, initial marking and final marking structures, as imported in MicroPM4Py from the PNML file, are:
[('start', '({'register request', 'reinitiate request'}, {'check ticket'})', '({'examine casually', 'examine thoroughly'}, {'decide'})', 'end', '({'decide'}, {'reinitiate request', 'pay compensation', 'reject request'})', '({'check ticket'}, {'decide'})', '({'register request', 'reinitiate request'}, {'examine casually', 'examine thoroughly'})'), (('decide', {2: 1, 5: 1}, {4: 1}), ('examine casually', {6: 1}, {2: 1}), ('examine thoroughly', {6: 1}, {2: 1}), ('register request', {0: 1}, {1: 1, 6: 1}), ('check ticket', {1: 1}, {5: 1}), ('reinitiate request', {4: 1}, {1: 1, 6: 1}), ('pay compensation', {4: 1}, {3: 1}), ('reject request', {4: 1}, {3: 1}))]
{0: 1}
{3: 1}
As in the process model, we execute the activity register request. We expect to enable both the check ticket activity and one between examine casually/examine throughly. This is exactly what happens:
{1: 1, 6: 1}
Then, let's try to execute the following list of activities: examine casually, check ticket, decide, pay compensation. That should lead to the final marking of the process model.
petrinet_exec.exec(net, m, 'examine casually')
petrinet_exec.exec(net, m, 'check ticket')
petrinet_exec.exec(net, m, 'decide')
petrinet_exec.exec(net, m, 'pay compensation')
This is exactly what happens:
{3: 1}

Token-based Replay: instead of executing the activity one-by-one, in MicroPM4Py there is the possibility to execute through token-based replay an entire trace, just to check whether it fits or not (so, returning a boolean value). Please reminder that there is support only for visible transitions, and the replay just tells if the trace is fit or not, no other information (missing/remaining tokens, fitness ...).

Let's try to execute token-based replay on a fit trace. The True value is printed correctly:

m = copy.copy(im)
res = petrinet_exec.ex_trace(net, m, fm, ['register request', 'examine casually', 'check ticket', 'decide', 'pay compensation'])
print(res)
A token-based replay on an unfit trace prints False as expected (here, the unfitness is the lack of execution of the activity check ticket).
m = copy.copy(im)
res = petrinet_exec.ex_trace(net, m, fm, ['register request', 'examine casually', 'decide', 'pay compensation'])
print(res)
A token-based replay on a trace that is fitting but without reaching (still) the final marking prints False (since the FM is not reached):
m = copy.copy(im)
res = petrinet_exec.ex_trace(net, m, fm, ['register request', 'examine casually', 'check ticket', 'decide'])
print(res)

Algorithms for Petri Nets discovery

Alpha Miner: in MicroPM4Py, we offer the well-known Alpha Miner algorithm to discover a process model. The input data structure is the DFG as described previously.

In the following, we try to provide an example of application of the Alpha Miner to the running example log.

First, the DFG needs to be calculated/imported from the log:

from micropm4py.log import xes_import_traces_file
dfg = xes_import_traces_file.imp_dfg_file('micro_tests/running-example.xes')
Then, we can apply the Alpha Miner algorithm to retrieve a Petri net along with an initial and a final marking:
from micropm4py.discovery import alpha
net, im, fm = alpha.alpha(dfg)
We could also provide a visualization (DOT format) of such Petri net:
from micropm4py.visualization import petri_print
petri_print.print_dot(net)

DFG mining: in MicroPM4Py, we offer the DFG mining approach to convert the DFG into a sound workflow net with duplicate transitions and arc weights equal to 1.

The algorithm is described here. The only limitation of this implementation (since the Petri Nets in MicroPM4Py do not support invisibles) is that only DFGs with an unique start and end activity are supported.

Essentially, the previous conditions can be obtained in any event log inserting the artificial start and end activities (in MicroPM4Py, they are denoted as >> and []).

To import a XES log directly into a DFG with artificial start/end activities, the following code can be used:

from micropm4py.log import xes_import_traces_file_standard
dfg = xes_import_traces_file_standard.imp_dfg_file_sten('micro_tests/running-example.xes')
The DFG mining algorithm can be applied as in the following code snippet:
from micropm4py.conversion.dfg import dfg_mining
net, im, fm = dfg_mining.apply(dfg)

The result is an accepting Petri net that can be used anywhere in MicroPM4Py.


Conversions from/to PM4Py

The MicroPM4Py is easily interoperable with the PM4Py library. Most objects in MicroPM4Py can be converted from/to objects of PM4Py.

To convert from/to PM4Py objects, obviously PM4Py needs to be also installed.

MicroPM4Py DFG => PM4Py DFG the following code snippet can be used:

from micropm4py.conversion.pm4py import dfg as dfg_conv
dfg_pm4, acti_count, start_activities, end_activities = dfg_conv.to(dfg_mp)
where:
  • The first element is the dictionary of activities following each other.
  • The second element contains the number of occurrences of the activities.
  • The third element is the set of start activities of the log.
  • The fourth element is the set of end activities of the log.

MicroPM4Py log => PM4Py log the following code snippet can be used (here, the log is the PM4Py log, and the log0 is the MicroPM4Py log):

from micropm4py.log import xes_import_traces_file
log = traces.to(log0)

PM4Py log => MicroPM4Py log the following code snippet can be used (here, the log is the PM4Py log, and the log1 is the MicroPM4Py log):

from micropm4py.log import xes_import_traces_file
log1 = traces.frm(log)

MicroPM4Py Petri net => PM4Py Petri net the following code snippet can be used (here, the net is the PM4Py Petri net, and the net0 is the MicroPM4Py Petri net):

from micropm4py.conversion.pm4py import petri
net, im, fm = petri.to(net0, im0, fm0)

PM4Py Petri net => MicroPM4Py Petri net the following code snippet can be used (here, the net is the PM4Py Petri net, and the net1 is the MicroPM4Py Petri net):

from micropm4py.conversion.pm4py import petri
net1, im1, fm1 = petri.frm(net, im, fm)