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 – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – 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 – pandas.DataFrame
case_id_key – Attribute to be used as case identifier (default: “case:concept:name”).
activity_key – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – 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', perf_aggregation_key: str = 'all') → 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 – 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 – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – Attribute to be used as case identifier (default: “case:concept:name”).
perf_aggregation_key – Selector for the type of aggregation (all, mean, median, max, min, sum, stdev)

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 – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – 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 – Noise threshold for the sequence encoding graph (1.0=no filtering, 0.0=maximum filtering) (default: 1.0).
activity_key – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – 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]#

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 – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – 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'
)

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 – Enables or disables multiprocessing in the Inductive Miner (default: constants.ENABLE_MULTIPROCESSING_DEFAULT).
noise_threshold – Noise threshold (default: 0.0).
activity_key – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – Attribute to be used as case identifier (default: “case:concept:name”).
disable_fallthroughs – 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 – Dependency threshold (default: 0.5).
and_threshold – AND threshold for parallelism (default: 0.65).
loop_two_threshold – Loop two threshold (default: 0.5).
activity_key – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – 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.

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.
noise_threshold – Noise threshold (default: 0.0).
multi_processing – Enables or disables multiprocessing in the Inductive Miner (default: constants.ENABLE_MULTIPROCESSING_DEFAULT).
activity_key – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – Attribute to be used as case identifier (default: “case:concept:name”).
disable_fallthroughs – 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.

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 – Dependency threshold (default: 0.5).
and_threshold – AND threshold for parallelism (default: 0.65).
loop_two_threshold – Loop two threshold (default: 0.5).
min_act_count – Minimum number of occurrences per activity to be included in the discovery (default: 1).
min_dfg_occurrences – Minimum number of occurrences per arc in the DFG to be included in the discovery (default: 1).
activity_key – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – Attribute to be used as case identifier (default: “case:concept:name”).
decoration – 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 – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – 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 | POWL) → 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, or POWL.
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 – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – 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.

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.
noise_threshold – Noise threshold (default: 0.0).
multi_processing – Enables or disables multiprocessing in the Inductive Miner (default: constants.ENABLE_MULTIPROCESSING_DEFAULT).
activity_key – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – Attribute to be used as case identifier (default: “case:concept:name”).
disable_fallthroughs – 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 – Direction in which the transition system is built (“forward” or “backward”) (default: “forward”).
window – Window size for state construction (e.g., 2, 3) (default: 2).
view – View to use in the construction of the states (“sequence”, “set”, “multiset”) (default: “sequence”).
activity_key – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – 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, max_path_length: int | None = None, 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.
max_path_length – maximum path length (each trace is trimmed afterwards).
activity_key – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – 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)

}#

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 – Noise threshold influencing the strictness of constraints (default: 0.0).
activity_key – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – 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 – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – 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 – Factoring threshold for filtering weights, accepts values 0 <= x < 1 (default: 0.0).
order_graph_filtering_threshold – Filtering threshold for the order graph, valid for the DYNAMIC_CLUSTERING variant, accepts values 0.5 < x <= 1 (default: None).
activity_key – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – 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 – 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 – Minimum number of events required to form a batch (default: 2).
activity_key – Attribute to be used for the activity (default: “concept:name”).
timestamp_key – Attribute to be used for the timestamp (default: “time:timestamp”).
case_id_key – Attribute to be used as case identifier (default: “case:concept:name”).
resource_key – 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.discovery.correlation_miner(df: DataFrame, annotation: str = 'frequency', activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp') → Tuple[dict, dict, dict][source]#

Applies the Correlation Miner to ‘discover’ the frequency/performance DFG from an event log without case ID.

The approach is described in: Pourmirza, Shaya, Remco Dijkman, and Paul Grefen. “Correlation miner: mining business process models and event correlations without case identifiers.” International Journal of Cooperative Information Systems 26.02 (2017): 1742002.

Parameters:

log – Pandas dataframe
annotation – annotation (‘frequency’ for the frequency DFG, or ‘performance’ for the performance DFG)
activity_key – attribute to be used for the activity
timestamp_key – attribute to be used for the timestamp

Return type:

Tuple[dict, dict, dict]

import pm4py

log = pm4py.read_xes("tests/input_data/running-example.xes")
log = log[["concept:name", "time:timestamp"]]

dfg, sa, ea = pm4py.correlation_miner(log)
pm4py.view_dfg(dfg, sa, ea, format="svg")

perf_dfg, sa, ea = pm4py.correlation_miner(log, annotation="performance")
pm4py.view_performance_dfg(perf_dfg, sa, ea, format="svg")

pm4py.discovery module#

}#

This Page

PMTk

PM4Py

Company