


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 			__lowerCAmelCase   (       snake_case_       ,     snake_case_       ,     unittest.TestCase    ):






  """simple docstring"""





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



  def     lowerCAmelCase__     (   self				:				int     )      ->       Dict:




     """simple docstring"""
     torch.manual_seed(0     )
     snake_case_    			=    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_    			=    DDIMScheduler(
         beta_start=0.00_085			, beta_end=0.012			, beta_schedule="scaled_linear"			, num_train_timesteps=1_0_0_0			, clip_sample=UpperCAmelCase__			, set_alpha_to_one=UpperCAmelCase__			, )
     torch.manual_seed(0     )
     snake_case_    			=    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_    			=    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			, )
     snake_case_    			=    CLIPTextModel(UpperCAmelCase__     )
     snake_case_    			=    CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"     )

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



  def     lowerCAmelCase__     (   self				:				Any			, _lowerCAmelCase				:				str			, _lowerCAmelCase				:				Dict=0     )      ->       Optional[int]:




     """simple docstring"""
     snake_case_    			=    floats_tensor((1, 3, 3_2, 3_2)			, rng=random.Random(UpperCAmelCase__     )     ).to(UpperCAmelCase__     )
     snake_case_    			=    image / 2 + 0.5
     if str(UpperCAmelCase__     ).startswith("mps"     ):
        snake_case_    			=    torch.manual_seed(UpperCAmelCase__     )
     else:
        snake_case_    			=    torch.Generator(device=UpperCAmelCase__     ).manual_seed(UpperCAmelCase__     )
     snake_case_    			=    {
         """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				:				int     )      ->       Tuple:




     """simple docstring"""
     snake_case_    			=    """cpu"""  # ensure determinism for the device-dependent torch.Generator

     snake_case_    			=    self.get_dummy_components()
     snake_case_    			=    CycleDiffusionPipeline(**UpperCAmelCase__     )
     snake_case_    			=    pipe.to(UpperCAmelCase__     )
     pipe.set_progress_bar_config(disable=UpperCAmelCase__     )

     snake_case_    			=    self.get_dummy_inputs(UpperCAmelCase__     )
     snake_case_    			=    pipe(**UpperCAmelCase__     )
     snake_case_    			=    output.images

     snake_case_    			=    images[0, -3:, -3:, -1]

     assert images.shape == (1, 3_2, 3_2, 3)
     snake_case_    			=    np.array([0.4_459, 0.4_943, 0.4_544, 0.6_643, 0.5_474, 0.4_327, 0.5_701, 0.5_959, 0.5_179]     )

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



  @unittest.skipIf(torch_device != "cuda"			, "This test requires a GPU"     )
  def     lowerCAmelCase__     (   self				:				int     )      ->       List[str]:




     """simple docstring"""
     snake_case_    			=    self.get_dummy_components()
     for name, module in components.items():
        if hasattr(UpperCAmelCase__			, "half"     ):
           snake_case_    			=    module.half()
     snake_case_    			=    CycleDiffusionPipeline(**UpperCAmelCase__     )
     snake_case_    			=    pipe.to(UpperCAmelCase__     )
     pipe.set_progress_bar_config(disable=UpperCAmelCase__     )

     snake_case_    			=    self.get_dummy_inputs(UpperCAmelCase__     )
     snake_case_    			=    pipe(**UpperCAmelCase__     )
     snake_case_    			=    output.images

     snake_case_    			=    images[0, -3:, -3:, -1]

     assert images.shape == (1, 3_2, 3_2, 3)
     snake_case_    			=    np.array([0.3_506, 0.4_543, 0.446, 0.4_575, 0.5_195, 0.4_155, 0.5_273, 0.518, 0.4_116]     )

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



  @skip_mps
  def     lowerCAmelCase__     (   self				:				int     )      ->       str:




     """simple docstring"""
     return super().test_save_load_local()



  @unittest.skip("non-deterministic pipeline"     )
  def     lowerCAmelCase__     (   self				:				List[str]     )      ->       Optional[int]:




     """simple docstring"""
     return super().test_inference_batch_single_identical()



  @skip_mps
  def     lowerCAmelCase__     (   self				:				Union[str, Any]     )      ->       List[Any]:




     """simple docstring"""
     return super().test_dict_tuple_outputs_equivalent()



  @skip_mps
  def     lowerCAmelCase__     (   self				:				Dict     )      ->       Dict:




     """simple docstring"""
     return super().test_save_load_optional_components()



  @skip_mps
  def     lowerCAmelCase__     (   self				:				Union[str, Any]     )      ->       Dict:




     """simple docstring"""
     return super().test_attention_slicing_forward_pass()






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






  """simple docstring"""





  def     lowerCAmelCase__     (   self				:				List[str]     )      ->       Dict:




     """simple docstring"""
     # clean up the VRAM after each test
     super().tearDown()
     gc.collect()
     torch.cuda.empty_cache()



  def     lowerCAmelCase__     (   self				:				Optional[int]     )      ->       Optional[int]:




     """simple docstring"""
     snake_case_    			=    load_image(
         "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
         "/cycle-diffusion/black_colored_car.png"     )
     snake_case_    			=    load_numpy(
         "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy"     )
     snake_case_    			=    init_image.resize((5_1_2, 5_1_2)     )

     snake_case_    			=    """CompVis/stable-diffusion-v1-4"""
     snake_case_    			=    DDIMScheduler.from_pretrained(UpperCAmelCase__			, subfolder="scheduler"     )
     snake_case_    			=    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()

     snake_case_    			=    """A black colored car"""
     snake_case_    			=    """A blue colored car"""

     snake_case_    			=    torch.manual_seed(0     )
     snake_case_    			=    pipe(
         prompt=UpperCAmelCase__			, source_prompt=UpperCAmelCase__			, image=UpperCAmelCase__			, num_inference_steps=1_0_0			, eta=0.1			, strength=0.85			, guidance_scale=3			, source_guidance_scale=1			, generator=UpperCAmelCase__			, output_type="np"			, )
     snake_case_    			=    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				:				Any     )      ->       Any:




     """simple docstring"""
     snake_case_    			=    load_image(
         "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
         "/cycle-diffusion/black_colored_car.png"     )
     snake_case_    			=    load_numpy(
         "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy"     )
     snake_case_    			=    init_image.resize((5_1_2, 5_1_2)     )

     snake_case_    			=    """CompVis/stable-diffusion-v1-4"""
     snake_case_    			=    DDIMScheduler.from_pretrained(UpperCAmelCase__			, subfolder="scheduler"     )
     snake_case_    			=    CycleDiffusionPipeline.from_pretrained(UpperCAmelCase__			, scheduler=UpperCAmelCase__			, safety_checker=UpperCAmelCase__     )

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

     snake_case_    			=    """A black colored car"""
     snake_case_    			=    """A blue colored car"""

     snake_case_    			=    torch.manual_seed(0     )
     snake_case_    			=    pipe(
         prompt=UpperCAmelCase__			, source_prompt=UpperCAmelCase__			, image=UpperCAmelCase__			, num_inference_steps=1_0_0			, eta=0.1			, strength=0.85			, guidance_scale=3			, source_guidance_scale=1			, generator=UpperCAmelCase__			, output_type="np"			, )
     snake_case_    			=    output.images

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












"""simple docstring"""




def 			lowerCAmelCase__     (     UpperCamelCase__     ,   UpperCamelCase__     ,   UpperCamelCase__     ,   UpperCamelCase__     ,   UpperCamelCase__     ,   ):



  '''simple docstring'''
  _a  :       Optional[Any]             =	[redshift, radiation_density, matter_density, dark_energy]
  if any(p < 0 for p in parameters				):
    raise ValueError("""All input parameters must be positive"""				)

  if any(p > 1 for p in parameters[1:4]				):
    raise ValueError("""Relative densities cannot be greater than one"""				)
  else:
    _a  :       Tuple             =	1 - (matter_density + radiation_density + dark_energy)

    _a  :       int             =	(
        radiation_density * (redshift + 1) ** 4
        + matter_density * (redshift + 1) ** 3
        + curvature * (redshift + 1) ** 2
        + dark_energy
    )

    _a  :       List[str]             =	hubble_constant * e_a ** (1 / 2)
    return hubble


if __name__ == "__main__":
 import doctest

 # run doctest
 doctest.testmod()

 # demo LCDM approximation
 _snake_case          =   0.3

 print(
     hubble_parameter(
         hubble_constant=68.3,
         radiation_density=1e-4,
         matter_density=matter_density,
         dark_energy=1 - matter_density,
         redshift=0,
     )
 )

