File size: 2,635 Bytes
53a0ef9 | 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 | import subprocess
import sys
import os
import json
import logging
logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
def get_python_exe():
"""Detects the best python executable to use (prefers venv)."""
# Prefer explicit venv path
venv_python = os.path.join(os.getcwd(), "venv/bin/python3")
if os.path.exists(venv_python):
return venv_python
return sys.executable
def run_benchmark():
python_exe = get_python_exe()
logging.info(f"Using Python executable: {python_exe}")
modes = ["fp32", "int8", "selective"]
summary = {}
os.makedirs("results", exist_ok=True)
os.makedirs("outputs", exist_ok=True)
for mode in modes:
logging.info(f"=== BENCHMARKING MODE: {mode} ===")
output_file = f"results/results_{mode}.json"
# Run in subprocess to ensure isolated memory measurement
cmd = [
python_exe, "src/optimize_tts.py",
"--mode", mode,
"--output_json", output_file,
"--text", "This is a benchmarking sample for CPU optimized MOSS TTS. It tests end-to-end latency."
]
try:
# Capturing outputs
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 0:
if os.path.exists(output_file):
with open(output_file, "r") as f:
summary[mode] = json.load(f)
logging.info(f"Success: {mode}")
else:
logging.error(f"Output file {output_file} not found for mode {mode}")
else:
logging.error(f"Failed to benchmark {mode}. Return code: {result.returncode}")
logging.error(f"STDERR: {result.stderr}")
except Exception as e:
logging.error(f"Subprocess error for {mode}: {e}")
# Print Summary Table
print("\n" + "="*80)
print(f"{'Quantization Mode':<20} | {'RAM (MB)':<12} | {'Latency (ms)':<15} | {'Load (s)':<10}")
print("-" * 80)
for mode in modes:
if mode in summary:
data = summary[mode]
print(f"{mode:<20} | {data['peak_ram_mb']:<12.2f} | {data['latency_ms']:<15.2f} | {data['load_time_sec']:<10.2f}")
else:
print(f"{mode:<20} | {'FAILED':<12} | {'N/A':<15} | {'N/A':<10}")
print("="*80 + "\n")
with open("results/benchmark_summary.json", "w") as f:
json.dump(summary, f, indent=4)
logging.info("Benchmark summary saved to results/benchmark_summary.json")
if __name__ == "__main__":
run_benchmark()
|