
import csv
import json
import logging
import os
import re
import subprocess
import tempfile
import unittest
import zipfile
from pathlib import Path
from typing import Optional
from unittest import mock

import torch

# We use TF to parse the logs
from accelerate import Accelerator
from accelerate.test_utils.testing import (
    MockingTestCase,
    TempDirTestCase,
    require_comet_ml,
    require_tensorboard,
    require_wandb,
    skip,
)
from accelerate.tracking import CometMLTracker, GeneralTracker
from accelerate.utils import ProjectConfiguration, is_comet_ml_available, is_tensorboard_available


if is_comet_ml_available():
   from comet_ml import OfflineExperiment

if is_tensorboard_available():
   import struct

   import tensorboard.compat.proto.event_pba as event_pba

_A  					= logging.getLogger(__name__)







@require_tensorboard
class      lowerCamelCase		(  unittest.TestCase  ):





     def 		UpperCAmelCase(self      :    List[str]   )						->						Dict:
         snake_case      =					"test_project_with_config"
         with tempfile.TemporaryDirectory() as dirpath:
             snake_case      =					Accelerator(log_with="tensorboard"				,    project_dir=_A   )
             snake_case      =					{"num_iterations": 1_2, "learning_rate": 1E-2, "some_boolean": False, "some_string": "some_value"}
             accelerator.init_trackers(_A				,    _A   )
             accelerator.end_training()
             for child in Path(f'{dirpath}/{project_name}'   ).glob("*/**"   ):
                 snake_case      =					list(filter(lambda _A   : x.is_file()				,    child.iterdir()   )   )[0]
             self.assertNotEqual(str(_A   )				,    ""   )





     def 		UpperCAmelCase(self      :    List[str]   )						->						Union[str, Any]:
         snake_case      =					"test_project_with_log"
         with tempfile.TemporaryDirectory() as dirpath:
             snake_case      =					Accelerator(log_with="tensorboard"				,    project_dir=_A   )
             accelerator.init_trackers(_A   )
             snake_case      =					{"total_loss": 0.1, "iteration": 1, "my_text": "some_value"}
             accelerator.log(_A				,    step=0   )
             accelerator.end_training()
             # Logged values are stored in the outermost-tfevents file and can be read in as a TFRecord
             # Names are randomly generated each time
             snake_case      =					list(filter(lambda _A   : x.is_file()				,    Path(f'{dirpath}/{project_name}'   ).iterdir()   )   )[0]
             self.assertNotEqual(str(_A   )				,    ""   )





     def 		UpperCAmelCase(self      :    Any   )						->						Tuple:
         snake_case      =					"test_project_with_log"
         with tempfile.TemporaryDirectory() as dirpath:
             snake_case      =					Accelerator(log_with="tensorboard"				,    project_dir=_A   )
             accelerator.init_trackers(_A   )
             snake_case      =					{"tensor": torch.tensor(1   )}
             accelerator.log(_A				,    step=0   )
             accelerator.end_training()
             # Logged values are stored in the outermost-tfevents file and can be read in as a TFRecord
             # Names are randomly generated each time
             snake_case      =					list(filter(lambda _A   : x.is_file()				,    Path(f'{dirpath}/{project_name}'   ).iterdir()   )   )[0]
             # Reading implementation based on https://github.com/pytorch/pytorch/issues/45327#issuecomment-703757685
             with open(_A				,    "rb"   ) as f:
                 snake_case      =					f.read()
             snake_case      =					False
             while data:
                 snake_case      =					struct.unpack("Q"				,    data[:8]   )

                 snake_case      =					data[1_2 : 1_2 + int(header[0]   )]  # 8+4
                 snake_case      =					data[1_2 + int(header[0]   ) + 4 :]
                 snake_case      =					event_pba.Event()

                 event.ParseFromString(_A   )
                 if event.HasField("summary"   ):
                     for value in event.summary.value:
                         if value.simple_value == 1.0 and value.tag == "tensor":
                             snake_case      =					True
             self.assertTrue(_A				,    "Converted tensor was not found in the log file!"   )





     def 		UpperCAmelCase(self      :    List[str]   )						->						Optional[Any]:
         with self.assertRaisesRegex(_A				,    "Logging with `tensorboard` requires a `logging_dir`"   ):
             snake_case      =					Accelerator(log_with="tensorboard"   )
         with tempfile.TemporaryDirectory() as dirpath:
             snake_case      =					Accelerator(log_with="tensorboard"				,    project_dir=_A   )





     def 		UpperCAmelCase(self      :    Optional[Any]   )						->						List[Any]:
         snake_case      =					ProjectConfiguration(total_limit=3_0   )
         with tempfile.TemporaryDirectory() as dirpath:
             snake_case      =					Accelerator(log_with="tensorboard"				,    project_dir=_A				,    project_config=_A   )







