


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 			SCREAMING_SNAKE_CASE__  (    _UpperCAmelCase						,   _UpperCAmelCase						,   unittest.TestCase      ):
      snake_case__    :					int					       =						CycleDiffusionPipeline
      snake_case__    :					Optional[int]					       =						TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
          '''negative_prompt''',
          '''height''',
          '''width''',
          '''negative_prompt_embeds''',
      }
      snake_case__    :					Tuple					       =						PipelineTesterMixin.required_optional_params - {'''latents'''}
      snake_case__    :					int					       =						TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'''source_prompt'''}      )
      snake_case__    :					str					       =						IMAGE_TO_IMAGE_IMAGE_PARAMS
      snake_case__    :					List[str]					       =						IMAGE_TO_IMAGE_IMAGE_PARAMS
      def 	SCREAMING_SNAKE_CASE							(	self    :	int					)		->      Optional[int]:
             torch.manual_seed(0					)
             a_ :		Optional[int]							=  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					,							)
             a_ :		Any							=  DDIMScheduler(
                 beta_start=0.00085					,							beta_end=0.012					,							beta_schedule='scaled_linear'					,							num_train_timesteps=1_0_0_0					,							clip_sample=lowercase_					,							set_alpha_to_one=lowercase_					,							)
             torch.manual_seed(0					)
             a_ :		Optional[int]							=  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					)
             a_ :		List[str]							=  CLIPTextConfig(
                 bos_token_id=0					,							eos_token_id=2					,							hidden_size=3_2					,							intermediate_size=3_7					,							layer_norm_eps=1E-05					,							num_attention_heads=4					,							num_hidden_layers=5					,							pad_token_id=1					,							vocab_size=1_0_0_0					,							)
             a_ :		int							=  CLIPTextModel(lowercase_					)
             a_ :		Dict							=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'					)

             a_ :		Union[str, Any]							=  {
                 """unet""": unet,
                 """scheduler""": scheduler,
                 """vae""": vae,
                 """text_encoder""": text_encoder,
                 """tokenizer""": tokenizer,
                 """safety_checker""": None,
                 """feature_extractor""": None,
             }
             return components
      def 	SCREAMING_SNAKE_CASE							(	self    :	List[Any]					,							SCREAMING_SNAKE_CASE__    :	str					,							SCREAMING_SNAKE_CASE__    :	int=0					)		->      Tuple:
             a_ :		Any							=  floats_tensor((1, 3, 3_2, 3_2)					,							rng=random.Random(lowercase_					)					).to(lowercase_					)
             a_ :		int							=  image / 2 + 0.5
             if str(lowercase_					).startswith('mps'					):
                    a_ :		Dict							=  torch.manual_seed(lowercase_					)
             else:
                    a_ :		List[str]							=  torch.Generator(device=lowercase_					).manual_seed(lowercase_					)
             a_ :		Optional[int]							=  {
                 """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 	SCREAMING_SNAKE_CASE							(	self    :	Optional[int]					)		->      int:
             a_ :		Tuple							=  """cpu"""  # ensure determinism for the device-dependent torch.Generator

             a_ :		Any							=  self.get_dummy_components()
             a_ :		List[Any]							=  CycleDiffusionPipeline(**lowercase_					)
             a_ :		Any							=  pipe.to(lowercase_					)
             pipe.set_progress_bar_config(disable=lowercase_					)

             a_ :		Dict							=  self.get_dummy_inputs(lowercase_					)
             a_ :		Any							=  pipe(**lowercase_					)
             a_ :		Dict							=  output.images

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

             assert images.shape == (1, 3_2, 3_2, 3)
             a_ :		List[Any]							=  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 	SCREAMING_SNAKE_CASE							(	self    :	List[Any]					)		->      str:
             a_ :		int							=  self.get_dummy_components()
             for name, module in components.items():
                    if hasattr(lowercase_					,							'half'					):
                           a_ :		List[str]							=  module.half()
             a_ :		str							=  CycleDiffusionPipeline(**lowercase_					)
             a_ :		Optional[Any]							=  pipe.to(lowercase_					)
             pipe.set_progress_bar_config(disable=lowercase_					)

             a_ :		str							=  self.get_dummy_inputs(lowercase_					)
             a_ :		List[Any]							=  pipe(**lowercase_					)
             a_ :		int							=  output.images

             a_ :		Union[str, Any]							=  images[0, -3:, -3:, -1]

             assert images.shape == (1, 3_2, 3_2, 3)
             a_ :		List[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 	SCREAMING_SNAKE_CASE							(	self    :	List[str]					)		->      Optional[int]:
             return super().test_save_load_local()
      @unittest.skip('non-deterministic pipeline'					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	Optional[Any]					)		->      Dict:
             return super().test_inference_batch_single_identical()
      @skip_mps
      def 	SCREAMING_SNAKE_CASE							(	self    :	Dict					)		->      str:
             return super().test_dict_tuple_outputs_equivalent()
      @skip_mps
      def 	SCREAMING_SNAKE_CASE							(	self    :	str					)		->      Dict:
             return super().test_save_load_optional_components()

      @skip_mps
      def 	SCREAMING_SNAKE_CASE							(	self    :	str					)		->      Dict:
             return super().test_attention_slicing_forward_pass()
@slow
@require_torch_gpu
class 			SCREAMING_SNAKE_CASE__  (    unittest.TestCase      ):
      def 	SCREAMING_SNAKE_CASE							(	self    :	Any					)		->      Dict:
             # clean up the VRAM after each test
             super().tearDown()
             gc.collect()
             torch.cuda.empty_cache()
      def 	SCREAMING_SNAKE_CASE							(	self    :	List[str]					)		->      str:
             a_ :		Union[str, Any]							=  load_image(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
                 '/cycle-diffusion/black_colored_car.png'					)
             a_ :		List[str]							=  load_numpy(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy'					)
             a_ :		int							=  init_image.resize((5_1_2, 5_1_2)					)

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

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

             a_ :		Tuple							=  """A black colored car"""
             a_ :		Optional[Any]							=  """A blue colored car"""

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

             a_ :		Any							=  """CompVis/stable-diffusion-v1-4"""
             a_ :		int							=  DDIMScheduler.from_pretrained(lowercase_					,							subfolder='scheduler'					)
             a_ :		Any							=  CycleDiffusionPipeline.from_pretrained(lowercase_					,							scheduler=lowercase_					,							safety_checker=lowercase_					)

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

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

             a_ :		Any							=  torch.manual_seed(0					)
             a_ :		Optional[Any]							=  pipe(
                 prompt=lowercase_					,							source_prompt=lowercase_					,							image=lowercase_					,							num_inference_steps=1_0_0					,							eta=0.1					,							strength=0.85					,							guidance_scale=3					,							source_guidance_scale=1					,							generator=lowercase_					,							output_type='np'					,							)
             a_ :		List[str]							=  output.images

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






def 	SCREAMING_SNAKE_CASE_						(      __A			:      int    )     ->		int:

       """simple docstring"""

       if not isinstance(__A  ,   __A    ):
              raise ValueError('Input must be an integer'    )
       if input_num <= 0:
              raise ValueError('Input must be positive'    )
       return sum(
           divisor for divisor in range(1  ,   input_num // 2 + 1    ) if input_num % divisor == 0    )


if __name__ == "__main__":
   import doctest

   doctest.testmod()


