imacatandClaude Opus 5
d35d866b4b
Adds the Split Miner 2.0 pipeline (Augusto, Dumas & La Rosa, 2021) alongside the existing 1.0 implementation: - refined_dfg: refined DFG from activity lifecycle events (Definition 6) - refined_concurrency: true concurrency from lifecycle overlap (Equation 5) - heuristics: fix improper completion from AND-split loop-edges, and detect OR-splits from mutual exclusiveness (Section 3.3) - miner.split_miner_2: the 2.0 entry point, reusing the 1.0 filtering, splits, joins and OR-join minimization steps Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
424 lines
12 KiB
Python
424 lines
12 KiB
Python
# Split Miner - BPMN process discovery from event logs.
|
|
# Authors:
|
|
# imacat@mail.imacat.idv.tw (imacat), 2026/3/12
|
|
# 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.
|
|
"""Tests for SM 2.0 heuristics (Section 3.3).
|
|
|
|
Heuristic 1: AND-split with loop-edge -> preceding XOR-split
|
|
to fix improper completion.
|
|
Heuristic 2: AND-split with pairwise mutual exclusiveness
|
|
-> OR-split.
|
|
|
|
Reference:
|
|
Augusto, A., Dumas, M., & La Rosa, M. (2021).
|
|
Automated Discovery of Process Models with True
|
|
Concurrency and Inclusive Choices. Section 3.3.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from split_miner.bpmn import (
|
|
BPMNModel,
|
|
EndEvent,
|
|
Gateway,
|
|
GatewayType,
|
|
Node,
|
|
StartEvent,
|
|
Task,
|
|
)
|
|
from split_miner.heuristics import (
|
|
fix_improper_completion,
|
|
detect_or_splits,
|
|
)
|
|
|
|
S: str = "start"
|
|
"""Lifecycle start constant."""
|
|
E: str = "end"
|
|
"""Lifecycle end constant."""
|
|
|
|
|
|
def _make_tasks(
|
|
*labels: str,
|
|
) -> dict[str, Task]:
|
|
"""Create Task objects from labels.
|
|
|
|
:param labels: The activity labels.
|
|
:return: A dict mapping label to Task.
|
|
"""
|
|
return {
|
|
label: Task(label, label)
|
|
for label in labels
|
|
}
|
|
|
|
|
|
class TestFixImproperCompletion(unittest.TestCase):
|
|
"""Tests Heuristic 1: fix improper completion.
|
|
|
|
For each AND-split with a loop-edge (leading to a
|
|
topologically earlier node), create a preceding
|
|
XOR-split to carry the loop-edge.
|
|
"""
|
|
|
|
def _make_model_with_and_loop(
|
|
self,
|
|
) -> tuple[BPMNModel, dict[str, Node]]:
|
|
"""Build a model with AND-split + loop-edge.
|
|
|
|
Structure:
|
|
start -> a -> AND-split -> b -> join -> e
|
|
-> c -> join
|
|
-> a (loop-edge)
|
|
join -> e -> end
|
|
|
|
The AND-split has 3 outgoing: b, c, a.
|
|
Edge AND-split -> a is a loop-edge (goes
|
|
back to the topologically earlier node a).
|
|
|
|
:return: The model and named nodes.
|
|
"""
|
|
start: StartEvent = StartEvent("start")
|
|
end: EndEvent = EndEvent("end")
|
|
model: BPMNModel = BPMNModel(start, end)
|
|
|
|
tasks: dict[str, Task] = _make_tasks(
|
|
"a", "b", "c", "e"
|
|
)
|
|
for t in tasks.values():
|
|
model.add_task(t)
|
|
|
|
and_split: Gateway = model.create_gateway(
|
|
GatewayType.AND
|
|
)
|
|
and_join: Gateway = model.create_gateway(
|
|
GatewayType.AND
|
|
)
|
|
|
|
model.add_edge(start, tasks["a"])
|
|
model.add_edge(tasks["a"], and_split)
|
|
model.add_edge(and_split, tasks["b"])
|
|
model.add_edge(and_split, tasks["c"])
|
|
# Loop-edge: AND-split -> a (back to a)
|
|
model.add_edge(and_split, tasks["a"])
|
|
model.add_edge(tasks["b"], and_join)
|
|
model.add_edge(tasks["c"], and_join)
|
|
model.add_edge(and_join, tasks["e"])
|
|
model.add_edge(tasks["e"], end)
|
|
|
|
nodes: dict[str, Node] = {
|
|
"start": start, "end": end,
|
|
"and_split": and_split,
|
|
"and_join": and_join,
|
|
}
|
|
nodes.update(tasks)
|
|
return model, nodes
|
|
|
|
def test_and_split_preserved(self) -> None:
|
|
"""AND-split still exists after fix.
|
|
|
|
:return: None.
|
|
"""
|
|
model: BPMNModel
|
|
nodes: dict[str, Node]
|
|
model, nodes = (
|
|
self._make_model_with_and_loop()
|
|
)
|
|
fix_improper_completion(model)
|
|
and_split: Node = nodes["and_split"]
|
|
self.assertIn(and_split, model.all_nodes)
|
|
|
|
def test_xor_split_created(self) -> None:
|
|
"""A new XOR-split is created before AND-split.
|
|
|
|
:return: None.
|
|
"""
|
|
model: BPMNModel
|
|
nodes: dict[str, Node]
|
|
model, nodes = (
|
|
self._make_model_with_and_loop()
|
|
)
|
|
fix_improper_completion(model)
|
|
and_split: Node = nodes["and_split"]
|
|
# AND-split should now have a single
|
|
# predecessor that is the new XOR-split.
|
|
preds: set[Node] = model.predecessors(
|
|
and_split
|
|
)
|
|
xor_preds: list[Node] = [
|
|
p for p in preds
|
|
if isinstance(p, Gateway)
|
|
and p.gateway_type == GatewayType.XOR
|
|
]
|
|
self.assertEqual(len(xor_preds), 1)
|
|
|
|
def test_loop_edge_moved_to_xor(self) -> None:
|
|
"""Loop-edge originates from XOR-split, not AND.
|
|
|
|
The loop-edge to a should now be routed through
|
|
the XOR-split: XOR -> a (loop), XOR -> AND -> b,c.
|
|
|
|
:return: None.
|
|
"""
|
|
model: BPMNModel
|
|
nodes: dict[str, Node]
|
|
model, nodes = (
|
|
self._make_model_with_and_loop()
|
|
)
|
|
fix_improper_completion(model)
|
|
and_split: Node = nodes["and_split"]
|
|
a: Task = nodes["a"]
|
|
# a should NOT be a successor of AND-split
|
|
and_successors: set[Node] = (
|
|
model.successors(and_split)
|
|
)
|
|
self.assertNotIn(a, and_successors)
|
|
|
|
def test_and_keeps_non_loop_edges(self) -> None:
|
|
"""AND-split keeps its non-loop outgoing edges.
|
|
|
|
b and c should still be successors of AND-split.
|
|
|
|
:return: None.
|
|
"""
|
|
model: BPMNModel
|
|
nodes: dict[str, Node]
|
|
model, nodes = (
|
|
self._make_model_with_and_loop()
|
|
)
|
|
fix_improper_completion(model)
|
|
and_split: Node = nodes["and_split"]
|
|
b: Task = nodes["b"]
|
|
c: Task = nodes["c"]
|
|
successors: set[Node] = (
|
|
model.successors(and_split)
|
|
)
|
|
self.assertIn(b, successors)
|
|
self.assertIn(c, successors)
|
|
|
|
def test_no_change_without_loop(self) -> None:
|
|
"""No changes when AND-split has no loop-edge.
|
|
|
|
:return: None.
|
|
"""
|
|
start: StartEvent = StartEvent("start")
|
|
end: EndEvent = EndEvent("end")
|
|
model: BPMNModel = BPMNModel(start, end)
|
|
tasks: dict[str, Task] = _make_tasks(
|
|
"a", "b", "c"
|
|
)
|
|
for t in tasks.values():
|
|
model.add_task(t)
|
|
and_split: Gateway = model.create_gateway(
|
|
GatewayType.AND
|
|
)
|
|
model.add_edge(start, and_split)
|
|
model.add_edge(and_split, tasks["a"])
|
|
model.add_edge(and_split, tasks["b"])
|
|
model.add_edge(tasks["a"], tasks["c"])
|
|
model.add_edge(tasks["b"], tasks["c"])
|
|
model.add_edge(tasks["c"], end)
|
|
edges_before: set[tuple[Node, Node]] = (
|
|
set(model.edges)
|
|
)
|
|
fix_improper_completion(model)
|
|
self.assertEqual(model.edges, edges_before)
|
|
|
|
|
|
class TestDetectOrSplits(unittest.TestCase):
|
|
"""Tests Heuristic 2: detect OR-splits.
|
|
|
|
For each AND-split, check if successor activities
|
|
are pairwise both concurrent and mutually exclusive
|
|
in different traces. If the majority of pairs
|
|
qualify, convert AND to OR.
|
|
"""
|
|
|
|
def _make_or_candidate_model(
|
|
self,
|
|
) -> tuple[
|
|
BPMNModel,
|
|
dict[str, Node],
|
|
dict[tuple[tuple[Task, str], ...], int],
|
|
]:
|
|
"""Build a model with an AND-split that should be OR.
|
|
|
|
Model: start -> a -> AND-split -> b -> join
|
|
-> c -> join
|
|
-> d -> join
|
|
join -> e -> end
|
|
|
|
Traces (from paper's Lrho_y):
|
|
- {As,Ae,Bs,Cs,Ds,Be,De,Ce,Es,Ee} x3
|
|
(B,C,D all present, overlapping)
|
|
- {As,Ae,Cs,Ds,Ce,De,Es,Ee} x2
|
|
(no B - B and C mutually exclusive)
|
|
- {As,Ae,Bs,Ds,De,Be,Es,Ee} x1
|
|
(no C - B and C mutually exclusive)
|
|
|
|
:return: The model, nodes, and traces.
|
|
"""
|
|
start: StartEvent = StartEvent("start")
|
|
end: EndEvent = EndEvent("end")
|
|
model: BPMNModel = BPMNModel(start, end)
|
|
|
|
tasks: dict[str, Task] = _make_tasks(
|
|
"a", "b", "c", "d", "e"
|
|
)
|
|
for t in tasks.values():
|
|
model.add_task(t)
|
|
|
|
and_split: Gateway = model.create_gateway(
|
|
GatewayType.AND
|
|
)
|
|
and_join: Gateway = model.create_gateway(
|
|
GatewayType.AND
|
|
)
|
|
|
|
model.add_edge(start, tasks["a"])
|
|
model.add_edge(tasks["a"], and_split)
|
|
model.add_edge(and_split, tasks["b"])
|
|
model.add_edge(and_split, tasks["c"])
|
|
model.add_edge(and_split, tasks["d"])
|
|
model.add_edge(tasks["b"], and_join)
|
|
model.add_edge(tasks["c"], and_join)
|
|
model.add_edge(tasks["d"], and_join)
|
|
model.add_edge(and_join, tasks["e"])
|
|
model.add_edge(tasks["e"], end)
|
|
|
|
a: Task = tasks["a"]
|
|
b: Task = tasks["b"]
|
|
c: Task = tasks["c"]
|
|
d: Task = tasks["d"]
|
|
e: Task = tasks["e"]
|
|
|
|
traces: dict[
|
|
tuple[tuple[Task, str], ...], int
|
|
] = {
|
|
# All three present, overlapping
|
|
((a, S), (a, E), (b, S), (c, S),
|
|
(d, S), (b, E), (d, E), (c, E),
|
|
(e, S), (e, E)): 3,
|
|
# No B (B,C mutually exclusive)
|
|
((a, S), (a, E), (c, S), (d, S),
|
|
(c, E), (d, E),
|
|
(e, S), (e, E)): 2,
|
|
# No C (B,C mutually exclusive)
|
|
((a, S), (a, E), (b, S), (d, S),
|
|
(d, E), (b, E),
|
|
(e, S), (e, E)): 1,
|
|
}
|
|
|
|
nodes: dict[str, Node] = {
|
|
"start": start, "end": end,
|
|
"and_split": and_split,
|
|
"and_join": and_join,
|
|
}
|
|
nodes.update(tasks)
|
|
return model, nodes, traces
|
|
|
|
def test_and_becomes_or(self) -> None:
|
|
"""AND-split is converted to OR-split.
|
|
|
|
B,C: concurrent 3x, exclusive 3x -> eligible.
|
|
B,D: concurrent 4x, exclusive 2x -> eligible.
|
|
C,D: concurrent 5x, exclusive 1x -> NOT eligible.
|
|
2/3 pairs eligible -> majority -> OR.
|
|
|
|
:return: None.
|
|
"""
|
|
model: BPMNModel
|
|
nodes: dict[str, Node]
|
|
traces: dict[
|
|
tuple[tuple[Task, str], ...], int
|
|
]
|
|
model, nodes, traces = (
|
|
self._make_or_candidate_model()
|
|
)
|
|
and_split: Gateway = nodes["and_split"]
|
|
detect_or_splits(model, traces)
|
|
self.assertEqual(
|
|
and_split.gateway_type, GatewayType.OR,
|
|
)
|
|
|
|
def test_join_also_becomes_or(self) -> None:
|
|
"""Corresponding join is also converted to OR.
|
|
|
|
:return: None.
|
|
"""
|
|
model: BPMNModel
|
|
nodes: dict[str, Node]
|
|
traces: dict[
|
|
tuple[tuple[Task, str], ...], int
|
|
]
|
|
model, nodes, traces = (
|
|
self._make_or_candidate_model()
|
|
)
|
|
and_join: Gateway = nodes["and_join"]
|
|
detect_or_splits(model, traces)
|
|
self.assertEqual(
|
|
and_join.gateway_type, GatewayType.OR,
|
|
)
|
|
|
|
def test_no_change_when_always_concurrent(
|
|
self,
|
|
) -> None:
|
|
"""AND-split stays AND when no mutual exclusion.
|
|
|
|
If all successors always co-occur, no pair
|
|
is mutually exclusive, so AND stays AND.
|
|
|
|
:return: None.
|
|
"""
|
|
start: StartEvent = StartEvent("start")
|
|
end: EndEvent = EndEvent("end")
|
|
model: BPMNModel = BPMNModel(start, end)
|
|
tasks: dict[str, Task] = _make_tasks(
|
|
"a", "b", "c"
|
|
)
|
|
for t in tasks.values():
|
|
model.add_task(t)
|
|
and_split: Gateway = model.create_gateway(
|
|
GatewayType.AND
|
|
)
|
|
and_join: Gateway = model.create_gateway(
|
|
GatewayType.AND
|
|
)
|
|
model.add_edge(start, tasks["a"])
|
|
model.add_edge(tasks["a"], and_split)
|
|
model.add_edge(and_split, tasks["b"])
|
|
model.add_edge(and_split, tasks["c"])
|
|
model.add_edge(tasks["b"], and_join)
|
|
model.add_edge(tasks["c"], and_join)
|
|
model.add_edge(and_join, end)
|
|
a: Task = tasks["a"]
|
|
b: Task = tasks["b"]
|
|
c: Task = tasks["c"]
|
|
traces: dict[
|
|
tuple[tuple[Task, str], ...], int
|
|
] = {
|
|
((a, S), (a, E), (b, S), (c, S),
|
|
(b, E), (c, E)): 5,
|
|
}
|
|
detect_or_splits(model, traces)
|
|
self.assertEqual(
|
|
and_split.gateway_type, GatewayType.AND,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|