diff --git a/src/split_miner/refined_dfg.py b/src/split_miner/refined_dfg.py index 76f2a84..cd7d758 100644 --- a/src/split_miner/refined_dfg.py +++ b/src/split_miner/refined_dfg.py @@ -73,11 +73,6 @@ class RefinedDirectlyFollowsGraph: ) -> None: """Build the refined DFG from lifecycle traces. - For each trace, scan for end events. After each - end event, collect all start events that occur - before the next end event. These form the - directly-follows pairs per Definition 6. - :param traces: The event log. """ for trace, count in traces.items(): @@ -98,17 +93,6 @@ class RefinedDirectlyFollowsGraph: if lifecycle == "end": self.__sinks.add(node) break - # Detect self-loops: activity with multiple - # complete lifecycles in a trace. - end_counts: dict[Node, int] = {} - for node, lifecycle in trace: - if lifecycle == "end": - end_counts[node] = ( - end_counts.get(node, 0) + 1 - ) - for node, cnt in end_counts.items(): - if cnt > 1: - self.__self_loops.add(node) # Definition 6: ax ->r ay iff ay starts # after ax ends with no other end event # between. @@ -121,12 +105,8 @@ class RefinedDirectlyFollowsGraph: ) -> None: """Scan a single trace for refined DF relations. - Walk through events. When we see an end event - for activity ax, record ax as a "pending source". - When we see a start event for ay, create edges - from all pending sources to ay. When we see - another end event, clear all pending sources - (since the new end event is "between"). + A self-relation ax ->r ax is recorded as a + self-loop instead of an edge. :param trace: The lifecycle trace. :param count: The trace frequency. @@ -144,16 +124,16 @@ class RefinedDirectlyFollowsGraph: # All pending sources directly-follow # to this activity. for src in pending: - if src != node: - pair: tuple[Node, Node] = ( - src, node - ) - self.__df_freq[pair] = ( - self.__df_freq.get( - pair, 0 - ) - + count - ) + if src == node: + self.__self_loops.add(node) + continue + pair: tuple[Node, Node] = ( + src, node + ) + self.__df_freq[pair] = ( + self.__df_freq.get(pair, 0) + + count + ) @property def nodes(self) -> set[Node]: @@ -190,8 +170,11 @@ class RefinedDirectlyFollowsGraph: def self_loops(self) -> set[Node]: """The set of self-loop nodes. - An activity is a self-loop if it completes - (has an end event) more than once in any trace. + An activity a is a self-loop if a ->r a holds per + Definition 6, that is, in some trace an end event + of a is followed by a start event of a with no + other end event in between. Such self-relations + are excluded from :attr:`edges`. :return: The self-loop nodes. """ diff --git a/tests/test_refined_dfg.py b/tests/test_refined_dfg.py index 68dff4b..11b735e 100644 --- a/tests/test_refined_dfg.py +++ b/tests/test_refined_dfg.py @@ -455,15 +455,16 @@ class TestForkAfterEnd(unittest.TestCase): class TestSelfLoop(unittest.TestCase): """Tests self-loop detection from lifecycle traces. - An activity that completes more than once in a trace - is a self-loop. + An activity a is a self-loop iff a ->r a holds per + Definition 6, that is, an end event of a is followed by + a start event of a with no other end event in between. """ def setUp(self) -> None: - """Set up trace with self-loop: A repeats. + """Set up trace with self-loop: A repeats at once. - Trace: As Ae Bs Be As Ae Cs Ce - Activity A completes twice. + Trace: As Ae As Ae Bs Be Cs Ce + A ends and immediately starts again, so A ->r A. :return: None. """ @@ -477,8 +478,8 @@ class TestSelfLoop(unittest.TestCase): tuple[tuple[Task, str], ...], int ] = { ((a, S), (a, E), - (b, S), (b, E), (a, S), (a, E), + (b, S), (b, E), (c, S), (c, E)): 1, } self.__dfg: RefinedDirectlyFollowsGraph = ( @@ -505,6 +506,27 @@ class TestSelfLoop(unittest.TestCase): self.__dfg.self_loops, ) + def test_no_self_edge(self) -> None: + """The self-loop edge A->A is not in the edges. + + :return: None. + """ + a: Task = self.__tasks["A"] + self.assertNotIn((a, a), self.__dfg.edges) + + def test_other_edges(self) -> None: + """The other edges are A->B and B->C. + + :return: None. + """ + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + self.assertEqual( + self.__dfg.edges, + {(a, b), (b, c)}, + ) + def test_no_self_loop_in_paper(self) -> None: """Paper example has no self-loops. @@ -514,5 +536,100 @@ class TestSelfLoop(unittest.TestCase): self.assertEqual(dfg.self_loops, set()) +class TestSelfLoopWithoutRepeatedEnd(unittest.TestCase): + """Tests a self-loop with a single end event. + + Definition 6 only requires an end event of a followed by + a start event of a, not a second end event of a. + """ + + def setUp(self) -> None: + """Set up trace: As Ae As Bs Be. + + A ends, then A starts again with no end event in + between, so A ->r A. + + :return: None. + """ + self.__tasks: dict[str, Task] = ( + _make_tasks("A", "B") + ) + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + traces: dict[ + tuple[tuple[Task, str], ...], int + ] = { + ((a, S), (a, E), (a, S), + (b, S), (b, E)): 1, + } + self.__dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph(traces) + ) + + def test_a_is_self_loop(self) -> None: + """A is a self-loop even though it ends once. + + :return: None. + """ + self.assertIn( + self.__tasks["A"], + self.__dfg.self_loops, + ) + + +class TestTwoLoopIsNotSelfLoop(unittest.TestCase): + """Tests that a 2-loop is not mistaken for a self-loop. + + An activity repeating after another activity completes + forms a 2-loop, not a self-loop. + """ + + def setUp(self) -> None: + """Set up trace: As Ae Bs Be As Ae. + + B ends between the end of A and the next start of + A, so A ->r A does not hold. + + :return: None. + """ + self.__tasks: dict[str, Task] = ( + _make_tasks("A", "B") + ) + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + traces: dict[ + tuple[tuple[Task, str], ...], int + ] = { + ((a, S), (a, E), + (b, S), (b, E), + (a, S), (a, E)): 1, + } + self.__dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph(traces) + ) + + def test_a_not_self_loop(self) -> None: + """A is not a self-loop. + + :return: None. + """ + self.assertNotIn( + self.__tasks["A"], + self.__dfg.self_loops, + ) + + def test_two_loop_edges(self) -> None: + """The edges form the 2-loop A->B and B->A. + + :return: None. + """ + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + self.assertEqual( + self.__dfg.edges, + {(a, b), (b, a)}, + ) + + if __name__ == "__main__": unittest.main()