




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

    '''simple docstring'''



    lowerCamelCase_      :	List[Any]	     =      CycleDiffusionPipeline
    lowerCamelCase_      :	Union[str, Any]	     =      TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
        """negative_prompt""",
        """height""",
        """width""",
        """negative_prompt_embeds""",
    }
    lowerCamelCase_      :	Any	     =      PipelineTesterMixin.required_optional_params - {"""latents"""}
    lowerCamelCase_      :	int	     =      TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"""source_prompt"""}     )
    lowerCamelCase_      :	Optional[Any]	     =      IMAGE_TO_IMAGE_IMAGE_PARAMS
    lowerCamelCase_      :	str	     =      IMAGE_TO_IMAGE_IMAGE_PARAMS







    def     _lowercase     (    self     )     ->      Tuple:
           torch.manual_seed(0     )
           lowerCamelCase   :      int			 =					UNetaDConditionModel(
               block_out_channels=(32, 64)     ,  layers_per_block=2     ,  sample_size=32     ,  in_channels=4     ,  out_channels=4     ,  down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")     ,  up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")     ,  cross_attention_dim=32     ,  )
           lowerCamelCase   :      Optional[int]			 =					DDIMScheduler(
               beta_start=0.00085     ,  beta_end=0.012     ,  beta_schedule="scaled_linear"     ,  num_train_timesteps=1000     ,  clip_sample=UpperCamelCase__     ,  set_alpha_to_one=UpperCamelCase__     ,  )
           torch.manual_seed(0     )
           lowerCamelCase   :      Union[str, Any]			 =					AutoencoderKL(
               block_out_channels=[32, 64]     ,  in_channels=3     ,  out_channels=3     ,  down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]     ,  up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]     ,  latent_channels=4     ,  )
           torch.manual_seed(0     )
           lowerCamelCase   :      Any			 =					CLIPTextConfig(
               bos_token_id=0     ,  eos_token_id=2     ,  hidden_size=32     ,  intermediate_size=37     ,  layer_norm_eps=1e-05     ,  num_attention_heads=4     ,  num_hidden_layers=5     ,  pad_token_id=1     ,  vocab_size=1000     ,  )
           lowerCamelCase   :      List[str]			 =					CLIPTextModel(UpperCamelCase__     )
           lowerCamelCase   :      Any			 =					CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"     )

           lowerCamelCase   :      str			 =					{
               "unet": unet,
               "scheduler": scheduler,
               "vae": vae,
               "text_encoder": text_encoder,
               "tokenizer": tokenizer,
               "safety_checker": None,
               "feature_extractor": None,
           }
           return components







    def     _lowercase     (    self     ,  UpperCamelCase__     ,  UpperCamelCase__=0     )     ->      List[str]:
           lowerCamelCase   :      List[str]			 =					floats_tensor((1, 3, 32, 32)     ,  rng=random.Random(UpperCamelCase__     )     ).to(UpperCamelCase__     )
           lowerCamelCase   :      List[str]			 =					image / 2 + 0.5
           if str(UpperCamelCase__     ).startswith("mps"     ):
                  lowerCamelCase   :      Optional[Any]			 =					torch.manual_seed(UpperCamelCase__     )
           else:
                  lowerCamelCase   :      Any			 =					torch.Generator(device=UpperCamelCase__     ).manual_seed(UpperCamelCase__     )
           lowerCamelCase   :      Union[str, Any]			 =					{
               "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     _lowercase     (    self     )     ->      Dict:
           lowerCamelCase   :      Union[str, Any]			 =					"cpu"  # ensure determinism for the device-dependent torch.Generator

           lowerCamelCase   :      Optional[Any]			 =					self.get_dummy_components()
           lowerCamelCase   :      Optional[Any]			 =					CycleDiffusionPipeline(**UpperCamelCase__     )
           lowerCamelCase   :      Optional[int]			 =					pipe.to(UpperCamelCase__     )
           pipe.set_progress_bar_config(disable=UpperCamelCase__     )

           lowerCamelCase   :      Optional[Any]			 =					self.get_dummy_inputs(UpperCamelCase__     )
           lowerCamelCase   :      str			 =					pipe(**UpperCamelCase__     )
           lowerCamelCase   :      List[Any]			 =					output.images

           lowerCamelCase   :      str			 =					images[0, -3:, -3:, -1]

           assert images.shape == (1, 32, 32, 3)
           lowerCamelCase   :      List[str]			 =					np.array([0.4459, 0.4943, 0.4544, 0.6643, 0.5474, 0.4327, 0.5701, 0.5959, 0.5179]     )

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







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

           lowerCamelCase   :      Tuple			 =					self.get_dummy_inputs(UpperCamelCase__     )
           lowerCamelCase   :      Optional[Any]			 =					pipe(**UpperCamelCase__     )
           lowerCamelCase   :      Union[str, Any]			 =					output.images

           lowerCamelCase   :      Tuple			 =					images[0, -3:, -3:, -1]

           assert images.shape == (1, 32, 32, 3)
           lowerCamelCase   :      Union[str, Any]			 =					np.array([0.3506, 0.4543, 0.446, 0.4575, 0.5195, 0.4155, 0.5273, 0.518, 0.4116]     )

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







    @skip_mps
    def     _lowercase     (    self     )     ->      Any:
           return super().test_save_load_local()







    @unittest.skip("non-deterministic pipeline"     )
    def     _lowercase     (    self     )     ->      Optional[int]:
           return super().test_inference_batch_single_identical()







    @skip_mps
    def     _lowercase     (    self     )     ->      List[Any]:
           return super().test_dict_tuple_outputs_equivalent()







    @skip_mps
    def     _lowercase     (    self     )     ->      int:
           return super().test_save_load_optional_components()







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




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

    '''simple docstring'''



    def     _lowercase     (    self     )     ->      Dict:
           # clean up the VRAM after each test
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()







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

           lowerCamelCase   :      int			 =					"CompVis/stable-diffusion-v1-4"
           lowerCamelCase   :      Optional[Any]			 =					DDIMScheduler.from_pretrained(UpperCamelCase__     ,  subfolder="scheduler"     )
           lowerCamelCase   :      Any			 =					CycleDiffusionPipeline.from_pretrained(
               UpperCamelCase__     ,  scheduler=UpperCamelCase__     ,  safety_checker=UpperCamelCase__     ,  torch_dtype=torch.floataa     ,  revision="fp16"     )

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

           lowerCamelCase   :      List[Any]			 =					"A black colored car"
           lowerCamelCase   :      List[str]			 =					"A blue colored car"

           lowerCamelCase   :      Dict			 =					torch.manual_seed(0     )
           lowerCamelCase   :      Any			 =					pipe(
               prompt=UpperCamelCase__     ,  source_prompt=UpperCamelCase__     ,  image=UpperCamelCase__     ,  num_inference_steps=100     ,  eta=0.1     ,  strength=0.85     ,  guidance_scale=3     ,  source_guidance_scale=1     ,  generator=UpperCamelCase__     ,  output_type="np"     ,  )
           lowerCamelCase   :      Any			 =					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     _lowercase     (    self     )     ->      List[str]:
           lowerCamelCase   :      Optional[Any]			 =					load_image(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
               "/cycle-diffusion/black_colored_car.png"     )
           lowerCamelCase   :      Union[str, Any]			 =					load_numpy(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy"     )
           lowerCamelCase   :      Optional[int]			 =					init_image.resize((512, 512)     )

           lowerCamelCase   :      Tuple			 =					"CompVis/stable-diffusion-v1-4"
           lowerCamelCase   :      Optional[int]			 =					DDIMScheduler.from_pretrained(UpperCamelCase__     ,  subfolder="scheduler"     )
           lowerCamelCase   :      Optional[int]			 =					CycleDiffusionPipeline.from_pretrained(UpperCamelCase__     ,  scheduler=UpperCamelCase__     ,  safety_checker=UpperCamelCase__     )

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

           lowerCamelCase   :      List[Any]			 =					"A black colored car"
           lowerCamelCase   :      List[Any]			 =					"A blue colored car"

           lowerCamelCase   :      Any			 =					torch.manual_seed(0     )
           lowerCamelCase   :      Any			 =					pipe(
               prompt=UpperCamelCase__     ,  source_prompt=UpperCamelCase__     ,  image=UpperCamelCase__     ,  num_inference_steps=100     ,  eta=0.1     ,  strength=0.85     ,  guidance_scale=3     ,  source_guidance_scale=1     ,  generator=UpperCamelCase__     ,  output_type="np"     ,  )
           lowerCamelCase   :      Dict			 =					output.images

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












from typing import TYPE_CHECKING

from ....utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


SCREAMING_SNAKE_CASE__       :      Optional[Any]  =   {'configuration_mmbt': ['MMBTConfig']}

try:
       if not is_torch_available():
              raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
       pass
else:
       SCREAMING_SNAKE_CASE__       :      List[Any]  =   ['MMBTForClassification', 'MMBTModel', 'ModalEmbeddings']


if TYPE_CHECKING:
       from .configuration_mmbt import MMBTConfig

       try:
              if not is_torch_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .modeling_mmbt import MMBTForClassification, MMBTModel, ModalEmbeddings

else:
       import sys

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






