





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					):
					lowerCAmelCase_							=	CycleDiffusionPipeline
					lowerCAmelCase_							=	TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
					    '''negative_prompt''',
					    '''height''',
					    '''width''',
					    '''negative_prompt_embeds''',
					}
					lowerCAmelCase_							=	PipelineTesterMixin.required_optional_params - {'''latents'''}
					lowerCAmelCase_							=	TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'''source_prompt'''}					)
					lowerCAmelCase_							=	IMAGE_TO_IMAGE_IMAGE_PARAMS
					lowerCAmelCase_							=	IMAGE_TO_IMAGE_IMAGE_PARAMS




					def 	lowerCAmelCase			(		self  )		->							Tuple:
										torch.manual_seed(0  )
										_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				,						)
										_snake_case									=				DDIMScheduler(
										    beta_start=0.0_00_85				,						beta_end=0.0_12				,						beta_schedule='scaled_linear'				,						num_train_timesteps=1000				,						clip_sample=lowerCAmelCase_				,						set_alpha_to_one=lowerCAmelCase_				,						)
										torch.manual_seed(0  )
										_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  )
										_snake_case									=				CLIPTextConfig(
										    bos_token_id=0				,						eos_token_id=2				,						hidden_size=32				,						intermediate_size=37				,						layer_norm_eps=1E-05				,						num_attention_heads=4				,						num_hidden_layers=5				,						pad_token_id=1				,						vocab_size=1000				,						)
										_snake_case									=				CLIPTextModel(lowerCAmelCase_  )
										_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				,						lowerCAmelCase_				,						lowerCAmelCase_=0  )		->							str:
										_snake_case									=				floats_tensor((1, 3, 32, 32)				,						rng=random.Random(lowerCAmelCase_  )  ).to(lowerCAmelCase_  )
										_snake_case									=				image / 2 + 0.5
										if str(lowerCAmelCase_  ).startswith('mps'  ):
															_snake_case									=				torch.manual_seed(lowerCAmelCase_  )
										else:
															_snake_case									=				torch.Generator(device=lowerCAmelCase_  ).manual_seed(lowerCAmelCase_  )
										_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  )		->							Dict:
										_snake_case									=				'cpu'  # ensure determinism for the device-dependent torch.Generator

										_snake_case									=				self.get_dummy_components()
										_snake_case									=				CycleDiffusionPipeline(**lowerCAmelCase_  )
										_snake_case									=				pipe.to(lowerCAmelCase_  )
										pipe.set_progress_bar_config(disable=lowerCAmelCase_  )

										_snake_case									=				self.get_dummy_inputs(lowerCAmelCase_  )
										_snake_case									=				pipe(**lowerCAmelCase_  )
										_snake_case									=				output.images

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

										assert images.shape == (1, 32, 32, 3)
										_snake_case									=				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 	lowerCAmelCase			(		self  )		->							Union[str, Any]:
										_snake_case									=				self.get_dummy_components()
										for name, module in components.items():
															if hasattr(lowerCAmelCase_				,						'half'  ):
																				_snake_case									=				module.half()
										_snake_case									=				CycleDiffusionPipeline(**lowerCAmelCase_  )
										_snake_case									=				pipe.to(lowerCAmelCase_  )
										pipe.set_progress_bar_config(disable=lowerCAmelCase_  )

										_snake_case									=				self.get_dummy_inputs(lowerCAmelCase_  )
										_snake_case									=				pipe(**lowerCAmelCase_  )
										_snake_case									=				output.images

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

										assert images.shape == (1, 32, 32, 3)
										_snake_case									=				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 	lowerCAmelCase			(		self  )		->							Any:
										return super().test_save_load_local()




					@unittest.skip('non-deterministic pipeline'  )
					def 	lowerCAmelCase			(		self  )		->							Optional[Any]:
										return super().test_inference_batch_single_identical()




					@skip_mps
					def 	lowerCAmelCase			(		self  )		->							int:
										return super().test_dict_tuple_outputs_equivalent()




					@skip_mps
					def 	lowerCAmelCase			(		self  )		->							int:
										return super().test_save_load_optional_components()






					@skip_mps
					def 	lowerCAmelCase			(		self  )		->							Optional[int]:
										return super().test_attention_slicing_forward_pass()





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




					def 	lowerCAmelCase			(		self  )		->							Tuple:
										# clean up the VRAM after each test
										super().tearDown()
										gc.collect()
										torch.cuda.empty_cache()




					def 	lowerCAmelCase			(		self  )		->							int:
										_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((512, 512)  )

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

										_snake_case									=				'A black colored car'
										_snake_case									=				'A blue colored car'

										_snake_case									=				torch.manual_seed(0  )
										_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'				,						)
										_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  )		->							Tuple:
										_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((512, 512)  )

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

										_snake_case									=				'A black colored car'
										_snake_case									=				'A blue colored car'

										_snake_case									=				torch.manual_seed(0  )
										_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'				,						)
										_snake_case									=				output.images

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










