





"""simple docstring"""





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 					_lowerCAmelCase   (     a  ,   unittest.TestCase       ):
       """simple docstring"""




       __magic_name__     :str		  =					KandinskyImgaImgPipeline
       __magic_name__     :List[Any]		  =					["""prompt""", """image_embeds""", """negative_image_embeds""", """image"""]
       __magic_name__     :Optional[Any]		  =					[
           """prompt""",
           """negative_prompt""",
           """image_embeds""",
           """negative_image_embeds""",
           """image""",
       ]
       __magic_name__     :Any		  =					[
           """generator""",
           """height""",
           """width""",
           """strength""",
           """guidance_scale""",
           """negative_prompt""",
           """num_inference_steps""",
           """return_dict""",
           """guidance_scale""",
           """num_images_per_prompt""",
           """output_type""",
           """return_dict""",
       ]
       __magic_name__     :List[str]		  =					False






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         return 3_2






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         return 3_2






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         return self.time_input_dim






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         return self.time_input_dim * 4






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         return 1_0_0






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:Tuple											=							XLMRobertaTokenizerFast.from_pretrained('YiYiXu/tiny-random-mclip-base'      )
         return tokenizer






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         torch.manual_seed(0      )
         lowerCAmelCase__			:Any											=							MCLIPConfig(
             numDims=self.cross_attention_dim    , transformerDimensions=self.text_embedder_hidden_size    , hidden_size=self.text_embedder_hidden_size    , intermediate_size=3_7    , num_attention_heads=4    , num_hidden_layers=5    , vocab_size=1_0_0_5    , )

         lowerCAmelCase__			:Tuple											=							MultilingualCLIP(__UpperCAmelCase      )
         lowerCAmelCase__			:Dict											=							text_encoder.eval()

         return text_encoder






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         torch.manual_seed(0      )

         lowerCAmelCase__			:str											=							{
             '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,
         }

         lowerCAmelCase__			:int											=							UNetaDConditionModel(**__UpperCAmelCase      )
         return model






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         return {
             "block_out_channels": [3_2, 6_4],
             "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": 1_2,
             "out_channels": 3,
             "up_block_types": [
                 "AttnUpDecoderBlock2D",
                 "UpDecoderBlock2D",
             ],
             "vq_embed_dim": 4,
         }






       @property
       def      snake_case						(    self      ):



         '''simple docstring'''

         torch.manual_seed(0      )
         lowerCAmelCase__			:int											=							VQModel(**self.dummy_movq_kwargs      )
         return model






       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:int											=							self.dummy_text_encoder
         lowerCAmelCase__			:Any											=							self.dummy_tokenizer
         lowerCAmelCase__			:Optional[Any]											=							self.dummy_unet
         lowerCAmelCase__			:List[Any]											=							self.dummy_movq

         lowerCAmelCase__			:List[Any]											=							{
             'num_train_timesteps': 1_0_0_0,
             'beta_schedule': 'linear',
             'beta_start': 0.0_00_85,
             'beta_end': 0.0_12,
             'clip_sample': False,
             'set_alpha_to_one': False,
             'steps_offset': 0,
             'prediction_type': 'epsilon',
             'thresholding': False,
         }

         lowerCAmelCase__			:List[str]											=							DDIMScheduler(**__UpperCAmelCase      )

         lowerCAmelCase__			:Any											=							{
             'text_encoder': text_encoder,
             'tokenizer': tokenizer,
             'unet': unet,
             'scheduler': scheduler,
             'movq': movq,
         }

         return components






       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase=0      ):



         '''simple docstring'''

         lowerCAmelCase__			:List[str]											=							floats_tensor((1, self.cross_attention_dim)    , rng=random.Random(__UpperCAmelCase      )      ).to(__UpperCAmelCase      )
         lowerCAmelCase__			:str											=							floats_tensor((1, self.cross_attention_dim)    , rng=random.Random(seed + 1      )      ).to(__UpperCAmelCase      )
         # create init_image
         lowerCAmelCase__			:str											=							floats_tensor((1, 3, 6_4, 6_4)    , rng=random.Random(__UpperCAmelCase      )      ).to(__UpperCAmelCase      )
         lowerCAmelCase__			:Optional[Any]											=							image.cpu().permute(0    , 2    , 3    , 1      )[0]
         lowerCAmelCase__			:Tuple											=							Image.fromarray(np.uinta(__UpperCAmelCase      )      ).convert('RGB'      ).resize((2_5_6, 2_5_6)      )

         if str(__UpperCAmelCase      ).startswith('mps'      ):
           lowerCAmelCase__			:Tuple											=							torch.manual_seed(__UpperCAmelCase      )
         else:
           lowerCAmelCase__			:int											=							torch.Generator(device=__UpperCAmelCase      ).manual_seed(__UpperCAmelCase      )
         lowerCAmelCase__			:Dict											=							{
             'prompt': 'horse',
             'image': init_image,
             'image_embeds': image_embeds,
             'negative_image_embeds': negative_image_embeds,
             'generator': generator,
             'height': 6_4,
             'width': 6_4,
             'num_inference_steps': 1_0,
             'guidance_scale': 7.0,
             'strength': 0.2,
             'output_type': 'np',
         }
         return inputs






       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:str											=							'cpu'

         lowerCAmelCase__			:Dict											=							self.get_dummy_components()

         lowerCAmelCase__			:List[Any]											=							self.pipeline_class(**__UpperCAmelCase      )
         lowerCAmelCase__			:List[Any]											=							pipe.to(__UpperCAmelCase      )

         pipe.set_progress_bar_config(disable=__UpperCAmelCase      )

         lowerCAmelCase__			:Dict											=							pipe(**self.get_dummy_inputs(__UpperCAmelCase      )      )
         lowerCAmelCase__			:List[str]											=							output.images

         lowerCAmelCase__			:Any											=							pipe(
             **self.get_dummy_inputs(__UpperCAmelCase      )    , return_dict=__UpperCAmelCase    , )[0]

         lowerCAmelCase__			:List[Any]											=							image[0, -3:, -3:, -1]
         lowerCAmelCase__			:List[str]											=							image_from_tuple[0, -3:, -3:, -1]

         assert image.shape == (1, 6_4, 6_4, 3)

         lowerCAmelCase__			:Union[str, Any]											=							np.array(
             [0.61_47_49_43, 0.6_07_35_39, 0.43_30_85_44, 0.5_92_82_69, 0.47_49_35_95, 0.46_75_59_73, 0.4_61_38_38, 0.45_36_87_97, 0.50_11_92_33]      )
         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 					_lowerCAmelCase   (     unittest.TestCase       ):
       """simple docstring"""




       def      snake_case						(    self      ):



         '''simple docstring'''

         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()






       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:Any											=							load_numpy(
             'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
             '/kandinsky/kandinsky_img2img_frog.npy'      )

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

         lowerCAmelCase__			:Any											=							KandinskyPriorPipeline.from_pretrained(
             'kandinsky-community/kandinsky-2-1-prior'    , torch_dtype=torch.floataa      )
         pipe_prior.to(__UpperCAmelCase      )

         lowerCAmelCase__			:Dict											=							KandinskyImgaImgPipeline.from_pretrained(
             'kandinsky-community/kandinsky-2-1'    , torch_dtype=torch.floataa      )
         lowerCAmelCase__			:Optional[int]											=							pipeline.to(__UpperCAmelCase      )

         pipeline.set_progress_bar_config(disable=__UpperCAmelCase      )

         lowerCAmelCase__			:str											=							torch.Generator(device='cpu'      ).manual_seed(0      )
         lowerCAmelCase__							,				lowerCAmelCase__			:Any											=							pipe_prior(
             __UpperCAmelCase    , generator=__UpperCAmelCase    , num_inference_steps=5    , negative_prompt=''    , ).to_tuple()

         lowerCAmelCase__			:Union[str, Any]											=							pipeline(
             __UpperCAmelCase    , image=__UpperCAmelCase    , image_embeds=__UpperCAmelCase    , negative_image_embeds=__UpperCAmelCase    , generator=__UpperCAmelCase    , num_inference_steps=1_0_0    , height=7_6_8    , width=7_6_8    , strength=0.2    , output_type='np'    , )

         lowerCAmelCase__			:List[Any]											=							output.images[0]

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

         assert_mean_pixel_difference(__UpperCAmelCase    , __UpperCAmelCase      )







