Source code for pm4py.algo.discovery.inductive.variants.im_clean.cuts.sequence

    This file is part of PM4Py (More Info:

    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
    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 <>.
from collections import Counter
from itertools import product

from pm4py.algo.discovery.inductive.variants.im_clean import utils as im_utils
from pm4py.objects.log.obj import EventLog, Trace

[docs]def detect(alphabet, transitive_predecessors, transitive_successors): ''' This method finds a xor cut in the dfg. Implementation follows function XorCut on page 188 of "Robust Process Mining with Guarantees" by Sander J.J. Leemans (ISBN: 978-90-386-4257-4) Basic Steps: 1. create a group per activity 2. merge pairwise reachable nodes (based on transitive relations) 3. merge pairwise unreachable nodes (based on transitive relations) 4. sort the groups based on their reachability Parameters ---------- alphabet characters occurring in the dfg transitive_predecessors dictionary mapping activities to their (transitive) predecessors, according to the DFG transitive_successors dictionary mapping activities to their (transitive) successors, according to the DFG Returns ------- A list of sets of activities, i.e., forming a maximal sequence cut None if no cut is found. ''' groups = [{a} for a in alphabet] if len(groups) == 0: return None for a, b in product(alphabet, alphabet): if (b in transitive_successors[a] and a in transitive_successors[b]) or ( b not in transitive_successors[a] and a not in transitive_successors[b]): groups = im_utils.__merge_groups_for_acts(a, b, groups) groups = list(sorted(groups, key=lambda g: len( transitive_predecessors[next(iter(g))]) + (len(alphabet) - len(transitive_successors[next(iter(g))])))) return groups if len(groups) > 1 else None
[docs]def project(log, groups, activity_key): ''' This method projects the log based on a presumed sequence cut and a list of activity groups Parameters ---------- log original log groups list of activity sets to be used in projection (activities can only appear in one group) activity_key key to use in the event to derive the activity name Returns ------- list of corresponding logs according to the sequence cut. ''' # refactored to support both IM and IMf logs = list() for group in groups: logs.append(EventLog()) for t in log: i = 0 split_point = 0 act_union = set() while i < len(groups): new_split_point = find_split_point(t, groups[i], split_point, act_union, activity_key) trace_i = Trace() j = split_point while j < new_split_point: if t[j][activity_key] in groups[i]: trace_i.append(t[j]) j = j + 1 logs[i].append(trace_i) split_point = new_split_point act_union = act_union.union(set(groups[i])) i = i + 1 return logs
[docs]def find_split_point(t, group, start, ignore, activity_key): least_cost = 0 position_with_least_cost = start cost = 0 i = start while i < len(t): if t[i][activity_key] in group: cost = cost - 1 elif t[i][activity_key] not in ignore: cost = cost + 1 if cost < least_cost: least_cost = cost position_with_least_cost = i + 1 i = i + 1 return position_with_least_cost
def _is_strict_subset(A, B): return A != B and A.issubset(B) def _is_strict_superset(A, B): return A != B and A.issuperset(B)
[docs]def project_dfg(dfg_sa_ea_actcount, groups): start_activities = [] end_activities = [] activities = [] dfgs = [] skippable = [] for g in groups: skippable.append(False) activities_idx = {} for gind, g in enumerate(groups): for act in g: activities_idx[act] = int(gind) i = 0 while i < len(groups): to_succ_arcs = Counter() from_prev_arcs = Counter() if i < len(groups) - 1: for arc in dfg_sa_ea_actcount.dfg: if arc[0] in groups[i] and arc[1] in groups[i + 1]: to_succ_arcs[arc[0]] += dfg_sa_ea_actcount.dfg[arc] if i > 0: for arc in dfg_sa_ea_actcount.dfg: if arc[0] in groups[i - 1] and arc[1] in groups[i]: from_prev_arcs[arc[1]] += dfg_sa_ea_actcount.dfg[arc] if i == 0: start_activities.append({}) for act in dfg_sa_ea_actcount.start_activities: if act in groups[i]: start_activities[i][act] = dfg_sa_ea_actcount.start_activities[act]; else: j = i while j < activities_idx[act]: skippable[j] = True j = j + 1 else: start_activities.append(from_prev_arcs) if i == len(groups) - 1: end_activities.append({}) for act in dfg_sa_ea_actcount.end_activities: if act in groups[i]: end_activities[i][act] = dfg_sa_ea_actcount.end_activities[act] else: j = activities_idx[act] + 1 while j <= i: skippable[j] = True j = j + 1 else: end_activities.append(to_succ_arcs) activities.append({}) for act in groups[i]: activities[i][act] = dfg_sa_ea_actcount.act_count[act] dfgs.append({}) for arc in dfg_sa_ea_actcount.dfg: if arc[0] in groups[i] and arc[1] in groups[i]: dfgs[i][arc] = dfg_sa_ea_actcount.dfg[arc] i = i + 1 i = 0 while i < len(dfgs): dfgs[i] = im_utils.DfgSaEaActCount(dfgs[i], start_activities[i], end_activities[i], activities[i]) i = i + 1 for arc in dfg_sa_ea_actcount.dfg: z = activities_idx[arc[1]] j = activities_idx[arc[0]] + 1 while j < z: skippable[j] = False j = j + 1 return [dfgs, skippable]