



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    (			lowerCAmelCase_				,						unittest.TestCase						):
			UpperCamelCase   					=     KandinskyImgaImgPipeline
			UpperCamelCase   					=     ['''prompt''', '''image_embeds''', '''negative_image_embeds''', '''image''']
			UpperCamelCase   					=     [
			    '''prompt''',
			    '''negative_prompt''',
			    '''image_embeds''',
			    '''negative_image_embeds''',
			    '''image''',
			]
			UpperCamelCase   					=     [
			    '''generator''',
			    '''height''',
			    '''width''',
			    '''strength''',
			    '''guidance_scale''',
			    '''negative_prompt''',
			    '''num_inference_steps''',
			    '''return_dict''',
			    '''guidance_scale''',
			    '''num_images_per_prompt''',
			    '''output_type''',
			    '''return_dict''',
			]
			UpperCamelCase   					=     False



			@property
			def     _lowerCamelCase       (						self   :  Dict)   ->    Optional[Any]:



						"""simple docstring"""






						return 32



			@property
			def     _lowerCamelCase       (						self   :  Dict)   ->    int:



						"""simple docstring"""






						return 32



			@property
			def     _lowerCamelCase       (						self   :  List[str])   ->    Tuple:



						"""simple docstring"""






						return self.time_input_dim



			@property
			def     _lowerCamelCase       (						self   :  str)   ->    Optional[Any]:



						"""simple docstring"""






						return self.time_input_dim * 4



			@property
			def     _lowerCamelCase       (						self   :  str)   ->    Optional[Any]:



						"""simple docstring"""






						return 1_00



			@property
			def     _lowerCamelCase       (						self   :  Optional[Any])   ->    List[Any]:



						"""simple docstring"""






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



			@property
			def     _lowerCamelCase       (						self   :  Optional[int])   ->    Tuple:



						"""simple docstring"""






						torch.manual_seed(0)
						_UpperCAmelCase		       =   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 ,		)

						_UpperCAmelCase		       =   MultilingualCLIP(_snake_case)
						_UpperCAmelCase		       =   text_encoder.eval()

						return text_encoder



			@property
			def     _lowerCamelCase       (						self   :  Optional[int])   ->    Dict:



						"""simple docstring"""






						torch.manual_seed(0)

						_UpperCAmelCase		       =   {
						    '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,
						}

						_UpperCAmelCase		       =   UNetaDConditionModel(**_snake_case)
						return model



			@property
			def     _lowerCamelCase       (						self   :  Optional[Any])   ->    int:



						"""simple docstring"""






						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     _lowerCamelCase       (						self   :  Tuple)   ->    List[str]:



						"""simple docstring"""






						torch.manual_seed(0)
						_UpperCAmelCase		       =   VQModel(**self.dummy_movq_kwargs)
						return model



			def     _lowerCamelCase       (						self   :  Union[str, Any])   ->    Optional[int]:



						"""simple docstring"""






						_UpperCAmelCase		       =   self.dummy_text_encoder
						_UpperCAmelCase		       =   self.dummy_tokenizer
						_UpperCAmelCase		       =   self.dummy_unet
						_UpperCAmelCase		       =   self.dummy_movq

						_UpperCAmelCase		       =   {
						    'num_train_timesteps': 10_00,
						    'beta_schedule': 'linear',
						    'beta_start': 0.0_0_0_8_5,
						    'beta_end': 0.0_1_2,
						    'clip_sample': False,
						    'set_alpha_to_one': False,
						    'steps_offset': 0,
						    'prediction_type': 'epsilon',
						    'thresholding': False,
						}

						_UpperCAmelCase		       =   DDIMScheduler(**_snake_case)

						_UpperCAmelCase		       =   {
						    'text_encoder': text_encoder,
						    'tokenizer': tokenizer,
						    'unet': unet,
						    'scheduler': scheduler,
						    'movq': movq,
						}

						return components



			def     _lowerCamelCase       (						self   :  int ,		A   :  Any ,		A   :  Dict=0)   ->    Dict:



						"""simple docstring"""






						_UpperCAmelCase		       =   floats_tensor((1, self.cross_attention_dim) ,		rng=random.Random(_snake_case)).to(_snake_case)
						_UpperCAmelCase		       =   floats_tensor((1, self.cross_attention_dim) ,		rng=random.Random(seed + 1)).to(_snake_case)
						# create init_image
						_UpperCAmelCase		       =   floats_tensor((1, 3, 64, 64) ,		rng=random.Random(_snake_case)).to(_snake_case)
						_UpperCAmelCase		       =   image.cpu().permute(0 ,		2 ,		3 ,		1)[0]
						_UpperCAmelCase		       =   Image.fromarray(np.uinta(_snake_case)).convert('RGB').resize((2_56, 2_56))

						if str(_snake_case).startswith('mps'):
									_UpperCAmelCase		       =   torch.manual_seed(_snake_case)
						else:
									_UpperCAmelCase		       =   torch.Generator(device=_snake_case).manual_seed(_snake_case)
						_UpperCAmelCase		       =   {
						    '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     _lowerCamelCase       (						self   :  Optional[int])   ->    Dict:



						"""simple docstring"""






						_UpperCAmelCase		       =   'cpu'

						_UpperCAmelCase		       =   self.get_dummy_components()

						_UpperCAmelCase		       =   self.pipeline_class(**_snake_case)
						_UpperCAmelCase		       =   pipe.to(_snake_case)

						pipe.set_progress_bar_config(disable=_snake_case)

						_UpperCAmelCase		       =   pipe(**self.get_dummy_inputs(_snake_case))
						_UpperCAmelCase		       =   output.images

						_UpperCAmelCase		       =   pipe(
						    **self.get_dummy_inputs(_snake_case) ,		return_dict=_snake_case ,		)[0]

						_UpperCAmelCase		       =   image[0, -3:, -3:, -1]
						_UpperCAmelCase		       =   image_from_tuple[0, -3:, -3:, -1]

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

						_UpperCAmelCase		       =   np.array(
						    [0.6_1_4_7_4_9_4_3, 0.6_0_7_3_5_3_9, 0.4_3_3_0_8_5_4_4, 0.5_9_2_8_2_6_9, 0.4_7_4_9_3_5_9_5, 0.4_6_7_5_5_9_7_3, 0.4_6_1_3_8_3_8, 0.4_5_3_6_8_7_9_7, 0.5_0_1_1_9_2_3_3])
						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						):



			def     _lowerCamelCase       (						self   :  List[Any])   ->    Any:



						"""simple docstring"""






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



			def     _lowerCamelCase       (						self   :  Dict)   ->    List[str]:



						"""simple docstring"""






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

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

						_UpperCAmelCase		       =   KandinskyPriorPipeline.from_pretrained(
						    'kandinsky-community/kandinsky-2-1-prior' ,		torch_dtype=torch.floataa)
						pipe_prior.to(_snake_case)

						_UpperCAmelCase		       =   KandinskyImgaImgPipeline.from_pretrained(
						    'kandinsky-community/kandinsky-2-1' ,		torch_dtype=torch.floataa)
						_UpperCAmelCase		       =   pipeline.to(_snake_case)

						pipeline.set_progress_bar_config(disable=_snake_case)

						_UpperCAmelCase		       =   torch.Generator(device='cpu').manual_seed(0)
						_UpperCAmelCase		       =   pipe_prior(
						    _snake_case ,		generator=_snake_case ,		num_inference_steps=5 ,		negative_prompt='' ,		).to_tuple()

						_UpperCAmelCase		       =   pipeline(
						    _snake_case ,		image=_snake_case ,		image_embeds=_snake_case ,		negative_image_embeds=_snake_case ,		generator=_snake_case ,		num_inference_steps=1_00 ,		height=7_68 ,		width=7_68 ,		strength=0.2 ,		output_type='np' ,		)

						_UpperCAmelCase		       =   output.images[0]

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

						assert_mean_pixel_difference(_snake_case ,		_snake_case)





