



"""simple docstring"""







import gc
import random
import unittest

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import (
    DiffusionPipeline,
    UnCLIPImageVariationPipeline,
    UnCLIPScheduler,
    UNetaDConditionModel,
    UNetaDModel,
)
from diffusers.pipelines.unclip.text_proj import UnCLIPTextProjModel
from diffusers.utils import floats_tensor, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, load_image, require_torch_gpu, skip_mps

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()






class      UpperCAmelCase__		(      _snake_case   ,unittest.TestCase     ):
		lowerCAmelCase_       :	Union[str, Any] 							=     UnCLIPImageVariationPipeline
		lowerCAmelCase_       :	Optional[int] 							=     IMAGE_VARIATION_PARAMS - {"""height""", """width""", """guidance_scale"""}
		lowerCAmelCase_       :	List[Any] 							=     IMAGE_VARIATION_BATCH_PARAMS

		lowerCAmelCase_       :	Any 							=     [
		    """generator""",
		    """return_dict""",
		    """decoder_num_inference_steps""",
		    """super_res_num_inference_steps""",
		]
		lowerCAmelCase_       :	str 							=     False



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






									'''simple docstring'''

									return 32



		@property
		def 					A_		( self  :    Union[str, Any]				)			->						Union[str, Any]:






									'''simple docstring'''

									return 32



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






									'''simple docstring'''

									return self.time_input_dim



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






									'''simple docstring'''

									return self.time_input_dim * 4



		@property
		def 					A_		( self  :    str				)			->						Tuple:






									'''simple docstring'''

									return 100



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






									'''simple docstring'''

									A         =					CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'				)
									return tokenizer



		@property
		def 					A_		( self  :    Tuple				)			->						Dict:






									'''simple docstring'''

									torch.manual_seed(0				)
									A         =					CLIPTextConfig(
									    bos_token_id=0      ,							eos_token_id=2      ,							hidden_size=self.text_embedder_hidden_size      ,							projection_dim=self.text_embedder_hidden_size      ,							intermediate_size=37      ,							layer_norm_eps=1E-0_5      ,							num_attention_heads=4      ,							num_hidden_layers=5      ,							pad_token_id=1      ,							vocab_size=1_000      ,							)
									return CLIPTextModelWithProjection(snake_case_				)



		@property
		def 					A_		( self  :    str				)			->						Any:






									'''simple docstring'''

									torch.manual_seed(0				)
									A         =					CLIPVisionConfig(
									    hidden_size=self.text_embedder_hidden_size      ,							projection_dim=self.text_embedder_hidden_size      ,							num_hidden_layers=5      ,							num_attention_heads=4      ,							image_size=32      ,							intermediate_size=37      ,							patch_size=1      ,							)
									return CLIPVisionModelWithProjection(snake_case_				)



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






									'''simple docstring'''

									torch.manual_seed(0				)

									A         =					{
									    "clip_embeddings_dim": self.text_embedder_hidden_size,
									    "time_embed_dim": self.time_embed_dim,
									    "cross_attention_dim": self.cross_attention_dim,
									}

									A         =					UnCLIPTextProjModel(**snake_case_				)
									return model



		@property
		def 					A_		( self  :    str				)			->						Tuple:






									'''simple docstring'''

									torch.manual_seed(0				)

									A         =					{
									    "sample_size": 32,
									    # RGB in channels
									    "in_channels": 3,
									    # Out channels is double in channels because predicts mean and variance
									    "out_channels": 6,
									    "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,
									    "cross_attention_dim": self.cross_attention_dim,
									    "attention_head_dim": 4,
									    "resnet_time_scale_shift": "scale_shift",
									    "class_embed_type": "identity",
									}

									A         =					UNetaDConditionModel(**snake_case_				)
									return model



		@property
		def 					A_		( self  :    Optional[int]				)			->						List[str]:






									'''simple docstring'''

									return {
									    "sample_size": 64,
									    "layers_per_block": 1,
									    "down_block_types": ("ResnetDownsampleBlock2D", "ResnetDownsampleBlock2D"),
									    "up_block_types": ("ResnetUpsampleBlock2D", "ResnetUpsampleBlock2D"),
									    "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
									    "in_channels": 6,
									    "out_channels": 3,
									}



		@property
		def 					A_		( self  :    Tuple				)			->						Tuple:






									'''simple docstring'''

									torch.manual_seed(0				)

									A         =					UNetaDModel(**self.dummy_super_res_kwargs				)
									return model



		@property
		def 					A_		( self  :    int				)			->						Any:






									'''simple docstring'''

									torch.manual_seed(1				)

									A         =					UNetaDModel(**self.dummy_super_res_kwargs				)
									return model



		def 					A_		( self  :    Tuple				)			->						Union[str, Any]:






									'''simple docstring'''

									A         =					self.dummy_decoder
									A         =					self.dummy_text_proj
									A         =					self.dummy_text_encoder
									A         =					self.dummy_tokenizer
									A         =					self.dummy_super_res_first
									A         =					self.dummy_super_res_last

									A         =					UnCLIPScheduler(
									    variance_type='learned_range'      ,							prediction_type='epsilon'      ,							num_train_timesteps=1_000      ,							)

									A         =					UnCLIPScheduler(
									    variance_type='fixed_small_log'      ,							prediction_type='epsilon'      ,							num_train_timesteps=1_000      ,							)

									A         =					CLIPImageProcessor(crop_size=32      ,							size=32				)

									A         =					self.dummy_image_encoder

									return {
									    "decoder": decoder,
									    "text_encoder": text_encoder,
									    "tokenizer": tokenizer,
									    "text_proj": text_proj,
									    "feature_extractor": feature_extractor,
									    "image_encoder": image_encoder,
									    "super_res_first": super_res_first,
									    "super_res_last": super_res_last,
									    "decoder_scheduler": decoder_scheduler,
									    "super_res_scheduler": super_res_scheduler,
									}



		def 					A_		( self  :    str      ,							snake_case  :    str      ,							snake_case  :    str=0      ,							snake_case  :    List[Any]=True				)			->						str:






									'''simple docstring'''

									A         =					floats_tensor((1, 3, 32, 32)      ,							rng=random.Random(snake_case_				)				).to(snake_case_				)
									if str(snake_case_				).startswith('mps'				):
																A         =					torch.manual_seed(snake_case_				)
									else:
																A         =					torch.Generator(device=snake_case_				).manual_seed(snake_case_				)

									if pil_image:
																A         =					input_image * 0.5 + 0.5
																A         =					input_image.clamp(0      ,							1				)
																A         =					input_image.cpu().permute(0      ,							2      ,							3      ,							1				).float().numpy()
																A         =					DiffusionPipeline.numpy_to_pil(snake_case_				)[0]

									return {
									    "image": input_image,
									    "generator": generator,
									    "decoder_num_inference_steps": 2,
									    "super_res_num_inference_steps": 2,
									    "output_type": "np",
									}



		def 					A_		( self  :    List[Any]				)			->						List[str]:






									'''simple docstring'''

									A         =					"cpu"

									A         =					self.get_dummy_components()

									A         =					self.pipeline_class(**snake_case_				)
									A         =					pipe.to(snake_case_				)

									pipe.set_progress_bar_config(disable=snake_case_				)

									A         =					self.get_dummy_inputs(snake_case_      ,							pil_image=snake_case_				)

									A         =					pipe(**snake_case_				)
									A         =					output.images

									A         =					self.get_dummy_inputs(snake_case_      ,							pil_image=snake_case_				)

									A         =					pipe(
									    **snake_case_      ,							return_dict=snake_case_      ,							)[0]

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

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

									A         =					np.array(
									    [
									        0.9997,
									        0.0002,
									        0.9997,
									        0.9997,
									        0.9969,
									        0.0023,
									        0.9997,
									        0.9969,
									        0.9970,
									    ]				)

									assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-2
									assert np.abs(image_from_tuple_slice.flatten() - expected_slice				).max() < 1E-2



		def 					A_		( self  :    Tuple				)			->						List[str]:






									'''simple docstring'''

									A         =					"cpu"

									A         =					self.get_dummy_components()

									A         =					self.pipeline_class(**snake_case_				)
									A         =					pipe.to(snake_case_				)

									pipe.set_progress_bar_config(disable=snake_case_				)

									A         =					self.get_dummy_inputs(snake_case_      ,							pil_image=snake_case_				)

									A         =					pipe(**snake_case_				)
									A         =					output.images

									A         =					self.get_dummy_inputs(snake_case_      ,							pil_image=snake_case_				)

									A         =					pipe(
									    **snake_case_      ,							return_dict=snake_case_      ,							)[0]

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

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

									A         =					np.array([0.9997, 0.0003, 0.9997, 0.9997, 0.9970, 0.0024, 0.9997, 0.9971, 0.9971]				)

									assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-2
									assert np.abs(image_from_tuple_slice.flatten() - expected_slice				).max() < 1E-2



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






									'''simple docstring'''

									A         =					"cpu"

									A         =					self.get_dummy_components()

									A         =					self.pipeline_class(**snake_case_				)
									A         =					pipe.to(snake_case_				)

									pipe.set_progress_bar_config(disable=snake_case_				)

									A         =					self.get_dummy_inputs(snake_case_      ,							pil_image=snake_case_				)
									A         =					[
									    pipeline_inputs["image"],
									    pipeline_inputs["image"],
									]

									A         =					pipe(**snake_case_				)
									A         =					output.images

									A         =					self.get_dummy_inputs(snake_case_      ,							pil_image=snake_case_				)
									A         =					[
									    tuple_pipeline_inputs["image"],
									    tuple_pipeline_inputs["image"],
									]

									A         =					pipe(
									    **snake_case_      ,							return_dict=snake_case_      ,							)[0]

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

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

									A         =					np.array(
									    [
									        0.9997,
									        0.9989,
									        0.0008,
									        0.0021,
									        0.9960,
									        0.0018,
									        0.0014,
									        0.0002,
									        0.9933,
									    ]				)

									assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-2
									assert np.abs(image_from_tuple_slice.flatten() - expected_slice				).max() < 1E-2



		def 					A_		( self  :    Tuple				)			->						List[str]:






									'''simple docstring'''

									A         =					torch.device('cpu'				)






									class      UpperCAmelCase__		:
											lowerCAmelCase_       :	List[str] 							=     1



									A         =					self.get_dummy_components()

									A         =					self.pipeline_class(**snake_case_				)
									A         =					pipe.to(snake_case_				)

									pipe.set_progress_bar_config(disable=snake_case_				)

									A         =					torch.Generator(device=snake_case_				).manual_seed(0				)
									A         =					pipe.decoder.dtype
									A         =					1

									A         =					(
									    batch_size,
									    pipe.decoder.config.in_channels,
									    pipe.decoder.config.sample_size,
									    pipe.decoder.config.sample_size,
									)
									A         =					pipe.prepare_latents(
									    snake_case_      ,							dtype=snake_case_      ,							device=snake_case_      ,							generator=snake_case_      ,							latents=snake_case_      ,							scheduler=DummyScheduler()				)

									A         =					(
									    batch_size,
									    pipe.super_res_first.config.in_channels // 2,
									    pipe.super_res_first.config.sample_size,
									    pipe.super_res_first.config.sample_size,
									)
									A         =					pipe.prepare_latents(
									    snake_case_      ,							dtype=snake_case_      ,							device=snake_case_      ,							generator=snake_case_      ,							latents=snake_case_      ,							scheduler=DummyScheduler()				)

									A         =					self.get_dummy_inputs(snake_case_      ,							pil_image=snake_case_				)

									A         =					pipe(
									    **snake_case_      ,							decoder_latents=snake_case_      ,							super_res_latents=snake_case_				).images

									A         =					self.get_dummy_inputs(snake_case_      ,							pil_image=snake_case_				)
									# Don't pass image, instead pass embedding
									A         =					pipeline_inputs.pop('image'				)
									A         =					pipe.image_encoder(snake_case_				).image_embeds

									A         =					pipe(
									    **snake_case_      ,							decoder_latents=snake_case_      ,							super_res_latents=snake_case_      ,							image_embeddings=snake_case_      ,							).images

									# make sure passing text embeddings manually is identical
									assert np.abs(img_out_a - img_out_a				).max() < 1E-4



		@skip_mps
		def 					A_		( self  :    str				)			->						Any:






									'''simple docstring'''

									A         =					torch_device == "cpu"

									# Check is relaxed because there is not a torch 2.0 sliced attention added kv processor
									A         =					1E-2

									self._test_attention_slicing_forward_pass(
									    test_max_difference=snake_case_      ,							expected_max_diff=snake_case_				)



		@skip_mps
		def 					A_		( self  :    int				)			->						List[str]:






									'''simple docstring'''

									A         =					torch_device == "cpu"
									A         =					True
									A         =					[
									    "decoder_num_inference_steps",
									    "super_res_num_inference_steps",
									]

									self._test_inference_batch_single_identical(
									    test_max_difference=snake_case_      ,							relax_max_difference=snake_case_      ,							additional_params_copy_to_batched_inputs=snake_case_      ,							)



		def 					A_		( self  :    Union[str, Any]				)			->						List[Any]:






									'''simple docstring'''

									A         =					[
									    "decoder_num_inference_steps",
									    "super_res_num_inference_steps",
									]

									if torch_device == "mps":
																# TODO: MPS errors with larger batch sizes
																A         =					[2, 3]
																self._test_inference_batch_consistent(
																    batch_sizes=snake_case_      ,							additional_params_copy_to_batched_inputs=snake_case_      ,							)
									else:
																self._test_inference_batch_consistent(
																    additional_params_copy_to_batched_inputs=snake_case_				)



		@skip_mps
		def 					A_		( self  :    str				)			->						List[Any]:






									'''simple docstring'''

									return super().test_dict_tuple_outputs_equivalent()



		@skip_mps
		def 					A_		( self  :    Tuple				)			->						Dict:






									'''simple docstring'''

									return super().test_save_load_local()



		@skip_mps
		def 					A_		( self  :    Any				)			->						Tuple:






									'''simple docstring'''

									return super().test_save_load_optional_components()






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



		def 					A_		( self  :    Any				)			->						Optional[Any]:






									'''simple docstring'''

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



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






									'''simple docstring'''

									A         =					load_image(
									    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/unclip/cat.png'				)
									A         =					load_numpy(
									    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
									    '/unclip/karlo_v1_alpha_cat_variation_fp16.npy'				)

									A         =					UnCLIPImageVariationPipeline.from_pretrained(
									    'kakaobrain/karlo-v1-alpha-image-variations'      ,							torch_dtype=torch.floataa				)
									A         =					pipeline.to(snake_case_				)
									pipeline.set_progress_bar_config(disable=snake_case_				)

									A         =					torch.Generator(device='cpu'				).manual_seed(0				)
									A         =					pipeline(
									    snake_case_      ,							generator=snake_case_      ,							output_type='np'      ,							)

									A         =					output.images[0]

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

									assert_mean_pixel_difference(snake_case_      ,							snake_case_      ,							15				)








"""simple docstring"""







import os
import unittest

from transformers import LayoutLMTokenizer, LayoutLMTokenizerFast
from transformers.models.layoutlm.tokenization_layoutlm import VOCAB_FILES_NAMES
from transformers.testing_utils import require_tokenizers

from ...test_tokenization_common import TokenizerTesterMixin






@require_tokenizers
class      UpperCAmelCase__		(      UpperCamelCase   ,unittest.TestCase     ):
		lowerCAmelCase_       :	Tuple 							=     LayoutLMTokenizer
		lowerCAmelCase_       :	Any 							=     LayoutLMTokenizerFast
		lowerCAmelCase_       :	Optional[int] 							=     True
		lowerCAmelCase_       :	List[Any] 							=     True



		def 					A_		( self  :    Optional[Any]				)			->						Optional[Any]:






									'''simple docstring'''

									super().setUp()

									A         =					[
									    '[UNK]',
									    '[CLS]',
									    '[SEP]',
									    'want',
									    '##want',
									    '##ed',
									    'wa',
									    'un',
									    'runn',
									    '##ing',
									    ',',
									    'low',
									    'lowest',
									]
									A         =					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]				)				)



		def 					A_		( self  :    str      ,							**snake_case  :    Tuple				)			->						Union[str, Any]:






									'''simple docstring'''

									return LayoutLMTokenizer.from_pretrained(self.tmpdirname      ,							**snake_case				)



		def 					A_		( self  :    List[str]      ,							snake_case  :    int				)			->						List[Any]:






									'''simple docstring'''

									A         =					'UNwant\u00E9d,running'
									A         =					'unwanted, running'
									return input_text, output_text



		def 					A_		( self  :    Tuple				)			->						Optional[int]:






									'''simple docstring'''

									A         =					self.tokenizer_class(self.vocab_file				)

									A         =					tokenizer.tokenize('UNwant\u00E9d,running'				)
									self.assertListEqual(snake_case      ,							['un', '##want', '##ed', ',', 'runn', '##ing']				)
									self.assertListEqual(tokenizer.convert_tokens_to_ids(snake_case				)      ,							[7, 4, 5, 10, 8, 9]				)



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






									'''simple docstring'''

									pass



