





'''simple docstring'''

import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModelWithProjection

from diffusers import (
    AutoencoderKL,
    DPMSolverMultistepScheduler,
    PNDMScheduler,
    StableDiffusionImageVariationPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, nightly, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()


class      SCREAMING_SNAKE_CASE__      (   snake_case__							,snake_case__							,snake_case__							,unittest.TestCase		):







					"""simple docstring"""






					SCREAMING_SNAKE_CASE							=	StableDiffusionImageVariationPipeline
					SCREAMING_SNAKE_CASE							=	IMAGE_VARIATION_PARAMS
					SCREAMING_SNAKE_CASE							=	IMAGE_VARIATION_BATCH_PARAMS
					SCREAMING_SNAKE_CASE							=	frozenset([]		)
					# TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
					SCREAMING_SNAKE_CASE							=	frozenset([]		)
					def 		lowerCamelCase_   (   self       :      Optional[Any]		):







							"""simple docstring"""
							torch.manual_seed(0		)
							__UpperCAmelCase	:							Optional[int]     =     UNetaDConditionModel(
							    block_out_channels=(32, 64)		,	layers_per_block=2		,	sample_size=32		,	in_channels=4		,	out_channels=4		,	down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")		,	up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")		,	cross_attention_dim=32		,	)
							__UpperCAmelCase	:							List[Any]     =     PNDMScheduler(skip_prk_steps=UpperCAmelCase_		)
							torch.manual_seed(0		)
							__UpperCAmelCase	:							List[Any]     =     AutoencoderKL(
							    block_out_channels=[32, 64]		,	in_channels=3		,	out_channels=3		,	down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]		,	up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]		,	latent_channels=4		,	)
							torch.manual_seed(0		)
							__UpperCAmelCase	:							Union[str, Any]     =     CLIPVisionConfig(
							    hidden_size=32		,	projection_dim=32		,	intermediate_size=37		,	layer_norm_eps=1e-05		,	num_attention_heads=4		,	num_hidden_layers=5		,	image_size=32		,	patch_size=4		,	)
							__UpperCAmelCase	:							Union[str, Any]     =     CLIPVisionModelWithProjection(UpperCAmelCase_		)
							__UpperCAmelCase	:							Optional[Any]     =     CLIPImageProcessor(crop_size=32		,	size=32		)

							__UpperCAmelCase	:							Tuple     =     {
							    "unet": unet,
							    "scheduler": scheduler,
							    "vae": vae,
							    "image_encoder": image_encoder,
							    "feature_extractor": feature_extractor,
							    "safety_checker": None,
							}
							return components
					def 		lowerCamelCase_   (   self       :      Dict		,	UpperCAmelCase_       :      Optional[Any]		,	UpperCAmelCase_       :      Optional[Any]=0		):







							"""simple docstring"""
							__UpperCAmelCase	:							Union[str, Any]     =     floats_tensor((1, 3, 32, 32)		,	rng=random.Random(UpperCAmelCase_		)		)
							__UpperCAmelCase	:							int     =     image.cpu().permute(0		,	2		,	3		,	1		)[0]
							__UpperCAmelCase	:							Any     =     Image.fromarray(np.uinta(UpperCAmelCase_		)		).convert("RGB"		).resize((32, 32)		)
							if str(UpperCAmelCase_		).startswith("mps"		):
									__UpperCAmelCase	:							Union[str, Any]     =     torch.manual_seed(UpperCAmelCase_		)
							else:
									__UpperCAmelCase	:							List[str]     =     torch.Generator(device=UpperCAmelCase_		).manual_seed(UpperCAmelCase_		)
							__UpperCAmelCase	:							Any     =     {
							    "image": image,
							    "generator": generator,
							    "num_inference_steps": 2,
							    "guidance_scale": 6.0,
							    "output_type": "numpy",
							}
							return inputs
					def 		lowerCamelCase_   (   self       :      Tuple		):







							"""simple docstring"""
							__UpperCAmelCase	:							Optional[int]     =     "cpu"  # ensure determinism for the device-dependent torch.Generator
							__UpperCAmelCase	:							str     =     self.get_dummy_components()
							__UpperCAmelCase	:							Optional[int]     =     StableDiffusionImageVariationPipeline(**UpperCAmelCase_		)
							__UpperCAmelCase	:							Tuple     =     sd_pipe.to(UpperCAmelCase_		)
							sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_		)

							__UpperCAmelCase	:							Optional[int]     =     self.get_dummy_inputs(UpperCAmelCase_		)
							__UpperCAmelCase	:							Union[str, Any]     =     sd_pipe(**UpperCAmelCase_		).images
							__UpperCAmelCase	:							str     =     image[0, -3:, -3:, -1]

							assert image.shape == (1, 64, 64, 3)
							__UpperCAmelCase	:							List[Any]     =     np.array([0.5239, 0.5723, 0.4796, 0.5049, 0.5550, 0.4685, 0.5329, 0.4891, 0.4921]		)

							assert np.abs(image_slice.flatten() - expected_slice		).max() < 1e-3
					def 		lowerCamelCase_   (   self       :      Union[str, Any]		):







							"""simple docstring"""
							__UpperCAmelCase	:							List[Any]     =     "cpu"  # ensure determinism for the device-dependent torch.Generator
							__UpperCAmelCase	:							Tuple     =     self.get_dummy_components()
							__UpperCAmelCase	:							Union[str, Any]     =     StableDiffusionImageVariationPipeline(**UpperCAmelCase_		)
							__UpperCAmelCase	:							str     =     sd_pipe.to(UpperCAmelCase_		)
							sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_		)

							__UpperCAmelCase	:							List[Any]     =     self.get_dummy_inputs(UpperCAmelCase_		)
							__UpperCAmelCase	:							Any     =     2 * [inputs["image"]]
							__UpperCAmelCase	:							Optional[int]     =     sd_pipe(**UpperCAmelCase_		)

							__UpperCAmelCase	:							Union[str, Any]     =     output.images

							__UpperCAmelCase	:							List[str]     =     image[-1, -3:, -3:, -1]

							assert image.shape == (2, 64, 64, 3)
							__UpperCAmelCase	:							str     =     np.array([0.6892, 0.5637, 0.5836, 0.5771, 0.6254, 0.6409, 0.5580, 0.5569, 0.5289]		)

							assert np.abs(image_slice.flatten() - expected_slice		).max() < 1e-3






					def 		lowerCamelCase_   (   self       :      Optional[int]		):







							"""simple docstring"""
							super().test_inference_batch_single_identical(expected_max_diff=3e-3		)






