zstd

import zstd original_data = b"This is some data that will be compressed using Zstandard. It\'s a fairly long string to demonstrate compression efficiency." * 100 # Compress data # level can range from -100 (ultra-fast) to 22 (slowest, best compression) # threads can be 0 (auto-tune) or a specific number compressed_data = zstd.compress(original_data, level=3, threads=0) print(f"Original size: {len(original_data)} bytes") print(f"Compressed size: {len(compressed_data)} bytes") print(f"Compression ratio: {len(compressed_data) / len(original_data):.2f}") # Decompress data decompressed_data = zstd.decompress(compressed_data) assert original_data == decompressed_data print("Decompression successful! Data matches original.") # Example of streaming compression for large data cctx = zstd.ZstdCompressor(level=1) dctx = zstd.ZstdDecompressor() chunk_size = len(original_data) // 5 compressed_chunks = [] # Stream compress for i in range(0, len(original_data), chunk_size): chunk = original_data[i:i + chunk_size] compressed_chunk = cctx.compress(chunk) compressed_chunks.append(compressed_chunk) # Important: flush the compressor to finalize the frame compressed_chunks.append(cctx.flush()) streaming_compressed_data = b''.join(compressed_chunks) # Stream decompress streaming_decompressed_data = dctx.decompress(streaming_compressed_data) assert original_data == streaming_decompressed_data print("Streaming decompression successful! Data matches original.")