# Copyright 2023 The HuggingFace Inc. team. 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.
from ..models.auto import AutoModelForSeqaSeqLM, AutoTokenizer
from .base import PipelineTool




class 					snake_case__						(       lowerCAmelCase_     ):






     """simple docstring"""


     _SCREAMING_SNAKE_CASE						=	"""philschmid/bart-large-cnn-samsum"""
     _SCREAMING_SNAKE_CASE						=	(
         """This is a tool that summarizes an English text. It takes an input `text` containing the text to summarize, """
         """and returns a summary of the text."""
     )
     _SCREAMING_SNAKE_CASE						=	"""summarizer"""
     _SCREAMING_SNAKE_CASE						=	AutoTokenizer
     _SCREAMING_SNAKE_CASE						=	AutoModelForSeqaSeqLM

     _SCREAMING_SNAKE_CASE						=	["""text"""]
     _SCREAMING_SNAKE_CASE						=	["""text"""]






     def    lowercase_   (       self  :  Optional[Any], _snake_case  :  str					)   ->Any:
            return self.pre_processor(_snake_case, return_tensors='pt', truncation=_snake_case					)






     def    lowercase_   (       self  :  int, _snake_case  :  List[Any]					)   ->Any:
            return self.model.generate(**_snake_case					)[0]






     def    lowercase_   (       self  :  int, _snake_case  :  int					)   ->str:
            return self.pre_processor.decode(_snake_case, skip_special_tokens=_snake_case, clean_up_tokenization_spaces=_snake_case					)






