Petri Net Management

To obtain a Petri net from event data, start with process discovery. You can then use the model for conformance checking or simulation.

Petri nets are one of the most common formalisms used to express process models. A Petri net is a directed bipartite graph in which the nodes represent transitions and places. Arcs connect places to transitions and transitions to places, each with an associated weight. A transition can fire if each of its input places contains a number of tokens at least equal to the weight of the arc connecting the place to the transition. When a transition fires, tokens are removed from the input places according to the weight of the input arc and are added to the output places according to the weight of the output arc.

A marking is a state in the Petri net that associates each place with a number of tokens. It is uniquely associated with a set of enabled transitions that could be fired according to the marking. Process discovery algorithms implemented in PM4Py return a Petri net along with an initial marking and a final marking. The initial marking represents the starting state of the process execution, while the final marking is the state that should be reached at the end of the process.

Importing and Exporting

Petri nets, along with their initial and final markings, can be imported and exported from the PNML file format. The following code demonstrates how to import a Petri net along with its initial and final markings. First, we import the log. Then, the Petri net is visualized using the Petri Net visualizer. Additionally, the Petri net can be exported with its initial marking or both initial and final markings.

import os
import pm4py
from pm4py.objects.petri_net.obj import Marking, PetriNet

if __name__ == "__main__":
    net: PetriNet
    initial_marking: Marking
    final_marking: Marking
    net, initial_marking, final_marking = pm4py.read_pnml(os.path.join("tests","input_data","running-example.pnml"))
    pm4py.view_petri_net(net, initial_marking, final_marking)

    pm4py.write_pnml(net, initial_marking, final_marking, "petri.pnml")

Visualization of the object referenced in the example.

Petri Net Properties

This section explains how to obtain the properties of a Petri net. For example, you can retrieve the enabled transitions in a particular marking. Additionally, you can inspect the list of places, transitions, or arcs. The list of transitions enabled in a particular marking can be obtained using the code shown below.

from pm4py.objects.petri_net import semantics
from pm4py.objects.petri_net.obj import PetriNet

if __name__ == "__main__":
    transitions: set[PetriNet.Transition] = semantics.enabled_transitions(net, initial_marking)

The function print(transitions) reports that only the transition register_request is enabled in the initial marking of the given Petri net. To obtain all places, transitions, and arcs of the Petri net, use the following code:

from pm4py.objects.petri_net.obj import PetriNet
if __name__ == "__main__":
    places: set[PetriNet.Place] = net.places
    transitions: set[PetriNet.Transition] = net.transitions
    arcs: set[PetriNet.Arc] = net.arcs

Each place has a name and a set of input/output arcs (connected to a transition). Each transition has a name, label, and a set of input/output arcs (connected to a place). The code below prints the name of each place, along with the name and label of each corresponding transition for each input arc.

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

Creating a New Petri Net

This section provides an overview of the code necessary to create a new Petri net with places, transitions, and arcs. To create a Petri net object in PM4Py, you need to provide a name. The code below creates a Petri net named new_petri_net.

# creating an empty Petri net
from pm4py.objects.petri_net.obj import PetriNet, Marking

if __name__ == "__main__":
    net: PetriNet = PetriNet("new_petri_net")

In addition, three places are created: source, sink, and p_1. These places are added to the Petri net.

from pm4py.objects.petri_net.obj import PetriNet
if __name__ == "__main__":
    # creating source, p_1 and sink places
    source: PetriNet.Place = PetriNet.Place("source")
    sink: PetriNet.Place = PetriNet.Place("sink")
    p_1: PetriNet.Place = 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 places, transitions can also be created. However, each transition needs to be assigned a name and a label.

from pm4py.objects.petri_net.obj import PetriNet
if __name__ == "__main__":
   # Create transitions
    t_1: PetriNet.Transition = PetriNet.Transition("name_1", "label_1")
    t_2: PetriNet.Transition = 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) are required. The following code demonstrates how to add arcs. The first parameter specifies the arc's starting point, the second parameter its target, and the third parameter specifies the Petri net to which it belongs.

if __name__ == "__main__":
    # Add arcs
    from pm4py.objects.petri_net.utils import petri_utils
    petri_utils.add_arc_from_to(source, t_1, net)
    petri_utils.add_arc_from_to(t_1, p_1, net)
    petri_utils.add_arc_from_to(p_1, t_2, net)
    petri_utils.add_arc_from_to(t_2, sink, net)

