pm4py.util package#

This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General Public License as published by the Free Software Foundation, either version 3 of the License, or any later version.

This program 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 Affero General Public License for more details.

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

Subpackages#

Submodules#

pm4py.util.business_hours module#

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

class pm4py.util.business_hours.BusinessHours(datetime1, datetime2, **kwargs)[source]#

Bases: object

business_seconds_from_week_start(dt)[source]#: Calculate business seconds from the start of the week to the given datetime.

get_seconds()[source]#: Calculate total business seconds between datetime1 and datetime2.

pm4py.util.business_hours.soj_time_business_hours_diff(st: datetime, et: datetime, business_hour_slots: List[Tuple[int]], work_calendar=None) → float[source]#

Calculates the difference between the provided timestamps based on business hours.

Parameters#

stdatetime: Start timestamp
etdatetime: End timestamp
business_hour_slotsList[Tuple[int]]: Work schedule as list of tuples (start, end) in seconds since week start
work_calendar: Work calendar (unused in this implementation)

Returns#

float: Difference in business hours (seconds)

pm4py.util.colors module#

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.util.colors.get_corr_hex(num)[source]#

Gets correspondence between a number and an hexadecimal string

Parameters#

num: Number

Returns#

hex_string: Hexadecimal string

pm4py.util.colors.get_transitions_color(count_move_on_model, count_fit)[source]#

Gets the color associated to the transition

Parameters#

count_move_on_model: Number of move on models
count_fit: Number of fit moves

Returns#

color: Color associated to the transition

pm4py.util.colors.get_string_from_int_below_255(factor)[source]#

Gets a string from an integer below 255

Parameters#

factor: Factor

Returns#

stru: Length 2 string

pm4py.util.constants module#

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.util.constants.get_param_from_env(name, default)[source]#

pm4py.util.constants.get_default_timestamp_format()[source]#

pm4py.util.constants.get_default_xes_timestamp_format()[source]#

pm4py.util.constants.get_default_is_aware_enabled()[source]#

