


import gc
import random
import unittest

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

from diffusers import AutoencoderKL, DDIMScheduler, DDPMScheduler, StableUnCLIPImgaImgPipeline, UNetaDConditionModel
from diffusers.pipelines.pipeline_utils import DiffusionPipeline
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    floats_tensor,
    load_image,
    load_numpy,
    require_torch_gpu,
    skip_mps,
    slow,
    torch_device,
)

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()

class     snake_case      (							_UpperCamelCase						,			_UpperCamelCase						,			_UpperCamelCase						,			unittest.TestCase						):
			'''simple docstring'''

			UpperCAmelCase							:							Dict							=			StableUnCLIPImgaImgPipeline
			UpperCAmelCase							:							Union[str, Any]							=			TEXT_GUIDED_IMAGE_VARIATION_PARAMS
			UpperCAmelCase							:							List[Any]							=			TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
			UpperCAmelCase							:							List[str]							=			frozenset(
			    []						)  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
			UpperCAmelCase							:							str							=			frozenset([]						)

			def _lowercase			(						self  :  Optional[int]       )       ->       Optional[Any]:




								"""simple docstring"""






								SCREAMING_SNAKE_CASE_						   =							32
								SCREAMING_SNAKE_CASE_						   =							embedder_hidden_size

								# image encoding components

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

								torch.manual_seed(0       )
								SCREAMING_SNAKE_CASE_						   =							CLIPVisionModelWithProjection(
								    CLIPVisionConfig(
								        hidden_size=UpperCamelCase__				,				projection_dim=UpperCamelCase__				,				num_hidden_layers=5				,				num_attention_heads=4				,				image_size=32				,				intermediate_size=37				,				patch_size=1				,				)       )

								# regular denoising components

								torch.manual_seed(0       )
								SCREAMING_SNAKE_CASE_						   =							StableUnCLIPImageNormalizer(embedding_dim=UpperCamelCase__       )
								SCREAMING_SNAKE_CASE_						   =							DDPMScheduler(beta_schedule='''squaredcos_cap_v2'''       )

								torch.manual_seed(0       )
								SCREAMING_SNAKE_CASE_						   =							CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''       )

								torch.manual_seed(0       )
								SCREAMING_SNAKE_CASE_						   =							CLIPTextModel(
								    CLIPTextConfig(
								        bos_token_id=0				,				eos_token_id=2				,				hidden_size=UpperCamelCase__				,				projection_dim=32				,				intermediate_size=37				,				layer_norm_eps=1e-05				,				num_attention_heads=4				,				num_hidden_layers=5				,				pad_token_id=1				,				vocab_size=1_000				,				)       )

								torch.manual_seed(0       )
								SCREAMING_SNAKE_CASE_						   =							UNetaDConditionModel(
								    sample_size=32				,				in_channels=4				,				out_channels=4				,				down_block_types=('''CrossAttnDownBlock2D''', '''DownBlock2D''')				,				up_block_types=('''UpBlock2D''', '''CrossAttnUpBlock2D''')				,				block_out_channels=(32, 64)				,				attention_head_dim=(2, 4)				,				class_embed_type='''projection'''				,				projection_class_embeddings_input_dim=embedder_projection_dim * 2				,				cross_attention_dim=UpperCamelCase__				,				layers_per_block=1				,				upcast_attention=UpperCamelCase__				,				use_linear_projection=UpperCamelCase__				,				)

								torch.manual_seed(0       )
								SCREAMING_SNAKE_CASE_						   =							DDIMScheduler(
								    beta_schedule='''scaled_linear'''				,				beta_start=0.00_085				,				beta_end=0.012				,				prediction_type='''v_prediction'''				,				set_alpha_to_one=UpperCamelCase__				,				steps_offset=1				,				)

								torch.manual_seed(0       )
								SCREAMING_SNAKE_CASE_						   =							AutoencoderKL()

								SCREAMING_SNAKE_CASE_						   =							{
								    # image encoding components
								    '''feature_extractor''': feature_extractor,
								    '''image_encoder''': image_encoder.eval(),
								    # image noising components
								    '''image_normalizer''': image_normalizer.eval(),
								    '''image_noising_scheduler''': image_noising_scheduler,
								    # regular denoising components
								    '''tokenizer''': tokenizer,
								    '''text_encoder''': text_encoder.eval(),
								    '''unet''': unet.eval(),
								    '''scheduler''': scheduler,
								    '''vae''': vae.eval(),
								}

								return components

			def _lowercase			(						self  :  List[str]				,				lowerCAmelCase_  :  str				,				lowerCAmelCase_  :  List[Any]=0				,				lowerCAmelCase_  :  Optional[int]=True       )       ->       Tuple:




								"""simple docstring"""






								if str(UpperCamelCase__       ).startswith('''mps'''       ):
													SCREAMING_SNAKE_CASE_						   =							torch.manual_seed(UpperCamelCase__       )
								else:
													SCREAMING_SNAKE_CASE_						   =							torch.Generator(device=UpperCamelCase__       ).manual_seed(UpperCamelCase__       )

								SCREAMING_SNAKE_CASE_						   =							floats_tensor((1, 3, 32, 32)				,				rng=random.Random(UpperCamelCase__       )       ).to(UpperCamelCase__       )

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

								return {
								    "prompt": "An anime racoon running a marathon",
								    "image": input_image,
								    "generator": generator,
								    "num_inference_steps": 2,
								    "output_type": "np",
								}

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




								"""simple docstring"""






								SCREAMING_SNAKE_CASE_						   =							'''cpu'''  # ensure determinism for the device-dependent torch.Generator
								SCREAMING_SNAKE_CASE_						   =							self.get_dummy_components()
								SCREAMING_SNAKE_CASE_						   =							StableUnCLIPImgaImgPipeline(**UpperCamelCase__       )
								SCREAMING_SNAKE_CASE_						   =							sd_pipe.to(UpperCamelCase__       )
								sd_pipe.set_progress_bar_config(disable=UpperCamelCase__       )

								SCREAMING_SNAKE_CASE_						   =							self.get_dummy_inputs(UpperCamelCase__       )
								inputs.update({'''image_embeds''': None}       )
								SCREAMING_SNAKE_CASE_						   =							sd_pipe(**UpperCamelCase__       ).images
								SCREAMING_SNAKE_CASE_						   =							image[0, -3:, -3:, -1]

								assert image.shape == (1, 32, 32, 3)
								SCREAMING_SNAKE_CASE_						   =							np.array([0.3_872, 0.7_224, 0.5_601, 0.4_741, 0.6_872, 0.5_814, 0.4_636, 0.3_867, 0.5_078]       )

								assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-3

			def _lowercase			(						self  :  int       )       ->       int:




								"""simple docstring"""






								SCREAMING_SNAKE_CASE_						   =							torch_device in ['''cpu''', '''mps''']

								self._test_attention_slicing_forward_pass(test_max_difference=UpperCamelCase__       )

			def _lowercase			(						self  :  int       )       ->       List[Any]:




								"""simple docstring"""






								SCREAMING_SNAKE_CASE_						   =							torch_device in ['''cpu''', '''mps''']

								self._test_inference_batch_single_identical(test_max_difference=UpperCamelCase__       )




			@unittest.skipIf(
			    torch_device != '''cuda''' or not is_xformers_available()				,				reason='''XFormers attention is only available with CUDA and `xformers` installed'''				,				)
			def _lowercase			(						self  :  Any       )       ->       Any:




								"""simple docstring"""






								self._test_xformers_attention_forwardGenerator_pass(test_max_difference=UpperCamelCase__       )

