



import gc
import random
import unittest

import numpy as np
import torch
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
from diffusers.utils import floats_tensor, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()


class 	lowerCamelCase__ (		lowerCAmelCase_				,					unittest.TestCase   ):
						_lowerCAmelCase								= KandinskyPipeline
						_lowerCAmelCase								= [
						    '''prompt''',
						    '''image_embeds''',
						    '''negative_image_embeds''',
						]
						_lowerCAmelCase								= ['''prompt''', '''negative_prompt''', '''image_embeds''', '''negative_image_embeds''']
						_lowerCAmelCase								= [
						    '''generator''',
						    '''height''',
						    '''width''',
						    '''latents''',
						    '''guidance_scale''',
						    '''negative_prompt''',
						    '''num_inference_steps''',
						    '''return_dict''',
						    '''guidance_scale''',
						    '''num_images_per_prompt''',
						    '''output_type''',
						    '''return_dict''',
						]
						_lowerCAmelCase								= False


						@property
						def   _lowerCamelCase			( self						:  List[Any]			):
							return 3_2


						@property
						def   _lowerCamelCase			( self						:  Optional[Any]			):
							return 3_2


						@property
						def   _lowerCamelCase			( self						:  Tuple			):
							return self.time_input_dim


						@property
						def   _lowerCamelCase			( self						:  Optional[int]			):
							return self.time_input_dim * 4


						@property
						def   _lowerCamelCase			( self						:  Optional[int]			):
							return 1_0_0


						@property
						def   _lowerCamelCase			( self						:  Tuple			):
							a__:       Union[str, Any]											=XLMRobertaTokenizerFast.from_pretrained("YiYiXu/tiny-random-mclip-base"			)
							return tokenizer


						@property
						def   _lowerCamelCase			( self						:  int			):
							torch.manual_seed(0			)
							a__:       Dict											=MCLIPConfig(
							    numDims=self.cross_attention_dim      ,       transformerDimensions=self.text_embedder_hidden_size      ,       hidden_size=self.text_embedder_hidden_size      ,       intermediate_size=3_7      ,       num_attention_heads=4      ,       num_hidden_layers=5      ,       vocab_size=1_0_0_5      ,       )

							a__:       Dict											=MultilingualCLIP(__a			)
							a__:       Optional[int]											=text_encoder.eval()

							return text_encoder


						@property
						def   _lowerCamelCase			( self						:  Optional[int]			):
							torch.manual_seed(0			)

							a__:       Union[str, Any]											={
							    'in_channels': 4,
							    # Out channels is double in channels because predicts mean and variance
							    'out_channels': 8,
							    'addition_embed_type': 'text_image',
							    'down_block_types': ('ResnetDownsampleBlock2D', 'SimpleCrossAttnDownBlock2D'),
							    'up_block_types': ('SimpleCrossAttnUpBlock2D', 'ResnetUpsampleBlock2D'),
							    'mid_block_type': 'UNetMidBlock2DSimpleCrossAttn',
							    'block_out_channels': (self.block_out_channels_a, self.block_out_channels_a * 2),
							    'layers_per_block': 1,
							    'encoder_hid_dim': self.text_embedder_hidden_size,
							    'encoder_hid_dim_type': 'text_image_proj',
							    'cross_attention_dim': self.cross_attention_dim,
							    'attention_head_dim': 4,
							    'resnet_time_scale_shift': 'scale_shift',
							    'class_embed_type': None,
							}

							a__:       Tuple											=UNetaDConditionModel(**__a			)
							return model


						@property
						def   _lowerCamelCase			( self						:  List[str]			):
							return {
							    "block_out_channels": [3_2, 6_4],
							    "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
							    "in_channels": 3,
							    "latent_channels": 4,
							    "layers_per_block": 1,
							    "norm_num_groups": 8,
							    "norm_type": "spatial",
							    "num_vq_embeddings": 1_2,
							    "out_channels": 3,
							    "up_block_types": [
							        "AttnUpDecoderBlock2D",
							        "UpDecoderBlock2D",
							    ],
							    "vq_embed_dim": 4,
							}


						@property
						def   _lowerCamelCase			( self						:  Dict			):
							torch.manual_seed(0			)
							a__:       Union[str, Any]											=VQModel(**self.dummy_movq_kwargs			)
							return model


						def   _lowerCamelCase			( self						:  Tuple			):
							a__:       Optional[Any]											=self.dummy_text_encoder
							a__:       str											=self.dummy_tokenizer
							a__:       List[str]											=self.dummy_unet
							a__:       List[str]											=self.dummy_movq

							a__:       str											=DDIMScheduler(
							    num_train_timesteps=1_0_0_0      ,       beta_schedule="linear"      ,       beta_start=0.0_0_0_8_5      ,       beta_end=0.0_1_2      ,       clip_sample=__a      ,       set_alpha_to_one=__a      ,       steps_offset=1      ,       prediction_type="epsilon"      ,       thresholding=__a      ,       )

							a__:       List[str]											={
							    'text_encoder': text_encoder,
							    'tokenizer': tokenizer,
							    'unet': unet,
							    'scheduler': scheduler,
							    'movq': movq,
							}
							return components


						def   _lowerCamelCase			( self						:  Dict      ,       _a						:  List[Any]      ,       _a						:  Dict=0			):
							a__:       List[str]											=floats_tensor((1, self.cross_attention_dim)      ,       rng=random.Random(__a			)			).to(__a			)
							a__:       Union[str, Any]											=floats_tensor((1, self.cross_attention_dim)      ,       rng=random.Random(seed + 1			)			).to(__a			)
							if str(__a			).startswith("mps"			):
								a__:       Any											=torch.manual_seed(__a			)
							else:
								a__:       Tuple											=torch.Generator(device=__a			).manual_seed(__a			)
							a__:       Optional[int]											={
							    'prompt': 'horse',
							    'image_embeds': image_embeds,
							    'negative_image_embeds': negative_image_embeds,
							    'generator': generator,
							    'height': 6_4,
							    'width': 6_4,
							    'guidance_scale': 4.0,
							    'num_inference_steps': 2,
							    'output_type': 'np',
							}
							return inputs







						def   _lowerCamelCase			( self						:  List[str]			):
							a__:       Any											='cpu'

							a__:       Union[str, Any]											=self.get_dummy_components()

							a__:       Optional[Any]											=self.pipeline_class(**__a			)
							a__:       Tuple											=pipe.to(__a			)

							pipe.set_progress_bar_config(disable=__a			)

							a__:       List[str]											=pipe(**self.get_dummy_inputs(__a			)			)
							a__:       Tuple											=output.images

							a__:       List[str]											=pipe(
							    **self.get_dummy_inputs(__a			)      ,       return_dict=__a      ,       )[0]

							a__:       List[Any]											=image[0, -3:, -3:, -1]
							a__:       Tuple											=image_from_tuple[0, -3:, -3:, -1]

							assert image.shape == (1, 6_4, 6_4, 3)

							a__:       Union[str, Any]											=np.array(
							    [0.3_2_8_6_6_3, 1.0, 0.2_3_2_1_6_8_7_3, 1.0, 0.9_2_7_1_7_5_6_4, 0.4_6_3_9_0_4_6, 0.9_6_8_9_4_7_7_7, 0.3_1_7_1_3_3_7_8, 0.6_2_9_3_9_5_3]			)

							assert (
							    np.abs(image_slice.flatten() - expected_slice			).max() < 1e-2
							), F" expected_slice {expected_slice}, but got {image_slice.flatten()}"
							assert (
							    np.abs(image_from_tuple_slice.flatten() - expected_slice			).max() < 1e-2
							), F" expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}"


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


						def   _lowerCamelCase			( self						:  Union[str, Any]			):
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()







						def   _lowerCamelCase			( self						:  Optional[Any]			):
							a__:       Optional[int]											=load_numpy(
							    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
							    "/kandinsky/kandinsky_text2img_cat_fp16.npy"			)

							a__:       Union[str, Any]											=KandinskyPriorPipeline.from_pretrained(
							    "kandinsky-community/kandinsky-2-1-prior"      ,       torch_dtype=torch.floataa			)
							pipe_prior.to(__a			)

							a__:       List[Any]											=KandinskyPipeline.from_pretrained("kandinsky-community/kandinsky-2-1"      ,       torch_dtype=torch.floataa			)
							a__:       str											=pipeline.to(__a			)
							pipeline.set_progress_bar_config(disable=__a			)

							a__:       Union[str, Any]											='red cat, 4k photo'

							a__:       Optional[Any]											=torch.Generator(device="cuda"			).manual_seed(0			)
							a__:       Union[str, Any]											=pipe_prior(
							    __a      ,       generator=__a      ,       num_inference_steps=5      ,       negative_prompt=""      ,       ).to_tuple()

							a__:       List[str]											=torch.Generator(device="cuda"			).manual_seed(0			)
							a__:       Union[str, Any]											=pipeline(
							    __a      ,       image_embeds=__a      ,       negative_image_embeds=__a      ,       generator=__a      ,       num_inference_steps=1_0_0      ,       output_type="np"      ,       )

							a__:       str											=output.images[0]

							assert image.shape == (5_1_2, 5_1_2, 3)

							assert_mean_pixel_difference(__a      ,       __a			)










