



from typing import Optional

import numpy as np
import torch
from torch import nn
from transformers import GPTaConfig, GPTaLMHeadModel
from transformers.modeling_utils import ModuleUtilsMixin

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin



class        lowerCAmelCase__					(  _lowerCamelCase ,				_lowerCamelCase ,				_lowerCamelCase       ):
				A_   :     Union[str, Any]								=     [r'h\.\d+\.attn\.bias', r'h\.\d+\.attn\.masked_bias']


				@register_to_config
				def __init__(							self      :  Optional[Any]						, __UpperCamelCase      :  int						, __UpperCamelCase      :  int						, __UpperCamelCase      :  Optional[int] = None						, __UpperCamelCase      :  int = 50_257						, __UpperCamelCase      :  int = 1_024						, __UpperCamelCase      :  int = 768						, __UpperCamelCase      :  int = 12						, __UpperCamelCase      :  int = 12						, __UpperCamelCase      :  Optional[int] = None						, __UpperCamelCase      :  str = "gelu_new"						, __UpperCamelCase      :  float = 0.1						, __UpperCamelCase      :  float = 0.1						, __UpperCamelCase      :  float = 0.1						, __UpperCamelCase      :  float = 1e-5						, __UpperCamelCase      :  float = 0.0_2						, __UpperCamelCase      :  bool = True						, __UpperCamelCase      :  bool = True						, __UpperCamelCase      :  bool = False						, __UpperCamelCase      :  bool = False						, )		->     int:
											super().__init__()

											A      						=    prefix_length

											if prefix_inner_dim != n_embd and prefix_hidden_dim is None:
																		raise ValueError(
																		    f'''`prefix_hidden_dim` cannot be `None` when `prefix_inner_dim`: {prefix_hidden_dim} and'''
																		    f''' `n_embd`: {n_embd} are not equal.'''       )

											A      						=    prefix_inner_dim
											A      						=    prefix_hidden_dim

											A      						=    (
											    nn.Linear(self.prefix_inner_dim						, self.prefix_hidden_dim       )
											    if self.prefix_hidden_dim is not None
											    else nn.Identity()
											)
											A      						=    (
											    nn.Linear(self.prefix_hidden_dim						, __UpperCamelCase       ) if self.prefix_hidden_dim is not None else nn.Identity()
											)

											A      						=    GPTaConfig(
											    vocab_size=__UpperCamelCase						, n_positions=__UpperCamelCase						, n_embd=__UpperCamelCase						, n_layer=__UpperCamelCase						, n_head=__UpperCamelCase						, n_inner=__UpperCamelCase						, activation_function=__UpperCamelCase						, resid_pdrop=__UpperCamelCase						, embd_pdrop=__UpperCamelCase						, attn_pdrop=__UpperCamelCase						, layer_norm_epsilon=__UpperCamelCase						, initializer_range=__UpperCamelCase						, scale_attn_weights=__UpperCamelCase						, use_cache=__UpperCamelCase						, scale_attn_by_inverse_layer_idx=__UpperCamelCase						, reorder_and_upcast_attn=__UpperCamelCase						, )
											A      						=    GPTaLMHeadModel(__UpperCamelCase       )


				def __UpperCamelCase						(							self      :  int						, __UpperCamelCase      :  torch.Tensor						, __UpperCamelCase      :  torch.Tensor						, __UpperCamelCase      :  Optional[torch.Tensor] = None						, __UpperCamelCase      :  Optional[torch.Tensor] = None						, )		->     Optional[Any]:
											A      						=    self.transformer.transformer.wte(__UpperCamelCase       )
											A      						=    self.encode_prefix(__UpperCamelCase       )
											A      						=    self.decode_prefix(__UpperCamelCase       )
											A      						=    torch.cat((prefix_embeds, embedding_text)						, dim=1       )

											if labels is not None:
																		A      						=    self.get_dummy_token(input_ids.shape[0]						, input_ids.device       )
																		A      						=    torch.cat((dummy_token, input_ids)						, dim=1       )
											A      						=    self.transformer(inputs_embeds=__UpperCamelCase						, labels=__UpperCamelCase						, attention_mask=__UpperCamelCase       )
											if self.prefix_hidden_dim is not None:
																		return out, hidden
											else:
																		return out


				def __UpperCamelCase						(							self      :  Dict						, __UpperCamelCase      :  int						, __UpperCamelCase      :  torch.device       )		->     torch.Tensor:
											return torch.zeros(__UpperCamelCase						, self.prefix_length						, dtype=torch.intaa						, device=__UpperCamelCase       )


				def __UpperCamelCase						(							self      :  Optional[int]						, __UpperCamelCase      :  Tuple       )		->     Optional[int]:
											return self.encode_prefix(__UpperCamelCase       )


				@torch.no_grad()
				def __UpperCamelCase						(							self      :  Union[str, Any]						, __UpperCamelCase      :  str						, __UpperCamelCase      :  Union[str, Any]						, __UpperCamelCase      :  Optional[Any]       )		->     Dict:
											A      						=    torch.split(__UpperCamelCase						, 1						, dim=0       )
											A      						=    []
											A      						=    []
											for feature in features:
																		A      						=    self.decode_prefix(feature.to(__UpperCamelCase       )       )  # back to the clip feature
																		# Only support beam search for now
																		A						,   A      						=    self.generate_beam(
																		    input_embeds=__UpperCamelCase						, device=__UpperCamelCase						, eos_token_id=__UpperCamelCase       )
																		generated_tokens.append(output_tokens[0]       )
																		generated_seq_lengths.append(seq_lengths[0]       )
											A      						=    torch.stack(__UpperCamelCase       )
											A      						=    torch.stack(__UpperCamelCase       )
											return generated_tokens, generated_seq_lengths


				@torch.no_grad()
				def __UpperCamelCase						(							self      :  Optional[int]						, __UpperCamelCase      :  Optional[int]=None						, __UpperCamelCase      :  Optional[Any]=None						, __UpperCamelCase      :  int=None						, __UpperCamelCase      :  int = 5						, __UpperCamelCase      :  int = 67						, __UpperCamelCase      :  float = 1.0						, __UpperCamelCase      :  Optional[int] = None						, )		->     Tuple:
											A      						=    eos_token_id
											A      						=    None
											A      						=    None
											A      						=    torch.ones(__UpperCamelCase						, device=__UpperCamelCase						, dtype=torch.int       )
											A      						=    torch.zeros(__UpperCamelCase						, device=__UpperCamelCase						, dtype=torch.bool       )

											if input_embeds is not None:
																		A      						=    input_embeds
											else:
																		A      						=    self.transformer.transformer.wte(__UpperCamelCase       )

											for i in range(__UpperCamelCase       ):
																		A      						=    self.transformer(inputs_embeds=__UpperCamelCase       )
																		A      						=    outputs.logits
																		A      						=    logits[:, -1, :] / (temperature if temperature > 0 else 1.0)
																		A      						=    logits.softmax(-1       ).log()

																		if scores is None:
																									A						,   A      						=    logits.topk(__UpperCamelCase						, -1       )
																									A      						=    generated.expand(__UpperCamelCase						, *generated.shape[1:]       )
																									A						,   A      						=    next_tokens.permute(1						, 0       ), scores.squeeze(0       )
																									if tokens is None:
																																A      						=    next_tokens
																									else:
																																A      						=    tokens.expand(__UpperCamelCase						, *tokens.shape[1:]       )
																																A      						=    torch.cat((tokens, next_tokens)						, dim=1       )
																		else:
																									A      						=    -float(np.inf       )
																									A      						=    0
																									A      						=    scores[:, None] + logits
																									seq_lengths[~is_stopped] += 1
																									A      						=    scores_sum / seq_lengths[:, None]
																									A						,   A      						=    scores_sum_average.view(-1       ).topk(__UpperCamelCase						, -1       )
																									A      						=    next_tokens // scores_sum.shape[1]
																									A      						=    seq_lengths[next_tokens_source]
																									A      						=    next_tokens % scores_sum.shape[1]
																									A      						=    next_tokens.unsqueeze(1       )
																									A      						=    tokens[next_tokens_source]
																									A      						=    torch.cat((tokens, next_tokens)						, dim=1       )
																									A      						=    generated[next_tokens_source]
																									A      						=    scores_sum_average * seq_lengths
																									A      						=    is_stopped[next_tokens_source]

																		A      						=    self.transformer.transformer.wte(next_tokens.squeeze()       ).view(generated.shape[0]						, 1						, -1       )
																		A      						=    torch.cat((generated, next_token_embed)						, dim=1       )
																		A      						=    is_stopped + next_tokens.eq(__UpperCamelCase       ).squeeze()
																		if is_stopped.all():
																									break

											A      						=    scores / seq_lengths
											A      						=    scores.argsort(descending=__UpperCamelCase       )
											# tokens tensors are already padded to max_seq_length
											A      						=    [tokens[i] for i in order]
											A      						=    torch.stack(__UpperCamelCase						, dim=0       )
											A      						=    torch.tensor([seq_lengths[i] for i in order]						, dtype=seq_lengths.dtype       )
											return output_texts, seq_lengths


