


import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyImgaImgPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()

class 							lowercase  (						_UpperCAmelCase    ,       unittest.TestCase			):
     lowerCamelCase			:			Optional[Any]      = KandinskyImgaImgPipeline
     lowerCamelCase			:			Union[str, Any]      = ['''prompt''', '''image_embeds''', '''negative_image_embeds''', '''image''']
     lowerCamelCase			:			Optional[Any]      = [
         '''prompt''',
         '''negative_prompt''',
         '''image_embeds''',
         '''negative_image_embeds''',
         '''image''',
     ]
     lowerCamelCase			:			List[Any]      = [
         '''generator''',
         '''height''',
         '''width''',
         '''strength''',
         '''guidance_scale''',
         '''negative_prompt''',
         '''num_inference_steps''',
         '''return_dict''',
         '''guidance_scale''',
         '''num_images_per_prompt''',
         '''output_type''',
         '''return_dict''',
     ]
     lowerCamelCase			:			int      = False







     @property
     def 							lowercase__       (						self :						Any      ):
           return 32







     @property
     def 							lowercase__       (						self :						Dict      ):
           return 32







     @property
     def 							lowercase__       (						self :						int      ):
           return self.time_input_dim







     @property
     def 							lowercase__       (						self :						Optional[int]      ):
           return self.time_input_dim * 4







     @property
     def 							lowercase__       (						self :						List[str]      ):
           return 1_00







     @property
     def 							lowercase__       (						self :						Union[str, Any]      ):
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		XLMRobertaTokenizerFast.from_pretrained('''YiYiXu/tiny-random-mclip-base'''      )
           return tokenizer







     @property
     def 							lowercase__       (						self :						Dict      ):
           torch.manual_seed(0      )
           SCREAMING_SNAKE_CASE__       :  str						     =		MCLIPConfig(
               numDims=self.cross_attention_dim   ,   transformerDimensions=self.text_embedder_hidden_size   ,   hidden_size=self.text_embedder_hidden_size   ,   intermediate_size=37   ,   num_attention_heads=4   ,   num_hidden_layers=5   ,   vocab_size=10_05   ,   )

           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		MultilingualCLIP(_lowercase      )
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		text_encoder.eval()

           return text_encoder







     @property
     def 							lowercase__       (						self :						int      ):
           torch.manual_seed(0      )

           SCREAMING_SNAKE_CASE__       :  Tuple						     =		{
               '''in_channels''': 4,
               # Out channels is double in channels because predicts mean and variance
               '''out_channels''': 8,
               '''addition_embed_type''': '''text_image''',
               '''down_block_types''': ('''ResnetDownsampleBlock2D''', '''SimpleCrossAttnDownBlock2D'''),
               '''up_block_types''': ('''SimpleCrossAttnUpBlock2D''', '''ResnetUpsampleBlock2D'''),
               '''mid_block_type''': '''UNetMidBlock2DSimpleCrossAttn''',
               '''block_out_channels''': (self.block_out_channels_a, self.block_out_channels_a * 2),
               '''layers_per_block''': 1,
               '''encoder_hid_dim''': self.text_embedder_hidden_size,
               '''encoder_hid_dim_type''': '''text_image_proj''',
               '''cross_attention_dim''': self.cross_attention_dim,
               '''attention_head_dim''': 4,
               '''resnet_time_scale_shift''': '''scale_shift''',
               '''class_embed_type''': None,
           }

           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		UNetaDConditionModel(**_lowercase      )
           return model







     @property
     def 							lowercase__       (						self :						List[str]      ):
           return {
               "block_out_channels": [32, 64],
               "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
               "in_channels": 3,
               "latent_channels": 4,
               "layers_per_block": 1,
               "norm_num_groups": 8,
               "norm_type": "spatial",
               "num_vq_embeddings": 12,
               "out_channels": 3,
               "up_block_types": [
                   "AttnUpDecoderBlock2D",
                   "UpDecoderBlock2D",
               ],
               "vq_embed_dim": 4,
           }







     @property
     def 							lowercase__       (						self :						Union[str, Any]      ):
           torch.manual_seed(0      )
           SCREAMING_SNAKE_CASE__       :  List[str]						     =		VQModel(**self.dummy_movq_kwargs      )
           return model







     def 							lowercase__       (						self :						int      ):
           SCREAMING_SNAKE_CASE__       :  Dict						     =		self.dummy_text_encoder
           SCREAMING_SNAKE_CASE__       :  str						     =		self.dummy_tokenizer
           SCREAMING_SNAKE_CASE__       :  List[str]						     =		self.dummy_unet
           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		self.dummy_movq

           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		{
               '''num_train_timesteps''': 10_00,
               '''beta_schedule''': '''linear''',
               '''beta_start''': 0.00085,
               '''beta_end''': 0.012,
               '''clip_sample''': False,
               '''set_alpha_to_one''': False,
               '''steps_offset''': 0,
               '''prediction_type''': '''epsilon''',
               '''thresholding''': False,
           }

           SCREAMING_SNAKE_CASE__       :  Any						     =		DDIMScheduler(**_lowercase      )

           SCREAMING_SNAKE_CASE__       :  str						     =		{
               '''text_encoder''': text_encoder,
               '''tokenizer''': tokenizer,
               '''unet''': unet,
               '''scheduler''': scheduler,
               '''movq''': movq,
           }

           return components







     def 							lowercase__       (						self :						Union[str, Any]   ,   _lowercase :						Union[str, Any]   ,   _lowercase :						Optional[int]=0      ):
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		floats_tensor((1, self.cross_attention_dim)   ,   rng=random.Random(_lowercase      )      ).to(_lowercase      )
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		floats_tensor((1, self.cross_attention_dim)   ,   rng=random.Random(seed + 1      )      ).to(_lowercase      )
           # create init_image
           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		floats_tensor((1, 3, 64, 64)   ,   rng=random.Random(_lowercase      )      ).to(_lowercase      )
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		image.cpu().permute(0   ,   2   ,   3   ,   1      )[0]
           SCREAMING_SNAKE_CASE__       :  Any						     =		Image.fromarray(np.uinta(_lowercase      )      ).convert('''RGB'''      ).resize((2_56, 2_56)      )

           if str(_lowercase      ).startswith('''mps'''      ):
                 SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		torch.manual_seed(_lowercase      )
           else:
                 SCREAMING_SNAKE_CASE__       :  int						     =		torch.Generator(device=_lowercase      ).manual_seed(_lowercase      )
           SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		{
               '''prompt''': '''horse''',
               '''image''': init_image,
               '''image_embeds''': image_embeds,
               '''negative_image_embeds''': negative_image_embeds,
               '''generator''': generator,
               '''height''': 64,
               '''width''': 64,
               '''num_inference_steps''': 10,
               '''guidance_scale''': 7.0,
               '''strength''': 0.2,
               '''output_type''': '''np''',
           }
           return inputs







     def 							lowercase__       (						self :						List[str]      ):
           SCREAMING_SNAKE_CASE__       :  Dict						     =		'''cpu'''

           SCREAMING_SNAKE_CASE__       :  Any						     =		self.get_dummy_components()

           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		self.pipeline_class(**_lowercase      )
           SCREAMING_SNAKE_CASE__       :  int						     =		pipe.to(_lowercase      )

           pipe.set_progress_bar_config(disable=_lowercase      )

           SCREAMING_SNAKE_CASE__       :  Any						     =		pipe(**self.get_dummy_inputs(_lowercase      )      )
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		output.images

           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		pipe(
               **self.get_dummy_inputs(_lowercase      )   ,   return_dict=_lowercase   ,   )[0]

           SCREAMING_SNAKE_CASE__       :  Any						     =		image[0, -3:, -3:, -1]
           SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		image_from_tuple[0, -3:, -3:, -1]

           assert image.shape == (1, 64, 64, 3)

           SCREAMING_SNAKE_CASE__       :  Tuple						     =		np.array(
               [0.61474943, 0.6073539, 0.43308544, 0.5928269, 0.47493595, 0.46755973, 0.4613838, 0.45368797, 0.50119233]      )
           assert (
               np.abs(image_slice.flatten() - expected_slice      ).max() < 1E-2
           ), f""" expected_slice {expected_slice}, but got {image_slice.flatten()}"""
           assert (
               np.abs(image_from_tuple_slice.flatten() - expected_slice      ).max() < 1E-2
           ), f""" expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}"""




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







     def 							lowercase__       (						self :						Optional[int]      ):
           # clean up the VRAM after each test
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()







     def 							lowercase__       (						self :						Union[str, Any]      ):
           SCREAMING_SNAKE_CASE__       :  int						     =		load_numpy(
               '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
               '''/kandinsky/kandinsky_img2img_frog.npy'''      )

           SCREAMING_SNAKE_CASE__       :  int						     =		load_image(
               '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main''' '''/kandinsky/cat.png'''      )
           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		'''A red cartoon frog, 4k'''

           SCREAMING_SNAKE_CASE__       :  int						     =		KandinskyPriorPipeline.from_pretrained(
               '''kandinsky-community/kandinsky-2-1-prior'''   ,   torch_dtype=torch.floataa      )
           pipe_prior.to(_lowercase      )

           SCREAMING_SNAKE_CASE__       :  Dict						     =		KandinskyImgaImgPipeline.from_pretrained(
               '''kandinsky-community/kandinsky-2-1'''   ,   torch_dtype=torch.floataa      )
           SCREAMING_SNAKE_CASE__       :  Dict						     =		pipeline.to(_lowercase      )

           pipeline.set_progress_bar_config(disable=_lowercase      )

           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		torch.Generator(device='''cpu'''      ).manual_seed(0      )
           SCREAMING_SNAKE_CASE__     ,       SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		pipe_prior(
               _lowercase   ,   generator=_lowercase   ,   num_inference_steps=5   ,   negative_prompt=''''''   ,   ).to_tuple()

           SCREAMING_SNAKE_CASE__       :  Dict						     =		pipeline(
               _lowercase   ,   image=_lowercase   ,   image_embeds=_lowercase   ,   negative_image_embeds=_lowercase   ,   generator=_lowercase   ,   num_inference_steps=1_00   ,   height=7_68   ,   width=7_68   ,   strength=0.2   ,   output_type='''np'''   ,   )

           SCREAMING_SNAKE_CASE__       :  Dict						     =		output.images[0]

           assert image.shape == (7_68, 7_68, 3)

           assert_mean_pixel_difference(_lowercase   ,   _lowercase      )











