



"""simple docstring"""

import gc
import random
import unittest

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModel,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import AutoencoderKL, DDIMScheduler, DDPMScheduler, StableUnCLIPImgaImgPipeline, UNetaDConditionModel
from diffusers.pipelines.pipeline_utils import DiffusionPipeline
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    floats_tensor,
    load_image,
    load_numpy,
    require_torch_gpu,
    skip_mps,
    slow,
    torch_device,
)

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()







class    __lowercase     (					_UpperCAmelCase					,    _UpperCAmelCase					,    _UpperCAmelCase					,    unittest.TestCase):







						"""simple docstring"""







						_A		:							Tuple							       =		StableUnCLIPImgaImgPipeline
						_A		:							Dict							       =		TEXT_GUIDED_IMAGE_VARIATION_PARAMS
						_A		:							Dict							       =		TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
						_A		:							str							       =		frozenset(
						    [])  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
						_A		:							int							       =		frozenset([])







						def 						__UpperCamelCase							(self       ):
											snake_case_					:		Tuple 					=					32
											snake_case_					:		Union[str, Any] 					=					embedder_hidden_size

											# image encoding components

											snake_case_					:		int 					=					CLIPImageProcessor(crop_size=32   ,		size=32       )

											torch.manual_seed(0       )
											snake_case_					:		Any 					=					CLIPVisionModelWithProjection(
											    CLIPVisionConfig(
											        hidden_size=lowercase__   ,		projection_dim=lowercase__   ,		num_hidden_layers=5   ,		num_attention_heads=4   ,		image_size=32   ,		intermediate_size=37   ,		patch_size=1   ,		)       )

											# regular denoising components

											torch.manual_seed(0       )
											snake_case_					:		Union[str, Any] 					=					StableUnCLIPImageNormalizer(embedding_dim=lowercase__       )
											snake_case_					:		Dict 					=					DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""       )

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

											torch.manual_seed(0       )
											snake_case_					:		Dict 					=					CLIPTextModel(
											    CLIPTextConfig(
											        bos_token_id=0   ,		eos_token_id=2   ,		hidden_size=lowercase__   ,		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       )
											snake_case_					:		Optional[Any] 					=					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=lowercase__   ,		layers_per_block=1   ,		upcast_attention=lowercase__   ,		use_linear_projection=lowercase__   ,		)

											torch.manual_seed(0       )
											snake_case_					:		Tuple 					=					DDIMScheduler(
											    beta_schedule="""scaled_linear"""   ,		beta_start=0.00085   ,		beta_end=0.012   ,		prediction_type="""v_prediction"""   ,		set_alpha_to_one=lowercase__   ,		steps_offset=1   ,		)

											torch.manual_seed(0       )
											snake_case_					:		int 					=					AutoencoderKL()

											snake_case_					:		str 					=					{
											    # image encoding components
											    """feature_extractor""": feature_extractor,
											    """image_encoder""": image_encoder.eval(),
											    # image noising components
											    """image_normalizer""": image_normalizer.eval(),
											    """image_noising_scheduler""": image_noising_scheduler,
											    # regular denoising components
											    """tokenizer""": tokenizer,
											    """text_encoder""": text_encoder.eval(),
											    """unet""": unet.eval(),
											    """scheduler""": scheduler,
											    """vae""": vae.eval(),
											}

											return components







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__=0   ,		lowercase__=True       ):
											if str(lowercase__       ).startswith("""mps"""       ):
																snake_case_					:		List[Any] 					=					torch.manual_seed(lowercase__       )
											else:
																snake_case_					:		Dict 					=					torch.Generator(device=lowercase__       ).manual_seed(lowercase__       )

											snake_case_					:		List[Any] 					=					floats_tensor((1, 3, 32, 32)   ,		rng=random.Random(lowercase__       )       ).to(lowercase__       )

											if pil_image:
																snake_case_					:		Tuple 					=					input_image * 0.5 + 0.5
																snake_case_					:		Optional[int] 					=					input_image.clamp(0   ,		1       )
																snake_case_					:		Dict 					=					input_image.cpu().permute(0   ,		2   ,		3   ,		1       ).float().numpy()
																snake_case_					:		Tuple 					=					DiffusionPipeline.numpy_to_pil(lowercase__       )[0]

											return {
											    "prompt": "An anime racoon running a marathon",
											    "image": input_image,
											    "generator": generator,
											    "num_inference_steps": 2,
											    "output_type": "np",
											}







						@skip_mps
						def 						__UpperCamelCase							(self       ):
											snake_case_					:		Union[str, Any] 					=					"""cpu"""  # ensure determinism for the device-dependent torch.Generator
											snake_case_					:		Union[str, Any] 					=					self.get_dummy_components()
											snake_case_					:		str 					=					StableUnCLIPImgaImgPipeline(**lowercase__       )
											snake_case_					:		Dict 					=					sd_pipe.to(lowercase__       )
											sd_pipe.set_progress_bar_config(disable=lowercase__       )

											snake_case_					:		int 					=					self.get_dummy_inputs(lowercase__       )
											inputs.update({"""image_embeds""": None}       )
											snake_case_					:		int 					=					sd_pipe(**lowercase__       ).images
											snake_case_					:		Dict 					=					image[0, -3:, -3:, -1]

											assert image.shape == (1, 32, 32, 3)
											snake_case_					:		Optional[int] 					=					np.array([0.3872, 0.7224, 0.5601, 0.4741, 0.6872, 0.5814, 0.4636, 0.3867, 0.5078]       )

											assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-3







						def 						__UpperCamelCase							(self       ):
											snake_case_					:		int 					=					torch_device in ["""cpu""", """mps"""]

											self._test_attention_slicing_forward_pass(test_max_difference=lowercase__       )







						def 						__UpperCamelCase							(self       ):
											snake_case_					:		Dict 					=					torch_device in ["""cpu""", """mps"""]

											self._test_inference_batch_single_identical(test_max_difference=lowercase__       )







						@unittest.skipIf(
						    torch_device != """cuda""" or not is_xformers_available()   ,		reason="""XFormers attention is only available with CUDA and `xformers` installed"""   ,		)
						def 						__UpperCamelCase							(self       ):
											self._test_xformers_attention_forwardGenerator_pass(test_max_difference=lowercase__       )







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







						"""simple docstring"""







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







						def 						__UpperCamelCase							(self       ):
											snake_case_					:		Optional[Any] 					=					load_image(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"""       )

											snake_case_					:		Any 					=					load_numpy(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_img2img_anime_turtle_fp16.npy"""       )

											snake_case_					:		str 					=					StableUnCLIPImgaImgPipeline.from_pretrained(
											    """fusing/stable-unclip-2-1-l-img2img"""   ,		torch_dtype=torch.floataa       )
											pipe.to(lowercase__       )
											pipe.set_progress_bar_config(disable=lowercase__       )
											# 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()

											snake_case_					:		List[Any] 					=					torch.Generator(device="""cpu"""       ).manual_seed(0       )
											snake_case_					:		List[str] 					=					pipe(lowercase__   ,		"""anime turle"""   ,		generator=lowercase__   ,		output_type="""np"""       )

											snake_case_					:		int 					=					output.images[0]

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

											assert_mean_pixel_difference(lowercase__   ,		lowercase__       )







						def 						__UpperCamelCase							(self       ):
											snake_case_					:		Union[str, Any] 					=					load_image(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"""       )

											snake_case_					:		str 					=					load_numpy(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_h_img2img_anime_turtle_fp16.npy"""       )

											snake_case_					:		Dict 					=					StableUnCLIPImgaImgPipeline.from_pretrained(
											    """fusing/stable-unclip-2-1-h-img2img"""   ,		torch_dtype=torch.floataa       )
											pipe.to(lowercase__       )
											pipe.set_progress_bar_config(disable=lowercase__       )
											# 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()

											snake_case_					:		Tuple 					=					torch.Generator(device="""cpu"""       ).manual_seed(0       )
											snake_case_					:		Dict 					=					pipe(lowercase__   ,		"""anime turle"""   ,		generator=lowercase__   ,		output_type="""np"""       )

											snake_case_					:		Dict 					=					output.images[0]

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

											assert_mean_pixel_difference(lowercase__   ,		lowercase__       )







						def 						__UpperCamelCase							(self       ):
											snake_case_					:		Optional[Any] 					=					load_image(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"""       )

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

											snake_case_					:		List[str] 					=					StableUnCLIPImgaImgPipeline.from_pretrained(
											    """fusing/stable-unclip-2-1-h-img2img"""   ,		torch_dtype=torch.floataa       )
											snake_case_					:		List[str] 					=					pipe.to(lowercase__       )
											pipe.set_progress_bar_config(disable=lowercase__       )
											pipe.enable_attention_slicing()
											pipe.enable_sequential_cpu_offload()

											snake_case_					:		str 					=					pipe(
											    lowercase__   ,		"""anime turtle"""   ,		num_inference_steps=2   ,		output_type="""np"""   ,		)

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









"""simple docstring"""

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 							SCREAMING_SNAKE_CASE__      (		SCREAMING_SNAKE_CASE__ :  List[Any]				,    SCREAMING_SNAKE_CASE__ :  Dict			):



					"""simple docstring"""





					assert isinstance(SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__			)
					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 							SCREAMING_SNAKE_CASE__      (		SCREAMING_SNAKE_CASE__ :  Union[str, Any]				,    SCREAMING_SNAKE_CASE__ :  Optional[Any]				,    SCREAMING_SNAKE_CASE__ :  Optional[Any]			):



					"""simple docstring"""





					snake_case_					:		Optional[Any] 					=					tmp_path / """cache"""
					snake_case_					:		Optional[int] 					=					{"""text""": """string"""}
					with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
										snake_case_					:		Tuple 					=					TextDatasetReader(SCREAMING_SNAKE_CASE__				,    cache_dir=SCREAMING_SNAKE_CASE__				,    keep_in_memory=SCREAMING_SNAKE_CASE__			).read()
					_check_text_dataset(SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__			)







@pytest.mark.parametrize(
    """features"""				,    [
        None,
        {"""text""": """string"""},
        {"""text""": """int32"""},
        {"""text""": """float32"""},
    ]				,    )
def 							SCREAMING_SNAKE_CASE__      (		SCREAMING_SNAKE_CASE__ :  Dict				,    SCREAMING_SNAKE_CASE__ :  List[str]				,    SCREAMING_SNAKE_CASE__ :  Union[str, Any]			):



					"""simple docstring"""





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







@pytest.mark.parametrize("""split"""				,    [None, NamedSplit("""train"""			), """train""", """test"""]			)
def 							SCREAMING_SNAKE_CASE__      (		SCREAMING_SNAKE_CASE__ :  Tuple				,    SCREAMING_SNAKE_CASE__ :  int				,    SCREAMING_SNAKE_CASE__ :  str			):



					"""simple docstring"""





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







@pytest.mark.parametrize("""path_type"""				,    [str, list]			)
def 							SCREAMING_SNAKE_CASE__      (		SCREAMING_SNAKE_CASE__ :  Tuple				,    SCREAMING_SNAKE_CASE__ :  List[str]				,    SCREAMING_SNAKE_CASE__ :  Dict			):



					"""simple docstring"""





					if issubclass(SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__			):
										snake_case_					:		List[str] 					=					text_path
					elif issubclass(SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__			):
										snake_case_					:		str 					=					[text_path]
					snake_case_					:		List[str] 					=					tmp_path / """cache"""
					snake_case_					:		List[str] 					=					{"""text""": """string"""}
					snake_case_					:		Dict 					=					TextDatasetReader(SCREAMING_SNAKE_CASE__				,    cache_dir=SCREAMING_SNAKE_CASE__			).read()
					_check_text_dataset(SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__			)







def 							SCREAMING_SNAKE_CASE__      (		SCREAMING_SNAKE_CASE__ :  str				,    SCREAMING_SNAKE_CASE__ :  Dict				,    SCREAMING_SNAKE_CASE__ :  List[str]=("train",)			):



					"""simple docstring"""





					assert isinstance(SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__			)
					for split in splits:
										snake_case_					:		Dict 					=					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 							SCREAMING_SNAKE_CASE__      (		SCREAMING_SNAKE_CASE__ :  List[Any]				,    SCREAMING_SNAKE_CASE__ :  Tuple				,    SCREAMING_SNAKE_CASE__ :  Any			):



					"""simple docstring"""





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







@pytest.mark.parametrize(
    """features"""				,    [
        None,
        {"""text""": """string"""},
        {"""text""": """int32"""},
        {"""text""": """float32"""},
    ]				,    )
def 							SCREAMING_SNAKE_CASE__      (		SCREAMING_SNAKE_CASE__ :  int				,    SCREAMING_SNAKE_CASE__ :  int				,    SCREAMING_SNAKE_CASE__ :  List[str]			):



					"""simple docstring"""





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







@pytest.mark.parametrize("""split"""				,    [None, NamedSplit("""train"""			), """train""", """test"""]			)
def 							SCREAMING_SNAKE_CASE__      (		SCREAMING_SNAKE_CASE__ :  Dict				,    SCREAMING_SNAKE_CASE__ :  Dict				,    SCREAMING_SNAKE_CASE__ :  Any			):



					"""simple docstring"""





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




