Trixy7/trixy_core/utils/decorators/timing.py

# -*- coding: utf-8 -*-
"""
Timing-Decorators für Performance-Messung.

Bietet @timed für einfache Zeitmessung und @profile für
detailliertes Profiling von Funktionen.
"""

from __future__ import annotations

import asyncio
import functools
import time
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Callable, ParamSpec, TypeVar


P = ParamSpec("P")
R = TypeVar("R")


@dataclass
class TimingResult:
    """
    Ergebnis einer Zeitmessung.

    Enthält alle relevanten Timing-Informationen für eine
    Funktionsausführung.
    """

    function_name: str
    duration_seconds: float
    start_time: datetime
    end_time: datetime
    args_count: int = 0
    kwargs_count: int = 0
    success: bool = True
    error: str | None = None

    @property
    def duration_ms(self) -> float:
        """Gibt die Dauer in Millisekunden zurück."""
        return self.duration_seconds * 1000

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert das Ergebnis in ein Dictionary."""
        return {
            "function": self.function_name,
            "duration_seconds": self.duration_seconds,
            "duration_ms": self.duration_ms,
            "start_time": self.start_time.isoformat(),
            "end_time": self.end_time.isoformat(),
            "success": self.success,
            "error": self.error,
        }


@dataclass
class ProfilingStats:
    """
    Aggregierte Profiling-Statistiken für eine Funktion.

    Sammelt Statistiken über mehrere Aufrufe hinweg.
    """

    function_name: str
    call_count: int = 0
    total_time: float = 0.0
    min_time: float = float("inf")
    max_time: float = 0.0
    error_count: int = 0
    last_call: datetime | None = None
    _times: list[float] = field(default_factory=list)

    @property
    def avg_time(self) -> float:
        """Durchschnittliche Ausführungszeit."""
        return self.total_time / self.call_count if self.call_count > 0 else 0.0

    @property
    def success_rate(self) -> float:
        """Erfolgsrate in Prozent."""
        if self.call_count == 0:
            return 100.0
        return ((self.call_count - self.error_count) / self.call_count) * 100

    def record(self, duration: float, success: bool = True) -> None:
        """
        Zeichnet eine Ausführung auf.

        Args:
            duration: Dauer in Sekunden.
            success: Ob die Ausführung erfolgreich war.
        """
        self.call_count += 1
        self.total_time += duration
        self.min_time = min(self.min_time, duration)
        self.max_time = max(self.max_time, duration)
        self.last_call = datetime.now()

        if not success:
            self.error_count += 1

        # Für Percentil-Berechnung (limitiert auf letzte 1000 Werte)
        self._times.append(duration)
        if len(self._times) > 1000:
            self._times.pop(0)

    def percentile(self, p: float) -> float:
        """
        Berechnet das p-te Percentil der Ausführungszeiten.

        Args:
            p: Percentil zwischen 0 und 100.

        Returns:
            Percentil-Wert in Sekunden.
        """
        if not self._times:
            return 0.0

        sorted_times = sorted(self._times)
        index = int((p / 100) * len(sorted_times))
        index = min(index, len(sorted_times) - 1)
        return sorted_times[index]

    def reset(self) -> None:
        """Setzt alle Statistiken zurück."""
        self.call_count = 0
        self.total_time = 0.0
        self.min_time = float("inf")
        self.max_time = 0.0
        self.error_count = 0
        self.last_call = None
        self._times.clear()

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert die Statistiken in ein Dictionary."""
        return {
            "function": self.function_name,
            "call_count": self.call_count,
            "total_time_seconds": self.total_time,
            "avg_time_seconds": self.avg_time,
            "min_time_seconds": self.min_time if self.min_time != float("inf") else 0,
            "max_time_seconds": self.max_time,
            "p50_seconds": self.percentile(50),
            "p95_seconds": self.percentile(95),
            "p99_seconds": self.percentile(99),
            "error_count": self.error_count,
            "success_rate": self.success_rate,
            "last_call": self.last_call.isoformat() if self.last_call else None,
        }


# Globaler Speicher für Profiling-Statistiken
_profiling_stats: dict[str, ProfilingStats] = {}


def get_profiling_stats(function_name: str | None = None) -> dict[str, ProfilingStats]:
    """
    Gibt Profiling-Statistiken zurück.

    Args:
        function_name: Optional spezifischer Funktionsname.

    Returns:
        Dictionary mit Statistiken.
    """
    if function_name:
        if function_name in _profiling_stats:
            return {function_name: _profiling_stats[function_name]}
        return {}
    return dict(_profiling_stats)


def reset_profiling_stats(function_name: str | None = None) -> None:
    """
    Setzt Profiling-Statistiken zurück.

    Args:
        function_name: Optional spezifischer Funktionsname.
    """
    if function_name:
        if function_name in _profiling_stats:
            _profiling_stats[function_name].reset()
    else:
        _profiling_stats.clear()