import copy
import tempfile
import unittest

from huggingface_hub import HfFolder, delete_repo
from parameterized import parameterized
from requests.exceptions import HTTPError

from transformers import AutoConfig, GenerationConfig
from transformers.testing_utils import TOKEN, USER, is_staging_test
class 			UpperCamelCase_		(  unittest.TestCase					):




					@parameterized.expand([(None,), ('foo.json',)]  )
					def 	lowerCAmelCase			(		self				,						lowerCAmelCase_  )		->							str:
										_snake_case									=				GenerationConfig(
										    do_sample=lowerCAmelCase_				,						temperature=0.7				,						length_penalty=1.0				,						bad_words_ids=[[1, 2, 3], [4, 5]]				,						)
										with tempfile.TemporaryDirectory() as tmp_dir:
															config.save_pretrained(lowerCAmelCase_				,						config_name=lowerCAmelCase_  )
															_snake_case									=				GenerationConfig.from_pretrained(lowerCAmelCase_				,						config_name=lowerCAmelCase_  )

										# Checks parameters that were specified
										self.assertEqual(loaded_config.do_sample				,						lowerCAmelCase_  )
										self.assertEqual(loaded_config.temperature				,						0.7  )
										self.assertEqual(loaded_config.length_penalty				,						1.0  )
										self.assertEqual(loaded_config.bad_words_ids				,						[[1, 2, 3], [4, 5]]  )

										# Checks parameters that were not specified (defaults)
										self.assertEqual(loaded_config.top_k				,						50  )
										self.assertEqual(loaded_config.max_length				,						20  )
										self.assertEqual(loaded_config.max_time				,						lowerCAmelCase_  )




					def 	lowerCAmelCase			(		self  )		->							Optional[int]:
										_snake_case									=				AutoConfig.from_pretrained('gpt2'  )
										_snake_case									=				GenerationConfig.from_model_config(lowerCAmelCase_  )
										_snake_case									=				GenerationConfig()

										# The generation config has loaded a few non-default parameters from the model config
										self.assertNotEqual(lowerCAmelCase_				,						lowerCAmelCase_  )

										# One of those parameters is eos_token_id -- check if it matches
										self.assertNotEqual(generation_config_from_model.eos_token_id				,						default_generation_config.eos_token_id  )
										self.assertEqual(generation_config_from_model.eos_token_id				,						model_config.eos_token_id  )




					def 	lowerCAmelCase			(		self  )		->							Tuple:
										_snake_case									=				GenerationConfig()
										_snake_case									=				{
										    'max_new_tokens': 1024,
										    'foo': 'bar',
										}
										_snake_case									=				copy.deepcopy(lowerCAmelCase_  )
										_snake_case									=				generation_config.update(**lowerCAmelCase_  )

										# update_kwargs was not modified (no side effects)
										self.assertEqual(lowerCAmelCase_				,						lowerCAmelCase_  )

										# update_kwargs was used to update the config on valid attributes
										self.assertEqual(generation_config.max_new_tokens				,						1024  )

										# `.update()` returns a dictionary of unused kwargs
										self.assertEqual(lowerCAmelCase_				,						{'foo': 'bar'}  )




					def 	lowerCAmelCase			(		self  )		->							Optional[int]:
										_snake_case									=				GenerationConfig()
										_snake_case									=				'bar'

										with tempfile.TemporaryDirectory('test-generation-config'  ) as tmp_dir:
															generation_config.save_pretrained(lowerCAmelCase_  )

															_snake_case									=				GenerationConfig.from_pretrained(lowerCAmelCase_  )
										# update_kwargs was used to update the config on valid attributes
										self.assertEqual(new_config.foo				,						'bar'  )

										_snake_case									=				GenerationConfig.from_model_config(lowerCAmelCase_  )
										assert not hasattr(lowerCAmelCase_				,						'foo'  )  # no new kwargs should be initialized if from config






					def 	lowerCAmelCase			(		self  )		->							List[Any]:
										_snake_case									=				GenerationConfig()
										self.assertEqual(default_config.temperature				,						1.0  )
										self.assertEqual(default_config.do_sample				,						lowerCAmelCase_  )
										self.assertEqual(default_config.num_beams				,						1  )

										_snake_case									=				GenerationConfig(
										    do_sample=lowerCAmelCase_				,						temperature=0.7				,						length_penalty=1.0				,						bad_words_ids=[[1, 2, 3], [4, 5]]				,						)
										self.assertEqual(config.temperature				,						0.7  )
										self.assertEqual(config.do_sample				,						lowerCAmelCase_  )
										self.assertEqual(config.num_beams				,						1  )

										with tempfile.TemporaryDirectory() as tmp_dir:
															config.save_pretrained(lowerCAmelCase_  )
															_snake_case									=				GenerationConfig.from_pretrained(lowerCAmelCase_				,						temperature=1.0  )

										self.assertEqual(loaded_config.temperature				,						1.0  )
										self.assertEqual(loaded_config.do_sample				,						lowerCAmelCase_  )
										self.assertEqual(loaded_config.num_beams				,						1  )  # default value