To complete the Petri net, you need to define an initial marking and possibly a final marking. In this case, we define the initial marking to contain one token in the source place and the final marking to contain one token in the sink place.

from pm4py.objects.petri_net.obj import Marking
# Adding tokens

if __name__ == "__main__":
    initial_marking: Marking = Marking()
    initial_marking[source] = 1
    final_marking: Marking = Marking()
    final_marking[sink] = 1

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

# Visualizing the Petri net
import pm4py

if __name__ == "__main__":
    pm4py.write_pnml(net, initial_marking, final_marking, "createdPetriNet1.pnml")

    pm4py.view_petri_net(net, initial_marking, final_marking)

To obtain a specific output format (e.g., SVG or PNG), a format parameter should be provided to the algorithm. The following code snippet demonstrates how to obtain an SVG representation of the Petri net, with the option to save the visualization of the model.

import pm4py

if __name__ == "__main__":
    pm4py.view_petri_net(net, initial_marking, final_marking, format="svg")
    pm4py.save_vis_petri_net(net, initial_marking, final_marking, "net.svg")

Maximal Decomposition

The decomposition technique described in this section is useful for conformance checking. Splitting a large model into smaller ones can reduce the size of the state space, thereby increasing the performance of conformance checking operations. We propose using the maximal decomposition of a Petri net as described in the following paper:

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

Let's demonstrate maximal decomposition using the Petri net extracted by the Alpha Miner on the Running Example log. First, we load the log and apply the Alpha Miner.

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

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

Next, the decomposition can be found using the following code:

from pm4py.objects.petri_net.utils.decomposition import decompose
from pm4py.objects.petri_net.obj import Marking, PetriNet

if __name__ == "__main__":
    list_nets: list[tuple[PetriNet, Marking, Marking]] = decompose(net, im, fm)

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

import pm4py

if __name__ == "__main__":
    for index, (subnet, s_im, s_fm) in enumerate(list_nets):
        pm4py.save_vis_petri_net(subnet, s_im, s_fm, str(index) + ".png")

A log that fits the original model (projected onto the activities of the net) can also be adapted to these nets. Any deviations from these models indicate deviations in the original model as well.

Reachability Graph

A reachability graph is a transition system that can be constructed for any Petri net along with an initial marking. It is the graph of all markings of the Petri net, with markings connected by edges corresponding to the transitions connecting them. The main goal of the reachability graph is to provide insight into the state space of the Petri net. For Petri nets containing a lot of concurrency, the reachability graph can become very large, making its computation unfeasible for such models.

The reachability graph, given the Petri net and the initial marking, can be computed using the following code:

from pm4py.objects.petri_net.utils import reachability_graph
from pm4py.objects.transition_system.obj import TransitionSystem

if __name__ == "__main__":
    ts: TransitionSystem = reachability_graph.construct_reachability_graph(net, im)

The visualization of the reachability graph is possible with the following code snippet:

from pm4py.visualization.transition_system import visualizer as ts_visualizer
from graphviz import Graph

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

Petri Nets with Reset/Inhibitor Arcs

Support for Petri nets with reset/inhibitor arcs is available through the arctype property of a PetriNet.Arc object. The arctype property can have two possible values:

reset: Defines a reset arc. A reset arc removes all tokens from its source place whenever the target transition is fired.
inhibitor: Defines an inhibitor arc. An inhibitor arc blocks the firing of transitions it is connected to if there is at least one token in the source place.

The corresponding semantics, identical to the classic semantics of Petri nets, are defined in pm4py.objects.petri_net.inhibitor_reset.semantics.

Data Petri Nets

Data Petri nets extend the concept of Petri nets by including an execution context in the marking object. In a Data Petri net, the execution of a transition may depend on the value of the execution context, not just on the tokens. Data Petri nets are extensively defined in the following scientific paper:

Mannhardt, Felix, et al. "Balanced multi-perspective checking of process conformance." Computing 98.4 (2016): 407-437.

The semantics of a Data Petri net require the specification of the execution context (a dictionary associating attribute keys with values), which is defined in pm4py.objects.petri_net.data_petri_nets.semantics. The following methods require the execution context:

semantics.enabled_transitions(pn, m, e): Checks the enabled transitions in the provided Petri net pn and marking m when the execution context is updated with information from the current event.
semantics.execute(t, pn, m, e): Executes the transition t in the marking m, updating the execution context with information from the current event.