

"""simple docstring"""

import timeit
from typing import Callable, Optional

from ..configuration_utils import PretrainedConfig
from ..models.auto.modeling_auto import MODEL_MAPPING, MODEL_WITH_LM_HEAD_MAPPING
from ..utils import is_pyanvml_available, is_torch_available, logging
from .benchmark_utils import (
    Benchmark,
    Memory,
    MemorySummary,
    measure_peak_memory_cpu,
    start_memory_tracing,
    stop_memory_tracing,
)


if is_torch_available():
							import torch

							from .benchmark_args import PyTorchBenchmarkArguments


if is_pyanvml_available():
							import pyanvml.pyanvml as nvml


a__			:						int							=			logging.get_logger(__name__)


class       _snake_case  (    snake_case__							):
						__snake_case			      =  42
						__snake_case			      =  42
						__snake_case			      =  'PyTorch'







						@property
						def 							_UpperCamelCase     (self				)      ->    Optional[Any]:
											return torch.__version__







						def 							_UpperCamelCase     (self  ,				UpperCAmelCase_  ,				UpperCAmelCase_  ,				UpperCAmelCase_				)      ->    str:
											__lowercase        =      self._prepare_inference_func(lowercase_  ,				lowercase_  ,				lowercase_				)
											return self._measure_speed(_inference				)







						def 							_UpperCamelCase     (self  ,				UpperCAmelCase_  ,				UpperCAmelCase_  ,				UpperCAmelCase_				)      ->    int:
											__lowercase        =      self._prepare_inference_func(lowercase_  ,				lowercase_  ,				lowercase_				)
											return self._measure_memory(_inference				)







						def 							_UpperCamelCase     (self  ,				UpperCAmelCase_  ,				UpperCAmelCase_  ,				UpperCAmelCase_				)      ->    Union[str, Any]:
											__lowercase        =      self._prepare_train_func(lowercase_  ,				lowercase_  ,				lowercase_				)
											return self._measure_speed(_train				)







						def 							_UpperCamelCase     (self  ,				UpperCAmelCase_  ,				UpperCAmelCase_  ,				UpperCAmelCase_				)      ->    Optional[int]:
											__lowercase        =      self._prepare_train_func(lowercase_  ,				lowercase_  ,				lowercase_				)
											return self._measure_memory(_train				)







						def 							_UpperCamelCase     (self  ,				UpperCAmelCase_  ,				UpperCAmelCase_  ,				UpperCAmelCase_				)      ->    int:
											__lowercase        =      self.config_dict[model_name]

											if self.args.torchscript:
																__lowercase        =      True

											__lowercase        =      (
											    hasattr(lowercase_  ,				'''architectures'''				)
											    and isinstance(config.architectures  ,				lowercase_				)
											    and len(config.architectures				) > 0
											)
											if not self.args.only_pretrain_model and has_model_class_in_config:
																try:
																					__lowercase        =      config.architectures[0]
																					__lowercase        =      __import__('''transformers'''  ,				fromlist=[model_class]				)
																					__lowercase        =      getattr(lowercase_  ,				lowercase_				)
																					__lowercase        =      model_cls(lowercase_				)
																except ImportError:
																					raise ImportError(
																					    f'''{model_class} does not exist. If you just want to test the pretrained model, you might want to'''
																					    ''' set `--only_pretrain_model` or `args.only_pretrain_model=True`.'''				)
											else:
																__lowercase        =      MODEL_MAPPING[config.__class__](lowercase_				)

											model.eval()
											model.to(self.args.device				)

											# encoder-decoder has vocab size saved differently
											__lowercase        =      config.vocab_size if hasattr(lowercase_  ,				'''vocab_size'''				) else config.encoder.vocab_size
											__lowercase        =      torch.randint(lowercase_  ,				(batch_size, sequence_length)  ,				dtype=torch.long  ,				device=self.args.device				)

											if self.args.fpaa:
																logger.info('''Running training in Mixed Precision...'''				)
																if not self.args.is_gpu:
																					raise ValueError('''Mixed precision is possible only for GPU.'''				)
																# amp seems to have memory leaks so that memory usage
																# is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
																model.half()

											if self.args.torchscript:
																with torch.no_grad():
																					__lowercase        =      torch.jit.trace(lowercase_  ,				lowercase_				)
											else:
																__lowercase        =      model

											def encoder_decoder_forward():
																with torch.no_grad():
																					__lowercase        =      inference_model(lowercase_  ,				decoder_input_ids=lowercase_				)
																return outputs

											def encoder_forward():
																with torch.no_grad():
																					__lowercase        =      inference_model(lowercase_				)
																return outputs

											__lowercase        =      encoder_decoder_forward if config.is_encoder_decoder else encoder_forward
											return _forward







						def 							_UpperCamelCase     (self  ,				UpperCAmelCase_  ,				UpperCAmelCase_  ,				UpperCAmelCase_				)      ->    Any:
											__lowercase        =      self.config_dict[model_name]

											__lowercase        =      (
											    hasattr(lowercase_  ,				'''architectures'''				)
											    and isinstance(config.architectures  ,				lowercase_				)
											    and len(config.architectures				) > 0
											)
											if not self.args.only_pretrain_model and has_model_class_in_config:
																try:
																					__lowercase        =      config.architectures[0]
																					__lowercase        =      __import__('''transformers'''  ,				fromlist=[model_class]				)
																					__lowercase        =      getattr(lowercase_  ,				lowercase_				)
																					__lowercase        =      model_cls(lowercase_				)
																except ImportError:
																					raise ImportError(
																					    f'''{model_class} does not exist. If you just want to test the pretrained model, you might want to'''
																					    ''' set `--only_pretrain_model` or `args.only_pretrain_model=True`.'''				)
											else:
																__lowercase        =      MODEL_WITH_LM_HEAD_MAPPING[config.__class__](lowercase_				)

											if self.args.torchscript:
																raise NotImplementedError('''Training for torchscript is currently not implemented'''				)
											else:
																__lowercase        =      model

											model.train()
											model.to(self.args.device				)

											# encoder-decoder has vocab size saved differently
											__lowercase        =      config.vocab_size if hasattr(lowercase_  ,				'''vocab_size'''				) else config.encoder.vocab_size
											__lowercase        =      torch.randint(lowercase_  ,				(batch_size, sequence_length)  ,				dtype=torch.long  ,				device=self.args.device				)

											if self.args.fpaa:
																logger.info('''Running training in Mixed Precision...'''				)
																if not self.args.is_gpu:
																					raise ValueError('''Mixed precision is possible only for GPU.'''				)

																# amp seems to have memory leaks so that memory usage
																# is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
																model.half()

											def compute_loss_and_backprob_encoder():
																__lowercase        =      train_model(lowercase_  ,				labels=lowercase_				)[0]
																loss.backward()
																return loss

											def compute_loss_and_backprob_encoder_decoder():
																__lowercase        =      train_model(lowercase_  ,				decoder_input_ids=lowercase_  ,				labels=lowercase_				)[0]
																loss.backward()
																return loss

											__lowercase        =      (
											    compute_loss_and_backprob_encoder_decoder
											    if config.is_encoder_decoder
											    else compute_loss_and_backprob_encoder
											)
											return _train







						def 							_UpperCamelCase     (self  ,				UpperCAmelCase_				)      ->    Dict:
											try:
																if self.args.is_tpu or self.args.torchscript:
																					# run additional 10 times to stabilize compilation for tpu and torchscript
																					logger.info('''Do inference on TPU or torchscript. Running model 5 times to stabilize compilation'''				)
																					timeit.repeat(
																					    lowercase_  ,				repeat=1  ,				number=5  ,				)

																# as written in https://docs.python.org/2/library/timeit.html#timeit.Timer.repeat, min should be taken rather than the average
																__lowercase        =      timeit.repeat(
																    lowercase_  ,				repeat=self.args.repeat  ,				number=1_0  ,				)

																if self.args.is_tpu and self.args.torch_xla_tpu_print_metrics:
																					import torch_xla.debug.metrics as met

																					self.print_fn(met.metrics_report()				)

																return min(lowercase_				) / 1_0.0
											except RuntimeError as e:
																self.print_fn(f'''Doesn\'t fit on GPU. {e}'''				)
																return "N/A"







						def 							_UpperCamelCase     (self  ,				UpperCAmelCase_				)      ->    Optional[int]:
											try:
																if self.args.trace_memory_line_by_line:
																					__lowercase        =      start_memory_tracing('''transformers'''				)

																if self.args.is_tpu:
																					# tpu
																					raise NotImplementedError(
																					    '''Memory Benchmarking is currently not implemented for TPU. Please disable memory benchmarking with'''
																					    ''' `--no-memory` or `args.memory=False`'''				)
																elif self.args.is_gpu:
																					if not is_pyanvml_available():
																										logger.warning(
																										    '''py3nvml not installed, we won\'t log GPU memory usage. '''
																										    '''Install py3nvml (pip install py3nvml) to log information about GPU.'''				)
																										__lowercase        =      "N/A"
																					else:
																										logger.info(
																										    '''Measuring total GPU usage on GPU device. Make sure to not have additional processes running'''
																										    ''' on the same GPU.'''				)
																										# init nvml
																										nvml.nvmlInit()
																										func()
																										__lowercase        =      nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx				)
																										__lowercase        =      nvml.nvmlDeviceGetMemoryInfo(lowercase_				)
																										__lowercase        =      meminfo.used
																										__lowercase        =      Memory(lowercase_				)
																										# shutdown nvml
																										nvml.nvmlShutdown()
																else:
																					# cpu
																					__lowercase        =      measure_peak_memory_cpu(lowercase_				)
																					__lowercase        =      Memory(lowercase_				) if isinstance(lowercase_  ,				lowercase_				) else memory_bytes

																if self.args.trace_memory_line_by_line:
																					__lowercase        =      stop_memory_tracing(lowercase_				)
																else:
																					__lowercase        =      None

																return memory, summary
											except RuntimeError as e:
																self.print_fn(f'''Doesn\'t fit on GPU. {e}'''				)
																return "N/A", None




