from collections import OrderedDict
from typing import Any, Mapping, Optional

from ... import PreTrainedTokenizer
from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig, OnnxConfigWithPast, OnnxSeqaSeqConfigWithPast
from ...onnx.utils import compute_effective_axis_dimension
from ...utils import TensorType, is_torch_available, logging


_SCREAMING_SNAKE_CASE     =     logging.get_logger(__name__)

_SCREAMING_SNAKE_CASE     =     {
    """google/bigbird-pegasus-large-arxiv""": (
        """https://huggingface.co/google/bigbird-pegasus-large-arxiv/resolve/main/config.json"""
    ),
    """google/bigbird-pegasus-large-pubmed""": (
        """https://huggingface.co/google/bigbird-pegasus-large-pubmed/resolve/main/config.json"""
    ),
    """google/bigbird-pegasus-large-bigpatent""": (
        """https://huggingface.co/google/bigbird-pegasus-large-bigpatent/resolve/main/config.json"""
    ),
    # See all BigBirdPegasus models at https://huggingface.co/models?filter=bigbird_pegasus
}







class        UpperCAmelCase_					(				lowercase__     ):
    UpperCAmelCase_      :							Any    			=				'bigbird_pegasus'
    UpperCAmelCase_      :							Optional[int]    			=				['past_key_values']
    UpperCAmelCase_      :							Optional[int]    			=				{
        'num_attention_heads': 'encoder_attention_heads',
        'hidden_size': 'd_model',
        'attention_probs_dropout_prob': 'attention_dropout',
    }




    def __init__(       self	: Dict   ,   lowerCAmelCase	: List[Any]=96103   ,   lowerCAmelCase	: List[str]=4096   ,   lowerCAmelCase	: Union[str, Any]=16   ,   lowerCAmelCase	: List[Any]=4096   ,   lowerCAmelCase	: List[Any]=16   ,   lowerCAmelCase	: Tuple=16   ,   lowerCAmelCase	: Dict=4096   ,   lowerCAmelCase	: Union[str, Any]=16   ,   lowerCAmelCase	: Union[str, Any]=0.0   ,   lowerCAmelCase	: List[Any]=0.0   ,   lowerCAmelCase	: List[Any]=True   ,   lowerCAmelCase	: List[Any]=True   ,   lowerCAmelCase	: Any="gelu_new"   ,   lowerCAmelCase	: Optional[int]=1024   ,   lowerCAmelCase	: str=0.1   ,   lowerCAmelCase	: Optional[int]=0.0   ,   lowerCAmelCase	: Optional[Any]=0.0   ,   lowerCAmelCase	: Any=0.02   ,   lowerCAmelCase	: str=2   ,   lowerCAmelCase	: Optional[Any]=0.0   ,   lowerCAmelCase	: Optional[int]=True   ,   lowerCAmelCase	: Optional[Any]=0   ,   lowerCAmelCase	: str=2   ,   lowerCAmelCase	: Tuple=1   ,   lowerCAmelCase	: Dict="block_sparse"   ,   lowerCAmelCase	: Optional[int]=64   ,   lowerCAmelCase	: List[str]=3   ,   lowerCAmelCase	: Tuple=False   ,   **lowerCAmelCase	: Any   ,   )						->			int:
      a__  		=  vocab_size
      a__  		=  max_position_embeddings
      a__  		=  d_model
      a__  		=  encoder_ffn_dim
      a__  		=  encoder_layers
      a__  		=  encoder_attention_heads
      a__  		=  decoder_ffn_dim
      a__  		=  decoder_layers
      a__  		=  decoder_attention_heads
      a__  		=  dropout
      a__  		=  attention_dropout
      a__  		=  activation_dropout
      a__  		=  activation_function
      a__  		=  init_std
      a__  		=  encoder_layerdrop
      a__  		=  decoder_layerdrop
      a__  		=  classifier_dropout
      a__  		=  use_cache
      a__  		=  encoder_layers
      a__  		=  scale_embedding  # scale factor will be sqrt(d_model) if True

      # extra config
      a__  		=  attention_type
      a__  		=  block_size
      a__  		=  num_random_blocks
      a__  		=  use_bias

      super().__init__(
          pad_token_id=SCREAMING_SNAKE_CASE__   ,   bos_token_id=SCREAMING_SNAKE_CASE__   ,   eos_token_id=SCREAMING_SNAKE_CASE__   ,   is_encoder_decoder=SCREAMING_SNAKE_CASE__   ,   decoder_start_token_id=SCREAMING_SNAKE_CASE__   ,   **SCREAMING_SNAKE_CASE__   ,   )