@slow
@require_torch_gpu
class      SCREAMING_SNAKE_CASE__      (   unittest.TestCase		):







					"""simple docstring"""






					def 		lowerCamelCase_   (   self       :      str		):







							"""simple docstring"""
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()
					def 		lowerCamelCase_   (   self       :      Union[str, Any]		,	UpperCAmelCase_       :      Optional[Any]		,	UpperCAmelCase_       :      Optional[Any]="cpu"		,	UpperCAmelCase_       :      str=torch.floataa		,	UpperCAmelCase_       :      Tuple=0		):







							"""simple docstring"""
							__UpperCAmelCase	:							Dict     =     torch.Generator(device=UpperCAmelCase_		).manual_seed(UpperCAmelCase_		)
							__UpperCAmelCase	:							List[str]     =     load_image(
							    "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
							    "/stable_diffusion_imgvar/input_image_vermeer.png"		)
							__UpperCAmelCase	:							Dict     =     np.random.RandomState(UpperCAmelCase_		).standard_normal((1, 4, 64, 64)		)
							__UpperCAmelCase	:							List[str]     =     torch.from_numpy(UpperCAmelCase_		).to(device=UpperCAmelCase_		,	dtype=UpperCAmelCase_		)
							__UpperCAmelCase	:							Optional[int]     =     {
							    "image": init_image,
							    "latents": latents,
							    "generator": generator,
							    "num_inference_steps": 3,
							    "guidance_scale": 7.5,
							    "output_type": "numpy",
							}
							return inputs
					def 		lowerCamelCase_   (   self       :      List[Any]		):







							"""simple docstring"""
							__UpperCAmelCase	:							Any     =     StableDiffusionImageVariationPipeline.from_pretrained(
							    "lambdalabs/sd-image-variations-diffusers"		,	safety_checker=UpperCAmelCase_		)
							__UpperCAmelCase	:							Optional[Any]     =     sd_pipe.to(UpperCAmelCase_		)
							sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_		)

							__UpperCAmelCase	:							List[Any]     =     self.get_inputs(UpperCAmelCase_		)
							__UpperCAmelCase	:							Optional[int]     =     sd_pipe(**UpperCAmelCase_		).images
							__UpperCAmelCase	:							Optional[Any]     =     image[0, -3:, -3:, -1].flatten()

							assert image.shape == (1, 512, 512, 3)
							__UpperCAmelCase	:							Union[str, Any]     =     np.array([0.84491, 0.90789, 0.75708, 0.78734, 0.83485, 0.70099, 0.66938, 0.68727, 0.61379]		)
							assert np.abs(image_slice - expected_slice		).max() < 6e-3
					def 		lowerCamelCase_   (   self       :      Optional[int]		):







							"""simple docstring"""
							__UpperCAmelCase	:							Dict     =     0

							def callback_fn(UpperCAmelCase_       :      int		,	UpperCAmelCase_       :      int		,	UpperCAmelCase_       :      torch.FloatTensor		) -> None:
									__UpperCAmelCase	:							List[Any]     =     True
									nonlocal number_of_steps
									number_of_steps += 1
									if step == 1:
											__UpperCAmelCase	:							int     =     latents.detach().cpu().numpy()
											assert latents.shape == (1, 4, 64, 64)
											__UpperCAmelCase	:							int     =     latents[0, -3:, -3:, -1]
											__UpperCAmelCase	:							Optional[Any]     =     np.array(
											    [-0.1621, 0.2837, -0.7979, -0.1221, -1.3057, 0.7681, -2.1191, 0.0464, 1.6309]		)

											assert np.abs(latents_slice.flatten() - expected_slice		).max() < 5e-2
									elif step == 2:
											__UpperCAmelCase	:							List[Any]     =     latents.detach().cpu().numpy()
											assert latents.shape == (1, 4, 64, 64)
											__UpperCAmelCase	:							str     =     latents[0, -3:, -3:, -1]
											__UpperCAmelCase	:							Optional[Any]     =     np.array([0.6299, 1.7500, 1.1992, -2.1582, -1.8994, 0.7334, -0.7090, 1.0137, 1.5273]		)

											assert np.abs(latents_slice.flatten() - expected_slice		).max() < 5e-2

							__UpperCAmelCase	:							Tuple     =     False

							__UpperCAmelCase	:							Tuple     =     StableDiffusionImageVariationPipeline.from_pretrained(
							    "fusing/sd-image-variations-diffusers"		,	safety_checker=UpperCAmelCase_		,	torch_dtype=torch.floataa		,	)
							pipe.to(UpperCAmelCase_		)
							pipe.set_progress_bar_config(disable=UpperCAmelCase_		)
							pipe.enable_attention_slicing()

							__UpperCAmelCase	:							Tuple     =     self.get_inputs(UpperCAmelCase_		,	dtype=torch.floataa		)
							pipe(**UpperCAmelCase_		,	callback=UpperCAmelCase_		,	callback_steps=1		)
							assert callback_fn.has_been_called
							assert number_of_steps == inputs["num_inference_steps"]






					def 		lowerCamelCase_   (   self       :      int		):







							"""simple docstring"""
							torch.cuda.empty_cache()
							torch.cuda.reset_max_memory_allocated()
							torch.cuda.reset_peak_memory_stats()

							__UpperCAmelCase	:							Union[str, Any]     =     "fusing/sd-image-variations-diffusers"
							__UpperCAmelCase	:							Optional[int]     =     StableDiffusionImageVariationPipeline.from_pretrained(
							    UpperCAmelCase_		,	safety_checker=UpperCAmelCase_		,	torch_dtype=torch.floataa		)
							__UpperCAmelCase	:							Dict     =     pipe.to(UpperCAmelCase_		)
							pipe.set_progress_bar_config(disable=UpperCAmelCase_		)
							pipe.enable_attention_slicing(1		)
							pipe.enable_sequential_cpu_offload()

							__UpperCAmelCase	:							List[str]     =     self.get_inputs(UpperCAmelCase_		,	dtype=torch.floataa		)
							__UpperCAmelCase	:							List[str]     =     pipe(**UpperCAmelCase_		)

							__UpperCAmelCase	:							Dict     =     torch.cuda.max_memory_allocated()
							# make sure that less than 2.6 GB is allocated
							assert mem_bytes < 2.6 * 10**9






