Trixy7/trixy_core/network/compression/compressor.py

# -*- coding: utf-8 -*-
"""
Basis-Kompressor-Interface.

Definiert die abstrakte Schnittstelle für alle
Komprimierungsimplementierungen.
"""

from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from enum import IntEnum
from typing import Any


class CompressionLevel(IntEnum):
    """Komprimierungsstufen."""
    NONE = 0           # Keine Komprimierung
    FASTEST = 1        # Schnellste Komprimierung
    FAST = 3           # Schnell
    DEFAULT = 6        # Standard
    BEST = 9           # Beste Komprimierung (langsamste)


@dataclass
class CompressionStats:
    """Statistiken über Komprimierungsoperationen."""
    total_compressed: int = 0        # Anzahl Komprimierungen
    total_decompressed: int = 0      # Anzahl Dekomprimierungen
    bytes_in: int = 0                # Eingabe-Bytes insgesamt
    bytes_out: int = 0               # Ausgabe-Bytes insgesamt
    compression_ratio: float = 0.0   # Durchschnittliches Verhältnis
    total_time_ms: float = 0.0       # Gesamtzeit in ms
    errors: int = 0                  # Fehler
    last_operation: datetime | None = None

    def update_compression(
        self,
        input_size: int,
        output_size: int,
        time_ms: float,
    ) -> None:
        """Aktualisiert Statistik nach Komprimierung."""
        self.total_compressed += 1
        self.bytes_in += input_size
        self.bytes_out += output_size
        self.total_time_ms += time_ms
        self.last_operation = datetime.now()

        if self.bytes_in > 0:
            self.compression_ratio = self.bytes_out / self.bytes_in

    def update_decompression(
        self,
        input_size: int,
        output_size: int,
        time_ms: float,
    ) -> None:
        """Aktualisiert Statistik nach Dekomprimierung."""
        self.total_decompressed += 1
        self.total_time_ms += time_ms
        self.last_operation = datetime.now()

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert zu Dictionary."""
        return {
            "total_compressed": self.total_compressed,
            "total_decompressed": self.total_decompressed,
            "bytes_in": self.bytes_in,
            "bytes_out": self.bytes_out,
            "compression_ratio": round(self.compression_ratio, 4),
            "total_time_ms": round(self.total_time_ms, 2),
            "errors": self.errors,
            "space_saved_percent": round(
                (1 - self.compression_ratio) * 100, 2
            ) if self.compression_ratio > 0 else 0,
        }


class Compressor(ABC):
    """
    Abstrakte Basis-Klasse für Kompressoren.

    Definiert die Schnittstelle die alle konkreten
    Komprimierungsimplementierungen erfüllen müssen.

    Beispiel:
        compressor = ZlibCompressor(level=CompressionLevel.BEST)
        compressed = compressor.compress(data)
        original = compressor.decompress(compressed)
    """

    def __init__(
        self,
        level: CompressionLevel = CompressionLevel.DEFAULT,
    ) -> None:
        """
        Initialisiert den Kompressor.

        Args:
            level: Komprimierungsstufe
        """
        self.level = level
        self.stats = CompressionStats()

    @property
    @abstractmethod
    def name(self) -> str:
        """Name des Komprimierungsalgorithmus."""
        pass

    @property
    @abstractmethod
    def content_type(self) -> str:
        """Content-Type für komprimierte Daten."""
        pass

    @property
    def min_size(self) -> int:
        """
        Minimale Größe für Komprimierung.

        Daten unter dieser Größe werden nicht komprimiert,
        da der Overhead größer als der Nutzen sein kann.
        """
        return 100  # 100 Bytes Standard

    def should_compress(self, data: bytes) -> bool:
        """
        Prüft ob Daten komprimiert werden sollten.

        Args:
            data: Zu prüfende Daten

        Returns:
            True wenn Komprimierung sinnvoll
        """
        return len(data) >= self.min_size

    @abstractmethod
    def compress(self, data: bytes) -> bytes:
        """
        Komprimiert Daten.

        Args:
            data: Unkomprimierte Daten

        Returns:
            Komprimierte Daten
        """
        pass

    @abstractmethod
    def decompress(self, data: bytes) -> bytes:
        """
        Dekomprimiert Daten.

        Args:
            data: Komprimierte Daten

        Returns:
            Ursprüngliche Daten
        """
        pass

    def compress_if_beneficial(self, data: bytes) -> tuple[bytes, bool]:
        """
        Komprimiert nur wenn es sich lohnt.

        Args:
            data: Zu komprimierende Daten

        Returns:
            Tuple (Daten, ob_komprimiert)
        """
        if not self.should_compress(data):
            return data, False

        compressed = self.compress(data)

        # Nur verwenden wenn kleiner
        if len(compressed) < len(data):
            return compressed, True

        return data, False

    def get_stats(self) -> CompressionStats:
        """Gibt Komprimierungsstatistiken zurück."""
        return self.stats

    def reset_stats(self) -> None:
        """Setzt Statistiken zurück."""
        self.stats = CompressionStats()

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}(level={self.level.name})"


class CompressorError(Exception):
    """Fehler bei Komprimierung/Dekomprimierung."""
    pass


class CompressionError(CompressorError):
    """Fehler bei Komprimierung."""
    pass


class DecompressionError(CompressorError):
    """Fehler bei Dekomprimierung."""
    pass