





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

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


_lowerCAmelCase							:					List[Any]		    =		logging.get_logger(__name__)

_lowerCAmelCase							:					Any		    =		{
    '''facebook/blenderbot_small-90M''': '''https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/config.json''',
    # See all BlenderbotSmall models at https://huggingface.co/models?filter=blenderbot_small
}


class     __magic_name__     (							__A							):





     """simple docstring"""



     __UpperCamelCase    			=   'blenderbot-small'
     __UpperCamelCase    			=   ['past_key_values']
     __UpperCamelCase    			=   {'num_attention_heads': 'encoder_attention_heads', 'hidden_size': 'd_model'}



     def __init__(							self				:int	,					snake_case				:int=50_265	,					snake_case				:Any=512	,					snake_case				:Dict=8	,					snake_case				:Dict=2_048	,					snake_case				:Optional[int]=16	,					snake_case				:List[str]=8	,					snake_case				:Dict=2_048	,					snake_case				:Dict=16	,					snake_case				:int=0.0	,					snake_case				:Optional[Any]=0.0	,					snake_case				:List[Any]=True	,					snake_case				:Any=True	,					snake_case				:Any="gelu"	,					snake_case				:Dict=512	,					snake_case				:Tuple=0.1	,					snake_case				:Dict=0.0	,					snake_case				:str=0.0	,					snake_case				:List[str]=0.02	,					snake_case				:int=1	,					snake_case				:str=False	,					snake_case				:List[Any]=0	,					snake_case				:str=1	,					snake_case				:List[str]=2	,					snake_case				:Optional[int]=2	,					**snake_case				:Union[str, Any]	,					):





        '''simple docstring'''




        A_    : Any								=	vocab_size
        A_    : int								=	max_position_embeddings
        A_    : Optional[Any]								=	d_model
        A_    : Any								=	encoder_ffn_dim
        A_    : Any								=	encoder_layers
        A_    : int								=	encoder_attention_heads
        A_    : Tuple								=	decoder_ffn_dim
        A_    : Any								=	decoder_layers
        A_    : Union[str, Any]								=	decoder_attention_heads
        A_    : Optional[Any]								=	dropout
        A_    : str								=	attention_dropout
        A_    : Optional[int]								=	activation_dropout
        A_    : List[Any]								=	activation_function
        A_    : Tuple								=	init_std
        A_    : Any								=	encoder_layerdrop
        A_    : List[Any]								=	decoder_layerdrop
        A_    : Union[str, Any]								=	use_cache
        A_    : Optional[Any]								=	encoder_layers
        A_    : Union[str, Any]								=	scale_embedding  # scale factor will be sqrt(d_model) if True

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