class        UpperCAmelCase_					(				lowercase__     ):




    @property
    def     __lowerCamelCase			(       self	: int )						->			Optional[int]:

      if self.task in ["default", "seq2seq-lm"]:
        a__  		=  OrderedDict(
            [
                ("input_ids", {0: "batch", 1: "encoder_sequence"}),
                ("attention_mask", {0: "batch", 1: "encoder_sequence"}),
            ] )

        if self.use_past:
          a__  		=  {0: """batch"""}
          a__  		=  {0: """batch""", 1: """past_decoder_sequence + sequence"""}
        else:
          a__  		=  {0: """batch""", 1: """decoder_sequence"""}
          a__  		=  {0: """batch""", 1: """decoder_sequence"""}

        if self.use_past:
          self.fill_with_past_key_values_(SCREAMING_SNAKE_CASE__   ,   direction="inputs" )
      elif self.task == "causal-lm":
        # TODO: figure this case out.
        a__  		=  OrderedDict(
            [
                ("input_ids", {0: "batch", 1: "encoder_sequence"}),
                ("attention_mask", {0: "batch", 1: "encoder_sequence"}),
            ] )
        if self.use_past:
          a__  		=  self.num_layers
          for i in range(SCREAMING_SNAKE_CASE__ ):
            a__  		=  {0: """batch""", 2: """past_sequence + sequence"""}
            a__  		=  {0: """batch""", 2: """past_sequence + sequence"""}
      else:
        a__  		=  OrderedDict(
            [
                ("input_ids", {0: "batch", 1: "encoder_sequence"}),
                ("attention_mask", {0: "batch", 1: "encoder_sequence"}),
                ("decoder_input_ids", {0: "batch", 1: "decoder_sequence"}),
                ("decoder_attention_mask", {0: "batch", 1: "decoder_sequence"}),
            ] )

      return common_inputs




    @property
    def     __lowerCamelCase			(       self	: Optional[int] )						->			str:

      if self.task in ["default", "seq2seq-lm"]:
        a__  		=  super().outputs
      else:
        a__  		=  super(SCREAMING_SNAKE_CASE__   ,   self ).outputs
        if self.use_past:
          a__  		=  self.num_layers
          for i in range(SCREAMING_SNAKE_CASE__ ):
            a__  		=  {0: """batch""", 2: """past_sequence + sequence"""}
            a__  		=  {0: """batch""", 2: """past_sequence + sequence"""}
      return common_outputs




    def     __lowerCamelCase			(       self	: List[Any]   ,   lowerCAmelCase	: List[str]   ,   lowerCAmelCase	: Dict = -1   ,   lowerCAmelCase	: int = -1   ,   lowerCAmelCase	: Union[str, Any] = False   ,   lowerCAmelCase	: Any = None   ,   )						->			Optional[int]:
      a__  		=  self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
          SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__ )

      # Generate decoder inputs
      a__  		=  seq_length if not self.use_past else 1
      a__  		=  self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
          SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__ )
      a__  		=  {f"decoder_{name}": tensor for name, tensor in decoder_inputs.items()}
      a__  		=  dict(**SCREAMING_SNAKE_CASE__   ,   **SCREAMING_SNAKE_CASE__ )

      if self.use_past:
        if not is_torch_available():
          raise ValueError("Cannot generate dummy past_keys inputs without PyTorch installed." )
        else:
          import torch
        a__  		=  common_inputs["""input_ids"""].shape
        a__  		=  common_inputs["""decoder_input_ids"""].shape[1]
        a__  		=  self.num_attention_heads
        a__  		=  (
            batch,
            num_encoder_attention_heads,
            encoder_seq_length,
            self._config.hidden_size // num_encoder_attention_heads,
        )
        a__  		=  decoder_seq_length + 3
        a__  		=  (
            batch,
            num_decoder_attention_heads,
            decoder_past_length,
            self._config.hidden_size // num_decoder_attention_heads,
        )

        a__  		=  torch.cat(
            [common_inputs["decoder_attention_mask"], torch.ones(SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__ )]   ,   dim=1 )

        a__  		=  []
        # If the number of encoder and decoder layers are present in the model configuration, both are considered
        a__  		=  self.num_layers
        a__  		=  min(SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__ )
        a__  		=  max(SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__ ) - min_num_layers
        a__  		=  """encoder""" if num_encoder_layers > num_decoder_layers else """decoder"""

        for _ in range(SCREAMING_SNAKE_CASE__ ):
          common_inputs["past_key_values"].append(
              (
                  torch.zeros(SCREAMING_SNAKE_CASE__ ),
                  torch.zeros(SCREAMING_SNAKE_CASE__ ),
                  torch.zeros(SCREAMING_SNAKE_CASE__ ),
                  torch.zeros(SCREAMING_SNAKE_CASE__ ),
              ) )
        # TODO: test this.
        a__  		=  encoder_shape if remaining_side_name == """encoder""" else decoder_shape
        for _ in range(SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__ ):
          common_inputs["past_key_values"].append((torch.zeros(SCREAMING_SNAKE_CASE__ ), torch.zeros(SCREAMING_SNAKE_CASE__ )) )
      return common_inputs




    def     __lowerCamelCase			(       self	: Union[str, Any]   ,   lowerCAmelCase	: List[Any]   ,   lowerCAmelCase	: Any = -1   ,   lowerCAmelCase	: str = -1   ,   lowerCAmelCase	: Optional[int] = False   ,   lowerCAmelCase	: Tuple = None   ,   )						->			Optional[Any]:
      a__  		=  self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
          SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__ )

      if self.use_past:
        if not is_torch_available():
          raise ValueError("Cannot generate dummy past_keys inputs without PyTorch installed." )
        else:
          import torch
        a__  		=  common_inputs["""input_ids"""].shape
        # Not using the same length for past_key_values
        a__  		=  seqlen + 2
        a__  		=  self.num_layers
        a__  		=  self.num_attention_heads
        a__  		=  (
            batch,
            num_encoder_attention_heads,
            past_key_values_length,
            self._config.hidden_size // num_encoder_attention_heads,
        )

        a__  		=  common_inputs["""attention_mask"""].dtype
        a__  		=  torch.cat(
            [common_inputs["attention_mask"], torch.ones(SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   dtype=SCREAMING_SNAKE_CASE__ )]   ,   dim=1 )
        a__  		=  [
            (torch.zeros(SCREAMING_SNAKE_CASE__ ), torch.zeros(SCREAMING_SNAKE_CASE__ )) for _ in range(SCREAMING_SNAKE_CASE__ )
        ]
      return common_inputs




    def     __lowerCamelCase			(       self	: Union[str, Any]   ,   lowerCAmelCase	: Optional[int]   ,   lowerCAmelCase	: Optional[Any] = -1   ,   lowerCAmelCase	: Tuple = -1   ,   lowerCAmelCase	: Optional[int] = False   ,   lowerCAmelCase	: Union[str, Any] = None   ,   )						->			str:
      a__  		=  compute_effective_axis_dimension(
          SCREAMING_SNAKE_CASE__   ,   fixed_dimension=OnnxConfig.default_fixed_batch   ,   num_token_to_add=0 )

      # If dynamic axis (-1) we forward with a fixed dimension of 8 tokens to avoid optimizations made by ONNX
      a__  		=  tokenizer.num_special_tokens_to_add(SCREAMING_SNAKE_CASE__ )
      a__  		=  compute_effective_axis_dimension(
          SCREAMING_SNAKE_CASE__   ,   fixed_dimension=OnnxConfig.default_fixed_sequence   ,   num_token_to_add=SCREAMING_SNAKE_CASE__ )

      # Generate dummy inputs according to compute batch and sequence
      a__  		=  [""" """.join([tokenizer.unk_token] ) * seq_length] * batch_size
      a__  		=  dict(tokenizer(SCREAMING_SNAKE_CASE__   ,   return_tensors=SCREAMING_SNAKE_CASE__ ) )
      return common_inputs




    def     __lowerCamelCase			(       self	: Optional[Any]   ,   lowerCAmelCase	: Tuple   ,   lowerCAmelCase	: str = -1   ,   lowerCAmelCase	: Any = -1   ,   lowerCAmelCase	: List[Any] = False   ,   lowerCAmelCase	: List[str] = None   ,   )						->			Union[str, Any]:

      if self.task in ["default", "seq2seq-lm"]:
        a__  		=  self._generate_dummy_inputs_for_default_and_seqaseq_lm(
            SCREAMING_SNAKE_CASE__   ,   batch_size=SCREAMING_SNAKE_CASE__   ,   seq_length=SCREAMING_SNAKE_CASE__   ,   is_pair=SCREAMING_SNAKE_CASE__   ,   framework=SCREAMING_SNAKE_CASE__ )

      elif self.task == "causal-lm":
        a__  		=  self._generate_dummy_inputs_for_causal_lm(
            SCREAMING_SNAKE_CASE__   ,   batch_size=SCREAMING_SNAKE_CASE__   ,   seq_length=SCREAMING_SNAKE_CASE__   ,   is_pair=SCREAMING_SNAKE_CASE__   ,   framework=SCREAMING_SNAKE_CASE__ )
      else:
        a__  		=  self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
            SCREAMING_SNAKE_CASE__   ,   batch_size=SCREAMING_SNAKE_CASE__   ,   seq_length=SCREAMING_SNAKE_CASE__   ,   is_pair=SCREAMING_SNAKE_CASE__   ,   framework=SCREAMING_SNAKE_CASE__ )

      return common_inputs




    def     __lowerCamelCase			(       self	: List[str]   ,   lowerCAmelCase	: Union[str, Any]   ,   lowerCAmelCase	: Union[str, Any]   ,   lowerCAmelCase	: Any   ,   lowerCAmelCase	: Tuple )						->			Optional[Any]:

      if self.task in ["default", "seq2seq-lm"]:
        a__  		=  super()._flatten_past_key_values_(SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__ )
      else:
        a__  		=  super(SCREAMING_SNAKE_CASE__   ,   self )._flatten_past_key_values_(
            SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__   ,   SCREAMING_SNAKE_CASE__ )






