




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 						__lowercase		(UpperCamelCase__					,		UpperCamelCase__					,		UpperCamelCase__					,		unittest.TestCase							):




      """simple docstring"""





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

      def      UpperCAmelCase				(				self     )		->    Union[str, Any]:
         snake_case				:    List[str]           =	3_2
         snake_case				:    Optional[int]           =	embedder_hidden_size

         # image encoding components

         snake_case				:    List[str]           =	CLIPImageProcessor(crop_size=3_2    ,				size=3_2     )

         torch.manual_seed(0     )
         snake_case				:    Union[str, Any]           =	CLIPVisionModelWithProjection(
             CLIPVisionConfig(
                 hidden_size=A    ,				projection_dim=A    ,				num_hidden_layers=5    ,				num_attention_heads=4    ,				image_size=3_2    ,				intermediate_size=3_7    ,				patch_size=1    ,				)     )

         # regular denoising components

         torch.manual_seed(0     )
         snake_case				:    Tuple           =	StableUnCLIPImageNormalizer(embedding_dim=A     )
         snake_case				:    Optional[int]           =	DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""     )

         torch.manual_seed(0     )
         snake_case				:    Dict           =	CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""     )

         torch.manual_seed(0     )
         snake_case				:    Optional[int]           =	CLIPTextModel(
             CLIPTextConfig(
                 bos_token_id=0    ,				eos_token_id=2    ,				hidden_size=A    ,				projection_dim=3_2    ,				intermediate_size=3_7    ,				layer_norm_eps=1e-0_5    ,				num_attention_heads=4    ,				num_hidden_layers=5    ,				pad_token_id=1    ,				vocab_size=1_0_0_0    ,				)     )

         torch.manual_seed(0     )
         snake_case				:    Any           =	UNetaDConditionModel(
             sample_size=3_2    ,				in_channels=4    ,				out_channels=4    ,				down_block_types=("""CrossAttnDownBlock2D""", """DownBlock2D""")    ,				up_block_types=("""UpBlock2D""", """CrossAttnUpBlock2D""")    ,				block_out_channels=(3_2, 6_4)    ,				attention_head_dim=(2, 4)    ,				class_embed_type="""projection"""    ,				projection_class_embeddings_input_dim=embedder_projection_dim * 2    ,				cross_attention_dim=A    ,				layers_per_block=1    ,				upcast_attention=A    ,				use_linear_projection=A    ,				)

         torch.manual_seed(0     )
         snake_case				:    int           =	DDIMScheduler(
             beta_schedule="""scaled_linear"""    ,				beta_start=0.0_00_85    ,				beta_end=0.0_12    ,				prediction_type="""v_prediction"""    ,				set_alpha_to_one=A    ,				steps_offset=1    ,				)

         torch.manual_seed(0     )
         snake_case				:    Optional[int]           =	AutoencoderKL()

         snake_case				:    Optional[Any]           =	{
             # 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    ,				A    ,				A=0    ,				A=True     )		->    List[str]:
         if str(A     ).startswith("""mps"""     ):
            snake_case				:    List[str]           =	torch.manual_seed(A     )
         else:
            snake_case				:    Optional[int]           =	torch.Generator(device=A     ).manual_seed(A     )

         snake_case				:    Optional[int]           =	floats_tensor((1, 3, 3_2, 3_2)    ,				rng=random.Random(A     )     ).to(A     )

         if pil_image:
            snake_case				:    str           =	input_image * 0.5 + 0.5
            snake_case				:    Optional[Any]           =	input_image.clamp(0    ,				1     )
            snake_case				:    List[Any]           =	input_image.cpu().permute(0    ,				2    ,				3    ,				1     ).float().numpy()
            snake_case				:    int           =	DiffusionPipeline.numpy_to_pil(A     )[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     )		->    Any:
         snake_case				:    Any           =	"""cpu"""  # ensure determinism for the device-dependent torch.Generator
         snake_case				:    str           =	self.get_dummy_components()
         snake_case				:    Optional[Any]           =	StableUnCLIPImgaImgPipeline(**A     )
         snake_case				:    List[str]           =	sd_pipe.to(A     )
         sd_pipe.set_progress_bar_config(disable=A     )

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

         assert image.shape == (1, 3_2, 3_2, 3)
         snake_case				:    Dict           =	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     )		->    Union[str, Any]:
         snake_case				:    Any           =	torch_device in ["""cpu""", """mps"""]

         self._test_attention_slicing_forward_pass(test_max_difference=A     )

      def      UpperCAmelCase				(				self     )		->    List[Any]:
         snake_case				:    str           =	torch_device in ["""cpu""", """mps"""]

         self._test_inference_batch_single_identical(test_max_difference=A     )

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




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




      """simple docstring"""





      def      UpperCAmelCase				(				self     )		->    List[Any]:
         # clean up the VRAM after each test
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()

      def      UpperCAmelCase				(				self     )		->    Tuple:
         snake_case				:    int           =	load_image(
             """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"""     )

         snake_case				:    int           =	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"""     )

         snake_case				:    str           =	StableUnCLIPImgaImgPipeline.from_pretrained(
             """fusing/stable-unclip-2-1-l-img2img"""    ,				torch_dtype=torch.floataa     )
         pipe.to(A     )
         pipe.set_progress_bar_config(disable=A     )
         # 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()

         snake_case				:    int           =	torch.Generator(device="""cpu"""     ).manual_seed(0     )
         snake_case				:    Tuple           =	pipe(A    ,				"""anime turle"""    ,				generator=A    ,				output_type="""np"""     )

         snake_case				:    Dict           =	output.images[0]

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

         assert_mean_pixel_difference(A    ,				A     )

      def      UpperCAmelCase				(				self     )		->    List[Any]:
         snake_case				:    Any           =	load_image(
             """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"""     )

         snake_case				:    Optional[Any]           =	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"""     )

         snake_case				:    List[Any]           =	StableUnCLIPImgaImgPipeline.from_pretrained(
             """fusing/stable-unclip-2-1-h-img2img"""    ,				torch_dtype=torch.floataa     )
         pipe.to(A     )
         pipe.set_progress_bar_config(disable=A     )
         # 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()

         snake_case				:    Optional[int]           =	torch.Generator(device="""cpu"""     ).manual_seed(0     )
         snake_case				:    str           =	pipe(A    ,				"""anime turle"""    ,				generator=A    ,				output_type="""np"""     )

         snake_case				:    Optional[Any]           =	output.images[0]

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

         assert_mean_pixel_difference(A    ,				A     )

      def      UpperCAmelCase				(				self     )		->    Optional[int]:
         snake_case				:    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()

         snake_case				:    str           =	StableUnCLIPImgaImgPipeline.from_pretrained(
             """fusing/stable-unclip-2-1-h-img2img"""    ,				torch_dtype=torch.floataa     )
         snake_case				:    Union[str, Any]           =	pipe.to(A     )
         pipe.set_progress_bar_config(disable=A     )
         pipe.enable_attention_slicing()
         pipe.enable_sequential_cpu_offload()

         snake_case				:    List[str]           =	pipe(
             A    ,				"""anime turtle"""    ,				num_inference_steps=2    ,				output_type="""np"""    ,				)

         snake_case				:    str           =	torch.cuda.max_memory_allocated()
         # make sure that less than 7 GB is allocated
         assert mem_bytes < 7 * 1_0**9











