



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    UpperCAmelCase__(				lowerCamelCase  ,				lowerCamelCase  ,				lowerCamelCase  ,				unittest.TestCase						):

   '''simple docstring'''




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


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




         """simple docstring"""

         lowercase__  =	32
         lowercase__  =	embedder_hidden_size

         # image encoding components

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

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

         # regular denoising components

         torch.manual_seed(0)
         lowercase__  =	StableUnCLIPImageNormalizer(embedding_dim=lowerCAmelCase)
         lowercase__  =	DDPMScheduler(beta_schedule='squaredcos_cap_v2')

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

         torch.manual_seed(0)
         lowercase__  =	CLIPTextModel(
             CLIPTextConfig(
                 bos_token_id=0      ,					eos_token_id=2      ,					hidden_size=lowerCAmelCase      ,					projection_dim=32      ,					intermediate_size=37      ,					layer_norm_eps=1E-0_5      ,					num_attention_heads=4      ,					num_hidden_layers=5      ,					pad_token_id=1      ,					vocab_size=10_00      ,					))

         torch.manual_seed(0)
         lowercase__  =	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=lowerCAmelCase      ,					layers_per_block=1      ,					upcast_attention=lowerCAmelCase      ,					use_linear_projection=lowerCAmelCase      ,					)

         torch.manual_seed(0)
         lowercase__  =	DDIMScheduler(
             beta_schedule='scaled_linear'      ,					beta_start=0.0_00_85      ,					beta_end=0.0_12      ,					prediction_type='v_prediction'      ,					set_alpha_to_one=lowerCAmelCase      ,					steps_offset=1      ,					)

         torch.manual_seed(0)
         lowercase__  =	AutoencoderKL()

         lowercase__  =	{
             # 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       UpperCAmelCase		(    self	:      Union[str, Any]      ,					lowerCAmelCase	:      Any      ,					lowerCAmelCase	:      Dict=0      ,					lowerCAmelCase	:      Union[str, Any]=True)						->					Tuple:




         """simple docstring"""

         if str(lowerCAmelCase).startswith('mps'):
               lowercase__  =	torch.manual_seed(lowerCAmelCase)
         else:
               lowercase__  =	torch.Generator(device=lowerCAmelCase).manual_seed(lowerCAmelCase)

         lowercase__  =	floats_tensor((1, 3, 32, 32)      ,					rng=random.Random(lowerCAmelCase)).to(lowerCAmelCase)

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

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


   @skip_mps
   def       UpperCAmelCase		(    self	:      Tuple)						->					Optional[Any]:




         """simple docstring"""

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

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

         assert image.shape == (1, 32, 32, 3)
         lowercase__  =	np.array([0.38_72, 0.72_24, 0.56_01, 0.47_41, 0.68_72, 0.58_14, 0.46_36, 0.38_67, 0.50_78])

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


   def       UpperCAmelCase		(    self	:      Tuple)						->					Any:




         """simple docstring"""

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

         self._test_attention_slicing_forward_pass(test_max_difference=lowerCAmelCase)


   def       UpperCAmelCase		(    self	:      Optional[Any])						->					List[Any]:




         """simple docstring"""

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

         self._test_inference_batch_single_identical(test_max_difference=lowerCAmelCase)



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




         """simple docstring"""

         self._test_xformers_attention_forwardGenerator_pass(test_max_difference=lowerCAmelCase)



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

   '''simple docstring'''




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




         """simple docstring"""

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


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




         """simple docstring"""

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

         lowercase__  =	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')

         lowercase__  =	StableUnCLIPImgaImgPipeline.from_pretrained(
             'fusing/stable-unclip-2-1-l-img2img'      ,					torch_dtype=torch.floataa)
         pipe.to(lowerCAmelCase)
         pipe.set_progress_bar_config(disable=lowerCAmelCase)
         # 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()

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

         lowercase__  =	output.images[0]

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

         assert_mean_pixel_difference(lowerCAmelCase      ,					lowerCAmelCase)


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




         """simple docstring"""

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

         lowercase__  =	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')

         lowercase__  =	StableUnCLIPImgaImgPipeline.from_pretrained(
             'fusing/stable-unclip-2-1-h-img2img'      ,					torch_dtype=torch.floataa)
         pipe.to(lowerCAmelCase)
         pipe.set_progress_bar_config(disable=lowerCAmelCase)
         # 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()

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

         lowercase__  =	output.images[0]

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

         assert_mean_pixel_difference(lowerCAmelCase      ,					lowerCAmelCase)



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




         """simple docstring"""

         lowercase__  =	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()

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

         lowercase__  =	pipe(
             lowerCAmelCase      ,					'anime turtle'      ,					num_inference_steps=2      ,					output_type='np'      ,					)

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





import json
import os
import unittest

from transformers.models.ctrl.tokenization_ctrl import VOCAB_FILES_NAMES, CTRLTokenizer

from ...test_tokenization_common import TokenizerTesterMixin



class    UpperCAmelCase__(				lowerCamelCase  ,				unittest.TestCase						):

   '''simple docstring'''




   A     :	int												=     CTRLTokenizer
   A     :	Tuple												=     False
   A     :	Any												=     False


   def       UpperCAmelCase		(    self	:      int)						->					Any:




         """simple docstring"""

         super().setUp()

         # Adapted from Sennrich et al. 2015 and https://github.com/rsennrich/subword-nmt
         lowercase__  =	['adapt', 're@@', 'a@@', 'apt', 'c@@', 't', '<unk>']
         lowercase__  =	dict(zip(lowerCAmelCase      ,					range(len(lowerCAmelCase))))
         lowercase__  =	['#version: 0.2', 'a p', 'ap t</w>', 'r e', 'a d', 'ad apt</w>', '']
         lowercase__  =	{'unk_token': '<unk>'}

         lowercase__  =	os.path.join(self.tmpdirname      ,					VOCAB_FILES_NAMES['vocab_file'])
         lowercase__  =	os.path.join(self.tmpdirname      ,					VOCAB_FILES_NAMES['merges_file'])
         with open(self.vocab_file      ,					'w'      ,					encoding='utf-8') as fp:
               fp.write(json.dumps(lowerCAmelCase) + '\n')
         with open(self.merges_file      ,					'w'      ,					encoding='utf-8') as fp:
               fp.write('\n'.join(lowerCAmelCase))


   def       UpperCAmelCase		(    self	:      int      ,					**lowerCAmelCase	:      str)						->					Dict:




         """simple docstring"""

         kwargs.update(self.special_tokens_map)
         return CTRLTokenizer.from_pretrained(self.tmpdirname      ,					**lowerCAmelCase)


   def       UpperCAmelCase		(    self	:      List[str]      ,					lowerCAmelCase	:      Optional[int])						->					Optional[Any]:




         """simple docstring"""

         lowercase__  =	'adapt react readapt apt'
         lowercase__  =	'adapt react readapt apt'
         return input_text, output_text



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




         """simple docstring"""

         lowercase__  =	CTRLTokenizer(self.vocab_file      ,					self.merges_file      ,					**self.special_tokens_map)
         lowercase__  =	'adapt react readapt apt'
         lowercase__  =	'adapt re@@ a@@ c@@ t re@@ adapt apt'.split()
         lowercase__  =	tokenizer.tokenize(lowerCAmelCase)
         self.assertListEqual(lowerCAmelCase      ,					lowerCAmelCase)

         lowercase__  =	tokens + [tokenizer.unk_token]

         lowercase__  =	[0, 1, 2, 4, 5, 1, 0, 3, 6]
         self.assertListEqual(tokenizer.convert_tokens_to_ids(lowerCAmelCase)      ,					lowerCAmelCase)