import tempfile
import unittest

import numpy as np
import torch

from diffusers import DiffusionPipeline
from diffusers.models.attention_processor import Attention, AttnAddedKVProcessor





class 		__SCREAMING_SNAKE_CASE      (				unittest.TestCase      ):






				"""simple docstring"""

				def 		__lowerCamelCase(				self						,	SCREAMING_SNAKE_CASE__ = False      ):






					"""simple docstring"""

					_snake_case					:							Any					     =   10

					if only_cross_attention:
						_snake_case					:							Tuple					     =   12
					else:
						# when only cross attention is not set, the cross attention dim must be the same as the query dim
						_snake_case					:							Any					     =   query_dim

					return {
					    "query_dim": query_dim,
					    "cross_attention_dim": cross_attention_dim,
					    "heads": 2,
					    "dim_head": 4,
					    "added_kv_proj_dim": 6,
					    "norm_num_groups": 1,
					    "only_cross_attention": only_cross_attention,
					    "processor": AttnAddedKVProcessor(),
					}







				def 		__lowerCamelCase(				self						,	SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__      ):






					"""simple docstring"""

					_snake_case					:							Any					     =   2

					_snake_case					:							List[str]					     =   torch.rand(SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__						,	3						,	2      )
					_snake_case					:							List[str]					     =   torch.rand(SCREAMING_SNAKE_CASE__						,	4						,	SCREAMING_SNAKE_CASE__      )
					_snake_case					:							Optional[int]					     =   None

					return {
					    "hidden_states": hidden_states,
					    "encoder_hidden_states": encoder_hidden_states,
					    "attention_mask": attention_mask,
					}







				def 		__lowerCamelCase(				self      ):






					"""simple docstring"""

					torch.manual_seed(0      )

					_snake_case					:							Optional[int]					     =   self.get_constructor_arguments(only_cross_attention=SCREAMING_SNAKE_CASE__      )
					_snake_case					:							List[str]					     =   Attention(**SCREAMING_SNAKE_CASE__      )

					self.assertTrue(attn.to_k is not None      )
					self.assertTrue(attn.to_v is not None      )

					_snake_case					:							Dict					     =   self.get_forward_arguments(
					    query_dim=constructor_args["""query_dim"""]						,	added_kv_proj_dim=constructor_args["""added_kv_proj_dim"""]      )

					_snake_case					:							Any					     =   attn(**SCREAMING_SNAKE_CASE__      )

					# only self attention

					torch.manual_seed(0      )

					_snake_case					:							Any					     =   self.get_constructor_arguments(only_cross_attention=SCREAMING_SNAKE_CASE__      )
					_snake_case					:							List[str]					     =   Attention(**SCREAMING_SNAKE_CASE__      )

					self.assertTrue(attn.to_k is None      )
					self.assertTrue(attn.to_v is None      )

					_snake_case					:							str					     =   self.get_forward_arguments(
					    query_dim=constructor_args["""query_dim"""]						,	added_kv_proj_dim=constructor_args["""added_kv_proj_dim"""]      )

					_snake_case					:							Any					     =   attn(**SCREAMING_SNAKE_CASE__      )

					self.assertTrue((only_cross_attn_out != self_and_cross_attn_out).all()      )