@is_staging_test
class 			UpperCamelCase_		(  unittest.TestCase					):




					@classmethod
					def 	lowerCAmelCase			(		cls  )		->							List[str]:
										_snake_case									=				TOKEN
										HfFolder.save_token(lowerCAmelCase_  )




					@classmethod
					def 	lowerCAmelCase			(		cls  )		->							Union[str, Any]:
										try:
															delete_repo(token=cls._token				,						repo_id='test-generation-config'  )
										except HTTPError:
															pass

										try:
															delete_repo(token=cls._token				,						repo_id='valid_org/test-generation-config-org'  )
										except HTTPError:
															pass




					def 	lowerCAmelCase			(		self  )		->							int:
										_snake_case									=				GenerationConfig(
										    do_sample=lowerCAmelCase_				,						temperature=0.7				,						length_penalty=1.0				,						)
										config.push_to_hub('test-generation-config'				,						use_auth_token=self._token  )

										_snake_case									=				GenerationConfig.from_pretrained(F'''{USER}/test-generation-config'''  )
										for k, v in config.to_dict().items():
															if k != "transformers_version":
																				self.assertEqual(lowerCAmelCase_				,						getattr(lowerCAmelCase_				,						lowerCAmelCase_  )  )

        # Reset repo
										delete_repo(token=self._token				,						repo_id='test-generation-config'  )

										# Push to hub via save_pretrained
										with tempfile.TemporaryDirectory() as tmp_dir:
															config.save_pretrained(
															    lowerCAmelCase_				,						repo_id='test-generation-config'				,						push_to_hub=lowerCAmelCase_				,						use_auth_token=self._token  )

										_snake_case									=				GenerationConfig.from_pretrained(F'''{USER}/test-generation-config'''  )
										for k, v in config.to_dict().items():
															if k != "transformers_version":
																				self.assertEqual(lowerCAmelCase_				,						getattr(lowerCAmelCase_				,						lowerCAmelCase_  )  )






					def 	lowerCAmelCase			(		self  )		->							List[str]:
										_snake_case									=				GenerationConfig(
										    do_sample=lowerCAmelCase_				,						temperature=0.7				,						length_penalty=1.0				,						)
										config.push_to_hub('valid_org/test-generation-config-org'				,						use_auth_token=self._token  )

										_snake_case									=				GenerationConfig.from_pretrained('valid_org/test-generation-config-org'  )
										for k, v in config.to_dict().items():
															if k != "transformers_version":
																				self.assertEqual(lowerCAmelCase_				,						getattr(lowerCAmelCase_				,						lowerCAmelCase_  )  )

        # Reset repo
										delete_repo(token=self._token				,						repo_id='valid_org/test-generation-config-org'  )

										# Push to hub via save_pretrained
										with tempfile.TemporaryDirectory() as tmp_dir:
															config.save_pretrained(
															    lowerCAmelCase_				,						repo_id='valid_org/test-generation-config-org'				,						push_to_hub=lowerCAmelCase_				,						use_auth_token=self._token  )

										_snake_case									=				GenerationConfig.from_pretrained('valid_org/test-generation-config-org'  )
										for k, v in config.to_dict().items():
															if k != "transformers_version":
																				self.assertEqual(lowerCAmelCase_				,						getattr(lowerCAmelCase_				,						lowerCAmelCase_  )  )