def timed(
    callback: Callable[[TimingResult], None] | None = None,
    log_args: bool = False,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """
    Decorator zur Zeitmessung einer Funktion.

    Misst die Ausführungszeit und ruft optional einen Callback
    mit dem Ergebnis auf.

    Args:
        callback: Optionale Funktion, die mit TimingResult aufgerufen wird.
        log_args: Ob Argument-Anzahl mitgeloggt werden soll.

    Returns:
        Decorator-Funktion.

    Example:
        @timed()
        def slow_function():
            time.sleep(1)

        @timed(callback=lambda r: print(f"Took {r.duration_ms}ms"))
        def another_function():
            pass
    """
    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            start_time = datetime.now()
            start = time.perf_counter()
            error_msg = None
            success = True

            try:
                result = func(*args, **kwargs)
                return result
            except Exception as e:
                success = False
                error_msg = str(e)
                raise
            finally:
                end = time.perf_counter()
                end_time = datetime.now()
                duration = end - start

                timing_result = TimingResult(
                    function_name=func.__qualname__,
                    duration_seconds=duration,
                    start_time=start_time,
                    end_time=end_time,
                    args_count=len(args) if log_args else 0,
                    kwargs_count=len(kwargs) if log_args else 0,
                    success=success,
                    error=error_msg,
                )

                if callback:
                    callback(timing_result)

        return wrapper
    return decorator


def async_timed(
    callback: Callable[[TimingResult], None] | None = None,
    log_args: bool = False,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """
    Async-Decorator zur Zeitmessung.

    Wie @timed, aber für async Funktionen.

    Args:
        callback: Optionale Funktion, die mit TimingResult aufgerufen wird.
        log_args: Ob Argument-Anzahl mitgeloggt werden soll.

    Returns:
        Decorator-Funktion.

    Example:
        @async_timed()
        async def slow_async_function():
            await asyncio.sleep(1)
    """
    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(func)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            start_time = datetime.now()
            start = time.perf_counter()
            error_msg = None
            success = True

            try:
                result = await func(*args, **kwargs)
                return result
            except Exception as e:
                success = False
                error_msg = str(e)
                raise
            finally:
                end = time.perf_counter()
                end_time = datetime.now()
                duration = end - start

                timing_result = TimingResult(
                    function_name=func.__qualname__,
                    duration_seconds=duration,
                    start_time=start_time,
                    end_time=end_time,
                    args_count=len(args) if log_args else 0,
                    kwargs_count=len(kwargs) if log_args else 0,
                    success=success,
                    error=error_msg,
                )

                if callback:
                    callback(timing_result)

        return wrapper
    return decorator


def profile(
    name: str | None = None,
    threshold_seconds: float | None = None,
    callback: Callable[[ProfilingStats], None] | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """
    Decorator für detailliertes Profiling.

    Sammelt Statistiken über mehrere Aufrufe hinweg.

    Args:
        name: Optionaler Name für die Statistiken (Default: Funktionsname).
        threshold_seconds: Optionaler Schwellenwert für Warnungen.
        callback: Optionaler Callback bei Schwellenwert-Überschreitung.

    Returns:
        Decorator-Funktion.

    Example:
        @profile(threshold_seconds=1.0)
        def database_query():
            # Langsame Abfrage
            pass

        # Statistiken abrufen
        stats = get_profiling_stats("database_query")
    """
    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        stats_name = name or func.__qualname__

        # Statistik-Objekt erstellen falls nicht vorhanden
        if stats_name not in _profiling_stats:
            _profiling_stats[stats_name] = ProfilingStats(function_name=stats_name)

        stats = _profiling_stats[stats_name]

        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            start = time.perf_counter()
            success = True

            try:
                result = func(*args, **kwargs)
                return result
            except Exception:
                success = False
                raise
            finally:
                duration = time.perf_counter() - start
                stats.record(duration, success)

                # Schwellenwert-Prüfung
                if threshold_seconds and duration > threshold_seconds:
                    if callback:
                        callback(stats)

        # Statistik-Zugriff über die Funktion ermöglichen
        wrapper.stats = stats  # type: ignore
        wrapper.reset_stats = stats.reset  # type: ignore

        return wrapper

    return decorator


def async_profile(
    name: str | None = None,
    threshold_seconds: float | None = None,
    callback: Callable[[ProfilingStats], None] | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """
    Async-Decorator für detailliertes Profiling.

    Args:
        name: Optionaler Name für die Statistiken.
        threshold_seconds: Optionaler Schwellenwert für Warnungen.
        callback: Optionaler Callback bei Schwellenwert-Überschreitung.

    Returns:
        Decorator-Funktion.
    """
    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        stats_name = name or func.__qualname__

        if stats_name not in _profiling_stats:
            _profiling_stats[stats_name] = ProfilingStats(function_name=stats_name)

        stats = _profiling_stats[stats_name]

        @functools.wraps(func)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            start = time.perf_counter()
            success = True

            try:
                result = await func(*args, **kwargs)
                return result
            except Exception:
                success = False
                raise
            finally:
                duration = time.perf_counter() - start
                stats.record(duration, success)

                if threshold_seconds and duration > threshold_seconds:
                    if callback:
                        callback(stats)

        wrapper.stats = stats  # type: ignore
        wrapper.reset_stats = stats.reset  # type: ignore

        return wrapper

    return decorator