'''
This file is part of PM4Py (More Info: https://pm4py.fit.fraunhofer.de).
PM4Py is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
PM4Py is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with PM4Py. If not, see <https://www.gnu.org/licenses/>.
'''
__doc__ = """
The ``pm4py.stats`` module contains the statistics offered in ``pm4py``
"""
from typing import Dict, Union, List, Tuple, Collection
from typing import Set, Optional
from collections import Counter
import pandas as pd
from pm4py.objects.log.obj import EventLog, Trace, EventStream
from pm4py.util.pandas_utils import check_is_pandas_dataframe, check_pandas_dataframe_columns, insert_ev_in_tr_index
from pm4py.utils import get_properties, __event_log_deprecation_warning
from pm4py.util import xes_constants, constants
from pm4py.objects.petri_net.obj import PetriNet
from pm4py.objects.process_tree.obj import ProcessTree
from copy import copy
import deprecation
[docs]def get_start_activities(log: Union[EventLog, pd.DataFrame], activity_key: str = "concept:name", timestamp_key: str = "time:timestamp", case_id_key: str = "case:concept:name") -> Dict[str, int]:
"""
Returns the start activities from a log object
:param log: Log object
:param activity_key: attribute to be used for the activity
:param timestamp_key: attribute to be used for the timestamp
:param case_id_key: attribute to be used as case identifier
:rtype: ``Dict[str, int]``
.. code-block:: python3
import pm4py
start_activities = pm4py.get_start_activities(dataframe, activity_key='concept:name', case_id_key='case:concept:name', timestamp_key='time:timestamp')
"""
if type(log) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!")
__event_log_deprecation_warning(log)
properties = get_properties(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
if check_is_pandas_dataframe(log):
check_pandas_dataframe_columns(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
from pm4py.statistics.start_activities.pandas import get
return get.get_start_activities(log, parameters=properties)
else:
from pm4py.statistics.start_activities.log import get
return get.get_start_activities(log, parameters=properties)
[docs]def get_end_activities(log: Union[EventLog, pd.DataFrame], activity_key: str = "concept:name", timestamp_key: str = "time:timestamp", case_id_key: str = "case:concept:name") -> Dict[str, int]:
"""
Returns the end activities of a log
:param log: Log object
:param activity_key: attribute to be used for the activity
:param timestamp_key: attribute to be used for the timestamp
:param case_id_key: attribute to be used as case identifier
:rtype: ``Dict[str, int]``
.. code-block:: python3
import pm4py
end_activities = pm4py.get_end_activities(dataframe, activity_key='concept:name', case_id_key='case:concept:name', timestamp_key='time:timestamp')
"""
if type(log) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!")
__event_log_deprecation_warning(log)
properties = get_properties(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
if check_is_pandas_dataframe(log):
check_pandas_dataframe_columns(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
from pm4py.statistics.end_activities.pandas import get
return get.get_end_activities(log, parameters=properties)
else:
from pm4py.statistics.end_activities.log import get
return get.get_end_activities(log, parameters=properties)
[docs]def get_event_attributes(log: Union[EventLog, pd.DataFrame]) -> List[str]:
"""
Returns the attributes at the event level of the log
:param log: Log object
:rtype: ``List[str]``
.. code-block:: python3
import pm4py
event_attributes = pm4py.get_event_attributes(dataframe)
"""
if type(log) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!")
__event_log_deprecation_warning(log)
if check_is_pandas_dataframe(log):
check_pandas_dataframe_columns(log)
return list(log.columns)
else:
from pm4py.statistics.attributes.log import get
return list(get.get_all_event_attributes_from_log(log))
[docs]def get_trace_attributes(log: Union[EventLog, pd.DataFrame]) -> List[str]:
"""
Gets the attributes at the trace level of a log object
:param log: Log object
:rtype: ``List[str]``
.. code-block:: python3
import pm4py
trace_attributes = pm4py.get_trace_attributes(dataframe)
"""
if type(log) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!")
__event_log_deprecation_warning(log)
from pm4py.util import constants
if check_is_pandas_dataframe(log):
check_pandas_dataframe_columns(log)
return [x for x in list(log.columns) if x.startswith(constants.CASE_ATTRIBUTE_PREFIX)]
else:
from pm4py.statistics.attributes.log import get
return list(get.get_all_trace_attributes_from_log(log))
[docs]def get_event_attribute_values(log: Union[EventLog, pd.DataFrame], attribute: str, count_once_per_case=False, case_id_key: str = "case:concept:name") -> Dict[str, int]:
"""
Returns the values for a specified (event) attribute
:param log: Log object
:param attribute: attribute
:param count_once_per_case: If True, consider only an occurrence of the given attribute value inside a case (if there are multiple events sharing the same attribute value, count only 1 occurrence)
:param case_id_key: attribute to be used as case identifier
:rtype: ``Dict[str, int]``
.. code-block:: python3
import pm4py
activities = pm4py.get_event_attribute_values(dataframe, 'concept:name', case_id_key='case:concept:name')
"""
if type(log) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!")
__event_log_deprecation_warning(log)
parameters = get_properties(log, case_id_key=case_id_key)
parameters["keep_once_per_case"] = count_once_per_case
if check_is_pandas_dataframe(log):
check_pandas_dataframe_columns(log, case_id_key=case_id_key)
from pm4py.statistics.attributes.pandas import get
return get.get_attribute_values(log, attribute, parameters=parameters)
else:
from pm4py.statistics.attributes.log import get
return get.get_attribute_values(log, attribute, parameters=parameters)
[docs]def get_trace_attribute_values(log: Union[EventLog, pd.DataFrame], attribute: str, case_id_key: str = "case:concept:name") -> Dict[str, int]:
"""
Returns the values for a specified trace attribute
:param log: Log object
:param attribute: Attribute
:param case_id_key: attribute to be used as case identifier
:rtype: ``Dict[str, int]``
.. code-block:: python3
import pm4py
tr_attr_values = pm4py.get_trace_attribute_values(dataframe, 'case:attribute', case_id_key='case:concept:name')
"""
if type(log) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!")
__event_log_deprecation_warning(log)
parameters = get_properties(log, case_id_key=case_id_key)
if check_is_pandas_dataframe(log):
check_pandas_dataframe_columns(log, case_id_key=case_id_key)
from pm4py.statistics.attributes.pandas import get
if attribute not in log and constants.CASE_ATTRIBUTE_PREFIX + attribute in log:
# if "attribute" does not exist as column, but "case:attribute" exists, then use that
attribute = constants.CASE_ATTRIBUTE_PREFIX + attribute
ret = get.get_attribute_values(log, attribute, parameters=parameters)
return ret
else:
from pm4py.statistics.attributes.log import get
ret = get.get_trace_attribute_values(log, attribute, parameters=parameters)
if not ret:
# if the provided attribute does not exist, but starts with "case:", try to get the attribute values
# removing the "case:" at the beginning
if attribute.startswith(constants.CASE_ATTRIBUTE_PREFIX):
attribute = attribute.split(constants.CASE_ATTRIBUTE_PREFIX)[-1]
ret = get.get_trace_attribute_values(log, attribute, parameters=parameters)
return ret
[docs]def get_variants(log: Union[EventLog, pd.DataFrame], activity_key: str = "concept:name", timestamp_key: str = "time:timestamp", case_id_key: str = "case:concept:name") -> Dict[Tuple[str], List[Trace]]:
"""
Gets the variants from the log
:param log: Event log
:param activity_key: attribute to be used for the activity
:param timestamp_key: attribute to be used for the timestamp
:param case_id_key: attribute to be used as case identifier
:rtype: ``Dict[Tuple[str], List[Trace]]``
.. code-block:: python3
import pm4py
variants = pm4py.get_variants(dataframe, activity_key='concept:name', case_id_key='case:concept:name', timestamp_key='time:timestamp')
"""
return get_variants_as_tuples(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
[docs]def get_variants_as_tuples(log: Union[EventLog, pd.DataFrame], activity_key: str = "concept:name", timestamp_key: str = "time:timestamp", case_id_key: str = "case:concept:name") -> Dict[Tuple[str], List[Trace]]:
"""
Gets the variants from the log (where the keys are tuples and not strings)
:param log: Event log
:param activity_key: attribute to be used for the activity
:param timestamp_key: attribute to be used for the timestamp
:param case_id_key: attribute to be used as case identifier
:rtype: ``Dict[Tuple[str], List[Trace]]``
.. code-block:: python3
import pm4py
variants = pm4py.get_variants_as_tuples(dataframe, activity_key='concept:name', case_id_key='case:concept:name', timestamp_key='time:timestamp')
"""
if type(log) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!")
__event_log_deprecation_warning(log)
properties = get_properties(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
if check_is_pandas_dataframe(log):
check_pandas_dataframe_columns(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
from pm4py.statistics.variants.pandas import get
return get.get_variants_count(log, parameters=properties)
else:
from pm4py.statistics.variants.log import get
return get.get_variants(log, parameters=properties)
[docs]def get_stochastic_language(*args, **kwargs) -> Dict[List[str], float]:
"""
Gets the stochastic language from the provided object
:param args: Pandas dataframe / event log / accepting Petri net / process tree
:param kwargs: keyword arguments
:rtype: ``Dict[List[str], float]``
.. code-block:: python3
import pm4py
log = pm4py.read_xes('tests/input_data/running-example.xes')
language_log = pm4py.get_stochastic_language(log)
print(language_log)
net, im, fm = pm4py.read_pnml('tests/input_data/running-example.pnml')
language_model = pm4py.get_stochastic_language(net, im, fm)
print(language_model)
"""
from pm4py.statistics.variants.log import get
if isinstance(args[0], EventLog) or isinstance(args[0], EventStream) or isinstance(args[0], pd.DataFrame):
from pm4py.objects.conversion.log import converter as log_converter
log = log_converter.apply(args[0])
return get.get_language(log)
elif isinstance(args[0], PetriNet) or isinstance(args[0], ProcessTree) or isinstance(args[0], dict):
import pm4py
log = pm4py.play_out(*args, **kwargs)
return get.get_language(log)
else:
raise Exception("unsupported input")
[docs]def get_minimum_self_distances(log: Union[EventLog, pd.DataFrame], activity_key: str = "concept:name", timestamp_key: str = "time:timestamp", case_id_key: str = "case:concept:name") -> Dict[str, int]:
'''
This algorithm computes the minimum self-distance for each activity observed in an event log.
The self distance of a in <a> is infinity, of a in <a,a> is 0, in <a,b,a> is 1, etc.
The minimum self distance is the minimal observed self distance value in the event log.
:param log: event log (either pandas.DataFrame, EventLog or EventStream)
:param activity_key: attribute to be used for the activity
:param timestamp_key: attribute to be used for the timestamp
:param case_id_key: attribute to be used as case identifier
:rtype: ``Dict[str, int]``
.. code-block:: python3
import pm4py
msd = pm4py.get_minimum_self_distances(dataframe, activity_key='concept:name', case_id_key='case:concept:name', timestamp_key='time:timestamp')
'''
if type(log) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!")
__event_log_deprecation_warning(log)
if check_is_pandas_dataframe(log):
check_pandas_dataframe_columns(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
properties = get_properties(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
from pm4py.algo.discovery.minimum_self_distance import algorithm as msd_algo
return msd_algo.apply(log, parameters=properties)
[docs]def get_minimum_self_distance_witnesses(log: Union[EventLog, pd.DataFrame], activity_key: str = "concept:name", timestamp_key: str = "time:timestamp", case_id_key: str = "case:concept:name") -> Dict[str, Set[str]]:
"""
This function derives the minimum self distance witnesses.
The self distance of a in <a> is infinity, of a in <a,a> is 0, in <a,b,a> is 1, etc.
The minimum self distance is the minimal observed self distance value in the event log.
A 'witness' is an activity that witnesses the minimum self distance.
For example, if the minimum self distance of activity a in some log L is 2, then,
if trace <a,b,c,a> is in log L, b and c are a witness of a.
:param log: Event Log to use
:param activity_key: attribute to be used for the activity
:param timestamp_key: attribute to be used for the timestamp
:param case_id_key: attribute to be used as case identifier
:rtype: ``Dict[str, Set[str]]``
.. code-block:: python3
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')
"""
if type(log) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!")
__event_log_deprecation_warning(log)
if check_is_pandas_dataframe(log):
check_pandas_dataframe_columns(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
from pm4py.algo.discovery.minimum_self_distance import algorithm as msd_algo
from pm4py.algo.discovery.minimum_self_distance import utils as msdw_algo
return msdw_algo.derive_msd_witnesses(log, msd_algo.apply(log, parameters=get_properties(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)))
[docs]def get_case_arrival_average(log: Union[EventLog, pd.DataFrame], activity_key: str = "concept:name", timestamp_key: str = "time:timestamp", case_id_key: str = "case:concept:name") -> float:
"""
Gets the average difference between the start times of two consecutive cases
:param log: log object
:param activity_key: attribute to be used for the activity
:param timestamp_key: attribute to be used for the timestamp
:param case_id_key: attribute to be used as case identifier
:rtype: ``float``
.. code-block:: python3
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')
"""
if type(log) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!")
__event_log_deprecation_warning(log)
properties = get_properties(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
if check_is_pandas_dataframe(log):
check_pandas_dataframe_columns(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
from pm4py.statistics.traces.generic.pandas import case_arrival
return case_arrival.get_case_arrival_avg(log, parameters=properties)
else:
from pm4py.statistics.traces.generic.log import case_arrival
return case_arrival.get_case_arrival_avg(log, parameters=properties)
[docs]def get_rework_cases_per_activity(log: Union[EventLog, pd.DataFrame], activity_key: str = "concept:name", timestamp_key: str = "time:timestamp", case_id_key: str = "case:concept:name") -> Dict[str, int]:
"""
Find out for which activities of the log the rework (more than one occurrence in the trace for the activity)
occurs.
The output is a dictionary associating to each of the aforementioned activities
the number of cases for which the rework occurred.
:param log: Log object
:param activity_key: attribute to be used for the activity
:param timestamp_key: attribute to be used for the timestamp
:param case_id_key: attribute to be used as case identifier
:rtype: ``Dict[str, int]``
.. code-block:: python3
import pm4py
rework = pm4py.get_rework_cases_per_activity(dataframe, activity_key='concept:name', case_id_key='case:concept:name', timestamp_key='time:timestamp')
"""
if type(log) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!")
__event_log_deprecation_warning(log)
properties = get_properties(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
if check_is_pandas_dataframe(log):
check_pandas_dataframe_columns(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
from pm4py.statistics.rework.pandas import get as rework_get
return rework_get.apply(log, parameters=properties)
else:
from pm4py.statistics.rework.log import get as rework_get
return rework_get.apply(log, parameters=properties)
[docs]@deprecation.deprecated(deprecated_in="2.3.0", removed_in="3.0.0", details="the get_case_overlap function will be removed in a future release.")
def get_case_overlap(log: Union[EventLog, pd.DataFrame], activity_key: str = "concept:name", timestamp_key: str = "time:timestamp", case_id_key: str = "case:concept:name") -> List[int]:
"""
Associates to each case in the log the number of cases concurrently open
:param log: Log object
:param activity_key: attribute to be used for the activity
:param timestamp_key: attribute to be used for the timestamp
:param case_id_key: attribute to be used as case identifier
:rtype: ``List[int]``
.. code-block:: python3
import pm4py
overlap = pm4py.get_case_overlap(dataframe, activity_key='concept:name', case_id_key='case:concept:name', timestamp_key='time:timestamp')
"""
if type(log) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!")
__event_log_deprecation_warning(log)
properties = get_properties(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
if check_is_pandas_dataframe(log):
check_pandas_dataframe_columns(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
from pm4py.statistics.overlap.cases.pandas import get as cases_overlap
return cases_overlap.apply(log, parameters=properties)
else:
from pm4py.statistics.overlap.cases.log import get as cases_overlap
return cases_overlap.apply(log, parameters=properties)
[docs]def get_cycle_time(log: Union[EventLog, pd.DataFrame], activity_key: str = "concept:name", timestamp_key: str = "time:timestamp", case_id_key: str = "case:concept:name") -> float:
"""
Calculates the cycle time of the event log.
The definition that has been followed is the one proposed in:
https://www.presentationeze.com/presentations/lean-manufacturing-just-in-time/lean-manufacturing-just-in-time-full-details/process-cycle-time-analysis/calculate-cycle-time/#:~:text=Cycle%20time%20%3D%20Average%20time%20between,is%2024%20minutes%20on%20average.
So:
Cycle time = Average time between completion of units.
Example taken from the website:
Consider a manufacturing facility, which is producing 100 units of product per 40 hour week.
The average throughput rate is 1 unit per 0.4 hours, which is one unit every 24 minutes.
Therefore the cycle time is 24 minutes on average.
:param log: Log object
:param activity_key: attribute to be used for the activity
:param timestamp_key: attribute to be used for the timestamp
:param case_id_key: attribute to be used as case identifier
:rtype: ``float``
.. code-block:: python3
import pm4py
cycle_time = pm4py.get_cycle_time(dataframe, activity_key='concept:name', case_id_key='case:concept:name', timestamp_key='time:timestamp')
"""
if type(log) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!")
__event_log_deprecation_warning(log)
properties = get_properties(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
if check_is_pandas_dataframe(log):
check_pandas_dataframe_columns(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
from pm4py.statistics.traces.cycle_time.pandas import get as cycle_time
return cycle_time.apply(log, parameters=properties)
else:
from pm4py.statistics.traces.cycle_time.log import get as cycle_time
return cycle_time.apply(log, parameters=properties)
[docs]def get_all_case_durations(log: Union[EventLog, pd.DataFrame], business_hours: bool = False, business_hour_slots=constants.DEFAULT_BUSINESS_HOUR_SLOTS, activity_key: str = "concept:name", timestamp_key: str = "time:timestamp", case_id_key: str = "case:concept:name") -> List[float]:
"""
Gets the durations of the cases in the event log
:param log: Event log
:param business_hours: Enables/disables the computation based on the business hours (default: False)
:param business_hour_slots: work schedule of the company, provided as a list of tuples where each tuple represents one time slot of business hours. One slot i.e. one tuple consists of one start and one end time given in seconds since week start, e.g. [(7 * 60 * 60, 17 * 60 * 60), ((24 + 7) * 60 * 60, (24 + 12) * 60 * 60), ((24 + 13) * 60 * 60, (24 + 17) * 60 * 60),] meaning that business hours are Mondays 07:00 - 17:00 and Tuesdays 07:00 - 12:00 and 13:00 - 17:00
:param activity_key: attribute to be used for the activity
:param timestamp_key: attribute to be used for the timestamp
:param case_id_key: attribute to be used as case identifier
:rtype: ``List[float]``
.. code-block:: python3
import pm4py
case_durations = pm4py.get_all_case_durations(dataframe, activity_key='concept:name', case_id_key='case:concept:name', timestamp_key='time:timestamp')
"""
if type(log) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!")
__event_log_deprecation_warning(log)
properties = get_properties(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
properties["business_hours"] = business_hours
properties["business_hour_slots"] = business_hour_slots
if check_is_pandas_dataframe(log):
check_pandas_dataframe_columns(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
from pm4py.statistics.traces.generic.pandas import case_statistics
cd = case_statistics.get_cases_description(log, parameters=properties)
return sorted([x["caseDuration"] for x in cd.values()])
else:
from pm4py.statistics.traces.generic.log import case_statistics
return case_statistics.get_all_case_durations(log, parameters=properties)
[docs]def get_case_duration(log: Union[EventLog, pd.DataFrame], case_id: str, business_hours: bool = False, business_hour_slots=constants.DEFAULT_BUSINESS_HOUR_SLOTS, activity_key: str = "concept:name", timestamp_key: str = "time:timestamp", case_id_key: Optional[str] = None) -> float:
"""
Gets the duration of a specific case
:param log: Event log
:param case_id: Case identifier
:param business_hours: Enables/disables the computation based on the business hours (default: False)
:param business_hour_slots: work schedule of the company, provided as a list of tuples where each tuple represents one time slot of business hours. One slot i.e. one tuple consists of one start and one end time given in seconds since week start, e.g. [(7 * 60 * 60, 17 * 60 * 60), ((24 + 7) * 60 * 60, (24 + 12) * 60 * 60), ((24 + 13) * 60 * 60, (24 + 17) * 60 * 60),] meaning that business hours are Mondays 07:00 - 17:00 and Tuesdays 07:00 - 12:00 and 13:00 - 17:00
:param activity_key: attribute to be used for the activity
:param timestamp_key: attribute to be used for the timestamp
:param case_id_key: attribute to be used as case identifier
:rtype: ``float``
.. code-block:: python3
import pm4py
duration = pm4py.get_case_duration(dataframe, 'case 1', activity_key='concept:name', case_id_key='case:concept:name', timestamp_key='time:timestamp')
"""
if type(log) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!")
__event_log_deprecation_warning(log)
properties = get_properties(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
properties["business_hours"] = business_hours
properties["business_hour_slots"] = business_hour_slots
if check_is_pandas_dataframe(log):
check_pandas_dataframe_columns(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
from pm4py.statistics.traces.generic.pandas import case_statistics
cd = case_statistics.get_cases_description(log, parameters=properties)
return cd[case_id]["caseDuration"]
else:
from pm4py.statistics.traces.generic.log import case_statistics
cd = case_statistics.get_cases_description(log, parameters=properties)
return cd[case_id]["caseDuration"]
[docs]def get_activity_position_summary(log: Union[EventLog, pd.DataFrame], activity: str, activity_key: str = "concept:name", timestamp_key: str = "time:timestamp", case_id_key: str = "case:concept:name") -> Dict[int, int]:
"""
Given an event log, returns a dictionary which summarize the positions
of the activities in the different cases of the event log.
E.g., if an activity happens 1000 times in the position 1 (the second event of a case),
and 500 times in the position 2 (the third event of a case), then the returned dictionary would be:
{1: 1000, 2: 500}
:param log: Event log object / Pandas dataframe
:param activity: Activity to consider
:param activity_key: attribute to be used for the activity
:param timestamp_key: attribute to be used for the timestamp
:param case_id_key: attribute to be used as case identifier
:rtype: ``Dict[int, int]``
.. code-block:: python3
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')
"""
if type(log) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!")
__event_log_deprecation_warning(log)
if check_is_pandas_dataframe(log):
check_pandas_dataframe_columns(log, activity_key=activity_key, timestamp_key=timestamp_key, case_id_key=case_id_key)
log = insert_ev_in_tr_index(log, case_id_key, "@@index_in_trace")
ret = log[log[activity_key] == activity]["@@index_in_trace"].value_counts().to_dict()
return ret
else:
ret = Counter()
for trace in log:
for i in range(len(trace)):
this_act = trace[i][activity_key]
if this_act == activity:
ret[i] += 1
return dict(ret)