
"""simple docstring"""


import gc
import random
import unittest

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

import diffusers
from diffusers import (
    AutoencoderKL,
    EulerDiscreteScheduler,
    StableDiffusionLatentUpscalePipeline,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.schedulers import KarrasDiffusionSchedulers
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()






def 					_snake_case (						snake_case__      :   Union[str, Any]   ):
	A        =      [tensor.shape for tensor in tensor_list]
	return all(shape == shapes[0] for shape in shapes[1:]   )







class  lowerCAmelCase_ (					_lowercase						,    _lowercase						,    _lowercase						,    unittest.TestCase			):



						'''simple docstring'''


						_lowerCamelCase:			Any					=			StableDiffusionLatentUpscalePipeline
						_lowerCamelCase:			Dict					=			TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
						    '''height''',
						    '''width''',
						    '''cross_attention_kwargs''',
						    '''negative_prompt_embeds''',
						    '''prompt_embeds''',
						}
						_lowerCamelCase:			List[str]					=			PipelineTesterMixin.required_optional_params - {'''num_images_per_prompt'''}
						_lowerCamelCase:			int					=			TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
						_lowerCamelCase:			Union[str, Any]					=			frozenset(
						    []			)  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
						_lowerCamelCase:			Optional[int]					=			frozenset([]			)

						_lowerCamelCase:			Tuple					=			True


						@property
						def   _SCREAMING_SNAKE_CASE					(       self   :					str					)   ->      str:
							A        =      1
							A        =      4
							A        =      (16, 16)

							A        =      floats_tensor((batch_size, num_channels) + sizes     ,rng=random.Random(0					)					).to(A_					)
							return image


						def   _SCREAMING_SNAKE_CASE					(       self   :					int					)   ->      str:
							torch.manual_seed(0					)
							A        =      UNetaDConditionModel(
							    act_fn='gelu'     ,attention_head_dim=8     ,norm_num_groups=A_     ,block_out_channels=[32, 32, 64, 64]     ,time_cond_proj_dim=160     ,conv_in_kernel=1     ,conv_out_kernel=1     ,cross_attention_dim=32     ,down_block_types=(
							        'KDownBlock2D',
							        'KCrossAttnDownBlock2D',
							        'KCrossAttnDownBlock2D',
							        'KCrossAttnDownBlock2D',
							    )     ,in_channels=8     ,mid_block_type=A_     ,only_cross_attention=A_     ,out_channels=5     ,resnet_time_scale_shift='scale_shift'     ,time_embedding_type='fourier'     ,timestep_post_act='gelu'     ,up_block_types=('KCrossAttnUpBlock2D', 'KCrossAttnUpBlock2D', 'KCrossAttnUpBlock2D', 'KUpBlock2D')     ,)
							A        =      AutoencoderKL(
							    block_out_channels=[32, 32, 64, 64]     ,in_channels=3     ,out_channels=3     ,down_block_types=[
							        'DownEncoderBlock2D',
							        'DownEncoderBlock2D',
							        'DownEncoderBlock2D',
							        'DownEncoderBlock2D',
							    ]     ,up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D', 'UpDecoderBlock2D', 'UpDecoderBlock2D']     ,latent_channels=4     ,)
							A        =      EulerDiscreteScheduler(prediction_type='sample'					)
							A        =      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=1000     ,hidden_act='quick_gelu'     ,projection_dim=512     ,)
							A        =      CLIPTextModel(A_					)
							A        =      CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'					)

							A        =      {
							    'unet': model.eval(),
							    'vae': vae.eval(),
							    'scheduler': scheduler,
							    'text_encoder': text_encoder,
							    'tokenizer': tokenizer,
							}

							return components


						def   _SCREAMING_SNAKE_CASE					(       self   :					Any     ,A_   :					Optional[int]     ,A_   :					Union[str, Any]=0					)   ->      List[Any]:
							if str(A_					).startswith('mps'					):
								A        =      torch.manual_seed(A_					)
							else:
								A        =      torch.Generator(device=A_					).manual_seed(A_					)
							A        =      {
							    'prompt': 'A painting of a squirrel eating a burger',
							    'image': self.dummy_image.cpu(),
							    'generator': generator,
							    'num_inference_steps': 2,
							    'output_type': 'numpy',
							}
							return inputs


						def   _SCREAMING_SNAKE_CASE					(       self   :					Dict					)   ->      str:
							A        =      'cpu'

							A        =      self.get_dummy_components()
							A        =      self.pipeline_class(**A_					)
							pipe.to(A_					)
							pipe.set_progress_bar_config(disable=A_					)

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

							self.assertEqual(image.shape     ,(1, 256, 256, 3)					)
							A        =      np.array(
							    [0.47_22_24_12, 0.41_92_16_33, 0.44_71_74_34, 0.46_87_41_92, 0.42_58_82_58, 0.46_15_07_26, 0.4_67_75_34, 0.45_58_38_32, 0.48_57_90_55]					)
							A        =      np.abs(image_slice.flatten() - expected_slice					).max()
							self.assertLessEqual(A_     ,1e-3					)


						def   _SCREAMING_SNAKE_CASE					(       self   :					Optional[int]					)   ->      Any:
							super().test_attention_slicing_forward_pass(expected_max_diff=7e-3					)


						def   _SCREAMING_SNAKE_CASE					(       self   :					Union[str, Any]					)   ->      Optional[Any]:
							super().test_cpu_offload_forward_pass(expected_max_diff=3e-3					)


						def   _SCREAMING_SNAKE_CASE					(       self   :					Optional[int]					)   ->      Dict:
							super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3					)


						def   _SCREAMING_SNAKE_CASE					(       self   :					int					)   ->      Optional[int]:
							super().test_inference_batch_single_identical(expected_max_diff=7e-3					)


						def   _SCREAMING_SNAKE_CASE					(       self   :					str					)   ->      Dict:
							super().test_pt_np_pil_outputs_equivalent(expected_max_diff=3e-3					)


						def   _SCREAMING_SNAKE_CASE					(       self   :					Optional[int]					)   ->      Any:
							super().test_save_load_local(expected_max_difference=3e-3					)


						def   _SCREAMING_SNAKE_CASE					(       self   :					str					)   ->      Dict:
							super().test_save_load_optional_components(expected_max_difference=3e-3					)




						def   _SCREAMING_SNAKE_CASE					(       self   :					Dict					)   ->      Any:
							A        =      [
							    'DDIMScheduler',
							    'DDPMScheduler',
							    'PNDMScheduler',
							    'HeunDiscreteScheduler',
							    'EulerAncestralDiscreteScheduler',
							    'KDPM2DiscreteScheduler',
							    'KDPM2AncestralDiscreteScheduler',
							    'DPMSolverSDEScheduler',
							]
							A        =      self.get_dummy_components()
							A        =      self.pipeline_class(**A_					)

							# make sure that PNDM does not need warm-up
							pipe.scheduler.register_to_config(skip_prk_steps=A_					)

							pipe.to(A_					)
							pipe.set_progress_bar_config(disable=A_					)
							A        =      self.get_dummy_inputs(A_					)
							A        =      2

							A        =      []
							for scheduler_enum in KarrasDiffusionSchedulers:
								if scheduler_enum.name in skip_schedulers:
									# no sigma schedulers are not supported
									# no schedulers
									continue

								A        =      getattr(A_     ,scheduler_enum.name					)
								A        =      scheduler_cls.from_config(pipe.scheduler.config					)
								A        =      pipe(**A_					)[0]
								outputs.append(A_					)

							assert check_same_shape(A_					)







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



						'''simple docstring'''


						def   _SCREAMING_SNAKE_CASE					(       self   :					Union[str, Any]					)   ->      int:
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()


						def   _SCREAMING_SNAKE_CASE					(       self   :					Optional[Any]					)   ->      int:
							A        =      torch.manual_seed(33					)

							A        =      StableDiffusionPipeline.from_pretrained('CompVis/stable-diffusion-v1-4'     ,torch_dtype=torch.floataa					)
							pipe.to('cuda'					)

							A        =      StableDiffusionLatentUpscalePipeline.from_pretrained(
							    'stabilityai/sd-x2-latent-upscaler'     ,torch_dtype=torch.floataa					)
							upscaler.to('cuda'					)

							A        =      'a photo of an astronaut high resolution, unreal engine, ultra realistic'

							A        =      pipe(A_     ,generator=A_     ,output_type='latent'					).images

							A        =      upscaler(
							    prompt=A_     ,image=A_     ,num_inference_steps=20     ,guidance_scale=0     ,generator=A_     ,output_type='np'     ,).images[0]

							A        =      load_numpy(
							    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/astronaut_1024.npy'					)
							assert np.abs((expected_image - image).mean()					) < 5e-2




						def   _SCREAMING_SNAKE_CASE					(       self   :					List[str]					)   ->      Optional[Any]:
							A        =      torch.manual_seed(33					)

							A        =      StableDiffusionLatentUpscalePipeline.from_pretrained(
							    'stabilityai/sd-x2-latent-upscaler'     ,torch_dtype=torch.floataa					)
							upscaler.to('cuda'					)

							A        =      'the temple of fire by Ross Tran and Gerardo Dottori, oil on canvas'

							A        =      load_image(
							    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_512.png'					)

							A        =      upscaler(
							    prompt=A_     ,image=A_     ,num_inference_steps=20     ,guidance_scale=0     ,generator=A_     ,output_type='np'     ,).images[0]

							A        =      load_numpy(
							    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_1024.npy'					)
							assert np.abs((expected_image - image).max()					) < 5e-2






'''simple docstring'''







from ...configuration_utils import PretrainedConfig
from ...utils import logging


lowerCAmelCase_					:    Dict      	=			logging.get_logger(__name__)

lowerCAmelCase_					:    int      	=			{
    'bigcode/gpt_bigcode-santacoder': 'https://huggingface.co/bigcode/gpt_bigcode-santacoder/resolve/main/config.json',
}
class 				__SCREAMING_SNAKE_CASE  (lowerCamelCase_      ):






				"""simple docstring"""







				__a            ='gpt_bigcode'
				__a            =['past_key_values']
				__a            ={
				    'hidden_size': 'n_embd',
				    'max_position_embeddings': 'n_positions',
				    'num_attention_heads': 'n_head',
				    'num_hidden_layers': 'n_layer',
				}




				def __init__(  self     :					Optional[Any]					,				__a     :					Tuple=5_02_57					,				__a     :					str=10_24					,				__a     :					Dict=7_68					,				__a     :					Tuple=12					,				__a     :					str=12					,				__a     :					Optional[int]=None					,				__a     :					Dict="gelu_pytorch_tanh"					,				__a     :					Tuple=0.1					,				__a     :					Tuple=0.1					,				__a     :					Union[str, Any]=0.1					,				__a     :					Tuple=1e-5					,				__a     :					str=0.02					,				__a     :					Dict=True					,				__a     :					Union[str, Any]=True					,				__a     :					Optional[int]=5_02_56					,				__a     :					Optional[int]=5_02_56					,				__a     :					Union[str, Any]=True					,				__a     :					Dict=True					,				__a     :					Union[str, Any]=True					,				**__a     :					List[Any]					,				):
								_a								= vocab_size
								_a								= n_positions
								_a								= n_embd
								_a								= n_layer
								_a								= n_head
								_a								= n_inner
								_a								= activation_function
								_a								= resid_pdrop
								_a								= embd_pdrop
								_a								= attn_pdrop
								_a								= layer_norm_epsilon
								_a								= initializer_range
								_a								= scale_attn_weights
								_a								= use_cache
								_a								= attention_softmax_in_fpaa
								_a								= scale_attention_softmax_in_fpaa
								_a								= multi_query

								_a								= bos_token_id
								_a								= eos_token_id

								super().__init__(bos_token_id=__a					,				eos_token_id=__a					,				**__a					)