from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
)


lowerCamelCase	:	str				=   {
    'configuration_roformer': ['ROFORMER_PRETRAINED_CONFIG_ARCHIVE_MAP', 'RoFormerConfig', 'RoFormerOnnxConfig'],
    'tokenization_roformer': ['RoFormerTokenizer'],
}

try:
       if not is_tokenizers_available():
              raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
       pass
else:
       lowerCamelCase	:	List[Any]				=   ['RoFormerTokenizerFast']

try:
       if not is_torch_available():
              raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
       pass
else:
       lowerCamelCase	:	Dict				=   [
           'ROFORMER_PRETRAINED_MODEL_ARCHIVE_LIST',
           'RoFormerForCausalLM',
           'RoFormerForMaskedLM',
           'RoFormerForMultipleChoice',
           'RoFormerForQuestionAnswering',
           'RoFormerForSequenceClassification',
           'RoFormerForTokenClassification',
           'RoFormerLayer',
           'RoFormerModel',
           'RoFormerPreTrainedModel',
           'load_tf_weights_in_roformer',
       ]


try:
       if not is_tf_available():
              raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
       pass
else:
       lowerCamelCase	:	Any				=   [
           'TF_ROFORMER_PRETRAINED_MODEL_ARCHIVE_LIST',
           'TFRoFormerForCausalLM',
           'TFRoFormerForMaskedLM',
           'TFRoFormerForMultipleChoice',
           'TFRoFormerForQuestionAnswering',
           'TFRoFormerForSequenceClassification',
           'TFRoFormerForTokenClassification',
           'TFRoFormerLayer',
           'TFRoFormerModel',
           'TFRoFormerPreTrainedModel',
       ]


