




import gc
import random
import unittest

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

from diffusers import DDIMScheduler, KandinskyInpaintPipeline, 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 (       __magic_name__			,				unittest.TestCase ):

			"""simple docstring"""







			snake_case_   		=       KandinskyInpaintPipeline
			snake_case_   		=       ['''prompt''', '''image_embeds''', '''negative_image_embeds''', '''image''', '''mask_image''']
			snake_case_   		=       [
			    '''prompt''',
			    '''negative_prompt''',
			    '''image_embeds''',
			    '''negative_image_embeds''',
			    '''image''',
			    '''mask_image''',
			]
			snake_case_   		=       [
			    '''generator''',
			    '''height''',
			    '''width''',
			    '''latents''',
			    '''guidance_scale''',
			    '''negative_prompt''',
			    '''num_inference_steps''',
			    '''return_dict''',
			    '''guidance_scale''',
			    '''num_images_per_prompt''',
			    '''output_type''',
			    '''return_dict''',
			]
			snake_case_   		=       False


			@property
			def       lowercase_    (      self							)    ->      Any:



							'''simple docstring'''
							return 32


			@property
			def       lowercase_    (      self							)    ->      str:



							'''simple docstring'''
							return 32


			@property
			def       lowercase_    (      self							)    ->      Any:



							'''simple docstring'''
							return self.time_input_dim


			@property
			def       lowercase_    (      self							)    ->      Tuple:



							'''simple docstring'''
							return self.time_input_dim * 4


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



							'''simple docstring'''
							return 100


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



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


			@property
			def       lowercase_    (      self							)    ->      Any:



							'''simple docstring'''
							torch.manual_seed(0							)
							__lowerCamelCase		       =    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=1_005			,						)

							__lowerCamelCase		       =    MultilingualCLIP(lowerCamelCase__							)
							__lowerCamelCase		       =    text_encoder.eval()

							return text_encoder


			@property
			def       lowercase_    (      self							)    ->      Any:



							'''simple docstring'''
							torch.manual_seed(0							)

							__lowerCamelCase		       =    {
							    'in_channels': 9,
							    # 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		       =    UNetaDConditionModel(**lowerCamelCase__							)
							return model


			@property
			def       lowercase_    (      self							)    ->      List[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       lowercase_    (      self							)    ->      Dict:



							'''simple docstring'''
							torch.manual_seed(0							)
							__lowerCamelCase		       =    VQModel(**self.dummy_movq_kwargs							)
							return model


			def       lowercase_    (      self							)    ->      Tuple:



							'''simple docstring'''
							__lowerCamelCase		       =    self.dummy_text_encoder
							__lowerCamelCase		       =    self.dummy_tokenizer
							__lowerCamelCase		       =    self.dummy_unet
							__lowerCamelCase		       =    self.dummy_movq

							__lowerCamelCase		       =    DDIMScheduler(
							    num_train_timesteps=1_000			,						beta_schedule='linear'			,						beta_start=0.0_00_85			,						beta_end=0.0_12			,						clip_sample=lowerCamelCase__			,						set_alpha_to_one=lowerCamelCase__			,						steps_offset=1			,						prediction_type='epsilon'			,						thresholding=lowerCamelCase__			,						)

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

							return components


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__=0							)    ->      List[str]:



							'''simple docstring'''
							__lowerCamelCase		       =    floats_tensor((1, self.cross_attention_dim)			,						rng=random.Random(lowerCamelCase__							)							).to(lowerCamelCase__							)
							__lowerCamelCase		       =    floats_tensor((1, self.cross_attention_dim)			,						rng=random.Random(seed + 1							)							).to(lowerCamelCase__							)
							# create init_image
							__lowerCamelCase		       =    floats_tensor((1, 3, 64, 64)			,						rng=random.Random(lowerCamelCase__							)							).to(lowerCamelCase__							)
							__lowerCamelCase		       =    image.cpu().permute(0			,						2			,						3			,						1							)[0]
							__lowerCamelCase		       =    Image.fromarray(np.uinta(lowerCamelCase__							)							).convert('RGB'							).resize((256, 256)							)
							# create mask
							__lowerCamelCase		       =    np.ones((64, 64)			,						dtype=np.floataa							)
							__lowerCamelCase		       =    0

							if str(lowerCamelCase__							).startswith('mps'							):
											__lowerCamelCase		       =    torch.manual_seed(lowerCamelCase__							)
							else:
											__lowerCamelCase		       =    torch.Generator(device=lowerCamelCase__							).manual_seed(lowerCamelCase__							)
							__lowerCamelCase		       =    {
							    'prompt': 'horse',
							    'image': init_image,
							    'mask_image': mask,
							    'image_embeds': image_embeds,
							    'negative_image_embeds': negative_image_embeds,
							    'generator': generator,
							    'height': 64,
							    'width': 64,
							    'num_inference_steps': 2,
							    'guidance_scale': 4.0,
							    'output_type': 'np',
							}
							return inputs


			def       lowercase_    (      self							)    ->      int:



							'''simple docstring'''
							__lowerCamelCase		       =    'cpu'

							__lowerCamelCase		       =    self.get_dummy_components()

							__lowerCamelCase		       =    self.pipeline_class(**lowerCamelCase__							)
							__lowerCamelCase		       =    pipe.to(lowerCamelCase__							)

							pipe.set_progress_bar_config(disable=lowerCamelCase__							)

							__lowerCamelCase		       =    pipe(**self.get_dummy_inputs(lowerCamelCase__							)							)
							__lowerCamelCase		       =    output.images

							__lowerCamelCase		       =    pipe(
							    **self.get_dummy_inputs(lowerCamelCase__							)			,						return_dict=lowerCamelCase__			,						)[0]

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

							print(f"""image.shape {image.shape}"""							)

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

							__lowerCamelCase		       =    np.array(
							    [0.8_32_69_19, 0.73_79_04_67, 0.20_91_85_81, 0.9_30_96_12, 0.5_51_17_91, 0.43_71_33_28, 0.5_51_33_21, 0.49_92_29_34, 0.59_49_77_86]							)

							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()}"""






			def       lowercase_    (      self							)    ->      Optional[int]:



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







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

			"""simple docstring"""







			def       lowercase_    (      self							)    ->      List[str]:



							'''simple docstring'''
							# clean up the VRAM after each test
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()






			def       lowercase_    (      self							)    ->      Dict:



							'''simple docstring'''
							__lowerCamelCase		       =    load_numpy(
							    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
							    '/kandinsky/kandinsky_inpaint_cat_with_hat_fp16.npy'							)

							__lowerCamelCase		       =    load_image(
							    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main' '/kandinsky/cat.png'							)
							__lowerCamelCase		       =    np.ones((768, 768)			,						dtype=np.floataa							)
							__lowerCamelCase		       =    0

							__lowerCamelCase		       =    'a hat'

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

							__lowerCamelCase		       =    KandinskyInpaintPipeline.from_pretrained(
							    'kandinsky-community/kandinsky-2-1-inpaint'			,						torch_dtype=torch.floataa							)
							__lowerCamelCase		       =    pipeline.to(lowerCamelCase__							)
							pipeline.set_progress_bar_config(disable=lowerCamelCase__							)

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

							__lowerCamelCase		       =    pipeline(
							    lowerCamelCase__			,						image=lowerCamelCase__			,						mask_image=lowerCamelCase__			,						image_embeds=lowerCamelCase__			,						negative_image_embeds=lowerCamelCase__			,						generator=lowerCamelCase__			,						num_inference_steps=100			,						height=768			,						width=768			,						output_type='np'			,						)

							__lowerCamelCase		       =    output.images[0]

							assert image.shape == (768, 768, 3)

							assert_mean_pixel_difference(lowerCamelCase__			,						lowerCamelCase__							)








"""simple docstring"""




import json
import os
import shutil
import tempfile
import unittest

import numpy as np

from transformers import BertTokenizerFast
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES, BertTokenizer
from transformers.testing_utils import require_tokenizers, require_vision
from transformers.utils import IMAGE_PROCESSOR_NAME, is_vision_available


if is_vision_available():
					from PIL import Image

					from transformers import VisionTextDualEncoderProcessor, ViTImageProcessor






@require_tokenizers
@require_vision
class     __magic_name__							(       unittest.TestCase				):






				'''simple docstring'''





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							lowerCamelCase			      =					tempfile.mkdtemp()

							# fmt: off
							lowerCamelCase			      =					["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """want""", """##want""", """##ed""", """wa""", """un""", """runn""", """##ing""", """,""", """low""", """lowest"""]
							# fmt: on
							lowerCamelCase			      =					os.path.join(self.tmpdirname		, VOCAB_FILES_NAMES["""vocab_file"""] )
							with open(self.vocab_file		, """w"""		, encoding="""utf-8""" ) as vocab_writer:
										vocab_writer.write("""""".join([x + """\n""" for x in vocab_tokens] ) )

							lowerCamelCase			      =					{
							    """do_resize""": True,
							    """size""": {"""height""": 18, """width""": 18},
							    """do_normalize""": True,
							    """image_mean""": [0.5, 0.5, 0.5],
							    """image_std""": [0.5, 0.5, 0.5],
							}
							lowerCamelCase			      =					os.path.join(self.tmpdirname		, _a )
							with open(self.image_processor_file		, """w"""		, encoding="""utf-8""" ) as fp:
										json.dump(_a		, _a )





				def    _lowerCAmelCase				(       self		, **_a ):





							"""simple docstring"""
							return BertTokenizer.from_pretrained(self.tmpdirname		, **_a )





				def    _lowerCAmelCase				(       self		, **_a ):





							"""simple docstring"""
							return ViTImageProcessor.from_pretrained(self.tmpdirname		, **_a )





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							shutil.rmtree(self.tmpdirname )





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							lowerCamelCase			      =					[np.random.randint(255		, size=(3, 30, 400)		, dtype=np.uinta )]

							lowerCamelCase			      =					[Image.fromarray(np.moveaxis(_a		, 0		, -1 ) ) for x in image_inputs]

							return image_inputs





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							lowerCamelCase			      =					self.get_tokenizer()
							lowerCamelCase			      =					self.get_image_processor()

							lowerCamelCase			      =					VisionTextDualEncoderProcessor(tokenizer=_a		, image_processor=_a )

							processor.save_pretrained(self.tmpdirname )
							lowerCamelCase			      =					VisionTextDualEncoderProcessor.from_pretrained(self.tmpdirname )

							self.assertEqual(processor.tokenizer.get_vocab()		, tokenizer.get_vocab() )
							self.assertIsInstance(processor.tokenizer		, (BertTokenizer, BertTokenizerFast) )

							self.assertEqual(processor.image_processor.to_json_string()		, image_processor.to_json_string() )
							self.assertIsInstance(processor.image_processor		, _a )





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							lowerCamelCase			      =					VisionTextDualEncoderProcessor(
							    tokenizer=self.get_tokenizer()		, image_processor=self.get_image_processor() )
							processor.save_pretrained(self.tmpdirname )

							lowerCamelCase			      =					self.get_tokenizer(bos_token="""(BOS)"""		, eos_token="""(EOS)""" )
							lowerCamelCase			      =					self.get_image_processor(do_normalize=_a		, padding_value=1.0 )

							lowerCamelCase			      =					VisionTextDualEncoderProcessor.from_pretrained(
							    self.tmpdirname		, bos_token="""(BOS)"""		, eos_token="""(EOS)"""		, do_normalize=_a		, padding_value=1.0 )

							self.assertEqual(processor.tokenizer.get_vocab()		, tokenizer_add_kwargs.get_vocab() )
							self.assertIsInstance(processor.tokenizer		, (BertTokenizer, BertTokenizerFast) )

							self.assertEqual(processor.image_processor.to_json_string()		, image_processor_add_kwargs.to_json_string() )
							self.assertIsInstance(processor.image_processor		, _a )





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							lowerCamelCase			      =					self.get_image_processor()
							lowerCamelCase			      =					self.get_tokenizer()

							lowerCamelCase			      =					VisionTextDualEncoderProcessor(tokenizer=_a		, image_processor=_a )

							lowerCamelCase			      =					self.prepare_image_inputs()

							lowerCamelCase			      =					image_processor(_a		, return_tensors="""np""" )
							lowerCamelCase			      =					processor(images=_a		, return_tensors="""np""" )

							for key in input_feat_extract.keys():
										self.assertAlmostEqual(input_feat_extract[key].sum()		, input_processor[key].sum()		, delta=1e-2 )





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							lowerCamelCase			      =					self.get_image_processor()
							lowerCamelCase			      =					self.get_tokenizer()

							lowerCamelCase			      =					VisionTextDualEncoderProcessor(tokenizer=_a		, image_processor=_a )

							lowerCamelCase			      =					"""lower newer"""

							lowerCamelCase			      =					processor(text=_a )

							lowerCamelCase			      =					tokenizer(_a )

							for key in encoded_tok.keys():
										self.assertListEqual(encoded_tok[key]		, encoded_processor[key] )





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							lowerCamelCase			      =					self.get_image_processor()
							lowerCamelCase			      =					self.get_tokenizer()

							lowerCamelCase			      =					VisionTextDualEncoderProcessor(tokenizer=_a		, image_processor=_a )

							lowerCamelCase			      =					"""lower newer"""
							lowerCamelCase			      =					self.prepare_image_inputs()

							lowerCamelCase			      =					processor(text=_a		, images=_a )

							self.assertListEqual(list(inputs.keys() )		, ["""input_ids""", """token_type_ids""", """attention_mask""", """pixel_values"""] )

							# test if it raises when no input is passed
							with self.assertRaises(_a ):
										processor()





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							lowerCamelCase			      =					self.get_image_processor()
							lowerCamelCase			      =					self.get_tokenizer()

							lowerCamelCase			      =					VisionTextDualEncoderProcessor(tokenizer=_a		, image_processor=_a )

							lowerCamelCase			      =					[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

							lowerCamelCase			      =					processor.batch_decode(_a )
							lowerCamelCase			      =					tokenizer.batch_decode(_a )

							self.assertListEqual(_a		, _a )





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							lowerCamelCase			      =					self.get_image_processor()
							lowerCamelCase			      =					self.get_tokenizer()

							lowerCamelCase			      =					VisionTextDualEncoderProcessor(tokenizer=_a		, image_processor=_a )

							lowerCamelCase			      =					"""lower newer"""
							lowerCamelCase			      =					self.prepare_image_inputs()

							lowerCamelCase			      =					processor(text=_a		, images=_a )

							self.assertListEqual(list(inputs.keys() )		, processor.model_input_names )



