



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




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


   def 			UpperCAmelCase		(						self  :			Optional[Any])	->				Tuple:




         """simple docstring"""

         torch.manual_seed(0)
         lowercase__     =  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			,				)
         lowercase__     =  DDIMScheduler(
             beta_start=0.0_00_85			,				beta_end=0.0_12			,				beta_schedule='scaled_linear'			,				num_train_timesteps=10_00			,				clip_sample=lowerCAmelCase			,				set_alpha_to_one=lowerCAmelCase			,				)
         torch.manual_seed(0)
         lowercase__     =  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)
         lowercase__     =  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=10_00			,				)
         lowercase__     =  CLIPTextModel(lowerCAmelCase)
         lowercase__     =  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip')

         lowercase__     =  {
             'unet': unet,
             'scheduler': scheduler,
             'vae': vae,
             'text_encoder': text_encoder,
             'tokenizer': tokenizer,
             'safety_checker': None,
             'feature_extractor': None,
         }
         return components


   def 			UpperCAmelCase		(						self  :			List[str]			,				lowerCAmelCase  :			Optional[Any]			,				lowerCAmelCase  :			int=0)	->				Any:




         """simple docstring"""

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




         """simple docstring"""

         lowercase__     =  'cpu'  # ensure determinism for the device-dependent torch.Generator

         lowercase__     =  self.get_dummy_components()
         lowercase__     =  CycleDiffusionPipeline(**lowerCAmelCase)
         lowercase__     =  pipe.to(lowerCAmelCase)
         pipe.set_progress_bar_config(disable=lowerCAmelCase)

         lowercase__     =  self.get_dummy_inputs(lowerCAmelCase)
         lowercase__     =  pipe(**lowerCAmelCase)
         lowercase__     =  output.images

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

         assert images.shape == (1, 32, 32, 3)
         lowercase__     =  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  :			Dict)	->				Optional[Any]:




         """simple docstring"""

         lowercase__     =  self.get_dummy_components()
         for name, module in components.items():
               if hasattr(lowerCAmelCase			,				'half'):
                     lowercase__     =  module.half()
         lowercase__     =  CycleDiffusionPipeline(**lowerCAmelCase)
         lowercase__     =  pipe.to(lowerCAmelCase)
         pipe.set_progress_bar_config(disable=lowerCAmelCase)

         lowercase__     =  self.get_dummy_inputs(lowerCAmelCase)
         lowercase__     =  pipe(**lowerCAmelCase)
         lowercase__     =  output.images

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

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




         """simple docstring"""

         return super().test_save_load_local()


   @unittest.skip('non-deterministic pipeline')
   def 			UpperCAmelCase		(						self  :			List[Any])	->				Tuple:




         """simple docstring"""

         return super().test_inference_batch_single_identical()


   @skip_mps
   def 			UpperCAmelCase		(						self  :			Any)	->				str:




         """simple docstring"""

         return super().test_dict_tuple_outputs_equivalent()


   @skip_mps
   def 			UpperCAmelCase		(						self  :			int)	->				List[str]:




         """simple docstring"""

         return super().test_save_load_optional_components()



   @skip_mps
   def 			UpperCAmelCase		(						self  :			Dict)	->				int:




         """simple docstring"""

         return super().test_attention_slicing_forward_pass()



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

   '''simple docstring'''




   def 			UpperCAmelCase		(						self  :			Optional[Any])	->				List[str]:




         """simple docstring"""

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


   def 			UpperCAmelCase		(						self  :			Optional[int])	->				Dict:




         """simple docstring"""

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

         lowercase__     =  'CompVis/stable-diffusion-v1-4'
         lowercase__     =  DDIMScheduler.from_pretrained(lowerCAmelCase			,				subfolder='scheduler')
         lowercase__     =  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()

         lowercase__     =  'A black colored car'
         lowercase__     =  'A blue colored car'

         lowercase__     =  torch.manual_seed(0)
         lowercase__     =  pipe(
             prompt=lowerCAmelCase			,				source_prompt=lowerCAmelCase			,				image=lowerCAmelCase			,				num_inference_steps=1_00			,				eta=0.1			,				strength=0.85			,				guidance_scale=3			,				source_guidance_scale=1			,				generator=lowerCAmelCase			,				output_type='np'			,				)
         lowercase__     =  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  :			str)	->				Optional[int]:




         """simple docstring"""

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

         lowercase__     =  'CompVis/stable-diffusion-v1-4'
         lowercase__     =  DDIMScheduler.from_pretrained(lowerCAmelCase			,				subfolder='scheduler')
         lowercase__     =  CycleDiffusionPipeline.from_pretrained(lowerCAmelCase			,				scheduler=lowerCAmelCase			,				safety_checker=lowerCAmelCase)

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

         lowercase__     =  'A black colored car'
         lowercase__     =  'A blue colored car'

         lowercase__     =  torch.manual_seed(0)
         lowercase__     =  pipe(
             prompt=lowerCAmelCase			,				source_prompt=lowerCAmelCase			,				image=lowerCAmelCase			,				num_inference_steps=1_00			,				eta=0.1			,				strength=0.85			,				guidance_scale=3			,				source_guidance_scale=1			,				generator=lowerCAmelCase			,				output_type='np'			,				)
         lowercase__     =  output.images

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





from typing import Mapping

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxSeqaSeqConfigWithPast
from ...utils import logging


