Trixy7/trixy_core/network/compression/lz4.py

# -*- coding: utf-8 -*-
"""
LZ4-Kompressor.

Implementiert schnelle Komprimierung mit dem LZ4-Algorithmus.
Erfordert das optionale lz4-Paket.
"""

from __future__ import annotations

import time
from typing import Any

try:
    import lz4.frame
    import lz4.block
    LZ4_AVAILABLE = True
except ImportError:
    LZ4_AVAILABLE = False

from trixy_core.network.compression.compressor import (
    Compressor,
    CompressionLevel,
    CompressionError,
    DecompressionError,
)


if not LZ4_AVAILABLE:
    raise ImportError(
        "LZ4-Kompressor erfordert das lz4-Paket. "
        "Installieren mit: pip install lz4"
    )


class LZ4Compressor(Compressor):
    """
    Kompressor basierend auf LZ4.

    LZ4 bietet extrem schnelle Komprimierung bei moderater
    Komprimierungsrate. Ideal für Echtzeit-Anwendungen.

    Beispiel:
        compressor = LZ4Compressor(level=CompressionLevel.FAST)
        compressed = compressor.compress(data)
        original = compressor.decompress(compressed)
    """

    # Mapping von CompressionLevel zu LZ4-Levels
    _LEVEL_MAP = {
        CompressionLevel.NONE: 0,
        CompressionLevel.FASTEST: 0,
        CompressionLevel.FAST: 3,
        CompressionLevel.DEFAULT: 6,
        CompressionLevel.BEST: 12,
    }

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

        Args:
            level: Komprimierungsstufe
            use_frame: Frame-Format verwenden (empfohlen)
                       False für Block-Format (schneller, aber ohne Streaming)
        """
        super().__init__(level)
        self._use_frame = use_frame
        self._lz4_level = self._LEVEL_MAP.get(level, 6)

    @property
    def name(self) -> str:
        return "lz4"

    @property
    def content_type(self) -> str:
        return "application/x-lz4"

    @property
    def min_size(self) -> int:
        """LZ4 hat minimalen Overhead."""
        return 16

    def compress(self, data: bytes) -> bytes:
        """
        Komprimiert Daten mit LZ4.

        Args:
            data: Unkomprimierte Daten

        Returns:
            Komprimierte Daten
        """
        if not data:
            return b""

        start = time.monotonic()

        try:
            if self._use_frame:
                compressed = lz4.frame.compress(
                    data,
                    compression_level=self._lz4_level,
                )
            else:
                compressed = lz4.block.compress(
                    data,
                    compression=self._lz4_level,
                    store_size=True,
                )
        except Exception as e:
            self.stats.errors += 1
            raise CompressionError(f"LZ4-Komprimierung fehlgeschlagen: {e}") from e

        elapsed_ms = (time.monotonic() - start) * 1000
        self.stats.update_compression(len(data), len(compressed), elapsed_ms)

        return compressed

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

        Args:
            data: Komprimierte Daten

        Returns:
            Ursprüngliche Daten
        """
        if not data:
            return b""

        start = time.monotonic()

        try:
            if self._use_frame:
                decompressed = lz4.frame.decompress(data)
            else:
                decompressed = lz4.block.decompress(data)
        except Exception as e:
            self.stats.errors += 1
            raise DecompressionError(
                f"LZ4-Dekomprimierung fehlgeschlagen: {e}"
            ) from e

        elapsed_ms = (time.monotonic() - start) * 1000
        self.stats.update_decompression(len(data), len(decompressed), elapsed_ms)

        return decompressed


class LZ4HCCompressor(Compressor):
    """
    LZ4 High Compression Kompressor.

    Verwendet LZ4 HC-Modus für bessere Komprimierungsrate
    bei langsamerer Geschwindigkeit.
    """

    def __init__(
        self,
        level: CompressionLevel = CompressionLevel.BEST,
    ) -> None:
        super().__init__(level)
        # HC-Mode verwendet Levels 1-12
        self._hc_level = min(12, max(1, int(level) + 3))

    @property
    def name(self) -> str:
        return "lz4hc"

    @property
    def content_type(self) -> str:
        return "application/x-lz4hc"

    def compress(self, data: bytes) -> bytes:
        """Komprimiert mit LZ4 HC."""
        if not data:
            return b""

        start = time.monotonic()

        try:
            compressed = lz4.frame.compress(
                data,
                compression_level=lz4.frame.COMPRESSIONLEVEL_MAX,
            )
        except Exception as e:
            self.stats.errors += 1
            raise CompressionError(f"LZ4-HC-Komprimierung fehlgeschlagen: {e}") from e

        elapsed_ms = (time.monotonic() - start) * 1000
        self.stats.update_compression(len(data), len(compressed), elapsed_ms)

        return compressed

    def decompress(self, data: bytes) -> bytes:
        """Dekomprimiert LZ4-HC-Daten (identisch zu LZ4)."""
        if not data:
            return b""

        start = time.monotonic()

        try:
            decompressed = lz4.frame.decompress(data)
        except Exception as e:
            self.stats.errors += 1
            raise DecompressionError(
                f"LZ4-HC-Dekomprimierung fehlgeschlagen: {e}"
            ) from e

        elapsed_ms = (time.monotonic() - start) * 1000
        self.stats.update_decompression(len(data), len(decompressed), elapsed_ms)

        return decompressed