





'''simple docstring'''






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


A :			Dict							   =      logging.get_logger(__name__)

A :			Optional[int]							   =      {
    'facebook/blenderbot-3B': 'https://huggingface.co/facebook/blenderbot-3B/resolve/main/config.json',
    # See all Blenderbot models at https://huggingface.co/models?filter=blenderbot
}







class       lowerCamelCase       (    __UpperCAmelCase	):
					_SCREAMING_SNAKE_CASE				 =     "blenderbot"
					_SCREAMING_SNAKE_CASE				 =     ["past_key_values"]
					_SCREAMING_SNAKE_CASE				 =     {"num_attention_heads": "encoder_attention_heads", "hidden_size": "d_model"}


					def __init__(       self						:	Tuple							,  __snake_case						:	Union[str, Any]=80_08							,  __snake_case						:	str=1_28							,  __snake_case						:	Dict=2							,  __snake_case						:	Dict=1_02_40							,  __snake_case						:	Dict=32							,  __snake_case						:	List[Any]=24							,  __snake_case						:	List[str]=1_02_40							,  __snake_case						:	Optional[int]=32							,  __snake_case						:	str=0.0							,  __snake_case						:	str=0.0							,  __snake_case						:	List[Any]=True							,  __snake_case						:	Union[str, Any]=True							,  __snake_case						:	List[str]="gelu"							,  __snake_case						:	Any=25_60							,  __snake_case						:	Optional[Any]=0.1							,  __snake_case						:	str=0.0							,  __snake_case						:	List[Any]=0.0							,  __snake_case						:	Union[str, Any]=0.02							,  __snake_case						:	Optional[int]=1							,  __snake_case						:	Any=False							,  __snake_case						:	List[Any]=0							,  __snake_case						:	Dict=1							,  __snake_case						:	List[str]=2							,  __snake_case						:	Dict=3							,  __snake_case						:	str=2							,  **__snake_case						:	List[str]							,  ):





											'''simple docstring'''





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

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







