
'''simple docstring'''







import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, DDIMScheduler, LDMTextToImagePipeline, UNetaDConditionModel
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    load_numpy,
    nightly,
    require_torch_gpu,
    slow,
    torch_device,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()


class 						__UpperCAmelCase      (					_lowerCamelCase						,   unittest.TestCase						):
							__lowercase      		=       LDMTextToImagePipeline
							__lowercase      		=       TEXT_TO_IMAGE_PARAMS - {
							    """negative_prompt""",
							    """negative_prompt_embeds""",
							    """cross_attention_kwargs""",
							    """prompt_embeds""",
							}
							__lowercase      		=       PipelineTesterMixin.required_optional_params - {
							    """num_images_per_prompt""",
							    """callback""",
							    """callback_steps""",
							}
							__lowercase      		=       TEXT_TO_IMAGE_BATCH_PARAMS
							__lowercase      		=       False


							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								torch.manual_seed(0  )
								_snake_case   						=		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    ,   )
								_snake_case   						=		DDIMScheduler(
								    beta_start=0.00085    ,   beta_end=0.012    ,   beta_schedule='scaled_linear'    ,   clip_sample=lowerCAmelCase_    ,   set_alpha_to_one=lowerCAmelCase_    ,   )
								torch.manual_seed(0  )
								_snake_case   						=		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  )
								_snake_case   						=		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=10_00    ,   )
								_snake_case   						=		CLIPTextModel(lowerCAmelCase_  )
								_snake_case   						=		CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'  )

								_snake_case   						=		{
								    'unet': unet,
								    'scheduler': scheduler,
								    'vqvae': vae,
								    'bert': text_encoder,
								    'tokenizer': tokenizer,
								}
								return components


							def  lowerCamelCase				(       self    ,   lowerCAmelCase_    ,   lowerCAmelCase_=0  ):





								"""simple docstring"""






								if str(lowerCAmelCase_  ).startswith('mps'  ):
									_snake_case   						=		torch.manual_seed(lowerCAmelCase_  )
								else:
									_snake_case   						=		torch.Generator(device=lowerCAmelCase_  ).manual_seed(lowerCAmelCase_  )
								_snake_case   						=		{
								    'prompt': 'A painting of a squirrel eating a burger',
								    'generator': generator,
								    'num_inference_steps': 2,
								    'guidance_scale': 6.0,
								    'output_type': 'numpy',
								}
								return inputs




							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								_snake_case   						=		'cpu'  # ensure determinism for the device-dependent torch.Generator

								_snake_case   						=		self.get_dummy_components()
								_snake_case   						=		LDMTextToImagePipeline(**lowerCAmelCase_  )
								pipe.to(lowerCAmelCase_  )
								pipe.set_progress_bar_config(disable=lowerCAmelCase_  )

								_snake_case   						=		self.get_dummy_inputs(lowerCAmelCase_  )
								_snake_case   						=		pipe(**lowerCAmelCase_  ).images
								_snake_case   						=		image[0, -3:, -3:, -1]

								assert image.shape == (1, 16, 16, 3)
								_snake_case   						=		np.array([0.6101, 0.6156, 0.5622, 0.4895, 0.6661, 0.3804, 0.5748, 0.6136, 0.5014]  )

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


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


							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






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


							def  lowerCamelCase				(       self    ,   lowerCAmelCase_    ,   lowerCAmelCase_=torch.floataa    ,   lowerCAmelCase_=0  ):





								"""simple docstring"""






								_snake_case   						=		torch.manual_seed(lowerCAmelCase_  )
								_snake_case   						=		np.random.RandomState(lowerCAmelCase_  ).standard_normal((1, 4, 32, 32)  )
								_snake_case   						=		torch.from_numpy(lowerCAmelCase_  ).to(device=lowerCAmelCase_    ,   dtype=lowerCAmelCase_  )
								_snake_case   						=		{
								    'prompt': 'A painting of a squirrel eating a burger',
								    'latents': latents,
								    'generator': generator,
								    'num_inference_steps': 3,
								    'guidance_scale': 6.0,
								    'output_type': 'numpy',
								}
								return inputs




							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								_snake_case   						=		LDMTextToImagePipeline.from_pretrained('CompVis/ldm-text2im-large-256'  ).to(lowerCAmelCase_  )
								pipe.set_progress_bar_config(disable=lowerCAmelCase_  )

								_snake_case   						=		self.get_inputs(lowerCAmelCase_  )
								_snake_case   						=		pipe(**lowerCAmelCase_  ).images
								_snake_case   						=		image[0, -3:, -3:, -1].flatten()

								assert image.shape == (1, 2_56, 2_56, 3)
								_snake_case   						=		np.array([0.51825, 0.52850, 0.52543, 0.54258, 0.52304, 0.52569, 0.54363, 0.55276, 0.56878]  )
								_snake_case   						=		np.abs(expected_slice - image_slice  ).max()
								assert max_diff < 1E-3


