


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 	__lowercase    (__SCREAMING_SNAKE_CASE	,       __SCREAMING_SNAKE_CASE	,       unittest.TestCase  ):





							"""simple docstring"""







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






							def 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




										torch.manual_seed(0       )
										SCREAMING_SNAKE_CASE_		:     Dict					=					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			,						)
										SCREAMING_SNAKE_CASE_		:     int					=					DDIMScheduler(
										    beta_start=0.00_085			,						beta_end=0.012			,						beta_schedule='scaled_linear'			,						num_train_timesteps=1_0_0_0			,						clip_sample=lowerCAmelCase__			,						set_alpha_to_one=lowerCAmelCase__			,						)
										torch.manual_seed(0       )
										SCREAMING_SNAKE_CASE_		:     Any					=					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       )
										SCREAMING_SNAKE_CASE_		:     Optional[Any]					=					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			,						)
										SCREAMING_SNAKE_CASE_		:     Any					=					CLIPTextModel(lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     Union[str, Any]					=					CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'       )

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






							def 					UpperCamelCase__     (      self			,						lowerCAmelCase__			,						lowerCAmelCase__=0       ):





										"""simple docstring"""




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





										"""simple docstring"""




										SCREAMING_SNAKE_CASE_		:     Dict					=					'cpu'  # ensure determinism for the device-dependent torch.Generator

										SCREAMING_SNAKE_CASE_		:     str					=					self.get_dummy_components()
										SCREAMING_SNAKE_CASE_		:     int					=					CycleDiffusionPipeline(**lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     int					=					pipe.to(lowerCAmelCase__       )
										pipe.set_progress_bar_config(disable=lowerCAmelCase__       )

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

										SCREAMING_SNAKE_CASE_		:     Any					=					images[0, -3:, -3:, -1]

										assert images.shape == (1, 3_2, 3_2, 3)
										SCREAMING_SNAKE_CASE_		:     Optional[int]					=					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 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




										SCREAMING_SNAKE_CASE_		:     Union[str, Any]					=					self.get_dummy_components()
										for name, module in components.items():
													if hasattr(lowerCAmelCase__			,						'half'       ):
																SCREAMING_SNAKE_CASE_		:     Optional[Any]					=					module.half()
										SCREAMING_SNAKE_CASE_		:     Tuple					=					CycleDiffusionPipeline(**lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     str					=					pipe.to(lowerCAmelCase__       )
										pipe.set_progress_bar_config(disable=lowerCAmelCase__       )

										SCREAMING_SNAKE_CASE_		:     int					=					self.get_dummy_inputs(lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     Dict					=					pipe(**lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     Any					=					output.images

										SCREAMING_SNAKE_CASE_		:     str					=					images[0, -3:, -3:, -1]

										assert images.shape == (1, 3_2, 3_2, 3)
										SCREAMING_SNAKE_CASE_		:     Union[str, Any]					=					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 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




										return super().test_save_load_local()






							@unittest.skip('non-deterministic pipeline'       )
							def 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




										return super().test_inference_batch_single_identical()






							@skip_mps
							def 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




										return super().test_dict_tuple_outputs_equivalent()






							@skip_mps
							def 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




										return super().test_save_load_optional_components()






							@skip_mps
							def 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




										return super().test_attention_slicing_forward_pass()







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





							"""simple docstring"""







							def 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




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






							def 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




										SCREAMING_SNAKE_CASE_		:     Union[str, Any]					=					load_image(
										    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
										    '/cycle-diffusion/black_colored_car.png'       )
										SCREAMING_SNAKE_CASE_		:     str					=					load_numpy(
										    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy'       )
										SCREAMING_SNAKE_CASE_		:     Tuple					=					init_image.resize((5_1_2, 5_1_2)       )

										SCREAMING_SNAKE_CASE_		:     List[str]					=					'CompVis/stable-diffusion-v1-4'
										SCREAMING_SNAKE_CASE_		:     Union[str, Any]					=					DDIMScheduler.from_pretrained(lowerCAmelCase__			,						subfolder='scheduler'       )
										SCREAMING_SNAKE_CASE_		:     List[str]					=					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_		:     Dict					=					'A black colored car'
										SCREAMING_SNAKE_CASE_		:     Any					=					'A blue colored car'

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





										"""simple docstring"""




										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_		:     int					=					load_numpy(
										    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy'       )
										SCREAMING_SNAKE_CASE_		:     Union[str, Any]					=					init_image.resize((5_1_2, 5_1_2)       )

										SCREAMING_SNAKE_CASE_		:     Optional[int]					=					'CompVis/stable-diffusion-v1-4'
										SCREAMING_SNAKE_CASE_		:     Tuple					=					DDIMScheduler.from_pretrained(lowerCAmelCase__			,						subfolder='scheduler'       )
										SCREAMING_SNAKE_CASE_		:     List[str]					=					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_		:     Optional[Any]					=					'A black colored car'
										SCREAMING_SNAKE_CASE_		:     Any					=					'A blue colored car'

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

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









import copy
import os
from typing import Union

from ...configuration_utils import PretrainedConfig
from ...utils import logging


lowerCAmelCase__			:      Tuple								=logging.get_logger(__name__)

lowerCAmelCase__			:      Optional[int]								={
    'microsoft/git-base': 'https://huggingface.co/microsoft/git-base/resolve/main/config.json',
}







class 	__lowercase    (__SCREAMING_SNAKE_CASE  ):





							"""simple docstring"""







							_UpperCAmelCase      =						"""git_vision_model"""






							def __init__(      self			,						lowerCAmelCase__=7_6_8			,						lowerCAmelCase__=3_0_7_2			,						lowerCAmelCase__=1_2			,						lowerCAmelCase__=1_2			,						lowerCAmelCase__=3			,						lowerCAmelCase__=2_2_4			,						lowerCAmelCase__=1_6			,						lowerCAmelCase__="quick_gelu"			,						lowerCAmelCase__=1E-5			,						lowerCAmelCase__=0.0			,						lowerCAmelCase__=0.02			,						**lowerCAmelCase__			,						):





										"""simple docstring"""




										super().__init__(**lowerCAmelCase__       )

										SCREAMING_SNAKE_CASE_		:     Optional[int]					=					hidden_size
										SCREAMING_SNAKE_CASE_		:     Optional[int]					=					intermediate_size
										SCREAMING_SNAKE_CASE_		:     Any					=					num_hidden_layers
										SCREAMING_SNAKE_CASE_		:     Dict					=					num_attention_heads
										SCREAMING_SNAKE_CASE_		:     Tuple					=					num_channels
										SCREAMING_SNAKE_CASE_		:     Tuple					=					patch_size
										SCREAMING_SNAKE_CASE_		:     Any					=					image_size
										SCREAMING_SNAKE_CASE_		:     Dict					=					initializer_range
										SCREAMING_SNAKE_CASE_		:     List[str]					=					attention_dropout
										SCREAMING_SNAKE_CASE_		:     Any					=					layer_norm_eps
										SCREAMING_SNAKE_CASE_		:     Union[str, Any]					=					hidden_act






							@classmethod
							def 					UpperCamelCase__     (      cls			,						lowerCAmelCase__			,						**lowerCAmelCase__       ):





										"""simple docstring"""




										cls._set_token_in_kwargs(lowerCAmelCase__       )

										SCREAMING_SNAKE_CASE_		,      SCREAMING_SNAKE_CASE_		:     List[str]					=					cls.get_config_dict(lowerCAmelCase__			,						**lowerCAmelCase__       )

										# get the vision config dict if we are loading from GITConfig
										if config_dict.get('model_type'       ) == "git":
													SCREAMING_SNAKE_CASE_		:     Union[str, Any]					=					config_dict['vision_config']

										if "model_type" in config_dict and hasattr(cls			,						'model_type'       ) and config_dict["model_type"] != cls.model_type:
													logger.warning(
													    F'''You are using a model of type {config_dict['model_type']} to instantiate a model of type '''
													    F'''{cls.model_type}. This is not supported for all configurations of models and can yield errors.'''       )

										return cls.from_dict(lowerCAmelCase__			,						**lowerCAmelCase__       )







class 	__lowercase    (__SCREAMING_SNAKE_CASE  ):





							"""simple docstring"""







							_UpperCAmelCase      =						"""git"""






							def __init__(      self			,						lowerCAmelCase__=None			,						lowerCAmelCase__=3_0_5_2_2			,						lowerCAmelCase__=7_6_8			,						lowerCAmelCase__=6			,						lowerCAmelCase__=1_2			,						lowerCAmelCase__=3_0_7_2			,						lowerCAmelCase__="gelu"			,						lowerCAmelCase__=0.1			,						lowerCAmelCase__=0.1			,						lowerCAmelCase__=1_0_2_4			,						lowerCAmelCase__=0.02			,						lowerCAmelCase__=1E-12			,						lowerCAmelCase__=0			,						lowerCAmelCase__="absolute"			,						lowerCAmelCase__=True			,						lowerCAmelCase__=False			,						lowerCAmelCase__=1_0_1			,						lowerCAmelCase__=1_0_2			,						lowerCAmelCase__=None			,						**lowerCAmelCase__			,						):





										"""simple docstring"""




										super().__init__(bos_token_id=lowerCAmelCase__			,						eos_token_id=lowerCAmelCase__			,						pad_token_id=lowerCAmelCase__			,						**lowerCAmelCase__       )

										if vision_config is None:
													SCREAMING_SNAKE_CASE_		:     Dict					=					{}
													logger.info('vision_config is None. initializing the GitVisionConfig with default values.'       )

										SCREAMING_SNAKE_CASE_		:     str					=					GitVisionConfig(**lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     Dict					=					vocab_size
										SCREAMING_SNAKE_CASE_		:     Optional[Any]					=					hidden_size
										SCREAMING_SNAKE_CASE_		:     Tuple					=					num_hidden_layers
										SCREAMING_SNAKE_CASE_		:     Any					=					num_attention_heads
										SCREAMING_SNAKE_CASE_		:     Union[str, Any]					=					hidden_act
										SCREAMING_SNAKE_CASE_		:     Any					=					intermediate_size
										SCREAMING_SNAKE_CASE_		:     Tuple					=					hidden_dropout_prob
										SCREAMING_SNAKE_CASE_		:     str					=					attention_probs_dropout_prob
										SCREAMING_SNAKE_CASE_		:     Union[str, Any]					=					max_position_embeddings
										SCREAMING_SNAKE_CASE_		:     int					=					initializer_range
										SCREAMING_SNAKE_CASE_		:     int					=					layer_norm_eps
										SCREAMING_SNAKE_CASE_		:     List[str]					=					position_embedding_type
										SCREAMING_SNAKE_CASE_		:     List[str]					=					use_cache
										SCREAMING_SNAKE_CASE_		:     Any					=					tie_word_embeddings
										SCREAMING_SNAKE_CASE_		:     Dict					=					num_image_with_embedding

										SCREAMING_SNAKE_CASE_		:     List[str]					=					bos_token_id
										SCREAMING_SNAKE_CASE_		:     str					=					eos_token_id






							def 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




										SCREAMING_SNAKE_CASE_		:     Union[str, Any]					=					copy.deepcopy(self.__dict__       )
										SCREAMING_SNAKE_CASE_		:     str					=					self.vision_config.to_dict()
										SCREAMING_SNAKE_CASE_		:     Optional[int]					=					self.__class__.model_type
										return output