@require_wandb
@mock.patch.dict(os.environ							, {"WANDB_MODE": "offline"}  )
class      lowerCamelCase		(  A_							, A_  ):





     def 		UpperCAmelCase(self      :    int   )						->						Optional[Any]:
         super().setUp()
         # wandb let's us override where logs are stored to via the WANDB_DIR env var
         self.add_mocks(mock.patch.dict(os.environ				,    {"WANDB_DIR": self.tmpdir}   )   )





     @staticmethod
     def 		UpperCAmelCase(_A      :    str				,    _A      :    str				,    _A      :    bool = True   )						->						Any:
         snake_case      =					rf'{section} ([\S\s]*?)\n\n'
         if record:
             snake_case      =					rf'Record: {pattern}'
         snake_case      =					re.findall(_A				,    _A   )[0]
         # A config
         if section == "config" or section == "history":
             snake_case      =					re.findall(r"\"([a-zA-Z0-9_.,]+)"				,    _A   )
             return {key: val for key, val in zip(cleaned_record[0::2]				,    cleaned_record[1::2]   )}
         # Everything else
         else:
             return dict(re.findall(r"(\w+): \"([^\s]+)\""				,    _A   )   )





     @skip
     def 		UpperCAmelCase(self      :    Optional[int]   )						->						Dict:
         snake_case      =					"test_project_with_config"
         snake_case      =					Accelerator(log_with="wandb"   )
         snake_case      =					{"num_iterations": 1_2, "learning_rate": 1E-2, "some_boolean": False, "some_string": "some_value"}
         snake_case      =					{"wandb": {"tags": ["my_tag"]}}
         accelerator.init_trackers(_A				,    _A				,    _A   )
         snake_case      =					{"total_loss": 0.1, "iteration": 1, "my_text": "some_value"}
         accelerator.log(_A				,    step=0   )
         accelerator.end_training()
         # The latest offline log is stored at wandb/latest-run/*.wandb
         for child in Path(f'{self.tmpdir}/wandb/latest-run'   ).glob("*"   ):
             if child.is_file() and child.suffix == ".wandb":
                 snake_case      =					subprocess.check_output(
                     ["wandb", "sync", "--view", "--verbose", str(_A   )]				,    env=os.environ.copy()   ).decode("utf8"				,    "ignore"   )
                 break

        # Check HPS through careful parsing and cleaning
         snake_case      =					self.parse_log(_A				,    "config"   )
         self.assertEqual(logged_items["num_iterations"]				,    "12"   )
         self.assertEqual(logged_items["learning_rate"]				,    "0.01"   )
         self.assertEqual(logged_items["some_boolean"]				,    "false"   )
         self.assertEqual(logged_items["some_string"]				,    "some_value"   )
         self.assertEqual(logged_items["some_string"]				,    "some_value"   )

         # Run tags
         snake_case      =					self.parse_log(_A				,    "run"				,    _A   )
         self.assertEqual(logged_items["tags"]				,    "my_tag"   )

         # Actual logging
         snake_case      =					self.parse_log(_A				,    "history"   )
         self.assertEqual(logged_items["total_loss"]				,    "0.1"   )
         self.assertEqual(logged_items["iteration"]				,    "1"   )
         self.assertEqual(logged_items["my_text"]				,    "some_value"   )
         self.assertEqual(logged_items["_step"]				,    "0"   )



