types_.py

"""Shared data types used across Split Miner phases."""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Iterable, Literal

NodeKind = Literal["task", "xor", "and", "or", "start", "end"]


@dataclass
class Node:
    id: str
    kind: NodeKind
    label: str = ""


@dataclass
class WorkingGraph:
    """Mutable adjacency representation used between phases 4-7.

    Edges are unweighted directed (source_id, target_id) pairs. Multiple
    parallel edges between the same pair of nodes are not supported.
    """

    nodes: dict[str, Node] = field(default_factory=dict)
    out_edges: dict[str, list[str]] = field(default_factory=dict)
    in_edges: dict[str, list[str]] = field(default_factory=dict)

    start_id: str = ""
    end_id: str = ""

    # Concurrency relation from phase 2 (symmetric pairs of task ids).
    concurrency: set[frozenset[str]] = field(default_factory=set)

    # Self-loops detected in phase 1: task ids that should carry a self-loop
    # in the final BPMN (reattached after split/join discovery).
    self_loops: set[str] = field(default_factory=set)

    _id_counter: int = 0

    def fresh_id(self, prefix: str) -> str:
        self._id_counter += 1
        return f"{prefix}_{self._id_counter}"

    def add_node(self, kind: NodeKind, label: str = "", node_id: str | None = None) -> str:
        if node_id is None:
            node_id = self.fresh_id(kind)
        self.nodes[node_id] = Node(id=node_id, kind=kind, label=label)
        self.out_edges.setdefault(node_id, [])
        self.in_edges.setdefault(node_id, [])
        return node_id

    def add_edge(self, src: str, tgt: str) -> None:
        if tgt not in self.out_edges[src]:
            self.out_edges[src].append(tgt)
        if src not in self.in_edges[tgt]:
            self.in_edges[tgt].append(src)

    def remove_edge(self, src: str, tgt: str) -> None:
        if tgt in self.out_edges.get(src, []):
            self.out_edges[src].remove(tgt)
        if src in self.in_edges.get(tgt, []):
            self.in_edges[tgt].remove(src)

    def remove_node(self, node_id: str) -> None:
        for s in list(self.in_edges.get(node_id, [])):
            self.remove_edge(s, node_id)
        for t in list(self.out_edges.get(node_id, [])):
            self.remove_edge(node_id, t)
        self.in_edges.pop(node_id, None)
        self.out_edges.pop(node_id, None)
        self.nodes.pop(node_id, None)

    def successors(self, node_id: str) -> list[str]:
        return list(self.out_edges.get(node_id, []))

    def predecessors(self, node_id: str) -> list[str]:
        return list(self.in_edges.get(node_id, []))

    def edges(self) -> list[tuple[str, str]]:
        return [(s, t) for s, ts in self.out_edges.items() for t in ts]

    def is_concurrent(self, a: str, b: str) -> bool:
        return frozenset((a, b)) in self.concurrency


# A DFG is just a frequency dict over directed task pairs.
DFG = dict[tuple[str, str], int]


def activities(dfg: DFG, extra: Iterable[str] = ()) -> set[str]:
    s: set[str] = set(extra)
    for a, b in dfg.keys():
        s.add(a)
        s.add(b)
    return s