@slow
@require_torch_gpu
class     snake_case      (							unittest.TestCase						):
			'''simple docstring'''

			def _lowercase			(						self  :  List[str]       )       ->       Optional[int]:




								"""simple docstring"""






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

			def _lowercase			(						self  :  str       )       ->       int:




								"""simple docstring"""






								SCREAMING_SNAKE_CASE_						   =							load_image(
								    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'''       )

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

								SCREAMING_SNAKE_CASE_						   =							StableUnCLIPImgaImgPipeline.from_pretrained(
								    '''fusing/stable-unclip-2-1-l-img2img'''				,				torch_dtype=torch.floataa       )
								pipe.to(UpperCamelCase__       )
								pipe.set_progress_bar_config(disable=UpperCamelCase__       )
								# stable unclip will oom when integration tests are run on a V100,
								# so turn on memory savings
								pipe.enable_attention_slicing()
								pipe.enable_sequential_cpu_offload()

								SCREAMING_SNAKE_CASE_						   =							torch.Generator(device='''cpu'''       ).manual_seed(0       )
								SCREAMING_SNAKE_CASE_						   =							pipe(UpperCamelCase__				,				'''anime turle'''				,				generator=UpperCamelCase__				,				output_type='''np'''       )

								SCREAMING_SNAKE_CASE_						   =							output.images[0]

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

								assert_mean_pixel_difference(UpperCamelCase__				,				UpperCamelCase__       )

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




								"""simple docstring"""






								SCREAMING_SNAKE_CASE_						   =							load_image(
								    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'''       )

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

								SCREAMING_SNAKE_CASE_						   =							StableUnCLIPImgaImgPipeline.from_pretrained(
								    '''fusing/stable-unclip-2-1-h-img2img'''				,				torch_dtype=torch.floataa       )
								pipe.to(UpperCamelCase__       )
								pipe.set_progress_bar_config(disable=UpperCamelCase__       )
								# stable unclip will oom when integration tests are run on a V100,
								# so turn on memory savings
								pipe.enable_attention_slicing()
								pipe.enable_sequential_cpu_offload()

								SCREAMING_SNAKE_CASE_						   =							torch.Generator(device='''cpu'''       ).manual_seed(0       )
								SCREAMING_SNAKE_CASE_						   =							pipe(UpperCamelCase__				,				'''anime turle'''				,				generator=UpperCamelCase__				,				output_type='''np'''       )

								SCREAMING_SNAKE_CASE_						   =							output.images[0]

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

								assert_mean_pixel_difference(UpperCamelCase__				,				UpperCamelCase__       )




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




								"""simple docstring"""






								SCREAMING_SNAKE_CASE_						   =							load_image(
								    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'''       )

								torch.cuda.empty_cache()
								torch.cuda.reset_max_memory_allocated()
								torch.cuda.reset_peak_memory_stats()

								SCREAMING_SNAKE_CASE_						   =							StableUnCLIPImgaImgPipeline.from_pretrained(
								    '''fusing/stable-unclip-2-1-h-img2img'''				,				torch_dtype=torch.floataa       )
								SCREAMING_SNAKE_CASE_						   =							pipe.to(UpperCamelCase__       )
								pipe.set_progress_bar_config(disable=UpperCamelCase__       )
								pipe.enable_attention_slicing()
								pipe.enable_sequential_cpu_offload()

								SCREAMING_SNAKE_CASE_						   =							pipe(
								    UpperCamelCase__				,				'''anime turtle'''				,				num_inference_steps=2				,				output_type='''np'''				,				)

								SCREAMING_SNAKE_CASE_						   =							torch.cuda.max_memory_allocated()
								# make sure that less than 7 GB is allocated
								assert mem_bytes < 7 * 10**9



'''simple docstring'''






from __future__ import annotations

import unittest

from transformers import is_tf_available
from transformers.testing_utils import require_sentencepiece, require_tf, require_tokenizers, slow


if is_tf_available():
   import numpy as np
   import tensorflow as tf

   from transformers import TFCamembertModel







@require_tf
@require_sentencepiece
@require_tokenizers
class lowercase_       (		unittest.TestCase			):







   """simple docstring"""

   @slow
   def 	__UpperCAmelCase   (      self      :			List[str]     )  ->			Any:
        _A    		=  TFCamembertModel.from_pretrained('jplu/tf-camembert-base'     )

        _A    		=  tf.convert_to_tensor(
            [[5, 1_21, 11, 6_60, 16, 7_30, 2_55_43, 1_10, 83, 6]], dtype=tf.intaa, )  # J'aime le camembert !"

        _A    		=  model(UpperCamelCase__     )['last_hidden_state']
        _A    		=  tf.TensorShape((1, 10, 7_68)     )
        self.assertEqual(output.shape, UpperCamelCase__     )
        # compare the actual values for a slice.
        _A    		=  tf.convert_to_tensor(
            [[[-0.0_254, 0.0_235, 0.1_027], [0.0_606, -0.1_811, -0.0_418], [-0.1_561, -0.1_127, 0.2_687]]], dtype=tf.floataa, )
        # camembert = torch.hub.load('pytorch/fairseq', 'camembert.v0')
        # camembert.eval()
        # expected_slice = roberta.model.forward(input_ids)[0][:, :3, :3].detach()

        self.assertTrue(np.allclose(output[:, :3, :3].numpy(), expected_slice.numpy(), atol=1e-4     )     )