def 						lowercase_			(   self ,		A__ ,		A__		)							->							Any:
    """simple docstring"""

    snake_case      =					run_name
    snake_case      =					OfflineExperiment(project_name=A__ ,		offline_directory=A__		)
    logger.info(F'Initialized offline CometML project {self.run_name}'		)
    logger.info("Make sure to log any initial configurations with `self.store_init_configuration` before training!"		)







@require_comet_ml
@mock.patch.object(A_							, "__init__"							, A_  )
class      lowerCamelCase		(  unittest.TestCase  ):





     @staticmethod
     def 		UpperCAmelCase(_A      :    Tuple				,    _A      :    str				,    _A      :    bool = False   )						->						Optional[Any]:
         for log in log_list:
             snake_case      =					json.loads(_A   )["payload"]
             if is_param and "param" in j.keys():
                 if j["param"]["paramName"] == key:
                     return j["param"]["paramValue"]
             if "log_other" in j.keys():
                 if j["log_other"]["key"] == key:
                     return j["log_other"]["val"]
             if "metric" in j.keys():
                 if j["metric"]["metricName"] == key:
                     return j["metric"]["metricValue"]





     def 		UpperCAmelCase(self      :    str   )						->						Union[str, Any]:
         with tempfile.TemporaryDirectory() as d:
             snake_case      =					CometMLTracker("test_project_with_config"				,    _A   )
             snake_case      =					Accelerator(log_with=_A   )
             snake_case      =					{"num_iterations": 1_2, "learning_rate": 1E-2, "some_boolean": False, "some_string": "some_value"}
             accelerator.init_trackers(_A				,    _A   )
             accelerator.end_training()
             snake_case      =					os.listdir(_A   )[0]  # Comet is nice, it's just a zip file here
             # We parse the raw logs
             snake_case      =					os.path.join(_A				,    _A   )
             snake_case      =					zipfile.ZipFile(_A				,    "r"   )
             snake_case      =					archive.open("messages.json"   ).read().decode("utf-8"   )
         snake_case      =					log.split("\n"   )[:-1]
         self.assertEqual(self.get_value_from_key(_A				,    "num_iterations"				,    _A   )				,    1_2   )
         self.assertEqual(self.get_value_from_key(_A				,    "learning_rate"				,    _A   )				,    0.01   )
         self.assertEqual(self.get_value_from_key(_A				,    "some_boolean"				,    _A   )				,    _A   )
         self.assertEqual(self.get_value_from_key(_A				,    "some_string"				,    _A   )				,    "some_value"   )





     def 		UpperCAmelCase(self      :    Dict   )						->						Optional[Any]:
         with tempfile.TemporaryDirectory() as d:
             snake_case      =					CometMLTracker("test_project_with_config"				,    _A   )
             snake_case      =					Accelerator(log_with=_A   )
             accelerator.init_trackers(_A   )
             snake_case      =					{"total_loss": 0.1, "iteration": 1, "my_text": "some_value"}
             accelerator.log(_A				,    step=0   )
             accelerator.end_training()
             snake_case      =					os.listdir(_A   )[0]  # Comet is nice, it's just a zip file here
             # We parse the raw logs
             snake_case      =					os.path.join(_A				,    _A   )
             snake_case      =					zipfile.ZipFile(_A				,    "r"   )
             snake_case      =					archive.open("messages.json"   ).read().decode("utf-8"   )
         snake_case      =					log.split("\n"   )[:-1]
         self.assertEqual(self.get_value_from_key(_A				,    "curr_step"				,    _A   )				,    0   )
         self.assertEqual(self.get_value_from_key(_A				,    "total_loss"   )				,    0.1   )
         self.assertEqual(self.get_value_from_key(_A				,    "iteration"   )				,    1   )
         self.assertEqual(self.get_value_from_key(_A				,    "my_text"   )				,    "some_value"   )







