pm4py.algo.conformance.alignments.dfg.variants package#

PM4Py – A Process Mining Library for Python

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

Submodules#

pm4py.algo.conformance.alignments.dfg.variants.classic module#

PM4Py – A Process Mining Library for Python

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.algo.conformance.alignments.dfg.variants.classic.Parameters(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: Enum

CASE_ID_KEY = 'pm4py:param:case_id_key'#

ACTIVITY_KEY = 'pm4py:param:activity_key'#

TIMESTAMP_KEY = 'pm4py:param:timestamp_key'#

SYNC_COST_FUNCTION = 'sync_cost_function'#

MODEL_MOVE_COST_FUNCTION = 'model_move_cost_function'#

LOG_MOVE_COST_FUNCTION = 'log_move_cost_function'#

INTERNAL_LOG_MOVE_COST_FUNCTION = 'internal_log_move_cost_function'#

PARAMETER_VARIANT_DELIMITER = 'variant_delimiter'#

class pm4py.algo.conformance.alignments.dfg.variants.classic.Outputs(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: Enum

ALIGNMENT = 'alignment'#

COST = 'cost'#

VISITED = 'visited_states'#

CLOSED = 'closed'#

INTERNAL_COST = 'internal_cost'#

pm4py.algo.conformance.alignments.dfg.variants.classic.apply(obj: EventLog | Trace, dfg: Dict[Tuple[str, str], int], sa: Dict[str, int], ea: Dict[str, int], parameters: Dict[str | Parameters, Any] | None = None) → Dict[str, Any] | List[Dict[str, Any]][source]#

Applies the alignment algorithm provided a log/trace object, and a connected DFG

Parameters#

obj: Event log / Trace
dfg: Connected directly-Follows Graph
sa: Start activities
ea: End activities
parameters: Parameters of the algorithm: - Parameters.SYNC_COST_FUNCTION: for each activity that is in both the trace and the model, provide the non-negative cost of a sync move - Parameters.MODEL_MOVE_COST_FUNCTION: for each activity that is in the model, provide the non-negative cost of a model move - Parameters.LOG_MOVE_COST_FUNCTION: for each activity that is in the trace, provide the cost of a log move that is returned in the alignment to the user (but not used internally for ordering the states) - Parameters.INTERNAL_LOG_MOVE_COST_FUNCTION: for each activity that is in the trace, provide the cost of a log move that is used internally for ordering the states in the search algorithm. - Parameters.ACTIVITY_KEY: the attribute of the log that is the activity

Returns#

ali: Result of the alignment

pm4py.algo.conformance.alignments.dfg.variants.classic.apply_log(log, dfg, sa, ea, parameters=None)[source]#

Applies the alignment algorithm provided a log object, and a connected DFG

Parameters#

log: Event log
dfg: Connected DFG
sa: Start activities
ea: End activities
parameters: Parameters of the algorithm: - Parameters.SYNC_COST_FUNCTION: for each activity that is in both the trace and the model, provide the non-negative cost of a sync move - Parameters.MODEL_MOVE_COST_FUNCTION: for each activity that is in the model, provide the non-negative cost of a model move - Parameters.LOG_MOVE_COST_FUNCTION: for each activity that is in the trace, provide the cost of a log move that is returned in the alignment to the user (but not used internally for ordering the states) - Parameters.INTERNAL_LOG_MOVE_COST_FUNCTION: for each activity that is in the trace, provide the cost of a log move that is used internally for ordering the states in the search algorithm. - Parameters.ACTIVITY_KEY: the attribute of the log that is the activity

Returns#

aligned_traces: For each trace, contains a dictionary describing the alignment

pm4py.algo.conformance.alignments.dfg.variants.classic.apply_trace(trace, dfg, sa, ea, parameters=None)[source]#

Applies the alignment algorithm provided a trace of a log, and a connected DFG

Parameters#

trace: Trace
dfg: Connected DFG
sa: Start activities
ea: End activities
parameters: Parameters of the algorithm: - Parameters.SYNC_COST_FUNCTION: for each activity that is in both the trace and the model, provide the non-negative cost of a sync move - Parameters.MODEL_MOVE_COST_FUNCTION: for each activity that is in the model, provide the non-negative cost of a model move - Parameters.LOG_MOVE_COST_FUNCTION: for each activity that is in the trace, provide the cost of a log move that is returned in the alignment to the user (but not used internally for ordering the states) - Parameters.INTERNAL_LOG_MOVE_COST_FUNCTION: for each activity that is in the trace, provide the cost of a log move that is used internally for ordering the states in the search algorithm. - Parameters.ACTIVITY_KEY: the attribute of the log that is the activity

Returns#

ali: Dictionary describing the alignment

pm4py.algo.conformance.alignments.dfg.variants.classic.apply_from_variants_list_dfg_string(var_list, dfg_serialization, parameters=None)[source]#

pm4py.algo.conformance.alignments.dfg.variants.classic.apply_from_variants_list(var_list, dfg, sa, ea, parameters=None)[source]#

pm4py.algo.conformance.alignments.dfg.variants.classic.apply_from_variant(variant, dfg, sa, ea, parameters=None)[source]#

pm4py.algo.conformance.alignments.dfg.variants.classic.dijkstra_to_end_node(dfg, sa, ea, start_node, end_node, activities_model, sync_cost_function, model_move_cost_function)[source]#

Gets the cost of the minimum path from a node to the end node

Parameters#

dfg: Connected DFG
sa: Start activities
ea: End activities
start_node: Start node of the graph (connected to all the start activities)
end_node: End node of the graph (connected to all the end activities)
activities_model: Set of the activities contained in the DFG
sync_cost_function: Given an activity, provides the cost when the activity is executed in a sync way
model_move_cost_function: Given an activity, provides the cost when the activity is executed as a move-on-model

Returns#

cost_to_end_node: Dictionary associating to each node the cost to the end node

pm4py.algo.conformance.alignments.dfg.variants.classic.project_log_on_dfg(log: EventLog | DataFrame, dfg: Dict[Tuple[str, str], int], sa: Dict[str, int], ea: Dict[str, int], parameters: Dict[str | Parameters, Any] | None = None) → EventLog[source]#

Projects the traces of an event log to the specified DFG, in order to assess the conformance of the different directly-follows relationships and their performance (as the timestamps are recorded). The result is a event log where each ‘projected’ trace can be replayed on the given DFG. Each event of a ‘projected’ trace has the ‘@@is_conforming’ attribute set to: - True when the activity is mimicked by the original trace (sync move) - False when the activity is not reflected in the original trace (move-on-model) Move-on-log (activities of the trace that are not mimicked by the DFG) are skipped altogether.

Minimum Viable Example:

import pm4py from pm4py.algo.conformance.alignments.dfg.variants import classic as dfg_alignments

log = pm4py.read_xes(“tests/input_data/receipt.xes”, return_legacy_log_object=True) filtered_log = pm4py.filter_variants_top_k(log, 1) dfg, sa, ea = pm4py.discover_dfg(filtered_log)

projected_log = dfg_alignments.project_log_on_dfg(log, dfg, sa, ea) pm4py.write_xes(projected_log, “projected_log.xes”)

Parameters#

log

Event log

dfg

Directly-Follows Graph

sa

Start activities

ea

End activities

parameters

Parameters of the method, including:

Parameters.ACTIVITY_KEY => the attribute of the event log to be used as activity
Parameters.TIMESTAMP_KEY => the attribute of the event log to be used as timestamp

Returns#

projected_log: Projected event log with the aforementioned features

pm4py.algo.conformance.alignments.dfg.variants.classic.project_alignments_on_dfg(log: EventLog | DataFrame, dfg: Dict[Tuple[str, str], int], sa: Dict[str, int], ea: Dict[str, int], parameters: Dict[str | Parameters, Any] | None = None) → Tuple[Dict[Tuple[str, str], Dict[str, int]], Dict[str, Dict[str, int]]][source]#

Projects alignment results onto the directly-follows graph (DFG) and activity dictionary.

This method performs a DFG-based alignment between an event log (or DataFrame) and a process model represented by a directly-follows graph. The alignment moves are categorized into three types:

SYNC (Synchronous Move): Both the log event and the model activity match.

MM (Model Move): The model activity is executed without a corresponding log event.

LM (Log Move): A log event occurs without a corresponding model activity.

After computing the alignments using the provided log and DFG (via the apply_log function), the method “projects” these results by aggregating:

For each DFG edge (transition), the number of SYNC and MM moves.

For each activity, the number of SYNC, MM, and LM moves.

Parameters#

logUnion[EventLog, pd.DataFrame]: The event log or DataFrame containing the events.
dfgDict[Tuple[str, str], int]: A dictionary representing the directly-follows graph where each key is a tuple of activities (from, to) and each value is the frequency of that transition.
saDict[str, int]: A dictionary of start activities with their corresponding frequencies.
eaDict[str, int]: A dictionary of end activities with their corresponding frequencies.
parametersOptional[Dict[Union[str, Parameters], Any]], optional: Additional parameters to control the alignment process, by default None.

Returns#

Tuple[Dict[Tuple[str, str], Dict[str, int]], Dict[str, Dict[str, int]]]

A tuple containing:

conformance_dfg: A dictionary where each key is an edge (tuple of activities) from the DFG, and the value is a dictionary with keys:

‘sync’ : count of synchronous moves on that edge. ‘mm’ : count of model moves on that edge.
conformance_activities: A dictionary where each key is an activity and the value is a dictionary with keys:

‘sync’ : count of synchronous moves for the activity. ‘mm’ : count of model moves for the activity. ‘lm’ : count of log moves for the activity.