class 		__SCREAMING_SNAKE_CASE      (				unittest.TestCase      ):






				"""simple docstring"""

				def 		__lowerCamelCase(				self      ):






					"""simple docstring"""

					_snake_case					:							str					     =   DiffusionPipeline.from_pretrained("""hf-internal-testing/tiny-stable-diffusion-pipe"""						,	safety_checker=SCREAMING_SNAKE_CASE__      )

					_snake_case					:							Tuple					     =   pipe(
					    """foo"""						,	num_inference_steps=2						,	generator=torch.Generator("""cpu"""      ).manual_seed(0      )						,	output_type="""np"""						,	).images

					# the initial conversion succeeds
					_snake_case					:							List[str]					     =   DiffusionPipeline.from_pretrained(
					    """hf-internal-testing/tiny-stable-diffusion-pipe"""						,	device_map="""sequential"""						,	safety_checker=SCREAMING_SNAKE_CASE__      )

					_snake_case					:							Union[str, Any]					     =   pipe(
					    """foo"""						,	num_inference_steps=2						,	generator=torch.Generator("""cpu"""      ).manual_seed(0      )						,	output_type="""np"""						,	).images

					with tempfile.TemporaryDirectory() as tmpdir:
						# save the converted model
						pipe.save_pretrained(SCREAMING_SNAKE_CASE__      )

						# can also load the converted weights
						_snake_case					:							str					     =   DiffusionPipeline.from_pretrained(SCREAMING_SNAKE_CASE__						,	device_map="""sequential"""						,	safety_checker=SCREAMING_SNAKE_CASE__      )

					_snake_case					:							int					     =   pipe(
					    """foo"""						,	num_inference_steps=2						,	generator=torch.Generator("""cpu"""      ).manual_seed(0      )						,	output_type="""np"""						,	).images

					self.assertTrue(np.allclose(SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__						,	atol=1e-5      )      )
					self.assertTrue(np.allclose(SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__						,	atol=1e-5      )      )





