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



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






       def      __lowerCAmelCase			(						self					) ->Tuple:
           torch.manual_seed(0					)
           SCREAMING_SNAKE_CASE							:							Tuple								=      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							,    )
           SCREAMING_SNAKE_CASE							:							Tuple								=      DDIMScheduler(
               beta_start=0.0_0_0_8_5							,    beta_end=0.0_1_2							,    beta_schedule='''scaled_linear'''							,    num_train_timesteps=1000							,    clip_sample=__lowerCAmelCase							,    set_alpha_to_one=__lowerCAmelCase							,    )
           torch.manual_seed(0					)
           SCREAMING_SNAKE_CASE							:							str								=      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					)
           SCREAMING_SNAKE_CASE							:							Tuple								=      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							,    )
           SCREAMING_SNAKE_CASE							:							Any								=      CLIPTextModel(__lowerCAmelCase					)
           SCREAMING_SNAKE_CASE							:							int								=      CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''					)

           SCREAMING_SNAKE_CASE							:							Optional[Any]								=      {
               '''unet''': unet,
               '''scheduler''': scheduler,
               '''vae''': vae,
               '''text_encoder''': text_encoder,
               '''tokenizer''': tokenizer,
               '''safety_checker''': None,
               '''feature_extractor''': None,
           }
           return components






       def      __lowerCAmelCase			(						self							,    _lowerCamelCase							,    _lowerCamelCase=0					) ->Optional[int]:
           SCREAMING_SNAKE_CASE							:							Union[str, Any]								=      floats_tensor((1, 3, 32, 32)							,    rng=random.Random(__lowerCAmelCase					)					).to(__lowerCAmelCase					)
           SCREAMING_SNAKE_CASE							:							Tuple								=      image / 2 + 0.5
           if str(__lowerCAmelCase					).startswith('''mps'''					):
               SCREAMING_SNAKE_CASE							:							Union[str, Any]								=      torch.manual_seed(__lowerCAmelCase					)
           else:
               SCREAMING_SNAKE_CASE							:							str								=      torch.Generator(device=__lowerCAmelCase					).manual_seed(__lowerCAmelCase					)
           SCREAMING_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      __lowerCAmelCase			(						self					) ->str:
           SCREAMING_SNAKE_CASE							:							int								=      '''cpu'''  # ensure determinism for the device-dependent torch.Generator

           SCREAMING_SNAKE_CASE							:							Dict								=      self.get_dummy_components()
           SCREAMING_SNAKE_CASE							:							str								=      CycleDiffusionPipeline(**__lowerCAmelCase					)
           SCREAMING_SNAKE_CASE							:							Tuple								=      pipe.to(__lowerCAmelCase					)
           pipe.set_progress_bar_config(disable=__lowerCAmelCase					)

           SCREAMING_SNAKE_CASE							:							int								=      self.get_dummy_inputs(__lowerCAmelCase					)
           SCREAMING_SNAKE_CASE							:							str								=      pipe(**__lowerCAmelCase					)
           SCREAMING_SNAKE_CASE							:							Optional[int]								=      output.images

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

           assert images.shape == (1, 32, 32, 3)
           SCREAMING_SNAKE_CASE							:							str								=      np.array([0.4_4_5_9, 0.4_9_4_3, 0.4_5_4_4, 0.6_6_4_3, 0.5_4_7_4, 0.4_3_2_7, 0.5_7_0_1, 0.5_9_5_9, 0.5_1_7_9]					)

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






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

           SCREAMING_SNAKE_CASE							:							Union[str, Any]								=      self.get_dummy_inputs(__lowerCAmelCase					)
           SCREAMING_SNAKE_CASE							:							Tuple								=      pipe(**__lowerCAmelCase					)
           SCREAMING_SNAKE_CASE							:							str								=      output.images

           SCREAMING_SNAKE_CASE							:							List[Any]								=      images[0, -3:, -3:, -1]

           assert images.shape == (1, 32, 32, 3)
           SCREAMING_SNAKE_CASE							:							str								=      np.array([0.3_5_0_6, 0.4_5_4_3, 0.4_4_6, 0.4_5_7_5, 0.5_1_9_5, 0.4_1_5_5, 0.5_2_7_3, 0.5_1_8, 0.4_1_1_6]					)

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






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






       @unittest.skip('''non-deterministic pipeline'''					)
       def      __lowerCAmelCase			(						self					) ->Optional[Any]:
           return super().test_inference_batch_single_identical()






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






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






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






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



       """simple docstring"""






       def      __lowerCAmelCase			(						self					) ->List[Any]:
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()






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

           SCREAMING_SNAKE_CASE							:							int								=      '''CompVis/stable-diffusion-v1-4'''
           SCREAMING_SNAKE_CASE							:							Union[str, Any]								=      DDIMScheduler.from_pretrained(__lowerCAmelCase							,    subfolder='''scheduler'''					)
           SCREAMING_SNAKE_CASE							:							Optional[int]								=      CycleDiffusionPipeline.from_pretrained(
               __lowerCAmelCase							,    scheduler=__lowerCAmelCase							,    safety_checker=__lowerCAmelCase							,    torch_dtype=torch.floataa							,    revision='''fp16'''					)

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

           SCREAMING_SNAKE_CASE							:							Optional[int]								=      '''A black colored car'''
           SCREAMING_SNAKE_CASE							:							str								=      '''A blue colored car'''

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

           SCREAMING_SNAKE_CASE							:							Dict								=      '''CompVis/stable-diffusion-v1-4'''
           SCREAMING_SNAKE_CASE							:							Dict								=      DDIMScheduler.from_pretrained(__lowerCAmelCase							,    subfolder='''scheduler'''					)
           SCREAMING_SNAKE_CASE							:							List[Any]								=      CycleDiffusionPipeline.from_pretrained(__lowerCAmelCase							,    scheduler=__lowerCAmelCase							,    safety_checker=__lowerCAmelCase					)

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

           SCREAMING_SNAKE_CASE							:							int								=      '''A black colored car'''
           SCREAMING_SNAKE_CASE							:							Optional[int]								=      '''A blue colored car'''

           SCREAMING_SNAKE_CASE							:							Any								=      torch.manual_seed(0					)
           SCREAMING_SNAKE_CASE							:							List[str]								=      pipe(
               prompt=__lowerCAmelCase							,    source_prompt=__lowerCAmelCase							,    image=__lowerCAmelCase							,    num_inference_steps=100							,    eta=0.1							,    strength=0.8_5							,    guidance_scale=3							,    source_guidance_scale=1							,    generator=__lowerCAmelCase							,    output_type='''np'''							,    )
           SCREAMING_SNAKE_CASE							:							List[str]								=      output.images

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