import gc
import unittest

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

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

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


enable_full_determinism()



class        A__(a_,  a_,  unittest.TestCase							):
						"""simple docstring"""






						_A    :				Optional[Any]       					=			StableDiffusionSAGPipeline
						_A    :				Any       					=			TEXT_TO_IMAGE_PARAMS
						_A    :				Dict       					=			TEXT_TO_IMAGE_BATCH_PARAMS
						_A    :				Dict       					=			TEXT_TO_IMAGE_IMAGE_PARAMS
						_A    :				int       					=			TEXT_TO_IMAGE_IMAGE_PARAMS
						_A    :				Optional[int]       					=			False





						def 			UpperCamelCase__					(    self    )     -> List[Any]:
									torch.manual_seed(0    )
									a_						:						Tuple 							= 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  ,      )
									a_						:						int 							= DDIMScheduler(
									    beta_start=0.0_0_0_8_5  ,      beta_end=0.0_1_2  ,      beta_schedule="""scaled_linear"""  ,      clip_sample=_lowercase  ,      set_alpha_to_one=_lowercase  ,      )
									torch.manual_seed(0    )
									a_						:						Tuple 							= 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    )
									a_						:						str 							= 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=1_000  ,      )
									a_						:						Union[str, Any] 							= CLIPTextModel(_lowercase    )
									a_						:						int 							= CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""    )

									a_						:						Union[str, Any] 							= {
									    """unet""": unet,
									    """scheduler""": scheduler,
									    """vae""": vae,
									    """text_encoder""": text_encoder,
									    """tokenizer""": tokenizer,
									    """safety_checker""": None,
									    """feature_extractor""": None,
									}
									return components





						def 			UpperCamelCase__					(    self  ,      _lowercase  ,      _lowercase=0    )     -> Dict:
									if str(_lowercase    ).startswith("""mps"""    ):
												a_						:						Optional[int] 							= torch.manual_seed(_lowercase    )
									else:
												a_						:						str 							= torch.Generator(device=_lowercase    ).manual_seed(_lowercase    )
									a_						:						Tuple 							= {
									    """prompt""": """.""",
									    """generator""": generator,
									    """num_inference_steps""": 2,
									    """guidance_scale""": 1.0,
									    """sag_scale""": 1.0,
									    """output_type""": """numpy""",
									}
									return inputs





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







