



'''simple docstring'''





import math

import torch
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from .attention_processor import Attention
from .embeddings import get_timestep_embedding
from .modeling_utils import ModelMixin






class __A    (						A     ,     A		):



 '''simple docstring'''







 @register_to_config
 def __init__(self				,     A = 128				,     A = 256				,     A = 2000.0				,     A = 768				,     A = 12				,     A = 12				,     A = 64				,     A = 2_048				,     A = 0.1				,     )					->  Optional[Any]:


      """simple docstring"""



      super().__init__()

      _a											=						nn.Sequential(
          nn.Linear(A				,     d_model * 4				,     bias=A )				,     nn.SiLU()				,     nn.Linear(d_model * 4				,     d_model * 4				,     bias=A )				,     nn.SiLU()				,     )

      _a											=						nn.Embedding(A				,     A )
      _a											=						False

      _a											=						nn.Linear(A				,     A				,     bias=A )

      _a											=						nn.Dropout(p=A )

      _a											=						nn.ModuleList()
      for lyr_num in range(A ):
           # FiLM conditional T5 decoder
           _a											=						DecoderLayer(d_model=A				,     d_kv=A				,     num_heads=A				,     d_ff=A				,     dropout_rate=A )
           self.decoders.append(A )

      _a											=						TaLayerNorm(A )

      _a											=						nn.Dropout(p=A )
      _a											=						nn.Linear(A				,     A				,     bias=A )



 def       a__						(self				,     A				,     A )					->  Any:


      """simple docstring"""



      _a											=						torch.mul(query_input.unsqueeze(-1 )				,     key_input.unsqueeze(-2 ) )
      return mask.unsqueeze(-3 )



 def       a__						(self				,     A				,     A				,     A )					->  Union[str, Any]:


      """simple docstring"""



      _a						,					_a						,					_a											=						decoder_input_tokens.shape
      assert decoder_noise_time.shape == (batch,)

      # decoder_noise_time is in [0, 1), so rescale to expected timing range.
      _a											=						get_timestep_embedding(
          decoder_noise_time * self.config.max_decoder_noise_time				,     embedding_dim=self.config.d_model				,     max_period=self.config.max_decoder_noise_time				,     ).to(dtype=self.dtype )

      _a											=						self.conditioning_emb(A ).unsqueeze(1 )

      assert conditioning_emb.shape == (batch, 1, self.config.d_model * 4)

      _a											=						decoder_input_tokens.shape[1]

      # If we want to use relative positions for audio context, we can just offset
      # this sequence by the length of encodings_and_masks.
      _a											=						torch.broadcast_to(
          torch.arange(A				,     device=decoder_input_tokens.device )				,     (batch, seq_length)				,     )

      _a											=						self.position_encoding(A )

      _a											=						self.continuous_inputs_projection(A )
      inputs += position_encodings
      _a											=						self.dropout(A )

      # decoder: No padding present.
      _a											=						torch.ones(
          decoder_input_tokens.shape[:2]				,     device=decoder_input_tokens.device				,     dtype=inputs.dtype )

      # Translate encoding masks to encoder-decoder masks.
      _a											=						[(x, self.encoder_decoder_mask(A				,     A )) for x, y in encodings_and_masks]

      # cross attend style: concat encodings
      _a											=						torch.cat([x[0] for x in encodings_and_encdec_masks]				,     dim=1 )
      _a											=						torch.cat([x[1] for x in encodings_and_encdec_masks]				,     dim=-1 )

      for lyr in self.decoders:
           _a											=						lyr(
               A				,     conditioning_emb=A				,     encoder_hidden_states=A				,     encoder_attention_mask=A				,     )[0]

      _a											=						self.decoder_norm(A )
      _a											=						self.post_dropout(A )

      _a											=						self.spec_out(A )
      return spec_out






