



import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()






class 						__a (     SCREAMING_SNAKE_CASE	,    SCREAMING_SNAKE_CASE	,    SCREAMING_SNAKE_CASE	,    unittest.TestCase				):
	SCREAMING_SNAKE_CASE       =       StableUnCLIPPipeline
	SCREAMING_SNAKE_CASE       =       TEXT_TO_IMAGE_PARAMS
	SCREAMING_SNAKE_CASE       =       TEXT_TO_IMAGE_BATCH_PARAMS
	SCREAMING_SNAKE_CASE       =       TEXT_TO_IMAGE_IMAGE_PARAMS
	SCREAMING_SNAKE_CASE       =       TEXT_TO_IMAGE_IMAGE_PARAMS

	# TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
	SCREAMING_SNAKE_CASE       =       False



	def 		UpperCamelCase							(						self							:    Optional[int])->							List[str]:
						__lowerCAmelCase	   =32
						__lowerCAmelCase	   =embedder_hidden_size

						# prior components

						torch.manual_seed(0)
						__lowerCAmelCase	   =CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""")

						torch.manual_seed(0)
						__lowerCAmelCase	   =CLIPTextModelWithProjection(
						    CLIPTextConfig(
						        bos_token_id=0				,	eos_token_id=2				,	hidden_size=snake_case_				,	projection_dim=snake_case_				,	intermediate_size=37				,	layer_norm_eps=1e-05				,	num_attention_heads=4				,	num_hidden_layers=5				,	pad_token_id=1				,	vocab_size=10_00				,	))

						torch.manual_seed(0)
						__lowerCAmelCase	   =PriorTransformer(
						    num_attention_heads=2				,	attention_head_dim=12				,	embedding_dim=snake_case_				,	num_layers=1				,	)

						torch.manual_seed(0)
						__lowerCAmelCase	   =DDPMScheduler(
						    variance_type="""fixed_small_log"""				,	prediction_type="""sample"""				,	num_train_timesteps=10_00				,	clip_sample=snake_case_				,	clip_sample_range=5.0				,	beta_schedule="""squaredcos_cap_v2"""				,	)

						# regular denoising components

						torch.manual_seed(0)
						__lowerCAmelCase	   =StableUnCLIPImageNormalizer(embedding_dim=snake_case_)
						__lowerCAmelCase	   =DDPMScheduler(beta_schedule="""squaredcos_cap_v2""")

						torch.manual_seed(0)
						__lowerCAmelCase	   =CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""")

						torch.manual_seed(0)
						__lowerCAmelCase	   =CLIPTextModel(
						    CLIPTextConfig(
						        bos_token_id=0				,	eos_token_id=2				,	hidden_size=snake_case_				,	projection_dim=32				,	intermediate_size=37				,	layer_norm_eps=1e-05				,	num_attention_heads=4				,	num_hidden_layers=5				,	pad_token_id=1				,	vocab_size=10_00				,	))

						torch.manual_seed(0)
						__lowerCAmelCase	   =UNetaDConditionModel(
						    sample_size=32				,	in_channels=4				,	out_channels=4				,	down_block_types=("""CrossAttnDownBlock2D""", """DownBlock2D""")				,	up_block_types=("""UpBlock2D""", """CrossAttnUpBlock2D""")				,	block_out_channels=(32, 64)				,	attention_head_dim=(2, 4)				,	class_embed_type="""projection"""				,	projection_class_embeddings_input_dim=embedder_projection_dim * 2				,	cross_attention_dim=snake_case_				,	layers_per_block=1				,	upcast_attention=snake_case_				,	use_linear_projection=snake_case_				,	)

						torch.manual_seed(0)
						__lowerCAmelCase	   =DDIMScheduler(
						    beta_schedule="""scaled_linear"""				,	beta_start=0.0_0_0_8_5				,	beta_end=0.0_1_2				,	prediction_type="""v_prediction"""				,	set_alpha_to_one=snake_case_				,	steps_offset=1				,	)

						torch.manual_seed(0)
						__lowerCAmelCase	   =AutoencoderKL()

						__lowerCAmelCase	   ={
						    # prior components
						    """prior_tokenizer""": prior_tokenizer,
						    """prior_text_encoder""": prior_text_encoder,
						    """prior""": prior,
						    """prior_scheduler""": prior_scheduler,
						    # image noising components
						    """image_normalizer""": image_normalizer,
						    """image_noising_scheduler""": image_noising_scheduler,
						    # regular denoising components
						    """tokenizer""": tokenizer,
						    """text_encoder""": text_encoder,
						    """unet""": unet,
						    """scheduler""": scheduler,
						    """vae""": vae,
						}

						return components



	def 		UpperCamelCase							(						self							:    Optional[Any]				,	snake_case_							:    Tuple				,	snake_case_							:    Dict=0)->							List[str]:
						if str(snake_case_).startswith("""mps"""):
											__lowerCAmelCase	   =torch.manual_seed(snake_case_)
						else:
											__lowerCAmelCase	   =torch.Generator(device=snake_case_).manual_seed(snake_case_)
						__lowerCAmelCase	   ={
						    """prompt""": """A painting of a squirrel eating a burger""",
						    """generator""": generator,
						    """num_inference_steps""": 2,
						    """prior_num_inference_steps""": 2,
						    """output_type""": """numpy""",
						}
						return inputs



	def 		UpperCamelCase							(						self							:    Any)->							List[str]:
						__lowerCAmelCase	   =torch_device == """cpu"""

						self._test_attention_slicing_forward_pass(test_max_difference=snake_case_)



	def 		UpperCamelCase							(						self							:    Dict)->							Dict:
						__lowerCAmelCase	   =torch_device in ["""cpu""", """mps"""]

						self._test_inference_batch_single_identical(test_max_difference=snake_case_)






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



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



	def 		UpperCamelCase							(						self							:    Dict)->							int:
						__lowerCAmelCase	   =load_numpy(
						    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy""")

						__lowerCAmelCase	   =StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""				,	torch_dtype=torch.floataa)
						pipe.to(snake_case_)
						pipe.set_progress_bar_config(disable=snake_case_)
						# stable unclip will oom when integration tests are run on a V100,
						# so turn on memory savings
						pipe.enable_attention_slicing()
						pipe.enable_sequential_cpu_offload()

						__lowerCAmelCase	   =torch.Generator(device="""cpu""").manual_seed(0)
						__lowerCAmelCase	   =pipe("""anime turle"""				,	generator=snake_case_				,	output_type="""np""")

						__lowerCAmelCase	   =output.images[0]

						assert image.shape == (7_68, 7_68, 3)

						assert_mean_pixel_difference(snake_case_				,	snake_case_)



	def 		UpperCamelCase							(						self							:    Optional[Any])->							str:
						torch.cuda.empty_cache()
						torch.cuda.reset_max_memory_allocated()
						torch.cuda.reset_peak_memory_stats()

						__lowerCAmelCase	   =StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""				,	torch_dtype=torch.floataa)
						__lowerCAmelCase	   =pipe.to(snake_case_)
						pipe.set_progress_bar_config(disable=snake_case_)
						pipe.enable_attention_slicing()
						pipe.enable_sequential_cpu_offload()

						__lowerCAmelCase	   =pipe(
						    """anime turtle"""				,	prior_num_inference_steps=2				,	num_inference_steps=2				,	output_type="""np"""				,	)

						__lowerCAmelCase	   =torch.cuda.max_memory_allocated()
						# make sure that less than 7 GB is allocated
						assert mem_bytes < 7 * 10**9









from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


lowercase_	       =			{
    '''configuration_blip_2''': [
        '''BLIP_2_PRETRAINED_CONFIG_ARCHIVE_MAP''',
        '''Blip2Config''',
        '''Blip2QFormerConfig''',
        '''Blip2VisionConfig''',
    ],
    '''processing_blip_2''': ['''Blip2Processor'''],
}

try:
				if not is_torch_available():
								raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
				pass
else:
				lowercase_	       =			[
				    '''BLIP_2_PRETRAINED_MODEL_ARCHIVE_LIST''',
				    '''Blip2Model''',
				    '''Blip2QFormerModel''',
				    '''Blip2PreTrainedModel''',
				    '''Blip2ForConditionalGeneration''',
				    '''Blip2VisionModel''',
				]

if TYPE_CHECKING:
				from .configuration_blip_a import (
				    BLIP_2_PRETRAINED_CONFIG_ARCHIVE_MAP,
				    BlipaConfig,
				    BlipaQFormerConfig,
				    BlipaVisionConfig,
				)
				from .processing_blip_a import BlipaProcessor

				try:
								if not is_torch_available():
												raise OptionalDependencyNotAvailable()
				except OptionalDependencyNotAvailable:
								pass
				else:
								from .modeling_blip_a import (
								    BLIP_2_PRETRAINED_MODEL_ARCHIVE_LIST,
								    BlipaForConditionalGeneration,
								    BlipaModel,
								    BlipaPreTrainedModel,
								    BlipaQFormerModel,
								    BlipaVisionModel,
								)

else:
				import sys

				lowercase_	       =			_LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)




