






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     _SCREAMING_SNAKE_CASE       (			_a		,   _a		,   unittest.TestCase      ):
	snake_case__							:     Optional[int]     	=  CycleDiffusionPipeline
	snake_case__							:     Union[str, Any]     	=  TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
	    """negative_prompt""",
	    """height""",
	    """width""",
	    """negative_prompt_embeds""",
	}
	snake_case__							:     Any     	=  PipelineTesterMixin.required_optional_params - {"""latents"""}
	snake_case__							:     Tuple     	=  TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"""source_prompt"""}      )
	snake_case__							:     Tuple     	=  IMAGE_TO_IMAGE_IMAGE_PARAMS
	snake_case__							:     List[Any]     	=  IMAGE_TO_IMAGE_IMAGE_PARAMS





	def 						_A (		self				:					str	):
		torch.manual_seed(0	)
		UpperCamelCase     :Optional[int]          =	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 ,  )
		UpperCamelCase     :Union[str, Any]          =	DDIMScheduler(
		    beta_start=0.00085 ,  beta_end=0.012 ,  beta_schedule="""scaled_linear""" ,  num_train_timesteps=1_000 ,  clip_sample=__lowerCamelCase ,  set_alpha_to_one=__lowerCamelCase ,  )
		torch.manual_seed(0	)
		UpperCamelCase     :Dict          =	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	)
		UpperCamelCase     :List[Any]          =	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=1_000 ,  )
		UpperCamelCase     :List[Any]          =	CLIPTextModel(__lowerCamelCase	)
		UpperCamelCase     :Optional[int]          =	CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""	)

		UpperCamelCase     :List[str]          =	{
		    """unet""": unet,
		    """scheduler""": scheduler,
		    """vae""": vae,
		    """text_encoder""": text_encoder,
		    """tokenizer""": tokenizer,
		    """safety_checker""": None,
		    """feature_extractor""": None,
		}
		return components





	def 						_A (		self				:					List[Any] ,  __lowerCamelCase				:					int ,  __lowerCamelCase				:					Union[str, Any]=0	):
		UpperCamelCase     :Union[str, Any]          =	floats_tensor((1, 3, 32, 32) ,  rng=random.Random(__lowerCamelCase	)	).to(__lowerCamelCase	)
		UpperCamelCase     :List[str]          =	image / 2 + 0.5
		if str(__lowerCamelCase	).startswith("""mps"""	):
			UpperCamelCase     :Union[str, Any]          =	torch.manual_seed(__lowerCamelCase	)
		else:
			UpperCamelCase     :List[Any]          =	torch.Generator(device=__lowerCamelCase	).manual_seed(__lowerCamelCase	)
		UpperCamelCase     :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 						_A (		self				:					Optional[int]	):
		UpperCamelCase     :Dict          =	"""cpu"""  # ensure determinism for the device-dependent torch.Generator

		UpperCamelCase     :str          =	self.get_dummy_components()
		UpperCamelCase     :int          =	CycleDiffusionPipeline(**__lowerCamelCase	)
		UpperCamelCase     :Dict          =	pipe.to(__lowerCamelCase	)
		pipe.set_progress_bar_config(disable=__lowerCamelCase	)

		UpperCamelCase     :Tuple          =	self.get_dummy_inputs(__lowerCamelCase	)
		UpperCamelCase     :Any          =	pipe(**__lowerCamelCase	)
		UpperCamelCase     :Dict          =	output.images

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

		assert images.shape == (1, 32, 32, 3)
		UpperCamelCase     :int          =	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 						_A (		self				:					Optional[int]	):
		UpperCamelCase     :Dict          =	self.get_dummy_components()
		for name, module in components.items():
			if hasattr(__lowerCamelCase ,  """half"""	):
				UpperCamelCase     :Optional[Any]          =	module.half()
		UpperCamelCase     :int          =	CycleDiffusionPipeline(**__lowerCamelCase	)
		UpperCamelCase     :Optional[Any]          =	pipe.to(__lowerCamelCase	)
		pipe.set_progress_bar_config(disable=__lowerCamelCase	)

		UpperCamelCase     :Any          =	self.get_dummy_inputs(__lowerCamelCase	)
		UpperCamelCase     :Optional[int]          =	pipe(**__lowerCamelCase	)
		UpperCamelCase     :List[str]          =	output.images

		UpperCamelCase     :Optional[Any]          =	images[0, -3:, -3:, -1]

		assert images.shape == (1, 32, 32, 3)
		UpperCamelCase     :List[str]          =	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 						_A (		self				:					List[str]	):
		return super().test_save_load_local()





	@unittest.skip("""non-deterministic pipeline"""	)
	def 						_A (		self				:					Optional[int]	):
		return super().test_inference_batch_single_identical()





	@skip_mps
	def 						_A (		self				:					Optional[int]	):
		return super().test_dict_tuple_outputs_equivalent()





	@skip_mps
	def 						_A (		self				:					List[str]	):
		return super().test_save_load_optional_components()





	@skip_mps
	def 						_A (		self				:					Tuple	):
		return super().test_attention_slicing_forward_pass()

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





	def 						_A (		self				:					Optional[int]	):
		# clean up the VRAM after each test
		super().tearDown()
		gc.collect()
		torch.cuda.empty_cache()





	def 						_A (		self				:					Tuple	):
		UpperCamelCase     :int          =	load_image(
		    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
		    """/cycle-diffusion/black_colored_car.png"""	)
		UpperCamelCase     :List[Any]          =	load_numpy(
		    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy"""	)
		UpperCamelCase     :Any          =	init_image.resize((512, 512)	)

		UpperCamelCase     :Dict          =	"""CompVis/stable-diffusion-v1-4"""
		UpperCamelCase     :List[Any]          =	DDIMScheduler.from_pretrained(__lowerCamelCase ,  subfolder="""scheduler"""	)
		UpperCamelCase     :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()

		UpperCamelCase     :List[Any]          =	"""A black colored car"""
		UpperCamelCase     :Tuple          =	"""A blue colored car"""

		UpperCamelCase     :Tuple          =	torch.manual_seed(0	)
		UpperCamelCase     :List[Any]          =	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""" ,  )
		UpperCamelCase     :Optional[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 						_A (		self				:					Optional[Any]	):
		UpperCamelCase     :Dict          =	load_image(
		    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
		    """/cycle-diffusion/black_colored_car.png"""	)
		UpperCamelCase     :Any          =	load_numpy(
		    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy"""	)
		UpperCamelCase     :Optional[Any]          =	init_image.resize((512, 512)	)

		UpperCamelCase     :str          =	"""CompVis/stable-diffusion-v1-4"""
		UpperCamelCase     :Union[str, Any]          =	DDIMScheduler.from_pretrained(__lowerCamelCase ,  subfolder="""scheduler"""	)
		UpperCamelCase     :str          =	CycleDiffusionPipeline.from_pretrained(__lowerCamelCase ,  scheduler=__lowerCamelCase ,  safety_checker=__lowerCamelCase	)

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

		UpperCamelCase     :Dict          =	"""A black colored car"""
		UpperCamelCase     :str          =	"""A blue colored car"""

		UpperCamelCase     :List[Any]          =	torch.manual_seed(0	)
		UpperCamelCase     :Optional[int]          =	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""" ,  )
		UpperCamelCase     :Optional[int]          =	output.images

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




'''simple docstring'''


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


UpperCAmelCase_		:   Dict							   =		logging.get_logger(__name__)

UpperCAmelCase_		:   Optional[int]							   =		{
    'google/canine-s': 'https://huggingface.co/google/canine-s/resolve/main/config.json',
    # See all CANINE models at https://huggingface.co/models?filter=canine
}





class     lowercase__  (						_snake_case					):





							'''simple docstring'''


							A_							:      Optional[int]      	=  """canine"""






							def __init__(    self    ,					__snake_case=768    ,					__snake_case=12    ,					__snake_case=12    ,					__snake_case=3072    ,					__snake_case="gelu"    ,					__snake_case=0.1    ,					__snake_case=0.1    ,					__snake_case=1_6384    ,					__snake_case=16    ,					__snake_case=0.02    ,					__snake_case=1e-12    ,					__snake_case=0    ,					__snake_case=0XE000    ,					__snake_case=0XE001    ,					__snake_case=4    ,					__snake_case=4    ,					__snake_case=8    ,					__snake_case=1_6384    ,					__snake_case=128    ,					**__snake_case    ,					):
											super().__init__(pad_token_id=__snake_case    ,					bos_token_id=__snake_case    ,					eos_token_id=__snake_case    ,					**__snake_case  )

											_SCREAMING_SNAKE_CASE :       str     		=      max_position_embeddings
											_SCREAMING_SNAKE_CASE :       Any     		=      hidden_size
											_SCREAMING_SNAKE_CASE :       Dict     		=      num_hidden_layers
											_SCREAMING_SNAKE_CASE :       Tuple     		=      num_attention_heads
											_SCREAMING_SNAKE_CASE :       List[str]     		=      intermediate_size
											_SCREAMING_SNAKE_CASE :       Tuple     		=      hidden_act
											_SCREAMING_SNAKE_CASE :       Union[str, Any]     		=      hidden_dropout_prob
											_SCREAMING_SNAKE_CASE :       str     		=      attention_probs_dropout_prob
											_SCREAMING_SNAKE_CASE :       Dict     		=      initializer_range
											_SCREAMING_SNAKE_CASE :       Optional[int]     		=      type_vocab_size
											_SCREAMING_SNAKE_CASE :       Optional[Any]     		=      layer_norm_eps

											# Character config:
											_SCREAMING_SNAKE_CASE :       Any     		=      downsampling_rate
											_SCREAMING_SNAKE_CASE :       List[str]     		=      upsampling_kernel_size
											_SCREAMING_SNAKE_CASE :       Union[str, Any]     		=      num_hash_functions
											_SCREAMING_SNAKE_CASE :       Optional[Any]     		=      num_hash_buckets
											_SCREAMING_SNAKE_CASE :       Union[str, Any]     		=      local_transformer_stride