@nightly
@require_torch_gpu
class      SCREAMING_SNAKE_CASE__      (   unittest.TestCase		):







					"""simple docstring"""






					def 		lowerCamelCase_   (   self       :      int		):







							"""simple docstring"""
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()
					def 		lowerCamelCase_   (   self       :      List[str]		,	UpperCAmelCase_       :      Any		,	UpperCAmelCase_       :      Dict="cpu"		,	UpperCAmelCase_       :      Dict=torch.floataa		,	UpperCAmelCase_       :      Any=0		):







							"""simple docstring"""
							__UpperCAmelCase	:							Optional[int]     =     torch.Generator(device=UpperCAmelCase_		).manual_seed(UpperCAmelCase_		)
							__UpperCAmelCase	:							List[Any]     =     load_image(
							    "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
							    "/stable_diffusion_imgvar/input_image_vermeer.png"		)
							__UpperCAmelCase	:							Dict     =     np.random.RandomState(UpperCAmelCase_		).standard_normal((1, 4, 64, 64)		)
							__UpperCAmelCase	:							Any     =     torch.from_numpy(UpperCAmelCase_		).to(device=UpperCAmelCase_		,	dtype=UpperCAmelCase_		)
							__UpperCAmelCase	:							List[str]     =     {
							    "image": init_image,
							    "latents": latents,
							    "generator": generator,
							    "num_inference_steps": 50,
							    "guidance_scale": 7.5,
							    "output_type": "numpy",
							}
							return inputs
					def 		lowerCamelCase_   (   self       :      Dict		):







							"""simple docstring"""
							__UpperCAmelCase	:							List[str]     =     StableDiffusionImageVariationPipeline.from_pretrained("fusing/sd-image-variations-diffusers"		)
							sd_pipe.to(UpperCAmelCase_		)
							sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_		)

							__UpperCAmelCase	:							Optional[int]     =     self.get_inputs(UpperCAmelCase_		)
							__UpperCAmelCase	:							Any     =     sd_pipe(**UpperCAmelCase_		).images[0]

							__UpperCAmelCase	:							List[str]     =     load_numpy(
							    "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
							    "/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy"		)
							__UpperCAmelCase	:							Tuple     =     np.abs(expected_image - image		).max()
							assert max_diff < 1e-3






					def 		lowerCamelCase_   (   self       :      List[Any]		):







							"""simple docstring"""
							__UpperCAmelCase	:							Any     =     StableDiffusionImageVariationPipeline.from_pretrained("fusing/sd-image-variations-diffusers"		)
							__UpperCAmelCase	:							Tuple     =     DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config		)
							sd_pipe.to(UpperCAmelCase_		)
							sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_		)

							__UpperCAmelCase	:							Any     =     self.get_inputs(UpperCAmelCase_		)
							__UpperCAmelCase	:							Union[str, Any]     =     25
							__UpperCAmelCase	:							int     =     sd_pipe(**UpperCAmelCase_		).images[0]

							__UpperCAmelCase	:							Dict     =     load_numpy(
							    "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
							    "/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy"		)
							__UpperCAmelCase	:							List[str]     =     np.abs(expected_image - image		).max()
							assert max_diff < 1e-3











