




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



						@property
						def 	__a					(			self	:   List[Any]  ):

								"""simple docstring"""

								return 32



						@property
						def 	__a					(			self	:   Any  ):

								"""simple docstring"""

								return 32



						@property
						def 	__a					(			self	:   Optional[Any]  ):

								"""simple docstring"""

								return self.time_input_dim



						@property
						def 	__a					(			self	:   Optional[Any]  ):

								"""simple docstring"""

								return self.time_input_dim * 4



						@property
						def 	__a					(			self	:   int  ):

								"""simple docstring"""

								return 1_00



						@property
						def 	__a					(			self	:   Dict  ):

								"""simple docstring"""

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



						@property
						def 	__a					(			self	:   Optional[int]  ):

								"""simple docstring"""

								torch.manual_seed(0  )
								SCREAMING_SNAKE_CASE_					      =							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_					      =							MultilingualCLIP(snake_case__  )
								SCREAMING_SNAKE_CASE_					      =							text_encoder.eval()

								return text_encoder



						@property
						def 	__a					(			self	:   int  ):

								"""simple docstring"""

								torch.manual_seed(0  )

								SCREAMING_SNAKE_CASE_					      =							{
								    '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_					      =							UNetaDConditionModel(**snake_case__  )
								return model



						@property
						def 	__a					(			self	:   Any  ):

								"""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 	__a					(			self	:   str  ):

								"""simple docstring"""

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



						def 	__a					(			self	:   str  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							self.dummy_text_encoder
								SCREAMING_SNAKE_CASE_					      =							self.dummy_tokenizer
								SCREAMING_SNAKE_CASE_					      =							self.dummy_unet
								SCREAMING_SNAKE_CASE_					      =							self.dummy_movq

								SCREAMING_SNAKE_CASE_					      =							{
								    'num_train_timesteps': 10_00,
								    '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,
								}

								SCREAMING_SNAKE_CASE_					      =							DDIMScheduler(**snake_case__  )

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

								return components



						def 	__a					(			self	:   List[str]					,							snake_case__	:   int					,							snake_case__	:   Optional[int]=0  ):

								"""simple docstring"""

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

								if str(snake_case__  ).startswith('mps'  ):
										SCREAMING_SNAKE_CASE_					      =							torch.manual_seed(snake_case__  )
								else:
										SCREAMING_SNAKE_CASE_					      =							torch.Generator(device=snake_case__  ).manual_seed(snake_case__  )
								SCREAMING_SNAKE_CASE_					      =							{
								    '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 	__a					(			self	:   str  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							'cpu'

								SCREAMING_SNAKE_CASE_					      =							self.get_dummy_components()

								SCREAMING_SNAKE_CASE_					      =							self.pipeline_class(**snake_case__  )
								SCREAMING_SNAKE_CASE_					      =							pipe.to(snake_case__  )

								pipe.set_progress_bar_config(disable=snake_case__  )

								SCREAMING_SNAKE_CASE_					      =							pipe(**self.get_dummy_inputs(snake_case__  )  )
								SCREAMING_SNAKE_CASE_					      =							output.images

								SCREAMING_SNAKE_CASE_					      =							pipe(
								    **self.get_dummy_inputs(snake_case__  )					,							return_dict=snake_case__					,							)[0]

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

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

								SCREAMING_SNAKE_CASE_					      =							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    lowercase_		(unittest.TestCase ):



						def 	__a					(			self	:   Tuple  ):

								"""simple docstring"""

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






						def 	__a					(			self	:   Union[str, Any]  ):

								"""simple docstring"""

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

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

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

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

								pipeline.set_progress_bar_config(disable=snake_case__  )

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

								SCREAMING_SNAKE_CASE_					      =							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'					,							)

								SCREAMING_SNAKE_CASE_					      =							output.images[0]

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

								assert_mean_pixel_difference(snake_case__					,							snake_case__  )





from __future__ import annotations

import inspect
import unittest

import numpy as np

from transformers import ResNetConfig
from transformers.testing_utils import require_tf, require_vision, slow
from transformers.utils import cached_property, is_tf_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
						import tensorflow as tf

						from transformers import TFResNetForImageClassification, TFResNetModel
						from transformers.models.resnet.modeling_tf_resnet import TF_RESNET_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
						from PIL import Image

						from transformers import AutoImageProcessor



class    lowercase_		:



						def __init__(			self	:   Optional[Any]					,							snake_case__	:   List[Any]					,							snake_case__	:   Tuple=3					,							snake_case__	:   Union[str, Any]=32					,							snake_case__	:   Union[str, Any]=3					,							snake_case__	:   str=10					,							snake_case__	:   List[Any]=[10, 20, 30, 40]					,							snake_case__	:   Optional[Any]=[1, 1, 2, 1]					,							snake_case__	:   str=True					,							snake_case__	:   str=True					,							snake_case__	:   List[str]="relu"					,							snake_case__	:   Tuple=3					,							snake_case__	:   str=None					,							):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							parent
								SCREAMING_SNAKE_CASE_					      =							batch_size
								SCREAMING_SNAKE_CASE_					      =							image_size
								SCREAMING_SNAKE_CASE_					      =							num_channels
								SCREAMING_SNAKE_CASE_					      =							embeddings_size
								SCREAMING_SNAKE_CASE_					      =							hidden_sizes
								SCREAMING_SNAKE_CASE_					      =							depths
								SCREAMING_SNAKE_CASE_					      =							is_training
								SCREAMING_SNAKE_CASE_					      =							use_labels
								SCREAMING_SNAKE_CASE_					      =							hidden_act
								SCREAMING_SNAKE_CASE_					      =							num_labels
								SCREAMING_SNAKE_CASE_					      =							scope
								SCREAMING_SNAKE_CASE_					      =							len(snake_case__  )



						def 	__a					(			self	:   Any  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]  )

								SCREAMING_SNAKE_CASE_					      =							None
								if self.use_labels:
										SCREAMING_SNAKE_CASE_					      =							ids_tensor([self.batch_size]					,							self.num_labels  )

								SCREAMING_SNAKE_CASE_					      =							self.get_config()

								return config, pixel_values, labels



						def 	__a					(			self	:   int  ):

								"""simple docstring"""

								return ResNetConfig(
								    num_channels=self.num_channels					,							embeddings_size=self.embeddings_size					,							hidden_sizes=self.hidden_sizes					,							depths=self.depths					,							hidden_act=self.hidden_act					,							num_labels=self.num_labels					,							image_size=self.image_size					,							)



						def 	__a					(			self	:   str					,							snake_case__	:   str					,							snake_case__	:   Tuple					,							snake_case__	:   Optional[Any]  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							TFResNetModel(config=snake_case__  )
								SCREAMING_SNAKE_CASE_					      =							model(snake_case__  )
								# expected last hidden states: B, C, H // 32, W // 32
								self.parent.assertEqual(
								    result.last_hidden_state.shape					,							(self.batch_size, self.hidden_sizes[-1], self.image_size // 32, self.image_size // 32)					,							)



						def 	__a					(			self	:   int					,							snake_case__	:   Optional[Any]					,							snake_case__	:   int					,							snake_case__	:   Any  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							self.num_labels
								SCREAMING_SNAKE_CASE_					      =							TFResNetForImageClassification(snake_case__  )
								SCREAMING_SNAKE_CASE_					      =							model(snake_case__					,							labels=snake_case__  )
								self.parent.assertEqual(result.logits.shape					,							(self.batch_size, self.num_labels)  )






						def 	__a					(			self	:   Dict  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							self.prepare_config_and_inputs()
								SCREAMING_SNAKE_CASE_      ,				SCREAMING_SNAKE_CASE_      ,				SCREAMING_SNAKE_CASE_					      =							config_and_inputs
								SCREAMING_SNAKE_CASE_					      =							{'pixel_values': pixel_values}
								return config, inputs_dict



@require_tf
class    lowercase_		(SCREAMING_SNAKE_CASE__							,	SCREAMING_SNAKE_CASE__							,	unittest.TestCase ):
						lowerCAmelCase__			      =(TFResNetModel, TFResNetForImageClassification) if is_tf_available() else ()
						lowerCAmelCase__			      =(
						    {"feature-extraction": TFResNetModel, "image-classification": TFResNetForImageClassification}
						    if is_tf_available()
						    else {}
						)

						lowerCAmelCase__			      =False
						lowerCAmelCase__			      =False
						lowerCAmelCase__			      =False
						lowerCAmelCase__			      =False
						lowerCAmelCase__			      =False



						def 	__a					(			self	:   str  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							TFResNetModelTester(self  )
								SCREAMING_SNAKE_CASE_					      =							ConfigTester(self					,							config_class=snake_case__					,							has_text_modality=snake_case__  )



						def 	__a					(			self	:   Optional[Any]  ):

								"""simple docstring"""

								self.create_and_test_config_common_properties()
								self.config_tester.create_and_test_config_to_json_string()
								self.config_tester.create_and_test_config_to_json_file()
								self.config_tester.create_and_test_config_from_and_save_pretrained()
								self.config_tester.create_and_test_config_with_num_labels()
								self.config_tester.check_config_can_be_init_without_params()
								self.config_tester.check_config_arguments_init()



						def 	__a					(			self	:   List[Any]  ):

								"""simple docstring"""

								return



						@unittest.skip(reason='ResNet does not use inputs_embeds'  )
						def 	__a					(			self	:   Optional[Any]  ):

								"""simple docstring"""

								pass



						@unittest.skip(reason='ResNet does not support input and output embeddings'  )
						def 	__a					(			self	:   Tuple  ):

								"""simple docstring"""

								pass



						def 	__a					(			self	:   List[Any]  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_      ,				SCREAMING_SNAKE_CASE_					      =							self.model_tester.prepare_config_and_inputs_for_common()

								for model_class in self.all_model_classes:
										SCREAMING_SNAKE_CASE_					      =							model_class(snake_case__  )
										SCREAMING_SNAKE_CASE_					      =							inspect.signature(model.call  )
										# signature.parameters is an OrderedDict => so arg_names order is deterministic
										SCREAMING_SNAKE_CASE_					      =							[*signature.parameters.keys()]

										SCREAMING_SNAKE_CASE_					      =							['pixel_values']
										self.assertListEqual(arg_names[:1]					,							snake_case__  )



						def 	__a					(			self	:   Union[str, Any]  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_model(*snake_case__  )



						def 	__a					(			self	:   int  ):

								"""simple docstring"""

								def check_hidden_states_output(snake_case__	:   Optional[int]					,							snake_case__	:   int					,							snake_case__	:   Dict  ):
										SCREAMING_SNAKE_CASE_					      =							model_class(snake_case__  )
										SCREAMING_SNAKE_CASE_					      =							model(**self._prepare_for_class(snake_case__					,							snake_case__  )  )

										SCREAMING_SNAKE_CASE_					      =							outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

										SCREAMING_SNAKE_CASE_					      =							self.model_tester.num_stages
										self.assertEqual(len(snake_case__  )					,							expected_num_stages + 1  )

										# ResNet's feature maps are of shape (batch_size, num_channels, height, width)
										self.assertListEqual(
										    list(hidden_states[0].shape[-2:]  )					,							[self.model_tester.image_size // 4, self.model_tester.image_size // 4]					,							)

								SCREAMING_SNAKE_CASE_      ,				SCREAMING_SNAKE_CASE_					      =							self.model_tester.prepare_config_and_inputs_for_common()
								SCREAMING_SNAKE_CASE_					      =							['basic', 'bottleneck']
								for model_class in self.all_model_classes:
										for layer_type in layers_type:
												SCREAMING_SNAKE_CASE_					      =							layer_type
												SCREAMING_SNAKE_CASE_					      =							True
												check_hidden_states_output(snake_case__					,							snake_case__					,							snake_case__  )

												# check that output_hidden_states also work using config
												del inputs_dict["output_hidden_states"]
												SCREAMING_SNAKE_CASE_					      =							True

												check_hidden_states_output(snake_case__					,							snake_case__					,							snake_case__  )



						def 	__a					(			self	:   List[Any]  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_image_classification(*snake_case__  )






						@slow
						def 	__a					(			self	:   Tuple  ):

								"""simple docstring"""

								for model_name in TF_RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
										SCREAMING_SNAKE_CASE_					      =							TFResNetModel.from_pretrained(snake_case__  )
										self.assertIsNotNone(snake_case__  )


def 					_a							(					)->	Tuple:
		SCREAMING_SNAKE_CASE_					      =							Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png' )
		return image




@require_tf
@require_vision
class    lowercase_		(unittest.TestCase ):



						@cached_property
						def 	__a					(			self	:   List[Any]  ):

								"""simple docstring"""

								return (
								    AutoImageProcessor.from_pretrained(TF_RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[0]  )
								    if is_vision_available()
								    else None
								)






						@slow
						def 	__a					(			self	:   Any  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							TFResNetForImageClassification.from_pretrained(TF_RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[0]  )

								SCREAMING_SNAKE_CASE_					      =							self.default_image_processor
								SCREAMING_SNAKE_CASE_					      =							prepare_img()
								SCREAMING_SNAKE_CASE_					      =							image_processor(images=snake_case__					,							return_tensors='tf'  )

								# forward pass
								SCREAMING_SNAKE_CASE_					      =							model(**snake_case__  )

								# verify the logits
								SCREAMING_SNAKE_CASE_					      =							tf.TensorShape((1, 10_00)  )
								self.assertEqual(outputs.logits.shape					,							snake_case__  )

								SCREAMING_SNAKE_CASE_					      =							tf.constant([-11.10_69, -9.78_77, -8.37_77]  )

								self.assertTrue(np.allclose(outputs.logits[0, :3].numpy()					,							snake_case__					,							atol=1e-4  )  )