
"""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




from __future__ import absolute_import, division, print_function, unicode_literals

from torch import nn
from torch.nn import CrossEntropyLoss, MSELoss

from transformers import RobertaConfig
from transformers.file_utils import add_start_docstrings, add_start_docstrings_to_model_forward
from transformers.models.roberta.modeling_roberta import (
    ROBERTA_INPUTS_DOCSTRING,
    ROBERTA_START_DOCSTRING,
    RobertaEmbeddings,
)

from .modeling_highway_bert import BertPreTrainedModel, DeeBertModel, HighwayException, entropy



@add_start_docstrings(
    'The RoBERTa Model transformer with early exiting (DeeRoBERTa). '   ,					a   ,					)
class   UpperCAmelCase_    (       a):
							lowerCamelCase__   			=		RobertaConfig
							lowerCamelCase__   			=		'roberta'







							def __init__(   self,			__a):

														'''simple docstring'''

														super().__init__(__a)

														_lowerCAmelCase     :							Optional[Any]    =    RobertaEmbeddings(__a)
														self.init_weights()



@add_start_docstrings(
    'RoBERTa Model (with early exiting - DeeRoBERTa) with a classifier on top,\n    also takes care of multi-layer training. '   ,					a   ,					)
class   UpperCAmelCase_    (       a):
							lowerCamelCase__   			=		RobertaConfig
							lowerCamelCase__   			=		'roberta'







							def __init__(   self,			__a):

														'''simple docstring'''

														super().__init__(__a)
														_lowerCAmelCase     :							Optional[int]    =    config.num_labels
														_lowerCAmelCase     :							Optional[int]    =    config.num_hidden_layers

														_lowerCAmelCase     :							Optional[int]    =    DeeRobertaModel(__a)
														_lowerCAmelCase     :							Union[str, Any]    =    nn.Dropout(config.hidden_dropout_prob)
														_lowerCAmelCase     :							List[str]    =    nn.Linear(config.hidden_size,			self.config.num_labels)







							@add_start_docstrings_to_model_forward(__a)
							def    snake_case__      (   self,			__a=None,			__a=None,			__a=None,			__a=None,			__a=None,			__a=None,			__a=None,			__a=-1,			__a=False,			):

														'''simple docstring'''

														_lowerCAmelCase     :							Union[str, Any]    =    self.num_layers
														try:
																					_lowerCAmelCase     :							List[Any]    =    self.roberta(
																					    __a,			attention_mask=__a,			token_type_ids=__a,			position_ids=__a,			head_mask=__a,			inputs_embeds=__a,			)

																					_lowerCAmelCase     :							List[Any]    =    outputs[1]

																					_lowerCAmelCase     :							Dict    =    self.dropout(__a)
																					_lowerCAmelCase     :							Dict    =    self.classifier(__a)
																					_lowerCAmelCase     :							Optional[Any]    =    (logits,) + outputs[2:]  # add hidden states and attention if they are here
														except HighwayException as e:
																					_lowerCAmelCase     :							Tuple    =    e.message
																					_lowerCAmelCase     :							Union[str, Any]    =    e.exit_layer
																					_lowerCAmelCase     :							List[Any]    =    outputs[0]

														if not self.training:
																					_lowerCAmelCase     :							int    =    entropy(__a)
																					_lowerCAmelCase     :							List[Any]    =    []
																					_lowerCAmelCase     :							str    =    []
														if labels is not None:
																					if self.num_labels == 1:
																												#  We are doing regression
																												_lowerCAmelCase     :							Optional[Any]    =    MSELoss()
																												_lowerCAmelCase     :							int    =    loss_fct(logits.view(-1),			labels.view(-1))
																					else:
																												_lowerCAmelCase     :							Optional[Any]    =    CrossEntropyLoss()
																												_lowerCAmelCase     :							Optional[Any]    =    loss_fct(logits.view(-1,			self.num_labels),			labels.view(-1))

																					# work with highway exits
																					_lowerCAmelCase     :							Optional[int]    =    []
																					for highway_exit in outputs[-1]:
																												_lowerCAmelCase     :							Any    =    highway_exit[0]
																												if not self.training:
																																			highway_logits_all.append(__a)
																																			highway_entropy.append(highway_exit[2])
																												if self.num_labels == 1:
																																			#  We are doing regression
																																			_lowerCAmelCase     :							List[str]    =    MSELoss()
																																			_lowerCAmelCase     :							List[Any]    =    loss_fct(highway_logits.view(-1),			labels.view(-1))
																												else:
																																			_lowerCAmelCase     :							Dict    =    CrossEntropyLoss()
																																			_lowerCAmelCase     :							Optional[Any]    =    loss_fct(highway_logits.view(-1,			self.num_labels),			labels.view(-1))
																												highway_losses.append(__a)

																					if train_highway:
																												_lowerCAmelCase     :							int    =    (sum(highway_losses[:-1]),) + outputs
																												# exclude the final highway, of course
																					else:
																												_lowerCAmelCase     :							Any    =    (loss,) + outputs
														if not self.training:
																					_lowerCAmelCase     :							Optional[Any]    =    outputs + ((original_entropy, highway_entropy), exit_layer)
																					if output_layer >= 0:
																												_lowerCAmelCase     :							Optional[Any]    =    (
																												    (outputs[0],) + (highway_logits_all[output_layer],) + outputs[2:]
																												)  # use the highway of the last layer

														return outputs  # (loss), logits, (hidden_states), (attentions), entropy
