diff --git a/analyze_teacache.py b/analyze_teacache.py new file mode 100644 index 0000000..1597e42 --- /dev/null +++ b/analyze_teacache.py @@ -0,0 +1,200 @@ +try: + import json + import argparse + import pandas as pd + import matplotlib.pyplot as plt + from pathlib import Path + + def analyze_runs(data): + if not data: + print("错误：JSON文件中没有数据。") + return None, None + + # 预处理 + df = pd.DataFrame(data) + + # 过滤掉没有 'coefficients' 的无效运行记录 + df = df.dropna(subset=['coefficients']) + if df.empty: + print("错误：未找到包含'coefficients'的有效运行记录。") + return None, None + + # 计算缓存命中率 + # 为了避免除以0的错误，当 total_inferences 为0时，命中率也为0 + df['hit_ratio'] = (df['cache_hits'] / df['total_inferences']).fillna(0) + + # 2. 寻找基准和计算节省时间 + # 基准运行指的是不开启缓存的运行 (rel_l1_thresh == 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 + + # 3. 计算两种综合效率得分 + # 得分越高越好 + + # LPIPS得分 = 节省的时间 / LPIPS距离 (LPIPS越小越好) + if 'lpips_distance' in df.columns: + # 避免除以0或空值 + df['score_lpips'] = (df['time_saved'] / 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'] + + # 4. 找出各项最佳参数 + 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: + best_indices["最低LPIPS (最佳画质)"] = cached_runs['lpips_distance'].idxmin() + best_indices["最佳综合效率 (LPIPS)"] = cached_runs['score_lpips'].idxmax() + + # 整理结果 + for name, idx in best_indices.items(): + # .loc[idx] 可以获取索引对应的整行数据 + 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']:.2f}", + } + } + + 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): + """ + 使用matplotlib创建数据可视化图表。 + """ + if df is None or results is None: + print("数据不足，无法生成图表。") + return + + # 过滤掉无缓存的基准点，让图表更清晰 + plot_df = df[df['rel_l1_thresh'] > 0].copy() + if plot_df.empty: + print("无缓存运行数据，无法生成图表。") + return + + # 图表1: 时间 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 # s是点的大小 + ) + fig1.colorbar(scatter1, label='综合效率得分 (LPIPS Score)') + + ax1.set_title('生成速度 vs 图像质量 (LPIPS)', fontsize=16) + ax1.set_xlabel('生成时间 (秒) - 越低越好', fontsize=12) + ax1.set_ylabel('LPIPS 距离 - 越低越好', 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, ec='black', marker='*', label='最佳LPIPS综合效率') + ax1.text(best_point['generation_time'], best_point['lpips_distance'], ' 最佳综合点', color='red', ha='left') + + ax1.legend() + ax1.grid(True) + + # 图表2: 时间 vs 缓存命中率 (速度与效率的权衡) + 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='综合效率得分 (Hit Ratio Score)') + + ax2.set_title('生成速度 vs 缓存命中率', fontsize=16) + ax2.set_xlabel('生成时间 (秒) - 越低越好', fontsize=12) + ax2.set_ylabel('缓存命中率 - 越高越好', fontsize=12) + ax2.yaxis.set_major_formatter(plt.FuncFormatter(lambda y, _: '{:.0%}'.format(y))) # 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, ec='black', marker='*', label='最佳命中率综合效率') + ax2.text(best_point_hr['generation_time'], best_point_hr['hit_ratio'], ' 最佳综合点', color='blue', ha='left') + + ax2.legend() + ax2.grid(True) + + print("\n正在生成可视化图表，请在弹出的窗口中查看。关闭图表窗口后程序将结束。") + plt.tight_layout() + plt.show() + + + if __name__ == "__main__": + parser = argparse.ArgumentParser(description="分析 TeaCache 的 JSON 输出文件，并找出最优参数。") + parser.add_argument("json_file", type=Path, help="指向 teacache_analysis.json 文件的路径。") + 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) + print_results(best_results) + create_plots(df, best_results) +except: + pass