



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 		lowercase						(			snake_case__			,   snake_case__			,   snake_case__			,   unittest.TestCase):






	"""simple docstring"""


	a__				: Any		       =							StableUnCLIPPipeline
	a__				: str		       =							TEXT_TO_IMAGE_PARAMS
	a__				: Dict		       =							TEXT_TO_IMAGE_BATCH_PARAMS
	a__				: str		       =							TEXT_TO_IMAGE_IMAGE_PARAMS
	a__				: List[Any]		       =							TEXT_TO_IMAGE_IMAGE_PARAMS

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




	def   _SCREAMING_SNAKE_CASE		(     self     :						Union[str, Any]					)      ->			Optional[int]:
							UpperCAmelCase_=       32
							UpperCAmelCase_=       embedder_hidden_size

							# prior components

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

							torch.manual_seed(0					)
							UpperCAmelCase_=       CLIPTextModelWithProjection(
							    CLIPTextConfig(
							        bos_token_id=0						,  eos_token_id=2						,  hidden_size=__UpperCAmelCase						,  projection_dim=__UpperCAmelCase						,  intermediate_size=37						,  layer_norm_eps=1E-05						,  num_attention_heads=4						,  num_hidden_layers=5						,  pad_token_id=1						,  vocab_size=1_000						,  )					)

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

							torch.manual_seed(0					)
							UpperCAmelCase_=       DDPMScheduler(
							    variance_type="""fixed_small_log"""						,  prediction_type="""sample"""						,  num_train_timesteps=1_000						,  clip_sample=__UpperCAmelCase						,  clip_sample_range=5.0						,  beta_schedule="""squaredcos_cap_v2"""						,  )

							# regular denoising components

							torch.manual_seed(0					)
							UpperCAmelCase_=       StableUnCLIPImageNormalizer(embedding_dim=__UpperCAmelCase					)
							UpperCAmelCase_=       DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""					)

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

							torch.manual_seed(0					)
							UpperCAmelCase_=       CLIPTextModel(
							    CLIPTextConfig(
							        bos_token_id=0						,  eos_token_id=2						,  hidden_size=__UpperCAmelCase						,  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=1_000						,  )					)

							torch.manual_seed(0					)
							UpperCAmelCase_=       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=__UpperCAmelCase						,  layers_per_block=1						,  upcast_attention=__UpperCAmelCase						,  use_linear_projection=__UpperCAmelCase						,  )

							torch.manual_seed(0					)
							UpperCAmelCase_=       DDIMScheduler(
							    beta_schedule="""scaled_linear"""						,  beta_start=0.00_085						,  beta_end=0.012						,  prediction_type="""v_prediction"""						,  set_alpha_to_one=__UpperCAmelCase						,  steps_offset=1						,  )

							torch.manual_seed(0					)
							UpperCAmelCase_=       AutoencoderKL()

							UpperCAmelCase_=       {
							    # 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   _SCREAMING_SNAKE_CASE		(     self     :						int						,  __UpperCAmelCase     :						Optional[int]						,  __UpperCAmelCase     :						List[Any]=0					)      ->			Any:
							if str(__UpperCAmelCase					).startswith("""mps"""					):
													UpperCAmelCase_=       torch.manual_seed(__UpperCAmelCase					)
							else:
													UpperCAmelCase_=       torch.Generator(device=__UpperCAmelCase					).manual_seed(__UpperCAmelCase					)
							UpperCAmelCase_=       {
							    """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   _SCREAMING_SNAKE_CASE		(     self     :						Any					)      ->			int:
							UpperCAmelCase_=       torch_device == """cpu"""

							self._test_attention_slicing_forward_pass(test_max_difference=__UpperCAmelCase					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						Optional[int]					)      ->			List[Any]:
							UpperCAmelCase_=       torch_device in ["""cpu""", """mps"""]

							self._test_inference_batch_single_identical(test_max_difference=__UpperCAmelCase					)






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






	"""simple docstring"""


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




	def   _SCREAMING_SNAKE_CASE		(     self     :						str					)      ->			Optional[int]:
							UpperCAmelCase_=       load_numpy(
							    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy"""					)

							UpperCAmelCase_=       StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""						,  torch_dtype=torch.floataa					)
							pipe.to(__UpperCAmelCase					)
							pipe.set_progress_bar_config(disable=__UpperCAmelCase					)
							# 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()

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

							UpperCAmelCase_=       output.images[0]

							assert image.shape == (768, 768, 3)

							assert_mean_pixel_difference(__UpperCAmelCase						,  __UpperCAmelCase					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						int					)      ->			str:
							torch.cuda.empty_cache()
							torch.cuda.reset_max_memory_allocated()
							torch.cuda.reset_peak_memory_stats()

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

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

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