try:
       if not is_flax_available():
              raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
       pass
else:
       lowerCamelCase	:	Union[str, Any]				=   [
           'FLAX_ROFORMER_PRETRAINED_MODEL_ARCHIVE_LIST',
           'FlaxRoFormerForMaskedLM',
           'FlaxRoFormerForMultipleChoice',
           'FlaxRoFormerForQuestionAnswering',
           'FlaxRoFormerForSequenceClassification',
           'FlaxRoFormerForTokenClassification',
           'FlaxRoFormerModel',
           'FlaxRoFormerPreTrainedModel',
       ]


if TYPE_CHECKING:
       from .configuration_roformer import ROFORMER_PRETRAINED_CONFIG_ARCHIVE_MAP, RoFormerConfig, RoFormerOnnxConfig
       from .tokenization_roformer import RoFormerTokenizer

       try:
              if not is_tokenizers_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .tokenization_roformer_fast import RoFormerTokenizerFast

       try:
              if not is_torch_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .modeling_roformer import (
                  ROFORMER_PRETRAINED_MODEL_ARCHIVE_LIST,
                  RoFormerForCausalLM,
                  RoFormerForMaskedLM,
                  RoFormerForMultipleChoice,
                  RoFormerForQuestionAnswering,
                  RoFormerForSequenceClassification,
                  RoFormerForTokenClassification,
                  RoFormerLayer,
                  RoFormerModel,
                  RoFormerPreTrainedModel,
                  load_tf_weights_in_roformer,
              )

       try:
              if not is_tf_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .modeling_tf_roformer import (
                  TF_ROFORMER_PRETRAINED_MODEL_ARCHIVE_LIST,
                  TFRoFormerForCausalLM,
                  TFRoFormerForMaskedLM,
                  TFRoFormerForMultipleChoice,
                  TFRoFormerForQuestionAnswering,
                  TFRoFormerForSequenceClassification,
                  TFRoFormerForTokenClassification,
                  TFRoFormerLayer,
                  TFRoFormerModel,
                  TFRoFormerPreTrainedModel,
              )

       try:
              if not is_flax_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .modeling_flax_roformer import (
                  FLAX_ROFORMER_PRETRAINED_MODEL_ARCHIVE_LIST,
                  FlaxRoFormerForMaskedLM,
                  FlaxRoFormerForMultipleChoice,
                  FlaxRoFormerForQuestionAnswering,
                  FlaxRoFormerForSequenceClassification,
                  FlaxRoFormerForTokenClassification,
                  FlaxRoFormerModel,
                  FlaxRoFormerPreTrainedModel,
              )


else:
       import sys

       lowerCamelCase	:	Optional[int]				=   _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)





