Trixy7/trixy_core/service/graph.py

# -*- coding: utf-8 -*-
"""
Abhängigkeitsgraph für Service-Visualisierung.

Ermöglicht die Erstellung von DOT- und Mermaid-Diagrammen
zur Visualisierung von Service-Abhängigkeiten.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum, auto
from typing import TYPE_CHECKING, Any, Iterator

from trixy_core.service.enums import ServicePriority, ServiceGroup, ServiceState

if TYPE_CHECKING:
    from trixy_core.service.iservice import IService


class GraphFormat(Enum):
    """
    Ausgabeformate für den Graphen.
    """

    DOT = auto()
    """GraphViz DOT-Format."""

    MERMAID = auto()
    """Mermaid-Diagramm-Format."""

    JSON = auto()
    """JSON-Struktur."""

    ASCII = auto()
    """ASCII-Baum-Darstellung."""


@dataclass
class GraphNode:
    """
    Knoten im Abhängigkeitsgraphen.
    """

    name: str
    """Name des Services."""

    priority: ServicePriority
    """Priorität des Services."""

    group: ServiceGroup
    """Gruppe des Services."""

    state: ServiceState
    """Aktueller Zustand."""

    dependencies: list[str] = field(default_factory=list)
    """Liste der Abhängigkeiten."""

    dependents: list[str] = field(default_factory=list)
    """Liste der abhängigen Services."""

    metadata: dict[str, Any] = field(default_factory=dict)
    """Zusätzliche Metadaten."""

    def __hash__(self) -> int:
        """Hash für Set-Verwendung."""
        return hash(self.name)

    def __eq__(self, other: object) -> bool:
        """Gleichheitsvergleich."""
        if isinstance(other, GraphNode):
            return self.name == other.name
        return False


@dataclass
class GraphEdge:
    """
    Kante im Abhängigkeitsgraphen.
    """

    source: str
    """Quellknoten (abhängiger Service)."""

    target: str
    """Zielknoten (Abhängigkeit)."""

    edge_type: str = "depends_on"
    """Typ der Beziehung."""

    def __hash__(self) -> int:
        """Hash für Set-Verwendung."""
        return hash((self.source, self.target))

    def __eq__(self, other: object) -> bool:
        """Gleichheitsvergleich."""
        if isinstance(other, GraphEdge):
            return self.source == other.source and self.target == other.target
        return False


@dataclass
class GraphStyle:
    """
    Styling-Optionen für die Graphen-Ausgabe.
    """

    # Farben nach Priorität
    priority_colors: dict[ServicePriority, str] = field(default_factory=lambda: {
        ServicePriority.CRITICAL: "#ff6b6b",
        ServicePriority.CORE: "#feca57",
        ServicePriority.NETWORK: "#48dbfb",
        ServicePriority.MANAGER: "#1dd1a1",
        ServicePriority.PLUGIN: "#a55eea",
        ServicePriority.OPTIONAL: "#c8d6e5",
    })

    # Farben nach Zustand
    state_colors: dict[ServiceState, str] = field(default_factory=lambda: {
        ServiceState.STOPPED: "#c8d6e5",
        ServiceState.STARTING: "#feca57",
        ServiceState.RUNNING: "#1dd1a1",
        ServiceState.STOPPING: "#ff9f43",
        ServiceState.FAILED: "#ff6b6b",
    })

    # Formen nach Gruppe
    group_shapes: dict[ServiceGroup, str] = field(default_factory=lambda: {
        ServiceGroup.CRITICAL: "doubleoctagon",
        ServiceGroup.NETWORK: "hexagon",
        ServiceGroup.MANAGER: "box",
        ServiceGroup.AUDIO: "ellipse",
        ServiceGroup.ML: "diamond",
        ServiceGroup.UTILITY: "oval",
        ServiceGroup.PLUGIN: "component",
    })

    edge_color: str = "#576574"
    """Farbe für Kanten."""

    font_name: str = "Arial"
    """Schriftart."""

    font_size: int = 12
    """Schriftgröße."""


class DependencyGraph:
    """
    Abhängigkeitsgraph für Services.

    Analysiert und visualisiert Service-Abhängigkeiten.

    Example:
        graph = DependencyGraph()
        graph.add_services(service_container.services)

        # DOT-Format exportieren
        dot = graph.to_dot()
        with open("services.dot", "w") as f:
            f.write(dot)

        # Mermaid-Diagramm
        mermaid = graph.to_mermaid()
        print(mermaid)

        # Zyklen erkennen
        cycles = graph.find_cycles()
    """

    def __init__(self, style: GraphStyle | None = None) -> None:
        """
        Initialisiert den Graphen.

        Args:
            style: Optionale Styling-Konfiguration.
        """
        self._nodes: dict[str, GraphNode] = {}
        self._edges: set[GraphEdge] = set()
        self._style = style or GraphStyle()

    @property
    def nodes(self) -> dict[str, GraphNode]:
        """Alle Knoten."""
        return dict(self._nodes)

    @property
    def edges(self) -> set[GraphEdge]:
        """Alle Kanten."""
        return set(self._edges)

    @property
    def node_count(self) -> int:
        """Anzahl der Knoten."""
        return len(self._nodes)

    @property
    def edge_count(self) -> int:
        """Anzahl der Kanten."""
        return len(self._edges)

    def add_service(self, service: "IService") -> GraphNode:
        """
        Fügt einen Service zum Graphen hinzu.

        Args:
            service: Der Service.

        Returns:
            Der erstellte GraphNode.
        """
        node = GraphNode(
            name=service.name,
            priority=service.PRIORITY,
            group=service.GROUP,
            state=service.state,
            dependencies=list(service.DEPENDENCIES),
        )
        self._nodes[service.name] = node

        # Kanten für Abhängigkeiten
        for dep in service.DEPENDENCIES:
            edge = GraphEdge(source=service.name, target=dep)
            self._edges.add(edge)

        return node

    def add_services(self, services: dict[str, "IService"]) -> None:
        """
        Fügt mehrere Services hinzu.

        Args:
            services: Dictionary der Services.
        """
        # Erst alle Nodes erstellen
        for service in services.values():
            self.add_service(service)

        # Dann Dependents berechnen
        self._compute_dependents()

    def _compute_dependents(self) -> None:
        """Berechnet die abhängigen Services für jeden Knoten."""
        for node in self._nodes.values():
            node.dependents = []

        for edge in self._edges:
            if edge.target in self._nodes:
                self._nodes[edge.target].dependents.append(edge.source)

    def get_node(self, name: str) -> GraphNode | None:
        """
        Gibt einen Knoten zurück.

        Args:
            name: Name des Knotens.

        Returns:
            GraphNode oder None.
        """
        return self._nodes.get(name)

    def get_dependencies(self, name: str) -> list[str]:
        """
        Gibt die Abhängigkeiten eines Knotens zurück.

        Args:
            name: Name des Knotens.

        Returns:
            Liste der Abhängigkeiten.
        """
        node = self._nodes.get(name)
        if node:
            return list(node.dependencies)
        return []

    def get_dependents(self, name: str) -> list[str]:
        """
        Gibt die abhängigen Services zurück.

        Args:
            name: Name des Knotens.

        Returns:
            Liste der abhängigen Services.
        """
        node = self._nodes.get(name)
        if node:
            return list(node.dependents)
        return []

    def get_all_dependencies(self, name: str) -> set[str]:
        """
        Gibt alle transitiven Abhängigkeiten zurück.

        Args:
            name: Name des Knotens.

        Returns:
            Set aller Abhängigkeiten (direkt und transitiv).
        """
        result: set[str] = set()
        visited: set[str] = set()

        def collect(node_name: str) -> None:
            if node_name in visited:
                return
            visited.add(node_name)

            node = self._nodes.get(node_name)
            if node:
                for dep in node.dependencies:
                    result.add(dep)
                    collect(dep)

        collect(name)
        return result

    def get_start_order(self) -> list[str]:
        """
        Berechnet die optimale Startreihenfolge.

        Returns:
            Liste der Service-Namen in Startreihenfolge.
        """
        # Topologische Sortierung mit Kahn's Algorithmus
        in_degree: dict[str, int] = {name: 0 for name in self._nodes}

        for edge in self._edges:
            if edge.source in in_degree and edge.target in self._nodes:
                in_degree[edge.source] += 1

        # Queue mit Knoten ohne Abhängigkeiten
        queue = [
            name for name, degree in in_degree.items()
            if degree == 0
        ]

        # Nach Priorität sortieren
        queue.sort(key=lambda n: (self._nodes[n].priority, n))

        result: list[str] = []

        while queue:
            # Nächsten Knoten verarbeiten
            current = queue.pop(0)
            result.append(current)

            # Abhängige aktualisieren
            for node in self._nodes.values():
                if current in node.dependencies:
                    in_degree[node.name] -= 1
                    if in_degree[node.name] == 0:
                        queue.append(node.name)
                        queue.sort(key=lambda n: (self._nodes[n].priority, n))

        return result

    def get_stop_order(self) -> list[str]:
        """
        Berechnet die optimale Stopp-Reihenfolge.

        Returns:
            Liste der Service-Namen in Stopp-Reihenfolge.
        """
        return list(reversed(self.get_start_order()))

    def find_cycles(self) -> list[list[str]]:
        """
        Findet Zyklen im Graphen.

        Returns:
            Liste von Zyklen (jeder Zyklus als Liste von Knoten).
        """
        cycles: list[list[str]] = []
        visited: set[str] = set()
        rec_stack: set[str] = set()
        path: list[str] = []

        def dfs(node_name: str) -> bool:
            visited.add(node_name)
            rec_stack.add(node_name)
            path.append(node_name)

            node = self._nodes.get(node_name)
            if node:
                for dep in node.dependencies:
                    if dep not in visited:
                        if dfs(dep):
                            return True
                    elif dep in rec_stack:
                        # Zyklus gefunden
                        cycle_start = path.index(dep)
                        cycle = path[cycle_start:] + [dep]
                        cycles.append(cycle)
                        return True

            path.pop()
            rec_stack.remove(node_name)
            return False

        for name in self._nodes:
            if name not in visited:
                dfs(name)

        return cycles

    def get_roots(self) -> list[str]:
        """
        Gibt die Wurzelknoten zurück (ohne Abhängigkeiten).

        Returns:
            Liste der Wurzelknoten.
        """
        return [
            name for name, node in self._nodes.items()
            if not node.dependencies
        ]

    def get_leaves(self) -> list[str]:
        """
        Gibt die Blattknoten zurück (ohne Abhängige).

        Returns:
            Liste der Blattknoten.
        """
        return [
            name for name, node in self._nodes.items()
            if not node.dependents
        ]

    def to_dot(self, title: str = "Service Dependencies") -> str:
        """
        Exportiert als GraphViz DOT-Format.

        Args:
            title: Titel des Graphen.

        Returns:
            DOT-String.
        """
        lines: list[str] = [
            f'digraph "{title}" {{',
            f'    rankdir=TB;',
            f'    node [fontname="{self._style.font_name}", fontsize={self._style.font_size}];',
            f'    edge [color="{self._style.edge_color}"];',
            "",
        ]

        # Cluster nach Gruppen
        groups: dict[ServiceGroup, list[GraphNode]] = {}
        for node in self._nodes.values():
            groups.setdefault(node.group, []).append(node)

        for group, nodes in groups.items():
            lines.append(f'    subgraph cluster_{group.name.lower()} {{')
            lines.append(f'        label="{group.name}";')
            lines.append(f'        style=rounded;')
            lines.append(f'        bgcolor="#f8f9fa";')

            for node in nodes:
                color = self._style.state_colors.get(node.state, "#c8d6e5")
                shape = self._style.group_shapes.get(node.group, "box")
                lines.append(
                    f'        "{node.name}" ['
                    f'shape={shape}, '
                    f'style=filled, '
                    f'fillcolor="{color}", '
                    f'label="{node.name}\\n({node.priority.name})"'
                    f'];'
                )

            lines.append('    }')
            lines.append('')

        # Kanten
        for edge in self._edges:
            lines.append(f'    "{edge.source}" -> "{edge.target}";')

        lines.append('}')
        return '\n'.join(lines)

    def to_mermaid(self, title: str = "Service Dependencies") -> str:
        """
        Exportiert als Mermaid-Diagramm.

        Args:
            title: Titel des Graphen.

        Returns:
            Mermaid-String.
        """
        lines: list[str] = [
            "```mermaid",
            "graph TD",
            f"    %% {title}",
            "",
        ]

        # Knoten mit Styling
        for name, node in self._nodes.items():
            # Mermaid-Shapes basierend auf Gruppe
            shape_start, shape_end = self._get_mermaid_shape(node.group)
            state_class = node.state.name.lower()

            lines.append(
                f'    {name}{shape_start}"{name}<br/>({node.priority.name})"{shape_end}'
            )

        lines.append("")

        # Kanten
        for edge in self._edges:
            lines.append(f'    {edge.source} --> {edge.target}')

        lines.append("")

        # Styling
        lines.append("    %% Styles")
        for state, color in self._style.state_colors.items():
            lines.append(f"    classDef {state.name.lower()} fill:{color}")

        # Klassen zuweisen
        for state in ServiceState:
            nodes_with_state = [
                name for name, node in self._nodes.items()
                if node.state == state
            ]
            if nodes_with_state:
                lines.append(
                    f"    class {','.join(nodes_with_state)} {state.name.lower()}"
                )

        lines.append("```")
        return '\n'.join(lines)

    def _get_mermaid_shape(self, group: ServiceGroup) -> tuple[str, str]:
        """
        Gibt die Mermaid-Shape-Zeichen für eine Gruppe zurück.

        Args:
            group: Die Service-Gruppe.

        Returns:
            Tupel (Start, Ende) für die Shape.
        """
        shapes = {
            ServiceGroup.CRITICAL: ("[[", "]]"),  # Subroutine
            ServiceGroup.NETWORK: ("{{", "}}"),  # Hexagon
            ServiceGroup.MANAGER: ("[", "]"),    # Rectangle
            ServiceGroup.AUDIO: ("([", "])"),    # Stadium
            ServiceGroup.ML: ("{", "}"),         # Rhombus
            ServiceGroup.UTILITY: ("(", ")"),    # Rounded
            ServiceGroup.PLUGIN: ("((", "))"),   # Circle
        }
        return shapes.get(group, ("[", "]"))

    def to_json(self) -> dict[str, Any]:
        """
        Exportiert als JSON-Struktur.

        Returns:
            Dictionary-Repräsentation.
        """
        return {
            "nodes": [
                {
                    "name": node.name,
                    "priority": node.priority.name,
                    "group": node.group.name,
                    "state": node.state.name,
                    "dependencies": node.dependencies,
                    "dependents": node.dependents,
                    "metadata": node.metadata,
                }
                for node in self._nodes.values()
            ],
            "edges": [
                {
                    "source": edge.source,
                    "target": edge.target,
                    "type": edge.edge_type,
                }
                for edge in self._edges
            ],
            "statistics": {
                "node_count": self.node_count,
                "edge_count": self.edge_count,
                "roots": self.get_roots(),
                "leaves": self.get_leaves(),
                "has_cycles": len(self.find_cycles()) > 0,
            },
        }

    def to_ascii(self, root: str | None = None) -> str:
        """
        Exportiert als ASCII-Baum.

        Args:
            root: Optionaler Wurzelknoten.

        Returns:
            ASCII-Baum-String.
        """
        lines: list[str] = []
        visited: set[str] = set()

        def render_tree(name: str, prefix: str = "", is_last: bool = True) -> None:
            if name in visited:
                connector = "└── " if is_last else "├── "
                lines.append(f"{prefix}{connector}{name} (circular)")
                return

            visited.add(name)

            node = self._nodes.get(name)
            if not node:
                return

            connector = "└── " if is_last else "├── "
            state_marker = "●" if node.state == ServiceState.RUNNING else "○"
            lines.append(f"{prefix}{connector}{state_marker} {name} [{node.priority.name}]")

            # Abhängige rendern
            dependents = node.dependents
            for i, dep in enumerate(dependents):
                is_last_dep = i == len(dependents) - 1
                new_prefix = prefix + ("    " if is_last else "│   ")
                render_tree(dep, new_prefix, is_last_dep)

        # Startpunkte bestimmen
        if root and root in self._nodes:
            roots = [root]
        else:
            roots = self.get_roots()

        for i, root_name in enumerate(sorted(roots)):
            is_last_root = i == len(roots) - 1
            render_tree(root_name, "", is_last_root)

        return '\n'.join(lines)

    def export(
        self,
        format: GraphFormat,
        title: str = "Service Dependencies",
    ) -> str:
        """
        Exportiert den Graphen im angegebenen Format.

        Args:
            format: Ausgabeformat.
            title: Titel des Graphen.

        Returns:
            Export-String.
        """
        if format == GraphFormat.DOT:
            return self.to_dot(title)
        elif format == GraphFormat.MERMAID:
            return self.to_mermaid(title)
        elif format == GraphFormat.JSON:
            import json
            return json.dumps(self.to_json(), indent=2)
        elif format == GraphFormat.ASCII:
            return self.to_ascii()
        else:
            raise ValueError(f"Unbekanntes Format: {format}")

    def validate(self) -> list[str]:
        """
        Validiert den Graphen.

        Returns:
            Liste von Warnungen/Fehlern.
        """
        issues: list[str] = []

        # Fehlende Abhängigkeiten prüfen
        for edge in self._edges:
            if edge.target not in self._nodes:
                issues.append(
                    f"Fehlende Abhängigkeit: {edge.source} -> {edge.target}"
                )

        # Zyklen prüfen
        cycles = self.find_cycles()
        for cycle in cycles:
            issues.append(f"Zyklische Abhängigkeit: {' -> '.join(cycle)}")

        # Verwaiste Knoten
        orphans = [
            name for name, node in self._nodes.items()
            if not node.dependencies and not node.dependents
        ]
        for orphan in orphans:
            issues.append(f"Verwaister Service: {orphan}")

        return issues