a__      :				Union[str, Any]   	=			logging.get_logger(__name__)

a__      :				Optional[Any]   	=			{
    "google/umt5-small": "https://huggingface.co/google/umt5-small/resolve/main/config.json",
    # See all umt5 models at https://huggingface.co/models?filter=umt5
}



class       UpperCAmelCase__(							lowerCamelCase	):

   '''simple docstring'''




   A    :  Union[str, Any]       				=    "umt5"
   A    :  List[str]       				=    ["past_key_values"]


   def __init__(						self  :			List[Any]			,				lowerCAmelCase  :			Optional[int]=25_01_12			,				lowerCAmelCase  :			str=5_12			,				lowerCAmelCase  :			List[Any]=64			,				lowerCAmelCase  :			Optional[int]=10_24			,				lowerCAmelCase  :			Union[str, Any]=8			,				lowerCAmelCase  :			Tuple=None			,				lowerCAmelCase  :			Optional[Any]=6			,				lowerCAmelCase  :			int=32			,				lowerCAmelCase  :			int=1_28			,				lowerCAmelCase  :			List[str]=0.1			,				lowerCAmelCase  :			List[str]=1E-6			,				lowerCAmelCase  :			Optional[int]=1.0			,				lowerCAmelCase  :			Optional[Any]="gated-gelu"			,				lowerCAmelCase  :			List[Any]=True			,				lowerCAmelCase  :			List[str]=True			,				lowerCAmelCase  :			List[Any]="T5Tokenizer"			,				lowerCAmelCase  :			str=True			,				lowerCAmelCase  :			Optional[int]=0			,				lowerCAmelCase  :			Tuple=1			,				lowerCAmelCase  :			Any=0			,				**lowerCAmelCase  :			int			,				)	->				str:




         """simple docstring"""

         super().__init__(
             is_encoder_decoder=lowerCAmelCase			,				tokenizer_class=lowerCAmelCase			,				tie_word_embeddings=lowerCAmelCase			,				pad_token_id=lowerCAmelCase			,				eos_token_id=lowerCAmelCase			,				decoder_start_token_id=lowerCAmelCase			,				**lowerCAmelCase			,				)
         lowercase__     =  vocab_size
         lowercase__     =  d_model
         lowercase__     =  d_kv
         lowercase__     =  d_ff
         lowercase__     =  num_layers
         lowercase__     =  (
             num_decoder_layers if num_decoder_layers is not None else self.num_layers
         )  # default = symmetry
         lowercase__     =  num_heads
         lowercase__     =  relative_attention_num_buckets
         lowercase__     =  relative_attention_max_distance
         lowercase__     =  dropout_rate
         lowercase__     =  layer_norm_epsilon
         lowercase__     =  initializer_factor
         lowercase__     =  feed_forward_proj
         lowercase__     =  use_cache

         lowercase__     =  self.feed_forward_proj.split('-')
         lowercase__     =  act_info[-1]
         lowercase__     =  act_info[0] == 'gated'

         if len(lowerCAmelCase) > 1 and act_info[0] != "gated" or len(lowerCAmelCase) > 2:
               raise ValueError(
                   f'''`feed_forward_proj`: {feed_forward_proj} is not a valid activation function of the dense layer.'''
                   'Please make sure `feed_forward_proj` is of the format `gated-{ACT_FN}` or `{ACT_FN}`, e.g. '
                   '\'gated-gelu\' or \'relu\'')

         if feed_forward_proj == "gated-gelu":
               lowercase__     =  'gelu_new'


   @property
   def 			UpperCAmelCase		(						self  :			Union[str, Any])	->				Dict:




         """simple docstring"""

         return self.d_model


   @property
   def 			UpperCAmelCase		(						self  :			List[str])	->				Union[str, Any]:




         """simple docstring"""

         return self.num_heads



   @property
   def 			UpperCAmelCase		(						self  :			Optional[int])	->				Optional[Any]:




         """simple docstring"""

         return self.num_layers



class       UpperCAmelCase__(							lowerCamelCase	):

   '''simple docstring'''




   @property
   # Copied from transformers.models.t5.configuration_t5.T5OnnxConfig.inputs
   def 			UpperCAmelCase		(						self  :			Optional[int])	->				Mapping[str, Mapping[int, str]]:




         """simple docstring"""

         lowercase__     =  {
             'input_ids': {0: 'batch', 1: 'encoder_sequence'},
             'attention_mask': {0: 'batch', 1: 'encoder_sequence'},
         }
         if self.use_past:
               lowercase__     =  'past_encoder_sequence + sequence'
               lowercase__     =  {0: 'batch'}
               lowercase__     =  {0: 'batch', 1: 'past_decoder_sequence + sequence'}
         else:
               lowercase__     =  {0: 'batch', 1: 'decoder_sequence'}
               lowercase__     =  {0: 'batch', 1: 'decoder_sequence'}

         if self.use_past:
               self.fill_with_past_key_values_(lowerCAmelCase			,				direction='inputs')

         return common_inputs


   @property
   # Copied from transformers.models.t5.configuration_t5.T5OnnxConfig.default_onnx_opset
   def 			UpperCAmelCase		(						self  :			int)	->				int:




         """simple docstring"""

         return 13



   @property
   def 			UpperCAmelCase		(						self  :			Optional[Any])	->				float:




         """simple docstring"""

         return 5E-4
