split-miner/src/split_miner/refined_dfg.py

# 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
        }