class __A    (						nn.Module		):



 '''simple docstring'''







 def __init__(self				,     A				,     A				,     A				,     A				,     A				,     A=1E-6 )					->  str:


      """simple docstring"""



      super().__init__()
      _a											=						nn.ModuleList()

      # cond self attention: layer 0
      self.layer.append(
          TaLayerSelfAttentionCond(d_model=A				,     d_kv=A				,     num_heads=A				,     dropout_rate=A ) )

      # cross attention: layer 1
      self.layer.append(
          TaLayerCrossAttention(
              d_model=A				,     d_kv=A				,     num_heads=A				,     dropout_rate=A				,     layer_norm_epsilon=A				,     ) )

      # Film Cond MLP + dropout: last layer
      self.layer.append(
          TaLayerFFCond(d_model=A				,     d_ff=A				,     dropout_rate=A				,     layer_norm_epsilon=A ) )



 def       a__						(self				,     A				,     A=None				,     A=None				,     A=None				,     A=None				,     A=None				,     )					->  Tuple:


      """simple docstring"""



      _a											=						self.layer[0](
          A				,     conditioning_emb=A				,     attention_mask=A				,     )

      if encoder_hidden_states is not None:
           _a											=						torch.where(encoder_attention_mask > 0				,     0				,     -1E10 ).to(
               encoder_hidden_states.dtype )

           _a											=						self.layer[1](
               A				,     key_value_states=A				,     attention_mask=A				,     )

      # Apply Film Conditional Feed Forward layer
      _a											=						self.layer[-1](A				,     A )

      return (hidden_states,)






class __A    (						nn.Module		):



 '''simple docstring'''







 def __init__(self				,     A				,     A				,     A				,     A )					->  List[Any]:


      """simple docstring"""



      super().__init__()
      _a											=						TaLayerNorm(A )
      _a											=						TaFiLMLayer(in_features=d_model * 4				,     out_features=A )
      _a											=						Attention(query_dim=A				,     heads=A				,     dim_head=A				,     out_bias=A				,     scale_qk=A )
      _a											=						nn.Dropout(A )



 def       a__						(self				,     A				,     A=None				,     A=None				,     )					->  Union[str, Any]:


      """simple docstring"""



      _a											=						self.layer_norm(A )

      if conditioning_emb is not None:
           _a											=						self.FiLMLayer(A				,     A )

      # Self-attention block
      _a											=						self.attention(A )

      _a											=						hidden_states + self.dropout(A )

      return hidden_states






class __A    (						nn.Module		):



 '''simple docstring'''







 def __init__(self				,     A				,     A				,     A				,     A				,     A )					->  Optional[int]:


      """simple docstring"""



      super().__init__()
      _a											=						Attention(query_dim=A				,     heads=A				,     dim_head=A				,     out_bias=A				,     scale_qk=A )
      _a											=						TaLayerNorm(A				,     eps=A )
      _a											=						nn.Dropout(A )



 def       a__						(self				,     A				,     A=None				,     A=None				,     )					->  Optional[Any]:


      """simple docstring"""



      _a											=						self.layer_norm(A )
      _a											=						self.attention(
          A				,     encoder_hidden_states=A				,     attention_mask=attention_mask.squeeze(1 )				,     )
      _a											=						hidden_states + self.dropout(A )
      return layer_output






class __A    (						nn.Module		):



 '''simple docstring'''







 def __init__(self				,     A				,     A				,     A				,     A )					->  List[str]:


      """simple docstring"""



      super().__init__()
      _a											=						TaDenseGatedActDense(d_model=A				,     d_ff=A				,     dropout_rate=A )
      _a											=						TaFiLMLayer(in_features=d_model * 4				,     out_features=A )
      _a											=						TaLayerNorm(A				,     eps=A )
      _a											=						nn.Dropout(A )



 def       a__						(self				,     A				,     A=None )					->  Union[str, Any]:


      """simple docstring"""



      _a											=						self.layer_norm(A )
      if conditioning_emb is not None:
           _a											=						self.film(A				,     A )

      _a											=						self.DenseReluDense(A )
      _a											=						hidden_states + self.dropout(A )
      return hidden_states






class __A    (						nn.Module		):



 '''simple docstring'''







 def __init__(self				,     A				,     A				,     A )					->  int:


      """simple docstring"""



      super().__init__()
      _a											=						nn.Linear(A				,     A				,     bias=A )
      _a											=						nn.Linear(A				,     A				,     bias=A )
      _a											=						nn.Linear(A				,     A				,     bias=A )
      _a											=						nn.Dropout(A )
      _a											=						NewGELUActivation()



 def       a__						(self				,     A )					->  List[Any]:


      """simple docstring"""



      _a											=						self.act(self.wi_a(A ) )
      _a											=						self.wi_a(A )
      _a											=						hidden_gelu * hidden_linear
      _a											=						self.dropout(A )

      _a											=						self.wo(A )
      return hidden_states






