pm4py.convert module#

The pm4py.convert module contains the cross-conversions implemented in pm4py

pm4py.convert.convert_to_event_log(obj: DataFrame | EventStream, case_id_key: str = 'case:concept:name', **kwargs) → EventLog[source]#

Converts a DataFrame or EventStream object to an event log object.

Parameters:

  • obj – The DataFrame or EventStream object to convert.

  • case_id_key – The attribute to be used as the case identifier. Defaults to “case:concept:name”.

  • kwargs – Additional keyword arguments to pass to the converter.

Returns:

An EventLog object.

import pandas as pd
import pm4py

dataframe = pm4py.read_csv("tests/input_data/running-example.csv")
dataframe = pm4py.format_dataframe(dataframe, case_id_column='case:concept:name', activity_column='concept:name', timestamp_column='time:timestamp')
log = pm4py.convert_to_event_log(dataframe)
pm4py.convert.convert_to_event_stream(obj: EventLog | DataFrame, case_id_key: str = 'case:concept:name', **kwargs) → EventStream[source]#

Converts a log object or DataFrame to an event stream.

Parameters:

  • obj – The log object (EventLog) or DataFrame to convert.

  • case_id_key – The attribute to be used as the case identifier. Defaults to “case:concept:name”.

  • kwargs – Additional keyword arguments to pass to the converter.

Returns:

An EventStream object.

import pm4py

log = pm4py.read_xes("tests/input_data/running-example.xes")
event_stream = pm4py.convert_to_event_stream(log)
pm4py.convert.convert_to_dataframe(obj: EventStream | EventLog, **kwargs) → DataFrame[source]#

Converts a log object (EventStream or EventLog) to a Pandas DataFrame.

Parameters:

  • obj – The log object to convert.

  • kwargs – Additional keyword arguments to pass to the converter.

Returns:

A pd.DataFrame object.

import pm4py

log = pm4py.read_xes("tests/input_data/running-example.xes")
dataframe = pm4py.convert_to_dataframe(log)
pm4py.convert.convert_to_bpmn(*args: Tuple[PetriNet, Marking, Marking] | ProcessTree) → BPMN[source]#

Converts an object to a BPMN diagram.

As input, either a Petri net (with corresponding initial and final markings) or a process tree can be provided. A process tree can always be converted into a BPMN model, ensuring the quality of the resulting object. For Petri nets, the quality of the conversion largely depends on the net provided (e.g., sound WF-nets are likely to produce reasonable BPMN models).

Parameters:

args –

  • If converting a Petri net: a tuple of (PetriNet, Marking, Marking).

  • If converting a process tree: a single ProcessTree object.

Returns:

A BPMN object.

import pm4py

# Import a Petri net from a file
net, im, fm = pm4py.read_pnml("tests/input_data/running-example.pnml")
bpmn_graph = pm4py.convert_to_bpmn(net, im, fm)
pm4py.convert.convert_to_petri_net(*args: BPMN | ProcessTree | HeuristicsNet | POWL | dict) → Tuple[PetriNet, Marking, Marking][source]#

Converts an input model to an (accepting) Petri net.

The input objects can be a process tree, BPMN model, Heuristic net, POWL model, or a dictionary representing a Directly-Follows Graph (DFG). The output is a tuple containing the Petri net and the initial and final markings. The markings are only returned if they can be reasonably derived from the input model.

Parameters:

args –

  • If converting from a BPMN, ProcessTree, HeuristicsNet, or POWL: a single object of the respective type.

  • If converting from a DFG: a dictionary representing the DFG, followed by lists of start and end activities.

Returns:

A tuple of (PetriNet, Marking, Marking).

import pm4py

# Imports a process tree from a PTML file
process_tree = pm4py.read_ptml("tests/input_data/running-example.ptml")
net, im, fm = pm4py.convert_to_petri_net(process_tree)
pm4py.convert.convert_to_process_tree(*args: Tuple[PetriNet, Marking, Marking] | BPMN | ProcessTree | POWL) → ProcessTree[source]#

Converts an input model to a process tree.

The input models can be Petri nets (with markings) or BPMN models. For both input types, the conversion is not guaranteed to work and may raise an exception.

Parameters:

args –

  • If converting from a Petri net: a tuple of (PetriNet, Marking, Marking).

  • If converting from a BPMN or ProcessTree: a single object of the respective type.

Returns:

A ProcessTree object.

import pm4py

# Imports a BPMN file
bpmn_graph = pm4py.read_bpmn("tests/input_data/running-example.bpmn")
# Converts the BPMN to a process tree (through intermediate conversion to a Petri net)
process_tree = pm4py.convert_to_process_tree(bpmn_graph)
pm4py.convert.convert_to_powl(*args: Tuple[PetriNet, Marking, Marking] | BPMN | ProcessTree) → POWL[source]#

Converts an input model to a POWL model.

The input models can be Petri nets (with markings) or BPMN models or process trees. For both input types, the conversion is not guaranteed to work and may raise an exception.

Parameters:

args –

  • If converting from a Petri net: a tuple of (PetriNet, Marking, Marking).

  • If converting from a BPMN or ProcessTree: a single object of the respective type.

Returns:

A ProcessTree object.

import pm4py

# Imports a BPMN file
bpmn_graph = pm4py.read_bpmn("tests/input_data/running-example.bpmn")
# Converts the BPMN to a POWL (through intermediate conversion to a Petri net)
powl = pm4py.convert_to_powl(bpmn_graph)
print(powl)
pm4py.convert.convert_to_reachability_graph(*args: Tuple[PetriNet, Marking, Marking] | BPMN | ProcessTree) → TransitionSystem[source]#

Converts an input model to a reachability graph (transition system).