class       lowerCamelCase       (    __UpperCAmelCase	):


					@property
					def  SCREAMING_SNAKE_CASE_		(       self						:	Tuple							):





											'''simple docstring'''





											if self.task in ["default", "seq2seq-lm"]:
																	_snake_case: List[str]	       =		OrderedDict(
																	    [
																	        ('input_ids', {0: 'batch', 1: 'encoder_sequence'}),
																	        ('attention_mask', {0: 'batch', 1: 'encoder_sequence'}),
																	    ]							)
																	if self.use_past:
																							_snake_case: int	       =		{0: 'batch'}
																							_snake_case: List[Any]	       =		{0: 'batch', 1: 'past_decoder_sequence + sequence'}
																	else:
																							_snake_case: str	       =		{0: 'batch', 1: 'decoder_sequence'}
																							_snake_case: Tuple	       =		{0: 'batch', 1: 'decoder_sequence'}
																	if self.use_past:
																							self.fill_with_past_key_values_(lowercase__							,  direction='inputs'							)
											elif self.task == "causal-lm":
																	_snake_case: List[Any]	       =		OrderedDict(
																	    [
																	        ('input_ids', {0: 'batch', 1: 'encoder_sequence'}),
																	        ('attention_mask', {0: 'batch', 1: 'encoder_sequence'}),
																	    ]							)
																	if self.use_past:
																							_snake_case       ,    _snake_case: int	       =		self.num_layers
																							for i in range(lowercase__							):
																													_snake_case: Optional[Any]	       =		{0: 'batch', 2: 'past_sequence + sequence'}
																													_snake_case: Dict	       =		{0: 'batch', 2: 'past_sequence + sequence'}
											else:
																	_snake_case: 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
					# Copied from transformers.models.bart.configuration_bart.BartOnnxConfig.outputs
					def  SCREAMING_SNAKE_CASE_		(       self						:	Dict							):





											'''simple docstring'''





											if self.task in ["default", "seq2seq-lm"]:
																	_snake_case: Optional[Any]	       =		super().outputs
											else:
																	_snake_case: Optional[int]	       =		super(lowercase__							,  self							).outputs
																	if self.use_past:
																							_snake_case       ,    _snake_case: Optional[Any]	       =		self.num_layers
																							for i in range(lowercase__							):
																													_snake_case: int	       =		{0: 'batch', 2: 'past_sequence + sequence'}
																													_snake_case: Tuple	       =		{0: 'batch', 2: 'past_sequence + sequence'}
											return common_outputs


					def  SCREAMING_SNAKE_CASE_		(       self						:	str							,  __snake_case						:	PreTrainedTokenizer							,  __snake_case						:	int = -1							,  __snake_case						:	int = -1							,  __snake_case						:	bool = False							,  __snake_case						:	Optional[TensorType] = None							,  ):





											'''simple docstring'''





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

											if self.use_past:
																	if not is_torch_available():
																							raise ValueError('Cannot generate dummy past_keys inputs without PyTorch installed.'							)
																	else:
																							import torch
																	_snake_case       ,    _snake_case: Tuple	       =		common_inputs['input_ids'].shape
																	_snake_case: Dict	       =		common_inputs['decoder_input_ids'].shape[1]
																	_snake_case       ,    _snake_case: int	       =		self.num_attention_heads
																	_snake_case: Optional[int]	       =		(
																	    batch,
																	    num_encoder_attention_heads,
																	    encoder_seq_length,
																	    self._config.hidden_size // num_encoder_attention_heads,
																	)
																	_snake_case: Optional[Any]	       =		decoder_seq_length
																	_snake_case: Dict	       =		(
																	    batch,
																	    num_decoder_attention_heads,
																	    decoder_past_length,
																	    self._config.hidden_size // num_decoder_attention_heads,
																	)
																	_snake_case: Tuple	       =		torch.cat(
																	    [common_inputs['decoder_attention_mask'], torch.ones(lowercase__							,  lowercase__							)]							,  dim=1							)
																	_snake_case: str	       =		[]
																	_snake_case       ,    _snake_case: List[Any]	       =		self.num_layers

																	for _ in range(lowercase__							):
																							common_inputs["past_key_values"].append(
																							    (
																							        torch.zeros(lowercase__							),
																							        torch.zeros(lowercase__							),
																							        torch.zeros(lowercase__							),
																							        torch.zeros(lowercase__							),
																							    )							)
											return common_inputs


					def  SCREAMING_SNAKE_CASE_		(       self						:	Union[str, Any]							,  __snake_case						:	PreTrainedTokenizer							,  __snake_case						:	int = -1							,  __snake_case						:	int = -1							,  __snake_case						:	bool = False							,  __snake_case						:	Optional[TensorType] = None							,  ):





											'''simple docstring'''





											_snake_case: int	       =		self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
											    lowercase__							,  lowercase__							,  lowercase__							,  lowercase__							,  lowercase__							)

											if self.use_past:
																	if not is_torch_available():
																							raise ValueError('Cannot generate dummy past_keys inputs without PyTorch installed.'							)
																	else:
																							import torch
																	_snake_case       ,    _snake_case: Optional[Any]	       =		common_inputs['input_ids'].shape
																	_snake_case: List[str]	       =		seqlen
																	_snake_case       ,    _snake_case: List[str]	       =		self.num_layers
																	_snake_case       ,    _snake_case: Union[str, Any]	       =		self.num_attention_heads
																	_snake_case: str	       =		(
																	    batch,
																	    num_encoder_attention_heads,
																	    past_key_values_length,
																	    self._config.hidden_size // num_encoder_attention_heads,
																	)
																	_snake_case: Union[str, Any]	       =		common_inputs['attention_mask'].dtype
																	_snake_case: Any	       =		torch.cat(
																	    [common_inputs['attention_mask'], torch.ones(lowercase__							,  lowercase__							,  dtype=lowercase__							)]							,  dim=1							)
																	_snake_case: Any	       =		[
																	    (torch.zeros(lowercase__							), torch.zeros(lowercase__							)) for _ in range(lowercase__							)
																	]
											return common_inputs


					def  SCREAMING_SNAKE_CASE_		(       self						:	str							,  __snake_case						:	PreTrainedTokenizer							,  __snake_case						:	int = -1							,  __snake_case						:	int = -1							,  __snake_case						:	bool = False							,  __snake_case						:	Optional[TensorType] = None							,  ):





											'''simple docstring'''





											_snake_case: Optional[Any]	       =		compute_effective_axis_dimension(
											    lowercase__							,  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
											_snake_case: List[Any]	       =		tokenizer.num_special_tokens_to_add(lowercase__							)
											_snake_case: List[Any]	       =		compute_effective_axis_dimension(
											    lowercase__							,  fixed_dimension=OnnxConfig.default_fixed_sequence							,  num_token_to_add=lowercase__							)

											# Generate dummy inputs according to compute batch and sequence
											_snake_case: Dict	       =		[' '.join([tokenizer.unk_token]							) * seq_length] * batch_size
											_snake_case: int	       =		dict(tokenizer(lowercase__							,  return_tensors=lowercase__							)							)
											return common_inputs


					def  SCREAMING_SNAKE_CASE_		(       self						:	int							,  __snake_case						:	PreTrainedTokenizer							,  __snake_case						:	int = -1							,  __snake_case						:	int = -1							,  __snake_case						:	bool = False							,  __snake_case						:	Optional[TensorType] = None							,  ):





											'''simple docstring'''





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

											elif self.task == "causal-lm":
																	_snake_case: List[str]	       =		self._generate_dummy_inputs_for_causal_lm(
																	    lowercase__							,  batch_size=lowercase__							,  seq_length=lowercase__							,  is_pair=lowercase__							,  framework=lowercase__							)
											else:
																	_snake_case: Optional[Any]	       =		self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
																	    lowercase__							,  batch_size=lowercase__							,  seq_length=lowercase__							,  is_pair=lowercase__							,  framework=lowercase__							)

											return common_inputs


					def  SCREAMING_SNAKE_CASE_		(       self						:	List[Any]							,  __snake_case						:	List[str]							,  __snake_case						:	Dict							,  __snake_case						:	int							,  __snake_case						:	List[str]							):





											'''simple docstring'''





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




					def  SCREAMING_SNAKE_CASE_		(       self						:	List[Any]							,  __snake_case						:	Mapping[str, Mapping[int, str]]							,  __snake_case						:	str							):





											'''simple docstring'''





											if direction not in ["inputs", "outputs"]:
																	raise ValueError(f'''direction must either be \"inputs\" or \"outputs\", but {direction} was given'''							)

											_snake_case: Optional[Any]	       =		'past_key_values' if direction == 'inputs' else 'present'
											_snake_case       ,    _snake_case: List[str]	       =		self.num_layers

											_snake_case: Dict	       =		'past_encoder_sequence'
											_snake_case: Optional[Any]	       =		'past_decoder_sequence' if direction == 'inputs' else 'past_decoder_sequence + sequence'

											for i in range(lowercase__							):
																	_snake_case: Dict	       =		{0: 'batch', 2: decoder_sequence}
																	_snake_case: Optional[Any]	       =		{0: 'batch', 2: decoder_sequence}
																	_snake_case: List[str]	       =		{0: 'batch', 2: encoder_sequence}
																	_snake_case: Any	       =		{0: 'batch', 2: encoder_sequence}