class __A    (						nn.Module		):



 '''simple docstring'''







 def __init__(self				,     A				,     A=1E-6 )					->  int:


      """simple docstring"""



      super().__init__()
      _a											=						nn.Parameter(torch.ones(A ) )
      _a											=						eps



 def       a__						(self				,     A )					->  Optional[int]:


      """simple docstring"""



      _a											=						hidden_states.to(torch.floataa ).pow(2 ).mean(-1				,     keepdim=A )
      _a											=						hidden_states * torch.rsqrt(variance + self.variance_epsilon )

      # convert into half-precision if necessary
      if self.weight.dtype in [torch.floataa, torch.bfloataa]:
           _a											=						hidden_states.to(self.weight.dtype )

      return self.weight * hidden_states






class __A    (						nn.Module		):



 '''simple docstring'''







 def       a__						(self				,     A )					->  torch.Tensor:


      """simple docstring"""



      return 0.5 * input * (1.0 + torch.tanh(math.sqrt(2.0 / math.pi ) * (input + 0.044715 * torch.pow(A				,     3.0 )) ))






class __A    (						nn.Module		):



 '''simple docstring'''







 def __init__(self				,     A				,     A )					->  str:


      """simple docstring"""



      super().__init__()
      _a											=						nn.Linear(A				,     out_features * 2				,     bias=A )



 def       a__						(self				,     A				,     A )					->  Any:


      """simple docstring"""



      _a											=						self.scale_bias(A )
      _a						,					_a											=						torch.chunk(A				,     2				,     -1 )
      _a											=						x * (1 + scale) + shift
      return x








'''simple docstring'''





import random
import unittest

import numpy as np

from diffusers import (
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionImgaImgPipeline,
    PNDMScheduler,
)
from diffusers.utils import floats_tensor
from diffusers.utils.testing_utils import (
    is_onnx_available,
    load_image,
    nightly,
    require_onnxruntime,
    require_torch_gpu,
)

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
  import onnxruntime as ort






class __A    (						A     ,     unittest.TestCase		):



 '''simple docstring'''







 __lowerCamelCase					: Dict    = 'hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline'



 def       a__						(self				,     A=0 )					->  str:


      """simple docstring"""



      _a											=						floats_tensor((1, 3, 128, 128)				,     rng=random.Random(A ) )
      _a											=						np.random.RandomState(A )
      _a											=						{
          '''prompt''': '''A painting of a squirrel eating a burger''',
          '''image''': image,
          '''generator''': generator,
          '''num_inference_steps''': 3,
          '''strength''': 0.75,
          '''guidance_scale''': 7.5,
          '''output_type''': '''numpy''',
      }
      return inputs



 def       a__						(self )					->  str:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      pipe.set_progress_bar_config(disable=A )

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

      assert image.shape == (1, 128, 128, 3)
      _a											=						np.array([0.69643, 0.58484, 0.50314, 0.58760, 0.55368, 0.59643, 0.51529, 0.41217, 0.49087] )
      assert np.abs(image_slice - expected_slice ).max() < 1E-1



 def       a__						(self )					->  Tuple:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      _a											=						PNDMScheduler.from_config(pipe.scheduler.config				,     skip_prk_steps=A )
      pipe.set_progress_bar_config(disable=A )

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

      assert image.shape == (1, 128, 128, 3)
      _a											=						np.array([0.61737, 0.54642, 0.53183, 0.54465, 0.52742, 0.60525, 0.49969, 0.40655, 0.48154] )

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



 def       a__						(self )					->  int:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      _a											=						LMSDiscreteScheduler.from_config(pipe.scheduler.config )
      pipe.set_progress_bar_config(disable=A )

      # warmup pass to apply optimizations
      _a											=						pipe(**self.get_dummy_inputs() )

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

      assert image.shape == (1, 128, 128, 3)
      _a											=						np.array([0.52761, 0.59977, 0.49033, 0.49619, 0.54282, 0.50311, 0.47600, 0.40918, 0.45203] )

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



 def       a__						(self )					->  Dict:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      _a											=						EulerDiscreteScheduler.from_config(pipe.scheduler.config )
      pipe.set_progress_bar_config(disable=A )

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

      assert image.shape == (1, 128, 128, 3)
      _a											=						np.array([0.52911, 0.60004, 0.49229, 0.49805, 0.54502, 0.50680, 0.47777, 0.41028, 0.45304] )

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



 def       a__						(self )					->  Tuple:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      _a											=						EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config )
      pipe.set_progress_bar_config(disable=A )

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

      assert image.shape == (1, 128, 128, 3)
      _a											=						np.array([0.52911, 0.60004, 0.49229, 0.49805, 0.54502, 0.50680, 0.47777, 0.41028, 0.45304] )

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



 def       a__						(self )					->  List[str]:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      _a											=						DPMSolverMultistepScheduler.from_config(pipe.scheduler.config )
      pipe.set_progress_bar_config(disable=A )

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

      assert image.shape == (1, 128, 128, 3)
      _a											=						np.array([0.65331, 0.58277, 0.48204, 0.56059, 0.53665, 0.56235, 0.50969, 0.40009, 0.46552] )

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