import pytest

import datasets


# Import fixture modules as plugins
__A       =						['''tests.fixtures.files''', '''tests.fixtures.hub''', '''tests.fixtures.fsspec''']



def 	__a					(   lowerCAmelCase_       : Optional[Any]      ,lowerCAmelCase_       : Any    )				->			Tuple:






						'''simple docstring'''



						for item in items:
												if any(marker in item.keywords for marker in ["""integration""", """unit"""]    ):
																		continue
												item.add_marker(pytest.mark.unit    )



def 	__a					(   lowerCAmelCase_       : Tuple    )				->			Optional[Any]:






						'''simple docstring'''



						config.addinivalue_line("""markers"""      ,"""torchaudio_latest: mark test to run with torchaudio>=0.12"""    )



@pytest.fixture(autouse=lowerCAmelCase_    )
def 	__a					(   lowerCAmelCase_       : List[str]      ,lowerCAmelCase_       : List[str]    )				->			int:






						'''simple docstring'''



						UpperCAmelCase_=       tmp_path_factory.getbasetemp() / """cache"""
						UpperCAmelCase_=       test_hf_cache_home / """datasets"""
						UpperCAmelCase_=       test_hf_cache_home / """metrics"""
						UpperCAmelCase_=       test_hf_cache_home / """modules"""
						monkeypatch.setattr("""datasets.config.HF_DATASETS_CACHE"""      ,str(lowerCAmelCase_    )    )
						monkeypatch.setattr("""datasets.config.HF_METRICS_CACHE"""      ,str(lowerCAmelCase_    )    )
						monkeypatch.setattr("""datasets.config.HF_MODULES_CACHE"""      ,str(lowerCAmelCase_    )    )
						UpperCAmelCase_=       test_hf_datasets_cache / """downloads"""
						monkeypatch.setattr("""datasets.config.DOWNLOADED_DATASETS_PATH"""      ,str(lowerCAmelCase_    )    )
						UpperCAmelCase_=       test_hf_datasets_cache / """downloads""" / """extracted"""
						monkeypatch.setattr("""datasets.config.EXTRACTED_DATASETS_PATH"""      ,str(lowerCAmelCase_    )    )



@pytest.fixture(autouse=lowerCAmelCase_      ,scope="""session"""    )
def 	__a					(   )				->			Optional[int]:






						'''simple docstring'''



						datasets.disable_progress_bar()



@pytest.fixture(autouse=lowerCAmelCase_    )
def 	__a					(   lowerCAmelCase_       : int    )				->			str:






						'''simple docstring'''



						monkeypatch.setattr("""datasets.config.HF_UPDATE_DOWNLOAD_COUNTS"""      ,lowerCAmelCase_    )



@pytest.fixture
def 	__a					(   lowerCAmelCase_       : List[str]    )				->			Union[str, Any]:






						'''simple docstring'''



						monkeypatch.setattr("""sqlalchemy.util.deprecations.SILENCE_UBER_WARNING"""      ,lowerCAmelCase_    )




