imacat/split-miner
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix self-loop detection to use the Definition 6 self-relation

Browse Source 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
imacat
2026-08-20 10:00:45 +08:00
co-authored by Claude Fable 5
parent 11045c57d3
commit 79bb561693
2 changed files with 140 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/split_miner/refined_dfg.py
+17 -34
View File
@@ -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.
"""
tests/test_refined_dfg.py
+123 -6
View File
@@ -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()