feat(pm4py): incongruent parameters specification (start/end node)

incongruent parameters specification (start/end node)
process-intelligence-solutions · Oct 4, 2021 · 374e8f5 · 374e8f5
1 parent 6348fff
commit 374e8f5
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 12 deletions.
diff --git a/pm4py/algo/simulation/playout/dfg/variants/performance.py b/pm4py/algo/simulation/playout/dfg/variants/performance.py
@@ -16,6 +16,8 @@ class Parameters(Enum):
     CASE_ID_KEY = constants.PARAMETER_CONSTANT_CASEID_KEY
     CASE_ARRIVAL_RATE = "case_arrival_rate"
     PERFORMANCE_DFG = "performance_dfg"
+    PARAM_ARTIFICIAL_START_ACTIVITY = constants.PARAM_ARTIFICIAL_START_ACTIVITY
+    PARAM_ARTIFICIAL_END_ACTIVITY = constants.PARAM_ARTIFICIAL_END_ACTIVITY
 
 
 def dict_based_choice(dct: Dict[str, float]) -> str:
@@ -86,13 +88,16 @@ def apply(frequency_dfg: Dict[Tuple[str, str], int], start_activities: Dict[str,
     performance_dfg = copy(exec_utils.get_param_value(Parameters.PERFORMANCE_DFG, parameters, None))
     frequency_dfg = copy(frequency_dfg)
 
+    artificial_start_activity = exec_utils.get_param_value(Parameters.PARAM_ARTIFICIAL_START_ACTIVITY, parameters, constants.DEFAULT_ARTIFICIAL_START_ACTIVITY)
+    artificial_end_activity = exec_utils.get_param_value(Parameters.PARAM_ARTIFICIAL_END_ACTIVITY, parameters, constants.DEFAULT_ARTIFICIAL_END_ACTIVITY)
+
     for sa in start_activities:
-        frequency_dfg[("▶", sa)] = start_activities[sa]
-        performance_dfg[("▶", sa)] = 0
+        frequency_dfg[(artificial_start_activity, sa)] = start_activities[sa]
+        performance_dfg[(artificial_start_activity, sa)] = 0
 
     for ea in end_activities:
-        frequency_dfg[(ea, "■")] = end_activities[ea]
-        performance_dfg[(ea, "■")] = 0
+        frequency_dfg[(ea, artificial_end_activity)] = end_activities[ea]
+        performance_dfg[(ea, artificial_end_activity)] = 0
 
     choices = {}
     for el in frequency_dfg:
@@ -112,10 +117,10 @@ def apply(frequency_dfg: Dict[Tuple[str, str], int], start_activities: Dict[str,
         curr_t = curr_st
         trace = Trace(attributes={case_id_key: str(i)})
         log.append(trace)
-        curr_act = "▶"
+        curr_act = artificial_start_activity
         while True:
             next_act = dict_based_choice(choices[curr_act])
-            if next_act == "■" or next_act is None:
+            if next_act == artificial_end_activity or next_act is None:
                 break
             perf = performance_dfg[(curr_act, next_act)]
             if type(perf) is dict:

diff --git a/pm4py/objects/conversion/dfg/variants/to_petri_net_invisibles_no_duplicates.py b/pm4py/objects/conversion/dfg/variants/to_petri_net_invisibles_no_duplicates.py
@@ -5,12 +5,14 @@
 from pm4py.objects.dfg.utils import dfg_utils
 from pm4py.objects.petri_net.obj import PetriNet, Marking
 from pm4py.objects.petri_net.utils import petri_utils
-from pm4py.util import exec_utils
+from pm4py.util import exec_utils, constants
 
 
 class Parameters(Enum):
     START_ACTIVITIES = 'start_activities'
     END_ACTIVITIES = 'end_activities'
+    PARAM_ARTIFICIAL_START_ACTIVITY = constants.PARAM_ARTIFICIAL_START_ACTIVITY
+    PARAM_ARTIFICIAL_END_ACTIVITY = constants.PARAM_ARTIFICIAL_END_ACTIVITY
 
 
 def apply(dfg: Dict[Tuple[str, str], int], parameters: Optional[Dict[Any, Any]] = None):
@@ -43,12 +45,14 @@ def apply(dfg: Dict[Tuple[str, str], int], parameters: Optional[Dict[Any, Any]]
                                                       dfg)})
     end_activities = exec_utils.get_param_value(Parameters.END_ACTIVITIES, parameters,
                                                 {x: 1 for x in dfg_utils.infer_end_activities(dfg)})
+    artificial_start_activity = exec_utils.get_param_value(Parameters.PARAM_ARTIFICIAL_START_ACTIVITY, parameters, constants.DEFAULT_ARTIFICIAL_START_ACTIVITY)
+    artificial_end_activity = exec_utils.get_param_value(Parameters.PARAM_ARTIFICIAL_END_ACTIVITY, parameters, constants.DEFAULT_ARTIFICIAL_END_ACTIVITY)
 
     enriched_dfg = copy(dfg)
     for act in start_activities:
-        enriched_dfg[("▶", act)] = start_activities[act]
+        enriched_dfg[(artificial_start_activity, act)] = start_activities[act]
     for act in end_activities:
-        enriched_dfg[(act, "■")] = end_activities[act]
+        enriched_dfg[(act, artificial_end_activity)] = end_activities[act]
     activities = set(x[1] for x in enriched_dfg).union(set(x[0] for x in enriched_dfg))
     net = PetriNet("")
     im = Marking()
@@ -60,7 +64,7 @@ def apply(dfg: Dict[Tuple[str, str], int], parameters: Optional[Dict[Any, Any]]
         pl1 = PetriNet.Place("source_" + act)
         pl2 = PetriNet.Place("sink_" + act)
         trans = PetriNet.Transition("trans_" + act, act)
-        if act in ["▶", "■"]:
+        if act in [artificial_start_activity, artificial_end_activity]:
             trans.label = None
         net.places.add(pl1)
         net.places.add(pl2)
@@ -75,7 +79,7 @@ def apply(dfg: Dict[Tuple[str, str], int], parameters: Optional[Dict[Any, Any]]
         net.transitions.add(hidden)
         petri_utils.add_arc_from_to(right_places[arc[0]], hidden, net)
         petri_utils.add_arc_from_to(hidden, left_places[arc[1]], net)
-    im[left_places["▶"]] = 1
-    fm[right_places["■"]] = 1
+    im[left_places[artificial_start_activity]] = 1
+    fm[right_places[artificial_end_activity]] = 1
 
     return net, im, fm
diff --git a/pm4py/util/constants.py b/pm4py/util/constants.py
@@ -34,6 +34,11 @@
 
 DEFAULT_ENCODING = "utf-8"
 
+PARAM_ARTIFICIAL_START_ACTIVITY = "pm4py:param:art_start_act"
+PARAM_ARTIFICIAL_END_ACTIVITY = "pm4py:param:art_end_act"
+DEFAULT_ARTIFICIAL_START_ACTIVITY = "▶"
+DEFAULT_ARTIFICIAL_END_ACTIVITY = "■"
+
 from enum import Enum