class      lowerCamelCase		(  A_  ):
     UpperCAmelCase__				:							Dict           =		[
         "total_loss",
         "iteration",
         "my_text",
         "learning_rate",
         "num_iterations",
         "some_boolean",
         "some_string",
     ]

     UpperCAmelCase__				:							Tuple           =		"my_custom_tracker"
     UpperCAmelCase__				:							Optional[int]           =		False





     def __init__(self      :    str				,    _A      :    str   )						->						Union[str, Any]:
         snake_case      =					open(f'{dir}/log.csv'				,    "w+"   )
         snake_case      =					csv.DictWriter(self.f				,    fieldnames=self._col_names   )
         self.writer.writeheader()





     @property
     def 		UpperCAmelCase(self      :    List[Any]   )						->						Optional[Any]:
         return self.writer





     def 		UpperCAmelCase(self      :    Any				,    _A      :    dict   )						->						List[str]:
         logger.info("Call init"   )
         self.writer.writerow(_A   )





     def 		UpperCAmelCase(self      :    List[Any]				,    _A      :    dict				,    _A      :    Optional[int]   )						->						Dict:
         logger.info("Call log"   )
         self.writer.writerow(_A   )





     def 		UpperCAmelCase(self      :    int   )						->						Optional[int]:
         self.f.close()







class      lowerCamelCase		(  unittest.TestCase  ):





     def 		UpperCAmelCase(self      :    List[str]   )						->						List[Any]:
         with tempfile.TemporaryDirectory() as d:
             snake_case      =					MyCustomTracker(_A   )
             snake_case      =					Accelerator(log_with=_A   )
             snake_case      =					{"num_iterations": 1_2, "learning_rate": 1E-2, "some_boolean": False, "some_string": "some_value"}
             accelerator.init_trackers("Some name"				,    _A   )
             accelerator.end_training()
             with open(f'{d}/log.csv'				,    "r"   ) as f:
                 snake_case      =					csv.DictReader(_A   )
                 snake_case      =					next(_A   )
                 snake_case      =					{
                     "total_loss": "",
                     "iteration": "",
                     "my_text": "",
                     "learning_rate": "0.01",
                     "num_iterations": "12",
                     "some_boolean": "False",
                     "some_string": "some_value",
                 }
                 self.assertDictEqual(_A				,    _A   )





     def 		UpperCAmelCase(self      :    List[Any]   )						->						int:
         with tempfile.TemporaryDirectory() as d:
             snake_case      =					MyCustomTracker(_A   )
             snake_case      =					Accelerator(log_with=_A   )
             accelerator.init_trackers("Some name"   )
             snake_case      =					{"total_loss": 0.1, "iteration": 1, "my_text": "some_value"}
             accelerator.log(_A				,    step=0   )
             accelerator.end_training()
             with open(f'{d}/log.csv'				,    "r"   ) as f:
                 snake_case      =					csv.DictReader(_A   )
                 snake_case      =					next(_A   )
                 snake_case      =					{
                     "total_loss": "0.1",
                     "iteration": "1",
                     "my_text": "some_value",
                     "learning_rate": "",
                     "num_iterations": "",
                     "some_boolean": "",
                     "some_string": "",
                 }
                 self.assertDictEqual(_A				,    _A   )



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


_A  					= logging.get_logger(__name__)

