





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       A__     ( __snake_case ,     __snake_case ,     unittest.TestCase    ):
      _UpperCAmelCase			:str							=     CycleDiffusionPipeline
      _UpperCAmelCase			:Union[str, Any]							=     TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
          'negative_prompt',
          'height',
          'width',
          'negative_prompt_embeds',
      }
      _UpperCAmelCase			:List[Any]							=     PipelineTesterMixin.required_optional_params - {'latents'}
      _UpperCAmelCase			:int							=     TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'source_prompt'}    )
      _UpperCAmelCase			:str							=     IMAGE_TO_IMAGE_IMAGE_PARAMS
      _UpperCAmelCase			:Optional[Any]							=     IMAGE_TO_IMAGE_IMAGE_PARAMS





      def  __UpperCamelCase(				self     ):
       '''simple docstring'''




       torch.manual_seed(0     )
       UpperCamelCase			:      Any 			=     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 ,							)
       UpperCamelCase			:      Dict 			=     DDIMScheduler(
           beta_start=0.0_00_85 ,							beta_end=0.0_12 ,							beta_schedule="scaled_linear" ,							num_train_timesteps=1000 ,							clip_sample=A_ ,							set_alpha_to_one=A_ ,							)
       torch.manual_seed(0     )
       UpperCamelCase			:      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     )
       UpperCamelCase			:      List[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 ,							)
       UpperCamelCase			:      Dict 			=     CLIPTextModel(A_     )
       UpperCamelCase			:      Optional[int] 			=     CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"     )

       UpperCamelCase			:      Dict 			=     {
           "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     ):
       '''simple docstring'''




       UpperCamelCase			:      Union[str, Any] 			=     floats_tensor((1, 3, 32, 32) ,							rng=random.Random(A_     )     ).to(A_     )
       UpperCamelCase			:      Dict 			=     image / 2 + 0.5
       if str(A_     ).startswith("mps"     ):
        UpperCamelCase			:      Dict 			=     torch.manual_seed(A_     )
       else:
        UpperCamelCase			:      List[str] 			=     torch.Generator(device=A_     ).manual_seed(A_     )
       UpperCamelCase			:      Tuple 			=     {
           "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     ):
       '''simple docstring'''




       UpperCamelCase			:      List[Any] 			=     "cpu"  # ensure determinism for the device-dependent torch.Generator

       UpperCamelCase			:      Optional[int] 			=     self.get_dummy_components()
       UpperCamelCase			:      List[Any] 			=     CycleDiffusionPipeline(**A_     )
       UpperCamelCase			:      int 			=     pipe.to(A_     )
       pipe.set_progress_bar_config(disable=A_     )

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

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

       assert images.shape == (1, 32, 32, 3)
       UpperCamelCase			:      Union[str, Any] 			=     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     ):
       '''simple docstring'''




       UpperCamelCase			:      List[str] 			=     self.get_dummy_components()
       for name, module in components.items():
        if hasattr(A_ ,							"half"     ):
         UpperCamelCase			:      List[Any] 			=     module.half()
       UpperCamelCase			:      int 			=     CycleDiffusionPipeline(**A_     )
       UpperCamelCase			:      Optional[Any] 			=     pipe.to(A_     )
       pipe.set_progress_bar_config(disable=A_     )

       UpperCamelCase			:      Tuple 			=     self.get_dummy_inputs(A_     )
       UpperCamelCase			:      str 			=     pipe(**A_     )
       UpperCamelCase			:      Dict 			=     output.images

       UpperCamelCase			:      Dict 			=     images[0, -3:, -3:, -1]

       assert images.shape == (1, 32, 32, 3)
       UpperCamelCase			:      List[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     ):
       '''simple docstring'''




       return super().test_save_load_local()





      @unittest.skip("non-deterministic pipeline"     )
      def  __UpperCamelCase(				self     ):
       '''simple docstring'''




       return super().test_inference_batch_single_identical()





      @skip_mps
      def  __UpperCamelCase(				self     ):
       '''simple docstring'''




       return super().test_dict_tuple_outputs_equivalent()





      @skip_mps
      def  __UpperCamelCase(				self     ):
       '''simple docstring'''




       return super().test_save_load_optional_components()





      @skip_mps
      def  __UpperCamelCase(				self     ):
       '''simple docstring'''




       return super().test_attention_slicing_forward_pass()



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





      def  __UpperCamelCase(				self     ):
       '''simple docstring'''




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





      def  __UpperCamelCase(				self     ):
       '''simple docstring'''




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

       UpperCamelCase			:      str 			=     "CompVis/stable-diffusion-v1-4"
       UpperCamelCase			:      Union[str, Any] 			=     DDIMScheduler.from_pretrained(A_ ,							subfolder="scheduler"     )
       UpperCamelCase			:      Dict 			=     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()

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

       UpperCamelCase			:      Optional[Any] 			=     torch.manual_seed(0     )
       UpperCamelCase			:      Any 			=     pipe(
           prompt=A_ ,							source_prompt=A_ ,							image=A_ ,							num_inference_steps=100 ,							eta=0.1 ,							strength=0.85 ,							guidance_scale=3 ,							source_guidance_scale=1 ,							generator=A_ ,							output_type="np" ,							)
       UpperCamelCase			:      Dict 			=     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     ):
       '''simple docstring'''




       UpperCamelCase			:      str 			=     load_image(
           "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
           "/cycle-diffusion/black_colored_car.png"     )
       UpperCamelCase			:      Union[str, Any] 			=     load_numpy(
           "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy"     )
       UpperCamelCase			:      Union[str, Any] 			=     init_image.resize((512, 512)     )

       UpperCamelCase			:      str 			=     "CompVis/stable-diffusion-v1-4"
       UpperCamelCase			:      List[Any] 			=     DDIMScheduler.from_pretrained(A_ ,							subfolder="scheduler"     )
       UpperCamelCase			:      Optional[int] 			=     CycleDiffusionPipeline.from_pretrained(A_ ,							scheduler=A_ ,							safety_checker=A_     )

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

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

       UpperCamelCase			:      List[str] 			=     torch.manual_seed(0     )
       UpperCamelCase			:      List[str] 			=     pipe(
           prompt=A_ ,							source_prompt=A_ ,							image=A_ ,							num_inference_steps=100 ,							eta=0.1 ,							strength=0.85 ,							guidance_scale=3 ,							source_guidance_scale=1 ,							generator=A_ ,							output_type="np" ,							)
       UpperCamelCase			:      Union[str, Any] 			=     output.images

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








from ..utils import DummyObject, requires_backends



class       A__     ( metaclass=__snake_case    ):
      _UpperCAmelCase			:Tuple							=     ['note_seq']





      def __init__(				self ,							*A_ ,							**A_     ):
       '''simple docstring'''




       requires_backends(self ,							["note_seq"]     )





      @classmethod
      def  __UpperCamelCase(				cls ,							*A_ ,							**A_     ):
       '''simple docstring'''




       requires_backends(cls ,							["note_seq"]     )





      @classmethod
      def  __UpperCamelCase(				cls ,							*A_ ,							**A_     ):
       '''simple docstring'''




       requires_backends(cls ,							["note_seq"]     )
