185 lines
8.0 KiB
Python
185 lines
8.0 KiB
Python
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)
|