'''simple docstring'''







import sys
from collections.abc import Mapping

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter






class        a__     (      TensorFormatter[Mapping, np.ndarray, Mapping]    ):
     '''simple docstring'''







     def __init__(self     :						Tuple					,					lowercase__     :						Dict=None					,					**lowercase__     :						List[Any]       )	->							str:
           super().__init__(features=lowercase__       )
           __A		 =   np_array_kwargs






     def      _UpperCamelCase (self     :						int					,					lowercase__     :						Tuple       )	->							List[Any]:
           if isinstance(lowercase__					,					lowercase__       ):
                 if column and all(
                     isinstance(lowercase__					,					np.ndarray       ) and x.shape == column[0].shape and x.dtype == column[0].dtype for x in column       ):
                       return np.stack(lowercase__       )
                 else:
                       # don't use np.array(column, dtype=object)
                       # since it fails in certain cases
                       # see https://stackoverflow.com/q/51005699
                       __A		 =   np.empty(len(lowercase__       )					,					dtype=lowercase__       )
                       __A		 =   column
                       return out
           return column






     def      _UpperCamelCase (self     :						Optional[Any]					,					lowercase__     :						Optional[int]       )	->							Tuple:
           if isinstance(lowercase__					,					(str, bytes, type(lowercase__       ))       ):
                 return value
           elif isinstance(lowercase__					,					(np.character, np.ndarray)       ) and np.issubdtype(value.dtype					,					np.character       ):
                 return value
           elif isinstance(lowercase__					,					np.number       ):
                 return value

           __A		 =   {}

           if isinstance(lowercase__					,					np.ndarray       ) and np.issubdtype(value.dtype					,					np.integer       ):
                 __A		 =   {"dtype": np.intaa}
           elif isinstance(lowercase__					,					np.ndarray       ) and np.issubdtype(value.dtype					,					np.floating       ):
                 __A		 =   {"dtype": np.floataa}
           elif config.PIL_AVAILABLE and "PIL" in sys.modules:
                 import PIL.Image

                 if isinstance(lowercase__					,					PIL.Image.Image       ):
                       return np.asarray(lowercase__					,					**self.np_array_kwargs       )

           return np.asarray(lowercase__					,					**{**default_dtype, **self.np_array_kwargs}       )






     def      _UpperCamelCase (self     :						List[Any]					,					lowercase__     :						str       )	->							Optional[Any]:
           # support for torch, tf, jax etc.
           if config.TORCH_AVAILABLE and "torch" in sys.modules:
                 import torch

                 if isinstance(lowercase__					,					torch.Tensor       ):
                       return self._tensorize(data_struct.detach().cpu().numpy()[()]       )
           if hasattr(lowercase__					,					"__array__"       ) and not isinstance(lowercase__					,					(np.ndarray, np.character, np.number)       ):
                 __A		 =   data_struct.__array__()
           # support for nested types like struct of list of struct
           if isinstance(lowercase__					,					np.ndarray       ):
                 if data_struct.dtype == object:
                       return self._consolidate([self.recursive_tensorize(lowercase__       ) for substruct in data_struct]       )
           if isinstance(lowercase__					,					(list, tuple)       ):
                 return self._consolidate([self.recursive_tensorize(lowercase__       ) for substruct in data_struct]       )
           return self._tensorize(lowercase__       )






     def      _UpperCamelCase (self     :						Optional[Any]					,					lowercase__     :						dict       )	->							Dict:
           return map_nested(self._recursive_tensorize					,					lowercase__					,					map_list=lowercase__       )






     def      _UpperCamelCase (self     :						Optional[int]					,					lowercase__     :						pa.Table       )	->							Mapping:
           __A		 =   self.numpy_arrow_extractor().extract_row(lowercase__       )
           __A		 =   self.python_features_decoder.decode_row(lowercase__       )
           return self.recursive_tensorize(lowercase__       )






     def      _UpperCamelCase (self     :						List[Any]					,					lowercase__     :						pa.Table       )	->							np.ndarray:
           __A		 =   self.numpy_arrow_extractor().extract_column(lowercase__       )
           __A		 =   self.python_features_decoder.decode_column(lowercase__					,					pa_table.column_names[0]       )
           __A		 =   self.recursive_tensorize(lowercase__       )
           __A		 =   self._consolidate(lowercase__       )
           return column






     def      _UpperCamelCase (self     :						Optional[Any]					,					lowercase__     :						pa.Table       )	->							Mapping:
           __A		 =   self.numpy_arrow_extractor().extract_batch(lowercase__       )
           __A		 =   self.python_features_decoder.decode_batch(lowercase__       )
           __A		 =   self.recursive_tensorize(lowercase__       )
           for column_name in batch:
                 __A		 =   self._consolidate(batch[column_name]       )
           return batch


