




import gc
import random
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, CycleDiffusionPipeline, DDIMScheduler, UNetaDConditionModel
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

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


enable_full_determinism()




class 						__lowercase		(lowerCamelCase__					,		lowerCamelCase__					,		unittest.TestCase							):




      """simple docstring"""





      _snake_case												=						CycleDiffusionPipeline
      _snake_case												=						TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
          """negative_prompt""",
          """height""",
          """width""",
          """negative_prompt_embeds""",
      }
      _snake_case												=						PipelineTesterMixin.required_optional_params - {"""latents"""}
      _snake_case												=						TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"""source_prompt"""}							)
      _snake_case												=						IMAGE_TO_IMAGE_IMAGE_PARAMS
      _snake_case												=						IMAGE_TO_IMAGE_IMAGE_PARAMS

      def      UpperCAmelCase				(				self     )		->    int:
         torch.manual_seed(0     )
         snake_case				:    Union[str, Any]           =	UNetaDConditionModel(
             block_out_channels=(3_2, 6_4)    ,				layers_per_block=2    ,				sample_size=3_2    ,				in_channels=4    ,				out_channels=4    ,				down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")    ,				up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")    ,				cross_attention_dim=3_2    ,				)
         snake_case				:    List[str]           =	DDIMScheduler(
             beta_start=0.0_00_85    ,				beta_end=0.0_12    ,				beta_schedule="""scaled_linear"""    ,				num_train_timesteps=1_0_0_0    ,				clip_sample=A    ,				set_alpha_to_one=A    ,				)
         torch.manual_seed(0     )
         snake_case				:    Dict           =	AutoencoderKL(
             block_out_channels=[3_2, 6_4]    ,				in_channels=3    ,				out_channels=3    ,				down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]    ,				up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]    ,				latent_channels=4    ,				)
         torch.manual_seed(0     )
         snake_case				:    Union[str, Any]           =	CLIPTextConfig(
             bos_token_id=0    ,				eos_token_id=2    ,				hidden_size=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    ,				)
         snake_case				:    str           =	CLIPTextModel(A     )
         snake_case				:    Union[str, Any]           =	CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""     )

         snake_case				:    List[str]           =	{
             "unet": unet,
             "scheduler": scheduler,
             "vae": vae,
             "text_encoder": text_encoder,
             "tokenizer": tokenizer,
             "safety_checker": None,
             "feature_extractor": None,
         }
         return components

      def      UpperCAmelCase				(				self    ,				A    ,				A=0     )		->    Tuple:
         snake_case				:    List[str]           =	floats_tensor((1, 3, 3_2, 3_2)    ,				rng=random.Random(A     )     ).to(A     )
         snake_case				:    Dict           =	image / 2 + 0.5
         if str(A     ).startswith("""mps"""     ):
            snake_case				:    Optional[Any]           =	torch.manual_seed(A     )
         else:
            snake_case				:    Union[str, Any]           =	torch.Generator(device=A     ).manual_seed(A     )
         snake_case				:    Dict           =	{
             "prompt": "An astronaut riding an elephant",
             "source_prompt": "An astronaut riding a horse",
             "image": image,
             "generator": generator,
             "num_inference_steps": 2,
             "eta": 0.1,
             "strength": 0.8,
             "guidance_scale": 3,
             "source_guidance_scale": 1,
             "output_type": "numpy",
         }
         return inputs

      def      UpperCAmelCase				(				self     )		->    Any:
         snake_case				:    List[str]           =	"cpu"  # ensure determinism for the device-dependent torch.Generator

         snake_case				:    Optional[int]           =	self.get_dummy_components()
         snake_case				:    List[Any]           =	CycleDiffusionPipeline(**A     )
         snake_case				:    List[str]           =	pipe.to(A     )
         pipe.set_progress_bar_config(disable=A     )

         snake_case				:    Optional[int]           =	self.get_dummy_inputs(A     )
         snake_case				:    List[Any]           =	pipe(**A     )
         snake_case				:    Optional[Any]           =	output.images

         snake_case				:    Any           =	images[0, -3:, -3:, -1]

         assert images.shape == (1, 3_2, 3_2, 3)
         snake_case				:    List[str]           =	np.array([0.44_59, 0.49_43, 0.45_44, 0.66_43, 0.54_74, 0.43_27, 0.57_01, 0.59_59, 0.51_79]     )

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

      @unittest.skipIf(torch_device != """cuda"""    ,				"""This test requires a GPU"""     )
      def      UpperCAmelCase				(				self     )		->    List[str]:
         snake_case				:    Optional[int]           =	self.get_dummy_components()
         for name, module in components.items():
            if hasattr(A    ,				"""half"""     ):
               snake_case				:    int           =	module.half()
         snake_case				:    List[Any]           =	CycleDiffusionPipeline(**A     )
         snake_case				:    int           =	pipe.to(A     )
         pipe.set_progress_bar_config(disable=A     )

         snake_case				:    Union[str, Any]           =	self.get_dummy_inputs(A     )
         snake_case				:    Optional[Any]           =	pipe(**A     )
         snake_case				:    List[Any]           =	output.images

         snake_case				:    Optional[int]           =	images[0, -3:, -3:, -1]

         assert images.shape == (1, 3_2, 3_2, 3)
         snake_case				:    Any           =	np.array([0.35_06, 0.45_43, 0.4_46, 0.45_75, 0.51_95, 0.41_55, 0.52_73, 0.5_18, 0.41_16]     )

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

      @skip_mps
      def      UpperCAmelCase				(				self     )		->    Optional[Any]:
         return super().test_save_load_local()

      @unittest.skip("""non-deterministic pipeline"""     )
      def      UpperCAmelCase				(				self     )		->    Dict:
         return super().test_inference_batch_single_identical()

      @skip_mps
      def      UpperCAmelCase				(				self     )		->    Union[str, Any]:
         return super().test_dict_tuple_outputs_equivalent()

      @skip_mps
      def      UpperCAmelCase				(				self     )		->    List[str]:
         return super().test_save_load_optional_components()

      @skip_mps
      def      UpperCAmelCase				(				self     )		->    Any:
         return super().test_attention_slicing_forward_pass()




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




      """simple docstring"""





      def      UpperCAmelCase				(				self     )		->    str:
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()

      def      UpperCAmelCase				(				self     )		->    Dict:
         snake_case				:    Any           =	load_image(
             """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
             """/cycle-diffusion/black_colored_car.png"""     )
         snake_case				:    Tuple           =	load_numpy(
             """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy"""     )
         snake_case				:    List[str]           =	init_image.resize((5_1_2, 5_1_2)     )

         snake_case				:    List[str]           =	"CompVis/stable-diffusion-v1-4"
         snake_case				:    str           =	DDIMScheduler.from_pretrained(A    ,				subfolder="""scheduler"""     )
         snake_case				:    List[Any]           =	CycleDiffusionPipeline.from_pretrained(
             A    ,				scheduler=A    ,				safety_checker=A    ,				torch_dtype=torch.floataa    ,				revision="""fp16"""     )

         pipe.to(A     )
         pipe.set_progress_bar_config(disable=A     )
         pipe.enable_attention_slicing()

         snake_case				:    Dict           =	"A black colored car"
         snake_case				:    int           =	"A blue colored car"

         snake_case				:    Union[str, Any]           =	torch.manual_seed(0     )
         snake_case				:    Optional[Any]           =	pipe(
             prompt=A    ,				source_prompt=A    ,				image=A    ,				num_inference_steps=1_0_0    ,				eta=0.1    ,				strength=0.85    ,				guidance_scale=3    ,				source_guidance_scale=1    ,				generator=A    ,				output_type="""np"""    ,				)
         snake_case				:    Tuple           =	output.images

         # the values aren't exactly equal, but the images look the same visually
         assert np.abs(image - expected_image     ).max() < 5e-1

      def      UpperCAmelCase				(				self     )		->    Union[str, Any]:
         snake_case				:    str           =	load_image(
             """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
             """/cycle-diffusion/black_colored_car.png"""     )
         snake_case				:    List[Any]           =	load_numpy(
             """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy"""     )
         snake_case				:    Optional[int]           =	init_image.resize((5_1_2, 5_1_2)     )

         snake_case				:    Union[str, Any]           =	"CompVis/stable-diffusion-v1-4"
         snake_case				:    List[str]           =	DDIMScheduler.from_pretrained(A    ,				subfolder="""scheduler"""     )
         snake_case				:    Tuple           =	CycleDiffusionPipeline.from_pretrained(A    ,				scheduler=A    ,				safety_checker=A     )

         pipe.to(A     )
         pipe.set_progress_bar_config(disable=A     )
         pipe.enable_attention_slicing()

         snake_case				:    Union[str, Any]           =	"A black colored car"
         snake_case				:    Optional[Any]           =	"A blue colored car"

         snake_case				:    Optional[int]           =	torch.manual_seed(0     )
         snake_case				:    Optional[Any]           =	pipe(
             prompt=A    ,				source_prompt=A    ,				image=A    ,				num_inference_steps=1_0_0    ,				eta=0.1    ,				strength=0.85    ,				guidance_scale=3    ,				source_guidance_scale=1    ,				generator=A    ,				output_type="""np"""    ,				)
         snake_case				:    Union[str, Any]           =	output.images

         assert np.abs(image - expected_image     ).max() < 2e-2











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__)





