pm4py package #

Parameters:

log – The log object to convert.
activity_column (str) – The name of the column representing activities.
timestamp_column (str) – 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 (str) – 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.

Return type:

DiGraph

Parameters:

ocel (OCEL) – The object-centric event log to convert.
variant (str) – 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.

Return type:

DiGraph

Parameters:

log – The log object to convert (EventLog, EventStream, or Pandas DataFrame).
include_df (bool) – Whether to include the directly-follows relation in the graph. Defaults to True.
case_id_key (str) – 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 (str) – The attribute to be used as the activity identifier. Defaults to “concept:name”.
timestamp_key (str) – The attribute to be used as the timestamp. Defaults to “time:timestamp”.
case_id_key (str) – The attribute to be used as the case identifier. Defaults to “case:concept:name”.
start_timestamp_key (str) – 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.

Return type:

DiGraph

Parameters:

net (PetriNet) – The Petri net to convert.
im (Marking) – The initial marking of the Petri net.
fm (Marking) – 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 (PetriNet) – The Petri net to convert.
im (Marking) – The initial marking of the Petri net.
fm (Marking) – The final marking of the Petri net.
type (str) – The target Petri net type. Options are “classic”, “reset”, “inhibitor”, “reset_inhibitor”. Defaults to “classic”.

Returns:

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

pm4py.discovery module#

The pm4py.discovery module contains the process discovery algorithms implemented in pm4py.

pm4py.discovery.discover_dfg(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Tuple[dict, dict, dict][source]#

Discovers a Directly-Follows Graph (DFG) from a log.

This method returns a tuple containing: - A dictionary with pairs of directly-following activities as keys and the frequency of the relationship as values. - A dictionary of start activities with their respective frequencies. - A dictionary of end activities with their respective frequencies.

Parameters:

log – Event log or Pandas DataFrame.
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).

Returns:

A tuple of three dictionaries: (dfg, start_activities, end_activities).

Return type:

Tuple[dict, dict, dict]

import pm4py

dfg, start_activities, end_activities = pm4py.discover_dfg(
    dataframe,
    case_id_key='case:concept:name',
    activity_key='concept:name',
    timestamp_key='time:timestamp'
)