'''simple docstring'''

import os
from shutil import copyfile
from typing import List, Optional, Tuple

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging


if is_sentencepiece_available():
			from .tokenization_xglm import XGLMTokenizer
else:
			lowerCAmelCase__      :   List[str]      	=			None


lowerCAmelCase__      :   Union[str, Any]      	=			logging.get_logger(__name__)

lowerCAmelCase__      :   Tuple      	=			{"vocab_file": "sentencepiece.bpe.model", "tokenizer_file": "tokenizer.json"}

lowerCAmelCase__      :   List[str]      	=			{
    "vocab_file": {
        "facebook/xglm-564M": "https://huggingface.co/facebook/xglm-564M/resolve/main/sentencepiece.bpe.model",
    },
    "tokenizer_file": {
        "facebook/xglm-564M": "https://huggingface.co/facebook/xglm-564M/resolve/main/tokenizer.json",
    },
}

lowerCAmelCase__      :   int      	=			{
    "facebook/xglm-564M": 20_48,
}


class      SCREAMING_SNAKE_CASE__      (   snake_case__		):







					"""simple docstring"""






					SCREAMING_SNAKE_CASE							=	VOCAB_FILES_NAMES
					SCREAMING_SNAKE_CASE							=	PRETRAINED_VOCAB_FILES_MAP
					SCREAMING_SNAKE_CASE							=	PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
					SCREAMING_SNAKE_CASE							=	['''input_ids''', '''attention_mask''']
					SCREAMING_SNAKE_CASE							=	XGLMTokenizer
					def __init__(   self       :      Optional[int]		,	UpperCAmelCase_       :      List[Any]=None		,	UpperCAmelCase_       :      str=None		,	UpperCAmelCase_       :      str="<s>"		,	UpperCAmelCase_       :      List[Any]="</s>"		,	UpperCAmelCase_       :      Dict="</s>"		,	UpperCAmelCase_       :      Optional[Any]="<s>"		,	UpperCAmelCase_       :      List[str]="<unk>"		,	UpperCAmelCase_       :      Union[str, Any]="<pad>"		,	**UpperCAmelCase_       :      List[Any]		,	):







							"""simple docstring"""
							# Compatibility with the original tokenizer
							__UpperCAmelCase	:							Union[str, Any]     =     7
							__UpperCAmelCase	:							Tuple     =     [f"<madeupword{i}>" for i in range(self.num_madeup_words		)]

							__UpperCAmelCase	:							List[Any]     =     kwargs.get("additional_special_tokens"		,	[]		)
							kwargs["additional_special_tokens"] += [
							    word for word in madeup_words if word not in kwargs["additional_special_tokens"]
							]

							super().__init__(
							    UpperCAmelCase_		,	tokenizer_file=UpperCAmelCase_		,	bos_token=UpperCAmelCase_		,	eos_token=UpperCAmelCase_		,	sep_token=UpperCAmelCase_		,	cls_token=UpperCAmelCase_		,	unk_token=UpperCAmelCase_		,	pad_token=UpperCAmelCase_		,	**UpperCAmelCase_		,	)

							__UpperCAmelCase	:							Dict     =     vocab_file
							__UpperCAmelCase	:							Optional[int]     =     False if not self.vocab_file else True
					def 		lowerCamelCase_   (   self       :      int		,	UpperCAmelCase_       :      List[int]		,	UpperCAmelCase_       :      Optional[List[int]] = None		):







							"""simple docstring"""

							if token_ids_a is None:
									return [self.sep_token_id] + token_ids_a
							__UpperCAmelCase	:							Any     =     [self.sep_token_id]
							return sep + token_ids_a + sep + sep + token_ids_a
					def 		lowerCamelCase_   (   self       :      Optional[Any]		,	UpperCAmelCase_       :      List[int]		,	UpperCAmelCase_       :      Optional[List[int]] = None		):







							"""simple docstring"""
							__UpperCAmelCase	:							List[Any]     =     [self.sep_token_id]

							if token_ids_a is None:
									return len(sep + token_ids_a		) * [0]
							return len(sep + token_ids_a + sep + sep + token_ids_a		) * [0]






					def 		lowerCamelCase_   (   self       :      List[Any]		,	UpperCAmelCase_       :      str		,	UpperCAmelCase_       :      Optional[str] = None		):







							"""simple docstring"""
							if not self.can_save_slow_tokenizer:
									raise ValueError(
									    "Your fast tokenizer does not have the necessary information to save the vocabulary for a slow "
									    "tokenizer."		)

							if not os.path.isdir(UpperCAmelCase_		):
									logger.error(f"Vocabulary path ({save_directory}) should be a directory."		)
									return
							__UpperCAmelCase	:							str     =     os.path.join(
							    UpperCAmelCase_		,	(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]		)

							if os.path.abspath(self.vocab_file		) != os.path.abspath(UpperCAmelCase_		):
									copyfile(self.vocab_file		,	UpperCAmelCase_		)

							return (out_vocab_file,)