The input models can be Petri nets (with markings), BPMN models, or process trees. The output is the state-space of the model, encoded as a TransitionSystem object.

Parameters:

args –

  • If converting from a Petri net: a tuple of (PetriNet, Marking, Marking).

  • If converting from a BPMN or ProcessTree: a single object of the respective type.

Returns:

A TransitionSystem object.

import pm4py

# Reads a Petri net from a file
net, im, fm = pm4py.read_pnml("tests/input_data/running-example.pnml")
# Converts it to a reachability graph
reach_graph = pm4py.convert_to_reachability_graph(net, im, fm)
pm4py.convert.convert_log_to_ocel(log: EventLog | EventStream | DataFrame, activity_column: str = 'concept:name', timestamp_column: str = 'time:timestamp', object_types: Collection[str] | None = None, obj_separator: str = ' AND ', additional_event_attributes: Collection[str] | None = None, additional_object_attributes: Dict[str, Collection[str]] | None = None) → OCEL[source]#

Converts an event log to an object-centric event log (OCEL) with one or more object types.

Parameters:

  • log – The log object to convert.

  • activity_column – The name of the column representing activities.

  • timestamp_column – The name of the column representing timestamps.

  • object_types – A collection of column names to consider as object types. If None, defaults are used.

  • obj_separator – The separator used between different objects in the same column. Defaults to “ AND “.

  • additional_event_attributes – Additional attribute names to include as event attributes in the OCEL.

  • additional_object_attributes – Additional attributes per object type to include as object attributes in the OCEL. Should be a dictionary mapping object types to lists of attribute names.

Returns:

An OCEL object.

pm4py.convert.convert_ocel_to_networkx(ocel: OCEL, variant: str = 'ocel_to_nx') → DiGraph[source]#

Converts an OCEL to a NetworkX DiGraph object.

Parameters:

  • ocel – The object-centric event log to convert.

  • variant – The variant of the conversion to use. Options: - “ocel_to_nx”: Graph containing event and object IDs and two types of relations (REL=related objects, DF=directly-follows). - “ocel_features_to_nx”: Graph containing different types of interconnections at the object level.

Returns:

A nx.DiGraph object representing the OCEL.

pm4py.convert.convert_log_to_networkx(log: EventLog | EventStream | DataFrame, include_df: bool = True, case_id_key: str = 'concept:name', other_case_attributes_as_nodes: Collection[str] | None = None, event_attributes_as_nodes: Collection[str] | None = None) → DiGraph[source]#

Converts an event log to a NetworkX DiGraph object.

The nodes of the graph include events, cases, and optionally log attributes. The edges represent: - BELONGS_TO: Connecting each event to its corresponding case. - DF: Connecting events that directly follow each other (if enabled). - ATTRIBUTE_EDGE: Connecting cases/events to their attribute values.

Parameters:

  • log – The log object to convert (EventLog, EventStream, or Pandas DataFrame).

  • include_df – Whether to include the directly-follows relation in the graph. Defaults to True.

  • case_id_key – The attribute to be used as the case identifier. Defaults to “concept:name”.

  • other_case_attributes_as_nodes – Attributes at the case level to include as nodes, excluding the case ID.

  • event_attributes_as_nodes – Attributes at the event level to include as nodes.

Returns:

A nx.DiGraph object representing the event log.

pm4py.convert.convert_log_to_time_intervals(log: EventLog | DataFrame, filter_activity_couple: Tuple[str, str] | None = None, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', start_timestamp_key: str = 'time:timestamp') → List[List[Any]][source]#

Extracts a list of time intervals from an event log.

Each interval contains two temporally consecutive events within the same case and measures the time between them (complete timestamp of the first event against the start timestamp of the second event).

Parameters:

  • log – The log object to convert.

  • filter_activity_couple – Optional tuple to filter intervals by a specific pair of activities.

  • activity_key – The attribute to be used as the activity identifier. Defaults to “concept:name”.

  • timestamp_key – The attribute to be used as the timestamp. Defaults to “time:timestamp”.

  • case_id_key – The attribute to be used as the case identifier. Defaults to “case:concept:name”.

  • start_timestamp_key – The attribute to be used as the start timestamp in the interval. Defaults to “time:timestamp”.

Returns:

A list of intervals, where each interval is a list containing relevant information about the time gap.

import pm4py

log = pm4py.read_xes('tests/input_data/receipt.xes')
time_intervals = pm4py.convert_log_to_time_intervals(log)
print(len(time_intervals))
time_intervals = pm4py.convert_log_to_time_intervals(
    log,
    filter_activity_couple=('Confirmation of receipt', 'T02 Check confirmation of receipt')
)
print(len(time_intervals))
pm4py.convert.convert_petri_net_to_networkx(net: PetriNet, im: Marking, fm: Marking) → DiGraph[source]#

Converts a Petri net to a NetworkX DiGraph.

Each place and transition in the Petri net is represented as a node in the graph.

Parameters:

  • net – The Petri net to convert.

  • im – The initial marking of the Petri net.

  • fm – The final marking of the Petri net.

Returns:

A nx.DiGraph object representing the Petri net.

pm4py.convert.convert_petri_net_type(net: PetriNet, im: Marking, fm: Marking, type: str = 'classic') → Tuple[PetriNet, Marking, Marking][source]#

Changes the internal type of a Petri net.

Supports conversion to different Petri net types such as classic, reset, inhibitor, and reset_inhibitor nets.

Parameters:

  • net – The Petri net to convert.

  • im – The initial marking of the Petri net.

  • fm – The final marking of the Petri net.

  • type – The target Petri net type. Options are “classic”, “reset”, “inhibitor”, “reset_inhibitor”. Defaults to “classic”.

Returns:

A tuple of the converted (PetriNet, Marking, Marking).