





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'''

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






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




           '''simple docstring'''


           torch.manual_seed(0       )
           SCREAMING_SNAKE_CASE		=						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		=						DDIMScheduler(
               beta_start=0.00085						,beta_end=0.012						,beta_schedule="""scaled_linear"""						,num_train_timesteps=1000						,clip_sample=lowerCamelCase__						,set_alpha_to_one=lowerCamelCase__						,)
           torch.manual_seed(0       )
           SCREAMING_SNAKE_CASE		=						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		=						CLIPTextConfig(
               bos_token_id=0						,eos_token_id=2						,hidden_size=32						,intermediate_size=37						,layer_norm_eps=1e-0_5						,num_attention_heads=4						,num_hidden_layers=5						,pad_token_id=1						,vocab_size=1000						,)
           SCREAMING_SNAKE_CASE		=						CLIPTextModel(lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""       )

           SCREAMING_SNAKE_CASE		=						{
               """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						:    Union[str, Any]						,lowerCamelCase__						:    Any						,lowerCamelCase__						:    Tuple=0       )     ->		Union[str, Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						floats_tensor((1, 3, 32, 32)						,rng=random.Random(lowerCamelCase__       )       ).to(lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						image / 2 + 0.5
           if str(lowerCamelCase__       ).startswith("""mps"""       ):
                 SCREAMING_SNAKE_CASE		=						torch.manual_seed(lowerCamelCase__       )
           else:
                 SCREAMING_SNAKE_CASE		=						torch.Generator(device=lowerCamelCase__       ).manual_seed(lowerCamelCase__       )
           SCREAMING_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 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[Any]       )     ->		str:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						"""cpu"""  # ensure determinism for the device-dependent torch.Generator

           SCREAMING_SNAKE_CASE		=						self.get_dummy_components()
           SCREAMING_SNAKE_CASE		=						CycleDiffusionPipeline(**lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						pipe.to(lowerCamelCase__       )
           pipe.set_progress_bar_config(disable=lowerCamelCase__       )

           SCREAMING_SNAKE_CASE		=						self.get_dummy_inputs(lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						pipe(**lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						output.images

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

           assert images.shape == (1, 32, 32, 3)
           SCREAMING_SNAKE_CASE		=						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						:    Dict       )     ->		Tuple:




           '''simple docstring'''


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

           SCREAMING_SNAKE_CASE		=						self.get_dummy_inputs(lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						pipe(**lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						output.images

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

           assert images.shape == (1, 32, 32, 3)
           SCREAMING_SNAKE_CASE		=						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						:    str       )     ->		Optional[int]:




           '''simple docstring'''


           return super().test_save_load_local()






     @unittest.skip("""non-deterministic pipeline"""       )
     def 				SCREAMING_SNAKE_CASE__  (							self						:    Any       )     ->		List[Any]:




           '''simple docstring'''


           return super().test_inference_batch_single_identical()






     @skip_mps
     def 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[Any]       )     ->		List[Any]:




           '''simple docstring'''


           return super().test_dict_tuple_outputs_equivalent()






     @skip_mps
     def 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[int]       )     ->		Optional[Any]:




           '''simple docstring'''


           return super().test_save_load_optional_components()






     @skip_mps
     def 				SCREAMING_SNAKE_CASE__  (							self						:    Any       )     ->		Optional[Any]:




           '''simple docstring'''


           return super().test_attention_slicing_forward_pass()

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



     '''simple docstring'''

     def 				SCREAMING_SNAKE_CASE__  (							self						:    List[Any]       )     ->		Union[str, Any]:




           '''simple docstring'''


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






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Any       )     ->		int:




           '''simple docstring'''


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

           SCREAMING_SNAKE_CASE		=						"""CompVis/stable-diffusion-v1-4"""
           SCREAMING_SNAKE_CASE		=						DDIMScheduler.from_pretrained(lowerCamelCase__						,subfolder="""scheduler"""       )
           SCREAMING_SNAKE_CASE		=						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		=						"""A black colored car"""
           SCREAMING_SNAKE_CASE		=						"""A blue colored car"""

           SCREAMING_SNAKE_CASE		=						torch.manual_seed(0       )
           SCREAMING_SNAKE_CASE		=						pipe(
               prompt=lowerCamelCase__						,source_prompt=lowerCamelCase__						,image=lowerCamelCase__						,num_inference_steps=100						,eta=0.1						,strength=0.85						,guidance_scale=3						,source_guidance_scale=1						,generator=lowerCamelCase__						,output_type="""np"""						,)
           SCREAMING_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 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[int]       )     ->		List[str]:




           '''simple docstring'''


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

           SCREAMING_SNAKE_CASE		=						"""CompVis/stable-diffusion-v1-4"""
           SCREAMING_SNAKE_CASE		=						DDIMScheduler.from_pretrained(lowerCamelCase__						,subfolder="""scheduler"""       )
           SCREAMING_SNAKE_CASE		=						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		=						"""A black colored car"""
           SCREAMING_SNAKE_CASE		=						"""A blue colored car"""

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

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









from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_sentencepiece_available,
    is_tokenizers_available,
    is_torch_available,
)


SCREAMING_SNAKE_CASE_						   =       {
    """configuration_llama""": ["""LLAMA_PRETRAINED_CONFIG_ARCHIVE_MAP""", """LlamaConfig"""],
}

try:
      if not is_sentencepiece_available():
            raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
      pass
else:
      SCREAMING_SNAKE_CASE_						   =       ["""LlamaTokenizer"""]

try:
      if not is_tokenizers_available():
            raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
      pass
else:
      SCREAMING_SNAKE_CASE_						   =       ["""LlamaTokenizerFast"""]

try:
      if not is_torch_available():
            raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
      pass
else:
      SCREAMING_SNAKE_CASE_						   =       [
          """LlamaForCausalLM""",
          """LlamaModel""",
          """LlamaPreTrainedModel""",
          """LlamaForSequenceClassification""",
      ]


if TYPE_CHECKING:
      from .configuration_llama import LLAMA_PRETRAINED_CONFIG_ARCHIVE_MAP, LlamaConfig

      try:
            if not is_sentencepiece_available():
                  raise OptionalDependencyNotAvailable()
      except OptionalDependencyNotAvailable:
            pass
      else:
            from .tokenization_llama import LlamaTokenizer

      try:
            if not is_tokenizers_available():
                  raise OptionalDependencyNotAvailable()
      except OptionalDependencyNotAvailable:
            pass
      else:
            from .tokenization_llama_fast import LlamaTokenizerFast

      try:
            if not is_torch_available():
                  raise OptionalDependencyNotAvailable()
      except OptionalDependencyNotAvailable:
            pass
      else:
            from .modeling_llama import LlamaForCausalLM, LlamaForSequenceClassification, LlamaModel, LlamaPreTrainedModel


else:
      import sys

      SCREAMING_SNAKE_CASE_						   =       _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)


