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

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 the business hours

Parameters#

st

Start timestamp

et

Complete timestamp

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. [

] meaning that business hours are Mondays 07:00 - 17:00 and Tuesdays 07:00 - 12:00 and 13:00 - 17:00

work_calendar

work calendar (it permits querying if a given day is a working day in a given culture)

Returns#

diff: Difference in business hours

pm4py.util.business_hours.get_overlapping_time(timespan1_begin: datetime, timespan1_end: datetime, timespan2_begin: datetime, timespan2_end: datetime) → float[source]#

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

Bases: object

get_seconds()[source]#

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.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)[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.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