class     __magic_name__     (							__A							):





     """simple docstring"""



     @property
     def   SCREAMING_SNAKE_CASE						(							self				:str     ):





        '''simple docstring'''




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

           if self.use_past:
              A_    : Optional[Any]								=	{0: "batch"}
              A_    : List[str]								=	{0: "batch", 1: "past_decoder_sequence + sequence"}
           else:
              A_    : Any								=	{0: "batch", 1: "decoder_sequence"}
              A_    : Optional[int]								=	{0: "batch", 1: "decoder_sequence"}

           if self.use_past:
              self.fill_with_past_key_values_(__UpperCAmelCase	,					direction="inputs"     )
        elif self.task == "causal-lm":
           # TODO: figure this case out.
           A_    : Union[str, Any]								=	OrderedDict(
               [
                   ("input_ids", {0: "batch", 1: "encoder_sequence"}),
                   ("attention_mask", {0: "batch", 1: "encoder_sequence"}),
               ]     )
           if self.use_past:
              A_					,  A_    : Optional[Any]								=	self.num_layers
              for i in range(__UpperCAmelCase     ):
                 A_    : Tuple								=	{0: "batch", 2: "past_sequence + sequence"}
                 A_    : Optional[int]								=	{0: "batch", 2: "past_sequence + sequence"}
        else:
           A_    : List[str]								=	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   SCREAMING_SNAKE_CASE						(							self				:Union[str, Any]     ):





        '''simple docstring'''




        if self.task in ["default", "seq2seq-lm"]:
           A_    : Dict								=	super().outputs
        else:
           A_    : str								=	super(__UpperCAmelCase	,					self     ).outputs
           if self.use_past:
              A_					,  A_    : Dict								=	self.num_layers
              for i in range(__UpperCAmelCase     ):
                 A_    : int								=	{0: "batch", 2: "past_sequence + sequence"}
                 A_    : Union[str, Any]								=	{0: "batch", 2: "past_sequence + sequence"}
        return common_outputs



     def   SCREAMING_SNAKE_CASE						(							self				:str	,					snake_case				:int	,					snake_case				:List[str] = -1	,					snake_case				:Any = -1	,					snake_case				:Any = False	,					snake_case				:Tuple = None	,					):





        '''simple docstring'''




        A_    : int								=	self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
            __UpperCAmelCase	,					__UpperCAmelCase	,					__UpperCAmelCase	,					__UpperCAmelCase	,					__UpperCAmelCase     )

        # Generate decoder inputs
        A_    : Union[str, Any]								=	seq_length if not self.use_past else 1
        A_    : Optional[Any]								=	self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
            __UpperCAmelCase	,					__UpperCAmelCase	,					__UpperCAmelCase	,					__UpperCAmelCase	,					__UpperCAmelCase     )
        A_    : Tuple								=	{f"decoder_{name}": tensor for name, tensor in decoder_inputs.items()}
        A_    : int								=	dict(**__UpperCAmelCase	,					**__UpperCAmelCase     )

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

           A_    : int								=	torch.cat(
               [common_inputs["decoder_attention_mask"], torch.ones(__UpperCAmelCase	,					__UpperCAmelCase     )]	,					dim=1     )

           A_    : Any								=	[]
           # If the number of encoder and decoder layers are present in the model configuration, both are considered
           A_					,  A_    : Tuple								=	self.num_layers
           A_    : str								=	min(__UpperCAmelCase	,					__UpperCAmelCase     )
           A_    : Dict								=	max(__UpperCAmelCase	,					__UpperCAmelCase     ) - min_num_layers
           A_    : Dict								=	"encoder" if num_encoder_layers > num_decoder_layers else "decoder"

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



     def   SCREAMING_SNAKE_CASE						(							self				:Dict	,					snake_case				:Union[str, Any]	,					snake_case				:Tuple = -1	,					snake_case				:Union[str, Any] = -1	,					snake_case				:str = False	,					snake_case				:int = None	,					):





        '''simple docstring'''




        A_    : int								=	self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
            __UpperCAmelCase	,					__UpperCAmelCase	,					__UpperCAmelCase	,					__UpperCAmelCase	,					__UpperCAmelCase     )

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

           A_    : Any								=	common_inputs["attention_mask"].dtype
           A_    : Union[str, Any]								=	torch.cat(
               [common_inputs["attention_mask"], torch.ones(__UpperCAmelCase	,					__UpperCAmelCase	,					dtype=__UpperCAmelCase     )]	,					dim=1     )
           A_    : Union[str, Any]								=	[
               (torch.zeros(__UpperCAmelCase     ), torch.zeros(__UpperCAmelCase     )) for _ in range(__UpperCAmelCase     )
           ]
        return common_inputs



     def   SCREAMING_SNAKE_CASE						(							self				:Tuple	,					snake_case				:Tuple	,					snake_case				:int = -1	,					snake_case				:List[str] = -1	,					snake_case				:str = False	,					snake_case				:int = None	,					):





        '''simple docstring'''




        A_    : List[str]								=	compute_effective_axis_dimension(
            __UpperCAmelCase	,					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_    : Union[str, Any]								=	tokenizer.num_special_tokens_to_add(__UpperCAmelCase     )
        A_    : List[Any]								=	compute_effective_axis_dimension(
            __UpperCAmelCase	,					fixed_dimension=OnnxConfig.default_fixed_sequence	,					num_token_to_add=__UpperCAmelCase     )

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



     def   SCREAMING_SNAKE_CASE						(							self				:str	,					snake_case				:Tuple	,					snake_case				:Union[str, Any] = -1	,					snake_case				:Dict = -1	,					snake_case				:Union[str, Any] = False	,					snake_case				:str = None	,					):





        '''simple docstring'''




        if self.task in ["default", "seq2seq-lm"]:
           A_    : List[str]								=	self._generate_dummy_inputs_for_default_and_seqaseq_lm(
               __UpperCAmelCase	,					batch_size=__UpperCAmelCase	,					seq_length=__UpperCAmelCase	,					is_pair=__UpperCAmelCase	,					framework=__UpperCAmelCase     )

        elif self.task == "causal-lm":
           A_    : List[str]								=	self._generate_dummy_inputs_for_causal_lm(
               __UpperCAmelCase	,					batch_size=__UpperCAmelCase	,					seq_length=__UpperCAmelCase	,					is_pair=__UpperCAmelCase	,					framework=__UpperCAmelCase     )
        else:
           A_    : Dict								=	self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
               __UpperCAmelCase	,					batch_size=__UpperCAmelCase	,					seq_length=__UpperCAmelCase	,					is_pair=__UpperCAmelCase	,					framework=__UpperCAmelCase     )

        return common_inputs





     def   SCREAMING_SNAKE_CASE						(							self				:Union[str, Any]	,					snake_case				:Any	,					snake_case				:str	,					snake_case				:str	,					snake_case				:Optional[int]     ):





        '''simple docstring'''




        if self.task in ["default", "seq2seq-lm"]:
           A_    : int								=	super()._flatten_past_key_values_(__UpperCAmelCase	,					__UpperCAmelCase	,					__UpperCAmelCase	,					__UpperCAmelCase     )
        else:
           A_    : Optional[Any]								=	super(__UpperCAmelCase	,					self     )._flatten_past_key_values_(
               __UpperCAmelCase	,					__UpperCAmelCase	,					__UpperCAmelCase	,					__UpperCAmelCase     )