from dataclasses import dataclass, field
from typing import Tuple

from ..utils import cached_property, is_tf_available, logging, requires_backends
from .benchmark_args_utils import BenchmarkArguments


if is_tf_available():
     import tensorflow as tf


a_    :  Any						  =		logging.get_logger(__name__)





@dataclass
class  lowerCamelCase__	(    UpperCAmelCase_):
     """simple docstring"""



     _A      =      [
         'no_inference',
         'no_cuda',
         'no_tpu',
         'no_speed',
         'no_memory',
         'no_env_print',
         'no_multi_process',
     ]







     def __init__(self					,      **__a     ):






           '''simple docstring'''



           for deprecated_arg in self.deprecated_args:
                 if deprecated_arg in kwargs:
                       lowerCamelCase             =		deprecated_arg[3:]
                       lowerCamelCase             =		not kwargs.pop(__a     )
                       logger.warning(
                           F"""{deprecated_arg} is depreciated. Please use --no-{positive_arg} or"""
                           F""" {positive_arg}={kwargs[positive_arg]}"""     )
           lowerCamelCase             =		kwargs.pop("tpu_name"					,      self.tpu_name     )
           lowerCamelCase             =		kwargs.pop("device_idx"					,      self.device_idx     )
           lowerCamelCase             =		kwargs.pop("eager_mode"					,      self.eager_mode     )
           lowerCamelCase             =		kwargs.pop("use_xla"					,      self.use_xla     )
           super().__init__(**__a     )

     _A      =      field(
         default=UpperCAmelCase_							,     metadata={'help': 'Name of TPU'}							,     )
     _A      =      field(
         default=0							,     metadata={'help': 'CPU / GPU device index. Defaults to 0.'}							,     )
     _A      =      field(default=UpperCAmelCase_							,     metadata={'help': 'Benchmark models in eager model.'})
     _A      =      field(
         default=UpperCAmelCase_							,     metadata={
             'help': 'Benchmark models using XLA JIT compilation. Note that `eager_model` has to be set to `False`.'
         }							,     )







     @cached_property
     def     _a   (self     ):






           '''simple docstring'''



           requires_backends(self					,      ["tf"]     )
           lowerCamelCase             =		None
           if self.tpu:
                 try:
                       if self.tpu_name:
                             lowerCamelCase             =		tf.distribute.cluster_resolver.TPUClusterResolver(self.tpu_name     )
                       else:
                             lowerCamelCase             =		tf.distribute.cluster_resolver.TPUClusterResolver()
                 except ValueError:
                       lowerCamelCase             =		None
           return tpu







     @cached_property
     def     _a   (self     ):






           '''simple docstring'''



           requires_backends(self					,      ["tf"]     )
           if self.is_tpu:
                 tf.config.experimental_connect_to_cluster(self._setup_tpu     )
                 tf.tpu.experimental.initialize_tpu_system(self._setup_tpu     )

                 lowerCamelCase             =		tf.distribute.TPUStrategy(self._setup_tpu     )
           else:
                 # currently no multi gpu is allowed
                 if self.is_gpu:
                       # TODO: Currently only single GPU is supported
                       tf.config.set_visible_devices(self.gpu_list[self.device_idx]					,      "GPU"     )
                       lowerCamelCase             =		tf.distribute.OneDeviceStrategy(device=F"""/gpu:{self.device_idx}"""     )
                 else:
                       tf.config.set_visible_devices([]					,      "GPU"     )  # disable GPU
                       lowerCamelCase             =		tf.distribute.OneDeviceStrategy(device=F"""/cpu:{self.device_idx}"""     )

           return strategy







     @property
     def     _a   (self     ):






           '''simple docstring'''



           requires_backends(self					,      ["tf"]     )
           return self._setup_tpu is not None







     @property
     def     _a   (self     ):






           '''simple docstring'''



           requires_backends(self					,      ["tf"]     )
           return self._setup_strategy







     @property
     def     _a   (self     ):






           '''simple docstring'''



           requires_backends(self					,      ["tf"]     )
           return tf.config.list_physical_devices("GPU"     )







     @property
     def     _a   (self     ):






           '''simple docstring'''



           requires_backends(self					,      ["tf"]     )
           if self.cuda:
                 return len(self.gpu_list     )
           return 0







     @property
     def     _a   (self     ):






           '''simple docstring'''



           return self.n_gpu > 0