import json
import os
from dataclasses import dataclass

import numpy as np
import pyarrow as pa

import datasets
from utils import get_duration


__UpperCAmelCase						  =     10_00_00_00_00_00
__UpperCAmelCase						  =     1_00_00

__UpperCAmelCase  ,							__UpperCAmelCase						  =     os.path.split(__file__)
__UpperCAmelCase						  =     os.path.join(RESULTS_BASEPATH, '''results''', RESULTS_FILENAME.replace('''.py''', '''.json'''))





def 						__lowerCamelCase       (      __magic_name__	:  int     , __magic_name__	:  int						):
							a__:       Any											=pa.Table.from_pydict({"col": [0] * chunk_size}						)
							a__:       Optional[Any]											=pa.concat_tables([table] * (num_examples // chunk_size)						)
							return datasets.Dataset(__magic_name__     , fingerprint="table_100B"						)


@dataclass
class 	lowerCamelCase__ :
						_lowerCAmelCase								= 42
						_lowerCAmelCase								= 42
						_lowerCAmelCase								= 42
						_lowerCAmelCase								= 42


						def   _lowerCamelCase			( self						:  Union[str, Any]			):
													a__:       Tuple											=np.random.default_rng(self.seed			)
													a__:       Tuple											=rng.integers(low=self.low      ,       high=self.high      ,       size=self.size			).tolist()


						def __iter__( self						:  Any			):
													return iter(self._sampled_values			)







						def __len__( self						:  Any			):
													return self.size





@get_duration
def 						__lowerCamelCase       (      __magic_name__	:  datasets.Dataset						):
							a__:       Union[str, Any]											=dataset[0]





@get_duration
def 						__lowerCamelCase       (      __magic_name__	:  datasets.Dataset						):
							a__:       Optional[Any]											=dataset[-1]





@get_duration
def 						__lowerCamelCase       (      __magic_name__	:  datasets.Dataset						):
							a__:       List[str]											=dataset[range(len(__magic_name__						) // 2     , len(__magic_name__						) // 2 + 1_024						)]





@get_duration
def 						__lowerCamelCase       (      __magic_name__	:  datasets.Dataset						):
							a__:       Optional[int]											=dataset[RandIter(0     , len(__magic_name__						)     , 1_024     , seed=42						)]





def 						__lowerCamelCase       (      ):
							a__:       Dict											={"num examples": SPEED_TEST_N_EXAMPLES}
							a__:       Union[str, Any]											=(get_first_row, get_last_row, get_batch_of_aaaa_rows, get_batch_of_aaaa_random_rows)
							print("generating dataset"						)
							a__:       Dict											=generate_100B_dataset(num_examples=__magic_name__     , chunk_size=__magic_name__						)
							print("Functions"						)
							for func in functions:
														print(func.__name__						)
														a__:       Dict											=func(__magic_name__						)

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


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





