365 lines
12 KiB
Python
365 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
批次性能测试结果可视化
|
|
生成批次大小 vs 性能指标的对比图表
|
|
"""
|
|
|
|
import json
|
|
import matplotlib.pyplot as plt
|
|
import numpy as np
|
|
import os
|
|
from datetime import datetime
|
|
|
|
# 设置中文字体
|
|
plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei', 'DejaVu Sans']
|
|
plt.rcParams['axes.unicode_minus'] = False
|
|
|
|
def load_results(json_file):
|
|
"""加载测试结果"""
|
|
with open(json_file, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
|
|
def create_throughput_chart(results, output_dir):
|
|
"""创建吞吐量对比图表"""
|
|
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))
|
|
|
|
# 提取成功的测试数据
|
|
batch_sizes = []
|
|
throughputs = []
|
|
latencies = []
|
|
|
|
for test in results['batch_tests']:
|
|
if test['success']:
|
|
batch_sizes.append(test['batch_size'])
|
|
throughputs.append(test['avg_throughput'])
|
|
latencies.append(test['avg_latency_ms'])
|
|
|
|
if not batch_sizes:
|
|
print("⚠️ 没有成功的测试数据")
|
|
return
|
|
|
|
# 吞吐量图表
|
|
ax1.plot(batch_sizes, throughputs, 'o-', color='#4ECDC4',
|
|
linewidth=2, markersize=10, label='吞吐量')
|
|
ax1.set_title('批次大小 vs 吞吐量', fontsize=14, fontweight='bold')
|
|
ax1.set_xlabel('批次大小', fontsize=12)
|
|
ax1.set_ylabel('吞吐量 (FPS)', fontsize=12)
|
|
ax1.grid(True, alpha=0.3)
|
|
ax1.legend()
|
|
|
|
# 添加数值标签
|
|
for x, y in zip(batch_sizes, throughputs):
|
|
ax1.text(x, y + max(throughputs)*0.02, f'{y:.1f}',
|
|
ha='center', va='bottom', fontweight='bold')
|
|
|
|
# 找到最佳批次大小
|
|
best_idx = np.argmax(throughputs)
|
|
ax1.scatter([batch_sizes[best_idx]], [throughputs[best_idx]],
|
|
color='red', s=200, marker='*', zorder=5,
|
|
label=f'最佳: Batch {batch_sizes[best_idx]}')
|
|
ax1.legend()
|
|
|
|
# 延迟图表
|
|
ax2.plot(batch_sizes, latencies, 'o-', color='#FF6B6B',
|
|
linewidth=2, markersize=10, label='延迟')
|
|
ax2.set_title('批次大小 vs 延迟', fontsize=14, fontweight='bold')
|
|
ax2.set_xlabel('批次大小', fontsize=12)
|
|
ax2.set_ylabel('延迟 (ms)', fontsize=12)
|
|
ax2.grid(True, alpha=0.3)
|
|
ax2.legend()
|
|
|
|
# 添加数值标签
|
|
for x, y in zip(batch_sizes, latencies):
|
|
ax2.text(x, y + max(latencies)*0.02, f'{y:.1f}',
|
|
ha='center', va='bottom', fontweight='bold')
|
|
|
|
# 找到最低延迟
|
|
best_latency_idx = np.argmin(latencies)
|
|
ax2.scatter([batch_sizes[best_latency_idx]], [latencies[best_latency_idx]],
|
|
color='green', s=200, marker='*', zorder=5,
|
|
label=f'最低: Batch {batch_sizes[best_latency_idx]}')
|
|
ax2.legend()
|
|
|
|
plt.tight_layout()
|
|
plt.savefig(os.path.join(output_dir, 'batch_throughput_latency.png'),
|
|
dpi=300, bbox_inches='tight')
|
|
plt.show()
|
|
print(f"✅ 生成图表: batch_throughput_latency.png")
|
|
|
|
def create_gpu_utilization_chart(results, output_dir):
|
|
"""创建 GPU 利用率和内存使用图表"""
|
|
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))
|
|
|
|
# 提取数据
|
|
batch_sizes = []
|
|
gpu_utils = []
|
|
gpu_memories = []
|
|
|
|
for test in results['batch_tests']:
|
|
if test['success']:
|
|
batch_sizes.append(test['batch_size'])
|
|
gpu_utils.append(test['avg_gpu_util'])
|
|
gpu_memories.append(test['avg_gpu_memory_mb'])
|
|
|
|
if not batch_sizes:
|
|
return
|
|
|
|
# GPU 利用率图表
|
|
ax1.bar(batch_sizes, gpu_utils, color='#95E1D3', alpha=0.8, edgecolor='black')
|
|
ax1.set_title('批次大小 vs GPU 利用率', fontsize=14, fontweight='bold')
|
|
ax1.set_xlabel('批次大小', fontsize=12)
|
|
ax1.set_ylabel('GPU 利用率 (%)', fontsize=12)
|
|
ax1.grid(True, alpha=0.3, axis='y')
|
|
|
|
# 添加数值标签
|
|
for x, y in zip(batch_sizes, gpu_utils):
|
|
ax1.text(x, y + max(gpu_utils)*0.02, f'{y:.1f}%',
|
|
ha='center', va='bottom', fontweight='bold')
|
|
|
|
# GPU 内存使用图表
|
|
ax2.bar(batch_sizes, gpu_memories, color='#F38181', alpha=0.8, edgecolor='black')
|
|
ax2.set_title('批次大小 vs GPU 内存使用', fontsize=14, fontweight='bold')
|
|
ax2.set_xlabel('批次大小', fontsize=12)
|
|
ax2.set_ylabel('GPU 内存 (MB)', fontsize=12)
|
|
ax2.grid(True, alpha=0.3, axis='y')
|
|
|
|
# 添加数值标签
|
|
for x, y in zip(batch_sizes, gpu_memories):
|
|
ax2.text(x, y + max(gpu_memories)*0.02, f'{y:.0f}',
|
|
ha='center', va='bottom', fontweight='bold')
|
|
|
|
plt.tight_layout()
|
|
plt.savefig(os.path.join(output_dir, 'batch_gpu_metrics.png'),
|
|
dpi=300, bbox_inches='tight')
|
|
plt.show()
|
|
print(f"✅ 生成图表: batch_gpu_metrics.png")
|
|
|
|
def create_efficiency_chart(results, output_dir):
|
|
"""创建效率分析图表"""
|
|
fig, ax = plt.subplots(1, 1, figsize=(10, 6))
|
|
|
|
# 提取数据
|
|
batch_sizes = []
|
|
efficiencies = [] # FPS per GPU utilization
|
|
|
|
for test in results['batch_tests']:
|
|
if test['success'] and test['avg_gpu_util'] > 0:
|
|
batch_sizes.append(test['batch_size'])
|
|
efficiency = test['avg_throughput'] / test['avg_gpu_util']
|
|
efficiencies.append(efficiency)
|
|
|
|
if not batch_sizes:
|
|
return
|
|
|
|
# 效率图表
|
|
ax.plot(batch_sizes, efficiencies, 'o-', color='#AA96DA',
|
|
linewidth=2, markersize=10, label='效率 (FPS/GPU%)')
|
|
ax.set_title('批次大小 vs 性能效率', fontsize=14, fontweight='bold')
|
|
ax.set_xlabel('批次大小', fontsize=12)
|
|
ax.set_ylabel('效率 (FPS / GPU利用率%)', fontsize=12)
|
|
ax.grid(True, alpha=0.3)
|
|
ax.legend()
|
|
|
|
# 添加数值标签
|
|
for x, y in zip(batch_sizes, efficiencies):
|
|
ax.text(x, y + max(efficiencies)*0.02, f'{y:.2f}',
|
|
ha='center', va='bottom', fontweight='bold')
|
|
|
|
# 找到最高效率
|
|
best_idx = np.argmax(efficiencies)
|
|
ax.scatter([batch_sizes[best_idx]], [efficiencies[best_idx]],
|
|
color='gold', s=200, marker='*', zorder=5,
|
|
label=f'最高效率: Batch {batch_sizes[best_idx]}')
|
|
ax.legend()
|
|
|
|
plt.tight_layout()
|
|
plt.savefig(os.path.join(output_dir, 'batch_efficiency.png'),
|
|
dpi=300, bbox_inches='tight')
|
|
plt.show()
|
|
print(f"✅ 生成图表: batch_efficiency.png")
|
|
|
|
def create_comprehensive_table(results, output_dir):
|
|
"""创建综合性能对比表格"""
|
|
fig, ax = plt.subplots(figsize=(14, 8))
|
|
ax.axis('tight')
|
|
ax.axis('off')
|
|
|
|
# 准备表格数据
|
|
headers = ['批次大小', '吞吐量\n(FPS)', '延迟\n(ms)', 'GPU利用率\n(%)',
|
|
'GPU内存\n(MB)', '测试时长\n(s)', '总帧数']
|
|
|
|
table_data = []
|
|
for test in results['batch_tests']:
|
|
if test['success']:
|
|
row = [
|
|
test['batch_size'],
|
|
f"{test['avg_throughput']:.1f}",
|
|
f"{test['avg_latency_ms']:.1f}",
|
|
f"{test['avg_gpu_util']:.1f}",
|
|
f"{test['avg_gpu_memory_mb']:.0f}",
|
|
f"{test['test_duration']:.1f}",
|
|
test['total_frames']
|
|
]
|
|
table_data.append(row)
|
|
else:
|
|
row = [test['batch_size'], '失败', '-', '-', '-', '-', '-']
|
|
table_data.append(row)
|
|
|
|
# 创建表格
|
|
table = ax.table(cellText=table_data, colLabels=headers,
|
|
cellLoc='center', loc='center',
|
|
colWidths=[0.12, 0.15, 0.12, 0.15, 0.15, 0.15, 0.16])
|
|
|
|
table.auto_set_font_size(False)
|
|
table.set_fontsize(10)
|
|
table.scale(1, 2)
|
|
|
|
# 设置表头样式
|
|
for i in range(len(headers)):
|
|
table[(0, i)].set_facecolor('#4ECDC4')
|
|
table[(0, i)].set_text_props(weight='bold', color='white')
|
|
|
|
# 设置行颜色
|
|
for i in range(1, len(table_data) + 1):
|
|
for j in range(len(headers)):
|
|
if i % 2 == 0:
|
|
table[(i, j)].set_facecolor('#F0F0F0')
|
|
else:
|
|
table[(i, j)].set_facecolor('white')
|
|
|
|
plt.title('批次性能测试综合对比表', fontsize=16, fontweight='bold', pad=20)
|
|
plt.savefig(os.path.join(output_dir, 'batch_performance_table.png'),
|
|
dpi=300, bbox_inches='tight')
|
|
plt.show()
|
|
print(f"✅ 生成图表: batch_performance_table.png")
|
|
|
|
def generate_summary_report(results, output_dir):
|
|
"""生成总结报告"""
|
|
report = f"""
|
|
动态批次 TensorRT 性能测试总结报告
|
|
{'='*60}
|
|
|
|
测试时间: {results['timestamp']}
|
|
引擎路径: {results['engine_path']}
|
|
|
|
"""
|
|
|
|
if 'summary' in results and results['summary']:
|
|
summary = results['summary']
|
|
report += f"""测试概况:
|
|
{'='*60}
|
|
总测试数: {summary['total_tests']}
|
|
成功测试: {summary['successful_tests']}
|
|
失败测试: {summary['failed_tests']}
|
|
|
|
"""
|
|
|
|
if 'best_throughput' in summary:
|
|
report += f"""最佳吞吐量配置:
|
|
批次大小: {summary['best_throughput']['batch_size']}
|
|
吞吐量: {summary['best_throughput']['fps']:.1f} FPS
|
|
|
|
"""
|
|
|
|
if 'best_latency' in summary:
|
|
report += f"""最低延迟配置:
|
|
批次大小: {summary['best_latency']['batch_size']}
|
|
延迟: {summary['best_latency']['latency_ms']:.1f}ms
|
|
|
|
"""
|
|
|
|
report += f"""详细测试结果:
|
|
{'='*60}
|
|
"""
|
|
|
|
for test in results['batch_tests']:
|
|
if test['success']:
|
|
report += f"""
|
|
批次大小: {test['batch_size']}
|
|
吞吐量: {test['avg_throughput']:.1f} FPS
|
|
延迟: {test['avg_latency_ms']:.1f}ms
|
|
GPU 利用率: {test['avg_gpu_util']:.1f}%
|
|
GPU 内存: {test['avg_gpu_memory_mb']:.0f}MB (峰值: {test['max_gpu_memory_mb']:.0f}MB)
|
|
测试时长: {test['test_duration']:.1f}s
|
|
总帧数: {test['total_frames']}
|
|
"""
|
|
else:
|
|
report += f"""
|
|
批次大小: {test['batch_size']} - 测试失败
|
|
错误信息: {test.get('error_message', '未知错误')}
|
|
"""
|
|
|
|
report += f"""
|
|
|
|
推荐配置:
|
|
{'='*60}
|
|
"""
|
|
|
|
# 分析并给出推荐
|
|
successful_tests = [t for t in results['batch_tests'] if t['success']]
|
|
if successful_tests:
|
|
best_throughput = max(successful_tests, key=lambda x: x['avg_throughput'])
|
|
best_latency = min(successful_tests, key=lambda x: x['avg_latency_ms'])
|
|
|
|
report += f"""
|
|
✅ 追求最大吞吐量: 使用批次大小 {best_throughput['batch_size']} ({best_throughput['avg_throughput']:.1f} FPS)
|
|
✅ 追求最低延迟: 使用批次大小 {best_latency['batch_size']} ({best_latency['avg_latency_ms']:.1f}ms)
|
|
✅ 平衡性能与延迟: 建议使用批次大小 4-8
|
|
|
|
注意事项:
|
|
⚠️ 批次大小越大，吞吐量越高，但单帧延迟也会增加
|
|
⚠️ 实际部署时需要根据业务需求选择合适的批次大小
|
|
⚠️ GPU 内存占用随批次大小增加而增加，需要确保显存充足
|
|
"""
|
|
|
|
# 保存报告
|
|
report_file = os.path.join(output_dir, 'batch_performance_summary.txt')
|
|
with open(report_file, 'w', encoding='utf-8') as f:
|
|
f.write(report)
|
|
|
|
print(report)
|
|
print(f"\n📁 总结报告已保存: {report_file}")
|
|
|
|
def main():
|
|
"""主函数"""
|
|
# 查找最新的测试结果文件
|
|
results_dir = "batch_test_results"
|
|
if not os.path.exists(results_dir):
|
|
print("❌ 未找到测试结果目录")
|
|
print("请先运行 run_batch_performance_test.py")
|
|
return
|
|
|
|
json_files = [f for f in os.listdir(results_dir)
|
|
if f.startswith('batch_test_results_') and f.endswith('.json')]
|
|
if not json_files:
|
|
print("❌ 未找到测试结果文件")
|
|
return
|
|
|
|
# 使用最新的结果文件
|
|
latest_file = sorted(json_files)[-1]
|
|
json_path = os.path.join(results_dir, latest_file)
|
|
|
|
print(f"📊 加载测试结果: {json_path}")
|
|
results = load_results(json_path)
|
|
|
|
# 创建可视化输出目录
|
|
viz_dir = os.path.join(results_dir, "visualizations")
|
|
os.makedirs(viz_dir, exist_ok=True)
|
|
|
|
print("\n🎨 生成可视化图表...")
|
|
|
|
# 生成各种图表
|
|
create_throughput_chart(results, viz_dir)
|
|
create_gpu_utilization_chart(results, viz_dir)
|
|
create_efficiency_chart(results, viz_dir)
|
|
create_comprehensive_table(results, viz_dir)
|
|
|
|
# 生成总结报告
|
|
generate_summary_report(results, viz_dir)
|
|
|
|
print(f"\n✅ 所有可视化图表已生成完成！")
|
|
print(f"📁 输出目录: {viz_dir}")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|