Files

1147 lines
48 KiB
Python

# Copyright (c) 2025 Fudan University and StepFun Co., Ltd. All rights reserved.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
CIVIL Benchmark Evaluation Tool
-------------------------------
Evaluates generated images against reference images using multiple metrics:
- ArcFace similarity for facial recognition
- CLIP image and text similarity
- Aesthetic scoring
- FID (Fréchet Inception Distance)
"""
import os
import json
import time
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Tuple, Union
import numpy as np
from tqdm import tqdm
from colorama import init, Fore, Style
# Import metrics
from metrics.arcface_metric import ArcFace_Metrics
from metrics.clip_metric import CLIP_Metric
from metrics.aesthetic_metric import Aesthetic_Metric
# from metrics.anatomy_metric import Anatomy_Metric
from metrics.fid_metric import FID_Metric
from PIL import Image
# Initialize colorama for cross-platform colored terminal output
init()
def pad_image(img: Image.Image, ratio: float = 1.5) -> Image.Image:
"""
Pad an image to a square with black background.
Args:
img: PIL Image to be padded
ratio: Padding ratio relative to the longer side
Returns:
PIL Image: Padded square image
"""
w, h = img.size
# Find the longer side
longer_side = max(h, w)
# Calculate new size (ratio times the longer side)
new_size = int(longer_side * ratio)
# Create a new black square image
padded_img = Image.new('RGB', (new_size, new_size), (0, 0, 0))
# Calculate offsets to center the original image
x_offset = (new_size - w) // 2
y_offset = (new_size - h) // 2
# Paste the original image onto the new padded image
padded_img.paste(img, (x_offset, y_offset))
return padded_img
class BenchEval_Geo:
"""
Benchmark evaluation tool for celebrity portrait datasets.
"""
def __init__(
self,
target_dir: str,
output_dir: str = "./output/celeb_evaluation",
ori_file_name: str = "ori.jpg",
output_file_name: str = "output_0.jpg",
ref_1_file_name: str = "ref_1.jpg",
ref_2_file_name: str = "ref_2.jpg",
ref_3_file_name: str = None,
ref_4_file_name: str = None,
meta_file_name: str = "meta.json",
caption_keyword: str = "caption_en",
names_keyword: str = "name",
padding: bool = True,
save_individual_results: bool = True,
):
"""
Initialize the benchmark evaluation.
Args:
target_dir: Directory containing image folders to evaluate
output_dir: Directory to save individual evaluation results
ori_file_name: Filename for original images, set to None to skip original image comparison
output_file_name: Filename for generated output images
ref_1_file_name: Filename for first reference image
ref_2_file_name: Filename for second reference image
ref_3_file_name: Filename for third reference image, set to None to skip
ref_4_file_name: Filename for fourth reference image, set to None to skip
meta_file_name: Filename for metadata JSON
caption_keyword: Key in metadata to extract caption text
names_keyword: Key in metadata to extract celebrity names, set to None to skip name-based tests
padding: Whether to pad images to squares
save_individual_results: Whether to save per-image results
"""
# Initialize metrics
self._init_metrics()
# Set file paths and config
self._init_paths(target_dir, output_dir)
self._init_config(
ori_file_name, output_file_name,
ref_1_file_name, ref_2_file_name, ref_3_file_name, ref_4_file_name,
meta_file_name, caption_keyword, names_keyword, padding, save_individual_results
)
# Initialize result storage and collectors
self._init_results()
def _init_metrics(self):
"""Initialize all metric calculation components"""
self.metrics = ArcFace_Metrics()
self.clip_metric = CLIP_Metric()
self.aesthetic_metrics = Aesthetic_Metric()
# self.anatomy_metrics = Anatomy_Metric()
self.fid_metrics = FID_Metric()
def _init_paths(self, target_dir, output_dir):
"""Initialize paths and create output directory if needed"""
self.target_dir = target_dir
self.output_dir = output_dir
self.list_dir = [d for d in os.listdir(target_dir) if os.path.isdir(os.path.join(target_dir, d))]
def _init_config(self,
ori_file_name, output_file_name,
ref_1_file_name, ref_2_file_name, ref_3_file_name, ref_4_file_name,
meta_file_name, caption_keyword, names_keyword, padding, save_individual_results):
"""Initialize configuration settings"""
# Create output directory if needed
if save_individual_results and not os.path.exists(self.output_dir):
os.makedirs(self.output_dir)
# Original and output image config
self.ori_file_name = ori_file_name
self.ori_flag = self._is_valid_filename(ori_file_name)
self.output_file_name = output_file_name
# Reference image config
self.ref_1_file_name = ref_1_file_name
self.ref_2_file_name = ref_2_file_name
self.ref_3_file_name = ref_3_file_name
self.ref_4_file_name = ref_4_file_name
# Reference image flags
self.ref_1_flag = self._is_valid_filename(ref_1_file_name)
self.ref_2_flag = self._is_valid_filename(ref_2_file_name)
self.ref_3_flag = self._is_valid_filename(ref_3_file_name)
self.ref_4_flag = self._is_valid_filename(ref_4_file_name)
self.meta_file_name = meta_file_name
# Set options
self.caption_keyword = caption_keyword
self.names_keyword = names_keyword
self.name_flag = self._is_valid_filename(names_keyword)
self.padding = padding
self.save_individual_results = save_individual_results
def _is_valid_filename(self, filename):
"""Check if a filename is valid (not None, empty, or 'None')"""
return filename is not None and filename != "" and filename != "None"
def _init_results(self):
"""Initialize the results structure and metric collectors"""
self.results = {
"samples": [],
"summary": {},
"timestamp": datetime.now().isoformat()
}
# Initialize metric collectors based on flags
self.collectors = MetricCollectors(
use_original=self.ori_flag,
use_names=self.name_flag
)
def _load_image(self, path: str, warn_if_fail=True) -> Optional[Image.Image]:
"""Safe image loading with error handling"""
try:
return Image.open(path)
except Exception as e:
if warn_if_fail:
print(f"{Fore.RED}Error loading image {path}: {e}{Style.RESET_ALL}")
return None
def _load_metadata(self, path: str) -> Optional[Dict]:
"""Safe metadata loading with error handling"""
try:
with open(path, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
print(f"{Fore.RED}Error loading metadata {path}: {e}{Style.RESET_ALL}")
return None
def _save_individual_result(self, output_img: Image.Image, meta_id: str, metrics_data: Dict):
"""Save individual image result and metadata"""
try:
# Save image
output_img.save(os.path.join(self.output_dir, f"{meta_id}.jpg"))
# Save metrics as JSON
with open(os.path.join(self.output_dir, f"{meta_id}.json"), "w") as f:
json.dump(metrics_data, f, indent=4)
except Exception as e:
print(f"{Fore.RED}Error saving individual result for {meta_id}: {e}{Style.RESET_ALL}")
def _load_reference_images(self, dir_path):
"""Load available reference images based on configuration flags"""
refs = []
ref_count = 0
# Process each possible reference image
ref_configs = [
(self.ref_1_flag, self.ref_1_file_name, True), # Required ref
(self.ref_2_flag, self.ref_2_file_name, False), # Optional refs
(self.ref_3_flag, self.ref_3_file_name, False),
(self.ref_4_flag, self.ref_4_file_name, False)
]
for flag, filename, is_required in ref_configs:
if not flag:
continue
ref_img = self._load_image(
os.path.join(dir_path, filename),
warn_if_fail=is_required
)
if ref_img:
if self.padding:
ref_img = pad_image(ref_img)
refs.append(ref_img)
ref_count += 1
return refs, ref_count
def _calculate_metrics_for_sample(self, output_img, ori_img, refs, caption, names, dir_idx, directory):
"""Calculate all metrics for a sample and return the results"""
face_models = ["arcface", "facenet", "adaface"]
face_metrics = {}
# Calculate metrics for each face model
for model in face_models:
face_metrics[model] = self._calculate_face_metrics(
model, output_img, ori_img, refs, names
)
# Calculate other metrics
clip_metrics = self._calculate_clip_metrics(output_img, ori_img, caption)
aesthetic_score = self.aesthetic_metrics(output_img)
# anatomical_score = self.anatomy_metrics(output_img, return_scores=True)[0]
anatomical_score = 0.0 # Placeholder since anatomy metric is commented out
# Validate metric calculation success
if all(face_metrics[model]["sim_ref"] is None for model in face_models):
print(f"{Fore.RED}Face similarity calculation failed for {directory}{Style.RESET_ALL}")
return None
# Log sample metrics periodically
self._log_sample_metrics(
dir_idx, directory, face_metrics, clip_metrics,
aesthetic_score, anatomical_score, len(refs)
)
return {
"face_metrics": face_metrics,
"clip_i": clip_metrics["clip_i"],
"clip_t": clip_metrics["clip_t"],
"aesthetic_score": aesthetic_score,
"anatomical_score": anatomical_score
}
def _calculate_face_metrics(self, model, output_img, ori_img, refs, names):
"""Calculate face metrics for a specific model"""
metrics = {}
# Original image similarity if available
if self.ori_flag:
_, sim_ori, non_matched_mean_ori = self.metrics.compare_faces_with_confusion_matrix(
output_img, ori_img, model=model
)
metrics["sim_ori"] = sim_ori
# Only test unmatched faces when ref_2 is available
if self.ref_2_flag:
metrics["non_matched_ori"] = non_matched_mean_ori
else:
metrics["non_matched_ori"] = 0.0
# Calculate copy_geo metric when original image is available
try:
copy_geo_scores = self.metrics.calculate_copy_geo_multi_person(
output_img, ori_img, refs, model=model
)
metrics["copy_geo"] = copy_geo_scores
except Exception as e:
print(f"{Fore.YELLOW}Warning: Failed to calculate Copy-Geo using {model}: {str(e)}{Style.RESET_ALL}")
metrics["copy_geo"] = None
else:
metrics["sim_ori"] = None
metrics["non_matched_ori"] = None
metrics["copy_geo"] = None
# Reference similarity metrics
_, sim_ref, non_matched_mean_ref = self.metrics.compare_faces_with_confusion_matrix_with_ref(
output_img, refs, model=model
)
metrics["sim_ref"] = sim_ref
if self.ref_2_flag:
metrics["non_matched_ref"] = non_matched_mean_ref
else:
metrics["non_matched_ref"] = 0.0
# Cluster similarity metrics if names are available
if self.name_flag:
# _, sim_cluster, non_matched_mean_cluster = self.metrics.compare_faces_with_confusion_matrix_with_ref_names(
# output_img, names, model=model
# )
sim_cluster = 0.0 # Placeholder since name-based metric is commented out
metrics["sim_cluster"] = sim_cluster
else:
metrics["sim_cluster"] = None
return metrics
def _calculate_clip_metrics(self, output_img, ori_img, caption):
"""Calculate CLIP similarity metrics"""
clip_i = None
if self.ori_flag:
clip_i = self.clip_metric.compute_clip_i(ori_img, output_img)
clip_t = self.clip_metric.compute_clip_t(output_img, caption)
return {
"clip_i": clip_i,
"clip_t": clip_t
}
def _log_sample_metrics(self, dir_idx, directory, face_metrics, clip_metrics,
aesthetic_score, anatomical_score, ref_count):
"""Log metrics for a sample at periodic intervals"""
# Only log occasionally to avoid console spam
if dir_idx % max(1, len(self.list_dir) // 10) != 0:
return
print(f"\n{Fore.CYAN}Sample: {directory}{Style.RESET_ALL}")
print(f"{Fore.YELLOW}Using {ref_count} reference images for this sample{Style.RESET_ALL}")
for model in face_metrics:
model_metrics = face_metrics[model]
# Print face similarity with original if available
if self.ori_flag and model_metrics["sim_ori"] is not None:
print(f" {model.upper()} Face similarity (original): "
f"{model_metrics['sim_ori'].mean():.4f}")
# Print face similarity with reference
if model_metrics["sim_ref"] is not None:
print(f" {model.upper()} Face similarity (reference): "
f"{model_metrics['sim_ref'].mean():.4f}")
# Print cluster similarity if available
if self.name_flag and model_metrics["sim_cluster"] is not None:
print(f" {model.upper()} Cluster similarity: "
f"{model_metrics['sim_cluster'].mean():.4f}")
# Print CLIP and other scores
clip_i, clip_t = clip_metrics.get("clip_i"), clip_metrics.get("clip_t")
if self.ori_flag:
print(f" CLIP image: {clip_i:.4f}", end="")
print(f", CLIP text: {clip_t:.4f}")
print(f" Aesthetic score: {aesthetic_score:.4f}")
print(f" Anatomy score: {float(anatomical_score):.4f}\n")
def _collect_sample_metrics(self, all_metrics, metadata, ref_count, output_img):
"""Process metrics from a sample for collection and result storage"""
face_metrics = all_metrics["face_metrics"]
# Collect metrics for each model
for model in face_metrics:
model_metrics = face_metrics[model]
self.collectors.add_model_metrics(model, model_metrics, self.ori_flag, self.name_flag)
# Collect model-independent metrics
self.collectors.add_general_metrics(
clip_i=all_metrics["clip_i"] if self.ori_flag else None,
clip_t=all_metrics["clip_t"],
aesthetic_score=all_metrics["aesthetic_score"],
anatomy_score=all_metrics["anatomical_score"]
)
# Create sample data for results
sample_data = self._create_sample_data(
metadata, all_metrics, ref_count
)
# Save individual result if enabled
if self.save_individual_results:
self._save_individual_result(
output_img,
metadata.get("id", os.path.basename(os.path.normpath(metadata.get("path", "")))),
sample_data
)
# Add to results
self.results["samples"].append(sample_data)
return sample_data
def _create_sample_data(self, metadata, all_metrics, ref_count):
"""Create a dictionary of sample data for results"""
face_metrics = all_metrics["face_metrics"]
# Basic sample data
sample_id = metadata.get("id", os.path.basename(os.path.normpath(metadata.get("path", ""))))
sample_data = {
"id": sample_id,
"anatomy_score": float(all_metrics["anatomical_score"]),
"caption": metadata.get(self.caption_keyword, ""),
"clip_t": float(all_metrics["clip_t"]),
"aes_score": float(all_metrics["aesthetic_score"]),
"ref_count": ref_count
}
# Add conditional metrics
if self.ori_flag:
sample_data["clip_i"] = float(all_metrics["clip_i"])
# Add model-specific metrics
for model in face_metrics:
model_metrics = face_metrics[model]
# Reference similarity
if model_metrics["sim_ref"] is not None:
sample_data[f"{model}_sim_ref"] = float(model_metrics["sim_ref"].mean())
# Original similarity if available
if self.ori_flag and model_metrics["sim_ori"] is not None:
sample_data[f"{model}_sim_ori"] = float(model_metrics["sim_ori"].mean())
# Cluster similarity if available
if self.name_flag and model_metrics["sim_cluster"] is not None:
sample_data[f"{model}_sim_cluster"] = float(model_metrics["sim_cluster"].mean())
# Copy-geo metrics if available
if self.ori_flag and model_metrics["copy_geo"] is not None:
sample_data[f"{model}_copy_geo"] = float(model_metrics["copy_geo"].mean())
return sample_data
def _calculate_fid(self, original_images, generated_images):
"""Calculate FID score between original and generated images"""
if not self.ori_flag:
return None
print(f"\n{Fore.CYAN}Calculating FID score...{Style.RESET_ALL}")
try:
if len(original_images) > 1 and len(generated_images) > 1:
fid_score = self.fid_metrics.calculate_fid(original_images, generated_images)
print(f"{Fore.GREEN}FID Score: {fid_score:.4f}{Style.RESET_ALL}")
return float(fid_score)
else:
print(f"{Fore.YELLOW}Not enough samples to calculate FID score{Style.RESET_ALL}")
return None
except Exception as e:
print(f"{Fore.RED}Error calculating FID: {str(e)}{Style.RESET_ALL}")
return None
def _compute_statistics(self, values, remove_outliers=True, method="iqr", threshold=1.5, z_threshold=3.0):
"""Calculate comprehensive statistics for a list of values with outlier removal"""
if not values:
return None
values_array = np.array(values)
original_count = len(values_array)
# Remove outliers if requested
if remove_outliers:
filtered_values, outliers_removed = self._remove_outliers(
values_array, method, threshold, z_threshold
)
# Use filtered values for statistics
if len(filtered_values) < original_count * 0.5:
# If we've removed more than 50% of values, something is wrong
# Fall back to using median-based statistics which are robust to outliers
return self._compute_robust_statistics(values_array)
# Calculate statistics on filtered values
outliers_percent = (outliers_removed / original_count) * 100 if original_count > 0 else 0.0
return {
"mean": float(np.mean(filtered_values)),
"std": float(np.std(filtered_values)),
"min": float(np.min(filtered_values)),
"max": float(np.max(filtered_values)),
"median": float(np.median(filtered_values)),
"q1": float(np.percentile(filtered_values, 25)),
"q3": float(np.percentile(filtered_values, 75)),
"outliers_removed": int(outliers_removed),
"outliers_percent": float(outliers_percent)
}
# Return simple statistics if not removing outliers
return {
"mean": float(np.mean(values_array)),
"std": float(np.std(values_array)),
"min": float(np.min(values_array)),
"max": float(np.max(values_array)),
"median": float(np.median(values_array)),
"q1": float(np.percentile(values_array, 25)),
"q3": float(np.percentile(values_array, 75)),
"outliers_removed": 0,
"outliers_percent": 0.0
}
def _remove_outliers(self, values_array, method, threshold, z_threshold):
"""Remove outliers using the specified method and return filtered values"""
if method == "iqr":
# IQR method - commonly used for detecting outliers
q1 = np.percentile(values_array, 25)
q3 = np.percentile(values_array, 75)
iqr = q3 - q1
lower_bound = q1 - threshold * iqr
upper_bound = q3 + threshold * iqr
mask = (values_array >= lower_bound) & (values_array <= upper_bound)
filtered_values = values_array[mask]
elif method == "zscore":
# Z-score method - removes values beyond certain standard deviations
mean = np.mean(values_array)
std = np.std(values_array)
if std == 0: # Handle constant arrays
filtered_values = values_array
else:
z_scores = np.abs((values_array - mean) / std)
mask = z_scores < z_threshold
filtered_values = values_array[mask]
elif method == "percentile":
# Percentile method - removes top and bottom 5%
filtered_values = np.percentile(values_array, [5, 95])
filtered_values = values_array[(values_array >= filtered_values[0]) &
(values_array <= filtered_values[1])]
else:
filtered_values = values_array
outliers_removed = len(values_array) - len(filtered_values)
return filtered_values, outliers_removed
def _compute_robust_statistics(self, values_array):
"""Compute robust statistics that are not affected by outliers"""
return {
"mean": float(np.median(values_array)), # Use median instead of mean
"std": float(np.percentile(values_array, 75) - np.percentile(values_array, 25)), # Use IQR as std
"min": float(np.percentile(values_array, 5)), # 5th percentile instead of min
"max": float(np.percentile(values_array, 95)), # 95th percentile instead of max
"median": float(np.median(values_array)),
"q1": float(np.percentile(values_array, 25)),
"q3": float(np.percentile(values_array, 75)),
"outliers_removed": 0, # We're using percentiles instead of removal
"outliers_percent": 0.0
}
def _calculate_summary_statistics(self, successful_count, start_time, fid_score=None):
"""Calculate summary statistics for all collected metrics"""
if successful_count == 0:
return {}
# Calculate model-specific statistics
summary = self._calculate_model_statistics()
# Calculate average across all models
summary["average"] = self._calculate_average_model_statistics(summary)
# Add general metrics
summary["general"] = self._calculate_general_statistics(start_time, fid_score)
return summary
def _calculate_model_statistics(self):
"""Calculate statistics for each face recognition model"""
summary = {}
for model in ["arcface", "facenet", "adaface"]:
model_data = self.collectors.get_model_data(model)
if not model_data or not model_data.get("sim_ref_list"):
continue
model_stats = {
"sim_ref": self._compute_statistics(model_data["sim_ref_list"]),
"non_matched_ref": self._compute_statistics(model_data["unmatched_ref_list"])
}
# Add original image metrics if available
if self.ori_flag and model_data.get("sim_ori_list"):
model_stats.update({
"sim_ori": self._compute_statistics(model_data["sim_ori_list"]),
"non_matched_ori": self._compute_statistics(model_data["unmatched_list"])
})
# Add copy-geo metrics if available
if model_data.get("copy_geo_list"):
model_stats["copy_geo"] = self._compute_statistics(model_data["copy_geo_list"])
# Add cluster metrics if available
if self.name_flag and model_data.get("sim_cluster_list"):
model_stats["sim_cluster"] = self._compute_statistics(model_data["sim_cluster_list"])
# Calculate difference between original and cluster if both available
if self.ori_flag and model_data.get("sim_ori_list"):
diff_ori_cluster = [
ori - cluster for ori, cluster in zip(
model_data["sim_ori_list"], model_data["sim_cluster_list"]
)
]
model_stats["diff_ori_cluster"] = self._compute_statistics(diff_ori_cluster)
summary[model] = model_stats
return summary
def _calculate_average_model_statistics(self, model_summary):
"""Calculate average statistics across all models"""
models_with_data = [model for model in ["arcface", "facenet", "adaface"] if model in model_summary]
if not models_with_data:
return {}
# Combine data across models
combined_data = self._combine_model_data(models_with_data)
# Calculate statistics on combined data
avg_stats = {
"sim_ref": self._compute_statistics(combined_data["all_sim_ref"]),
"non_matched_ref": self._compute_statistics(combined_data["all_non_matched_ref"])
}
# Add conditional statistics
if combined_data["all_sim_ori"]:
avg_stats["sim_ori"] = self._compute_statistics(combined_data["all_sim_ori"])
if combined_data["all_non_matched_ori"]:
avg_stats["non_matched_ori"] = self._compute_statistics(combined_data["all_non_matched_ori"])
if combined_data["all_sim_cluster"]:
avg_stats["sim_cluster"] = self._compute_statistics(combined_data["all_sim_cluster"])
if combined_data["all_copy_geo"]:
avg_stats["copy_geo"] = self._compute_statistics(combined_data["all_copy_geo"])
if combined_data["all_diff_ori_cluster"]:
avg_stats["diff_ori_cluster"] = self._compute_statistics(combined_data["all_diff_ori_cluster"])
return avg_stats
def _combine_model_data(self, models):
"""Combine data from multiple models for averaging"""
all_sim_ref = []
all_non_matched_ref = []
all_sim_ori = []
all_non_matched_ori = []
all_sim_cluster = []
all_copy_geo = []
all_diff_ori_cluster = []
for model in models:
model_data = self.collectors.get_model_data(model)
# Combine reference similarity data
all_sim_ref.extend(model_data["sim_ref_list"])
all_non_matched_ref.extend(model_data["unmatched_ref_list"])
# Combine original image data if available
if self.ori_flag and model_data.get("sim_ori_list"):
all_sim_ori.extend(model_data["sim_ori_list"])
all_non_matched_ori.extend(model_data["unmatched_list"])
# Combine copy-geo data if available
if model_data.get("copy_geo_list"):
all_copy_geo.extend(model_data["copy_geo_list"])
# Combine cluster data if available
if self.name_flag and model_data.get("sim_cluster_list"):
all_sim_cluster.extend(model_data["sim_cluster_list"])
# Calculate differences if both original and cluster data are available
if self.ori_flag and self.name_flag and all_sim_ori and all_sim_cluster:
min_len = min(len(all_sim_ori), len(all_sim_cluster))
all_diff_ori_cluster = [all_sim_ori[i] - all_sim_cluster[i] for i in range(min_len)]
return {
"all_sim_ref": all_sim_ref,
"all_non_matched_ref": all_non_matched_ref,
"all_sim_ori": all_sim_ori,
"all_non_matched_ori": all_non_matched_ori,
"all_sim_cluster": all_sim_cluster,
"all_copy_geo": all_copy_geo,
"all_diff_ori_cluster": all_diff_ori_cluster
}
def _calculate_general_statistics(self, start_time, fid_score):
"""Calculate general statistics from collected metrics"""
general_data = self.collectors.get_general_data()
general_metrics = {
"clip_t": self._compute_statistics(general_data["sim_clip_t_list"]),
"aesthetic": self._compute_statistics(general_data["aes_list"]),
"anatomy": self._compute_statistics(general_data["anatomy_scores"]),
"total_samples": self.collectors.total_count,
"failed_samples": self.collectors.bad_count,
"successful_samples": self.collectors.successful_count,
"duration_seconds": time.time() - start_time
}
# Add CLIP image similarity if available
if self.ori_flag and general_data.get("sim_clip_i_list"):
general_metrics["clip_i"] = self._compute_statistics(general_data["sim_clip_i_list"])
# Add FID score if available
if self.ori_flag and fid_score is not None:
general_metrics["fid"] = fid_score
return general_metrics
def _print_summary_table(self, summary, start_time):
"""Print evaluation summary as a formatted table"""
reporter = SummaryReporter(
summary, self.ori_flag, self.name_flag, start_time
)
reporter.print_summary()
def __call__(self) -> Dict:
"""
Run the evaluation process on all directories.
Returns:
Dict with evaluation results
"""
print(f"\n{Fore.CYAN}╔══════════════════════════════════════╗")
print(f"║ CELEB BENCHMARK EVALUATION TOOL ║")
print(f"╚══════════════════════════════════════╝{Style.RESET_ALL}\n")
start_time = time.time()
# Collect images for FID
original_images = [] if self.ori_flag else None
generated_images = []
# Process all directories with progress bar
for dir_idx, directory in enumerate(tqdm(self.list_dir, desc="Evaluating samples", unit="sample")):
self.collectors.total_count += 1
# Process directory path and metadata
dir_path, metadata = self._process_directory_metadata(directory)
if not metadata:
self.collectors.bad_count += 1
continue
# Extract caption and names from metadata
caption, names = self._extract_metadata_text(metadata, directory)
# Load images
ori_img, output_img = self._load_primary_images(dir_path, directory)
if (self.ori_flag and not ori_img) or not output_img:
self.collectors.bad_count += 1
continue
# Load reference images
refs, ref_count = self._load_reference_images(dir_path)
if ref_count == 0:
print(f"{Fore.RED}No reference images found for {directory}, skipping...{Style.RESET_ALL}")
self.collectors.bad_count += 1
continue
# Calculate metrics for this sample
try:
all_metrics = self._calculate_metrics_for_sample(
output_img, ori_img, refs, caption, names, dir_idx, directory
)
if not all_metrics:
self.collectors.bad_count += 1
continue
# Collect metrics and add to results
self._collect_sample_metrics(all_metrics, metadata, ref_count, output_img)
self.collectors.successful_count += 1
# Collect images for FID
if self.ori_flag:
original_images.append(ori_img)
generated_images.append(output_img)
except Exception as e:
print(f"{Fore.RED}Error processing {directory}: {e}{Style.RESET_ALL}")
self.collectors.bad_count += 1
continue
# Calculate FID score
fid_score = self._calculate_fid(original_images, generated_images)
if fid_score is not None:
self.results["fid"] = fid_score
# Calculate summary statistics if we have successful samples
if self.collectors.successful_count > 0:
summary = self._calculate_summary_statistics(
self.collectors.successful_count, start_time, fid_score
)
# Store summary in results
self.results["summary"] = summary
# Print summary table
self._print_summary_table(summary, start_time)
# Return complete results
return self.results
def _process_directory_metadata(self, directory):
"""Process directory path and load metadata"""
dir_path = os.path.join(self.target_dir, directory)
meta_path = os.path.join(dir_path, self.meta_file_name)
# Check if metadata exists
if not os.path.exists(meta_path):
print(f"{Fore.YELLOW}Meta file not found for {directory}, skipping...{Style.RESET_ALL}")
print(f"{meta_path}")
return dir_path, None
# Load metadata
metadata = self._load_metadata(meta_path)
if metadata:
# Add directory path to metadata for reference
metadata["path"] = dir_path
return dir_path, metadata
def _extract_metadata_text(self, metadata, directory):
"""Extract caption and names from metadata"""
caption = metadata.get(self.caption_keyword, "")
names = metadata.get(self.names_keyword, []) if self.name_flag else []
# Log warnings for missing data
if not caption:
print(f"{Fore.YELLOW}Warning: No caption found for {directory}{Style.RESET_ALL}")
if self.name_flag and not names:
print(f"{Fore.YELLOW}Warning: No names found for {directory}{Style.RESET_ALL}")
return caption, names
def _load_primary_images(self, dir_path, directory):
"""Load original and output images"""
ori_img = None
if self.ori_flag:
ori_img = self._load_image(os.path.join(dir_path, self.ori_file_name))
if not ori_img:
print(f"{Fore.RED}Original image not found for {directory}, skipping...{Style.RESET_ALL}")
return None, None
output_img = self._load_image(os.path.join(dir_path, self.output_file_name))
if not output_img:
print(f"{Fore.RED}Output image not found for {directory}, skipping...{Style.RESET_ALL}")
return ori_img, None
return ori_img, output_img
class MetricCollectors:
"""Helper class to collect and organize metrics during evaluation"""
def __init__(self, use_original=True, use_names=False):
# Statistics
self.total_count = 0
self.bad_count = 0
self.successful_count = 0
# Initialize model-specific collectors
self.model_collectors = {}
for model in ["arcface", "facenet", "adaface"]:
self.model_collectors[model] = self._init_model_collectors(use_original, use_names)
# Initialize general metrics collectors
self.general_collectors = self._init_general_collectors(use_original)
def _init_model_collectors(self, use_original, use_names):
"""Initialize metric collectors for a specific model"""
collectors = {
"sim_ref_list": [],
"unmatched_ref_list": []
}
# Add collectors for original image metrics if needed
if use_original:
collectors.update({
"sim_ori_list": [],
"unmatched_list": [],
"copy_geo_list": []
})
# Add collectors for name/cluster metrics if needed
if use_names:
collectors["sim_cluster_list"] = []
return collectors
def _init_general_collectors(self, use_original):
"""Initialize collectors for general (non-model-specific) metrics"""
collectors = {
"sim_clip_t_list": [],
"aes_list": [],
"anatomy_scores": []
}
# Add collectors for original image metrics if needed
if use_original:
collectors["sim_clip_i_list"] = []
return collectors
def add_model_metrics(self, model, metrics, use_original, use_names):
"""Add metrics for a specific model to collectors"""
collectors = self.model_collectors[model]
# Always add reference similarity metrics
if metrics["sim_ref"] is not None:
collectors["sim_ref_list"].append(float(metrics["sim_ref"].mean()))
collectors["unmatched_ref_list"].append(float(metrics["non_matched_ref"]))
# Add original image metrics if available
if use_original and metrics["sim_ori"] is not None:
collectors["sim_ori_list"].append(float(metrics["sim_ori"].mean()))
collectors["unmatched_list"].append(float(metrics["non_matched_ori"]))
# Add copy-geo metrics if available
if metrics["copy_geo"] is not None:
collectors["copy_geo_list"].append(float(metrics["copy_geo"].mean()))
# Add cluster metrics if available
if use_names and metrics["sim_cluster"] is not None:
collectors["sim_cluster_list"].append(float(metrics["sim_cluster"].mean()))
def add_general_metrics(self, clip_i, clip_t, aesthetic_score, anatomy_score):
"""Add general metrics to collectors"""
# Always add text metrics
self.general_collectors["sim_clip_t_list"].append(float(clip_t))
self.general_collectors["aes_list"].append(float(aesthetic_score))
self.general_collectors["anatomy_scores"].append(float(anatomy_score))
# Add image metrics if available
if clip_i is not None:
self.general_collectors["sim_clip_i_list"].append(float(clip_i))
def get_model_data(self, model):
"""Get collected data for a specific model"""
return self.model_collectors.get(model, {})
def get_general_data(self):
"""Get collected general metrics data"""
return self.general_collectors
class SummaryReporter:
"""Helper class to generate and print evaluation summary reports"""
def __init__(self, summary, ori_flag, name_flag, start_time):
self.summary = summary
self.ori_flag = ori_flag
self.name_flag = name_flag
self.start_time = start_time
self.duration = time.time() - start_time
def print_summary(self):
"""Print the complete summary report"""
self._print_header()
self._print_model_metrics()
self._print_general_metrics()
self._print_footer()
def _print_header(self):
"""Print summary header with basic statistics"""
general = self.summary.get("general", {})
print(f"\n{Fore.CYAN}╔═══════════════════════════════════════════════════════╗")
print(f"║ EVALUATION SUMMARY RESULTS ║")
print(f"╠═══════════════════════════════════════════════════════╣")
print(f"║ {Fore.WHITE}Total samples:{Fore.CYAN} {general.get('total_samples', 0):<27} ║")
print(f"║ {Fore.WHITE}Successful samples:{Fore.CYAN} {general.get('successful_samples', 0):<27} ║")
print(f"║ {Fore.WHITE}Failed samples:{Fore.CYAN} {general.get('failed_samples', 0):<27} ║")
def _print_model_metrics(self):
"""Print metrics for each face recognition model"""
for model_name in ["arcface", "facenet", "adaface", "average"]:
if model_name not in self.summary:
continue
model_stats = self.summary[model_name]
print(f"╠═══════════════════════════════════════════════════════╣")
print(f"║ {Fore.YELLOW}{model_name.upper()} METRICS:{' ':>37} ║")
# Print similarity metrics with original if available
if self.ori_flag and "sim_ori" in model_stats:
print(f"║ {Fore.CYAN}Face similarity with original:{' ':>25} ║")
self._print_stat_row("sim_ori", model_stats)
# Print similarity metrics with references
print(f"║ {Fore.CYAN}Face similarity with reference:{' ':>24} ║")
self._print_stat_row("sim_ref", model_stats)
# Print similarity metrics with cluster if available
if self.name_flag and "sim_cluster" in model_stats:
print(f"║ {Fore.CYAN}Cluster similarity:{' ':>34} ║")
self._print_stat_row("sim_cluster", model_stats)
# Print difference metrics if available
if self.ori_flag and "copy_geo" in model_stats:
print(f"║ {Fore.CYAN}Difference (original vs reference):{' ':>18} ║")
self._print_stat_row("copy_geo", model_stats)
if self.ori_flag and self.name_flag and "diff_ori_cluster" in model_stats:
print(f"║ {Fore.CYAN}Difference (original vs cluster):{' ':>20} ║")
self._print_stat_row("diff_ori_cluster", model_stats)
# Print non-matched metrics if available
if self.ori_flag and "non_matched_ori" in model_stats:
print(f"║ {Fore.CYAN}Non-matched faces (original):{' ':>24} ║")
self._print_stat_row("non_matched_ori", model_stats)
print(f"║ {Fore.CYAN}Non-matched faces (reference):{' ':>23} ║")
self._print_stat_row("non_matched_ref", model_stats)
def _print_general_metrics(self):
"""Print general metrics"""
if "general" not in self.summary:
return
general = self.summary["general"]
print(f"╠═══════════════════════════════════════════════════════╣")
print(f"║ {Fore.YELLOW}GENERAL METRICS:{' ':>39} ║")
# Print CLIP image similarity if available
if self.ori_flag and "clip_i" in general:
print(f"║ {Fore.CYAN}CLIP image similarity:{' ':>31} ║")
self._print_stat_row("clip_i", general)
# Print CLIP text similarity
print(f"║ {Fore.CYAN}CLIP text similarity:{' ':>32} ║")
self._print_stat_row("clip_t", general)
# Print aesthetic score
print(f"║ {Fore.CYAN}Aesthetic score:{' ':>36} ║")
self._print_stat_row("aesthetic", general)
# Print anatomy score
print(f"║ {Fore.CYAN}Anatomy score:{' ':>38} ║")
self._print_stat_row("anatomy", general)
# Print FID if available
if self.ori_flag and "fid" in general:
print(f"║ {Fore.GREEN}FID score:{Fore.CYAN} {general['fid']:.4f}{' ':>21} ║")
def _print_footer(self):
"""Print summary footer"""
print(f"╠═══════════════════════════════════════════════════════╣")
print(f"║ {Fore.WHITE}Evaluation completed in {self.duration:.2f} seconds{' ':>13} ║")
print(f"╚═══════════════════════════════════════════════════════╝{Style.RESET_ALL}")
def _print_stat_row(self, metric_name, stat_dict, prefix=""):
"""Print statistics for a specific metric"""
if not stat_dict or metric_name not in stat_dict:
return
stats = stat_dict[metric_name]
if not stats:
return
# Print detailed stats for important metrics
if metric_name in ["sim_ori", "copy_geo"]:
print(f"║ {Fore.GREEN}{prefix}Mean:{Fore.CYAN} {stats['mean']:.4f}{' ':>30} ║")
print(f"║ {Fore.GREEN}{prefix}Median:{Fore.CYAN} {stats['median']:.4f}{' ':>30} ║")
print(f"║ {Fore.GREEN}{prefix}StdDev:{Fore.CYAN} {stats['std']:.4f}{' ':>30} ║")
print(f"║ {Fore.GREEN}{prefix}Min:{Fore.CYAN} {stats['min']:.4f}{' ':>30} ║")
print(f"║ {Fore.GREEN}{prefix}Max:{Fore.CYAN} {stats['max']:.4f}{' ':>30} ║")
print(f"║ {Fore.GREEN}{prefix}Q1:{Fore.CYAN} {stats['q1']:.4f}{' ':>30} ║")
print(f"║ {Fore.GREEN}{prefix}Q3:{Fore.CYAN} {stats['q3']:.4f}{' ':>30} ║")
else:
# Only print mean for other metrics
print(f"║ {Fore.GREEN}{prefix}Mean:{Fore.CYAN} {stats['mean']:.4f}{' ':>30} ║")
def v200_single(target_dir, output_dir):
evaler = BenchEval_Geo(
target_dir=target_dir,
output_dir=output_dir,
ori_file_name="ori.jpg",
output_file_name="out.jpg",
ref_1_file_name="ref_1.jpg",
# ref_2_file_name="ref_2.jpg",
ref_2_file_name=None,
caption_keyword="prompt",
names_keyword=None
)
evaler()
def v202(target_dir, output_dir):
evaler = BenchEval_Geo(
target_dir=target_dir,
output_dir=output_dir,
ori_file_name="ori.jpg",
output_file_name="out.jpg",
ref_1_file_name="ref_0.jpg",
ref_2_file_name="ref_1.jpg",
# ref_2_file_name=None,
caption_keyword="prompt",
names_keyword=None
)
evaler()
if __name__ == "__main__":
v202("../debug_test", "../debug_test/eval")