#!/opt/homebrew/bin/python3.11 """ 微基准测试 - 测试合约合规处理器的初始化开销 Micro Benchmark - Test initialization overhead of contract-compliant processors """ import sys import json import os import time import subprocess import statistics from datetime import datetime from typing import Dict, List, Any # Test configuration NUM_RUNS = 10 # More runs for statistical significance # Processors to test PROCESSORS = { "asr": { "legacy": "scripts/asr_processor.py", "contract": "scripts/asr_processor_contract_v2.py", }, "ocr": { "legacy": "scripts/ocr_processor.py", "contract": "scripts/ocr_processor_contract_v1.py", }, } def measure_import_time(script_path: str) -> float: """测量处理器导入时间""" test_code = f""" import sys import time start_time = time.time() try: # Import the module sys.path.insert(0, 'scripts') import {os.path.basename(script_path).replace(".py", "")} as processor elapsed = time.time() - start_time print(f"IMPORT_TIME:{{elapsed}}") except Exception as e: print(f"IMPORT_ERROR:{{e}}") sys.exit(1) """ try: result = subprocess.run( [sys.executable, "-c", test_code], capture_output=True, text=True, timeout=10, cwd=os.path.dirname(os.path.abspath(__file__)), ) for line in result.stdout.split("\n"): if line.startswith("IMPORT_TIME:"): return float(line.split(":")[1]) return float("inf") # Failed to measure except Exception as e: print(f"测量导入时间失败: {e}") return float("inf") def measure_health_check_time(script_path: str) -> float: """测量健康检查执行时间""" cmd = [sys.executable, script_path, "--check-health", "dummy.mp4", "dummy.json"] try: start_time = time.time() result = subprocess.run( cmd, capture_output=True, text=True, timeout=30, cwd=os.path.dirname(os.path.abspath(__file__)), ) elapsed = time.time() - start_time if result.returncode == 0: return elapsed else: return float("inf") except Exception as e: print(f"测量健康检查时间失败: {e}") return float("inf") def run_micro_benchmark(): """运行微基准测试""" print("=" * 80) print("微基准测试 - 合约合规处理器开销分析") print(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print("=" * 80) print() results = {} # Test each processor for processor_type in PROCESSORS: print(f"\n测试 {processor_type.upper()} 处理器...") print("-" * 40) processor_results = { "legacy": {"import_times": [], "health_check_times": [], "summary": {}}, "contract": {"import_times": [], "health_check_times": [], "summary": {}}, } # Test both versions for version in ["legacy", "contract"]: print(f"\n版本: {version}") script_path = PROCESSORS[processor_type][version] # Measure import time (multiple runs) print(" 测量导入时间...") import_times = [] for run in range(NUM_RUNS): import_time = measure_import_time(script_path) if import_time < float("inf"): import_times.append(import_time) print(f" 运行 #{run}: {import_time * 1000:.1f} ms") else: print(f" 运行 #{run}: 失败") # Measure health check time (multiple runs) print(" 测量健康检查时间...") health_check_times = [] for run in range(NUM_RUNS): health_check_time = measure_health_check_time(script_path) if health_check_time < float("inf"): health_check_times.append(health_check_time) print(f" 运行 #{run}: {health_check_time * 1000:.1f} ms") else: print(f" 运行 #{run}: 失败") # Store results processor_results[version]["import_times"] = import_times processor_results[version]["health_check_times"] = health_check_times # Calculate statistics if import_times: processor_results[version]["summary"]["import"] = { "runs": len(import_times), "min_ms": min(import_times) * 1000, "max_ms": max(import_times) * 1000, "avg_ms": statistics.mean(import_times) * 1000, "median_ms": statistics.median(import_times) * 1000, "std_dev_ms": statistics.stdev(import_times) * 1000 if len(import_times) > 1 else 0, } if health_check_times: processor_results[version]["summary"]["health_check"] = { "runs": len(health_check_times), "min_ms": min(health_check_times) * 1000, "max_ms": max(health_check_times) * 1000, "avg_ms": statistics.mean(health_check_times) * 1000, "median_ms": statistics.median(health_check_times) * 1000, "std_dev_ms": statistics.stdev(health_check_times) * 1000 if len(health_check_times) > 1 else 0, } results[processor_type] = processor_results # Calculate overhead if processor_results["legacy"]["summary"].get("import") and processor_results[ "contract" ]["summary"].get("import"): legacy_import = processor_results["legacy"]["summary"]["import"]["avg_ms"] contract_import = processor_results["contract"]["summary"]["import"][ "avg_ms" ] import_overhead = ((contract_import - legacy_import) / legacy_import) * 100 print(f"\n导入开销分析:") print(f" 传统版本: {legacy_import:.1f} ms") print(f" 合约版本: {contract_import:.1f} ms") print(f" 开销: {import_overhead:.1f}%") if import_overhead <= 5: print(f" ✅ 通过: 导入开销 ≤ 5%") else: print(f" ❌ 失败: 导入开销 > 5%") if processor_results["legacy"]["summary"].get( "health_check" ) and processor_results["contract"]["summary"].get("health_check"): legacy_hc = processor_results["legacy"]["summary"]["health_check"]["avg_ms"] contract_hc = processor_results["contract"]["summary"]["health_check"][ "avg_ms" ] hc_overhead = ((contract_hc - legacy_hc) / legacy_hc) * 100 print(f"\n健康检查开销分析:") print(f" 传统版本: {legacy_hc:.1f} ms") print(f" 合约版本: {contract_hc:.1f} ms") print(f" 开销: {hc_overhead:.1f}%") if hc_overhead <= 5: print(f" ✅ 通过: 健康检查开销 ≤ 5%") else: print(f" ❌ 失败: 健康检查开销 > 5%") # Generate final report print("\n" + "=" * 80) print("微基准测试完成报告") print("=" * 80) # Save detailed results report_file = f"/tmp/micro_benchmark_report_{int(time.time())}.json" with open(report_file, "w") as f: json.dump( { "timestamp": datetime.now().isoformat(), "test_config": { "num_runs": NUM_RUNS, "processors_tested": list(PROCESSORS.keys()), }, "results": results, }, f, indent=2, ensure_ascii=False, ) print(f"\n详细报告保存到: {report_file}") print("=" * 80) return True if __name__ == "__main__": success = run_micro_benchmark() sys.exit(0 if success else 1)