"""simple docstring"""




import os
import socket
from contextlib import contextmanager

import torch

from ..commands.config.default import write_basic_config  # noqa: F401
from ..state import PartialState
from .dataclasses import DistributedType
from .imports import is_deepspeed_available, is_tpu_available
from .transformer_engine import convert_model
from .versions import is_torch_version


if is_deepspeed_available():
				from deepspeed import DeepSpeedEngine

if is_tpu_available(check_device=False):
				import torch_xla.core.xla_model as xm






def      UpperCAmelCase    (		UpperCamelCase__ ):






							"""simple docstring"""





							if is_torch_version('<'				,			'2.0.0' ) or not hasattr(UpperCamelCase__				,			'_dynamo' ):
														return False
							return isinstance(UpperCamelCase__				,			torch._dynamo.eval_frame.OptimizedModule )






def      UpperCAmelCase    (		UpperCamelCase__				,			UpperCamelCase__ = True ):






							"""simple docstring"""





							A__		   =							(torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

							A__		   =							is_compiled_module(UpperCamelCase__ )
							if is_compiled:
														A__		   =							model
														A__		   =							model._orig_mod

							if is_deepspeed_available():
														options += (DeepSpeedEngine,)

							while isinstance(UpperCamelCase__				,			UpperCamelCase__ ):
														A__		   =							model.module

							if not keep_fpaa_wrapper:
														A__		   =							getattr(UpperCamelCase__				,			'forward' )
														A__		   =							model.__dict__.pop('_original_forward'				,			UpperCamelCase__ )
														if original_forward is not None:
																					while hasattr(UpperCamelCase__				,			'__wrapped__' ):
																												A__		   =							forward.__wrapped__
																												if forward == original_forward:
																																			break
																					A__		   =							forward
														if getattr(UpperCamelCase__				,			'_converted_to_transformer_engine'				,			UpperCamelCase__ ):
																					convert_model(UpperCamelCase__				,			to_transformer_engine=UpperCamelCase__ )

							if is_compiled:
														A__		   =							model
														A__		   =							compiled_model

							return model






def      UpperCAmelCase    (		):






							"""simple docstring"""





							PartialState().wait_for_everyone()






def      UpperCAmelCase    (		UpperCamelCase__				,			UpperCamelCase__ ):






							"""simple docstring"""





							if PartialState().distributed_type == DistributedType.TPU:
														xm.save(UpperCamelCase__				,			UpperCamelCase__ )
							elif PartialState().local_process_index == 0:
														torch.save(UpperCamelCase__				,			UpperCamelCase__ )






@contextmanager
def      UpperCAmelCase    (		**UpperCamelCase__ ):






							"""simple docstring"""





							for key, value in kwargs.items():
														A__		   =							str(UpperCamelCase__ )

							yield

							for key in kwargs:
														if key.upper() in os.environ:
																					del os.environ[key.upper()]






def      UpperCAmelCase    (		UpperCamelCase__ ):






							"""simple docstring"""





							if not hasattr(UpperCamelCase__				,			'__qualname__' ) and not hasattr(UpperCamelCase__				,			'__name__' ):
														A__		   =							getattr(UpperCamelCase__				,			'__class__'				,			UpperCamelCase__ )
							if hasattr(UpperCamelCase__				,			'__qualname__' ):
														return obj.__qualname__
							if hasattr(UpperCamelCase__				,			'__name__' ):
														return obj.__name__
							return str(UpperCamelCase__ )






def      UpperCAmelCase    (		UpperCamelCase__				,			UpperCamelCase__ ):






							"""simple docstring"""





							for key, value in source.items():
														if isinstance(UpperCamelCase__				,			UpperCamelCase__ ):
																					A__		   =							destination.setdefault(UpperCamelCase__				,			{} )
																					merge_dicts(UpperCamelCase__				,			UpperCamelCase__ )
														else:
																					A__		   =							value

							return destination






def      UpperCAmelCase    (		UpperCamelCase__ = None ):






							"""simple docstring"""





							if port is None:
														A__		   =							29_500
							with socket.socket(socket.AF_INET				,			socket.SOCK_STREAM ) as s:
														return s.connect_ex(('localhost', port) ) == 0





