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__						(					A__    ,		A__    ,		A__    ,		unittest.TestCase					):
					A	       =							StableUnCLIPPipeline
					A	       =							TEXT_TO_IMAGE_PARAMS
					A	       =							TEXT_TO_IMAGE_BATCH_PARAMS
					A	       =							TEXT_TO_IMAGE_IMAGE_PARAMS
					A	       =							TEXT_TO_IMAGE_IMAGE_PARAMS

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

					def 				__UpperCamelCase   (    self      :       Optional[int]    ):







							"""simple docstring"""






							SCREAMING_SNAKE_CASE_  :		Union[str, Any]				      =    32
							SCREAMING_SNAKE_CASE_  :		str				      =    embedder_hidden_size

							# prior components

							torch.manual_seed(0    )
							SCREAMING_SNAKE_CASE_  :		List[Any]				      =    CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"    )

							torch.manual_seed(0    )
							SCREAMING_SNAKE_CASE_  :		Dict				      =    CLIPTextModelWithProjection(
							    CLIPTextConfig(
							        bos_token_id=0,eos_token_id=2,hidden_size=_A,projection_dim=_A,intermediate_size=37,layer_norm_eps=1E-05,num_attention_heads=4,num_hidden_layers=5,pad_token_id=1,vocab_size=1000,)    )

							torch.manual_seed(0    )
							SCREAMING_SNAKE_CASE_  :		List[Any]				      =    PriorTransformer(
							    num_attention_heads=2,attention_head_dim=12,embedding_dim=_A,num_layers=1,)

							torch.manual_seed(0    )
							SCREAMING_SNAKE_CASE_  :		Optional[int]				      =    DDPMScheduler(
							    variance_type="fixed_small_log",prediction_type="sample",num_train_timesteps=1000,clip_sample=_A,clip_sample_range=5.0,beta_schedule="squaredcos_cap_v2",)

							# regular denoising components

							torch.manual_seed(0    )
							SCREAMING_SNAKE_CASE_  :		List[Any]				      =    StableUnCLIPImageNormalizer(embedding_dim=_A    )
							SCREAMING_SNAKE_CASE_  :		str				      =    DDPMScheduler(beta_schedule="squaredcos_cap_v2"    )

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

							torch.manual_seed(0    )
							SCREAMING_SNAKE_CASE_  :		Union[str, Any]				      =    CLIPTextModel(
							    CLIPTextConfig(
							        bos_token_id=0,eos_token_id=2,hidden_size=_A,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=1000,)    )

							torch.manual_seed(0    )
							SCREAMING_SNAKE_CASE_  :		Dict				      =    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=_A,layers_per_block=1,upcast_attention=_A,use_linear_projection=_A,)

							torch.manual_seed(0    )
							SCREAMING_SNAKE_CASE_  :		Any				      =    DDIMScheduler(
							    beta_schedule="scaled_linear",beta_start=0.00085,beta_end=0.012,prediction_type="v_prediction",set_alpha_to_one=_A,steps_offset=1,)

							torch.manual_seed(0    )
							SCREAMING_SNAKE_CASE_  :		Optional[int]				      =    AutoencoderKL()

							SCREAMING_SNAKE_CASE_  :		List[str]				      =    {
							    # 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[int],_A      :       int,_A      :       Optional[Any]=0    ):







							"""simple docstring"""






							if str(_A    ).startswith("mps"    ):
									SCREAMING_SNAKE_CASE_  :		int				      =    torch.manual_seed(_A    )
							else:
									SCREAMING_SNAKE_CASE_  :		Any				      =    torch.Generator(device=_A    ).manual_seed(_A    )
							SCREAMING_SNAKE_CASE_  :		Any				      =    {
							    "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      :       Optional[Any]    ):







							"""simple docstring"""






							SCREAMING_SNAKE_CASE_  :		Tuple				      =    torch_device == "cpu"

							self._test_attention_slicing_forward_pass(test_max_difference=_A    )






					def 				__UpperCamelCase   (    self      :       str    ):







							"""simple docstring"""






							SCREAMING_SNAKE_CASE_  :		Union[str, Any]				      =    torch_device in ["cpu", "mps"]

							self._test_inference_batch_single_identical(test_max_difference=_A    )







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

					def 				__UpperCamelCase   (    self      :       Optional[Any]    ):







							"""simple docstring"""






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

					def 				__UpperCamelCase   (    self      :       Any    ):







							"""simple docstring"""






							SCREAMING_SNAKE_CASE_  :		List[str]				      =    load_numpy(
							    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy"    )

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

							SCREAMING_SNAKE_CASE_  :		int				      =    torch.Generator(device="cpu"    ).manual_seed(0    )
							SCREAMING_SNAKE_CASE_  :		Optional[Any]				      =    pipe("anime turle",generator=_A,output_type="np"    )

							SCREAMING_SNAKE_CASE_  :		Optional[Any]				      =    output.images[0]

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

							assert_mean_pixel_difference(_A,_A    )






					def 				__UpperCamelCase   (    self      :       List[str]    ):







							"""simple docstring"""






							torch.cuda.empty_cache()
							torch.cuda.reset_max_memory_allocated()
							torch.cuda.reset_peak_memory_stats()

							SCREAMING_SNAKE_CASE_  :		Union[str, Any]				      =    StableUnCLIPPipeline.from_pretrained("fusing/stable-unclip-2-1-l",torch_dtype=torch.floataa    )
							SCREAMING_SNAKE_CASE_  :		int				      =    pipe.to(_A    )
							pipe.set_progress_bar_config(disable=_A    )
							pipe.enable_attention_slicing()
							pipe.enable_sequential_cpu_offload()

							SCREAMING_SNAKE_CASE_  :		List[Any]				      =    pipe(
							    "anime turtle",prior_num_inference_steps=2,num_inference_steps=2,output_type="np",)

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

import pytest

from datasets import Dataset, DatasetDict, Features, NamedSplit, Value
from datasets.io.text import TextDatasetReader

from ..utils import assert_arrow_memory_doesnt_increase, assert_arrow_memory_increases




def       _snake_case	(   lowerCAmelCase							:   str   , lowerCAmelCase							:   Union[str, Any]	):
		"""simple docstring"""







		assert isinstance(lowerCAmelCase   , lowerCAmelCase	)
		assert dataset.num_rows == 4
		assert dataset.num_columns == 1
		assert dataset.column_names == ["text"]
		for feature, expected_dtype in expected_features.items():
				assert dataset.features[feature].dtype == expected_dtype




@pytest.mark.parametrize("keep_in_memory"   , [False, True]	)
def       _snake_case	(   lowerCAmelCase							:   Tuple   , lowerCAmelCase							:   Union[str, Any]   , lowerCAmelCase							:   int	):
		"""simple docstring"""







		SCREAMING_SNAKE_CASE_  :		Union[str, Any]				      =    tmp_path / "cache"
		SCREAMING_SNAKE_CASE_  :		Union[str, Any]				      =    {"text": "string"}
		with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
				SCREAMING_SNAKE_CASE_  :		List[str]				      =    TextDatasetReader(lowerCAmelCase   , cache_dir=lowerCAmelCase   , keep_in_memory=lowerCAmelCase	).read()
		_check_text_dataset(lowerCAmelCase   , lowerCAmelCase	)




@pytest.mark.parametrize(
    "features"   , [
        None,
        {"text": "string"},
        {"text": "int32"},
        {"text": "float32"},
    ]   , )
def       _snake_case	(   lowerCAmelCase							:   Tuple   , lowerCAmelCase							:   Dict   , lowerCAmelCase							:   List[str]	):
		"""simple docstring"""







		SCREAMING_SNAKE_CASE_  :		Any				      =    tmp_path / "cache"
		SCREAMING_SNAKE_CASE_  :		Optional[Any]				      =    {"text": "string"}
		SCREAMING_SNAKE_CASE_  :		Any				      =    features.copy() if features else default_expected_features
		SCREAMING_SNAKE_CASE_  :		List[str]				      =    (
		    Features({feature: Value(lowerCAmelCase	) for feature, dtype in features.items()}	) if features is not None else None
		)
		SCREAMING_SNAKE_CASE_  :		Optional[Any]				      =    TextDatasetReader(lowerCAmelCase   , features=lowerCAmelCase   , cache_dir=lowerCAmelCase	).read()
		_check_text_dataset(lowerCAmelCase   , lowerCAmelCase	)




@pytest.mark.parametrize("split"   , [None, NamedSplit("train"	), "train", "test"]	)
def       _snake_case	(   lowerCAmelCase							:   Union[str, Any]   , lowerCAmelCase							:   Optional[Any]   , lowerCAmelCase							:   List[str]	):
		"""simple docstring"""







		SCREAMING_SNAKE_CASE_  :		Union[str, Any]				      =    tmp_path / "cache"
		SCREAMING_SNAKE_CASE_  :		List[str]				      =    {"text": "string"}
		SCREAMING_SNAKE_CASE_  :		List[str]				      =    TextDatasetReader(lowerCAmelCase   , cache_dir=lowerCAmelCase   , split=lowerCAmelCase	).read()
		_check_text_dataset(lowerCAmelCase   , lowerCAmelCase	)
		assert dataset.split == split if split else "train"




@pytest.mark.parametrize("path_type"   , [str, list]	)
def       _snake_case	(   lowerCAmelCase							:   int   , lowerCAmelCase							:   int   , lowerCAmelCase							:   Optional[int]	):
		"""simple docstring"""







		if issubclass(lowerCAmelCase   , lowerCAmelCase	):
				SCREAMING_SNAKE_CASE_  :		Union[str, Any]				      =    text_path
		elif issubclass(lowerCAmelCase   , lowerCAmelCase	):
				SCREAMING_SNAKE_CASE_  :		Union[str, Any]				      =    [text_path]
		SCREAMING_SNAKE_CASE_  :		int				      =    tmp_path / "cache"
		SCREAMING_SNAKE_CASE_  :		Optional[int]				      =    {"text": "string"}
		SCREAMING_SNAKE_CASE_  :		List[str]				      =    TextDatasetReader(lowerCAmelCase   , cache_dir=lowerCAmelCase	).read()
		_check_text_dataset(lowerCAmelCase   , lowerCAmelCase	)




def       _snake_case	(   lowerCAmelCase							:   Dict   , lowerCAmelCase							:   Dict   , lowerCAmelCase							:   List[str]=("train",)	):
		"""simple docstring"""







		assert isinstance(lowerCAmelCase   , lowerCAmelCase	)
		for split in splits:
				SCREAMING_SNAKE_CASE_  :		int				      =    dataset_dict[split]
				assert dataset.num_rows == 4
				assert dataset.num_columns == 1
				assert dataset.column_names == ["text"]
				for feature, expected_dtype in expected_features.items():
						assert dataset.features[feature].dtype == expected_dtype




@pytest.mark.parametrize("keep_in_memory"   , [False, True]	)
def       _snake_case	(   lowerCAmelCase							:   List[str]   , lowerCAmelCase							:   str   , lowerCAmelCase							:   Dict	):
		"""simple docstring"""







		SCREAMING_SNAKE_CASE_  :		List[str]				      =    tmp_path / "cache"
		SCREAMING_SNAKE_CASE_  :		Union[str, Any]				      =    {"text": "string"}
		with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
				SCREAMING_SNAKE_CASE_  :		List[Any]				      =    TextDatasetReader({"train": text_path}   , cache_dir=lowerCAmelCase   , keep_in_memory=lowerCAmelCase	).read()
		_check_text_datasetdict(lowerCAmelCase   , lowerCAmelCase	)




@pytest.mark.parametrize(
    "features"   , [
        None,
        {"text": "string"},
        {"text": "int32"},
        {"text": "float32"},
    ]   , )
def       _snake_case	(   lowerCAmelCase							:   Union[str, Any]   , lowerCAmelCase							:   List[Any]   , lowerCAmelCase							:   Tuple	):
		"""simple docstring"""







		SCREAMING_SNAKE_CASE_  :		List[Any]				      =    tmp_path / "cache"
		# CSV file loses col_1 string dtype information: default now is "int64" instead of "string"
		SCREAMING_SNAKE_CASE_  :		Tuple				      =    {"text": "string"}
		SCREAMING_SNAKE_CASE_  :		Any				      =    features.copy() if features else default_expected_features
		SCREAMING_SNAKE_CASE_  :		Dict				      =    (
		    Features({feature: Value(lowerCAmelCase	) for feature, dtype in features.items()}	) if features is not None else None
		)
		SCREAMING_SNAKE_CASE_  :		str				      =    TextDatasetReader({"train": text_path}   , features=lowerCAmelCase   , cache_dir=lowerCAmelCase	).read()
		_check_text_datasetdict(lowerCAmelCase   , lowerCAmelCase	)







@pytest.mark.parametrize("split"   , [None, NamedSplit("train"	), "train", "test"]	)
def       _snake_case	(   lowerCAmelCase							:   Optional[Any]   , lowerCAmelCase							:   Any   , lowerCAmelCase							:   Dict	):
		"""simple docstring"""







		if split:
				SCREAMING_SNAKE_CASE_  :		Optional[int]				      =    {split: text_path}
		else:
				SCREAMING_SNAKE_CASE_  :		List[Any]				      =    "train"
				SCREAMING_SNAKE_CASE_  :		Tuple				      =    {"train": text_path, "test": text_path}
		SCREAMING_SNAKE_CASE_  :		Any				      =    tmp_path / "cache"
		SCREAMING_SNAKE_CASE_  :		List[str]				      =    {"text": "string"}
		SCREAMING_SNAKE_CASE_  :		str				      =    TextDatasetReader(lowerCAmelCase   , cache_dir=lowerCAmelCase	).read()
		_check_text_datasetdict(lowerCAmelCase   , lowerCAmelCase   , splits=list(path.keys()	)	)
		assert all(dataset[split].split == split for split in path.keys()	)
