Data Compression Benchmark (python, written by Claude Code)

Claude Code wrote this python project from the task below. It installed and ran on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: Data Compression Benchmark

Write a program that benchmarks multiple compression algorithms on given data files, comparing compression ratio, speed, memory usage, and decompression speed across algorithms and compression levels.

FUNCTIONAL REQUIREMENTS:
- Accept one or more file paths as command-line arguments to use as benchmark data
- Support benchmarking multiple compression algorithms: DEFLATE (gzip), bzip2, LZMA (xz), LZ4 (if available), and zlib at various compression levels
- For each algorithm, test at multiple compression levels (e.g., levels 1, 5, 9 for gzip)
- Measure and report for each combination: compression ratio (compressed/original), compression speed (MB/s), decompression speed (MB/s), peak memory usage, and wall-clock time
- Run each benchmark multiple times (configurable via --iterations flag, default 3) and report min/mean/max for timing measurements
- Support a --quick flag to test only the default compression level for each algorithm
- Generate a summary comparison table sorted by a configurable metric via --sort flag (ratio, compress-speed, decompress-speed; default: ratio)
- Verify data integrity: decompress each result and verify it matches the original via checksum comparison
- Support benchmarking with different data types via --generate flag: text (English prose), csv (tabular data), json (structured data), binary (random bytes), and mixed
- Print results as a formatted table to console
- Save the full benchmark report as JSON with --output flag (default: compression_benchmark.json)
- If no input files are given, generate sample data files of each type (1MB each), benchmark all algorithms on each, and display a comprehensive comparison matrix
- Handle errors: unsupported algorithms on the platform, out-of-memory during compression, and algorithm-specific limitations

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

#!/usr/bin/env python3
"""
Data Compression Benchmark
Benchmarks DEFLATE, bzip2, LZMA, LZ4, zlib comparing compression ratio,
speed, and memory usage across compression levels.

Uses lz4, zstandard, and brotli alongside Python's built-in zlib, bz2, lzma.
"""

import os
import sys
import time
import zlib
import bz2
import lzma
import json
import tracemalloc
from dataclasses import dataclass, asdict
from typing import List, Dict, Any, Callable, Optional

import lz4.frame
import lz4.block
import zstandard as zstd
import brotli


@dataclass
class BenchmarkResult:
    """Stores the result of a single compression benchmark run."""
    algorithm: str
    level: int
    original_size: int
    compressed_size: int
    compression_ratio: float
    compression_time_ms: float
    decompression_time_ms: float
    compression_speed_mbps: float
    decompression_speed_mbps: float
    peak_memory_kb: float