class pm4py.util.constants.AvailableSerializations(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: Enum

EVENT_LOG = 'event_log'#

DATAFRAME = 'dataframe'#

PETRI_NET = 'petri_net'#

PROCESS_TREE = 'process_tree'#

BPMN = 'bpmn'#

DFG = 'dfg'#

pm4py.util.exec_utils module#

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.util.exec_utils.unroll(value)[source]#

pm4py.util.exec_utils.get_param_value(p, parameters, default)[source]#

pm4py.util.exec_utils.get_variant(variant)[source]#

pm4py.util.hie_utils module#

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.util.hie_utils.indent_representation(model_string, max_indent=9223372036854775807)[source]#

pm4py.util.labels_similarity module#

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.util.labels_similarity.label_sets_similarity(labels_a: List[str], labels_b: List[str], threshold: float = 0.75) → float[source]#

Dice-style similarity ∈ [0,1] that tolerates near-matches.

Each label in A is greedily paired with its most similar, still-unmatched label in B (SequenceMatcher ratio ≥ threshold).
Similarity = 2·|matched| / (|A| + |B|).

pm4py.util.labels_similarity.map_labels(labels_a: List[str], labels_b: List[str], threshold: float = 0.75) → Dict[str, str][source]#

Returns a dict {label_in_A: mapped_label}.

If an exact duplicate exists in B, map to that.
Otherwise pick the unused label in B with the highest similarity (SequenceMatcher ratio ≥ threshold).
If nothing reaches the threshold, map to itself (no change).

pm4py.util.ml_utils module#

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.util.ml_utils.DecisionTreeClassifier(*args, **kwargs)[source]#

pm4py.util.ml_utils.AffinityPropagation(*args, **kwargs)[source]#

pm4py.util.ml_utils.KMeans(*args, **kwargs)[source]#

pm4py.util.ml_utils.KNeighborsRegressor(*args, **kwargs)[source]#

pm4py.util.ml_utils.LocallyLinearEmbedding(*args, **kwargs)[source]#

pm4py.util.nx_utils module#

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

class pm4py.util.nx_utils.Parameters(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: Enum

SHOW_PROGRESS_BAR = 'show_progress_bar'#

pm4py.util.nx_utils.get_default_nx_environment()[source]#

pm4py.util.nx_utils.Graph(*args, **kwargs)[source]#

pm4py.util.nx_utils.DiGraph(*args, **kwargs)[source]#

pm4py.util.nx_utils.MultiGraph(*args, **kwargs)[source]#

pm4py.util.nx_utils.MultiDiGraph(*args, **kwargs)[source]#

pm4py.util.nx_utils.ancestors(*args, **kwargs)[source]#

pm4py.util.nx_utils.descendants(*args, **kwargs)[source]#

pm4py.util.nx_utils.connected_components(*args, **kwargs)[source]#

pm4py.util.nx_utils.bfs_tree(*args, **kwargs)[source]#

pm4py.util.nx_utils.contracted_nodes(*args, **kwargs)[source]#

pm4py.util.nx_utils.shortest_path(*args, **kwargs)[source]#

pm4py.util.nx_utils.strongly_connected_components(*args, **kwargs)[source]#

pm4py.util.nx_utils.has_path(*args, **kwargs)[source]#

pm4py.util.nx_utils.is_strongly_connected(*args, **kwargs)[source]#

pm4py.util.nx_utils.all_pairs_shortest_path(*args, **kwargs)[source]#

pm4py.util.nx_utils.all_pairs_dijkstra(*args, **kwargs)[source]#

pm4py.util.nx_utils.find_cliques(*args, **kwargs)[source]#

pm4py.util.nx_utils.degree_centrality(*args, **kwargs)[source]#

pm4py.util.nx_utils.greedy_modularity_communities(*args, **kwargs)[source]#

pm4py.util.nx_utils.maximum_flow_value(*args, **kwargs)[source]#

pm4py.util.nx_utils.minimum_weight_full_matching(*args, **kwargs)[source]#

pm4py.util.nx_utils.Edmonds(*args, **kwargs)[source]#

pm4py.util.nx_utils.neo4j_upload(nx_graph: DiGraph, session, clean_db: bool = True, parameters: Dict[Any, Any] | None = None)[source]#

Uploads a NetworkX DiGraph obtained from a traditional/object-centric event log to a Neo4J session

Parameters#

nx_graph: NetworkX graph
session: Neo4J session
clean_db: Cleans the database before uploading
parameters: Other optional parameters of the method, including: - Parameters.SHOW_PROGRESS_BAR => shows the percentage of nodes/edges uploaded to Neo4J

pm4py.util.nx_utils.neo4j_download(session, parameters: Dict[Any, Any] | None = None) → DiGraph[source]#

Downloads a NetworkX DiGraph starting from a Neo4J database.

Parameters#

session: Neo4J session
parameters: Optional parameters of the method.

Returns#

nx_graph: NetworkX DiGraph

pm4py.util.nx_utils.nx_to_ocel(nx_graph: DiGraph, parameters: Dict[Any, Any] | None = None)[source]#

Transforms a NetworkX DiGraph representing an OCEL to a proper OCEL.

Parameters#

nx_graph: NetworkX DiGraph
parameters: Optional parameters of the method

Returns#

ocel: Object-centric event log

pm4py.util.nx_utils.nx_to_event_log(nx_graph: DiGraph, parameters: Dict[Any, Any] | None = None)[source]#

Transforms a NetworkX DiGraph representing a traditional event log to a proper event log.

Parameters#

nx_graph: NetworkX DiGraph
parameters: Optional parameters of the method

Returns#

event_log: Traditional event log.

pm4py.util.pandas_utils module#

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.util.pandas_utils.get_default_dataframe_environment()[source]#

pm4py.util.pandas_utils.to_dict_records(df)[source]#

Pandas dataframe to dictionary (records method)

Parameters#

df: Dataframe

Returns#

list_dictio: List containing a dictionary for each row

pm4py.util.pandas_utils.to_dict_index(df)[source]#

Pandas dataframe to dictionary (index method)

Parameters#

df: Dataframe

Returns#

dict: dict like {index -> {column -> value}}

pm4py.util.pandas_utils.insert_index(df, column_name='@@index', copy_dataframe=True, reset_index=True)[source]#

Inserts the dataframe index in the specified column

Parameters#

df: Dataframe
column_name: Name of the column that should host the index
copy_dataframe: Establishes if the original dataframe should be copied before inserting the column

Returns#

df: Dataframe with index

pm4py.util.pandas_utils.insert_case_index(df, column_name='@@case_index', case_id='case:concept:name', copy_dataframe=True)[source]#

Inserts the case number in the dataframe

Parameters#

df: Dataframe
column_name: Name of the column that should host the case index
case_id: Case identifier
copy_dataframe: Establishes if the original dataframe should be copied before inserting the column

Returns#

df: Dataframe with case index

pm4py.util.pandas_utils.insert_ev_in_tr_index(df: DataFrame, case_id: str = 'case:concept:name', column_name: str = '@@index_in_trace', copy_dataframe=True) → DataFrame[source]#

Inserts a column that specify the index of the event inside the case

Parameters#

df: Dataframe
case_id: Column that hosts the case identifier
column_name: Name of the column that should host the index

Returns#

df: Dataframe with index

pm4py.util.pandas_utils.format_unique(values)[source]#

pm4py.util.pandas_utils.insert_feature_activity_position_in_trace(df: DataFrame, case_id: str = 'case:concept:name', activity_key: str = 'concept:name', prefix='@@position_')[source]#

Inserts additional columns @@position_ACT1, @@position_ACT2 … which are populated for every event having activity ACT1, ACT2 respectively, with the index of the event inside its case.

Parameters#

df: Pandas dataframe
case_id: Case idntifier
activity_key: Activity
prefix: Prefix of the “activity position in trace” feature (default: @@position_)

Returns#

df: Pandas dataframe

pm4py.util.pandas_utils.insert_case_arrival_finish_rate(log: DataFrame, case_id_column='case:concept:name', timestamp_column='time:timestamp', start_timestamp_column=None, arrival_rate_column='@@arrival_rate', finish_rate_column='@@finish_rate') → DataFrame[source]#

Inserts the arrival/finish rate in the dataframe.

Parameters#

log: Pandas dataframe

Returns#

log: Pandas dataframe enriched by arrival and finish rate

pm4py.util.pandas_utils.insert_case_service_waiting_time(log: DataFrame, case_id_column='case:concept:name', timestamp_column='time:timestamp', start_timestamp_column=None, diff_start_end_column='@@diff_start_end', service_time_column='@@service_time', sojourn_time_column='@@sojourn_time', waiting_time_column='@@waiting_time') → DataFrame[source]#

Inserts the service/waiting/sojourn time in the dataframe.

Parameters#

log: Pandas dataframe
parameters: Parameters of the method

Returns#

log: Pandas dataframe with service, waiting and sojourn time

pm4py.util.pandas_utils.check_is_pandas_dataframe(log)[source]#

Checks if a log object is a dataframe

Parameters#

log: Log object

Returns#

boolean: Is dataframe?

pm4py.util.pandas_utils.instantiate_dataframe(*args, **kwargs)[source]#

pm4py.util.pandas_utils.instantiate_dataframe_from_dict(*args, **kwargs)[source]#

pm4py.util.pandas_utils.instantiate_dataframe_from_records(*args, **kwargs)[source]#

pm4py.util.pandas_utils.get_grouper(*args, **kwargs)[source]#

pm4py.util.pandas_utils.get_total_seconds(difference)[source]#

pm4py.util.pandas_utils.convert_to_seconds(dt_column)[source]#

pm4py.util.pandas_utils.dataframe_column_string_to_datetime(*args, **kwargs)[source]#

pm4py.util.pandas_utils.read_csv(*args, **kwargs)[source]#

pm4py.util.pandas_utils.concat(*args, **kwargs)[source]#

pm4py.util.pandas_utils.merge(*args, **kwargs)[source]#

pm4py.util.pandas_utils.check_pandas_dataframe_columns(df, activity_key=None, case_id_key=None, timestamp_key=None, start_timestamp_key=None)[source]#

Checks if the dataframe contains all the required columns. If not, raise an exception

Parameters#

df: Pandas dataframe

pm4py.util.points_subset module#

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.util.points_subset.pick_chosen_points(m, n)[source]#

Pick chosen points in a list

Parameters#

m: Number of wanted points
n: Number of current points

Returns#

indexes: Indexes of chosen points

pm4py.util.points_subset.pick_chosen_points_list(m, lst, include_extremes=True)[source]#

Pick a chosen number of points from a list

Parameters#

m: Number of wanted points
lst: List

Returns#

reduced_lst: Reduced list

pm4py.util.regex module#

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.util.regex.get_new_char(label, shared_obj)[source]#

Get a new single character describing the activity, for the regex

Parameters#

label: Label of the transition
shared_obj: Shared object

class pm4py.util.regex.SharedObj[source]#: Bases: object

pm4py.util.string_distance module#

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.util.string_distance.levenshtein_distance(s1, s2)[source]#

pm4py.util.string_distance.levenshtein(stru1, stru2)[source]#

pm4py.util.string_distance.argmin_levenshtein(stru: str, list_stri: List[str]) → str | None[source]#

Given a string (stru), finds a string in a list of strings (list_stri) that minimizes the Levenshtein distance.

Parameters#

stru: String (that is compared)
list_stri: List of comparison strings

Returns#

argmin_dist: String (belonging to list_stri) that minimizes the Levenshtein distance with the ‘stru’ argument

pm4py.util.string_distance.argmax_levenshtein(stru: str, list_stri: List[str]) → str | None[source]#

Given a string (stru), finds a string in a list of strings (list_stri) that maximizes the Levenshtein distance.

Parameters#

stru: String (that is compared)
list_stri: List of comparison strings

Returns#

argmax_dist: String (belonging to list_stri) that maximizes the Levenshtein distance with the ‘stru’ argument

pm4py.util.typing module#

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.util.variants_util module#

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

class pm4py.util.variants_util.Parameters(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: Enum

ACTIVITY_KEY = 'pm4py:param:activity_key'#

PARAMETER_VARIANT_DELIMITER = 'variant_delimiter'#

pm4py.util.variants_util.variant_to_trace(variant, parameters=None)[source]#

pm4py.util.variants_util.get_activities_from_variant(variant, parameters=None)[source]#

pm4py.util.variants_util.get_variant_from_trace(trace, parameters=None)[source]#

pm4py.util.variants_util.aggregate_consecutive_activities_in_variants(variants: Dict[Collection[str], int | List], max_repetitions: int = 1) → Dict[Collection[str], int | List][source]#

Aggregate the consecutive activities in the variant.

For example, {(‘A’, ‘B’, ‘C’): 3, (‘A’, ‘B’, ‘B’, ‘B’, ‘C’): 2, (‘A’, ‘B’, ‘B’, ‘B’, ‘B’, ‘B’, ‘C’): 1} Would be reduced to: - {(‘A’, ‘B’, ‘C’): 6} if max_repetitions=1 - {(‘A’, ‘B’, ‘C’): 3, (‘A’, ‘B’, ‘B’, ‘C’): 3} if max_repetitions=2 - {(‘A’, ‘B’, ‘C’): 3, (‘A’, ‘B’, ‘B’, ‘B’, ‘C’): 3} if max_repetitions=3 - {(‘A’, ‘B’, ‘C’): 3, (‘A’, ‘B’, ‘B’, ‘B’, ‘C’): 2, (‘A’, ‘B’, ‘B’, ‘B’, ‘B’, ‘C’): 1} if max_repetitions=4

Parameters#

variants: Dictionary of variants (where each variant is associated to its count (pd.DataFrame) or the list of traces (EventLog))
max_repetitions: Maximum number of consecutive repetitions for an activity

Returns#

aggregated_variants: Dictionary of variants aggregated based on the limit of consecutive repetitions for the same activity

pm4py.util.vis_utils module#

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.util.vis_utils.human_readable_stat(timedelta, stat_locale: Dict[str, str] | None = None) → str[source]#

Transform a timedelta into a human readable string

Parameters#

timedelta: Timedelta

Returns#

string: Human readable string

pm4py.util.vis_utils.get_arc_penwidth(arc_measure, min_arc_measure, max_arc_measure)[source]#

Calculate arc width given the current arc measure value, the minimum arc measure value and the maximum arc measure value

Parameters#

arc_measure: Current arc measure value
min_arc_measure: Minimum measure value among all arcs
max_arc_measure: Maximum measure value among all arcs

Returns#

penwidth: Current arc width in the graph

pm4py.util.vis_utils.get_trans_freq_color(trans_count, min_trans_count, max_trans_count)[source]#

Gets transition frequency color

Parameters#

trans_count: Current transition count
min_trans_count: Minimum transition count
max_trans_count: Maximum transition count

Returns#

color: Frequency color for visible transition

pm4py.util.vis_utils.get_base64_from_gviz(gviz)[source]#

Get base 64 from string content of the file

Parameters#

gviz: Graphviz diagram

Returns#

base64: Base64 string

pm4py.util.vis_utils.get_base64_from_file(temp_file)[source]#

Get base 64 from string content of the file

Parameters#

temp_file: Temporary file path

Returns#

base64: Base64 string

pm4py.util.vis_utils.check_visualization_inside_jupyter()[source]#: Checks if the visualization of the model is performed inside a Jupyter notebook

pm4py.util.vis_utils.view_image_in_jupyter(file_name)[source]#

Visualizes a picture inside the Jupyter notebooks

Parameters#

file_name: Name of the file

pm4py.util.vis_utils.open_opsystem_image_viewer(file_name)[source]#

Visualizes a picture using the image viewer of the operating system

Parameters#

file_name: Name of the file

pm4py.util.vis_utils.get_corr_hex(num)[source]#

Gets correspondence between a number and an hexadecimal string

Parameters#

num: Number

Returns#

hex_string: Hexadecimal string

pm4py.util.vis_utils.value_to_color(val, min_value, max_value)[source]#

pm4py.util.xes_constants module#

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