@slow
@require_torch_gpu
class        A__(unittest.TestCase							):
						"""simple docstring"""






						def 			UpperCamelCase__					(    self    )     -> Optional[int]:
									# clean up the VRAM after each test
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()





						def 			UpperCamelCase__					(    self    )     -> Optional[int]:
									a_						:						Tuple 							= StableDiffusionSAGPipeline.from_pretrained("""CompVis/stable-diffusion-v1-4"""    )
									a_						:						List[str] 							= sag_pipe.to(_lowercase    )
									sag_pipe.set_progress_bar_config(disable=_lowercase    )

									a_						:						Optional[int] 							= """."""
									a_						:						int 							= torch.manual_seed(0    )
									a_						:						Any 							= sag_pipe(
									    [prompt]  ,      generator=_lowercase  ,      guidance_scale=7.5  ,      sag_scale=1.0  ,      num_inference_steps=20  ,      output_type="""np"""    )

									a_						:						Any 							= output.images

									a_						:						Optional[int] 							= image[0, -3:, -3:, -1]

									assert image.shape == (1, 512, 512, 3)
									a_						:						str 							= np.array([0.1_5_6_8, 0.1_7_3_8, 0.1_6_9_5, 0.1_6_9_3, 0.1_5_0_7, 0.1_7_0_5, 0.1_5_4_7, 0.1_7_5_1, 0.1_9_4_9]    )

									assert np.abs(image_slice.flatten() - expected_slice    ).max() < 5e-2





						def 			UpperCamelCase__					(    self    )     -> List[Any]:
									a_						:						Optional[int] 							= StableDiffusionSAGPipeline.from_pretrained("""stabilityai/stable-diffusion-2-1-base"""    )
									a_						:						List[str] 							= sag_pipe.to(_lowercase    )
									sag_pipe.set_progress_bar_config(disable=_lowercase    )

									a_						:						int 							= """."""
									a_						:						Dict 							= torch.manual_seed(0    )
									a_						:						Union[str, Any] 							= sag_pipe(
									    [prompt]  ,      generator=_lowercase  ,      guidance_scale=7.5  ,      sag_scale=1.0  ,      num_inference_steps=20  ,      output_type="""np"""    )

									a_						:						Optional[Any] 							= output.images

									a_						:						int 							= image[0, -3:, -3:, -1]

									assert image.shape == (1, 512, 512, 3)
									a_						:						Dict 							= np.array([0.3_4_5_9, 0.2_8_7_6, 0.2_5_3_7, 0.3_0_0_2, 0.2_6_7_1, 0.2_1_6_0, 0.3_0_2_6, 0.2_2_6_2, 0.2_3_7_1]    )

									assert np.abs(image_slice.flatten() - expected_slice    ).max() < 5e-2





						def 			UpperCamelCase__					(    self    )     -> Any:
									a_						:						int 							= StableDiffusionSAGPipeline.from_pretrained("""stabilityai/stable-diffusion-2-1-base"""    )
									a_						:						Optional[Any] 							= sag_pipe.to(_lowercase    )
									sag_pipe.set_progress_bar_config(disable=_lowercase    )

									a_						:						List[Any] 							= """."""
									a_						:						str 							= torch.manual_seed(0    )
									a_						:						int 							= sag_pipe(
									    [prompt]  ,      width=768  ,      height=512  ,      generator=_lowercase  ,      guidance_scale=7.5  ,      sag_scale=1.0  ,      num_inference_steps=20  ,      output_type="""np"""  ,      )

									a_						:						Any 							= output.images

									assert image.shape == (1, 512, 768, 3)


