import json import argparse import pandas as pd import matplotlib.pyplot as plt from pathlib import Path def analyze_runs(data, max_lpips_thresh=None): if not data: print("错误：JSON文件中没有数据。") return None, None df = pd.DataFrame(data) df = df.dropna(subset=['coefficients']) if df.empty: print("错误：未找到包含'coefficients'的有效运行记录。") return None, None if max_lpips_thresh is not None and max_lpips_thresh > 0: if 'lpips_distance' in df.columns: print("\n" + "-"*15 + f" 应用质量门槛 (LPIPS <= {max_lpips_thresh}) " + "-"*15) original_count = len(df) df = df.dropna(subset=['lpips_distance']) df = df[df['lpips_distance'] <= max_lpips_thresh] filtered_count = len(df) print(f"已从 {original_count} 条记录中筛选出 {filtered_count} 条符合条件的记录。") if df.empty: print("警告：应用阈值后，没有符合条件的运行记录。") return None, None else: print("警告: 数据中不包含 'lpips_distance'，无法应用质量门槛。") df['hit_ratio'] = (df['cache_hits'] / df['total_inferences']).fillna(0) # 寻找基准和计算节省时间 (仅用于“速度-命中率”得分) baseline_runs = df[df['rel_l1_thresh'] == 0] if baseline_runs.empty: print("警告：未找到基准运行 (rel_l1_thresh == 0)，无法计算'节省时间'相关的得分。") baseline_time = df['generation_time'].max() else: baseline_time = baseline_runs['generation_time'].min() print(f"基准运行时间 (最快无缓存): {baseline_time:.2f} 秒") df['time_saved'] = baseline_time - df['generation_time'] df.loc[df['time_saved'] < 0, 'time_saved'] = 0 if 'lpips_distance' in df.columns: # 命中率 / LPIPS距离 df['score_lpips'] = (df['hit_ratio'] / df['lpips_distance'].replace(0, float('inf'))).fillna(0) else: print("警告: 数据中不包含 'lpips_distance'，跳过LPIPS相关分析。") df['score_lpips'] = 0 # “速度-命中率”的得分公式保持不变 df['score_hit_ratio'] = df['time_saved'] * df['hit_ratio'] # 找出各项最佳参数 results = {} cached_runs = df[df['rel_l1_thresh'] > 0] if cached_runs.empty: print("警告: 未找到任何开启缓存的运行记录，无法推荐最优参数。") return df, None best_indices = { "最快生成速度": cached_runs['generation_time'].idxmin(), "最高缓存命中率": cached_runs['hit_ratio'].idxmax(), "最佳速度-命中率综合效率": cached_runs['score_hit_ratio'].idxmax(), # 名字更新 } if df['score_lpips'].sum() > 0 and 'lpips_distance' in cached_runs.columns and cached_runs['lpips_distance'].notna().any(): best_indices["最低LPIPS (最佳画质)"] = cached_runs['lpips_distance'].idxmin() best_indices["最佳质量-命中率综合效率 (LPIPS)"] = cached_runs['score_lpips'].idxmax() # 名字更新 for name, idx in best_indices.items(): best_run = df.loc[idx] results[name] = { "coefficients": best_run['coefficients'], "value": { "生成时间": f"{best_run['generation_time']:.2f}s", "命中率": f"{best_run['hit_ratio']:.2%}", "LPIPS": f"{best_run.get('lpips_distance', 'N/A'):.4f}" if pd.notna(best_run.get('lpips_distance')) else "N/A", "速度-命中率得分": f"{best_run['score_hit_ratio']:.2f}", "质量-命中率得分(LPIPS)": f"{best_run['score_lpips']:.4f}", } } return df, results def print_results(results): if not results: return print("\n" + "="*25 + " 分析结果 " + "="*25) for name, data in results.items(): print(f"\n--- {name} ---") print(f" 最佳Coefficients: {data['coefficients']}") print(" 相关指标:") for key, val in data['value'].items(): print(f" - {key}: {val}") print("\n" + "="*62) def create_plots(df, results, max_lpips_thresh=None): if df is None or results is None: print("Insufficient data to generate plots.") return plot_df = df[df['rel_l1_thresh'] > 0].copy() if plot_df.empty: print("No cache-enabled run data available, cannot generate plots.") return # Plot 1: Time vs LPIPS if 'score_lpips' in plot_df.columns and plot_df['lpips_distance'].notna().any(): plt.style.use('seaborn-v0_8-darkgrid') fig1, ax1 = plt.subplots(figsize=(12, 8)) scatter1 = ax1.scatter(plot_df['generation_time'], plot_df['lpips_distance'], c=plot_df['score_lpips'], cmap='viridis', alpha=0.7, s=50) fig1.colorbar(scatter1, label='Quality-Hit Ratio Score (Hit Ratio / LPIPS)') title = 'Generation Speed vs Image Quality (LPIPS)' if max_lpips_thresh: title += f'\n(Filtering applied, LPIPS <= {max_lpips_thresh})' ax1.set_title(title, fontsize=16) ax1.set_xlabel('Generation Time (seconds) - Lower is Better', fontsize=12) ax1.set_ylabel('LPIPS Distance - Lower is Better', fontsize=12) best_lpips_score_idx = plot_df['score_lpips'].idxmax() best_point = plot_df.loc[best_lpips_score_idx] ax1.scatter(best_point['generation_time'], best_point['lpips_distance'], color='red', s=150, edgecolor='black', marker='*', label='Best Quality-Hit Ratio Point') ax1.text(best_point['generation_time'], best_point['lpips_distance'], ' Best Combined Point', color='red', ha='left') ax1.legend() ax1.grid(True) # Plot 2: Time vs Cache Hit Ratio fig2, ax2 = plt.subplots(figsize=(12, 8)) scatter2 = ax2.scatter(plot_df['generation_time'], plot_df['hit_ratio'], c=plot_df['score_hit_ratio'], cmap='plasma', alpha=0.7, s=50) fig2.colorbar(scatter2, label='Speed-Hit Ratio Score (Time Saved * Hit Ratio)') ax2.set_title('Generation Speed vs Cache Hit Rate', fontsize=16) ax2.set_xlabel('Generation Time (seconds) - Lower is Better', fontsize=12) ax2.set_ylabel('Cache Hit Rate - Higher is Better', fontsize=12) ax2.yaxis.set_major_formatter(plt.FuncFormatter(lambda y, _: '{:.0%}'.format(y))) best_hr_score_idx = plot_df['score_hit_ratio'].idxmax() best_point_hr = plot_df.loc[best_hr_score_idx] ax2.scatter(best_point_hr['generation_time'], best_point_hr['hit_ratio'], color='blue', s=150, edgecolor='black', marker='*', label='Best Speed-Hit Ratio Point') ax2.text(best_point_hr['generation_time'], best_point_hr['hit_ratio'], ' Best Combined Point', color='blue', ha='left') ax2.legend() ax2.grid(True) print("\nGenerating visualization charts. Please check the pop-up windows. The program will end after you close the chart windows.") plt.tight_layout() plt.show() if __name__ == "__main__": parser = argparse.ArgumentParser(description="分析 TeaCache 的 JSON 输出文件，并找出最优参数。") parser.add_argument("json_file", default="teacache_analysis.json", nargs='?', type=Path, help="指向 teacache_analysis.json 文件的路径。") parser.add_argument( "--max_lpips", type=float, default=0.56, help="设置可接受的最大LPIPS距离阈值，用于过滤低质量数据。例如: --max_lpips 0.6" ) args = parser.parse_args() json_path = args.json_file if not json_path.is_file(): print(f"错误: 文件不存在 -> {json_path}") else: print(f"正在读取文件: {json_path}") with open(json_path, 'r', encoding='utf-8') as f: analysis_data = json.load(f) df, best_results = analyze_runs(analysis_data, args.max_lpips) print_results(best_results) create_plots(df, best_results, args.max_lpips)