pm4py.stats module#

The pm4py.stats module contains the statistical functionalities offered in pm4py.

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

Returns the start activities and their frequencies from a log object.

Parameters:

log – Log object (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A dictionary mapping start activity names to their frequencies.

import pm4py

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

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

Returns the end activities and their frequencies from a log object.

Parameters:

log – Log object (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A dictionary mapping end activity names to their frequencies.

import pm4py

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

pm4py.stats.get_event_attributes(log: EventLog | DataFrame) → List[str][source]#

Returns the list of event-level attributes in the log.

Parameters:: log – Log object (EventLog or pandas DataFrame).
Returns:: A list of event attribute names.

import pm4py

event_attributes = pm4py.get_event_attributes(dataframe)

pm4py.stats.get_trace_attributes(log: EventLog | DataFrame) → List[str][source]#

Returns the list of trace-level attributes in the log.

Parameters:: log – Log object (EventLog or pandas DataFrame).
Returns:: A list of trace attribute names.

import pm4py

trace_attributes = pm4py.get_trace_attributes(dataframe)

pm4py.stats.get_event_attribute_values(log: EventLog | DataFrame, attribute: str, count_once_per_case: bool = False, case_id_key: str = 'case:concept:name') → Dict[str, int][source]#

Returns the values and their frequencies for a specified event attribute.

Parameters:

log – Log object (EventLog or pandas DataFrame).
attribute – The event attribute to analyze.
count_once_per_case – If True, count each attribute value at most once per case.
case_id_key – Attribute to be used as the case identifier.

Returns:

A dictionary mapping attribute values to their frequencies.

import pm4py

activities = pm4py.get_event_attribute_values(
    dataframe,
    'concept:name',
    case_id_key='case:concept:name'
)

pm4py.stats.get_trace_attribute_values(log: EventLog | DataFrame, attribute: str, case_id_key: str = 'case:concept:name') → Dict[str, int][source]#

Returns the values and their frequencies for a specified trace attribute.

Parameters:

log – Log object (EventLog or pandas DataFrame).
attribute – The trace attribute to analyze.
case_id_key – Attribute to be used as the case identifier.

Returns:

A dictionary mapping trace attribute values to their frequencies.

import pm4py

tr_attr_values = pm4py.get_trace_attribute_values(
    dataframe,
    'case:attribute',
    case_id_key='case:concept:name'
)

pm4py.stats.get_variants(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', max_repetitions: int = 9223372036854775807) → Dict[Tuple[str], List[Trace]] | Dict[Tuple[str], int][source]#

Retrieves the variants from the log.

Parameters:

log – Event log (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.
max_repetitions – Maximum number of consecutive repetitions for an activity. Reduces variants by limiting consecutive activity repetitions.

Returns:

A dictionary mapping activity tuples to their counts or lists of traces.

import pm4py

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

pm4py.stats.get_variants_as_tuples(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', max_repetitions: int = 9223372036854775807) → Dict[Tuple[str], List[Trace]] | Dict[Tuple[str], int][source]#

Retrieves the variants from the log, where the variant keys are tuples.

Parameters:

log – Event log (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.
max_repetitions – Maximum number of consecutive repetitions for an activity. Reduces variants by limiting consecutive activity repetitions.

Returns:

A dictionary mapping activity tuples to their counts or lists of traces.

import pm4py

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

pm4py.stats.split_by_process_variant(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', variant_column: str = '@@variant_column', index_in_trace_column: str = '@@index_in_trace') → Iterator[Tuple[Collection[str], DataFrame]][source]#

Splits an event log into sub-dataframes for each process variant. The result is an iterator over the variants along with their corresponding sub-dataframes.

Parameters:

log – Event log (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.
variant_column – Name of the utility column that stores the variant’s tuple.
index_in_trace_column – Name of the utility column that stores the index of the event in the case.

Returns:

An iterator of tuples, each containing a variant and its corresponding sub-dataframe.

import pandas as pd
import pm4py

dataframe = pd.read_csv('tests/input_data/receipt.csv')
dataframe = pm4py.format_dataframe(dataframe)
for variant, subdf in pm4py.split_by_process_variant(dataframe):
    print(variant)
    print(subdf)

pm4py.stats.get_variants_paths_duration(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name', variant_column: str = '@@variant_column', variant_count: str = '@@variant_count', index_in_trace_column: str = '@@index_in_trace', cumulative_occ_path_column: str = '@@cumulative_occ_path_column', times_agg: str = 'mean') → DataFrame[source]#

Associates a pandas DataFrame aggregated by variants and their positions within each variant. Each row includes: - The variant - The position within the variant - The source activity of the path - The target activity of the path - An aggregation of the times between the two activities (e.g., mean) - The cumulative occurrences of the path within the case

Parameters:

log – Event log (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.
variant_column – Name of the utility column that stores the variant’s tuple.
variant_count – Name of the utility column that stores the variant’s occurrence count.
index_in_trace_column – Name of the utility column that stores the index of the event in the case.
cumulative_occ_path_column – Name of the column that stores the cumulative occurrences of the path within the case.
times_agg – Aggregation function to be used for time differences (e.g., “mean”, “median”).

Returns:

A pandas DataFrame with the aggregated variant paths and durations.

import pandas as pd
import pm4py

dataframe = pd.read_csv('tests/input_data/receipt.csv')
dataframe = pm4py.format_dataframe(dataframe)

var_paths_durs = pm4py.get_variants_paths_duration(dataframe)
print(var_paths_durs)

pm4py.stats.get_stochastic_language(*args, **kwargs) → Dict[List[str], float][source]#

Retrieves the stochastic language from the provided object.

The stochastic language represents the probabilities of different traces or sequences within the process.

Parameters:

args – The input object, which can be a pandas DataFrame, EventLog, accepting Petri net, or ProcessTree.
kwargs – Additional keyword arguments.

Returns:

A dictionary mapping sequences of activities to their probabilities.

import pm4py

# From an event log
log = pm4py.read_xes('tests/input_data/running-example.xes')
language_log = pm4py.get_stochastic_language(log)
print(language_log)

# From a Petri net
net, im, fm = pm4py.read_pnml('tests/input_data/running-example.pnml')
language_model = pm4py.get_stochastic_language(net, im, fm)
print(language_model)

pm4py.stats.get_minimum_self_distances(log: EventLog | DataFrame, 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 an activity a in a trace is defined as follows: - In a trace <a>, it’s infinity. - In a trace <a, a>, it’s 0. - In a trace <a, b, a>, it’s 1. - And so on.

The minimum self-distance for an activity is the smallest self-distance observed across all traces.

Parameters:

log – Event log (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

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

import pm4py

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

pm4py.stats.get_minimum_self_distance_witnesses(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Dict[str, Set[str]][source]#

Derives the minimum self-distance witnesses for each activity.

A ‘witness’ is an activity that occurs between two occurrences of the same activity at the minimum self-distance. For example, if the minimum self-distance of activity a is 2, then in a trace <a, b, c, a>, activities b and c are witnesses of a.

Parameters:

log – Event log (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A dictionary mapping each activity to a set of its witness activities.

import pm4py

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

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

Calculates the average time difference between the start times of two consecutive cases.

This metric is based on the definition: Cycle time = Average time between completion of units.

Example: In a manufacturing facility producing 100 units in a 40-hour week, the average throughput rate is 1 unit per 0.4 hours (24 minutes per unit). Therefore, the cycle time is 24 minutes on average.

Parameters:

log – Event log (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

The average case arrival time in the same units as the timestamp.

import pm4py

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

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

Identifies activities that have rework occurrences, i.e., activities that occur more than once within the same case. The output is a dictionary mapping each such activity to the number of cases in which rework occurred.

Parameters:

log – Log object (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A dictionary mapping each activity with rework to the number of cases where rework occurred.

import pm4py

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

pm4py.stats.get_case_overlap(log: EventLog | DataFrame, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → List[int]#

Associates each case in the log with the number of cases that are concurrently open.

Parameters:

log – Log object (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A list where each element corresponds to a case and indicates the number of overlapping cases.

import pm4py

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

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

Calculates the cycle time of the event log.

Cycle time is defined as the average time between the completion of units.

Example: In a manufacturing facility producing 100 units in a 40-hour week, the average throughput rate is 1 unit per 0.4 hours (24 minutes per unit). Therefore, the cycle time is 24 minutes on average.

Parameters:

log – Event log (EventLog or pandas DataFrame).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

The cycle time as a float.

import pm4py

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

pm4py.stats.get_service_time(log: EventLog | DataFrame, aggregation_measure: str = 'mean', activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', start_timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Dict[str, float][source]#

Computes the service time for each activity in the event log using the specified aggregation measure.

Service time refers to the duration an activity takes within a case.

Parameters:

log – Event log (EventLog or pandas DataFrame).
aggregation_measure – Aggregation function to apply (e.g., “mean”, “median”, “min”, “max”, “sum”).
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
start_timestamp_key – Attribute to be used for the start timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A dictionary mapping each activity to its aggregated service time.

import pm4py

log = pm4py.read_xes('tests/input_data/interval_event_log.xes')
mean_serv_time = pm4py.get_service_time(
    log,
    start_timestamp_key='start_timestamp',
    aggregation_measure='mean'
)
print(mean_serv_time)

median_serv_time = pm4py.get_service_time(
    log,
    start_timestamp_key='start_timestamp',
    aggregation_measure='median'
)
print(median_serv_time)

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

Retrieves the durations of all cases in the event log.

Parameters:

log – Event log (EventLog or pandas DataFrame).
business_hours – If True, computes durations based on business hours; otherwise, uses calendar time.
business_hour_slots –
Work schedule of the company as a list of tuples. Each tuple represents a time slot in seconds since the week start. Example: [

(7 * 60 * 60, 17 * 60 * 60), ((24 + 7) * 60 * 60, (24 + 12) * 60 * 60), ((24 + 13) * 60 * 60, (24 + 17) * 60 * 60),

] This example means: - Monday 07:00 - 17:00 - Tuesday 07:00 - 12:00 - Tuesday 13:00 - 17:00
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A sorted list of case durations.

import pm4py

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

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

Retrieves the duration of a specific case.

Parameters:

log – Event log (EventLog or pandas DataFrame).
case_id – Identifier of the case whose duration is to be retrieved.
business_hours – If True, computes duration based on business hours; otherwise, uses calendar time.
business_hour_slots –
Work schedule of the company as a list of tuples. Each tuple represents a time slot in seconds since the week start. Example: [

(7 * 60 * 60, 17 * 60 * 60), ((24 + 7) * 60 * 60, (24 + 12) * 60 * 60), ((24 + 13) * 60 * 60, (24 + 17) * 60 * 60),

] This example means: - Monday 07:00 - 17:00 - Tuesday 07:00 - 12:00 - Tuesday 13:00 - 17:00
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

The duration of the specified case.

import pm4py

duration = pm4py.get_case_duration(
    dataframe,
    'case_1',
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.stats.get_frequent_trace_segments(log: EventLog | DataFrame, min_occ: int, activity_key: str = 'concept:name', timestamp_key: str = 'time:timestamp', case_id_key: str = 'case:concept:name') → Counter[source]#

Retrieves frequent trace segments (sub-sequences of activities) from an event log. Each trace segment is preceded and followed by “…”, indicating that it can be part of a larger sequence.

Parameters:

log – Event log (EventLog or pandas DataFrame).
min_occ – Minimum number of occurrences for a trace segment to be included.
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A Counter object mapping trace segments to their occurrence counts.

import pm4py

log = pm4py.read_xes("tests/input_data/receipt.xes")
traces = pm4py.get_frequent_trace_segments(log, min_occ=100)
print(traces)

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

Summarizes the positions of a specific activity across all cases in the event log.

For each occurrence of the activity, records its position within the trace. For example, if ‘A’ occurs 1000 times in position 1 and 500 times in position 2, the returned dictionary will be {1: 1000, 2: 500}.

Parameters:

log – Event log object (EventLog or pandas DataFrame).
activity – The activity to analyze.
activity_key – Attribute to be used for the activity.
timestamp_key – Attribute to be used for the timestamp.
case_id_key – Attribute to be used as the case identifier.

Returns:

A dictionary mapping positions (0-based index) to the number of times the activity occurs in that position.

import pm4py

act_pos = pm4py.get_activity_position_summary(
    dataframe,
    'Act. A',
    activity_key='concept:name',
    case_id_key='case:concept:name',
    timestamp_key='time:timestamp'
)

pm4py.stats module#

This Page

PMTk

PM4Py

Company