"""simple docstring"""





from collections.abc import Iterator, MutableMapping
from dataclasses import dataclass
from typing import Generic, TypeVar

__A											=		TypeVar("""KEY""")
__A											=		TypeVar("""VAL""")






@dataclass(frozen=a  ,   slots=a       )
class 					_lowerCAmelCase   (     Generic[KEY, VAL]       ):
       """simple docstring"""




       __magic_name__     :KEY
       __magic_name__     :VAL






class 					_lowerCAmelCase   (     _Item       ):
       """simple docstring"""




       def __init__(    self      ):



         '''simple docstring'''

         super().__init__(__UpperCAmelCase    , __UpperCAmelCase      )






       def __bool__(    self      ):



         '''simple docstring'''

         return False






__A											=		_DeletedItem()






class 					_lowerCAmelCase   (     MutableMapping[KEY, VAL]       ):
       """simple docstring"""




       def __init__(    self    , __UpperCAmelCase = 8    , __UpperCAmelCase = 0.75      ):



         '''simple docstring'''

         lowerCAmelCase__			:List[str]											=							initial_block_size
         lowerCAmelCase__			:list[_Item | None]											=							[None] * initial_block_size
         assert 0.0 < capacity_factor < 1.0
         lowerCAmelCase__			:Tuple											=							capacity_factor
         lowerCAmelCase__			:str											=							0






       def      snake_case						(    self    , __UpperCAmelCase      ):



         '''simple docstring'''

         return hash(__UpperCAmelCase      ) % len(self._buckets      )






       def      snake_case						(    self    , __UpperCAmelCase      ):



         '''simple docstring'''

         return (ind + 1) % len(self._buckets      )






       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase      ):



         '''simple docstring'''

         lowerCAmelCase__			:Any											=							self._buckets[ind]
         if not stored:
           lowerCAmelCase__			:Dict											=							_Item(__UpperCAmelCase    , __UpperCAmelCase      )
           self._len += 1
           return True
         elif stored.key == key:
           lowerCAmelCase__			:Optional[Any]											=							_Item(__UpperCAmelCase    , __UpperCAmelCase      )
           return True
         else:
           return False






       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:int											=							len(self._buckets      ) * self._capacity_factor
         return len(self      ) >= int(__UpperCAmelCase      )






       def      snake_case						(    self      ):



         '''simple docstring'''

         if len(self._buckets      ) <= self._initial_block_size:
           return False
         lowerCAmelCase__			:Optional[Any]											=							len(self._buckets      ) * self._capacity_factor / 2
         return len(self      ) < limit






       def      snake_case						(    self    , __UpperCAmelCase      ):



         '''simple docstring'''

         lowerCAmelCase__			:Optional[int]											=							self._buckets
         lowerCAmelCase__			:Tuple											=							[None] * new_size
         lowerCAmelCase__			:List[Any]											=							0
         for item in old_buckets:
           if item:
             self._add_item(item.key    , item.val      )






       def      snake_case						(    self      ):



         '''simple docstring'''

         self._resize(len(self._buckets      ) * 2      )






       def      snake_case						(    self      ):



         '''simple docstring'''

         self._resize(len(self._buckets      ) // 2      )






       def      snake_case						(    self    , __UpperCAmelCase      ):



         '''simple docstring'''

         lowerCAmelCase__			:Optional[Any]											=							self._get_bucket_index(__UpperCAmelCase      )
         for _ in range(len(self._buckets      )      ):
           yield ind
           lowerCAmelCase__			:Tuple											=							self._get_next_ind(__UpperCAmelCase      )






       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase      ):



         '''simple docstring'''

         for ind in self._iterate_buckets(__UpperCAmelCase      ):
           if self._try_set(__UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase      ):
             break






       def __setitem__(    self    , __UpperCAmelCase    , __UpperCAmelCase      ):



         '''simple docstring'''

         if self._is_full():
           self._size_up()

         self._add_item(__UpperCAmelCase    , __UpperCAmelCase      )






       def __delitem__(    self    , __UpperCAmelCase      ):



         '''simple docstring'''

         for ind in self._iterate_buckets(__UpperCAmelCase      ):
           lowerCAmelCase__			:int											=							self._buckets[ind]
           if item is None:
             raise KeyError(__UpperCAmelCase      )
           if item is _deleted:
             continue
           if item.key == key:
             lowerCAmelCase__			:List[str]											=							_deleted
             self._len -= 1
             break
         if self._is_sparse():
           self._size_down()






       def __getitem__(    self    , __UpperCAmelCase      ):



         '''simple docstring'''

         for ind in self._iterate_buckets(__UpperCAmelCase      ):
           lowerCAmelCase__			:str											=							self._buckets[ind]
           if item is None:
             break
           if item is _deleted:
             continue
           if item.key == key:
             return item.val
         raise KeyError(__UpperCAmelCase      )






       def __len__(    self      ):



         '''simple docstring'''

         return self._len






       def __iter__(    self      ):



         '''simple docstring'''

         yield from (item.key for item in self._buckets if item)






       def __repr__(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:Tuple											=							' ,'.join(
             F"{item.key}: {item.val}" for item in self._buckets if item      )
         return F"HashMap({val_string})"
