483 lines
15 KiB
Python
483 lines
15 KiB
Python
# Split Miner - BPMN process discovery from event logs.
|
|
# Authors:
|
|
# imacat@mail.imacat.idv.tw (imacat), 2026/3/10
|
|
# AI assistance: Claude Code (Anthropic)
|
|
|
|
# Copyright (c) 2026 imacat.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
|
# implied. See the License for the specific language governing
|
|
# permissions and limitations under the License.
|
|
"""Detailed pipeline stage tests for the paper example.
|
|
|
|
Verifies each stage of the Split Miner pipeline using the
|
|
running example from Section 3 of the SM 1.0 paper.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from split_miner import (
|
|
BPMNModel,
|
|
Gateway,
|
|
GatewayType,
|
|
Node,
|
|
Task,
|
|
split_miner,
|
|
)
|
|
from split_miner.concurrency import PrunedDFG
|
|
from split_miner.dfg import DirectlyFollowsGraph
|
|
from split_miner.filtering import FilteredDFG
|
|
|
|
|
|
def _make_tasks(
|
|
labels: str,
|
|
) -> dict[str, Task]:
|
|
"""Create a Task for each single-character label.
|
|
|
|
:param labels: The labels as a string.
|
|
:return: A dict mapping label to Task.
|
|
"""
|
|
return {ch: Task(ch, ch) for ch in labels}
|
|
|
|
|
|
def _make_paper_node_log() -> tuple[
|
|
dict[tuple[Node, ...], int], dict[str, Task]
|
|
]:
|
|
"""Build the paper example log with Node objects.
|
|
|
|
:return: The Node-based traces and the task map.
|
|
"""
|
|
t: dict[str, Task] = _make_tasks("abcdefgh")
|
|
traces: dict[tuple[Node, ...], int] = {
|
|
(t["a"], t["b"], t["c"], t["g"],
|
|
t["e"], t["h"]): 10,
|
|
(t["a"], t["b"], t["c"], t["f"],
|
|
t["g"], t["h"]): 10,
|
|
(t["a"], t["b"], t["d"], t["g"],
|
|
t["e"], t["h"]): 10,
|
|
(t["a"], t["b"], t["d"], t["e"],
|
|
t["g"], t["h"]): 10,
|
|
(t["a"], t["b"], t["e"], t["c"],
|
|
t["g"], t["h"]): 10,
|
|
(t["a"], t["b"], t["e"], t["d"],
|
|
t["g"], t["h"]): 10,
|
|
(t["a"], t["c"], t["b"], t["e"],
|
|
t["g"], t["h"]): 10,
|
|
(t["a"], t["c"], t["b"], t["f"],
|
|
t["g"], t["h"]): 10,
|
|
(t["a"], t["d"], t["b"], t["e"],
|
|
t["g"], t["h"]): 10,
|
|
(t["a"], t["d"], t["b"], t["f"],
|
|
t["g"], t["h"]): 10,
|
|
}
|
|
return traces, t
|
|
|
|
|
|
def _make_paper_str_log() -> dict[
|
|
tuple[str, ...], int
|
|
]:
|
|
"""Build the paper example log with string labels.
|
|
|
|
:return: The string-based traces.
|
|
"""
|
|
return {
|
|
("a", "b", "c", "g", "e", "h"): 10,
|
|
("a", "b", "c", "f", "g", "h"): 10,
|
|
("a", "b", "d", "g", "e", "h"): 10,
|
|
("a", "b", "d", "e", "g", "h"): 10,
|
|
("a", "b", "e", "c", "g", "h"): 10,
|
|
("a", "b", "e", "d", "g", "h"): 10,
|
|
("a", "c", "b", "e", "g", "h"): 10,
|
|
("a", "c", "b", "f", "g", "h"): 10,
|
|
("a", "d", "b", "e", "g", "h"): 10,
|
|
("a", "d", "b", "f", "g", "h"): 10,
|
|
}
|
|
|
|
|
|
class TestDFGAllEdges(unittest.TestCase):
|
|
"""Tests for all DFG edge frequencies (Table 1)."""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up the test.
|
|
|
|
:return: None.
|
|
"""
|
|
traces: dict[tuple[Node, ...], int]
|
|
traces, self.__t = _make_paper_node_log()
|
|
self.__dfg: DirectlyFollowsGraph = (
|
|
DirectlyFollowsGraph(traces)
|
|
)
|
|
|
|
def test_all_edge_frequencies(self) -> None:
|
|
"""All 20 DFG edge frequencies match Table 1.
|
|
|
|
Verifies every directly-follows frequency from the
|
|
paper's example event log.
|
|
"""
|
|
t: dict[str, Task] = self.__t
|
|
expected: dict[tuple[Node, Node], int] = {
|
|
(t["a"], t["b"]): 60,
|
|
(t["a"], t["c"]): 20,
|
|
(t["a"], t["d"]): 20,
|
|
(t["b"], t["c"]): 20,
|
|
(t["b"], t["d"]): 20,
|
|
(t["b"], t["e"]): 40,
|
|
(t["b"], t["f"]): 20,
|
|
(t["c"], t["b"]): 20,
|
|
(t["c"], t["f"]): 10,
|
|
(t["c"], t["g"]): 20,
|
|
(t["d"], t["b"]): 20,
|
|
(t["d"], t["e"]): 10,
|
|
(t["d"], t["g"]): 20,
|
|
(t["e"], t["c"]): 10,
|
|
(t["e"], t["d"]): 10,
|
|
(t["e"], t["g"]): 30,
|
|
(t["e"], t["h"]): 20,
|
|
(t["f"], t["g"]): 30,
|
|
(t["g"], t["e"]): 20,
|
|
(t["g"], t["h"]): 80,
|
|
}
|
|
for (src, tgt), freq in expected.items():
|
|
self.assertEqual(
|
|
self.__dfg.df_frequency(src, tgt),
|
|
freq,
|
|
f"|{src.node_id} -> {tgt.node_id}|"
|
|
f" should be {freq}"
|
|
)
|
|
|
|
def test_all_edges(self) -> None:
|
|
"""The DFG has exactly 20 edges."""
|
|
t: dict[str, Task] = self.__t
|
|
expected: set[tuple[Node, Node]] = {
|
|
(t["a"], t["b"]), (t["a"], t["c"]),
|
|
(t["a"], t["d"]),
|
|
(t["b"], t["c"]), (t["b"], t["d"]),
|
|
(t["b"], t["e"]), (t["b"], t["f"]),
|
|
(t["c"], t["b"]), (t["c"], t["f"]),
|
|
(t["c"], t["g"]),
|
|
(t["d"], t["b"]), (t["d"], t["e"]),
|
|
(t["d"], t["g"]),
|
|
(t["e"], t["c"]), (t["e"], t["d"]),
|
|
(t["e"], t["g"]), (t["e"], t["h"]),
|
|
(t["f"], t["g"]),
|
|
(t["g"], t["e"]), (t["g"], t["h"]),
|
|
}
|
|
self.assertEqual(self.__dfg.edges, expected)
|
|
|
|
def test_edge_count(self) -> None:
|
|
"""The DFG has 20 edges."""
|
|
self.assertEqual(len(self.__dfg.edges), 20)
|
|
|
|
|
|
class TestPrunedDFGEdges(unittest.TestCase):
|
|
"""Tests for PDFG edge set (Section 3.2)."""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up the test.
|
|
|
|
:return: None.
|
|
"""
|
|
traces: dict[tuple[Node, ...], int]
|
|
traces, self.__t = _make_paper_node_log()
|
|
dfg: DirectlyFollowsGraph = (
|
|
DirectlyFollowsGraph(traces)
|
|
)
|
|
self.__pdfg: PrunedDFG = PrunedDFG(
|
|
dfg, epsilon=0.2
|
|
)
|
|
|
|
def test_pdfg_edges(self) -> None:
|
|
"""PDFG has 12 edges after concurrent pruning.
|
|
|
|
Concurrent pairs b||c, b||d, d||e, e||g are
|
|
removed along with their reverse edges.
|
|
"""
|
|
t: dict[str, Task] = self.__t
|
|
expected: set[tuple[Node, Node]] = {
|
|
(t["a"], t["b"]), (t["a"], t["c"]),
|
|
(t["a"], t["d"]),
|
|
(t["b"], t["e"]), (t["b"], t["f"]),
|
|
(t["c"], t["f"]), (t["c"], t["g"]),
|
|
(t["d"], t["g"]),
|
|
(t["e"], t["c"]), (t["e"], t["h"]),
|
|
(t["f"], t["g"]),
|
|
(t["g"], t["h"]),
|
|
}
|
|
self.assertEqual(self.__pdfg.edges, expected)
|
|
|
|
def test_pdfg_edge_count(self) -> None:
|
|
"""PDFG has 12 edges."""
|
|
self.assertEqual(len(self.__pdfg.edges), 12)
|
|
|
|
def test_concurrent_pairs(self) -> None:
|
|
"""All four concurrent pairs are detected."""
|
|
t: dict[str, Task] = self.__t
|
|
self.assertTrue(
|
|
self.__pdfg.is_concurrent(t["b"], t["c"])
|
|
)
|
|
self.assertTrue(
|
|
self.__pdfg.is_concurrent(t["b"], t["d"])
|
|
)
|
|
self.assertTrue(
|
|
self.__pdfg.is_concurrent(t["d"], t["e"])
|
|
)
|
|
self.assertTrue(
|
|
self.__pdfg.is_concurrent(t["e"], t["g"])
|
|
)
|
|
|
|
def test_not_concurrent(self) -> None:
|
|
"""Non-concurrent pairs."""
|
|
t: dict[str, Task] = self.__t
|
|
self.assertFalse(
|
|
self.__pdfg.is_concurrent(t["a"], t["b"])
|
|
)
|
|
self.assertFalse(
|
|
self.__pdfg.is_concurrent(t["c"], t["d"])
|
|
)
|
|
self.assertFalse(
|
|
self.__pdfg.is_concurrent(t["c"], t["f"])
|
|
)
|
|
|
|
|
|
class TestFilteredDFGEdges(unittest.TestCase):
|
|
"""Tests for filtered PDFG edge set (Section 3.3)."""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up the test.
|
|
|
|
:return: None.
|
|
"""
|
|
traces: dict[tuple[Node, ...], int]
|
|
traces, self.__t = _make_paper_node_log()
|
|
dfg: DirectlyFollowsGraph = (
|
|
DirectlyFollowsGraph(traces)
|
|
)
|
|
pdfg: PrunedDFG = PrunedDFG(dfg, epsilon=0.2)
|
|
self.__fdfg: FilteredDFG = FilteredDFG(
|
|
pdfg, eta=0.4
|
|
)
|
|
|
|
def test_filtered_edges(self) -> None:
|
|
"""Filtered PDFG has 10 edges.
|
|
|
|
Edges c->f and e->c are filtered out.
|
|
"""
|
|
t: dict[str, Task] = self.__t
|
|
expected: set[tuple[Node, Node]] = {
|
|
(t["a"], t["b"]), (t["a"], t["c"]),
|
|
(t["a"], t["d"]),
|
|
(t["b"], t["e"]), (t["b"], t["f"]),
|
|
(t["c"], t["g"]),
|
|
(t["d"], t["g"]),
|
|
(t["e"], t["h"]),
|
|
(t["f"], t["g"]),
|
|
(t["g"], t["h"]),
|
|
}
|
|
self.assertEqual(self.__fdfg.edges, expected)
|
|
|
|
def test_filtered_edge_count(self) -> None:
|
|
"""Filtered PDFG has 10 edges."""
|
|
self.assertEqual(len(self.__fdfg.edges), 10)
|
|
|
|
def test_removed_edges(self) -> None:
|
|
"""Edges c->f and e->c are not in filtered PDFG."""
|
|
t: dict[str, Task] = self.__t
|
|
self.assertNotIn(
|
|
(t["c"], t["f"]), self.__fdfg.edges
|
|
)
|
|
self.assertNotIn(
|
|
(t["e"], t["c"]), self.__fdfg.edges
|
|
)
|
|
|
|
|
|
class TestPaperExampleStructure(unittest.TestCase):
|
|
"""Tests for the final paper example structure."""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up the test.
|
|
|
|
:return: None.
|
|
"""
|
|
traces: dict[tuple[str, ...], int] = (
|
|
_make_paper_str_log()
|
|
)
|
|
self.__model: BPMNModel = split_miner(
|
|
traces, epsilon=0.2, eta=0.4
|
|
)
|
|
|
|
def test_node_count(self) -> None:
|
|
"""The final model has 16 nodes."""
|
|
self.assertEqual(
|
|
len(self.__model.all_nodes), 16
|
|
)
|
|
|
|
def test_edge_count(self) -> None:
|
|
"""The final model has 18 edges."""
|
|
self.assertEqual(len(self.__model.edges), 18)
|
|
|
|
def test_gateway_counts(self) -> None:
|
|
"""6 gateways: AND=1, XOR=4, OR=1."""
|
|
gw_types: list[GatewayType] = [
|
|
gw.gateway_type
|
|
for gw in self.__model.gateways.values()
|
|
]
|
|
self.assertEqual(len(gw_types), 6)
|
|
self.assertEqual(
|
|
gw_types.count(GatewayType.AND), 1
|
|
)
|
|
self.assertEqual(
|
|
gw_types.count(GatewayType.XOR), 4
|
|
)
|
|
self.assertEqual(
|
|
gw_types.count(GatewayType.OR), 1
|
|
)
|
|
|
|
def test_and_split_after_a(self) -> None:
|
|
"""Task a leads to an AND split gateway."""
|
|
a: Task = self.__model.get_task("a")
|
|
a_succs: set[Node] = (
|
|
self.__model.successors(a)
|
|
)
|
|
self.assertEqual(len(a_succs), 1)
|
|
and_gw: Node = next(iter(a_succs))
|
|
self.assertIsInstance(and_gw, Gateway)
|
|
assert isinstance(and_gw, Gateway)
|
|
self.assertEqual(
|
|
and_gw.gateway_type, GatewayType.AND
|
|
)
|
|
|
|
def test_and_split_successors(self) -> None:
|
|
"""AND split has successors b and XOR split."""
|
|
a: Task = self.__model.get_task("a")
|
|
and_gw: Node = next(
|
|
iter(self.__model.successors(a))
|
|
)
|
|
and_succs: set[Node] = (
|
|
self.__model.successors(and_gw)
|
|
)
|
|
self.assertEqual(len(and_succs), 2)
|
|
b: Task = self.__model.get_task("b")
|
|
self.assertIn(b, and_succs)
|
|
|
|
def test_xor_split_cd(self) -> None:
|
|
"""XOR split for c and d (successor of AND)."""
|
|
a: Task = self.__model.get_task("a")
|
|
and_gw: Node = next(
|
|
iter(self.__model.successors(a))
|
|
)
|
|
and_succs: set[Node] = (
|
|
self.__model.successors(and_gw)
|
|
)
|
|
b: Task = self.__model.get_task("b")
|
|
xor1: Node = (and_succs - {b}).pop()
|
|
self.assertIsInstance(xor1, Gateway)
|
|
assert isinstance(xor1, Gateway)
|
|
self.assertEqual(
|
|
xor1.gateway_type, GatewayType.XOR
|
|
)
|
|
xor1_succs: set[Node] = (
|
|
self.__model.successors(xor1)
|
|
)
|
|
c: Task = self.__model.get_task("c")
|
|
d: Task = self.__model.get_task("d")
|
|
self.assertEqual(xor1_succs, {c, d})
|
|
|
|
def test_xor_split_ef(self) -> None:
|
|
"""XOR split for e and f (after b)."""
|
|
b: Task = self.__model.get_task("b")
|
|
b_succs: set[Node] = (
|
|
self.__model.successors(b)
|
|
)
|
|
self.assertEqual(len(b_succs), 1)
|
|
xor2: Node = next(iter(b_succs))
|
|
self.assertIsInstance(xor2, Gateway)
|
|
assert isinstance(xor2, Gateway)
|
|
self.assertEqual(
|
|
xor2.gateway_type, GatewayType.XOR
|
|
)
|
|
xor2_succs: set[Node] = (
|
|
self.__model.successors(xor2)
|
|
)
|
|
e: Task = self.__model.get_task("e")
|
|
f: Task = self.__model.get_task("f")
|
|
self.assertEqual(xor2_succs, {e, f})
|
|
|
|
def test_xor_join_cd(self) -> None:
|
|
"""XOR join for c and d."""
|
|
c: Task = self.__model.get_task("c")
|
|
d: Task = self.__model.get_task("d")
|
|
c_succs: set[Node] = (
|
|
self.__model.successors(c)
|
|
)
|
|
d_succs: set[Node] = (
|
|
self.__model.successors(d)
|
|
)
|
|
self.assertEqual(len(c_succs), 1)
|
|
self.assertEqual(len(d_succs), 1)
|
|
self.assertEqual(c_succs, d_succs)
|
|
join: Node = next(iter(c_succs))
|
|
self.assertIsInstance(join, Gateway)
|
|
assert isinstance(join, Gateway)
|
|
self.assertEqual(
|
|
join.gateway_type, GatewayType.XOR
|
|
)
|
|
|
|
def test_or_join_to_g(self) -> None:
|
|
"""OR join for {XOR-join, f} leading to g."""
|
|
g: Task = self.__model.get_task("g")
|
|
g_preds: set[Node] = (
|
|
self.__model.predecessors(g)
|
|
)
|
|
self.assertEqual(len(g_preds), 1)
|
|
or_gw: Node = next(iter(g_preds))
|
|
self.assertIsInstance(or_gw, Gateway)
|
|
assert isinstance(or_gw, Gateway)
|
|
self.assertEqual(
|
|
or_gw.gateway_type, GatewayType.OR
|
|
)
|
|
or_preds: set[Node] = (
|
|
self.__model.predecessors(or_gw)
|
|
)
|
|
self.assertEqual(len(or_preds), 2)
|
|
f: Task = self.__model.get_task("f")
|
|
self.assertIn(f, or_preds)
|
|
|
|
def test_xor_join_to_h(self) -> None:
|
|
"""XOR join for {e, g} leading to h.
|
|
|
|
This was an OR-join that became XOR after
|
|
OR-joins minimization (Algorithm 9).
|
|
"""
|
|
h: Task = self.__model.get_task("h")
|
|
h_preds: set[Node] = (
|
|
self.__model.predecessors(h)
|
|
)
|
|
self.assertEqual(len(h_preds), 1)
|
|
join: Node = next(iter(h_preds))
|
|
self.assertIsInstance(join, Gateway)
|
|
assert isinstance(join, Gateway)
|
|
self.assertEqual(
|
|
join.gateway_type, GatewayType.XOR
|
|
)
|
|
join_preds: set[Node] = (
|
|
self.__model.predecessors(join)
|
|
)
|
|
self.assertEqual(len(join_preds), 2)
|
|
e: Task = self.__model.get_task("e")
|
|
g: Task = self.__model.get_task("g")
|
|
self.assertIn(e, join_preds)
|
|
self.assertIn(g, join_preds)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|