pm4py.utils module#

class pm4py.utils.Shared[source]#

Bases: object

RUSTXES_WARNING_SHOWN = False#
pm4py.utils.format_dataframe(df: DataFrame, case_id: str = 'case:concept:name', activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', start_timestamp_key: str = 'start_timestamp', timest_format: str | None = None) → DataFrame[source]#

Formats the dataframe appropriately for process mining purposes.

Parameters:

  • df – Dataframe.

  • case_id – Case identifier column.

  • activity_key – Activity column.

  • timestamp_key – Timestamp column.

  • start_timestamp_key – Start timestamp column.

  • timest_format – Timestamp format provided to Pandas.

Returns:

A formatted pandas DataFrame.

Return type:

pd.DataFrame

import pandas as pd
import pm4py

dataframe = pd.read_csv('event_log.csv')
dataframe = pm4py.format_dataframe(
    dataframe,
    case_id='case:concept:name',
    activity_key='concept:name',
    timestamp_key='time:timestamp',
    start_timestamp_key='start_timestamp',
    timest_format='%Y-%m-%d %H:%M:%S'
)
pm4py.utils.rebase(log_obj: EventLog | EventStream | DataFrame, case_id: str = 'case:concept:name', activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', start_timestamp_key: str = 'start_timestamp', timest_format: str | None = None) → EventLog | EventStream | DataFrame[source]#

Re-bases the log object by changing the case ID, activity, and timestamp attributes.

Parameters:

  • log_obj – Log object.

  • case_id – Case identifier.

  • activity_key – Activity.

  • timestamp_key – Timestamp.

  • start_timestamp_key – Start timestamp.

  • timest_format – Timestamp format provided to Pandas.

Returns:

A re-based log object.

Return type:

Union[EventLog, EventStream, pd.DataFrame].

import pm4py

rebased_dataframe = pm4py.rebase(
    dataframe,
    case_id='case:concept:name',
    activity_key='concept:name',
    timestamp_key='time:timestamp',
    start_timestamp_key='start_timestamp',
    timest_format='%Y-%m-%d %H:%M:%S'
)
pm4py.utils.parse_process_tree(tree_string: str) → ProcessTree[source]#

Parses a process tree from a string.

Parameters:

tree_string – String representing a process tree (e.g., “-> ( ‘A’, O ( ‘B’, ‘C’ ), ‘D’ )”). Operators are ‘->’ for sequence, ‘+’ for parallel, ‘X’ for XOR choice, ‘*’ for binary loop, and ‘O’ for choice.

Returns:

A ProcessTree object.

Return type:

ProcessTree

import pm4py

process_tree = pm4py.parse_process_tree("-> ( 'A', O ( 'B', 'C' ), 'D' )")
pm4py.utils.parse_powl_model_string(powl_string: str) → POWL[source]#

Parses a POWL model from a string representation of the process model (with the same format as the __repr__ and __str__ methods of the POWL model).

Parameters:

powl_string – POWL model expressed as a string (__repr__ of the POWL model).

Returns:

A POWL object.

Return type:

POWL

import pm4py

powl_model = pm4py.parse_powl_model_string('PO=(nodes={ NODE1, NODE2, NODE3 }, order={ NODE1-->NODE2 })')
print(powl_model)
pm4py.utils.serialize(*args) → Tuple[str, bytes][source]#

Serializes a PM4Py object into a bytes string.

Parameters:

args – PM4Py object(s) to serialize. Supported types include: - An EventLog object. - A Pandas DataFrame object. - A tuple consisting of (PetriNet, Marking, Marking). - A ProcessTree object. - A BPMN object. - A DFG, including the dictionary of directly-follows relations, start activities, and end activities.

Returns:

A tuple containing the serialization type as a string and the serialized bytes.

Return type:

Tuple[str, bytes]

import pm4py

net, im, fm = pm4py.discover_petri_net_inductive(dataframe)
serialization = pm4py.serialize(net, im, fm)
pm4py.utils.deserialize(ser_obj: Tuple[str, bytes]) → Any[source]#

Deserializes a bytes string back into a PM4Py object.

Parameters:

ser_obj – Serialized object as a tuple, consisting of a string indicating the type of the object and a bytes string representing the serialization.

Returns:

The deserialized PM4Py object.

Return type:

Any

import pm4py

net, im, fm = pm4py.discover_petri_net_inductive(dataframe)
serialization = pm4py.serialize(net, im, fm)
net, im, fm = pm4py.deserialize(serialization)
pm4py.utils.get_properties(log, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', resource_key: str = 'org:resource', group_key: str | None = None, start_timestamp_key: str | None = None, **kwargs)[source]#

Retrieves the properties from a log object.

Parameters:

  • log – Log object.

  • activity_key – Attribute to be used for the activity.

  • timestamp_key – Attribute to be used for the timestamp.

  • start_timestamp_key – (Optional) Attribute to be used for the start timestamp.

  • case_id_key – Attribute to be used as case identifier.

  • resource_key – (Optional) Attribute to be used as resource.

  • group_key – (Optional) Attribute to be used as group identifier.

  • kwargs – Additional keyword arguments.

Returns:

A dictionary of properties.

Return type:

Dict

pm4py.utils.set_classifier(log, classifier, classifier_attribute='@@classifier')#

Sets the specified classifier on an existing event log.

Parameters:

  • log – Log object.

  • classifier – Classifier to set. It can be: - A list of event attributes. - A single event attribute. - A classifier stored in the “classifiers” of the log object.

  • classifier_attribute – The attribute of the event that will store the concatenation of the attribute values for the given classifier.

Returns:

The updated log object as an EventLog or Pandas DataFrame.

Return type:

Union[EventLog, pd.DataFrame]

pm4py.utils.parse_event_log_string(traces: Collection[str], sep: str = ',', activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', return_legacy_log_object: bool = False) → EventLog | DataFrame[source]#

Parses a collection of traces expressed as strings (e.g., [“A,B,C,D”, “A,C,B,D”, “A,D”]) into a log object.

Parameters:

  • traces – Collection of traces expressed as strings.

  • sep – Separator used to split the activities in a string trace.

  • activity_key – The attribute to be used as activity.

  • timestamp_key – The attribute to be used as timestamp.

  • case_id_key – The attribute to be used as case identifier.

  • return_legacy_log_object – If True, returns a legacy log object (EventLog). If False, returns a Pandas DataFrame. Default is False.

Returns:

A log object, either as a legacy EventLog or a Pandas DataFrame.

Return type:

Union[EventLog, pd.DataFrame]

import pm4py

dataframe = pm4py.parse_event_log_string(["A,B,C,D", "A,C,B,D", "A,D"])
pm4py.utils.project_on_event_attribute(log: EventLog | DataFrame, attribute_key='concept:name', case_id_key=None) → List[List[str]][source]#

Projects the event log onto a specified event attribute. The result is a list containing a list for each case, where each case is represented as a list of values for the specified attribute.

Example:

`python pm4py.project_on_event_attribute(log, "concept:name") `

Output:

```python [

[‘register request’, ‘examine casually’, ‘check ticket’, ‘decide’, ‘reinitiate request’, ‘examine thoroughly’, ‘check ticket’, ‘decide’, ‘pay compensation’], [‘register request’, ‘check ticket’, ‘examine casually’, ‘decide’, ‘pay compensation’], [‘register request’, ‘examine thoroughly’, ‘check ticket’, ‘decide’, ‘reject request’], [‘register request’, ‘examine casually’, ‘check ticket’, ‘decide’, ‘pay compensation’], [‘register request’, ‘examine casually’, ‘check ticket’, ‘decide’, ‘reinitiate request’, ‘check ticket’, ‘examine casually’, ‘decide’, ‘reinitiate request’, ‘examine casually’, ‘check ticket’, ‘decide’, ‘reject request’], [‘register request’, ‘check ticket’, ‘examine thoroughly’, ‘decide’, ‘reject request’]

]#

param log:

Event log or Pandas DataFrame.

param attribute_key:

The attribute to be used for projection.

param case_id_key:

(Optional) The attribute to be used as case identifier.

return:

A list of lists containing the projected attribute values for each case.

rtype:

List[List[str]]

import pm4py

list_list_activities = pm4py.project_on_event_attribute(dataframe, 'concept:name')
pm4py.utils.sample_cases(log: EventLog | DataFrame, num_cases: int, case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Randomly samples a given number of cases from the event log.

Parameters:

  • log – Event log or Pandas DataFrame.

  • num_cases – Number of cases to sample.

  • case_id_key – Attribute to be used as case identifier.

Returns:

A sampled log object, either as an EventLog or a Pandas DataFrame.

Return type:

Union[EventLog, pd.DataFrame]

import pm4py

sampled_dataframe = pm4py.sample_cases(dataframe, 10, case_id_key='case:concept:name')
pm4py.utils.sample_events(log: EventStream | OCEL, num_events: int) → EventStream | OCEL | DataFrame[source]#

Randomly samples a given number of events from the event log.

Parameters:

  • log – Event stream, OCEL, or Pandas DataFrame.

  • num_events – Number of events to sample.

Returns:

A sampled log object, either as an EventStream, OCEL, or Pandas DataFrame.

Return type:

Union[EventStream, OCEL, pd.DataFrame]

import pm4py

sampled_dataframe = pm4py.sample_events(dataframe, 100)