From d35d866b4bb66bc6045b6df3926f61dbd6088704 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BE=9D=E7=91=AA=E8=B2=93?= Date: Thu, 12 Mar 2026 12:36:37 +0800 Subject: [PATCH] Add Split Miner 2.0 implementation 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 --- src/split_miner/heuristics.py | 323 ++++++++++++ src/split_miner/miner.py | 112 +++- src/split_miner/refined_concurrency.py | 329 ++++++++++++ src/split_miner/refined_dfg.py | 240 +++++++++ tests/test_heuristics.py | 423 +++++++++++++++ tests/test_refined_concurrency.py | 693 +++++++++++++++++++++++++ tests/test_refined_dfg.py | 518 ++++++++++++++++++ tests/test_split_miner_2.py | 332 ++++++++++++ 8 files changed, 2969 insertions(+), 1 deletion(-) create mode 100644 src/split_miner/heuristics.py create mode 100644 src/split_miner/refined_concurrency.py create mode 100644 src/split_miner/refined_dfg.py create mode 100644 tests/test_heuristics.py create mode 100644 tests/test_refined_concurrency.py create mode 100644 tests/test_refined_dfg.py create mode 100644 tests/test_split_miner_2.py diff --git a/src/split_miner/heuristics.py b/src/split_miner/heuristics.py new file mode 100644 index 0000000..e0ee69a --- /dev/null +++ b/src/split_miner/heuristics.py @@ -0,0 +1,323 @@ +# 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. +"""SM 2.0 heuristics for improper completion and OR-splits. + +Heuristic 1 (fix_improper_completion): For each AND-split +with a loop-edge (leading to a topologically earlier node), +create a preceding XOR-split and move the loop-edge to it. + +Heuristic 2 (detect_or_splits): For each AND-split, check +pairwise whether successor activities are both concurrent +and mutually exclusive in different traces. If the majority +of pairs qualify, convert the AND-split to 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 + +from collections.abc import Iterator + +from split_miner.bpmn import ( + BPMNModel, + Gateway, + GatewayType, + Node, + Task, +) + + +def fix_improper_completion( + model: BPMNModel, +) -> None: + """Fix improper completion from AND-split loop-edges. + + For each AND-split with an outgoing edge that is a + loop-edge (target is topologically before the source + in the process graph), insert a preceding XOR-split + and move the loop-edge to the XOR-split. + + :param model: The BPMN model to modify in place. + """ + topo_order: dict[Node, int] = ( + _compute_topo_order(model) + ) + # Collect AND-splits (list avoids mutation issues). + and_splits: list[Gateway] = [ + gw for gw in model.gateways.values() + if gw.gateway_type == GatewayType.AND + and len(model.outgoing_edges(gw)) > 1 + ] + for gw in and_splits: + loop_targets: list[Node] = [] + for _, tgt in model.outgoing_edges(gw): + gw_order: int = topo_order.get(gw, -1) + tgt_order: int = topo_order.get(tgt, -1) + if tgt_order <= gw_order: + loop_targets.append(tgt) + if not loop_targets: + continue + # Create XOR-split before AND-split. + xor: Gateway = model.create_gateway( + GatewayType.XOR + ) + # Redirect incoming edges of AND-split to XOR. + in_edges: set[tuple[Node, Node]] = ( + model.incoming_edges(gw) + ) + for src, _ in in_edges: + model.redirect_edge_target(src, gw, xor) + # XOR -> AND (forward path) + model.add_edge(xor, gw) + # Move loop-edges from AND to XOR. + for tgt in loop_targets: + model.remove_edge(gw, tgt) + model.add_edge(xor, tgt) + + +def _compute_topo_order( + model: BPMNModel, +) -> dict[Node, int]: + """Compute topological order via BFS from start. + + Nodes reachable from start get increasing order. + Nodes not reachable get order -1. + + :param model: The BPMN model. + :return: A dict mapping node to topological order. + """ + order: dict[Node, int] = {} + visited: set[Node] = set() + queue: list[Node] = [model.start] + idx: int = 0 + while queue: + node: Node = queue.pop(0) + if node in visited: + continue + visited.add(node) + order[node] = idx + idx += 1 + for succ in sorted( + model.successors(node), + key=lambda n: ( + type(n).__name__, n.node_id + ), + ): + if succ not in visited: + queue.append(succ) + return order + + +def detect_or_splits( + model: BPMNModel, + traces: dict[ + tuple[tuple[Node, str], ...], int + ], +) -> None: + """Detect AND-splits that should be OR-splits. + + For each AND-split, examine successor activities + pairwise. A pair is "eligible for inclusiveness" + if they are concurrent in some traces and mutually + exclusive in others, with at least 1 exclusive + observation per 2 concurrent observations (or + vice-versa). + + If the majority of pairs are eligible, convert the + AND-split (and its corresponding AND-join) to OR. + + :param model: The BPMN model to modify in place. + :param traces: The lifecycle traces. + """ + # Pre-compute activity presence per trace. + trace_activities: list[ + tuple[set[Node], int] + ] = [] + trace_overlaps: list[ + tuple[set[tuple[Node, Node]], int] + ] = [] + for trace, count in traces.items(): + activities: set[Node] = set() + overlaps: set[tuple[Node, Node]] = set() + active: set[Node] = set() + for node, lifecycle in trace: + if lifecycle == "start": + for other in active: + overlaps.add( + _ordered_pair(other, node) + ) + active.add(node) + elif lifecycle == "end": + active.discard(node) + activities.add(node) + trace_activities.append( + (activities, count) + ) + trace_overlaps.append( + (overlaps, count) + ) + + and_splits: list[Gateway] = [ + gw for gw in model.gateways.values() + if gw.gateway_type == GatewayType.AND + and len(model.outgoing_edges(gw)) > 1 + ] + + for gw in and_splits: + # Get successor tasks (skip gateways). + succ_tasks: list[Task] = _get_leaf_tasks( + model, gw + ) + if len(succ_tasks) < 2: + continue + # Check pairwise. + eligible: int = 0 + total: int = 0 + for i, a in enumerate(succ_tasks): + for b in succ_tasks[i + 1:]: + total += 1 + pair: tuple[Node, Node] = ( + _ordered_pair(a, b) + ) + conc: int = 0 + excl: int = 0 + for j, (acts, cnt) in enumerate( + trace_activities + ): + a_in: bool = a in acts + b_in: bool = b in acts + ovlps: set[ + tuple[Node, Node] + ] = trace_overlaps[j][0] + if a_in and b_in: + if pair in ovlps: + conc += cnt + elif a_in or b_in: + excl += cnt + if conc > 0 and excl > 0: + # Footnote 4: at least 1 per 2 + hi: int = max(conc, excl) + lo: int = min(conc, excl) + if hi <= 2 * lo: + eligible += 1 + if total > 0 and eligible > total / 2: + gw.gateway_type = GatewayType.OR + # Find and update corresponding join. + _update_corresponding_join(model, gw) + + +def _ordered_pair( + a: Node, b: Node +) -> tuple[Node, Node]: + """Return the pair in canonical order. + + :param a: First node. + :param b: Second node. + :return: The ordered pair. + """ + key_a: tuple[str, str] = ( + type(a).__name__, a.node_id + ) + key_b: tuple[str, str] = ( + type(b).__name__, b.node_id + ) + if key_a <= key_b: + return (a, b) + return (b, a) + + +def _get_leaf_tasks( + model: BPMNModel, gw: Gateway +) -> list[Task]: + """Get the leaf Task successors of a gateway. + + Follows through intermediate gateways to find + the actual Task nodes. + + :param model: The BPMN model. + :param gw: The gateway. + :return: The leaf tasks. + """ + result: list[Task] = [] + queue: list[Node] = list(model.successors(gw)) + visited: set[Node] = set() + while queue: + node: Node = queue.pop(0) + if node in visited: + continue + visited.add(node) + if isinstance(node, Task): + result.append(node) + elif isinstance(node, Gateway): + queue.extend(model.successors(node)) + return sorted( + result, + key=lambda t: t.node_id, + ) + + +def _update_corresponding_join( + model: BPMNModel, split_gw: Gateway +) -> None: + """Find and update the corresponding join gateway. + + Finds the AND-join that merges the paths from + the split gateway and converts it to OR. + + :param model: The BPMN model. + :param split_gw: The split gateway (now OR). + """ + # BFS from split to find the convergence point. + for node in _bfs_nodes(model, split_gw): + if not isinstance(node, Gateway): + continue + if node == split_gw: + continue + if node.gateway_type != GatewayType.AND: + continue + in_edges: set[tuple[Node, Node]] = ( + model.incoming_edges(node) + ) + if len(in_edges) > 1: + node.gateway_type = GatewayType.OR + return + + +def _bfs_nodes( + model: BPMNModel, start: Node +) -> Iterator[Node]: + """BFS traversal from a starting node. + + :param model: The BPMN model. + :param start: The starting node. + :return: Iterator of nodes in BFS order. + """ + visited: set[Node] = set() + queue: list[Node] = [start] + while queue: + node: Node = queue.pop(0) + if node in visited: + continue + visited.add(node) + yield node + for succ in model.successors(node): + if succ not in visited: + queue.append(succ) diff --git a/src/split_miner/miner.py b/src/split_miner/miner.py index ef89a62..6518c31 100644 --- a/src/split_miner/miner.py +++ b/src/split_miner/miner.py @@ -26,6 +26,12 @@ SM 1.0 paper): 4. Splits discovery 5. Joins discovery 6. OR-joins minimization + +Also provides Split Miner 2.0 which adds: +- Refined DFG using activity lifecycle (Definition 6) +- True concurrency via lifecycle overlap (Equation 5) +- Heuristic 1: fix improper completion from loop-edges +- Heuristic 2: detect OR-splits from mutual exclusiveness """ from __future__ import annotations @@ -41,8 +47,20 @@ from split_miner.bpmn import ( from split_miner.concurrency import PrunedDFG from split_miner.dfg import DirectlyFollowsGraph from split_miner.filtering import FilteredDFG +from split_miner.heuristics import ( + detect_or_splits, + fix_improper_completion, +) from split_miner.joins import discover_joins -from split_miner.or_minimization import replace_or_joins +from split_miner.or_minimization import ( + replace_or_joins, +) +from split_miner.refined_concurrency import ( + RefinedPrunedDFG, +) +from split_miner.refined_dfg import ( + RefinedDirectlyFollowsGraph, +) from split_miner.splits import discover_splits @@ -117,6 +135,98 @@ def split_miner( return model +def split_miner_2( + traces: dict[ + tuple[tuple[str, str], ...], int + ], + epsilon: float = 0.33, + eta: float = 0.8, +) -> BPMNModel: + """Run Split Miner 2.0 to discover a BPMN model. + + Uses lifecycle-aware traces (start/end events) for + refined DFG construction and true concurrency + discovery, plus heuristics for improper completion + and OR-split detection. + + :param traces: The input event log as a dict mapping + each trace (tuple of (activity_label, lifecycle) + pairs where lifecycle is "start" or "end") to + its frequency. + :param epsilon: Concurrency threshold (0 to 1). + Minimum overlap ratio to consider two activities + concurrent (Equation 5). + :param eta: Filtering percentile (0 to 1). + Lower values retain more edges. + :return: The discovered BPMN process model. + """ + # Pre-process: convert string traces to Node traces + task_map: dict[str, Task] = {} + node_traces: dict[ + tuple[tuple[Node, str], ...], int + ] = {} + for trace, count in traces.items(): + if not trace: + continue + node_list: list[tuple[Node, str]] = [] + for label, lifecycle in trace: + if label not in task_map: + task_map[label] = Task( + label, label + ) + node_list.append( + (task_map[label], lifecycle) + ) + node_trace: tuple[ + tuple[Node, str], ... + ] = tuple(node_list) + node_traces[node_trace] = ( + node_traces.get(node_trace, 0) + count + ) + + # Step 1: Refined DFG (Definition 6) + dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph(node_traces) + ) + + # Step 2: Refined concurrency (Equation 5) + pdfg: RefinedPrunedDFG = RefinedPrunedDFG( + dfg, node_traces, epsilon + ) + + # Step 3: Filtering (shared with SM 1.0) + fdfg: FilteredDFG = FilteredDFG( + pdfg, eta # type: ignore[arg-type] + ) + + # Step 4-6: Convert to BPMN and discover gateways + start: StartEvent = StartEvent("start") + end: EndEvent = EndEvent("end") + model: BPMNModel = _build_initial_model( + fdfg, start, end + ) + + # Step 4: Splits discovery + discover_splits(model, fdfg.is_concurrent) + + # Step 5: Joins discovery + discover_joins(model) + + # Step 6: OR-joins minimization + replace_or_joins(model) + + # Heuristic 1: Fix improper completion + fix_improper_completion(model) + + # Heuristic 2: Detect OR-splits + detect_or_splits(model, node_traces) + + # Restore self-loops (last step, same as SM 1.0) + _restore_self_loops(model, dfg.self_loops) + + return model + + def _build_initial_model( fdfg: FilteredDFG, start: StartEvent, diff --git a/src/split_miner/refined_concurrency.py b/src/split_miner/refined_concurrency.py new file mode 100644 index 0000000..693abbe --- /dev/null +++ b/src/split_miner/refined_concurrency.py @@ -0,0 +1,329 @@ +# 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. +"""Refined concurrency discovery (SM 2.0, Equation 5). + +Uses activity lifecycle overlap to discover true concurrency: +two activities A and B are concurrent iff +2·|A⊓B| / (|A|+|B|) >= epsilon, where |A⊓B| is the number +of traces where A and B have overlapping lifecycles, +and |A| is the number of traces containing A. + +Reference: + Augusto, A., Dumas, M., & La Rosa, M. (2021). + Automated Discovery of Process Models with True + Concurrency and Inclusive Choices. Section 3.2, + Equation 5. +""" +from __future__ import annotations + +from split_miner.bpmn import Node +from split_miner.refined_dfg import ( + RefinedDirectlyFollowsGraph, +) + + +def _canonical_pair( + a: Node, b: Node +) -> tuple[Node, Node]: + """Return the pair in canonical order. + + Ensures (A, B) and (B, A) map to the same + canonical tuple, preventing double-counting. + + :param a: First node. + :param b: Second node. + :return: The canonical pair. + """ + key_a: tuple[str, str] = ( + type(a).__name__, a.node_id + ) + key_b: tuple[str, str] = ( + type(b).__name__, b.node_id + ) + if key_a <= key_b: + return (a, b) + return (b, a) + + +class RefinedPrunedDFG: + """A pruned DFG using lifecycle-based concurrency. + + Uses Equation 5 from the SM 2.0 paper to detect + true concurrency via overlapping activity lifecycles, + then prunes edges between concurrent activities. + + :param dfg: The refined DFG. + :param traces: The lifecycle traces. + :param epsilon: The concurrency threshold (0 to 1). + """ + + def __init__( + self, + dfg: RefinedDirectlyFollowsGraph, + traces: dict[ + tuple[tuple[Node, str], ...], int + ], + epsilon: float, + ) -> None: + """Build a pruned DFG with lifecycle concurrency. + + :param dfg: The refined DFG. + :param traces: The lifecycle traces. + :param epsilon: The concurrency threshold. + """ + self.__dfg: RefinedDirectlyFollowsGraph = dfg + self.__concurrent: set[ + tuple[Node, Node] + ] = set() + self.__edges: set[tuple[Node, Node]] = set() + self.__edge_freq: dict[ + tuple[Node, Node], int + ] = {} + self.__discover_concurrency(traces, epsilon) + self.__build_pruned_edges() + + def __discover_concurrency( + self, + traces: dict[ + tuple[tuple[Node, str], ...], int + ], + epsilon: float, + ) -> None: + """Discover concurrency via lifecycle overlap. + + For each pair of activities, count the number + of trace instances where their lifecycles overlap + (one starts before the other ends). Apply + Equation 5: 2·|A⊓B| / (|A|+|B|) >= epsilon. + + :param traces: The lifecycle traces. + :param epsilon: The concurrency threshold. + """ + # Count overlaps and activity occurrences. + overlap_count: dict[ + tuple[Node, Node], int + ] = {} + activity_count: dict[Node, int] = {} + for trace, count in traces.items(): + self.__count_overlaps_in_trace( + trace, count, + overlap_count, activity_count, + ) + # Apply Equation 5 for each pair. + nodes: list[Node] = sorted( + self.__dfg.nodes, + key=lambda n: ( + type(n).__name__, n.node_id + ), + ) + for i, a in enumerate(nodes): + for b in nodes[i + 1:]: + pair: tuple[Node, Node] = ( + _canonical_pair(a, b) + ) + ab_overlap: int = ( + overlap_count.get(pair, 0) + ) + a_count: int = activity_count.get( + a, 0 + ) + b_count: int = activity_count.get( + b, 0 + ) + if a_count + b_count == 0: + continue + ratio: float = ( + 2.0 * ab_overlap + / (a_count + b_count) + ) + if ratio >= epsilon: + self.__concurrent.add((a, b)) + self.__concurrent.add((b, a)) + + @staticmethod + def __count_overlaps_in_trace( + trace: tuple[tuple[Node, str], ...], + count: int, + overlap_count: dict[ + tuple[Node, Node], int + ], + activity_count: dict[Node, int], + ) -> None: + """Count lifecycle overlaps in a single trace. + + An overlap between A and B occurs when one starts + before the other ends. Track active activities + (started but not yet ended) to detect overlaps. + + :param trace: The lifecycle trace. + :param count: The trace frequency. + :param overlap_count: Accumulated overlap counts + (mutated). + :param activity_count: Accumulated activity + counts (mutated). + """ + # Track which activities are currently active + # (started but not ended). + active: set[Node] = set() + # Track pairs already counted as overlapping + # in this trace (canonical order to avoid + # double-counting when self-loops cause + # overlap in both directions). + overlapped: set[tuple[Node, Node]] = set() + # Track activities seen in this trace (for + # per-trace counting per Equation 5). + seen: set[Node] = set() + for node, lifecycle in trace: + if lifecycle == "start": + # This activity overlaps with all + # currently active activities. + for other in active: + pair: tuple[Node, Node] = ( + _canonical_pair( + other, node + ) + ) + if pair not in overlapped: + overlapped.add(pair) + active.add(node) + elif lifecycle == "end": + active.discard(node) + seen.add(node) + # Count each activity once per trace + # (Equation 5: |A| = number of traces + # containing A). + for node in seen: + activity_count[node] = ( + activity_count.get(node, 0) + count + ) + # Add overlap counts (canonical pairs). + for pair in overlapped: + overlap_count[pair] = ( + overlap_count.get(pair, 0) + count + ) + + def __build_pruned_edges(self) -> None: + """Build the pruned edge set. + + Remove edges between concurrent activities. + For non-concurrent bidirectional pairs, remove + the less frequent edge. + """ + for a, b in self.__dfg.edges: + freq: int = self.__dfg.df_frequency(a, b) + if (a, b) in self.__concurrent: + continue + if (b, a) in self.__dfg.edges: + rev_freq: int = ( + self.__dfg.df_frequency(b, a) + ) + if freq < rev_freq: + continue + self.__edges.add((a, b)) + self.__edge_freq[(a, b)] = freq + + @property + def nodes(self) -> set[Node]: + """The set of nodes. + + :return: The nodes. + """ + return self.__dfg.nodes + + @property + def edges(self) -> set[tuple[Node, Node]]: + """The set of pruned edges. + + :return: The edges. + """ + return set(self.__edges) + + def edge_frequency( + self, a: Node, b: Node + ) -> int: + """Return the frequency of a pruned edge. + + :param a: The source node. + :param b: The target node. + :return: The frequency, or 0 if not present. + """ + return self.__edge_freq.get((a, b), 0) + + def is_concurrent( + self, a: Node, b: Node + ) -> bool: + """Check if two nodes are concurrent. + + :param a: The first node. + :param b: The second node. + :return: True if a || b. + """ + return (a, b) in self.__concurrent + + @property + def concurrent_pairs( + self, + ) -> set[tuple[Node, Node]]: + """The set of concurrent pairs (both directions). + + :return: The concurrent pairs. + """ + return set(self.__concurrent) + + @property + def sources(self) -> set[Node]: + """The source nodes. + + :return: The source nodes. + """ + return self.__dfg.sources + + @property + def sinks(self) -> set[Node]: + """The sink nodes. + + :return: The sink nodes. + """ + return self.__dfg.sinks + + def outgoing( + self, node: Node + ) -> set[tuple[Node, Node]]: + """Return the outgoing edges of a node. + + :param node: The node. + :return: The outgoing edges. + """ + return { + (a, b) for (a, b) in self.__edges + if a == node + } + + def incoming( + self, node: Node + ) -> set[tuple[Node, Node]]: + """Return the incoming edges of a node. + + :param node: The node. + :return: The incoming edges. + """ + return { + (a, b) for (a, b) in self.__edges + if b == node + } diff --git a/src/split_miner/refined_dfg.py b/src/split_miner/refined_dfg.py new file mode 100644 index 0000000..76f2a84 --- /dev/null +++ b/src/split_miner/refined_dfg.py @@ -0,0 +1,240 @@ +# 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. +"""Refined Directly-Follows Graph (SM 2.0, Definition 6). + +Uses activity lifecycle (start/end) events to build the +directly-follows relation: activity ay directly-follows ax +iff ay starts after ax ends with no other end events in +between. + +Reference: + Augusto, A., Dumas, M., & La Rosa, M. (2021). + Automated Discovery of Process Models with True + Concurrency and Inclusive Choices. Section 3.1, + Definition 6. +""" +from __future__ import annotations + +from split_miner.bpmn import Node + + +class RefinedDirectlyFollowsGraph: + """A refined DFG using activity lifecycle events. + + Built from lifecycle-aware traces per Definition 6 in the + SM 2.0 paper. Each trace event is a (Node, lifecycle) + pair where lifecycle is ``"start"`` or ``"end"``. + + :param traces: The event log as a dict mapping each + lifecycle trace to its frequency. + """ + + def __init__( + self, + traces: dict[ + tuple[tuple[Node, str], ...], int + ], + ) -> None: + """Build a refined DFG from lifecycle traces. + + :param traces: The event log as a dict mapping + each lifecycle trace to its frequency. + """ + self.__nodes: set[Node] = set() + self.__sources: set[Node] = set() + self.__sinks: set[Node] = set() + self.__df_freq: dict[ + tuple[Node, Node], int + ] = {} + self.__self_loops: set[Node] = set() + self.__build(traces) + + def __build( + self, + traces: dict[ + tuple[tuple[Node, str], ...], int + ], + ) -> 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(): + if not trace: + continue + # Collect nodes and find sources/sinks. + activities: set[Node] = set() + for node, _ in trace: + activities.add(node) + self.__nodes |= activities + # Source: first activity to start. + for node, lifecycle in trace: + if lifecycle == "start": + self.__sources.add(node) + break + # Sink: last activity to end. + for node, lifecycle in reversed(trace): + 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. + self.__scan_trace(trace, count) + + def __scan_trace( + self, + trace: tuple[tuple[Node, str], ...], + count: int, + ) -> 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"). + + :param trace: The lifecycle trace. + :param count: The trace frequency. + """ + pending: set[Node] = set() + for node, lifecycle in trace: + if lifecycle == "end": + # A new end event clears previous + # pending sources (they now have an + # end event between them and any + # future start). + pending.clear() + pending.add(node) + elif lifecycle == "start": + # 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 + ) + + @property + def nodes(self) -> set[Node]: + """The set of nodes. + + :return: The nodes. + """ + return set(self.__nodes) + + @property + def edges(self) -> set[tuple[Node, Node]]: + """The set of edges with positive frequency. + + :return: The edges. + """ + return { + (a, b) for (a, b), freq + in self.__df_freq.items() + if freq > 0 + } + + def df_frequency( + self, a: Node, b: Node + ) -> int: + """Return the directly-follows frequency. + + :param a: The source node. + :param b: The target node. + :return: The frequency. + """ + return self.__df_freq.get((a, b), 0) + + @property + 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. + + :return: The self-loop nodes. + """ + return set(self.__self_loops) + + @property + def sources(self) -> set[Node]: + """The source nodes (first to start in traces). + + :return: The source nodes. + """ + return set(self.__sources) + + @property + def sinks(self) -> set[Node]: + """The sink nodes (last to end in traces). + + :return: The sink nodes. + """ + return set(self.__sinks) + + def outgoing( + self, node: Node + ) -> set[tuple[Node, Node]]: + """Return the outgoing edges of a node. + + :param node: The node. + :return: The outgoing edges. + """ + return { + (a, b) for (a, b) in self.edges + if a == node + } + + def incoming( + self, node: Node + ) -> set[tuple[Node, Node]]: + """Return the incoming edges of a node. + + :param node: The node. + :return: The incoming edges. + """ + return { + (a, b) for (a, b) in self.edges + if b == node + } diff --git a/tests/test_heuristics.py b/tests/test_heuristics.py new file mode 100644 index 0000000..fd666d2 --- /dev/null +++ b/tests/test_heuristics.py @@ -0,0 +1,423 @@ +# 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() diff --git a/tests/test_refined_concurrency.py b/tests/test_refined_concurrency.py new file mode 100644 index 0000000..e940edc --- /dev/null +++ b/tests/test_refined_concurrency.py @@ -0,0 +1,693 @@ +# 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 refined concurrency discovery (SM 2.0, Equation 5). + +The refined concurrency oracle uses activity lifecycle overlap: +two activities A and B are concurrent iff +2·|A⊓B| / (|A|+|B|) >= epsilon, where |A⊓B| is the number +of overlapping lifecycle instances. + +Reference: + Augusto, A., Dumas, M., & La Rosa, M. (2021). + Automated Discovery of Process Models with True + Concurrency and Inclusive Choices. Section 3.2, + Equation 5. +""" +from __future__ import annotations + +import unittest + +from split_miner.bpmn import Task +from split_miner.refined_concurrency import RefinedPrunedDFG +from split_miner.refined_dfg import RefinedDirectlyFollowsGraph + +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 + } + + +def _make_paper_example() -> tuple[ + dict[str, Task], + dict[tuple[tuple[Task, str], ...], int], +]: + """Build the paper's example Lrho_x traces. + + Four traces with activities A-F, where B/C and + D/E have overlapping lifecycles. + + :return: The tasks and the lifecycle traces. + """ + t: dict[str, Task] = _make_tasks( + "A", "B", "C", "D", "E", "F" + ) + a: Task = t["A"] + b: Task = t["B"] + c: Task = t["C"] + d: Task = t["D"] + e: Task = t["E"] + f: Task = t["F"] + + traces: dict[ + tuple[tuple[Task, str], ...], int + ] = { + # Trace 1 + ((a, S), (a, E), (b, S), (c, S), + (c, E), (b, E), (e, S), (d, S), + (d, E), (e, E), (f, S), (f, E)): 1, + # Trace 2 + ((a, S), (a, E), (b, S), (c, S), + (b, E), (c, E), (e, S), (d, S), + (e, E), (d, E), (f, S), (f, E)): 1, + # Trace 3 + ((a, S), (a, E), (c, S), (b, S), + (b, E), (c, E), (d, S), (e, S), + (d, E), (e, E), (f, S), (f, E)): 1, + # Trace 4 + ((a, S), (a, E), (c, S), (b, S), + (c, E), (b, E), (d, S), (e, S), + (e, E), (d, E), (f, S), (f, E)): 1, + } + return t, traces + + +class TestNoOverlap(unittest.TestCase): + """Tests with purely sequential activities. + + No overlapping lifecycles means no concurrency. + """ + + def setUp(self) -> None: + """Set up sequential trace As Ae Bs Be Cs Ce. + + :return: None. + """ + self.__tasks: dict[str, Task] = ( + _make_tasks("A", "B", "C") + ) + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + traces: dict[ + tuple[tuple[Task, str], ...], int + ] = { + ((a, S), (a, E), + (b, S), (b, E), + (c, S), (c, E)): 1, + } + dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph(traces) + ) + self.__pruned: RefinedPrunedDFG = ( + RefinedPrunedDFG(dfg, traces, 0.5) + ) + + def test_no_concurrent_pairs(self) -> None: + """No concurrent pairs when sequential. + + :return: None. + """ + self.assertEqual( + self.__pruned.concurrent_pairs, set() + ) + + def test_edges_preserved(self) -> None: + """All DFG edges preserved (no pruning needed). + + :return: None. + """ + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + self.assertEqual( + self.__pruned.edges, + {(a, b), (b, c)}, + ) + + +class TestFullOverlap(unittest.TestCase): + """Tests with fully overlapping lifecycles. + + B and C always overlap: 2·|B⊓C|/(|B|+|C|) = 1.0. + """ + + def setUp(self) -> None: + """Set up: As Ae Bs Cs Be Ce (B and C overlap). + + :return: None. + """ + self.__tasks: dict[str, Task] = ( + _make_tasks("A", "B", "C") + ) + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + traces: dict[ + tuple[tuple[Task, str], ...], int + ] = { + ((a, S), (a, E), + (b, S), (c, S), + (b, E), (c, E)): 1, + } + dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph(traces) + ) + self.__pruned: RefinedPrunedDFG = ( + RefinedPrunedDFG(dfg, traces, 0.5) + ) + + def test_concurrent(self) -> None: + """B and C are concurrent (ratio=1.0 >= 0.5). + + :return: None. + """ + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + self.assertTrue( + self.__pruned.is_concurrent(b, c) + ) + self.assertTrue( + self.__pruned.is_concurrent(c, b) + ) + + def test_a_not_concurrent_with_b(self) -> None: + """A is not concurrent with B (no overlap). + + :return: None. + """ + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + self.assertFalse( + self.__pruned.is_concurrent(a, b) + ) + + +class TestPartialOverlap(unittest.TestCase): + """Tests with partial overlap across traces. + + B and C overlap in 1 of 2 traces. + Ratio = 2·1/(2+2) = 0.5. + """ + + def setUp(self) -> None: + """Set up two traces: one overlapping, one not. + + Trace 1: As Ae Bs Cs Be Ce (overlap) + Trace 2: As Ae Bs Be Cs Ce (no overlap) + + :return: None. + """ + self.__tasks: dict[str, Task] = ( + _make_tasks("A", "B", "C") + ) + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + self.__traces: dict[ + tuple[tuple[Task, str], ...], int + ] = { + # Trace 1: B and C overlap + ((a, S), (a, E), + (b, S), (c, S), + (b, E), (c, E)): 1, + # Trace 2: B and C sequential + ((a, S), (a, E), + (b, S), (b, E), + (c, S), (c, E)): 1, + } + + def test_concurrent_at_threshold(self) -> None: + """B||C with epsilon=0.5 (ratio=0.5 >= 0.5). + + :return: None. + """ + dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph(self.__traces) + ) + pruned: RefinedPrunedDFG = ( + RefinedPrunedDFG(dfg, self.__traces, 0.5) + ) + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + self.assertTrue( + pruned.is_concurrent(b, c) + ) + + def test_not_concurrent_above_threshold( + self, + ) -> None: + """B not ||C with epsilon=0.6 (ratio=0.5 < 0.6). + + :return: None. + """ + dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph(self.__traces) + ) + pruned: RefinedPrunedDFG = ( + RefinedPrunedDFG(dfg, self.__traces, 0.6) + ) + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + self.assertFalse( + pruned.is_concurrent(b, c) + ) + + +class TestPaperExample(unittest.TestCase): + """Tests concurrency on the paper's Lrho_x example. + + B/C and D/E overlap in all 4 traces. + Ratio = 2·4/(4+4) = 1.0. + """ + + def setUp(self) -> None: + """Set up the paper's example. + + :return: None. + """ + self.__tasks: dict[str, Task] + traces: dict[ + tuple[tuple[Task, str], ...], int + ] + self.__tasks, traces = ( + _make_paper_example() + ) + dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph(traces) + ) + self.__pruned: RefinedPrunedDFG = ( + RefinedPrunedDFG(dfg, traces, 0.5) + ) + + def test_b_c_concurrent(self) -> None: + """B and C are concurrent. + + :return: None. + """ + self.assertTrue( + self.__pruned.is_concurrent( + self.__tasks["B"], + self.__tasks["C"], + ) + ) + + def test_d_e_concurrent(self) -> None: + """D and E are concurrent. + + :return: None. + """ + self.assertTrue( + self.__pruned.is_concurrent( + self.__tasks["D"], + self.__tasks["E"], + ) + ) + + def test_a_b_not_concurrent(self) -> None: + """A and B are not concurrent. + + :return: None. + """ + self.assertFalse( + self.__pruned.is_concurrent( + self.__tasks["A"], + self.__tasks["B"], + ) + ) + + def test_b_d_not_concurrent(self) -> None: + """B and D are not concurrent. + + :return: None. + """ + self.assertFalse( + self.__pruned.is_concurrent( + self.__tasks["B"], + self.__tasks["D"], + ) + ) + + def test_edges_no_concurrent_edges(self) -> None: + """No edges between concurrent pairs. + + B->D, B->E, C->D, C->E exist in the DFG + but B||C and D||E, so no edges between B/C + or between D/E should appear. + + :return: None. + """ + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + d: Task = self.__tasks["D"] + e: Task = self.__tasks["E"] + # No edges between concurrent B/C + self.assertNotIn( + (b, c), self.__pruned.edges + ) + self.assertNotIn( + (c, b), self.__pruned.edges + ) + # No edges between concurrent D/E + self.assertNotIn( + (d, e), self.__pruned.edges + ) + self.assertNotIn( + (e, d), self.__pruned.edges + ) + + def test_pruned_edges(self) -> None: + """Pruned DFG has expected edges. + + A->B, A->C, B->D, B->E, C->D, C->E, D->F, + E->F (same as refined DFG since no concurrent + pairs have direct edges in the refined DFG). + + :return: None. + """ + t: dict[str, Task] = self.__tasks + expected: set[tuple[Task, Task]] = { + (t["A"], t["B"]), (t["A"], t["C"]), + (t["B"], t["D"]), (t["B"], t["E"]), + (t["C"], t["D"]), (t["C"], t["E"]), + (t["D"], t["F"]), (t["E"], t["F"]), + } + self.assertEqual( + self.__pruned.edges, expected + ) + + +class TestTraceFrequency(unittest.TestCase): + """Tests that trace frequency affects overlap count. + + A trace with frequency 3 where B and C overlap + contributes 3 to |B⊓C|. + """ + + def setUp(self) -> None: + """Set up traces with varying frequencies. + + Trace 1 (freq 3): B and C overlap. + Trace 2 (freq 7): B and C sequential. + |B⊓C|=3, |B|=10, |C|=10. + Ratio = 2·3/(10+10) = 0.3. + + :return: None. + """ + self.__tasks: dict[str, Task] = ( + _make_tasks("A", "B", "C") + ) + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + self.__traces: dict[ + tuple[tuple[Task, str], ...], int + ] = { + # Overlap, frequency 3 + ((a, S), (a, E), + (b, S), (c, S), + (b, E), (c, E)): 3, + # Sequential, frequency 7 + ((a, S), (a, E), + (b, S), (b, E), + (c, S), (c, E)): 7, + } + + def test_concurrent_low_epsilon(self) -> None: + """B||C with epsilon=0.3 (ratio=0.3 >= 0.3). + + :return: None. + """ + dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph(self.__traces) + ) + pruned: RefinedPrunedDFG = ( + RefinedPrunedDFG( + dfg, self.__traces, 0.3 + ) + ) + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + self.assertTrue( + pruned.is_concurrent(b, c) + ) + + def test_not_concurrent_high_epsilon( + self, + ) -> None: + """B not ||C with epsilon=0.5 (ratio=0.3 < 0.5). + + :return: None. + """ + dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph(self.__traces) + ) + pruned: RefinedPrunedDFG = ( + RefinedPrunedDFG( + dfg, self.__traces, 0.5 + ) + ) + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + self.assertFalse( + pruned.is_concurrent(b, c) + ) + + +class TestSelfLoopActivityCount(unittest.TestCase): + """Tests that |A| counts traces, not completions. + + Per Equation 5, |A| is the number of traces + containing activity A. An activity that completes + multiple times in a single trace (self-loop) should + still count as 1 for that trace, not as the number + of completions. + """ + + def setUp(self) -> None: + """Set up traces where B has a self-loop. + + Trace 1 (freq 1): B completes twice, overlaps + with C during its first execution. + As Ae Bs Cs Be Ce Bs Be + + |B⊓C| = 1 (1 trace with overlap). + |B| = 1 (1 trace containing B, NOT 2). + |C| = 1 (1 trace containing C). + Ratio = 2·1/(1+1) = 1.0. + + With the bug (counting completions): + |B| = 2, ratio = 2·1/(2+1) = 0.67. + + :return: None. + """ + self.__tasks: dict[str, Task] = ( + _make_tasks("A", "B", "C") + ) + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + self.__traces: dict[ + tuple[tuple[Task, str], ...], int + ] = { + ((a, S), (a, E), + (b, S), (c, S), + (b, E), (c, E), + (b, S), (b, E)): 1, + } + + def test_concurrent_with_self_loop( + self, + ) -> None: + """B||C even though B has a self-loop. + + Ratio is 2·1/(1+1) = 1.0, not 2·1/(2+1) + = 0.67. With per-trace counting, B||C at + epsilon=0.9. + + :return: None. + """ + dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph( + self.__traces + ) + ) + pruned: RefinedPrunedDFG = ( + RefinedPrunedDFG( + dfg, self.__traces, 0.9 + ) + ) + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + self.assertTrue( + pruned.is_concurrent(b, c) + ) + + def test_not_concurrent_bug_threshold( + self, + ) -> None: + """Verify the ratio is 1.0, not 0.67. + + If the bug existed (counting completions), + epsilon=0.9 would fail since 0.67 < 0.9. + This test passes because |B|=1 (per-trace), + giving ratio=1.0 >= 0.9. + + :return: None. + """ + dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph( + self.__traces + ) + ) + # Even at very high epsilon, should be + # concurrent since ratio is 1.0. + pruned: RefinedPrunedDFG = ( + RefinedPrunedDFG( + dfg, self.__traces, 1.0 + ) + ) + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + self.assertTrue( + pruned.is_concurrent(b, c) + ) + + +class TestSelfLoopOverlapCount(unittest.TestCase): + """Tests that |A⊓B| counts at most once per trace. + + Per Equation 5, |A⊓B| is the number of traces + where A and B have overlapping lifecycles. When + a self-loop activity overlaps with another activity + in both "directions" within one trace, it should + still count as 1 overlap, not 2. + """ + + def setUp(self) -> None: + """Set up traces where self-loop causes overlap. + + Trace 1 (freq 1): A overlaps B, then A + restarts while B is still active. + As Bs Ae As Be Ae + + Events: + - (A,s): active={A} + - (B,s): overlap A-B, active={A,B} + - (A,e): active={B} + - (A,s): overlap B-A, active={A,B} + - (B,e): active={A} + - (A,e): active={} + + Correct: |A⊓B| = 1 (one trace with overlap). + Bug: overlap counted as 2 (both directions). + + Trace 2 (freq 1): A and B sequential. + As Ae Bs Be + + Correct totals: |A⊓B|=1, |A|=2, |B|=2. + Ratio = 2·1/(2+2) = 0.5. + + Bug totals: |A⊓B|=2 (double-counted). + Bug ratio = 2·2/(2+2) = 1.0. + + :return: None. + """ + self.__tasks: dict[str, Task] = ( + _make_tasks("A", "B") + ) + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + self.__traces: dict[ + tuple[tuple[Task, str], ...], int + ] = { + # Trace 1: A self-loops, overlaps B + ((a, S), (b, S), (a, E), + (a, S), (b, E), (a, E)): 1, + # Trace 2: sequential + ((a, S), (a, E), + (b, S), (b, E)): 1, + } + + def test_not_concurrent_at_high_epsilon( + self, + ) -> None: + """A not ||B with epsilon=0.6 (ratio=0.5). + + Correct ratio is 2·1/(2+2) = 0.5 < 0.6. + With the bug (double-counted overlap), + ratio would be 2·2/(2+2) = 1.0 >= 0.6. + + :return: None. + """ + dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph( + self.__traces + ) + ) + pruned: RefinedPrunedDFG = ( + RefinedPrunedDFG( + dfg, self.__traces, 0.6 + ) + ) + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + self.assertFalse( + pruned.is_concurrent(a, b) + ) + + def test_concurrent_at_low_epsilon( + self, + ) -> None: + """A||B with epsilon=0.5 (ratio=0.5 >= 0.5). + + :return: None. + """ + dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph( + self.__traces + ) + ) + pruned: RefinedPrunedDFG = ( + RefinedPrunedDFG( + dfg, self.__traces, 0.5 + ) + ) + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + self.assertTrue( + pruned.is_concurrent(a, b) + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_refined_dfg.py b/tests/test_refined_dfg.py new file mode 100644 index 0000000..68dff4b --- /dev/null +++ b/tests/test_refined_dfg.py @@ -0,0 +1,518 @@ +# 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 the refined DFG (Definition 6, SM 2.0). + +The refined directly-follows relation uses activity lifecycle +(start/end) events: activity ay directly-follows ax iff +ay starts after ax ends with no other end events in between. + +Reference: + Augusto, A., Dumas, M., & La Rosa, M. (2021). + Automated Discovery of Process Models with True + Concurrency and Inclusive Choices. Section 3.1, + Definition 6. +""" +from __future__ import annotations + +import unittest + +from split_miner.bpmn import Task +from split_miner.refined_dfg import RefinedDirectlyFollowsGraph + +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 + } + + +def _make_paper_example() -> tuple[ + dict[str, Task], + RefinedDirectlyFollowsGraph, +]: + """Build the DFG from the paper's example Lrho_x. + + Four traces with activities A-F, where B/C and + D/E have overlapping lifecycles: + + Trace 1: As Ae Bs Cs Ce Be Es Ds De Ee Fs Fe + Trace 2: As Ae Bs Cs Be Ce Es Ds Ee De Fs Fe + Trace 3: As Ae Cs Bs Be Ce Ds Es De Ee Fs Fe + Trace 4: As Ae Cs Bs Ce Be Ds Es Ee De Fs Fe + + Expected DFG (Figure 3c): A->B, A->C, B->D, B->E, + C->D, C->E, D->F, E->F. + + :return: The tasks and the DFG. + """ + t: dict[str, Task] = _make_tasks( + "A", "B", "C", "D", "E", "F" + ) + a: Task = t["A"] + b: Task = t["B"] + c: Task = t["C"] + d: Task = t["D"] + e: Task = t["E"] + f: Task = t["F"] + + traces: dict[ + tuple[tuple[Task, str], ...], int + ] = { + # Trace 1 + ((a, S), (a, E), (b, S), (c, S), + (c, E), (b, E), (e, S), (d, S), + (d, E), (e, E), (f, S), (f, E)): 1, + # Trace 2 + ((a, S), (a, E), (b, S), (c, S), + (b, E), (c, E), (e, S), (d, S), + (e, E), (d, E), (f, S), (f, E)): 1, + # Trace 3 + ((a, S), (a, E), (c, S), (b, S), + (b, E), (c, E), (d, S), (e, S), + (d, E), (e, E), (f, S), (f, E)): 1, + # Trace 4 + ((a, S), (a, E), (c, S), (b, S), + (c, E), (b, E), (d, S), (e, S), + (e, E), (d, E), (f, S), (f, E)): 1, + } + dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph(traces) + ) + return t, dfg + + +class TestSequentialTrace(unittest.TestCase): + """Tests refined DFG with purely sequential traces. + + When activities don't overlap, the refined DFG matches + the SM 1.0 DFG. + """ + + def setUp(self) -> None: + """Set up a simple sequential trace. + + Trace: As Ae Bs Be Cs Ce + Expected: A->B, B->C. + + :return: None. + """ + self.__tasks: dict[str, Task] = ( + _make_tasks("A", "B", "C") + ) + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + traces: dict[ + tuple[tuple[Task, str], ...], int + ] = { + ((a, S), (a, E), + (b, S), (b, E), + (c, S), (c, E)): 1, + } + self.__dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph(traces) + ) + + def test_nodes(self) -> None: + """DFG has 3 nodes (A, B, C). + + :return: None. + """ + self.assertEqual(len(self.__dfg.nodes), 3) + + def test_edges(self) -> None: + """DFG has edges 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_sources(self) -> None: + """Source is A (first activity to start). + + :return: None. + """ + self.assertEqual( + self.__dfg.sources, + {self.__tasks["A"]}, + ) + + def test_sinks(self) -> None: + """Sink is C (last activity to end). + + :return: None. + """ + self.assertEqual( + self.__dfg.sinks, + {self.__tasks["C"]}, + ) + + +class TestOverlappingTrace(unittest.TestCase): + """Tests refined DFG with overlapping lifecycles. + + When activities overlap, no directly-follows relation + exists between them. + """ + + def setUp(self) -> None: + """Set up overlapping and sequential activities. + + Trace: As Bs Ae Be Cs Ce + A and B overlap. B ends last. + Expected: B->C only (not A->C, since A_end + precedes B_end which is an end event between + A_end and C_start). + + :return: None. + """ + self.__tasks: dict[str, Task] = ( + _make_tasks("A", "B", "C") + ) + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + traces: dict[ + tuple[tuple[Task, str], ...], int + ] = { + ((a, S), (b, S), (a, E), + (b, E), (c, S), (c, E)): 1, + } + self.__dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph(traces) + ) + + def test_no_edge_between_overlapping(self) -> None: + """No A->B or B->A edge (overlapping lifecycles). + + A starts before B, and A ends before B. But + A_end(3) > B_start(2), so A does not end before + B starts. Similarly B does not end before A + starts. So neither direction holds. + + :return: None. + """ + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + self.assertNotIn((a, b), self.__dfg.edges) + self.assertNotIn((b, a), self.__dfg.edges) + + def test_last_ender_follows_to_next(self) -> None: + """Only B->C exists (B is the last to end). + + B_end(4) precedes C_start(5) with no end events + between them. A_end(3) also precedes C_start(5) + but B_end(4) is between them, blocking A->C. + + :return: None. + """ + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + self.assertIn((b, c), self.__dfg.edges) + + def test_edge_count(self) -> None: + """Only 1 edge total. + + :return: None. + """ + self.assertEqual(len(self.__dfg.edges), 1) + + +class TestPaperExample(unittest.TestCase): + """Tests refined DFG on the paper's example Lrho_x. + + Reference: SM 2.0 paper, Section 3.1, Figure 3(c). + """ + + def setUp(self) -> None: + """Set up the paper's example. + + :return: None. + """ + self.__tasks: dict[str, Task] + self.__dfg: RefinedDirectlyFollowsGraph + self.__tasks, self.__dfg = ( + _make_paper_example() + ) + + def test_nodes(self) -> None: + """DFG has 6 nodes (A through F). + + :return: None. + """ + self.assertEqual(len(self.__dfg.nodes), 6) + + def test_edges(self) -> None: + """DFG has exactly 8 edges per Figure 3(c). + + A->B, A->C, B->D, B->E, C->D, C->E, D->F, + E->F. + + :return: None. + """ + t: dict[str, Task] = self.__tasks + expected: set[tuple[Task, Task]] = { + (t["A"], t["B"]), (t["A"], t["C"]), + (t["B"], t["D"]), (t["B"], t["E"]), + (t["C"], t["D"]), (t["C"], t["E"]), + (t["D"], t["F"]), (t["E"], t["F"]), + } + self.assertEqual(self.__dfg.edges, expected) + + def test_edge_count(self) -> None: + """DFG has 8 edges. + + :return: None. + """ + self.assertEqual(len(self.__dfg.edges), 8) + + def test_no_bidirectional_edges(self) -> None: + """No bidirectional edges exist. + + In the paper's example, overlapping lifecycles + prevent any bidirectional relations. + + :return: None. + """ + for src, tgt in self.__dfg.edges: + self.assertNotIn( + (tgt, src), self.__dfg.edges, + f"Bidirectional edge {src}->{tgt} " + f"and {tgt}->{src}", + ) + + def test_sources(self) -> None: + """Source is A. + + :return: None. + """ + self.assertEqual( + self.__dfg.sources, + {self.__tasks["A"]}, + ) + + def test_sinks(self) -> None: + """Sink is F. + + :return: None. + """ + self.assertEqual( + self.__dfg.sinks, + {self.__tasks["F"]}, + ) + + def test_a_to_b_frequency(self) -> None: + """A->B occurs in all 4 traces. + + :return: None. + """ + self.assertEqual( + self.__dfg.df_frequency( + self.__tasks["A"], + self.__tasks["B"], + ), + 4, + ) + + def test_a_to_c_frequency(self) -> None: + """A->C occurs in all 4 traces. + + :return: None. + """ + self.assertEqual( + self.__dfg.df_frequency( + self.__tasks["A"], + self.__tasks["C"], + ), + 4, + ) + + def test_b_to_d_frequency(self) -> None: + """B->D occurs in traces 1 and 4. + + :return: None. + """ + self.assertEqual( + self.__dfg.df_frequency( + self.__tasks["B"], + self.__tasks["D"], + ), + 2, + ) + + def test_d_to_f_frequency(self) -> None: + """D->F occurs in traces 2 and 4. + + :return: None. + """ + self.assertEqual( + self.__dfg.df_frequency( + self.__tasks["D"], + self.__tasks["F"], + ), + 2, + ) + + def test_e_to_f_frequency(self) -> None: + """E->F occurs in traces 1 and 3. + + :return: None. + """ + self.assertEqual( + self.__dfg.df_frequency( + self.__tasks["E"], + self.__tasks["F"], + ), + 2, + ) + + +class TestForkAfterEnd(unittest.TestCase): + """Tests that multiple activities can follow one end. + + When A ends and both B and C start (with no other end + events between), both A->B and A->C should exist. + """ + + def setUp(self) -> None: + """Set up a fork pattern. + + Trace: As Ae Bs Cs Be Ce + A ends, then B and C start (neither ends before + the other starts). + Expected: A->B, A->C. + + :return: None. + """ + self.__tasks: dict[str, Task] = ( + _make_tasks("A", "B", "C") + ) + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + traces: dict[ + tuple[tuple[Task, str], ...], int + ] = { + ((a, S), (a, E), (b, S), (c, S), + (b, E), (c, E)): 1, + } + self.__dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph(traces) + ) + + def test_fork_edges(self) -> None: + """Both A->B and A->C exist. + + A_end precedes both B_start and C_start with no + end events in between. + + :return: None. + """ + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + self.assertEqual( + self.__dfg.edges, + {(a, b), (a, c)}, + ) + + +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. + """ + + def setUp(self) -> None: + """Set up trace with self-loop: A repeats. + + Trace: As Ae Bs Be As Ae Cs Ce + Activity A completes twice. + + :return: None. + """ + self.__tasks: dict[str, Task] = ( + _make_tasks("A", "B", "C") + ) + a: Task = self.__tasks["A"] + b: Task = self.__tasks["B"] + c: Task = self.__tasks["C"] + traces: dict[ + tuple[tuple[Task, str], ...], int + ] = { + ((a, S), (a, E), + (b, S), (b, E), + (a, S), (a, E), + (c, S), (c, E)): 1, + } + self.__dfg: RefinedDirectlyFollowsGraph = ( + RefinedDirectlyFollowsGraph(traces) + ) + + def test_a_is_self_loop(self) -> None: + """A is detected as a self-loop. + + :return: None. + """ + self.assertIn( + self.__tasks["A"], + self.__dfg.self_loops, + ) + + def test_b_not_self_loop(self) -> None: + """B is not a self-loop. + + :return: None. + """ + self.assertNotIn( + self.__tasks["B"], + self.__dfg.self_loops, + ) + + def test_no_self_loop_in_paper(self) -> None: + """Paper example has no self-loops. + + :return: None. + """ + _, dfg = _make_paper_example() + self.assertEqual(dfg.self_loops, set()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_split_miner_2.py b/tests/test_split_miner_2.py new file mode 100644 index 0000000..3c6682c --- /dev/null +++ b/tests/test_split_miner_2.py @@ -0,0 +1,332 @@ +# 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. +"""Integration tests for Split Miner 2.0 pipeline. + +Verifies that split_miner_2() produces valid BPMN models +from lifecycle-aware event logs. + +Reference: + Augusto, A., Dumas, M., & La Rosa, M. (2021). + Automated Discovery of Process Models with True + Concurrency and Inclusive Choices. +""" +from __future__ import annotations + +import unittest + +from split_miner.bpmn import ( + BPMNModel, + Gateway, + GatewayType, + Node, + StartEvent, + Task, +) +from split_miner.miner import split_miner_2 + +S: str = "start" +"""Lifecycle start constant.""" +E: str = "end" +"""Lifecycle end constant.""" + + +class TestSequentialPipeline(unittest.TestCase): + """Tests SM 2.0 with a simple sequential log. + + A -> B -> C with no concurrency. + """ + + def setUp(self) -> None: + """Set up sequential lifecycle traces. + + :return: None. + """ + traces: dict[ + tuple[tuple[str, str], ...], int + ] = { + (("A", S), ("A", E), + ("B", S), ("B", E), + ("C", S), ("C", E)): 5, + } + self.__model: BPMNModel = split_miner_2( + traces + ) + + def test_has_start_and_end(self) -> None: + """Model has start and end events. + + :return: None. + """ + self.assertIsNotNone(self.__model.start) + self.assertIsNotNone(self.__model.end) + + def test_has_three_tasks(self) -> None: + """Model has 3 tasks (A, B, C). + + :return: None. + """ + self.assertEqual( + len(self.__model.tasks), 3 + ) + + def test_no_gateways(self) -> None: + """Sequential model has no gateways. + + :return: None. + """ + self.assertEqual( + len(self.__model.gateways), 0 + ) + + +class TestConcurrentPipeline(unittest.TestCase): + """Tests SM 2.0 with concurrent activities. + + A -> (B || C) -> D where B and C always overlap. + """ + + def setUp(self) -> None: + """Set up concurrent lifecycle traces. + + :return: None. + """ + traces: dict[ + tuple[tuple[str, str], ...], int + ] = { + (("A", S), ("A", E), + ("B", S), ("C", S), + ("B", E), ("C", E), + ("D", S), ("D", E)): 5, + } + self.__model: BPMNModel = split_miner_2( + traces + ) + + def test_has_four_tasks(self) -> None: + """Model has 4 tasks (A, B, C, D). + + :return: None. + """ + self.assertEqual( + len(self.__model.tasks), 4 + ) + + def test_has_gateways(self) -> None: + """Model has gateway(s) for concurrency. + + :return: None. + """ + self.assertGreater( + len(self.__model.gateways), 0 + ) + + def test_has_and_split(self) -> None: + """Model has an AND-split for B || C. + + :return: None. + """ + and_splits: list[Gateway] = [ + gw for gw + in self.__model.gateways.values() + if gw.gateway_type == GatewayType.AND + and len( + self.__model.outgoing_edges(gw) + ) > 1 + ] + self.assertGreater(len(and_splits), 0) + + +class TestPaperExample(unittest.TestCase): + """Tests SM 2.0 on the paper's Lrho_x example. + + Should produce a valid BPMN model with A-F. + """ + + def setUp(self) -> None: + """Set up the paper's example. + + :return: None. + """ + traces: dict[ + tuple[tuple[str, str], ...], int + ] = { + (("A", S), ("A", E), + ("B", S), ("C", S), + ("C", E), ("B", E), + ("E", S), ("D", S), + ("D", E), ("E", E), + ("F", S), ("F", E)): 1, + (("A", S), ("A", E), + ("B", S), ("C", S), + ("B", E), ("C", E), + ("E", S), ("D", S), + ("E", E), ("D", E), + ("F", S), ("F", E)): 1, + (("A", S), ("A", E), + ("C", S), ("B", S), + ("B", E), ("C", E), + ("D", S), ("E", S), + ("D", E), ("E", E), + ("F", S), ("F", E)): 1, + (("A", S), ("A", E), + ("C", S), ("B", S), + ("C", E), ("B", E), + ("D", S), ("E", S), + ("E", E), ("D", E), + ("F", S), ("F", E)): 1, + } + self.__model: BPMNModel = split_miner_2( + traces + ) + + def test_has_six_tasks(self) -> None: + """Model has 6 tasks (A through F). + + :return: None. + """ + self.assertEqual( + len(self.__model.tasks), 6 + ) + + def test_completes_without_error(self) -> None: + """Pipeline completes without error. + + :return: None. + """ + self.assertIsNotNone(self.__model) + + def test_start_connects_to_a(self) -> None: + """Start event connects to task A. + + :return: None. + """ + start_succs: set[Node] = ( + self.__model.successors( + self.__model.start + ) + ) + # Either directly to A or through a gateway + reachable: set[str] = set() + queue: list[Node] = list(start_succs) + visited: set[Node] = set() + while queue: + n: Node = queue.pop(0) + if n in visited: + continue + visited.add(n) + if isinstance(n, Task): + reachable.add(n.node_id) + elif isinstance(n, Gateway): + queue.extend( + self.__model.successors(n) + ) + self.assertIn("A", reachable) + + +class TestSelfLoopPipeline(unittest.TestCase): + """Tests SM 2.0 with a self-loop activity. + + A -> B (self-loop) -> C. B repeats in the same + trace without going through other activities. + The self-loop XOR back-edge is the only source of + gateways around B. + """ + + def setUp(self) -> None: + """Set up lifecycle traces with self-loop on B. + + Trace: As Ae Bs Be Bs Be Cs Ce + B completes twice. No other loops exist, + so only self-loop restoration adds gateways + around B. + + :return: None. + """ + traces: dict[ + tuple[tuple[str, str], ...], int + ] = { + (("A", S), ("A", E), + ("B", S), ("B", E), + ("B", S), ("B", E), + ("C", S), ("C", E)): 5, + } + self.__model: BPMNModel = split_miner_2( + traces + ) + + def test_self_loop_xor_join_before_b( + self, + ) -> None: + """Task B has an XOR-join predecessor. + + The self-loop restoration inserts an XOR-join + before B. Without restoration, B connects + directly to its predecessor (A or start). + + :return: None. + """ + b: Task = self.__model.get_task("B") + preds: set[Node] = ( + self.__model.predecessors(b) + ) + xor_joins: list[Gateway] = [ + p for p in preds + if isinstance(p, Gateway) + and p.gateway_type == GatewayType.XOR + and len( + self.__model.incoming_edges(p) + ) > 1 + ] + self.assertGreater( + len(xor_joins), 0, + "Self-loop task B should have an " + "XOR-join predecessor", + ) + + def test_self_loop_xor_split_after_b( + self, + ) -> None: + """Task B has an XOR-split successor. + + The self-loop restoration inserts an XOR-split + after B with a back-edge to the XOR-join. + + :return: None. + """ + b: Task = self.__model.get_task("B") + succs: set[Node] = ( + self.__model.successors(b) + ) + xor_splits: list[Gateway] = [ + s for s in succs + if isinstance(s, Gateway) + and s.gateway_type == GatewayType.XOR + and len( + self.__model.outgoing_edges(s) + ) > 1 + ] + self.assertGreater( + len(xor_splits), 0, + "Self-loop task B should have an " + "XOR-split successor", + ) + + +if __name__ == "__main__": + unittest.main()