_A  					= {
    "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      lowerCamelCase		(  A_  ):
     UpperCAmelCase__				:							List[Any]           =		"bigbird_pegasus"
     UpperCAmelCase__				:							Tuple           =		["past_key_values"]
     UpperCAmelCase__				:							Optional[Any]           =		{
         "num_attention_heads": "encoder_attention_heads",
         "hidden_size": "d_model",
         "attention_probs_dropout_prob": "attention_dropout",
     }





     def __init__(self      :    Any				,    _A      :    List[str]=9_6_1_0_3				,    _A      :    List[Any]=4_0_9_6				,    _A      :    str=1_6				,    _A      :    Tuple=4_0_9_6				,    _A      :    List[str]=1_6				,    _A      :    Optional[int]=1_6				,    _A      :    List[Any]=4_0_9_6				,    _A      :    Dict=1_6				,    _A      :    Tuple=0.0				,    _A      :    List[str]=0.0				,    _A      :    Dict=True				,    _A      :    int=True				,    _A      :    List[Any]="gelu_new"				,    _A      :    int=1_0_2_4				,    _A      :    int=0.1				,    _A      :    List[Any]=0.0				,    _A      :    Optional[int]=0.0				,    _A      :    Any=0.02				,    _A      :    Dict=2				,    _A      :    List[Any]=0.0				,    _A      :    Any=True				,    _A      :    str=0				,    _A      :    Dict=2				,    _A      :    Any=1				,    _A      :    List[str]="block_sparse"				,    _A      :    int=6_4				,    _A      :    Optional[int]=3				,    _A      :    List[Any]=False				,    **_A      :    Tuple				,    )						->						Any:
         snake_case      =					vocab_size
         snake_case      =					max_position_embeddings
         snake_case      =					d_model
         snake_case      =					encoder_ffn_dim
         snake_case      =					encoder_layers
         snake_case      =					encoder_attention_heads
         snake_case      =					decoder_ffn_dim
         snake_case      =					decoder_layers
         snake_case      =					decoder_attention_heads
         snake_case      =					dropout
         snake_case      =					attention_dropout
         snake_case      =					activation_dropout
         snake_case      =					activation_function
         snake_case      =					init_std
         snake_case      =					encoder_layerdrop
         snake_case      =					decoder_layerdrop
         snake_case      =					classifier_dropout
         snake_case      =					use_cache
         snake_case      =					encoder_layers
         snake_case      =					scale_embedding  # scale factor will be sqrt(d_model) if True

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

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







class      lowerCamelCase		(  A_  ):





     @property
     def 		UpperCAmelCase(self      :    Any   )						->						Mapping[str, Mapping[int, str]]:
         if self.task in ["default", "seq2seq-lm"]:
             snake_case      =					OrderedDict(
                 [
                     ("input_ids", {0: "batch", 1: "encoder_sequence"}),
                     ("attention_mask", {0: "batch", 1: "encoder_sequence"}),
                 ]   )

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

             if self.use_past:
                 self.fill_with_past_key_values_(_A				,    direction="inputs"   )
         elif self.task == "causal-lm":
             # TODO: figure this case out.
             snake_case      =					OrderedDict(
                 [
                     ("input_ids", {0: "batch", 1: "encoder_sequence"}),
                     ("attention_mask", {0: "batch", 1: "encoder_sequence"}),
                 ]   )
             if self.use_past:
                 snake_case     ,							snake_case      =					self.num_layers
                 for i in range(_A   ):
                     snake_case      =					{0: "batch", 2: "past_sequence + sequence"}
                     snake_case      =					{0: "batch", 2: "past_sequence + sequence"}
         else:
             snake_case      =					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 		UpperCAmelCase(self      :    str   )						->						Mapping[str, Mapping[int, str]]:
         if self.task in ["default", "seq2seq-lm"]:
             snake_case      =					super().outputs
         else:
             snake_case      =					super(_A				,    self   ).outputs
             if self.use_past:
                 snake_case     ,							snake_case      =					self.num_layers
                 for i in range(_A   ):
                     snake_case      =					{0: "batch", 2: "past_sequence + sequence"}
                     snake_case      =					{0: "batch", 2: "past_sequence + sequence"}
         return common_outputs





     def 		UpperCAmelCase(self      :    Any				,    _A      :    PreTrainedTokenizer				,    _A      :    int = -1				,    _A      :    int = -1				,    _A      :    bool = False				,    _A      :    Optional[TensorType] = None				,    )						->						Mapping[str, Any]:
         snake_case      =					self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
             _A				,    _A				,    _A				,    _A				,    _A   )

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

         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      =					common_inputs["input_ids"].shape
             snake_case      =					common_inputs["decoder_input_ids"].shape[1]
             snake_case     ,							snake_case      =					self.num_attention_heads
             snake_case      =					(
                 batch,
                 num_encoder_attention_heads,
                 encoder_seq_length,
                 self._config.hidden_size // num_encoder_attention_heads,
             )
             snake_case      =					decoder_seq_length + 3
             snake_case      =					(
                 batch,
                 num_decoder_attention_heads,
                 decoder_past_length,
                 self._config.hidden_size // num_decoder_attention_heads,
             )

             snake_case      =					torch.cat(
                 [common_inputs["decoder_attention_mask"], torch.ones(_A				,    _A   )]				,    dim=1   )

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

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





     def 		UpperCAmelCase(self      :    Tuple				,    _A      :    PreTrainedTokenizer				,    _A      :    int = -1				,    _A      :    int = -1				,    _A      :    bool = False				,    _A      :    Optional[TensorType] = None				,    )						->						Mapping[str, Any]:
         snake_case      =					self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
             _A				,    _A				,    _A				,    _A				,    _A   )

         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      =					common_inputs["input_ids"].shape
             # Not using the same length for past_key_values
             snake_case      =					seqlen + 2
             snake_case     ,							snake_case      =					self.num_layers
             snake_case     ,							snake_case      =					self.num_attention_heads
             snake_case      =					(
                 batch,
                 num_encoder_attention_heads,
                 past_key_values_length,
                 self._config.hidden_size // num_encoder_attention_heads,
             )

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





     def 		UpperCAmelCase(self      :    Dict				,    _A      :    PreTrainedTokenizer				,    _A      :    int = -1				,    _A      :    int = -1				,    _A      :    bool = False				,    _A      :    Optional[TensorType] = None				,    )						->						Mapping[str, Any]:
         # 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      =					compute_effective_axis_dimension(
             _A				,    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      =					tokenizer.num_special_tokens_to_add(_A   )
         snake_case      =					compute_effective_axis_dimension(
             _A				,    fixed_dimension=OnnxConfig.default_fixed_sequence				,    num_token_to_add=_A   )

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





     def 		UpperCAmelCase(self      :    Any				,    _A      :    PreTrainedTokenizer				,    _A      :    int = -1				,    _A      :    int = -1				,    _A      :    bool = False				,    _A      :    Optional[TensorType] = None				,    )						->						Mapping[str, Any]:
         if self.task in ["default", "seq2seq-lm"]:
             snake_case      =					self._generate_dummy_inputs_for_default_and_seqaseq_lm(
                 _A				,    batch_size=_A				,    seq_length=_A				,    is_pair=_A				,    framework=_A   )

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

         return common_inputs





     def 		UpperCAmelCase(self      :    List[Any]				,    _A      :    str				,    _A      :    Union[str, Any]				,    _A      :    Tuple				,    _A      :    List[Any]   )						->						Any:
         if self.task in ["default", "seq2seq-lm"]:
             snake_case      =					super()._flatten_past_key_values_(_A				,    _A				,    _A				,    _A   )
         else:
             snake_case      =					super(_A				,    self   )._flatten_past_key_values_(
                 _A				,    _A				,    _A				,    _A   )

