






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_					(    __A				,       __A				,       __A				,       unittest.TestCase				):




      """simple docstring"""




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






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


       '''simple docstring'''

       lowercase					:   List[Any]				  =32
       lowercase					:   Optional[Any]				  =embedder_hidden_size

       # image encoding components

       lowercase					:   List[str]				  =CLIPImageProcessor(crop_size=32     ,      size=32		)

       torch.manual_seed(0		)
       lowercase					:   str				  =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		)
       lowercase					:   int				  =StableUnCLIPImageNormalizer(embedding_dim=UpperCAmelCase		)
       lowercase					:   Union[str, Any]				  =DDPMScheduler(beta_schedule='''squaredcos_cap_v2'''		)

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

       torch.manual_seed(0		)
       lowercase					:   Tuple				  =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=1000     ,      )		)

       torch.manual_seed(0		)
       lowercase					:   int				  =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		)
       lowercase					:   int				  =DDIMScheduler(
           beta_schedule='''scaled_linear'''     ,      beta_start=0.0_0_0_8_5     ,      beta_end=0.0_1_2     ,      prediction_type='''v_prediction'''     ,      set_alpha_to_one=UpperCAmelCase     ,      steps_offset=1     ,      )

       torch.manual_seed(0		)
       lowercase					:   str				  =AutoencoderKL()

       lowercase					:   int				  ={
           # 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 				A__      ( self			:		Any     ,      UpperCAmelCase			:		List[str]     ,      UpperCAmelCase			:		List[Any]=0     ,      UpperCAmelCase			:		int=True		)  ->  Optional[Any]:


       '''simple docstring'''

       if str(UpperCAmelCase		).startswith('''mps'''		):
        lowercase					:   List[str]				  =torch.manual_seed(UpperCAmelCase		)
       else:
        lowercase					:   Any				  =torch.Generator(device=UpperCAmelCase		).manual_seed(UpperCAmelCase		)

       lowercase					:   Any				  =floats_tensor((1, 3, 32, 32)     ,      rng=random.Random(UpperCAmelCase		)		).to(UpperCAmelCase		)

       if pil_image:
        lowercase					:   str				  =input_image * 0.5 + 0.5
        lowercase					:   Tuple				  =input_image.clamp(0     ,      1		)
        lowercase					:   Tuple				  =input_image.cpu().permute(0     ,      2     ,      3     ,      1		).float().numpy()
        lowercase					:   Dict				  =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 				A__      ( self			:		Dict		)  ->  Dict:


       '''simple docstring'''

       lowercase					:   Union[str, Any]				  ='''cpu'''  # ensure determinism for the device-dependent torch.Generator
       lowercase					:   Optional[int]				  =self.get_dummy_components()
       lowercase					:   List[Any]				  =StableUnCLIPImgaImgPipeline(**UpperCAmelCase		)
       lowercase					:   str				  =sd_pipe.to(UpperCAmelCase		)
       sd_pipe.set_progress_bar_config(disable=UpperCAmelCase		)

       lowercase					:   Optional[Any]				  =self.get_dummy_inputs(UpperCAmelCase		)
       inputs.update({'''image_embeds''': None}		)
       lowercase					:   int				  =sd_pipe(**UpperCAmelCase		).images
       lowercase					:   str				  =image[0, -3:, -3:, -1]

       assert image.shape == (1, 32, 32, 3)
       lowercase					:   str				  =np.array([0.3_8_7_2, 0.7_2_2_4, 0.5_6_0_1, 0.4_7_4_1, 0.6_8_7_2, 0.5_8_1_4, 0.4_6_3_6, 0.3_8_6_7, 0.5_0_7_8]		)

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






      def 				A__      ( self			:		Dict		)  ->  Tuple:


       '''simple docstring'''

       lowercase					:   Union[str, Any]				  =torch_device in ['''cpu''', '''mps''']

       self._test_attention_slicing_forward_pass(test_max_difference=UpperCAmelCase		)






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


       '''simple docstring'''

       lowercase					:   int				  =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 				A__      ( self			:		Optional[Any]		)  ->  Union[str, Any]:


       '''simple docstring'''

       self._test_xformers_attention_forwardGenerator_pass(test_max_difference=UpperCAmelCase		)





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




      """simple docstring"""




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


       '''simple docstring'''

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






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


       '''simple docstring'''

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

       lowercase					:   Dict				  =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					:   Union[str, Any]				  =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()

       lowercase					:   Tuple				  =torch.Generator(device='''cpu'''		).manual_seed(0		)
       lowercase					:   Dict				  =pipe(UpperCAmelCase     ,      '''anime turle'''     ,      generator=UpperCAmelCase     ,      output_type='''np'''		)

       lowercase					:   int				  =output.images[0]

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

       assert_mean_pixel_difference(UpperCAmelCase     ,      UpperCAmelCase		)






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


       '''simple docstring'''

       lowercase					:   Optional[Any]				  =load_image(
           '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'''		)

       lowercase					:   Optional[int]				  =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					:   Tuple				  =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()

       lowercase					:   Any				  =torch.Generator(device='''cpu'''		).manual_seed(0		)
       lowercase					:   Optional[Any]				  =pipe(UpperCAmelCase     ,      '''anime turle'''     ,      generator=UpperCAmelCase     ,      output_type='''np'''		)

       lowercase					:   str				  =output.images[0]

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

       assert_mean_pixel_difference(UpperCAmelCase     ,      UpperCAmelCase		)






      def 				A__      ( self			:		str		)  ->  Tuple:


       '''simple docstring'''

       lowercase					:   Union[str, Any]				  =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					:   Any				  =StableUnCLIPImgaImgPipeline.from_pretrained(
           '''fusing/stable-unclip-2-1-h-img2img'''     ,      torch_dtype=torch.floataa		)
       lowercase					:   List[Any]				  =pipe.to(UpperCAmelCase		)
       pipe.set_progress_bar_config(disable=UpperCAmelCase		)
       pipe.enable_attention_slicing()
       pipe.enable_sequential_cpu_offload()

       lowercase					:   List[str]				  =pipe(
           UpperCAmelCase     ,      '''anime turtle'''     ,      num_inference_steps=2     ,      output_type='''np'''     ,      )

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











'''simple docstring'''



import json
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
from transformers import (
    CONFIG_MAPPING,
    IMAGE_PROCESSOR_MAPPING,
    AutoConfig,
    AutoImageProcessor,
    CLIPConfig,
    CLIPImageProcessor,
)
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER


sys.path.append(str(Path(__file__).parent.parent.parent.parent / 'utils'))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_image_processing import CustomImageProcessor  # noqa E402
class UpperCAmelCase_					(    unittest.TestCase				):




      """simple docstring"""




      def 				A__      ( self			:		int		)  ->  Any:


       '''simple docstring'''

       lowercase					:   Union[str, Any]				  =0






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


       '''simple docstring'''

       lowercase					:   Tuple				  =AutoImageProcessor.from_pretrained('''openai/clip-vit-base-patch32'''		)
       self.assertIsInstance(UpperCAmelCase     ,      UpperCAmelCase		)






      def 				A__      ( self			:		Any		)  ->  int:


       '''simple docstring'''

       with tempfile.TemporaryDirectory() as tmpdirname:
        lowercase					:   int				  =Path(UpperCAmelCase		) / '''preprocessor_config.json'''
        lowercase					:   str				  =Path(UpperCAmelCase		) / '''config.json'''
        json.dump(
            {'''image_processor_type''': '''CLIPImageProcessor''', '''processor_class''': '''CLIPProcessor'''}     ,      open(UpperCAmelCase     ,      '''w'''		)     ,      )
        json.dump({'''model_type''': '''clip'''}     ,      open(UpperCAmelCase     ,      '''w'''		)		)

        lowercase					:   str				  =AutoImageProcessor.from_pretrained(UpperCAmelCase		)
        self.assertIsInstance(UpperCAmelCase     ,      UpperCAmelCase		)






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


       '''simple docstring'''

       with tempfile.TemporaryDirectory() as tmpdirname:
        lowercase					:   int				  =Path(UpperCAmelCase		) / '''preprocessor_config.json'''
        lowercase					:   int				  =Path(UpperCAmelCase		) / '''config.json'''
        json.dump(
            {'''feature_extractor_type''': '''CLIPFeatureExtractor''', '''processor_class''': '''CLIPProcessor'''}     ,      open(UpperCAmelCase     ,      '''w'''		)     ,      )
        json.dump({'''model_type''': '''clip'''}     ,      open(UpperCAmelCase     ,      '''w'''		)		)

        lowercase					:   Optional[Any]				  =AutoImageProcessor.from_pretrained(UpperCAmelCase		)
        self.assertIsInstance(UpperCAmelCase     ,      UpperCAmelCase		)






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


       '''simple docstring'''

       with tempfile.TemporaryDirectory() as tmpdirname:
        lowercase					:   str				  =CLIPConfig()

        # Create a dummy config file with image_proceesor_type
        lowercase					:   Optional[int]				  =Path(UpperCAmelCase		) / '''preprocessor_config.json'''
        lowercase					:   Optional[Any]				  =Path(UpperCAmelCase		) / '''config.json'''
        json.dump(
            {'''image_processor_type''': '''CLIPImageProcessor''', '''processor_class''': '''CLIPProcessor'''}     ,      open(UpperCAmelCase     ,      '''w'''		)     ,      )
        json.dump({'''model_type''': '''clip'''}     ,      open(UpperCAmelCase     ,      '''w'''		)		)

        # remove image_processor_type to make sure config.json alone is enough to load image processor locally
        lowercase					:   Dict				  =AutoImageProcessor.from_pretrained(UpperCAmelCase		).to_dict()

        config_dict.pop('''image_processor_type'''		)
        lowercase					:   str				  =CLIPImageProcessor(**UpperCAmelCase		)

        # save in new folder
        model_config.save_pretrained(UpperCAmelCase		)
        config.save_pretrained(UpperCAmelCase		)

        lowercase					:   Optional[int]				  =AutoImageProcessor.from_pretrained(UpperCAmelCase		)

        # make sure private variable is not incorrectly saved
        lowercase					:   int				  =json.loads(config.to_json_string()		)
        self.assertTrue('''_processor_class''' not in dict_as_saved		)

       self.assertIsInstance(UpperCAmelCase     ,      UpperCAmelCase		)






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


       '''simple docstring'''

       with tempfile.TemporaryDirectory() as tmpdirname:
        lowercase					:   Dict				  =Path(UpperCAmelCase		) / '''preprocessor_config.json'''
        json.dump(
            {'''image_processor_type''': '''CLIPImageProcessor''', '''processor_class''': '''CLIPProcessor'''}     ,      open(UpperCAmelCase     ,      '''w'''		)     ,      )

        lowercase					:   Optional[Any]				  =AutoImageProcessor.from_pretrained(UpperCAmelCase		)
        self.assertIsInstance(UpperCAmelCase     ,      UpperCAmelCase		)






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


       '''simple docstring'''

       with self.assertRaisesRegex(
           UpperCAmelCase     ,      '''clip-base is not a local folder and is not a valid model identifier'''		):
        lowercase					:   Union[str, Any]				  =AutoImageProcessor.from_pretrained('''clip-base'''		)






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


       '''simple docstring'''

       with self.assertRaisesRegex(
           UpperCAmelCase     ,      R'''aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)'''		):
        lowercase					:   Any				  =AutoImageProcessor.from_pretrained(UpperCAmelCase     ,      revision='''aaaaaa'''		)






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


       '''simple docstring'''

       with self.assertRaisesRegex(
           UpperCAmelCase     ,      '''hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json.'''     ,      ):
        lowercase					:   Optional[int]				  =AutoImageProcessor.from_pretrained('''hf-internal-testing/config-no-model'''		)






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


       '''simple docstring'''

       with self.assertRaises(UpperCAmelCase		):
        lowercase					:   Dict				  =AutoImageProcessor.from_pretrained('''hf-internal-testing/test_dynamic_image_processor'''		)
       # If remote code is disabled, we can't load this config.
       with self.assertRaises(UpperCAmelCase		):
        lowercase					:   List[str]				  =AutoImageProcessor.from_pretrained(
            '''hf-internal-testing/test_dynamic_image_processor'''     ,      trust_remote_code=UpperCAmelCase		)

       lowercase					:   Union[str, Any]				  =AutoImageProcessor.from_pretrained(
           '''hf-internal-testing/test_dynamic_image_processor'''     ,      trust_remote_code=UpperCAmelCase		)
       self.assertEqual(image_processor.__class__.__name__     ,      '''NewImageProcessor'''		)

       # Test image processor can be reloaded.
       with tempfile.TemporaryDirectory() as tmp_dir:
        image_processor.save_pretrained(UpperCAmelCase		)
        lowercase					:   Any				  =AutoImageProcessor.from_pretrained(UpperCAmelCase     ,      trust_remote_code=UpperCAmelCase		)
       self.assertEqual(reloaded_image_processor.__class__.__name__     ,      '''NewImageProcessor'''		)






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


       '''simple docstring'''

       try:
        AutoConfig.register('''custom'''     ,      UpperCAmelCase		)
        AutoImageProcessor.register(UpperCAmelCase     ,      UpperCAmelCase		)
        # Trying to register something existing in the Transformers library will raise an error
        with self.assertRaises(UpperCAmelCase		):
         AutoImageProcessor.register(UpperCAmelCase     ,      UpperCAmelCase		)

        with tempfile.TemporaryDirectory() as tmpdirname:
         lowercase					:   Any				  =Path(UpperCAmelCase		) / '''preprocessor_config.json'''
         lowercase					:   str				  =Path(UpperCAmelCase		) / '''config.json'''
         json.dump(
             {'''feature_extractor_type''': '''CLIPFeatureExtractor''', '''processor_class''': '''CLIPProcessor'''}     ,      open(UpperCAmelCase     ,      '''w'''		)     ,      )
         json.dump({'''model_type''': '''clip'''}     ,      open(UpperCAmelCase     ,      '''w'''		)		)

         lowercase					:   Optional[int]				  =CustomImageProcessor.from_pretrained(UpperCAmelCase		)

        # Now that the config is registered, it can be used as any other config with the auto-API
        with tempfile.TemporaryDirectory() as tmp_dir:
         image_processor.save_pretrained(UpperCAmelCase		)
         lowercase					:   Dict				  =AutoImageProcessor.from_pretrained(UpperCAmelCase		)
         self.assertIsInstance(UpperCAmelCase     ,      UpperCAmelCase		)

       finally:
        if "custom" in CONFIG_MAPPING._extra_content:
         del CONFIG_MAPPING._extra_content["custom"]
        if CustomConfig in IMAGE_PROCESSOR_MAPPING._extra_content:
         del IMAGE_PROCESSOR_MAPPING._extra_content[CustomConfig]






      def 				A__      ( self			:		Any		)  ->  Any:


       '''simple docstring'''

       class UpperCAmelCase_					(    __A				):




             """simple docstring"""




             UpperCamelCase_								=			True

       try:
        AutoConfig.register('''custom'''     ,      UpperCAmelCase		)
        AutoImageProcessor.register(UpperCAmelCase     ,      UpperCAmelCase		)
        # If remote code is not set, the default is to use local
        lowercase					:   List[str]				  =AutoImageProcessor.from_pretrained('''hf-internal-testing/test_dynamic_image_processor'''		)
        self.assertEqual(image_processor.__class__.__name__     ,      '''NewImageProcessor'''		)
        self.assertTrue(image_processor.is_local		)

        # If remote code is disabled, we load the local one.
        lowercase					:   Tuple				  =AutoImageProcessor.from_pretrained(
            '''hf-internal-testing/test_dynamic_image_processor'''     ,      trust_remote_code=UpperCAmelCase		)
        self.assertEqual(image_processor.__class__.__name__     ,      '''NewImageProcessor'''		)
        self.assertTrue(image_processor.is_local		)

        # If remote is enabled, we load from the Hub
        lowercase					:   Dict				  =AutoImageProcessor.from_pretrained(
            '''hf-internal-testing/test_dynamic_image_processor'''     ,      trust_remote_code=UpperCAmelCase		)
        self.assertEqual(image_processor.__class__.__name__     ,      '''NewImageProcessor'''		)
        self.assertTrue(not hasattr(UpperCAmelCase     ,      '''is_local'''		)		)

       finally:
        if "custom" in CONFIG_MAPPING._extra_content:
         del CONFIG_MAPPING._extra_content["custom"]
        if CustomConfig in IMAGE_PROCESSOR_MAPPING._extra_content:
         del IMAGE_PROCESSOR_MAPPING._extra_content[CustomConfig]



