File size: 3,209 Bytes
d8967e2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | #!/usr/bin/env python3
'''
Performance Benchmark for SOMA Compression Algorithms
Measures speed and compression for all 15 algorithms
'''
import time
import json
import sys
from pathlib import Path
def load_sample_data():
'''Load sample messages for testing'''
return [
{
'role': 'user',
'content': 'Analyze this code for bugs and suggest improvements. The code processes user input.'
},
{
'role': 'assistant',
'content': 'I found several issues: 1) No input validation 2) Missing error handling 3) No type checking. Here is the corrected version with proper validation and error handling included.'
}
]
def benchmark_compressor(name, compress_func, messages, iterations=100):
'''Benchmark a single compressor'''
original_size = len(str(messages))
# Warmup
compress_func(messages.copy())
# Benchmark
start = time.time()
for _ in range(iterations):
result = compress_func(messages.copy())
elapsed = time.time() - start
compressed_size = len(str(result))
return {
'name': name,
'original_size': original_size,
'compressed_size': compressed_size,
'compression_ratio': compressed_size / original_size,
'compression_percent': (1 - compressed_size / original_size) * 100,
'avg_time_ms': (elapsed / iterations) * 1000,
'throughput_per_sec': iterations / elapsed
}
def main():
print("SOMA Compression Algorithms - Performance Benchmark")
print("=" * 70)
messages = load_sample_data()
# Add current directory to path
sys.path.insert(0, str(Path(__file__).parent))
compressors = []
# Try to import all compressors
try:
from performance_focused_compressor import compress_messages as perf
compressors.append(('Performance', perf))
except ImportError as e:
print(f"Could not import performance_focused: {e}")
try:
from aggressive_hybrid_compressor import compress_messages as agg
compressors.append(('Aggressive', agg))
except ImportError:
pass
try:
from code_focused_compressor import compress_messages as code
compressors.append(('Code-Focused', code))
except ImportError:
pass
if not compressors:
print("No compressors found! Ensure .py files are in same directory.")
return
print(f"\nTesting {len(compressors)} compressors...")
print(f"Original size: {len(str(messages))} bytes\n")
results = []
for name, func in compressors:
print(f"Benchmarking {name}...", end=' ', flush=True)
result = benchmark_compressor(name, func, messages)
results.append(result)
print(f"Done: {result['compression_percent']:.1f}% compression")
# Sort by compression
results.sort(key=lambda x: x['compression_percent'], reverse=True)
print("\n" + "=" * 70)
print("RESULTS")
print("=" * 70)
for r in results:
print(f"{r['name']}: {r['compression_percent']:.1f}% compression, "
f"{r['avg_time_ms']:.2f}ms avg")
print("\nBenchmark complete!")
if __name__ == '__main__':
main()
|