Source code for pm4py.algo.discovery.inductive.variants.im_clean.utils

'''
    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/>.
'''
from pm4py.objects.process_tree import obj as pt


[docs]def transform_dfg_to_directed_nx_graph(dfg, alphabet): import networkx as nx nx_graph = nx.DiGraph() nx_graph.add_nodes_from(alphabet) for a, b in dfg: nx_graph.add_edge(a, b) return nx_graph
def __merge_groups_for_acts(a, b, groups): group_a = None group_b = None for group in groups: if a in group: group_a = group if b in group: group_b = group groups = [group for group in groups if group != group_a and group != group_b] groups.append(group_a.union(group_b)) return groups def __filter_dfg_on_threshold(dfg, end_activities, threshold): outgoing_max_occ = {} for x, y in dfg.items(): act = x[0] if act not in outgoing_max_occ: outgoing_max_occ[act] = y else: outgoing_max_occ[act] = max(y, outgoing_max_occ[act]) if act in end_activities: outgoing_max_occ[act] = max(outgoing_max_occ[act], end_activities[act]) dfg_list = sorted([(x, y) for x, y in dfg.items()], key=lambda x: (x[1], x[0]), reverse=True) dfg_list = [x for x in dfg_list if x[1] > threshold * outgoing_max_occ[x[0][0]]] dfg_list = [x[0] for x in dfg_list] # filter the elements in the DFG dfg = {x: y for x, y in dfg.items() if x in dfg_list} return dfg def __flower(alphabet, root): operator = pt.ProcessTree(operator=pt.Operator.LOOP, parent=root) operator.children.append(pt.ProcessTree(parent=operator)) xor = pt.ProcessTree(operator=pt.Operator.XOR) operator.children.append(xor) for a in alphabet: tree = pt.ProcessTree(label=a, parent=xor) xor.children.append(tree) return operator
[docs]class DfgSaEaActCount(object): def __init__(self, dfg, sa, ea, act_count): self.dfg = dfg self.start_activities = sa self.end_activities = ea self.act_count = act_count def __str__(self): return str((self.dfg, self.start_activities, self.end_activities, self.act_count)) def __repr__(self): return str((self.dfg, self.start_activities, self.end_activities, self.act_count))