from __future__ import annotations from pathlib import Path import numpy as np import pytest import torch from PIL import Image from torchmetrics.image import PeakSignalNoiseRatio, StructuralSimilarityIndexMeasure def model_output_dir(parent_dir: Path, model: str) -> Path: safe_model_name = model.split("/")[-1].replace(".", "_") path = parent_dir / safe_model_name path.mkdir(parents=True, exist_ok=True) return path def assert_similarity( *, model_name: str, vllm_image: Image.Image, diffusers_image: Image.Image, width: int, height: int, ssim_threshold: float, psnr_threshold: float, ) -> None: requested_size = (width, height) if diffusers_image.size != requested_size: pytest.skip( "Skipping as diffusers baseline output is corrupt and not comparable: " f"dimensions do not match requested size; requested={requested_size}, got={diffusers_image.size}." ) assert vllm_image.size == diffusers_image.size, ( f"Online and diffusers output sizes mismatch: online={vllm_image.size}, diffusers={diffusers_image.size}" ) ssim_score, psnr_score = compute_image_ssim_psnr(prediction=vllm_image, reference=diffusers_image) print(f"{model_name} similarity metrics:") print(f" SSIM: value={ssim_score:.6f}, threshold>={ssim_threshold:.6f}, range=[-1, 1], higher_is_better=True") print( f" PSNR: value={psnr_score:.6f} dB, threshold>={psnr_threshold:.6f} dB, range=[0, +inf), higher_is_better=True" ) assert ssim_score >= ssim_threshold, ( f"SSIM below threshold for {model_name}: got {ssim_score:.6f}, expected >= {ssim_threshold:.6f}." ) assert psnr_score >= psnr_threshold, ( f"PSNR below threshold for {model_name}: got {psnr_score:.6f}, expected >= {psnr_threshold:.6f}." ) def compute_image_ssim_psnr( *, prediction: Image.Image, reference: Image.Image, ) -> tuple[float, float]: pred_tensor = _pil_to_batched_tensor(prediction) ref_tensor = _pil_to_batched_tensor(reference) ssim_metric = StructuralSimilarityIndexMeasure(data_range=1.0) psnr_metric = PeakSignalNoiseRatio(data_range=1.0) ssim_value = float(ssim_metric(pred_tensor, ref_tensor).item()) psnr_value = float(psnr_metric(pred_tensor, ref_tensor).item()) return ssim_value, psnr_value def _pil_to_batched_tensor(image: Image.Image) -> torch.Tensor: array = np.asarray(image.convert("RGB"), dtype=np.float32) / 255.0 tensor = torch.from_numpy(array).permute(2, 0, 1).unsqueeze(0) return tensor