@nightly
@require_onnxruntime
@require_torch_gpu
class __A    (						unittest.TestCase		):



 '''simple docstring'''







 @property
 def       a__						(self )					->  Tuple:


      """simple docstring"""



      return (
          "CUDAExecutionProvider",
          {
              "gpu_mem_limit": "15000000000",  # 15GB
              "arena_extend_strategy": "kSameAsRequested",
          },
      )



 @property
 def       a__						(self )					->  Any:


      """simple docstring"""



      _a											=						ort.SessionOptions()
      _a											=						False
      return options



 def       a__						(self )					->  str:


      """simple docstring"""



      _a											=						load_image(
          '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
          '''/img2img/sketch-mountains-input.jpg''' )
      _a											=						init_image.resize((768, 512) )
      # using the PNDM scheduler by default
      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(
          '''CompVis/stable-diffusion-v1-4'''				,     revision='''onnx'''				,     safety_checker=A				,     feature_extractor=A				,     provider=self.gpu_provider				,     sess_options=self.gpu_options				,     )
      pipe.set_progress_bar_config(disable=A )

      _a											=						'''A fantasy landscape, trending on artstation'''

      _a											=						np.random.RandomState(0 )
      _a											=						pipe(
          prompt=A				,     image=A				,     strength=0.75				,     guidance_scale=7.5				,     num_inference_steps=10				,     generator=A				,     output_type='''np'''				,     )
      _a											=						output.images
      _a											=						images[0, 255:258, 383:386, -1]

      assert images.shape == (1, 512, 768, 3)
      _a											=						np.array([0.4909, 0.5059, 0.5372, 0.4623, 0.4876, 0.5049, 0.4820, 0.4956, 0.5019] )
      # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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



 def       a__						(self )					->  Optional[int]:


      """simple docstring"""



      _a											=						load_image(
          '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
          '''/img2img/sketch-mountains-input.jpg''' )
      _a											=						init_image.resize((768, 512) )
      _a											=						LMSDiscreteScheduler.from_pretrained(
          '''runwayml/stable-diffusion-v1-5'''				,     subfolder='''scheduler'''				,     revision='''onnx''' )
      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(
          '''runwayml/stable-diffusion-v1-5'''				,     revision='''onnx'''				,     scheduler=A				,     safety_checker=A				,     feature_extractor=A				,     provider=self.gpu_provider				,     sess_options=self.gpu_options				,     )
      pipe.set_progress_bar_config(disable=A )

      _a											=						'''A fantasy landscape, trending on artstation'''

      _a											=						np.random.RandomState(0 )
      _a											=						pipe(
          prompt=A				,     image=A				,     strength=0.75				,     guidance_scale=7.5				,     num_inference_steps=20				,     generator=A				,     output_type='''np'''				,     )
      _a											=						output.images
      _a											=						images[0, 255:258, 383:386, -1]

      assert images.shape == (1, 512, 768, 3)
      _a											=						np.array([0.8043, 0.926, 0.9581, 0.8119, 0.8954, 0.913, 0.7209, 0.7463, 0.7431] )
      # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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