'''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_           =    logging.get_logger(__name__)

a_           =    {
    "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  UpperCAmelCase_					(						snake_case__       ):
 UpperCAmelCase_		     =    """blenderbot"""
 UpperCAmelCase_		     =    ["""past_key_values"""]
 UpperCAmelCase_		     =    {"""num_attention_heads""": """encoder_attention_heads""", """hidden_size""": """d_model"""}





 def __init__(     self			,							lowercase_=80_08			,							lowercase_=1_28			,							lowercase_=2			,							lowercase_=1_02_40			,							lowercase_=32			,							lowercase_=24			,							lowercase_=1_02_40			,							lowercase_=32			,							lowercase_=0.0			,							lowercase_=0.0			,							lowercase_=True			,							lowercase_=True			,							lowercase_="gelu"			,							lowercase_=25_60			,							lowercase_=0.1			,							lowercase_=0.0			,							lowercase_=0.0			,							lowercase_=0.02			,							lowercase_=1			,							lowercase_=False			,							lowercase_=0			,							lowercase_=1			,							lowercase_=2			,							lowercase_=3			,							lowercase_=2			,							**lowercase_			,							):
     snake_case_			:     str        =    vocab_size
     snake_case_			:     str        =    max_position_embeddings
     snake_case_			:     Optional[Any]        =    d_model
     snake_case_			:     Any        =    encoder_ffn_dim
     snake_case_			:     Optional[Any]        =    encoder_layers
     snake_case_			:     List[Any]        =    encoder_attention_heads
     snake_case_			:     List[Any]        =    decoder_ffn_dim
     snake_case_			:     Optional[int]        =    decoder_layers
     snake_case_			:     Optional[int]        =    decoder_attention_heads
     snake_case_			:     Any        =    dropout
     snake_case_			:     List[str]        =    attention_dropout
     snake_case_			:     int        =    activation_dropout
     snake_case_			:     Dict        =    activation_function
     snake_case_			:     Any        =    init_std
     snake_case_			:     Union[str, Any]        =    encoder_layerdrop
     snake_case_			:     Optional[int]        =    decoder_layerdrop
     snake_case_			:     Any        =    use_cache
     snake_case_			:     List[str]        =    encoder_layers
     snake_case_			:     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  UpperCAmelCase_					(						snake_case__       ):





 @property
 def 							snake_case__		(     self):
     if self.task in ["default", "seq2seq-lm"]:
         snake_case_			:     int        =    OrderedDict(
             [
                 ("input_ids", {0: "batch", 1: "encoder_sequence"}),
                 ("attention_mask", {0: "batch", 1: "encoder_sequence"}),
             ])
         if self.use_past:
             snake_case_			:     Any        =    {0: "batch"}
             snake_case_			:     Optional[int]        =    {0: "batch", 1: "past_decoder_sequence + sequence"}
         else:
             snake_case_			:     int        =    {0: "batch", 1: "decoder_sequence"}
             snake_case_			:     int        =    {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_			:     Tuple        =    OrderedDict(
             [
                 ("input_ids", {0: "batch", 1: "encoder_sequence"}),
                 ("attention_mask", {0: "batch", 1: "encoder_sequence"}),
             ])
         if self.use_past:
             snake_case_      ,			snake_case_			:     Any        =    self.num_layers
             for i in range(lowercase_):
                 snake_case_			:     Dict        =    {0: "batch", 2: "past_sequence + sequence"}
                 snake_case_			:     Dict        =    {0: "batch", 2: "past_sequence + sequence"}
     else:
         snake_case_			:     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 							snake_case__		(     self):
     if self.task in ["default", "seq2seq-lm"]:
         snake_case_			:     Optional[int]        =    super().outputs
     else:
         snake_case_			:     Dict        =    super(lowercase_			,							self).outputs
         if self.use_past:
             snake_case_      ,			snake_case_			:     Any        =    self.num_layers
             for i in range(lowercase_):
                 snake_case_			:     str        =    {0: "batch", 2: "past_sequence + sequence"}
                 snake_case_			:     int        =    {0: "batch", 2: "past_sequence + sequence"}
     return common_outputs





 def 							snake_case__		(     self			,							lowercase_			,							lowercase_ = -1			,							lowercase_ = -1			,							lowercase_ = False			,							lowercase_ = None			,							):
     snake_case_			:     List[Any]        =    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_			:     List[str]        =    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_			:     Any        =    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_			:     int        =    common_inputs["input_ids"].shape
         snake_case_			:     str        =    common_inputs["decoder_input_ids"].shape[1]
         snake_case_      ,			snake_case_			:     Tuple        =    self.num_attention_heads
         snake_case_			:     str        =    (
             batch,
             num_encoder_attention_heads,
             encoder_seq_length,
             self._config.hidden_size // num_encoder_attention_heads,
         )
         snake_case_			:     Dict        =    decoder_seq_length
         snake_case_			:     List[Any]        =    (
             batch,
             num_decoder_attention_heads,
             decoder_past_length,
             self._config.hidden_size // num_decoder_attention_heads,
         )
         snake_case_			:     Optional[Any]        =    torch.cat(
             [common_inputs["decoder_attention_mask"], torch.ones(lowercase_			,							lowercase_)]			,							dim=1)
         snake_case_			:     Tuple        =    []
         snake_case_      ,			snake_case_			:     Tuple        =    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 							snake_case__		(     self			,							lowercase_			,							lowercase_ = -1			,							lowercase_ = -1			,							lowercase_ = False			,							lowercase_ = None			,							):
     snake_case_			:     str        =    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_			:     Optional[int]        =    self.num_layers
         snake_case_      ,			snake_case_			:     Union[str, Any]        =    self.num_attention_heads
         snake_case_			:     Optional[int]        =    (
             batch,
             num_encoder_attention_heads,
             past_key_values_length,
             self._config.hidden_size // num_encoder_attention_heads,
         )
         snake_case_			:     Dict        =    common_inputs["attention_mask"].dtype
         snake_case_			:     Optional[Any]        =    torch.cat(
             [common_inputs["attention_mask"], torch.ones(lowercase_			,							lowercase_			,							dtype=lowercase_)]			,							dim=1)
         snake_case_			:     Tuple        =    [
             (torch.zeros(lowercase_), torch.zeros(lowercase_)) for _ in range(lowercase_)
         ]
     return common_inputs





 def 							snake_case__		(     self			,							lowercase_			,							lowercase_ = -1			,							lowercase_ = -1			,							lowercase_ = False			,							lowercase_ = None			,							):
     # Copied from OnnxConfig.generate_dummy_inputs
     # Did not use super(OnnxConfigWithPast, self).generate_dummy_inputs for code clarity.
     # If dynamic axis (-1) we forward with a fixed dimension of 2 samples to avoid optimizations made by ONNX
     snake_case_			:     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_			:     Tuple        =    tokenizer.num_special_tokens_to_add(lowercase_)
     snake_case_			:     Tuple        =    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_			:     Optional[int]        =    [" ".join([tokenizer.unk_token]) * seq_length] * batch_size
     snake_case_			:     Tuple        =    dict(tokenizer(lowercase_			,							return_tensors=lowercase_))
     return common_inputs





 def 							snake_case__		(     self			,							lowercase_			,							lowercase_ = -1			,							lowercase_ = -1			,							lowercase_ = False			,							lowercase_ = None			,							):
     if self.task in ["default", "seq2seq-lm"]:
         snake_case_			:     Optional[Any]        =    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_			:     Dict        =    self._generate_dummy_inputs_for_causal_lm(
             lowercase_			,							batch_size=lowercase_			,							seq_length=lowercase_			,							is_pair=lowercase_			,							framework=lowercase_)
     else:
         snake_case_			:     Dict        =    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 							snake_case__		(     self			,							lowercase_			,							lowercase_			,							lowercase_			,							lowercase_):
     if self.task in ["default", "seq2seq-lm"]:
         snake_case_			:     int        =    super()._flatten_past_key_values_(lowercase_			,							lowercase_			,							lowercase_			,							lowercase_)
     else:
         snake_case_			:     List[Any]        =    super(lowercase_			,							self)._flatten_past_key_values_(
             lowercase_			,							lowercase_			,							lowercase_			,							lowercase_)





 def 							snake_case__		(     self			,							lowercase_			,							lowercase_):
     if direction not in ["inputs", "outputs"]:
         raise ValueError(F'direction must either be "inputs" or "outputs", but {direction} was given')

     snake_case_			:     Union[str, Any]        =    "past_key_values" if direction == "inputs" else "present"
     snake_case_      ,			snake_case_			:     Union[str, Any]        =    self.num_layers

     snake_case_			:     List[Any]        =    "past_encoder_sequence"
     snake_case_			:     str        =    "past_decoder_sequence" if direction == "inputs" else "past_decoder_sequence + sequence"

     for i in range(lowercase_):
         snake_case_			:     Union[str, Any]        =    {0: "batch", 2: decoder_sequence}
         snake_case_			:     str        =    {0: "batch", 2: decoder_sequence}
         snake_case_			:     Any        =    {0: "batch", 2: encoder_sequence}
         snake_case_			:     Dict        =    {0: "batch", 2: encoder_sequence}