@nightly
@require_torch_gpu
class 						__UpperCAmelCase      (					unittest.TestCase						):


							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






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


							def  lowerCamelCase				(       self    ,   lowerCAmelCase_    ,   lowerCAmelCase_=torch.floataa    ,   lowerCAmelCase_=0  ):





								"""simple docstring"""






								_snake_case   						=		torch.manual_seed(lowerCAmelCase_  )
								_snake_case   						=		np.random.RandomState(lowerCAmelCase_  ).standard_normal((1, 4, 32, 32)  )
								_snake_case   						=		torch.from_numpy(lowerCAmelCase_  ).to(device=lowerCAmelCase_    ,   dtype=lowerCAmelCase_  )
								_snake_case   						=		{
								    'prompt': 'A painting of a squirrel eating a burger',
								    'latents': latents,
								    'generator': generator,
								    'num_inference_steps': 50,
								    'guidance_scale': 6.0,
								    'output_type': 'numpy',
								}
								return inputs




							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								_snake_case   						=		LDMTextToImagePipeline.from_pretrained('CompVis/ldm-text2im-large-256'  ).to(lowerCAmelCase_  )
								pipe.set_progress_bar_config(disable=lowerCAmelCase_  )

								_snake_case   						=		self.get_inputs(lowerCAmelCase_  )
								_snake_case   						=		pipe(**lowerCAmelCase_  ).images[0]

								_snake_case   						=		load_numpy(
								    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/ldm_text2img/ldm_large_256_ddim.npy'  )
								_snake_case   						=		np.abs(expected_image - image  ).max()
								assert max_diff < 1E-3






"""simple docstring"""






import json
import os
import tempfile

import datasets
from utils import generate_example_dataset, get_duration


__snake_case							:   Optional[int]      =    50_000
__snake_case							:   Dict      =    5_000

__snake_case      ,		__snake_case							:   Union[str, Any]      =    os.path.split(__file__)
__snake_case							:   Any      =    os.path.join(RESULTS_BASEPATH, 'results', RESULTS_FILENAME.replace('.py', '.json'))


@get_duration
def 			_lowercase				(  __snake_case					,__snake_case	)						->   Dict:
			for i in range(__snake_case	):
						__lowerCAmelCase      :  Union[str, Any]												=       dataset[i]


@get_duration
def 			_lowercase				(  __snake_case					,__snake_case					,__snake_case	)						->   Dict:
			for i in range(0					,len(__snake_case	)					,__snake_case	):
						__lowerCAmelCase      :  List[str]												=       dataset[i : i + batch_size]


@get_duration
def 			_lowercase				(  __snake_case					,__snake_case					,__snake_case	)						->   Dict:
			with dataset.formatted_as(type=__snake_case	):
						for i in range(__snake_case	):
									__lowerCAmelCase      :  Union[str, Any]												=       dataset[i]


@get_duration
def 			_lowercase				(  __snake_case					,__snake_case					,__snake_case					,__snake_case	)						->   str:
			with dataset.formatted_as(type=__snake_case	):
						for i in range(0					,__snake_case					,__snake_case	):
									__lowerCAmelCase      :  Optional[int]												=       dataset[i : i + batch_size]


def 			_lowercase				(  )						->   Union[str, Any]:
			__lowerCAmelCase      :  Optional[int]												=       {"num examples": SPEED_TEST_N_EXAMPLES}
			__lowerCAmelCase      :  Optional[int]												=       [
			    (read, {"length": SMALL_TEST}),
			    (read, {"length": SPEED_TEST_N_EXAMPLES}),
			    (read_batch, {"length": SPEED_TEST_N_EXAMPLES, "batch_size": 10}),
			    (read_batch, {"length": SPEED_TEST_N_EXAMPLES, "batch_size": 100}),
			    (read_batch, {"length": SPEED_TEST_N_EXAMPLES, "batch_size": 1_000}),
			    (read_formatted, {"type": "numpy", "length": SMALL_TEST}),
			    (read_formatted, {"type": "pandas", "length": SMALL_TEST}),
			    (read_formatted, {"type": "torch", "length": SMALL_TEST}),
			    (read_formatted, {"type": "tensorflow", "length": SMALL_TEST}),
			    (read_formatted_batch, {"type": "numpy", "length": SMALL_TEST, "batch_size": 10}),
			    (read_formatted_batch, {"type": "numpy", "length": SMALL_TEST, "batch_size": 1_000}),
			]

			__lowerCAmelCase      :  Any												=       [
			    (read, {"length": SMALL_TEST}),
			    (read, {"length": SPEED_TEST_N_EXAMPLES}),
			    (read_batch, {"length": SPEED_TEST_N_EXAMPLES, "batch_size": 10}),
			    (read_batch, {"length": SPEED_TEST_N_EXAMPLES, "batch_size": 100}),
			    (read_batch, {"length": SPEED_TEST_N_EXAMPLES, "batch_size": 1_000}),
			    (read_formatted, {"type": "numpy", "length": SMALL_TEST}),
			    (read_formatted_batch, {"type": "numpy", "length": SMALL_TEST, "batch_size": 10}),
			    (read_formatted_batch, {"type": "numpy", "length": SMALL_TEST, "batch_size": 1_000}),
			]
			with tempfile.TemporaryDirectory() as tmp_dir:
						print("generating dataset"	)
						__lowerCAmelCase      :  int												=       datasets.Features(
						    {"list": datasets.Sequence(datasets.Value("float32"	)	), "numbers": datasets.Value("float32"	)}	)
						__lowerCAmelCase      :  str												=       generate_example_dataset(
						    os.path.join(__snake_case					,"dataset.arrow"	)					,__snake_case					,num_examples=__snake_case					,seq_shapes={"list": (100,)}					,)
						print("first set of iterations"	)
						for func, kwargs in functions:
									print(func.__name__					,str(__snake_case	)	)
									__lowerCAmelCase      :  str												=       func(__snake_case					,**__snake_case	)

						print("shuffling dataset"	)
						__lowerCAmelCase      :  Optional[int]												=       dataset.shuffle()
						print("Second set of iterations (after shuffling"	)
						for func, kwargs in functions_shuffled:
									print("shuffled "					,func.__name__					,str(__snake_case	)	)
									__lowerCAmelCase      :  List[str]												=       func(
									    __snake_case					,**__snake_case	)

			with open(__snake_case					,"wb"	) as f:
						f.write(json.dumps(__snake_case	).encode("utf-8"	)	)


if __name__ == "__main__":  # useful to run the profiler
	benchmark_iterating()