import json import argparse import matplotlib.pyplot as plt import numpy as np from datetime import datetime from matplotlib.font_manager import FontProperties # 设置中文字体支持 plt.rcParams['font.sans-serif'] = ['SimHei', 'FangSong', 'Arial Unicode MS'] plt.rcParams['axes.unicode_minus'] = False def read_json(file_path): with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) return data def plot_cprofile_table(ax, cprofile_data): """绘制cProfile详细信息表格""" if not cprofile_data or 'points' not in cprofile_data: ax.text(0.5, 0.5, '无cProfile数据', ha='center', va='center', transform=ax.transAxes) ax.set_title('函数运行详细信息') return # 只取前15个最耗时的函数 fun_num = 15 points = cprofile_data['points'][:fun_num] # 准备表格数据 table_data = [] for point in points: # 简化函数名显示 func_name = point['function'][1:-1] func_name = func_name.split(',')[-1].strip() func_name = func_name[1:-1] # 限制函数名长度，但保留足够空间显示完整信息 func_name_display = func_name[:35] + '...' if len(func_name) > 35 else func_name # 将数据点添加到表格 table_data.append([ func_name_display, point['call_count'], f"{point['total_time']:.6f}", f"{point['cumulative_time']:.6f}" ]) # 创建表格 columns = ['函数名称', '调用次数', '总时间(s)', '累计时间(s)'] table = ax.table(cellText=table_data, colLabels=columns, cellLoc='center', loc='center') table.auto_set_font_size(False) table.set_fontsize(9) table.scale(1, 1.5) # 设置表格样式 for i in range(len(columns)): table[(0, i)].set_facecolor('#4CAF50') table[(0, i)].set_text_props(weight='bold', color='white') # 根据函数数量动态调整标题距离 # 每个函数行大约需要8个单位的高度，加上基础padding pad_value = 8 + len(points) * 3 ax.set_title(f'函数运行详细信息 (Top {fun_num} 函数)', fontsize=14, pad=pad_value) # 坐标轴掩藏 ax.set_xticks([]) ax.set_yticks([]) ax.spines['top'].set_visible(False) ax.spines['right'].set_visible(False) ax.spines['bottom'].set_visible(False) ax.spines['left'].set_visible(False) def plot_cprofile_bar_chart(ax, cprofile_data): """绘制cProfile总时间柱状图""" if not cprofile_data or 'points' not in cprofile_data: ax.text(0.5, 0.5, '无cProfile数据', ha='center', va='center', transform=ax.transAxes) ax.set_title('函数运行总时间') return # 只取前N个最耗时的函数 fun_num = 15 points = cprofile_data['points'][:fun_num] # 提取函数名和总时间 functions = [] total_times = [] for point in points: # 简化函数名显示 func_name = point['function'][1:-1] func_name = func_name.split(',')[-1].strip() func_name = func_name[1:-1] # 限制函数名长度 func_name_display = func_name[:35] + '...' if len(func_name) > 35 else func_name functions.append(func_name_display) total_times.append(point['total_time']) # 创建垂直柱状图，使用渐变色彩 x_pos = np.arange(len(functions)) colors = plt.cm.viridis(np.linspace(0, 1, len(functions))) bars = ax.bar(x_pos, total_times, color=colors) # 设置坐标轴 ax.set_xticks(x_pos) ax.set_xticklabels(functions, rotation=30, ha='right', fontsize=9) ax.set_ylabel('总时间 (秒)', fontsize=12) ax.set_title(f'函数运行总时间 (Top {fun_num} 函数)', fontsize=14, pad=20) # 改善网格线样式 ax.grid(axis='y', alpha=0.7, linestyle='--', linewidth=0.5) ax.set_axisbelow(True) # 在柱状图上添加数值标签，优化显示效果 for bar, time in zip(bars, total_times): height = bar.get_height() ax.text(bar.get_x() + bar.get_width()/2., height, f'{time:.6f}', ha='center', va='bottom', fontsize=8, rotation=0, weight='bold') # 添加边框 for spine in ax.spines.values(): spine.set_linewidth(0.5) def plot_memory_data(ax, memory_data): """绘制内存使用数据""" if not memory_data or 'points' not in memory_data: ax.text(0.5, 0.5, '无内存数据', ha='center', va='center', transform=ax.transAxes) ax.set_title('memory_profiler 内存使用情况') return timestamps = [point['timestamp'] for point in memory_data['points']] memory_values = [point['memory'] for point in memory_data['points']] # 转换时间戳为相对时间（秒） start_time = timestamps[0] if timestamps else 0 relative_times = [t - start_time for t in timestamps] # 查找内存峰值及其位置 max_memory = max(memory_values) if memory_values else 0 max_index = memory_values.index(max_memory) if memory_values else 0 max_time = relative_times[max_index] if relative_times else 0 # 绘制内存使用情况曲线 line, = ax.plot(relative_times, memory_values, marker='o', markersize=3, linewidth=1.5) # 在峰值点添加特殊标记 ax.plot(max_time, max_memory, marker='o', markersize=8, color='red', markerfacecolor='none', markeredgewidth=2, label=f'峰值: {max_memory:.2f} MB') ax.set_xlabel('时间 (秒)') ax.set_ylabel('内存使用量 (MB)') ax.set_title(f'进程内存使用情况 (峰值: {max_memory:.2f} MB at {max_time:.1f}s)') ax.grid(True, alpha=0.3) ax.legend() def plot_psutil_data(ax, psutil_data): """绘制psutil系统资源数据""" if not psutil_data or 'points' not in psutil_data: ax.text(0.5, 0.5, '无进程资源数据', ha='center', va='center', transform=ax.transAxes) ax.set_title('进程资源使用情况') return points = psutil_data['points'] if not points: ax.text(0.5, 0.5, '无进程资源数据', ha='center', va='center', transform=ax.transAxes) ax.set_title('进程资源使用情况') return # 提取数据并转换时间字符串为时间戳 timestamps = [] rss_values = [] vms_values = [] cpu_values = [] mem_pct_values = [] for point in points: # 解析时间字符串并转换为时间戳 try: dt = datetime.strptime(point['now_time'], '%Y-%m-%d %H:%M:%S') timestamps.append(dt.timestamp()) except ValueError: # 如果解析失败，使用索引作为后备 timestamps.append(len(timestamps)) # 转换时间戳为相对时间（秒） start_time = timestamps[0] if timestamps else 0 relative_times = [t - start_time for t in timestamps] for point in points: # 解析RSS内存值 rss_str = point['rss'].replace('MB', '') try: rss_values.append(float(rss_str)) except ValueError: rss_values.append(0) # 解析VMS内存值 vms_str = point['vms'].replace('MB', '') try: vms_values.append(float(vms_str)) except ValueError: vms_values.append(0) # 解析CPU百分比 cpu_str = point['cpu_pct'].replace('%', '') try: cpu_values.append(float(cpu_str)) except ValueError: cpu_values.append(0) # 解析内存占用率 mem_pct_str = point['mem_pct'].replace('%', '') try: mem_pct_values.append(float(mem_pct_str)) except ValueError: mem_pct_values.append(0) # 绘制双轴图 ax2 = ax.twinx() # 物理内存使用量（蓝色） line1, = ax.plot(relative_times, rss_values, marker='o', color='blue', label='物理内存(RSS)') # 虚拟内存使用量（绿色） line3, = ax.plot(relative_times, vms_values, marker='^', color='green', label='虚拟内存(VMS)') ax.set_xlabel('时间 (秒)') ax.set_ylabel('内存使用量 (MB)', color='blue') ax.tick_params(axis='y', labelcolor='blue') # CPU使用率（红色）和内存占用率（紫色） line2, = ax2.plot(relative_times, cpu_values, marker='s', color='red', label='CPU使用率') line4, = ax2.plot(relative_times, mem_pct_values, marker='d', color='purple', label='内存占用率') ax2.set_ylabel('百分比 (%)', color='red') ax2.tick_params(axis='y', labelcolor='red') # 合并图例 lines = [line1, line3, line2, line4] labels = [l.get_label() for l in lines] ax.legend(lines, labels, loc='upper left') ax.set_title('进程资源使用情况') ax.grid(True, alpha=0.3) def plot_network_data(ax, psutil_data): """绘制网络使用数据""" if not psutil_data or 'points' not in psutil_data: ax.text(0.5, 0.5, '无网络数据', ha='center', va='center', transform=ax.transAxes) ax.set_title('系统网络使用情况') return points = psutil_data['points'] if not points: ax.text(0.5, 0.5, '无网络数据', ha='center', va='center', transform=ax.transAxes) ax.set_title('系统网络使用情况') return # 提取数据并转换时间字符串为时间戳 from datetime import datetime timestamps = [] net_recv_values = [] net_send_values = [] for point in points: # 解析时间字符串并转换为时间戳 try: dt = datetime.strptime(point['now_time'], '%Y-%m-%d %H:%M:%S') timestamps.append(dt.timestamp()) except ValueError: # 如果解析失败，使用索引作为后备 timestamps.append(len(timestamps)) # 转换为相对时间（秒） start_time = timestamps[0] if timestamps else 0 relative_times = [t - start_time for t in timestamps] for point in points: # 解析网络接收值 net_recv_str = point['sys_net_recv'].replace('MB/s', '').replace('MB', '').replace(' ', '') try: net_recv_values.append(float(net_recv_str)) except ValueError: net_recv_values.append(0) # 解析网络发送值 net_send_str = point['sys_net_send'].replace('MB/s', '').replace('MB', '').replace(' ', '') try: net_send_values.append(float(net_send_str)) except ValueError: net_send_values.append(0) # 绘制网络使用情况 ax.plot(relative_times, net_recv_values, marker='o', color='blue', label='网络接收') ax.plot(relative_times, net_send_values, marker='s', color='red', label='网络发送') ax.set_xlabel('时间 (秒)') ax.set_ylabel('速率 (MB/s)') ax.set_title('系统网络使用情况') ax.grid(True, alpha=0.3) ax.legend() def plot_disk_io_data(ax, psutil_data): """绘制进程IO数据""" if not psutil_data or 'points' not in psutil_data: ax.text(0.5, 0.5, '无进程IO数据', ha='center', va='center', transform=ax.transAxes) ax.set_title('进程IO情况') return points = psutil_data['points'] if not points: ax.text(0.5, 0.5, '无进程IO数据', ha='center', va='center', transform=ax.transAxes) ax.set_title('进程IO情况') return # 提取数据并转换时间字符串为时间戳 timestamps = [] disk_read_values = [] disk_write_values = [] for point in points: # 解析时间字符串并转换为时间戳 try: dt = datetime.strptime(point['now_time'], '%Y-%m-%d %H:%M:%S') timestamps.append(dt.timestamp()) except ValueError: # 如果解析失败，使用索引作为后备 timestamps.append(len(timestamps)) # 转换为相对时间（秒） start_time = timestamps[0] if timestamps else 0 relative_times = [t - start_time for t in timestamps] for point in points: # 解析磁盘读取值 disk_read_str = point['disk_read'].replace('MB/s', '').replace('MB', '').replace(' ', '') try: disk_read_values.append(float(disk_read_str)) except ValueError: disk_read_values.append(0) # 解析磁盘写入值 disk_write_str = point['disk_write'].replace('MB/s', '').replace('MB', '').replace(' ', '') try: disk_write_values.append(float(disk_write_str)) except ValueError: disk_write_values.append(0) # 绘制磁盘IO情况 ax.plot(relative_times, disk_read_values, marker='o', color='blue', label='磁盘读取') ax.plot(relative_times, disk_write_values, marker='s', color='red', label='磁盘写入') ax.set_xlabel('时间 (秒)') ax.set_ylabel('速率 (MB/s)') ax.set_title('进程IO情况') ax.grid(True, alpha=0.3) ax.legend() def main(): parser = argparse.ArgumentParser(description='绘制性能分析结果图') parser.add_argument('--input','-i', default='./results/performance_analysis_report.json') parser.add_argument('--output', '-o', default='./results/performance_analysis_report.png') args = parser.parse_args() # 读取json数据 data = read_json(args.input) # 创建更大的图表 fig = plt.figure(figsize=(20, 18)) fig.suptitle('Python 性能分析报告', fontsize=20, fontweight='bold') # 创建子图 # cProfile 表格 ax1 = plt.subplot2grid((4, 2), (0, 0), colspan=1) # cProfile 柱状图 ax2 = plt.subplot2grid((4, 2), (0, 1), colspan=1) # 内存使用情况 ax3 = plt.subplot2grid((4, 2), (1, 0), colspan=1) # 系统资源监控 ax4 = plt.subplot2grid((4, 2), (1, 1), colspan=1) # 网络使用情况 ax5 = plt.subplot2grid((4, 2), (2, 0), colspan=1) # 磁盘读写情况 ax6 = plt.subplot2grid((4, 2), (2, 1), colspan=1) # 绘制各个子图 plot_cprofile_table(ax1, data.get('analysis_results', {}).get('cprofile', {})) plot_cprofile_bar_chart(ax2, data.get('analysis_results', {}).get('cprofile', {})) plot_memory_data(ax3, data.get('analysis_results', {}).get('memory_profile', {})) plot_psutil_data(ax4, data.get('analysis_results', {}).get('psutil', {})) plot_network_data(ax5, data.get('analysis_results', {}).get('psutil', {})) plot_disk_io_data(ax6, data.get('analysis_results', {}).get('psutil', {})) # 调整布局 plt.tight_layout() # 保存图像 plt.savefig(args.output, dpi=300, bbox_inches='tight') print(f"性能分析图表已保存至: {args.output}") if __name__ == '__main__': main()