def generate_test_data(size_bytes: int = 1_000_000) -> bytes:
    """Generate representative test data mixing text, repeated patterns, and random bytes."""
    parts = []
    text_block = (
        "The quick brown fox jumps over the lazy dog. "
        "Data compression reduces the size of data for storage or transmission. "
        "Lossless compression allows perfect reconstruction of the original data. "
    ) * 200
    parts.append(text_block.encode('utf-8'))

    repeated_pattern = (b'\xAB\xCD\xEF\x01\x23\x45' * 500)
    parts.append(repeated_pattern)

    import random
    random.seed(42)
    random_bytes = bytes(random.getrandbits(8) for _ in range(size_bytes // 4))
    parts.append(random_bytes)

    data = b''.join(parts)
    if len(data) > size_bytes:
        data = data[:size_bytes]
    elif len(data) < size_bytes:
        data = data + b'\x00' * (size_bytes - len(data))
    return data


def measure_memory(func: Callable, *args, **kwargs):
    """Measure peak memory usage of a function call."""
    tracemalloc.start()
    result = func(*args, **kwargs)
    _, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    return result, peak / 1024  # Convert to KB


def benchmark_zlib(data: bytes, level: int) -> BenchmarkResult:
    """Benchmark zlib (DEFLATE) compression."""
    start = time.perf_counter()
    compressed, peak_mem = measure_memory(zlib.compress, data, level)
    compress_time = (time.perf_counter() - start) * 1000

    start = time.perf_counter()
    decompressed = zlib.decompress(compressed)
    decompress_time = (time.perf_counter() - start) * 1000

    assert decompressed == data, "Decompression verification failed"

    orig_size = len(data)
    comp_size = len(compressed)
    return BenchmarkResult(
        algorithm="zlib/DEFLATE",
        level=level,
        original_size=orig_size,
        compressed_size=comp_size,
        compression_ratio=orig_size / comp_size if comp_size > 0 else 0,
        compression_time_ms=compress_time,
        decompression_time_ms=decompress_time,
        compression_speed_mbps=(orig_size / (1024 * 1024)) / (compress_time / 1000) if compress_time > 0 else 0,
        decompression_speed_mbps=(orig_size / (1024 * 1024)) / (decompress_time / 1000) if decompress_time > 0 else 0,
        peak_memory_kb=peak_mem,
    )


def benchmark_bzip2(data: bytes, level: int) -> BenchmarkResult:
    """Benchmark bzip2 compression."""
    start = time.perf_counter()
    compressed, peak_mem = measure_memory(bz2.compress, data, level)
    compress_time = (time.perf_counter() - start) * 1000

    start = time.perf_counter()
    decompressed = bz2.decompress(compressed)
    decompress_time = (time.perf_counter() - start) * 1000

    assert decompressed == data, "Decompression verification failed"

    orig_size = len(data)
    comp_size = len(compressed)
    return BenchmarkResult(
        algorithm="bzip2",
        level=level,
        original_size=orig_size,
        compressed_size=comp_size,
        compression_ratio=orig_size / comp_size if comp_size > 0 else 0,
        compression_time_ms=compress_time,
        decompression_time_ms=decompress_time,
        compression_speed_mbps=(orig_size / (1024 * 1024)) / (compress_time / 1000) if compress_time > 0 else 0,
        decompression_speed_mbps=(orig_size / (1024 * 1024)) / (decompress_time / 1000) if decompress_time > 0 else 0,
        peak_memory_kb=peak_mem,
    )


def benchmark_lzma(data: bytes, level: int) -> BenchmarkResult:
    """Benchmark LZMA compression."""
    preset = min(level, 9)
    start = time.perf_counter()
    compressed, peak_mem = measure_memory(lzma.compress, data, format=lzma.FORMAT_XZ, preset=preset)
    compress_time = (time.perf_counter() - start) * 1000

    start = time.perf_counter()
    decompressed = lzma.decompress(compressed)
    decompress_time = (time.perf_counter() - start) * 1000

    assert decompressed == data, "Decompression verification failed"

    orig_size = len(data)
    comp_size = len(compressed)
    return BenchmarkResult(
        algorithm="LZMA",
        level=preset,
        original_size=orig_size,
        compressed_size=comp_size,
        compression_ratio=orig_size / comp_size if comp_size > 0 else 0,
        compression_time_ms=compress_time,
        decompression_time_ms=decompress_time,
        compression_speed_mbps=(orig_size / (1024 * 1024)) / (compress_time / 1000) if compress_time > 0 else 0,
        decompression_speed_mbps=(orig_size / (1024 * 1024)) / (decompress_time / 1000) if decompress_time > 0 else 0,
        peak_memory_kb=peak_mem,
    )


def benchmark_lz4(data: bytes, level: int) -> BenchmarkResult:
    """Benchmark LZ4 compression using lz4.frame."""
    clevel = min(level, 16)
    start = time.perf_counter()
    compressed, peak_mem = measure_memory(
        lz4.frame.compress, data, compression_level=clevel
    )
    compress_time = (time.perf_counter() - start) * 1000

    start = time.perf_counter()
    decompressed = lz4.frame.decompress(compressed)
    decompress_time = (time.perf_counter() - start) * 1000

    assert decompressed == data, "Decompression verification failed"

    orig_size = len(data)
    comp_size = len(compressed)
    return BenchmarkResult(
        algorithm="LZ4",
        level=clevel,
        original_size=orig_size,
        compressed_size=comp_size,
        compression_ratio=orig_size / comp_size if comp_size > 0 else 0,
        compression_time_ms=compress_time,
        decompression_time_ms=decompress_time,
        compression_speed_mbps=(orig_size / (1024 * 1024)) / (compress_time / 1000) if compress_time > 0 else 0,
        decompression_speed_mbps=(orig_size / (1024 * 1024)) / (decompress_time / 1000) if decompress_time > 0 else 0,
        peak_memory_kb=peak_mem,
    )


def benchmark_zstandard(data: bytes, level: int) -> BenchmarkResult:
    """Benchmark Zstandard compression."""
    clevel = min(level, 22)
    cctx = zstd.ZstdCompressor(level=clevel)

    start = time.perf_counter()
    compressed, peak_mem = measure_memory(cctx.compress, data)
    compress_time = (time.perf_counter() - start) * 1000

    dctx = zstd.ZstdDecompressor()
    start = time.perf_counter()
    decompressed = dctx.decompress(compressed)
    decompress_time = (time.perf_counter() - start) * 1000

    assert decompressed == data, "Decompression verification failed"

    orig_size = len(data)
    comp_size = len(compressed)
    return BenchmarkResult(
        algorithm="Zstandard",
        level=clevel,
        original_size=orig_size,
        compressed_size=comp_size,
        compression_ratio=orig_size / comp_size if comp_size > 0 else 0,
        compression_time_ms=compress_time,
        decompression_time_ms=decompress_time,
        compression_speed_mbps=(orig_size / (1024 * 1024)) / (compress_time / 1000) if compress_time > 0 else 0,
        decompression_speed_mbps=(orig_size / (1024 * 1024)) / (decompress_time / 1000) if decompress_time > 0 else 0,
        peak_memory_kb=peak_mem,
    )


def benchmark_brotli(data: bytes, level: int) -> BenchmarkResult:
    """Benchmark Brotli compression."""
    quality = min(level, 11)
    start = time.perf_counter()
    compressed, peak_mem = measure_memory(brotli.compress, data, quality=quality)
    compress_time = (time.perf_counter() - start) * 1000

    start = time.perf_counter()
    decompressed = brotli.decompress(compressed)
    decompress_time = (time.perf_counter() - start) * 1000

    assert decompressed == data, "Decompression verification failed"

    orig_size = len(data)
    comp_size = len(compressed)
    return BenchmarkResult(
        algorithm="Brotli",
        level=quality,
        original_size=orig_size,
        compressed_size=comp_size,
        compression_ratio=orig_size / comp_size if comp_size > 0 else 0,
        compression_time_ms=compress_time,
        decompression_time_ms=decompress_time,
        compression_speed_mbps=(orig_size / (1024 * 1024)) / (compress_time / 1000) if compress_time > 0 else 0,
        decompression_speed_mbps=(orig_size / (1024 * 1024)) / (decompress_time / 1000) if decompress_time > 0 else 0,
        peak_memory_kb=peak_mem,
    )


def run_benchmarks(
    data_size: int = 1_000_000,
    levels: Optional[List[int]] = None,
    iterations: int = 3,
) -> List[BenchmarkResult]:
    """Run all compression benchmarks across specified levels."""
    if levels is None:
        levels = [1, 3, 6, 9]

    print(f"Generating {data_size / (1024 * 1024):.1f} MB of test data...")
    data = generate_test_data(data_size)
    print(f"Test data generated: {len(data)} bytes\n")

    algorithms = [
        ("zlib/DEFLATE", benchmark_zlib, 1, 9),
        ("bzip2", benchmark_bzip2, 1, 9),
        ("LZMA", benchmark_lzma, 0, 9),
        ("LZ4", benchmark_lz4, 0, 16),
        ("Zstandard", benchmark_zstandard, 1, 22),
        ("Brotli", benchmark_brotli, 0, 11),
    ]

    all_results: List[BenchmarkResult] = []

    for algo_name, bench_func, min_level, max_level in algorithms:
        print(f"--- Benchmarking {algo_name} ---")
        for level in levels:
            effective_level = max(min_level, min(level, max_level))
            best_result = None
            for i in range(iterations):
                result = bench_func(data, effective_level)
                if best_result is None or result.compression_time_ms < best_result.compression_time_ms:
                    best_result = result

            all_results.append(best_result)
            print(
                f"  Level {best_result.level:2d}: "
                f"ratio={best_result.compression_ratio:.2f}x  "
                f"compress={best_result.compression_time_ms:.1f}ms  "
                f"decompress={best_result.decompression_time_ms:.1f}ms  "
                f"speed={best_result.compression_speed_mbps:.1f} MB/s  "
                f"memory={best_result.peak_memory_kb:.0f} KB"
            )
        print()

    return all_results


def print_summary_table(results: List[BenchmarkResult]) -> None:
    """Print a formatted summary table of benchmark results."""
    header = (
        f"{'Algorithm':<15} {'Level':>5} {'Ratio':>8} {'Comp(ms)':>10} "
        f"{'Decomp(ms)':>11} {'Speed(MB/s)':>12} {'Memory(KB)':>11}"
    )
    print("=" * len(header))
    print("COMPRESSION BENCHMARK SUMMARY")
    print("=" * len(header))
    print(header)
    print("-" * len(header))

    for r in results:
        print(
            f"{r.algorithm:<15} {r.level:>5} {r.compression_ratio:>8.2f} "
            f"{r.compression_time_ms:>10.1f} {r.decompression_time_ms:>11.1f} "
            f"{r.compression_speed_mbps:>12.1f} {r.peak_memory_kb:>11.0f}"
        )
    print("=" * len(header))


def export_results(results: List[BenchmarkResult], filename: str = "benchmark_results.json") -> None:
    """Export benchmark results to JSON."""
    data = {
        "benchmark": "Data Compression Benchmark",
        "results": [asdict(r) for r in results],
    }
    with open(filename, 'w') as f:
        json.dump(data, f, indent=2)
    print(f"\nResults exported to {filename}")


def main():
    """Main entry point for the compression benchmark."""
    print("=" * 60)
    print("  Data Compression Benchmark")
    print("  DEFLATE / bzip2 / LZMA / LZ4 / Zstandard / Brotli")
    print("=" * 60)
    print()

    data_size = 2_000_000  # 2 MB test data
    levels = [1, 3, 6, 9]
    iterations = 3

    if len(sys.argv) > 1:
        try:
            data_size = int(sys.argv[1])
        except ValueError:
            print(f"Invalid data size: {sys.argv[1]}")
            sys.exit(1)

    results = run_benchmarks(data_size=data_size, levels=levels, iterations=iterations)
    print_summary_table(results)
    export_results(results)

    # Find best compression ratio and fastest algorithm
    best_ratio = max(results, key=lambda r: r.compression_ratio)
    fastest = min(results, key=lambda r: r.compression_time_ms)
    print(f"\nBest compression ratio: {best_ratio.algorithm} level {best_ratio.level} "
          f"({best_ratio.compression_ratio:.2f}x)")
    print(f"Fastest compression: {fastest.algorithm} level {fastest.level} "
          f"({fastest.compression_time_ms:.1f}ms)")


if __name__ == "__main__":
    main()

# Data Compression Benchmark - Python (Trial 1)

Benchmarks DEFLATE, bzip2, LZMA, LZ4, Zstandard, and Brotli compression algorithms,
comparing compression ratio, speed, and memory usage across compression levels.

## Dependencies

- **lz4** (4.3.3) - LZ4 compression bindings
- **zstandard** (0.22.0) - Zstandard compression bindings
- **brotli** (1.1.0) - Brotli compression bindings
- Python built-in: zlib, bz2, lzma

## Setup

```bash
pip install -r requirements.txt
```

## Usage

```bash
# Run with default 2MB test data
python benchmark.py

# Run with custom data size (bytes)
python benchmark.py 5000000
```

## Output

Results are printed as a summary table and exported to `benchmark_results.json`.

lz4==4.3.3
zstandard==0.22.0
brotli==1.1.0