'''simple docstring'''




import unittest

import numpy as np
import torch

from diffusers import PNDMPipeline, PNDMScheduler, UNetaDModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch, slow, torch_device


enable_full_determinism()






class 				a__(					unittest.TestCase	):


			'''simple docstring'''






			@property
			def 							a_       (   self):




					"""simple docstring"""


					torch.manual_seed(0)
					lowerCAmelCase   					=  UNetaDModel(
					    block_out_channels=(32, 64)       ,    layers_per_block=2       ,    sample_size=32       ,    in_channels=3       ,    out_channels=3       ,    down_block_types=("""DownBlock2D""", """AttnDownBlock2D""")       ,    up_block_types=("""AttnUpBlock2D""", """UpBlock2D""")       ,    )
					return model







			def 							a_       (   self):




					"""simple docstring"""


					lowerCAmelCase   					=  self.dummy_uncond_unet
					lowerCAmelCase   					=  PNDMScheduler()

					lowerCAmelCase   					=  PNDMPipeline(unet=__lowerCAmelCase       ,    scheduler=__lowerCAmelCase)
					pndm.to(__lowerCAmelCase)
					pndm.set_progress_bar_config(disable=__lowerCAmelCase)

					lowerCAmelCase   					=  torch.manual_seed(0)
					lowerCAmelCase   					=  pndm(generator=__lowerCAmelCase       ,    num_inference_steps=20       ,    output_type="""numpy""").images

					lowerCAmelCase   					=  torch.manual_seed(0)
					lowerCAmelCase   					=  pndm(generator=__lowerCAmelCase       ,    num_inference_steps=20       ,    output_type="""numpy"""       ,    return_dict=__lowerCAmelCase)[0]

					lowerCAmelCase   					=  image[0, -3:, -3:, -1]
					lowerCAmelCase   					=  image_from_tuple[0, -3:, -3:, -1]

					assert image.shape == (1, 32, 32, 3)
					lowerCAmelCase   					=  np.array([1.0, 1.0, 0.0, 1.0, 0.0, 1.0, 0.0, 0.0, 0.0])

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







@slow
@require_torch
class 				a__(					unittest.TestCase	):


			'''simple docstring'''






			def 							a_       (   self):




					"""simple docstring"""


					lowerCAmelCase   					=  """google/ddpm-cifar10-32"""

					lowerCAmelCase   					=  UNetaDModel.from_pretrained(__lowerCAmelCase)
					lowerCAmelCase   					=  PNDMScheduler()

					lowerCAmelCase   					=  PNDMPipeline(unet=__lowerCAmelCase       ,    scheduler=__lowerCAmelCase)
					pndm.to(__lowerCAmelCase)
					pndm.set_progress_bar_config(disable=__lowerCAmelCase)
					lowerCAmelCase   					=  torch.manual_seed(0)
					lowerCAmelCase   					=  pndm(generator=__lowerCAmelCase       ,    output_type="""numpy""").images

					lowerCAmelCase   					=  image[0, -3:, -3:, -1]

					assert image.shape == (1, 32, 32, 3)
					lowerCAmelCase   					=  np.array([0.1564, 0.14645, 0.1406, 0.14715, 0.12425, 0.14045, 0.13115, 0.12175, 0.125])

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




