Simulation

Create the model and its markings with Petri net management or process discovery. Compare the resulting behavior using log and model evaluation.

In PM4Py, we offer various simulation algorithms that, starting from a model, generate outputs that adhere to the model and the rules provided by the user.

Playout of a Petri Net

Playout APIs return a legacy EventLog, so len(simulated_log) counts traces. To analyze its events with Pandas, use pm4py.convert_to_dataframe(simulated_log).

A playout of a Petri net takes a Petri net and an initial marking as input, returning a list of process executions permitted by the process model. We offer two types of playouts:

The parameters for each variant are as follows:

Here's an example of applying the basic playout, given a Petri net, to generate a log with 50 traces:

from pm4py.algo.simulation.playout.petri_net import algorithm as simulator
from pm4py.objects.log.obj import EventLog

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

Here's an example of applying the extensive playout, given a Petri net, to generate a log with all executions of length ≤ 7:

from pm4py.algo.simulation.playout.petri_net import algorithm as simulator
from pm4py.objects.log.obj import EventLog

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

Monte Carlo Simulation

The Monte Carlo simulator requires a legacy EventLog. The example converts the DataFrame at that API call and retains the original DataFrame for timestamp analysis.

A time-related simulation allows you to determine the probability of process execution completion within a given time frame. This leads to a better estimation of Service Level Agreements or a better identification of process instances likely to have long throughput times.

This process begins with a performance DFG (e.g., the one discovered from the running example log):

import os
import pm4py
import pandas

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

Additionally, the case arrival ratio must be known. This refers to the average (or median) time between the arrival of two consecutive cases. It can be provided by the user or inferred from the event log using the following command:

import pm4py

if __name__ == "__main__":
    ratio: dict[str, int] = pm4py.get_rework_cases_per_activity(log)
    print(ratio)

Using the DFG mining approach, you can extract a Petri net model. This model is the “default” choice for Monte Carlo simulation, as its execution semantics are clear. Moreover, the Petri net generated by DFG mining is a sound workflow net, which ensures good properties for the model.

The DFG mining approach can be applied as follows:

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

if __name__ == "__main__":
    net: PetriNet
    im: Marking
    fm: Marking
    net, im, fm = pm4py.convert_to_petri_net(dfg_perf, sa, ea)

To perform a basic Monte Carlo simulation, the following code can be used. This example assumes a resource-constrained simulation, where a place can hold at most 1 token at a time. Later, we’ll show how to allow more tokens per place.

from pm4py.algo.simulation.montecarlo import algorithm as montecarlo_simulation
from pm4py.algo.conformance.tokenreplay.algorithm import Variants
from pm4py.objects.log.obj import EventLog

if __name__ == "__main__":
    parameters: dict = {}
    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: EventLog
    res: dict
    simulated_log, res = montecarlo_simulation.apply(
        pm4py.convert_to_event_log(log), net, im, fm, parameters=parameters
    )

The main outputs of the simulation process are:

The result dictionary, res, includes the following keys:

The last four items in the list are basic Python objects (e.g., floats and lists). The interval tree objects can be queried to obtain time-specific information. For example, the following code snippet prints, for a random transition, the number of overlapping intervals at 11 different time points (including the minimum and maximum timestamps):

import random
from typing import Any
if __name__ == "__main__":
    last_timestamp: float = log["time:timestamp"].max().timestamp()
    first_timestamp: float = log["time:timestamp"].min().timestamp()
    pick_trans: Any = random.choice(list(res["output_transitions_interval_trees"]))
    print(pick_trans)
    n_div: int = 10
    i: int = 0
    while i < n_div:
        timestamp = first_timestamp + (last_timestamp - first_timestamp)/n_div * i
        print("\t", timestamp, len(res["output_transitions_interval_trees"][pick_trans][timestamp]))
        i = i + 1

Similarly, the following code snippet prints, for a random place, the number of overlapping intervals at 11 different time points:

import random
from typing import Any
if __name__ == "__main__":
    last_timestamp: float = log["time:timestamp"].max().timestamp()
    first_timestamp: float = log["time:timestamp"].min().timestamp()
    pick_place: Any = random.choice(list(res["output_places_interval_trees"]))
    print(pick_place)
    n_div: int = 10
    i: int = 0
    while i < n_div:
        timestamp = first_timestamp + (last_timestamp - first_timestamp)/n_div * i
        print("\t", timestamp, len(res["output_places_interval_trees"][pick_place][timestamp]))
        i = i + 1

This information can be used to build graphs using tools like Microsoft Excel.

The simulation process can be summarized as follows:

• An event log and a model (DFG) are used as inputs.
• The simulation performs a replay operation between the log and the model.
• This replay operation constructs a stochastic map associating each transition with a probability distribution, selected to maximize the likelihood of the observed values. Users can specify a distribution (e.g., exponential) if needed.
• During replay, the frequency of each transition is calculated, allowing the weighted selection of enabled transitions.
• The simulation runs via generator-based case simulators interleaved by a virtual-time scheduler using a priority queue (no real-time sleeping or OS threads).
• Cases are started according to the case arrival ratio; at each step, the next case to advance is the one with the earliest next simulated time.

Several parameters are essential for performing a Monte Carlo simulation. These parameters, found in the petri_semaph_fifo class, are listed below in order of importance:

Extensive Playout of a Process Tree

An extensive playout operation allows the generation (up to the specified limits) of the entire language of the process model. Performing an extensive playout on a Petri net can be computationally expensive, as it requires exploring the reachability graph. Process trees, with their bottom-up structure, make it much easier to generate the entire language of an event log. The process starts from the language of the leaves (which is straightforward) and follows specific merge rules for the operators.

However, since the language of a process tree can be vast (especially when parallel operators are involved) or even infinite (when loops are present), extensive playouts are possible only within certain limits:

• If a loop is present, the maximum number of occurrences for that loop must be specified. This will stop the extensive playout operation after the given number of occurrences.
• Since the number of possible executions involving loops can still be very large, the maximum length of a trace can be specified. Traces that exceed this length are automatically discarded.
• To further limit the number of executions, the maximum number of traces returned by the algorithm can also be specified.

Additionally, the structure of the process tree makes it easy to infer the minimum trace length allowed by the process model (following the bottom-up approach).

Some reasonable default settings for the extensive playout are as follows:

• The maximum number of traces returned by the algorithm is set to 100,000.
• The maximum length of a trace output by the playout is, by default, set to the minimum trace length accepted by the process tree.
• The maximum number of loop occurrences is set to half the minimum trace length.

The list of parameters for the extensive playout is as follows:

Below is an example of how the playout can be executed. First, a log can be imported:

import pm4py
import os
import pandas

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

Next, a process tree can be discovered using the inductive miner algorithm:

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

In this example, we specify retrieving traces of length at most 3, and we set the maximum number of traces to 100,000.

from pm4py.algo.simulation.playout.process_tree import algorithm as tree_playout
from pm4py.objects.log.obj import EventLog
from typing import Any

if __name__ == "__main__":
    playout_variant: Any = tree_playout.Variants.EXTENSIVE
    param: Any = tree_playout.Variants.EXTENSIVE.value.Parameters

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

At this point, the extensive playout operation is complete.