pm4py.discovery.discover_directly_follows_graph(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Tuple[dict, dict, dict][source]#

pm4py.discovery.discover_dfg_typed(log: DataFrame, case_id_key: str = 'case:concept:name', activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp') → DirectlyFollowsGraph[source]#

Discovers a typed Directly-Follows Graph (DFG) from a log.

This method returns a typed DFG object, as specified in pm4py.objects.dfg.obj.py (DirectlyFollowsGraph Class). The DFG object includes the graph, start activities, and end activities. - The graph is a collection of triples of the form (a, b, f) representing an arc a->b with frequency f. - The start activities are a collection of tuples of the form (a, f) representing that activity a starts f cases. - The end activities are a collection of tuples of the form (a, f) representing that activity a ends f cases.

This method replaces pm4py.discover_dfg and pm4py.discover_directly_follows_graph. In future releases, these functions will adopt the same behavior as this function.

Parameters:

log (DataFrame) – pandas.DataFrame
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).

Returns:

A typed DFG object containing the graph, start activities, and end activities.

Return type:

DFG

import pm4py

dfg = pm4py.discover_dfg_typed(
    log,
    case_id_key='case:concept:name',
    activity_key='concept:name',
    timestamp_key='time:timestamp'
)

pm4py.discovery.discover_performance_dfg(log: EventLog | DataFrame, business_hours: bool = False, business_hour_slots=[(25200, 61200), (111600, 147600), (198000, 234000), (284400, 320400), (370800, 406800)], workcalendar=None, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Tuple[dict, dict, dict][source]#

Discovers a Performance Directly-Follows Graph from an event log.

This method returns a tuple containing: - A dictionary with pairs of directly-following activities as keys and the performance metrics of the relationship as values. - A dictionary of start activities with their respective frequencies. - A dictionary of end activities with their respective frequencies.

Parameters:

log – Event log or Pandas DataFrame.
business_hours (bool) – Enables or disables computation based on business hours (default: False).
business_hour_slots –
Work schedule of the company, provided as a list of tuples where each tuple represents one time slot of business hours. Each slot consists of a start and end time given in seconds since the week start. Example: ```python [

(7 * 60 * 60, 17 * 60 * 60), # Monday 07:00 - 17:00 ((24 + 7) * 60 * 60, (24 + 12) * 60 * 60), # Tuesday 07:00 - 12:00 ((24 + 13) * 60 * 60, (24 + 17) * 60 * 60) # Tuesday 13:00 - 17:00
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).

Returns:

A tuple of three dictionaries: (performance_dfg, start_activities, end_activities).

Return type:

Tuple[dict, dict, dict]

import pm4py

performance_dfg, start_activities, end_activities = pm4py.discover_performance_dfg(
    dataframe,
    case_id_key='case:concept:name',
    activity_key='concept:name',
    timestamp_key='time:timestamp'
)

pm4py.discovery.discover_petri_net_alpha(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Tuple[PetriNet, Marking, Marking][source]#

Discovers a Petri net using the Alpha Miner.

Parameters:

log – Event log or Pandas DataFrame.
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).

Returns:

A tuple containing the Petri net, initial marking, and final marking.

Return type:

Tuple[PetriNet, Marking, Marking]

import pm4py

net, im, fm = pm4py.discover_petri_net_alpha(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.discovery.discover_petri_net_ilp(log: EventLog | DataFrame, alpha: float = 1.0, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Tuple[PetriNet, Marking, Marking][source]#

Discovers a Petri net using the ILP Miner.

Parameters:

log – Event log or Pandas DataFrame.
alpha (float) – Noise threshold for the sequence encoding graph (1.0=no filtering, 0.0=maximum filtering) (default: 1.0).
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).

Returns:

A tuple containing the Petri net, initial marking, and final marking.

Return type:

Tuple[PetriNet, Marking, Marking]

import pm4py

net, im, fm = pm4py.discover_petri_net_ilp(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.discovery.discover_petri_net_alpha_plus(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Tuple[PetriNet, Marking, Marking][source]#

Discovers a Petri net using the Alpha+ algorithm.

Deprecated since version 2.3.0: This method will be removed in version 3.0.0. Use other discovery methods instead.

Parameters:

log – Event log or Pandas DataFrame.
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).

Returns:

A tuple containing the Petri net, initial marking, and final marking.

Return type:

Tuple[PetriNet, Marking, Marking]

import pm4py

net, im, fm = pm4py.discover_petri_net_alpha_plus(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

Deprecated since version 2.3.0: This will be removed in 3.0.0. This method will be removed in a future release.

pm4py.discovery.discover_petri_net_inductive(log: EventLog | DataFrame | DirectlyFollowsGraph, multi_processing: bool = False, noise_threshold: float = 0.0, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', disable_fallthroughs: bool = False) → Tuple[PetriNet, Marking, Marking][source]#

Discovers a Petri net using the Inductive Miner algorithm.

The Inductive Miner detects a ‘cut’ in the log (e.g., sequential, parallel, concurrent, loop) and recursively applies the algorithm to sublogs until a base case is found. Inductive miner models typically use hidden transitions for skipping or looping portions of the model, and each visible transition has a unique label.

Parameters:

log – Event log, Pandas DataFrame, or typed DFG.
multi_processing (bool) – Enables or disables multiprocessing in the Inductive Miner (default: constants.ENABLE_MULTIPROCESSING_DEFAULT).
noise_threshold (float) – Noise threshold (default: 0.0).
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).
disable_fallthroughs (bool) – Disables the Inductive Miner fall-throughs (default: False).

Returns:

A tuple containing the Petri net, initial marking, and final marking.

Return type:

Tuple[PetriNet, Marking, Marking]

import pm4py

net, im, fm = pm4py.discover_petri_net_inductive(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.discovery.discover_petri_net_heuristics(log: EventLog | DataFrame, dependency_threshold: float = 0.5, and_threshold: float = 0.65, loop_two_threshold: float = 0.5, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Tuple[PetriNet, Marking, Marking][source]#

Discovers a Petri net using the Heuristics Miner.

Heuristics Miner operates on the Directly-Follows Graph, handling noise and identifying common constructs such as dependencies between activities and parallelism. The output is a Heuristics Net, which can then be converted into a Petri net.

Parameters:

log – Event log or Pandas DataFrame.
dependency_threshold (float) – Dependency threshold (default: 0.5).
and_threshold (float) – AND threshold for parallelism (default: 0.65).
loop_two_threshold (float) – Loop two threshold (default: 0.5).
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).

Returns:

A tuple containing the Petri net, initial marking, and final marking.

Return type:

Tuple[PetriNet, Marking, Marking]

import pm4py

net, im, fm = pm4py.discover_petri_net_heuristics(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.discovery.discover_process_tree_inductive(log: EventLog | DataFrame | DirectlyFollowsGraph, noise_threshold: float = 0.0, multi_processing: bool = False, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', disable_fallthroughs: bool = False) → ProcessTree[source]#

Discovers a Process Tree using the Inductive Miner algorithm.

Parameters:

log – Event log, Pandas DataFrame, or typed DFG.
noise_threshold (float) – Noise threshold (default: 0.0).
multi_processing (bool) – Enables or disables multiprocessing in the Inductive Miner (default: constants.ENABLE_MULTIPROCESSING_DEFAULT).
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).
disable_fallthroughs (bool) – Disables the Inductive Miner fall-throughs (default: False).

Returns:

A ProcessTree object.

Return type:

ProcessTree

import pm4py

process_tree = pm4py.discover_process_tree_inductive(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.discovery.discover_heuristics_net(log: EventLog | DataFrame, dependency_threshold: float = 0.5, and_threshold: float = 0.65, loop_two_threshold: float = 0.5, min_act_count: int = 1, min_dfg_occurrences: int = 1, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', decoration: str = 'frequency') → HeuristicsNet[source]#

Discovers a Heuristics Net.

Parameters:

log – Event log or Pandas DataFrame.
dependency_threshold (float) – Dependency threshold (default: 0.5).
and_threshold (float) – AND threshold for parallelism (default: 0.65).
loop_two_threshold (float) – Loop two threshold (default: 0.5).
min_act_count (int) – Minimum number of occurrences per activity to be included in the discovery (default: 1).
min_dfg_occurrences (int) – Minimum number of occurrences per arc in the DFG to be included in the discovery (default: 1).
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).
decoration (str) – The decoration to be used (“frequency” or “performance”) (default: “frequency”).

Returns:

A HeuristicsNet object.

Return type:

HeuristicsNet

import pm4py

heu_net = pm4py.discover_heuristics_net(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.discovery.derive_minimum_self_distance(log: DataFrame | EventLog | EventStream, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Dict[str, int][source]#

Computes the minimum self-distance for each activity observed in an event log.

The self-distance of activity a in <a> is infinity, in <a, a> is 0, in <a, b, a> is 1, etc. The activity key ‘concept:name’ is used.

Parameters:

log – Event log or Pandas DataFrame.
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).

Returns:

A dictionary mapping each activity to its minimum self-distance.

Return type:

Dict[str, int]

import pm4py

msd = pm4py.derive_minimum_self_distance(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.discovery.discover_footprints(*args: EventLog | Tuple[PetriNet, Marking, Marking] | ProcessTree) → List[Dict[str, Any]] | Dict[str, Any][source]#

Discovers the footprints from the provided event log or process model.

Footprints are a high-level representation of the behavior captured in the event log or process model.

Parameters:: args – Event log, process model (Petri net and markings), or ProcessTree.
Returns:: A list of footprint dictionaries or a single footprint dictionary.
Return type:: Union[List[Dict[str, Any]], Dict[str, Any]]

import pm4py

footprints = pm4py.discover_footprints(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.discovery.discover_eventually_follows_graph(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Dict[Tuple[str, str], int][source]#

Generates the Eventually-Follows Graph from a log.

The Eventually-Follows Graph is a dictionary that maps each pair of activities to the number of times one activity eventually follows the other in the log.

Parameters:

log – Event log or Pandas DataFrame.
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).

Returns:

A dictionary mapping each pair of activities to the count of their eventually-follows relationship.

Return type:

Dict[Tuple[str, str], int]

import pm4py

efg = pm4py.discover_eventually_follows_graph(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.discovery.discover_bpmn_inductive(log: EventLog | DataFrame | DirectlyFollowsGraph, noise_threshold: float = 0.0, multi_processing: bool = False, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', disable_fallthroughs: bool = False) → BPMN[source]#

Discovers a BPMN model using the Inductive Miner algorithm.

Parameters:

log – Event log, Pandas DataFrame, or typed DFG.
noise_threshold (float) – Noise threshold (default: 0.0).
multi_processing (bool) – Enables or disables multiprocessing in the Inductive Miner (default: constants.ENABLE_MULTIPROCESSING_DEFAULT).
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).
disable_fallthroughs (bool) – Disables the Inductive Miner fall-throughs (default: False).

Returns:

A BPMN object representing the discovered BPMN model.

Return type:

BPMN

import pm4py

bpmn_graph = pm4py.discover_bpmn_inductive(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.discovery.discover_transition_system(log: EventLog | DataFrame, direction: str = 'forward', window: int = 2, view: str = 'sequence', activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → TransitionSystem[source]#

Discovers a Transition System from a log.

The Transition System is built based on the specified direction, window size, and view. It captures the transitions between states of activity sequences.

Parameters:

log – Event log or Pandas DataFrame.
direction (str) – Direction in which the transition system is built (“forward” or “backward”) (default: “forward”).
window (int) – Window size for state construction (e.g., 2, 3) (default: 2).
view (str) – View to use in the construction of the states (“sequence”, “set”, “multiset”) (default: “sequence”).
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).

Returns:

A TransitionSystem object representing the discovered transition system.

Return type:

TransitionSystem

import pm4py

transition_system = pm4py.discover_transition_system(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.discovery.discover_prefix_tree(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Trie[source]#

Discovers a Prefix Tree from the provided log.

A Prefix Tree represents all the unique prefixes of activity sequences in the log.

Parameters:

log – Event log or Pandas DataFrame.
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).

Returns:

A Trie object representing the discovered prefix tree.

Return type:

Trie

import pm4py

prefix_tree = pm4py.discover_prefix_tree(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.discovery.discover_temporal_profile(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Dict[Tuple[str, str], Tuple[float, float]][source]#

Discovers a Temporal Profile from a log.

Implements the approach described in: Stertz, Florian, Jürgen Mangler, and Stefanie Rinderle-Ma. “Temporal Conformance Checking at Runtime based on Time-infused Process Models.” arXiv preprint arXiv:2008.07262 (2020).

The output is a dictionary containing, for every pair of activities that eventually follow each other in at least one case of the log, the average and the standard deviation of the time difference between their timestamps.

Example: If the log has two cases: - Case 1: A (timestamp: 1980-01) → B (timestamp: 1980-03) → C (timestamp: 1980-06) - Case 2: A (timestamp: 1990-01) → B (timestamp: 1990-02) → D (timestamp: 1990-03)

The returned dictionary will contain: ``` {

(‘A’, ‘B’): (1.5 months, 0.5 months), (‘A’, ‘C’): (5 months, 0), (‘A’, ‘D’): (2 months, 0)

}#

type case_id_key:: str
type timestamp_key:: str
type activity_key:: str
param log:: Event log or Pandas DataFrame.
param activity_key:: Attribute to be used for the activity (default: “concept:name”).
param timestamp_key:: Attribute to be used for the timestamp (default: “time:timestamp”).
param case_id_key:: Attribute to be used as case identifier (default: “case:concept:name”).
return:: A dictionary mapping each pair of activities to a tuple of (average time difference, standard deviation).
rtype:: Dict[Tuple[str, str], Tuple[float, float]]

import pm4py

temporal_profile = pm4py.discover_temporal_profile(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.discovery.discover_log_skeleton(log: EventLog | DataFrame, noise_threshold: float = 0.0, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Dict[str, Any][source]#

Discovers a Log Skeleton from an event log.

A Log Skeleton is a declarative model consisting of six different constraints: - directly_follows: Specifies strict bounds on activities that directly follow each other. Example: ‘A should be directly followed by B’ and ‘B should be directly followed by C’. - always_before: Specifies that some activities may only be executed if certain other activities have been executed earlier in the case. Example: ‘C should always be preceded by A’. - always_after: Specifies that certain activities should always trigger the execution of some other activities later in the case. Example: ‘A should always be followed by C’. - equivalence: Specifies that a given pair of activities should occur the same number of times within a case. Example: ‘B and C should always occur the same number of times’. - never_together: Specifies that a given pair of activities should never occur together in a case. Example: ‘There should be no case containing both C and D’. - activ_occurrences: Specifies allowed numbers of occurrences per activity. Example: ‘Activity A can occur 1 or 2 times, and Activity B can occur 1 to 4 times’.

Reference paper: Verbeek, H. M. W., and R. Medeiros de Carvalho. “Log skeletons: A classification approach to process discovery.” arXiv preprint arXiv:1806.08247 (2018).

Parameters:

log – Event log or Pandas DataFrame.
noise_threshold (float) – Noise threshold influencing the strictness of constraints (default: 0.0).
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).

Returns:

A dictionary representing the Log Skeleton with various constraints.

Return type:

Dict[str, Any]

import pm4py

log_skeleton = pm4py.discover_log_skeleton(
    dataframe,
    noise_threshold=0.1,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.discovery.discover_declare(log: EventLog | DataFrame, allowed_templates: Set[str] | None = None, considered_activities: Set[str] | None = None, min_support_ratio: float | None = None, min_confidence_ratio: float | None = None, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Dict[str, Dict[Any, Dict[str, int]]][source]#

Discovers a DECLARE model from an event log.

Reference paper: F. M. Maggi, A. J. Mooij and W. M. P. van der Aalst, “User-guided discovery of declarative process models,” 2011 IEEE Symposium on Computational Intelligence and Data Mining (CIDM), Paris, France, 2011, pp. 192-199, doi: 10.1109/CIDM.2011.5949297.

Parameters:

log – Event log or Pandas DataFrame.
allowed_templates – (Optional) Set of DECLARE templates to consider for discovery.
considered_activities – (Optional) Set of activities to consider for discovery.
min_support_ratio – (Optional) Minimum percentage of cases for which the discovered rules apply.
min_confidence_ratio – (Optional) Minimum percentage of cases for which the discovered rules are valid, based on the rule’s support.
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).

Returns:

A dictionary representing the discovered DECLARE model with constraints and their parameters.

Return type:

Dict[str, Dict[Any, Dict[str, int]]]

import pm4py

declare_model = pm4py.discover_declare(log)

pm4py.discovery.discover_powl(log: EventLog | DataFrame, variant=None, filtering_weight_factor: float = 0.0, order_graph_filtering_threshold: float = None, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → POWL[source]#

Discovers a POWL (Partially Ordered Workflow Language) model from an event log.

Reference paper: Kourani, Humam, and Sebastiaan J. van Zelst. “POWL: partially ordered workflow language.” International Conference on Business Process Management. Cham: Springer Nature Switzerland, 2023.

Parameters:

log – Event log or Pandas DataFrame.
variant – Variant of the POWL discovery algorithm to use.
filtering_weight_factor (float) – Factoring threshold for filtering weights, accepts values 0 <= x < 1 (default: 0.0).
order_graph_filtering_threshold (float) – Filtering threshold for the order graph, valid for the DYNAMIC_CLUSTERING variant, accepts values 0.5 < x <= 1 (default: None).
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).

Returns:

A POWL object representing the discovered POWL model.

Return type:

POWL

import pm4py

log = pm4py.read_xes('tests/input_data/receipt.xes')
powl_model = pm4py.discover_powl(
    log,
    activity_key='concept:name'
)
print(powl_model)

pm4py.discovery.discover_batches(log: EventLog | DataFrame, merge_distance: int = 900, min_batch_size: int = 2, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', resource_key: str = 'org:resource') → List[Tuple[Tuple[str, str], int, Dict[str, Any]]][source]#

Discovers batches from the provided log.

An activity is executed in batches by a given resource when the resource performs the same activity multiple times in a short period. Identifying such activities may highlight repetitive tasks that could be automated.

The following batch categories are detected: - Simultaneous: All events in the batch have identical start and end timestamps. - Batching at Start: All events in the batch have identical start timestamps. - Batching at End: All events in the batch have identical end timestamps. - Sequential Batching: Consecutive events have the end of the first equal to the start of the second. - Concurrent Batching: Consecutive events that do not match sequentially.

Reference paper: Martin, N., Swennen, M., Depaire, B., Jans, M., Caris, A., & Vanhoof, K. (2015, December). Batch Processing: Definition and Event Log Identification. In SIMPDA (pp. 137-140).

Parameters:

log – Event log or Pandas DataFrame.
merge_distance (int) – Maximum time distance (in seconds) between non-overlapping intervals to consider them part of the same batch (default: 900 seconds, i.e., 15 minutes).
min_batch_size (int) – Minimum number of events required to form a batch (default: 2).
activity_key (str) – Attribute to be used for the activity (default: “concept:name”).
timestamp_key (str) – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key (str) – Attribute to be used as case identifier (default: “case:concept:name”).
resource_key (str) – Attribute to be used as resource (default: “org:resource”).

Returns:

A sorted list of tuples, each containing: - The (activity, resource) pair. - The number of batches for the given activity-resource. - A dictionary with batch details.

Return type:

List[Tuple[Tuple[str, str], int, Dict[str, Any]]]

import pm4py

batches = pm4py.discover_batches(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp',
    resource_key='org:resource'
)

pm4py.filtering module#

The pm4py.filtering module contains the filtering features offered in pm4py.

pm4py.filtering.filter_log_relative_occurrence_event_attribute(log: EventLog | DataFrame, min_relative_stake: float, attribute_key: str = 'concept:name', level: str = 'cases', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Filters the event log, keeping only the events that have an attribute value which occurs: - in at least the specified (min_relative_stake) percentage of events when level=”events”, - in at least the specified (min_relative_stake) percentage of cases when level=”cases”.

Parameters:

log – Event log or Pandas DataFrame.
min_relative_stake (float) – Minimum percentage of cases (expressed as a number between 0 and 1) in which the attribute should occur.
attribute_key (str) – The attribute to filter.
level (str) – The level of the filter (if level=”events”, then events; if level=”cases”, then cases).
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_log_relative_occurrence_event_attribute(
    dataframe,
    0.5,
    attribute_key='concept:name',
    level='cases',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.filtering.filter_start_activities(log: EventLog | DataFrame, activities: Set[str] | List[str], retain: bool = True, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Filters cases that have a start activity in the provided list.

Parameters:

log – Event log or Pandas DataFrame.
activities – Collection of start activities.
retain (bool) – If True, retains the traces containing the given start activities; if False, drops the traces.
activity_key (str) – Attribute to be used for the activity.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_start_activities(
    dataframe,
    ['Act. A'],
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.filtering.filter_end_activities(log: EventLog | DataFrame, activities: Set[str] | List[str], retain: bool = True, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Filters cases that have an end activity in the provided list.

Parameters:

log – Event log or Pandas DataFrame.
activities – Collection of end activities.
retain (bool) – If True, retains the traces containing the given end activities; if False, drops the traces.
activity_key (str) – Attribute to be used for the activity.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_end_activities(
    dataframe,
    ['Act. Z'],
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.filtering.filter_event_attribute_values(log: EventLog | DataFrame, attribute_key: str, values: Set[str] | List[str], level: str = 'case', retain: bool = True, case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Filters a log object based on the values of a specified event attribute.

Parameters:

log – Event log or Pandas DataFrame.
attribute_key (str) – Attribute to filter.
values – Admitted or forbidden values.
level (str) – Specifies how the filter should be applied (‘case’ filters the cases where at least one occurrence happens; ‘event’ filters the events, potentially trimming the cases).
retain (bool) – Specifies if the values should be kept or removed.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_event_attribute_values(
    dataframe,
    'concept:name',
    ['Act. A', 'Act. Z'],
    case_id_key='case:concept:name'
)

pm4py.filtering.filter_trace_attribute_values(log: EventLog | DataFrame, attribute_key: str, values: Set[str] | List[str], retain: bool = True, case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Filters a log based on the values of a specified trace attribute.

Parameters:

log – Event log or Pandas DataFrame.
attribute_key (str) – Attribute to filter.
values – Collection of values to filter.
retain (bool) – Boolean value indicating whether to keep or discard matching traces.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_trace_attribute_values(
    dataframe,
    'case:creator',
    ['Mike'],
    case_id_key='case:concept:name'
)

pm4py.filtering.filter_variants(log: EventLog | DataFrame, variants: Set[str] | List[str] | List[Tuple[str]], retain: bool = True, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Filters a log based on a specified set of variants.

Parameters:

log – Event log or Pandas DataFrame.
variants – Collection of variants to filter. A variant should be specified as a list of tuples of activity names, e.g., [(‘a’, ‘b’, ‘c’)].
retain (bool) – Boolean indicating whether to retain (if True) or remove (if False) traces conforming to the specified variants.
activity_key (str) – Attribute to be used for the activity.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_variants(
    dataframe,
    [('Act. A', 'Act. B', 'Act. Z'), ('Act. A', 'Act. C', 'Act. Z')],
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.filtering.filter_directly_follows_relation(log: EventLog | DataFrame, relations: List[str], retain: bool = True, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Retains traces that contain any of the specified ‘directly follows’ relations. For example, if relations == [(‘a’,’b’),(‘a’,’c’)] and log [<a,b,c>,<a,c,b>,<a,d,b>], the resulting log will contain traces describing [<a,b,c>,<a,c,b>].

Parameters:

log – Event log or Pandas DataFrame.
relations – List of activity name pairs, representing allowed or forbidden paths.
retain (bool) – Boolean indicating whether the paths should be kept (if True) or removed (if False).
activity_key (str) – Attribute to be used for the activity.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_directly_follows_relation(
    dataframe,
    [('A', 'B'), ('A', 'C')],
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.filtering.filter_eventually_follows_relation(log: EventLog | DataFrame, relations: List[str], retain: bool = True, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Retains traces that contain any of the specified ‘eventually follows’ relations. For example, if relations == [(‘a’,’b’),(‘a’,’c’)] and log [<a,b,c>,<a,c,b>,<a,d,b>], the resulting log will contain traces describing [<a,b,c>,<a,c,b>,<a,d,b>].

Parameters:

log – Event log or Pandas DataFrame.
relations – List of activity name pairs, representing allowed or forbidden paths.
retain (bool) – Boolean indicating whether the paths should be kept (if True) or removed (if False).
activity_key (str) – Attribute to be used for the activity.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_eventually_follows_relation(
    dataframe,
    [('A', 'B'), ('A', 'C')],
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.filtering.filter_time_range(log: EventLog | DataFrame, dt1: str, dt2: str, mode: str = 'events', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Filters a log based on a time interval.

Parameters:

log – Event log or Pandas DataFrame.
dt1 (str) – Left extreme of the interval.
dt2 (str) – Right extreme of the interval.
mode (str) – Modality of filtering (‘events’, ‘traces_contained’, ‘traces_intersecting’). - ‘events’: Any event that fits the time frame is retained. - ‘traces_contained’: Any trace completely contained in the timeframe is retained. - ‘traces_intersecting’: Any trace intersecting with the timeframe is retained.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe1 = pm4py.filter_time_range(
    dataframe,
    '2010-01-01 00:00:00',
    '2011-01-01 00:00:00',
    mode='traces_contained',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)
filtered_dataframe2 = pm4py.filter_time_range(
    dataframe,
    '2010-01-01 00:00:00',
    '2011-01-01 00:00:00',
    mode='traces_intersecting',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)
filtered_dataframe3 = pm4py.filter_time_range(
    dataframe,
    '2010-01-01 00:00:00',
    '2011-01-01 00:00:00',
    mode='events',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.filtering.filter_between(log: EventLog | DataFrame, act1: str | List[str], act2: str | List[str], activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Finds all the sub-cases leading from an event with activity “act1” to an event with activity “act2” in the log, and returns a log containing only them.

Example:

Log A B C D E F A B E F C A B F C B C B E F C

act1 = B act2 = C

Returned sub-cases: B C (from the first case) B E F C (from the second case) B F C (from the third case) B C (from the third case) B E F C (from the third case)

Parameters:

log – Event log or Pandas DataFrame.
act1 – Source activity or collection of activities.
act2 – Target activity or collection of activities.
activity_key (str) – Attribute to be used for the activity.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_between(
    dataframe,
    'A',
    'D',
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.filtering.filter_case_size(log: EventLog | DataFrame, min_size: int, max_size: int, case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Filters the event log, keeping cases that have a length (number of events) between min_size and max_size.

Parameters:

log – Event log or Pandas DataFrame.
min_size (int) – Minimum allowed number of events.
max_size (int) – Maximum allowed number of events.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_case_size(
    dataframe,
    5,
    10,
    case_id_key='case:concept:name'
)

pm4py.filtering.filter_case_performance(log: EventLog | DataFrame, min_performance: float, max_performance: float, timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Filters the event log, keeping cases that have a duration (the timestamp of the last event minus the timestamp of the first event) between min_performance and max_performance.

Parameters:

log – Event log or Pandas DataFrame.
min_performance (float) – Minimum allowed case duration.
max_performance (float) – Maximum allowed case duration.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_case_performance(
    dataframe,
    3600.0,
    86400.0,
    timestamp_key='time:timestamp',
    case_id_key='case:concept:name'
)

pm4py.filtering.filter_activities_rework(log: EventLog | DataFrame, activity: str, min_occurrences: int = 2, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Filters the event log, keeping cases where the specified activity occurs at least min_occurrences times.

Parameters:

log – Event log or Pandas DataFrame.
activity (str) – Activity to consider.
min_occurrences (int) – Minimum desired number of occurrences.
activity_key (str) – Attribute to be used for the activity.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_activities_rework(
    dataframe,
    'Approve Order',
    2,
    activity_key='concept:name',
    timestamp_key='time:timestamp',
    case_id_key='case:concept:name'
)

pm4py.filtering.filter_paths_performance(log: EventLog | DataFrame, path: Tuple[str, str], min_performance: float, max_performance: float, keep: bool = True, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Filters the event log based on the performance of specified paths.

If keep=True, retains cases having the specified path (tuple of 2 activities) with a duration between min_performance and max_performance.
If keep=False, discards cases having the specified path with a duration between min_performance and max_performance.

Parameters:

log – Event log or Pandas DataFrame.
path – Tuple of two activities (source_activity, target_activity).
min_performance (float) – Minimum allowed performance of the path.
max_performance (float) – Maximum allowed performance of the path.
keep (bool) – Boolean indicating whether to keep (if True) or discard (if False) the cases with the specified performance.
activity_key (str) – Attribute to be used for the activity.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_paths_performance(
    dataframe,
    ('A', 'D'),
    3600.0,
    86400.0,
    activity_key='concept:name',
    timestamp_key='time:timestamp',
    case_id_key='case:concept:name'
)

pm4py.filtering.filter_variants_top_k(log: EventLog | DataFrame, k: int, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Keeps the top-k variants of the log.

Parameters:

log – Event log or Pandas DataFrame.
k (int) – Number of variants to keep.
activity_key (str) – Attribute to be used for the activity.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_variants_top_k(
    dataframe,
    5,
    activity_key='concept:name',
    timestamp_key='time:timestamp',
    case_id_key='case:concept:name'
)

pm4py.filtering.filter_variants_by_coverage_percentage(log: EventLog | DataFrame, min_coverage_percentage: float, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Filters the variants of the log based on a coverage percentage. For example, if min_coverage_percentage=0.4 and the log has 1000 cases with: - 500 cases of variant 1, - 400 cases of variant 2, - 100 cases of variant 3, the filter keeps only the traces of variant 1 and variant 2.

Parameters:

log – Event log or Pandas DataFrame.
min_coverage_percentage (float) – Minimum allowed percentage of coverage.
activity_key (str) – Attribute to be used for the activity.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_variants_by_coverage_percentage(
    dataframe,
    0.1,
    activity_key='concept:name',
    timestamp_key='time:timestamp',
    case_id_key='case:concept:name'
)

pm4py.filtering.filter_prefixes(log: EventLog | DataFrame, activity: str, strict: bool = True, first_or_last: str = 'first', activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Filters the log, keeping the prefixes leading up to a given activity. For example, for a log with traces: - A,B,C,D - A,B,Z,A,B,C,D - A,B,C,D,C,E,C,F

The prefixes to “C” are respectively: - A,B - A,B,Z,A,B - A,B

Parameters:

log – Event log or Pandas DataFrame.
activity (str) – Target activity for the filter.
strict (bool) – Applies the filter strictly, cutting the occurrences of the selected activity.
first_or_last (str) – Decides if the first or last occurrence of an activity should be selected as the baseline for the filter.
activity_key (str) – Attribute to be used for the activity.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_prefixes(
    dataframe,
    'Act. C',
    activity_key='concept:name',
    timestamp_key='time:timestamp',
    case_id_key='case:concept:name'
)

pm4py.filtering.filter_suffixes(log: EventLog | DataFrame, activity: str, strict: bool = True, first_or_last: str = 'first', activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Filters the log, keeping the suffixes starting from a given activity. For example, for a log with traces: - A,B,C,D - A,B,Z,A,B,C,D - A,B,C,D,C,E,C,F

The suffixes from “C” are respectively: - D - D - D,C,E,C,F

Parameters:

log – Event log or Pandas DataFrame.
activity (str) – Target activity for the filter.
strict (bool) – Applies the filter strictly, cutting the occurrences of the selected activity.
first_or_last (str) – Decides if the first or last occurrence of an activity should be selected as the baseline for the filter.
activity_key (str) – Attribute to be used for the activity.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_suffixes(
    dataframe,
    'Act. C',
    activity_key='concept:name',
    timestamp_key='time:timestamp',
    case_id_key='case:concept:name'
)

pm4py.filtering.filter_ocel_event_attribute(ocel: OCEL, attribute_key: str, attribute_values: Collection[Any], positive: bool = True) → OCEL[source]#

Filters the object-centric event log based on the provided event attribute values.

Return type:

Parameters:

ocel (OCEL) – Object-centric event log.
attribute_key (str) – Attribute at the event level to filter.
attribute_values – Collection of attribute values to keep or remove.
positive (bool) – Determines whether the values should be kept (True) or removed (False).

Returns:

Filtered OCEL.

import pm4py

filtered_ocel = pm4py.filter_ocel_event_attribute(
    ocel,
    'ocel:activity',
    ['A', 'B', 'D']
)

pm4py.filtering.filter_ocel_object_attribute(ocel: OCEL, attribute_key: str, attribute_values: Collection[Any], positive: bool = True) → OCEL[source]#

Filters the object-centric event log based on the provided object attribute values.

Return type:

Parameters:

ocel (OCEL) – Object-centric event log.
attribute_key (str) – Attribute at the object level to filter.
attribute_values – Collection of attribute values to keep or remove.
positive (bool) – Determines whether the values should be kept (True) or removed (False).

Returns:

Filtered OCEL.

import pm4py

filtered_ocel = pm4py.filter_ocel_object_attribute(
    ocel,
    'ocel:type',
    ['order']
)

pm4py.filtering.filter_ocel_object_types_allowed_activities(ocel: OCEL, correspondence_dict: Dict[str, Collection[str]]) → OCEL[source]#

Filters an object-centric event log, keeping only the specified object types with the specified set of allowed activities.

Return type:

Parameters:

ocel (OCEL) – Object-centric event log.
correspondence_dict – Dictionary containing, for every object type of interest, a collection of allowed activities. Example: {“order”: [“Create Order”], “element”: [“Create Order”, “Create Delivery”]}.

Returns:

Filtered OCEL.

import pm4py

filtered_ocel = pm4py.filter_ocel_object_types_allowed_activities(
    ocel,
    {'order': ['create order', 'pay order'], 'item': ['create item', 'deliver item']}
)

pm4py.filtering.filter_ocel_object_per_type_count(ocel: OCEL, min_num_obj_type: Dict[str, int]) → OCEL[source]#

Filters the events of the object-centric logs that are related to at least the specified number of objects per type.

Example: pm4py.filter_object_per_type_count(ocel, {“order”: 1, “element”: 2})

Would keep the following events:

ocel:eid ocel:timestamp ocel:activity ocel:type:element ocel:type:order

0 e1 1980-01-01 Create Order [i4, i1, i3, i2] [o1] 1 e11 1981-01-01 Create Order [i6, i5] [o2] 2 e14 1981-01-04 Create Order [i8, i7] [o3]

Return type:

Parameters:

ocel (OCEL) – Object-centric event log.
min_num_obj_type – Minimum number of objects per type.

Returns:

Filtered OCEL.

import pm4py

filtered_ocel = pm4py.filter_ocel_object_per_type_count(
    ocel,
    {'order': 1, 'element': 2}
)

pm4py.filtering.filter_ocel_start_events_per_object_type(ocel: OCEL, object_type: str) → OCEL[source]#

Filters the events in which a new object of the given object type is spawned. For example, an event with activity “Create Order” might spawn new orders.

Return type:

Parameters:

ocel (OCEL) – Object-centric event log.
object_type (str) – Object type to consider.

Returns:

Filtered OCEL.

import pm4py

filtered_ocel = pm4py.filter_ocel_start_events_per_object_type(
    ocel,
    'delivery'
)

pm4py.filtering.filter_ocel_end_events_per_object_type(ocel: OCEL, object_type: str) → OCEL[source]#

Filters the events in which an object of the given object type terminates its lifecycle. For example, an event with activity “Pay Order” might terminate an order.

Return type:

Parameters:

ocel (OCEL) – Object-centric event log.
object_type (str) – Object type to consider.

Returns:

Filtered OCEL.

import pm4py

filtered_ocel = pm4py.filter_ocel_end_events_per_object_type(
    ocel,
    'delivery'
)

pm4py.filtering.filter_ocel_events_timestamp(ocel: OCEL, min_timest: datetime | str, max_timest: datetime | str, timestamp_key: str = 'ocel:timestamp') → OCEL[source]#

Filters the object-centric event log, keeping events within the provided timestamp range.

Return type:

Parameters:

ocel (OCEL) – Object-centric event log.
min_timest – Left extreme of the allowed timestamp interval (format: YYYY-mm-dd HH:MM:SS).
max_timest – Right extreme of the allowed timestamp interval (format: YYYY-mm-dd HH:MM:SS).
timestamp_key (str) – The attribute to use as timestamp (default: ocel:timestamp).

Returns:

Filtered OCEL.

import pm4py

filtered_ocel = pm4py.filter_ocel_events_timestamp(
    ocel,
    '1990-01-01 00:00:00',
    '2010-01-01 00:00:00'
)

pm4py.filtering.filter_four_eyes_principle(log: EventLog | DataFrame, activity1: str, activity2: str, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', resource_key: str = 'org:resource', keep_violations: bool = False) → EventLog | DataFrame[source]#

Filters out the cases of the log that violate the four-eyes principle on the provided activities.

Parameters:

log – Event log or Pandas DataFrame.
activity1 (str) – First activity.
activity2 (str) – Second activity.
activity_key (str) – Attribute to be used for the activity.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.
resource_key (str) – Attribute to be used as resource.
keep_violations (bool) – Boolean indicating whether to discard (if False) or retain (if True) the violations.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_four_eyes_principle(
    dataframe,
    'Act. A',
    'Act. B',
    activity_key='concept:name',
    resource_key='org:resource',
    timestamp_key='time:timestamp',
    case_id_key='case:concept:name'
)

pm4py.filtering.filter_activity_done_different_resources(log: EventLog | DataFrame, activity: str, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', resource_key: str = 'org:resource', keep_violations: bool = True) → EventLog | DataFrame[source]#

Filters the cases where an activity is performed by different resources multiple times.

Parameters:

log – Event log or Pandas DataFrame.
activity (str) – Activity to consider.
activity_key (str) – Attribute to be used for the activity.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.
resource_key (str) – Attribute to be used as resource.
keep_violations (bool) – Boolean indicating whether to discard (if False) or retain (if True) the violations.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

filtered_dataframe = pm4py.filter_activity_done_different_resources(
    dataframe,
    'Act. A',
    activity_key='concept:name',
    resource_key='org:resource',
    timestamp_key='time:timestamp',
    case_id_key='case:concept:name'
)

pm4py.filtering.filter_trace_segments(log: EventLog | DataFrame, admitted_traces: List[List[str]], positive: bool = True, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Filters an event log based on a set of trace segments. A trace is a sequence of activities and “…” where: - “…” before an activity indicates that other activities can precede the given activity. - “…” after an activity indicates that other activities can follow the given activity.

Examples: - pm4py.filter_trace_segments(log, [[“A”, “B”]]) retains only cases with the exact process variant A,B. - pm4py.filter_trace_segments(log, [[”…”, “A”, “B”]]) retains only cases ending with activities A,B. - pm4py.filter_trace_segments(log, [[“A”, “B”, “…”]]) retains only cases starting with activities A,B. - pm4py.filter_trace_segments(log, [[”…”, “A”, “B”, “C”, “…”], [”…”, “D”, “E”, “F”, “…”]]) retains cases where:

At any point, there is A followed by B followed by C,

And at any other point, there is D followed by E followed by F.

Parameters:

log – Event log or Pandas DataFrame.
admitted_traces – Collection of trace segments to admit based on the criteria above.
positive (bool) – Boolean indicating whether to keep (if True) or discard (if False) the cases satisfying the filter.
activity_key (str) – Attribute to be used for the activity.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

Returns:

Filtered event log or Pandas DataFrame.

import pm4py

log = pm4py.read_xes("tests/input_data/running-example.xes")

filtered_log = pm4py.filter_trace_segments(
    log,
    [["...", "check ticket", "decide", "reinitiate request", "..."]]
)
print(filtered_log)

pm4py.filtering.filter_ocel_object_types(ocel: OCEL, obj_types: Collection[str], positive: bool = True, level: int = 1) → OCEL[source]#

Filters the object types of an object-centric event log.

Return type:

Parameters:

ocel (OCEL) – Object-centric event log.
obj_types – Object types to keep or remove.
positive (bool) – Boolean indicating whether to keep (True) or remove (False) the specified object types.
level (int) – Recursively expands the set of object identifiers until the specified level.

Returns:

Filtered OCEL.

import pm4py

ocel = pm4py.read_ocel('log.jsonocel')
filtered_ocel = pm4py.filter_ocel_object_types(
    ocel,
    ['order']
)

pm4py.filtering.filter_ocel_objects(ocel: OCEL, object_identifiers: Collection[str], positive: bool = True, level: int = 1) → OCEL[source]#

Filters the object identifiers of an object-centric event log.

Return type:

Parameters:

ocel (OCEL) – Object-centric event log.
object_identifiers – Object identifiers to keep or remove.
positive (bool) – Boolean indicating whether to keep (True) or remove (False) the specified object identifiers.
level (int) – Recursively expands the set of object identifiers until the specified level.

Returns:

Filtered OCEL.

import pm4py

ocel = pm4py.read_ocel('log.jsonocel')
filtered_ocel = pm4py.filter_ocel_objects(
    ocel,
    ['o1'],
    level=1
)

pm4py.filtering.filter_ocel_events(ocel: OCEL, event_identifiers: Collection[str], positive: bool = True) → OCEL[source]#

Filters the event identifiers of an object-centric event log.

Return type:

Parameters:

ocel (OCEL) – Object-centric event log.
event_identifiers – Event identifiers to keep or remove.
positive (bool) – Boolean indicating whether to keep (True) or remove (False) the specified event identifiers.

Returns:

Filtered OCEL.

import pm4py

ocel = pm4py.read_ocel('log.jsonocel')
filtered_ocel = pm4py.filter_ocel_events(
    ocel,
    ['e1']
)

pm4py.filtering.filter_ocel_activities_connected_object_type(ocel: OCEL, object_type: str) → OCEL[source]#

Filter an OCEL on the set of activities executed on objects of the given object type.

Parameters:

ocel (OCEL) – object-centric event log
object_type (str) – object type

Return type:

OCEL

import pm4py

ocel = pm4py.read_ocel2("tests/input_data/ocel/ocel20_example.xmlocel")
filtered_ocel = pm4py.filter_ocel_activities_connected_object_type(ocel, "Purchase Order")
print(filtered_ocel)

pm4py.filtering.filter_ocel_cc_object(ocel: OCEL, object_id: str, conn_comp: List[List[str]] | None = None, return_conn_comp: bool = False) → OCEL | Tuple[OCEL, List[List[str]]][source]#

Returns the connected component of the object-centric event log to which the specified object belongs.

Parameters:

ocel (OCEL) – Object-centric event log.
object_id (str) – Object identifier.
conn_comp – (Optional) Precomputed connected components of the OCEL objects.
return_conn_comp (bool) – If True, returns the filtered OCEL along with the computed connected components.

Returns:

Filtered OCEL, optionally with the list of connected components.

import pm4py

ocel = pm4py.read_ocel('log.jsonocel')
filtered_ocel = pm4py.filter_ocel_cc_object(
    ocel,
    'order1'
)

pm4py.filtering.filter_ocel_cc_length(ocel: OCEL, min_cc_length: int, max_cc_length: int) → OCEL[source]#

Keeps only the objects in an OCEL belonging to a connected component with a length falling within the specified range.

Reference: Adams, Jan Niklas, et al. “Defining cases and variants for object-centric event data.” 2022 4th International Conference on Process Mining (ICPM). IEEE, 2022.

Return type:

Parameters:

ocel (OCEL) – Object-centric event log.
min_cc_length (int) – Minimum allowed length for the connected component.
max_cc_length (int) – Maximum allowed length for the connected component.

Returns:

Filtered OCEL.

import pm4py

filtered_ocel = pm4py.filter_ocel_cc_length(
    ocel,
    2,
    10
)

pm4py.filtering.filter_ocel_cc_otype(ocel: OCEL, otype: str, positive: bool = True) → OCEL[source]#

Filters the objects belonging to connected components that have at least one object of the specified type.

Reference: Adams, Jan Niklas, et al. “Defining cases and variants for object-centric event data.” 2022 4th International Conference on Process Mining (ICPM). IEEE, 2022.

Return type:

Parameters:

ocel (OCEL) – Object-centric event log.
otype (str) – Object type to consider.
positive (bool) – Boolean indicating whether to keep (True) or discard (False) the objects in these components.

Returns:

Filtered OCEL.

import pm4py

ocel = pm4py.read_ocel('log.jsonocel')
filtered_ocel = pm4py.filter_ocel_cc_otype(
    ocel,
    'order'
)

pm4py.filtering.filter_ocel_cc_activity(ocel: OCEL, activity: str) → OCEL[source]#

Filters the objects belonging to connected components that include at least one event with the specified activity.

Reference: Adams, Jan Niklas, et al. “Defining cases and variants for object-centric event data.” 2022 4th International Conference on Process Mining (ICPM). IEEE, 2022.

Return type:

Parameters:

ocel (OCEL) – Object-centric event log.
activity (str) – Activity to consider.

Returns:

Filtered OCEL.

import pm4py

ocel = pm4py.read_ocel('log.jsonocel')
filtered_ocel = pm4py.filter_ocel_cc_activity(
    ocel,
    'Create Order'
)

pm4py.hof module#

pm4py.hof.filter_log(f: Callable[[Any], bool], log: EventLog) → EventLog | EventStream[source]#

Filters the log according to a given (lambda) function.

Parameters:

f – function that specifies the filter criterion, may be a lambda
log (EventLog) – event log; either EventLog or EventStream Object

Return type:

Union[log_inst.EventLog, log_inst.EventStream]

Deprecated since version 2.3.0: This will be removed in 3.0.0. the EventLog class will be removed in a future release.

pm4py.hof.filter_trace(f: Callable[[Any], bool], trace: Trace) → Trace[source]#

Filters the trace according to a given (lambda) function.

Parameters:

f – function that specifies the filter criterion, may be a lambda
trace (Trace) – trace; PM4Py trace object

Return type:

log_inst.Trace

pm4py.hof.sort_log(log: EventLog, key, reverse: bool = False) → EventLog | EventStream[source]#

Sorts the event log according to a given key.

Parameters:

log (EventLog) – event log object; either EventLog or EventStream
key – sorting key
reverse (bool) – indicates whether sorting should be reversed or not

Return type:

Union[log_inst.EventLog, log_inst.EventStream]

Deprecated since version 2.3.0: This will be removed in 3.0.0. the EventLog class will be removed in a future release.

pm4py.hof.sort_trace(trace: Trace, key, reverse: bool = False) → Trace[source]#

Sorts the events in a trace according to a given key.

Parameters:

trace (Trace) – input trace
key – sorting key
reverse (bool) – indicates whether sorting should be reversed (default False)

Return type:

log_inst.Trace

Deprecated since version 2.3.0: This will be removed in 3.0.0. the EventLog class will be removed in a future release.

pm4py.llm module#

PM4Py – A Process Mining Library for Python

You should have received a copy of the GNU Affero General Public License along with this program. If not, see this software project’s root or visit <https://www.gnu.org/licenses/>.

Website: https://processintelligence.solutions Contact: info@processintelligence.solutions

pm4py.llm.openai_query(prompt: str, api_key: str | None = None, openai_model: str | None = None, api_url: str | None = None, **kwargs) → str[source]#

Executes the provided prompt, obtaining the answer from the OpenAI APIs.

Return type:

str

Parameters:

prompt (str) – The prompt to be executed.
api_key – (Optional) OpenAI API key.
openai_model – (Optional) OpenAI model to be used (default: “gpt-3.5-turbo”).
api_url – (Optional) OpenAI API URL.
**kwargs –
Additional parameters to pass to the OpenAI API.

Returns:

The response from the OpenAI API as a string.

import pm4py

resp = pm4py.llm.openai_query('What is the result of 3+3?', api_key="sk-382393", openai_model="gpt-3.5-turbo")
print(resp)

pm4py.llm.abstract_dfg(log_obj: DataFrame | EventLog | EventStream, max_len: int = 10000, include_performance: bool = True, relative_frequency: bool = False, response_header: bool = True, primary_performance_aggregation: str = 'mean', secondary_performance_aggregation: str | None = None, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → str[source]#

Obtains the DFG (Directly-Follows Graph) abstraction of a traditional event log.

Return type:

str

Parameters:

log_obj – The log object to abstract.
max_len (int) – Maximum length of the string abstraction (default: constants.OPENAI_MAX_LEN).
include_performance (bool) – Whether to include the performance of the paths in the abstraction.
relative_frequency (bool) – Whether to use relative instead of absolute frequency of the paths.
response_header (bool) – Whether to include a short header before the paths, describing the abstraction.
primary_performance_aggregation (str) – Primary aggregation method for the arc’s performance (default: “mean”). Other options: “median”, “min”, “max”, “sum”, “stdev”.
secondary_performance_aggregation – (Optional) Secondary aggregation method for the arc’s performance (default: None). Other options: “mean”, “median”, “min”, “max”, “sum”, “stdev”.
activity_key (str) – The column name to be used as activity.
timestamp_key (str) – The column name to be used as timestamp.
case_id_key (str) – The column name to be used as case identifier.

Returns:

The DFG abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes")
print(pm4py.llm.abstract_dfg(log))

pm4py.llm.abstract_variants(log_obj: DataFrame | EventLog | EventStream, max_len: int = 10000, include_performance: bool = True, relative_frequency: bool = False, response_header: bool = True, primary_performance_aggregation: str = 'mean', secondary_performance_aggregation: str | None = None, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → str[source]#

Obtains the variants abstraction of a traditional event log.

Return type:

str

Parameters:

log_obj – The log object to abstract.
max_len (int) – Maximum length of the string abstraction (default: constants.OPENAI_MAX_LEN).
include_performance (bool) – Whether to include the performance of the variants in the abstraction.
relative_frequency (bool) – Whether to use relative instead of absolute frequency of the variants.
response_header (bool) – Whether to include a short header before the variants, describing the abstraction.
primary_performance_aggregation (str) – Primary aggregation method for the variants’ performance (default: “mean”). Other options: “median”, “min”, “max”, “sum”, “stdev”.
secondary_performance_aggregation – (Optional) Secondary aggregation method for the variants’ performance (default: None). Other options: “mean”, “median”, “min”, “max”, “sum”, “stdev”.
activity_key (str) – The column name to be used as activity.
timestamp_key (str) – The column name to be used as timestamp.
case_id_key (str) – The column name to be used as case identifier.

Returns:

The variants abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes")
print(pm4py.llm.abstract_variants(log))

pm4py.llm.abstract_ocel(ocel: OCEL, include_timestamps: bool = True) → str[source]#

Obtains the abstraction of an object-centric event log, including the list of events and the objects of the OCEL.

Return type:

str

Parameters:

ocel (OCEL) – The object-centric event log to abstract.
include_timestamps (bool) – Whether to include timestamp information in the abstraction.

Returns:

The OCEL abstraction as a string.

import pm4py

ocel = pm4py.read_ocel("tests/input_data/ocel/example_log.jsonocel")
print(pm4py.llm.abstract_ocel(ocel))

pm4py.llm.abstract_ocel_ocdfg(ocel: OCEL, include_header: bool = True, include_timestamps: bool = True, max_len: int = 10000) → str[source]#

Obtains the abstraction of an object-centric event log, representing the object-centric directly-follows graph in text.

Return type:

str

Parameters:

ocel (OCEL) – The object-centric event log to abstract.
include_header (bool) – Whether to include a header in the abstraction.
include_timestamps (bool) – Whether to include timestamp information in the abstraction.
max_len (int) – Maximum length of the abstraction (default: constants.OPENAI_MAX_LEN).

Returns:

The object-centric DFG abstraction as a string.

import pm4py

ocel = pm4py.read_ocel("tests/input_data/ocel/example_log.jsonocel")
print(pm4py.llm.abstract_ocel_ocdfg(ocel))

pm4py.llm.abstract_ocel_features(ocel: OCEL, obj_type: str, include_header: bool = True, max_len: int = 10000, debug: bool = False, enable_object_lifecycle_paths: bool = True) → str[source]#

Obtains the abstraction of an object-centric event log, representing the features and their values in text.

Return type:

str

Parameters:

ocel (OCEL) – The object-centric event log to abstract.
obj_type (str) – The object type to consider in feature extraction.
include_header (bool) – Whether to include a header in the abstraction.
max_len (int) – Maximum length of the abstraction (default: constants.OPENAI_MAX_LEN).
debug (bool) – Enables debugging mode, providing insights into feature extraction steps.
enable_object_lifecycle_paths (bool) – Enables the “lifecycle paths” feature in the abstraction.

Returns:

The OCEL features abstraction as a string.

import pm4py

ocel = pm4py.read_ocel("tests/input_data/ocel/example_log.jsonocel")
print(pm4py.llm.abstract_ocel_features(ocel, obj_type="Resource"))

pm4py.llm.abstract_event_stream(log_obj: DataFrame | EventLog | EventStream, max_len: int = 10000, response_header: bool = True, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → str[source]#

Obtains the event stream abstraction of a traditional event log.

Return type:

str

Parameters:

log_obj – The log object to abstract.
max_len (int) – Maximum length of the string abstraction (default: constants.OPENAI_MAX_LEN).
response_header (bool) – Whether to include a short header before the event stream, describing the abstraction.
activity_key (str) – The column name to be used as activity.
timestamp_key (str) – The column name to be used as timestamp.
case_id_key (str) – The column name to be used as case identifier.

Returns:

The event stream abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes")
print(pm4py.llm.abstract_event_stream(log))

pm4py.llm.abstract_petri_net(net: PetriNet, im: Marking, fm: Marking, response_header: bool = True) → str[source]#

Obtains an abstraction of a Petri net.

Return type:

str

Parameters:

net (PetriNet) – The Petri net to abstract.
im (Marking) – The initial marking of the Petri net.
fm (Marking) – The final marking of the Petri net.
response_header (bool) – Whether to include a header in the abstraction.

Returns:

The Petri net abstraction as a string.

import pm4py

net, im, fm = pm4py.read_pnml('tests/input_data/running-example.pnml')
print(pm4py.llm.abstract_petri_net(net, im, fm))

pm4py.llm.abstract_log_attributes(log_obj: DataFrame | EventLog | EventStream, max_len: int = 10000, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → str[source]#

Abstracts the attributes of a log by reporting their names, types, and top values.

Return type:

str

Parameters:

log_obj – The log object whose attributes are to be abstracted.
max_len (int) – Maximum length of the string abstraction (default: constants.OPENAI_MAX_LEN).
activity_key (str) – The column name to be used as activity.
timestamp_key (str) – The column name to be used as timestamp.
case_id_key (str) – The column name to be used as case identifier.

Returns:

The log attributes abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes")
print(pm4py.llm.abstract_log_attributes(log))

pm4py.llm.abstract_log_features(log_obj: DataFrame | EventLog | EventStream, max_len: int = 10000, include_header: bool = True, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → str[source]#

Abstracts the machine learning features obtained from a log by reporting the top features until the desired length is achieved.

Return type:

str

Parameters:

log_obj – The log object from which to extract features.
max_len (int) – Maximum length of the string abstraction (default: constants.OPENAI_MAX_LEN).
include_header (bool) – Whether to include a header in the abstraction.
activity_key (str) – The column name to be used as activity.
timestamp_key (str) – The column name to be used as timestamp.
case_id_key (str) – The column name to be used as case identifier.

Returns:

The log features abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes")
print(pm4py.llm.abstract_log_features(log))

pm4py.llm.abstract_temporal_profile(temporal_profile: Dict[Tuple[str, str], Tuple[float, float]], include_header: bool = True) → str[source]#

Abstracts a temporal profile model into a descriptive string.

Return type:

str

Parameters:

temporal_profile – The temporal profile model to abstract.
include_header (bool) – Whether to include a header in the abstraction describing the temporal profile.

Returns:

The temporal profile abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes", return_legacy_log_object=True)
temporal_profile = pm4py.discover_temporal_profile(log)
text_abstr = pm4py.llm.abstract_temporal_profile(temporal_profile, include_header=True)
print(text_abstr)

pm4py.llm.abstract_case(case: Trace, include_case_attributes: bool = True, include_event_attributes: bool = True, include_timestamp: bool = True, include_header: bool = True, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp') → str[source]#

Textually abstracts a single case from an event log.

Return type:

str

Parameters:

case (Trace) – The case object to abstract.
include_case_attributes (bool) – Whether to include attributes at the case level.
include_event_attributes (bool) – Whether to include attributes at the event level.
include_timestamp (bool) – Whether to include event timestamps in the abstraction.
include_header (bool) – Whether to include a header in the abstraction.
activity_key (str) – The column name to be used as activity.
timestamp_key (str) – The column name to be used as timestamp.

Returns:

The case abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes", return_legacy_log_object=True)
print(pm4py.llm.abstract_case(log[0]))

pm4py.llm.abstract_declare(declare_model, include_header: bool = True) → str[source]#

Textually abstracts a DECLARE model.

Return type:

str

Parameters:

declare_model – The DECLARE model to abstract.
include_header (bool) – Whether to include a header in the abstraction.

Returns:

The DECLARE model abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes", return_legacy_log_object=True)
log_ske = pm4py.discover_declare(log)
print(pm4py.llm.abstract_declare(log_ske))

pm4py.llm.abstract_log_skeleton(log_skeleton, include_header: bool = True) → str[source]#

Textually abstracts a log skeleton process model.

Return type:

str

Parameters:

log_skeleton – The log skeleton to abstract.
include_header (bool) – Whether to include a header in the abstraction.

Returns:

The log skeleton abstraction as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/roadtraffic100traces.xes", return_legacy_log_object=True)
log_ske = pm4py.discover_log_skeleton(log)
print(pm4py.llm.abstract_log_skeleton(log_ske))

pm4py.llm.explain_visualization(vis_saver, *args, connector=<function openai_query>, **kwargs) → str[source]#

Explains a process mining visualization using LLMs by saving it as a .png image and providing the image to the Large Language Model along with a description.

Return type:

str

Parameters:

vis_saver – The visualizer function used to save the visualization to disk.
args – Positional arguments required by the visualizer function.
connector – (Optional) The connector method to communicate with the large language model (default: openai_query).
**kwargs –
Additional keyword arguments for the visualizer function or the connector (e.g., annotations, API key).

Returns:

The explanation of the visualization as a string.

import pm4py

log = pm4py.read_xes("tests/input_data/running-example.xes")
descr = pm4py.llm.explain_visualization(pm4py.save_vis_dotted_chart, log, api_key="sk-5HN", show_legend=False)
print(descr)

pm4py.meta module#

Process mining for Python

pm4py.ml module#

The pm4py.ml module contains the machine learning features offered in pm4py.

pm4py.ml.split_train_test(log: EventLog | DataFrame, train_percentage: float = 0.8, case_id_key: str = 'case:concept:name') → Tuple[EventLog, EventLog] | Tuple[DataFrame, DataFrame][source]#

Splits an event log into a training log and a test log for machine learning purposes.

This function separates the provided log into two parts based on the specified training percentage. It ensures that entire cases are included in either the training set or the test set.

Parameters:

log – The event log or Pandas DataFrame to be split.
train_percentage (float) – Fraction of cases to be included in the training log (between 0.0 and 1.0).
case_id_key (str) – Attribute to be used as the case identifier.

Returns:

A tuple containing the training and test event logs or DataFrames.

Return type:

Union[Tuple[EventLog, EventLog], Tuple[pd.DataFrame, pd.DataFrame]]

import pm4py

train_df, test_df = pm4py.split_train_test(dataframe, train_percentage=0.75)

pm4py.ml.get_prefixes_from_log(log: EventLog | DataFrame, length: int, case_id_key: str = 'case:concept:name') → EventLog | DataFrame[source]#

Retrieves prefixes of traces in a log up to a specified length.

The returned log contains prefixes of each trace: - If a trace has a length less than or equal to the specified length, it is included as-is. - If a trace exceeds the specified length, it is truncated to that length.

Parameters:

log – The event log or Pandas DataFrame from which to extract prefixes.
length (int) – The maximum length of prefixes to extract.
case_id_key (str) – Attribute to be used as the case identifier.

Returns:

A log containing the prefixes of the original log.

Return type:

Union[EventLog, pd.DataFrame]

import pm4py

trimmed_df = pm4py.get_prefixes_from_log(dataframe, length=5, case_id_key='case:concept:name')

pm4py.ml.extract_outcome_enriched_dataframe(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', start_timestamp_key: str = 'time:timestamp') → DataFrame[source]#

Enriches a dataframe with additional outcome-related columns computed from the entire case.

This function adds columns that model the outcome of each case by computing metrics such as arrival rates and service waiting times.

Parameters:

log – The event log or Pandas DataFrame to be enriched.
activity_key (str) – Attribute to be used for the activity.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as the case identifier.
start_timestamp_key (str) – Attribute to be used as the start timestamp.

Returns:

An enriched Pandas DataFrame with additional outcome-related columns.

Return type:

pd.DataFrame

import pm4py

enriched_df = pm4py.extract_outcome_enriched_dataframe(
    log,
    activity_key='concept:name',
    timestamp_key='time:timestamp',
    case_id_key='case:concept:name',
    start_timestamp_key='time:timestamp'
)

pm4py.ml.extract_features_dataframe(log: EventLog | DataFrame, str_tr_attr: List[str] | None = None, num_tr_attr: List[str] | None = None, str_ev_attr: List[str] | None = None, num_ev_attr: List[str] | None = None, str_evsucc_attr: List[str] | None = None, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str | None = None, resource_key: str = 'org:resource', include_case_id: bool = False, **kwargs) → DataFrame[source]#

Extracts a dataframe containing features for each case in the provided log object.

This function processes the log to generate a set of features that can be used for machine learning tasks. Features can include both case-level and event-level attributes, with options for one-hot encoding.

Parameters:

log – The event log or Pandas DataFrame from which to extract features.
str_tr_attr – (Optional) List of string attributes at the case level to extract as features.
num_tr_attr – (Optional) List of numeric attributes at the case level to extract as features.
str_ev_attr – (Optional) List of string attributes at the event level to extract as features (one-hot encoded).
num_ev_attr – (Optional) List of numeric attributes at the event level to extract as features (uses the last value per attribute in a case).
str_evsucc_attr – (Optional) List of string successor attributes at the event level to extract as features.
activity_key (str) – Attribute to be used as the activity identifier.
timestamp_key (str) – Attribute to be used for timestamps.
case_id_key – (Optional) Attribute to be used as the case identifier. If not provided, the default is used.
resource_key (str) – Attribute to be used as the resource identifier.
include_case_id (bool) – Whether to include the case identifier column in the features table.
**kwargs –
Additional keyword arguments to pass to the feature extraction algorithm.

Returns:

A Pandas DataFrame containing the extracted features for each case.

Return type:

pd.DataFrame

import pm4py

features_df = pm4py.extract_features_dataframe(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.ml.extract_ocel_features(ocel: OCEL, obj_type: str, enable_object_lifecycle_paths: bool = True, enable_object_work_in_progress: bool = False, object_str_attributes: Collection[str] | None = None, object_num_attributes: Collection[str] | None = None, include_obj_id: bool = False, debug: bool = False) → DataFrame[source]#

Extracts a set of features from an object-centric event log (OCEL) for objects of a specified type.

This function computes various features based on the lifecycle paths and work-in-progress metrics of objects within the OCEL. It also supports encoding of string and numeric object attributes.

The approach is based on: Berti, A., Herforth, J., Qafari, M.S. et al. Graph-based feature extraction on object-centric event logs. Int J Data Sci Anal (2023). https://doi.org/10.1007/s41060-023-00428-2

Parameters:

ocel (OCEL) – The object-centric event log from which to extract features.
obj_type (str) – The object type to consider for feature extraction.
enable_object_lifecycle_paths (bool) – Whether to enable the “lifecycle paths” feature.
enable_object_work_in_progress (bool) – Whether to enable the “work in progress” feature, which has a high computational cost.
object_str_attributes – (Optional) Collection of string attributes at the object level to one-hot encode.
object_num_attributes – (Optional) Collection of numeric attributes at the object level to encode.
include_obj_id (bool) – Whether to include the object identifier as a column in the features DataFrame.
debug (bool) – Whether to enable debugging mode to track the feature extraction process.

Returns:

A Pandas DataFrame containing the extracted features for the specified object type.

Return type:

pd.DataFrame

import pm4py

ocel = pm4py.read_ocel('log.jsonocel')
fea_df = pm4py.extract_ocel_features(ocel, "item")

pm4py.ml.extract_temporal_features_dataframe(log: EventLog | DataFrame, grouper_freq: str = 'W', activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str | None = None, start_timestamp_key: str = 'time:timestamp', resource_key: str = 'org:resource') → DataFrame[source]#

Extracts temporal features from a log object and returns them as a dataframe.

This function computes temporal metrics based on the specified grouping frequency, which can be daily (D), weekly (W), monthly (M), or yearly (Y). These features are useful for analyzing system dynamics and simulation in the context of process mining.

The approach is based on: Pourbafrani, Mahsa, Sebastiaan J. van Zelst, and Wil MP van der Aalst. “Supporting automatic system dynamics model generation for simulation in the context of process mining.” International Conference on Business Information Systems. Springer, Cham, 2020.

Parameters:

log – The event log or Pandas DataFrame from which to extract temporal features.
grouper_freq (str) – The frequency to use for grouping (e.g., ‘D’ for daily, ‘W’ for weekly, ‘M’ for monthly, ‘Y’ for yearly).
activity_key (str) – Attribute to be used as the activity identifier.
timestamp_key (str) – Attribute to be used for timestamps.
case_id_key – (Optional) Attribute to be used as the case identifier. If not provided, the default is used.
start_timestamp_key (str) – Attribute to be used as the start timestamp.
resource_key (str) – Attribute to be used as the resource identifier.

Returns:

A Pandas DataFrame containing the extracted temporal features.

Return type:

pd.DataFrame

import pm4py

temporal_features_df = pm4py.extract_temporal_features_dataframe(
    dataframe,
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.ml.extract_target_vector(log: EventLog | DataFrame, variant: str, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Tuple[Any, List[str]][source]#

Extracts the target vector from a log object for a specific machine learning use case.

Supported variants include: - ‘next_activity’: Predicts the next activity in a case. - ‘next_time’: Predicts the timestamp of the next activity. - ‘remaining_time’: Predicts the remaining time for the case.

Parameters:

log – The event log or Pandas DataFrame from which to extract the target vector.
variant (str) – The variant of the algorithm to use. Must be one of: ‘next_activity’, ‘next_time’, ‘remaining_time’.
activity_key (str) – Attribute to be used as the activity identifier.
timestamp_key (str) – Attribute to be used for timestamps.
case_id_key (str) – Attribute to be used as the case identifier.

Returns:

A tuple containing the target vector and a list of class labels (if applicable).

Return type:

Tuple[Any, List[str]]

Raises:

Exception – If an unsupported variant is provided.

import pm4py

vector_next_act, class_next_act = pm4py.extract_target_vector(
    log,
    'next_activity',
    activity_key='concept:name',
    timestamp_key='time:timestamp',
    case_id_key='case:concept:name'
)
vector_next_time, class_next_time = pm4py.extract_target_vector(
    log,
    'next_time',
    activity_key='concept:name',
    timestamp_key='time:timestamp',
    case_id_key='case:concept:name'
)
vector_rem_time, class_rem_time = pm4py.extract_target_vector(
    log,
    'remaining_time',
    activity_key='concept:name',
    timestamp_key='time:timestamp',
    case_id_key='case:concept:name'
)

pm4py.ocel module#

The pm4py.ocel module contains the object-centric process mining features offered in pm4py.

pm4py.ocel.ocel_get_object_types(ocel: OCEL) → List[str][source]#

Returns the list of object types contained in the object-centric event log (e.g., [“order”, “item”, “delivery”]).

Parameters:: ocel (OCEL) – Object-centric event log.
Returns:: List of object types.
Return type:: List[str]

import pm4py

object_types = pm4py.ocel_get_object_types(ocel)

pm4py.ocel.ocel_get_attribute_names(ocel: OCEL) → List[str][source]#

Returns the list of attributes at the event and object levels of an object-centric event log (e.g., [“cost”, “amount”, “name”]).

Parameters:: ocel (OCEL) – Object-centric event log.
Returns:: List of attribute names.
Return type:: List[str]

import pm4py

attribute_names = pm4py.ocel_get_attribute_names(ocel)

pm4py.ocel.ocel_flattening(ocel: OCEL, object_type: str) → DataFrame[source]#

Flattens the object-centric event log to a traditional event log based on a chosen object type. In the flattened log, the objects of the specified type are treated as cases, and each case contains the set of events related to that object. The flattened log follows the XES notations for case identifier, activity, and timestamp. Specifically: - “case:concept:name” is used for the case ID. - “concept:name” is used for the activity. - “time:timestamp” is used for the timestamp.

Parameters:

ocel (OCEL) – Object-centric event log.
object_type (str) – The object type to use as cases.

Returns:

Flattened traditional event log.

Return type:

pd.DataFrame

import pm4py

event_log = pm4py.ocel_flattening(ocel, 'items')

pm4py.ocel.ocel_object_type_activities(ocel: OCEL) → Dict[str, Collection[str]][source]#

Returns the set of activities performed for each object type.

Parameters:: ocel (OCEL) – Object-centric event log.
Returns:: Dictionary mapping object types to their associated activities.
Return type:: Dict[str, Collection[str]]

import pm4py

ot_activities = pm4py.ocel_object_type_activities(ocel)

pm4py.ocel.ocel_objects_ot_count(ocel: OCEL) → Dict[str, Dict[str, int]][source]#

Returns the count of related objects per type for each event.

Parameters:: ocel (OCEL) – Object-centric event log.
Returns:: Nested dictionary mapping events to object types and their counts.
Return type:: Dict[str, Dict[str, int]]

import pm4py

objects_ot_count = pm4py.ocel_objects_ot_count(ocel)

pm4py.ocel.ocel_temporal_summary(ocel: OCEL) → DataFrame[source]#

Returns the temporal summary of an object-centric event log. The temporal summary aggregates all events that occur at the same timestamp and reports the list of activities and involved objects.

Parameters:: ocel (OCEL) – Object-centric event log.
Returns:: Temporal summary DataFrame.
Return type:: pd.DataFrame

import pm4py

temporal_summary = pm4py.ocel_temporal_summary(ocel)

pm4py.ocel.ocel_objects_summary(ocel: OCEL) → DataFrame[source]#

Returns the objects summary of an object-centric event log.

Parameters:: ocel (OCEL) – Object-centric event log.
Returns:: Objects summary DataFrame containing lifecycle information and interacting objects.
Return type:: pd.DataFrame

import pm4py

objects_summary = pm4py.ocel_objects_summary(ocel)

pm4py.ocel.ocel_objects_interactions_summary(ocel: OCEL) → DataFrame[source]#

Returns the objects interactions summary of an object-centric event log. The summary includes a row for every combination of (event, related object, other related object). Properties such as the activity of the event and the object types of the two related objects are included.

Parameters:: ocel (OCEL) – Object-centric event log.
Returns:: Objects interactions summary DataFrame.
Return type:: pd.DataFrame

import pm4py

interactions_summary = pm4py.ocel_objects_interactions_summary(ocel)

pm4py.ocel.discover_ocdfg(ocel: OCEL, business_hours: bool = False, business_hour_slots: List[Tuple[int, int]] | None = [(25200, 61200), (111600, 147600), (198000, 234000), (284400, 320400), (370800, 406800)]) → Dict[str, Any][source]#

Discovers an Object-Centric Directly-Follows Graph (OC-DFG) from an object-centric event log.

Object-centric directly-follows multigraphs are a composition of directly-follows graphs for each object type. These graphs can be annotated with different metrics considering the entities of an object-centric event log (i.e., events, unique objects, total objects).

Reference paper: Berti, Alessandro, and Wil van der Aalst. “Extracting multiple viewpoint models from relational databases.” Data-Driven Process Discovery and Analysis. Springer, Cham, 2018. 24-51.

Parameters:

ocel (OCEL) – Object-centric event log.
business_hours (bool) – Enable the usage of business hours if set to True.
business_hour_slots (Optional[List[Tuple[int, int]]]) – Work schedule of the company, provided as a list of tuples where each tuple represents one time slot of business hours. Each tuple consists of a start and an end time given in seconds since week start, e.g., [(25200, 61200), (9072, 43200), (46800, 61200)] meaning that business hours are Mondays 07:00 - 17:00, Tuesdays 02:32 - 12:00, and Wednesdays 13:00 - 17:00.

Returns:

OC-DFG discovery result.

Return type:

Dict[str, Any]

import pm4py

ocdfg = pm4py.discover_ocdfg(ocel)

pm4py.ocel.discover_oc_petri_net(ocel: OCEL, inductive_miner_variant: str = 'im', diagnostics_with_tbr: bool = False) → Dict[str, Any][source]#

Discovers an object-centric Petri net from the provided object-centric event log.

Reference paper: van der Aalst, Wil MP, and Alessandro Berti. “Discovering object-centric Petri nets.” Fundamenta Informaticae 175.1-4 (2020): 1-40.

Parameters:

ocel (OCEL) – Object-centric event log.
inductive_miner_variant (str) – Variant of the inductive miner to use (“im” for traditional; “imd” for the faster inductive miner directly-follows).
diagnostics_with_tbr (bool) – Enable the computation of diagnostics using token-based replay if set to True.

Returns:

Discovered object-centric Petri net.

Return type:

Dict[str, Any]

import pm4py

ocpn = pm4py.discover_oc_petri_net(ocel)

pm4py.ocel.discover_objects_graph(ocel: OCEL, graph_type: str = 'object_interaction') → Set[Tuple[str, str]][source]#

Discovers an object graph from the provided object-centric event log.

Available graph types: - “object_interaction” - “object_descendants” - “object_inheritance” - “object_cobirth” - “object_codeath”

Parameters:

ocel (OCEL) – Object-centric event log.
graph_type (str) – Type of graph to consider. Options include “object_interaction”, “object_descendants”, “object_inheritance”, “object_cobirth”, “object_codeath”.

Returns:

Discovered object graph as a set of tuples.

Return type:

Set[Tuple[str, str]]

import pm4py

ocel = pm4py.read_ocel('trial.ocel')
obj_graph = pm4py.discover_objects_graph(ocel, graph_type='object_interaction')

pm4py.ocel.ocel_o2o_enrichment(ocel: OCEL, included_graphs: Collection[str] | None = None) → OCEL[source]#

Enriches the OCEL with information inferred from graph computations by inserting them into the O2O relations.

Parameters:

ocel (OCEL) – Object-centric event log.
included_graphs (Optional[Collection[str]]) – Types of graphs to include, provided as a list or set of strings. Options include “object_interaction_graph”, “object_descendants_graph”, “object_inheritance_graph”, “object_cobirth_graph”, “object_codeath_graph”.

Returns:

Enriched object-centric event log.

Return type:

import pm4py

ocel = pm4py.read_ocel('trial.ocel')
ocel = pm4py.ocel_o2o_enrichment(ocel)
print(ocel.o2o)

pm4py.ocel.ocel_e2o_lifecycle_enrichment(ocel: OCEL) → OCEL[source]#

Enriches the OCEL with lifecycle-based information, indicating when an object is created, terminated, or has other types of relations, by updating the E2O relations.

Parameters:: ocel (OCEL) – Object-centric event log.
Returns:: Enriched object-centric event log with lifecycle information.
Return type:: OCEL

import pm4py

ocel = pm4py.read_ocel('trial.ocel')
ocel = pm4py.ocel_e2o_lifecycle_enrichment(ocel)
print(ocel.relations)

pm4py.ocel.sample_ocel_objects(ocel: OCEL, num_objects: int) → OCEL[source]#

Returns a sampled object-centric event log containing a random subset of objects. Only events related to at least one of the sampled objects are included in the returned log. Note that this sampling may disrupt the relationships between different objects.

Parameters:

ocel (OCEL) – Object-centric event log.
num_objects (int) – Number of objects to include in the sampled event log.

Returns:

Sampled object-centric event log.

Return type:

import pm4py

ocel = pm4py.read_ocel('trial.ocel')
sampled_ocel = pm4py.sample_ocel_objects(ocel, 50)  # Keeps only 50 random objects

pm4py.ocel.sample_ocel_connected_components(ocel: OCEL, connected_components: int = 1, max_num_events_per_cc: int = 9223372036854775807, max_num_objects_per_cc: int = 9223372036854775807, max_num_e2o_relations_per_cc: int = 9223372036854775807) → OCEL[source]#

Returns a sampled object-centric event log containing a specified number of connected components. Users can also set maximum limits on the number of events, objects, and E2O relations per connected component.

Reference paper: Adams, Jan Niklas, et al. “Defining cases and variants for object-centric event data.” 2022 4th International Conference on Process Mining (ICPM). IEEE, 2022.

Parameters:

ocel (OCEL) – Object-centric event log.
connected_components (int) – Number of connected components to include in the sampled event log.
max_num_events_per_cc (int) – Maximum number of events allowed per connected component (default: sys.maxsize).
max_num_objects_per_cc (int) – Maximum number of objects allowed per connected component (default: sys.maxsize).
max_num_e2o_relations_per_cc (int) – Maximum number of event-to-object relationships allowed per connected component (default: sys.maxsize).

Returns:

Sampled object-centric event log containing the specified connected components.

Return type:

import pm4py

ocel = pm4py.read_ocel('trial.ocel')
sampled_ocel = pm4py.sample_ocel_connected_components(ocel, 5)  # Keeps only 5 connected components

pm4py.ocel.ocel_drop_duplicates(ocel: OCEL) → OCEL[source]#

Removes duplicate relations between events and objects that occur at the same time, have the same activity, and are linked to the same object identifier. This effectively cleans the OCEL by eliminating duplicate events.

Parameters:: ocel (OCEL) – Object-centric event log.
Returns:: Cleaned object-centric event log without duplicate relations.
Return type:: OCEL

import pm4py

ocel = pm4py.read_ocel('trial.ocel')
ocel = pm4py.ocel_drop_duplicates(ocel)

pm4py.ocel.ocel_merge_duplicates(ocel: OCEL, have_common_object: bool | None = False) → OCEL[source]#

Merges events in the OCEL that have the same activity and timestamp. Optionally, ensures that the events being merged share a common object.

Parameters:

ocel (OCEL) – Object-centric event log.
have_common_object (Optional[bool]) – If set to True, only merges events that share a common object. Defaults to False.

Returns:

Object-centric event log with merged duplicate events.

Return type:

import pm4py

ocel = pm4py.read_ocel('trial.ocel')
ocel = pm4py.ocel_merge_duplicates(ocel)

pm4py.ocel.ocel_sort_by_additional_column(ocel: OCEL, additional_column: str, primary_column: str = 'ocel:timestamp') → OCEL[source]#

Sorts the OCEL based on the primary timestamp column and an additional column to determine the order of events occurring at the same timestamp.

Parameters:

ocel (OCEL) – Object-centric event log.
additional_column (str) – Additional column to use for sorting.
primary_column (str) – Primary column to use for sorting (default: “ocel:timestamp”). Typically the timestamp column.

Returns:

Sorted object-centric event log.

Return type:

import pm4py

ocel = pm4py.read_ocel('trial.ocel')
ocel = pm4py.ocel_sort_by_additional_column(ocel, 'ordering')

pm4py.ocel.ocel_add_index_based_timedelta(ocel: OCEL) → OCEL[source]#

Adds a small time delta to the timestamp column based on the event index to ensure the correct ordering of events within any object-centric process mining solution.

Parameters:: ocel (OCEL) – Object-centric event log.
Returns:: Object-centric event log with index-based time deltas added.
Return type:: OCEL

import pm4py

ocel = pm4py.read_ocel('trial.ocel')
ocel = pm4py.ocel_add_index_based_timedelta(ocel)

pm4py.ocel.cluster_equivalent_ocel(ocel: OCEL, object_type: str, max_objs: int = 9223372036854775807) → Dict[str, Collection[OCEL]][source]#

Clusters the object-centric event log based on the ‘executions’ of a single object type. Equivalent ‘executions’ are grouped together in the output dictionary.

Parameters:

ocel (OCEL) – Object-centric event log.
object_type (str) – Reference object type for clustering.
max_objs (int) – Maximum number of objects (of the specified object type) to include per cluster. Defaults to sys.maxsize.

Returns:

Dictionary mapping cluster descriptions to collections of equivalent OCELs.

Return type:

Dict[str, Collection[OCEL]]

import pm4py

ocel = pm4py.read_ocel('trial.ocel')
clusters = pm4py.cluster_equivalent_ocel(ocel, "order")

pm4py.org module#

The pm4py.org module contains organizational analysis techniques offered in pm4py.

pm4py.org.discover_handover_of_work_network(log: EventLog | DataFrame, beta=0, resource_key: str = 'org:resource', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → SNA[source]#

Calculates the handover of work network of the event log.

The handover of work network is essentially the Directly-Follows Graph (DFG) of the event log, but using the resource as the nodes of the graph instead of activities. As such, resource information should be present in the event log.

Return type:

Parameters:

log – Event log or Pandas DataFrame.
beta (int) – Beta parameter for the Handover metric.
resource_key (str) – Attribute to be used for the resource.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

import pm4py

metric = pm4py.discover_handover_of_work_network(
    dataframe,
    beta=0,
    resource_key='org:resource',
    timestamp_key='time:timestamp',
    case_id_key='case:concept:name'
)

pm4py.org.discover_working_together_network(log: EventLog | DataFrame, resource_key: str = 'org:resource', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → SNA[source]#

Calculates the working together network of the process. Two resource nodes are connected in the graph if the resources collaborate on an instance of the process.

Return type:

Parameters:

log – Event log or Pandas DataFrame.
resource_key (str) – Attribute to be used for the resource.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

import pm4py

metric = pm4py.discover_working_together_network(
    dataframe,
    resource_key='org:resource',
    timestamp_key='time:timestamp',
    case_id_key='case:concept:name'
)

pm4py.org.discover_activity_based_resource_similarity(log: EventLog | DataFrame, activity_key: str = 'concept:name', resource_key: str = 'org:resource', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → SNA[source]#

Calculates similarity between the resources in the event log based on their activity profiles.

Return type:

Parameters:

log – Event log or Pandas DataFrame.
activity_key (str) – Attribute to be used for the activity.
resource_key (str) – Attribute to be used for the resource.
timestamp_key (str) – Attribute to be used for the timestamp.
case_id_key (str) – Attribute to be used as case identifier.

import pm4py

act_res_sim = pm4py.discover_activity_based_resource_similarity(
    dataframe,
    resource_key='org:resource',
    activity_key='concept:name',
    timestamp_key='time:timestamp',
    case_id_key='case:concept:name'
)

pm4py.org.discover_subcontracting_network(log: EventLog | DataFrame, n=2, resource_key: str = 'org:resource', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → SNA[source]#

Calculates the subcontracting network of the process.

Return type: