


"""simple docstring"""






import importlib
import importlib.metadata
import os
import warnings
from distutils.util import strtobool
from functools import lru_cache

import torch
from packaging import version
from packaging.version import parse

from .environment import parse_flag_from_env
from .versions import compare_versions, is_torch_version


try:
					import torch_xla.core.xla_model as xm  # noqa: F401

					lowercase  		=			True
except ImportError:
					lowercase  		=			False


# Cache this result has it's a C FFI call which can be pretty time-consuming
lowercase  		=			torch.distributed.is_available()



def 			A__		(					_UpperCAmelCase    :      str	)							->	int:

	'''simple docstring'''


	snake_case__  :						List[Any]		=      importlib.util.find_spec(_UpperCAmelCase	) is not None
	if package_exists:
		try:
			snake_case__  :						Tuple		=      importlib.metadata.metadata(_UpperCAmelCase	)
			return True
		except importlib.metadata.PackageNotFoundError:
			return False



def 			A__		(					)							->	bool:

	'''simple docstring'''


	return _torch_distributed_available



def 			A__		(					)							->	Union[str, Any]:

	'''simple docstring'''


	try:
		pass
	except ImportError:
		print(
		    "Intel(R) oneCCL Bindings for PyTorch* is required to run DDP on Intel(R) GPUs, but it is not"
		    " detected. If you see \"ValueError: Invalid backend: 'ccl'\" error, please install Intel(R) oneCCL"
		    " Bindings for PyTorch*."	)
	return (
	    importlib.util.find_spec("torch_ccl"	) is not None
	    or importlib.util.find_spec("oneccl_bindings_for_pytorch"	) is not None
	)



def 			A__		(					)							->	str:

	'''simple docstring'''


	return importlib.metadata.version("oneccl_bind_pt"	)



def 			A__		(					)							->	Optional[Any]:

	'''simple docstring'''


	return _is_package_available("transformer_engine"	)



@lru_cache
def 			A__		(					_UpperCAmelCase    :      int=True	)							->	Optional[int]:

	'''simple docstring'''


	if torch.cuda.is_available():
		return False
	if _tpu_available and check_device:
		try:
			# Will raise a RuntimeError if no XLA configuration is found
			snake_case__  :						str		=      xm.xla_device()
			return True
		except RuntimeError:
			return False
	return _tpu_available



def 			A__		(					)							->	Union[str, Any]:

	'''simple docstring'''


	return _is_package_available("deepspeed"	)



def 			A__		(					_UpperCAmelCase    :      List[Any]=False	)							->	List[str]:

	'''simple docstring'''


	if is_tpu_available():
		return not ignore_tpu
	if torch.cuda.is_available():
		return torch.cuda.is_bfaa_supported()
	if is_npu_available():
		return False
	return True



def 			A__		(					)							->	Tuple:

	'''simple docstring'''


	snake_case__  :						Any		=      _is_package_available("bitsandbytes"	)
	if package_exists:
		snake_case__  :						List[Any]		=      version.parse(importlib.metadata.version("bitsandbytes"	)	)
		return compare_versions(_UpperCAmelCase				,    ">="				,    "0.39.0"	)
	return False



def 			A__		(					)							->	Dict:

	'''simple docstring'''


	snake_case__  :						List[str]		=      _is_package_available("bitsandbytes"	)
	if package_exists:
		snake_case__  :						List[Any]		=      version.parse(importlib.metadata.version("bitsandbytes"	)	)
		return compare_versions(_UpperCAmelCase				,    ">="				,    "0.37.2"	)
	return False



def 			A__		(					)							->	List[str]:

	'''simple docstring'''


	return _is_package_available("bitsandbytes"	)



def 			A__		(					)							->	List[str]:

	'''simple docstring'''


	if strtobool(os.environ.get("ACCELERATE_USE_MEGATRON_LM"				,    "False"	)	) == 1:
		snake_case__  :						Union[str, Any]		=      importlib.util.find_spec("megatron"	) is not None
		if package_exists:
			try:
				snake_case__  :						List[str]		=      parse(importlib.metadata.version("megatron-lm"	)	)
				return compare_versions(_UpperCAmelCase				,    ">="				,    "2.2.0"	)
			except Exception as e:
				warnings.warn(F"""Parse Megatron version failed. Exception:{e}"""	)
				return False



def 			A__		(					)							->	int:

	'''simple docstring'''


	return _is_package_available("safetensors"	)



def 			A__		(					)							->	Optional[Any]:

	'''simple docstring'''


	return _is_package_available("transformers"	)



def 			A__		(					)							->	List[str]:

	'''simple docstring'''


	return _is_package_available("datasets"	)



def 			A__		(					)							->	Tuple:

	'''simple docstring'''


	return _is_package_available("aim"	)



def 			A__		(					)							->	Any:

	'''simple docstring'''


	return _is_package_available("tensorboard"	) or _is_package_available("tensorboardX"	)



def 			A__		(					)							->	Optional[int]:

	'''simple docstring'''


	return _is_package_available("wandb"	)



def 			A__		(					)							->	Optional[int]:

	'''simple docstring'''


	return _is_package_available("comet_ml"	)



def 			A__		(					)							->	Tuple:

	'''simple docstring'''


	return _is_package_available("boto3"	)



def 			A__		(					)							->	List[Any]:

	'''simple docstring'''


	if _is_package_available("rich"	):
		if "ACCELERATE_DISABLE_RICH" in os.environ:
			warnings.warn(
			    "`ACCELERATE_DISABLE_RICH` is deprecated and will be removed in v0.22.0 and deactivated by default. Please use `ACCELERATE_ENABLE_RICH` if you wish to use `rich`."	)
			return not parse_flag_from_env("ACCELERATE_DISABLE_RICH"				,    _UpperCAmelCase	)
		return parse_flag_from_env("ACCELERATE_ENABLE_RICH"				,    _UpperCAmelCase	)
	return False



def 			A__		(					)							->	Tuple:

	'''simple docstring'''


	return _is_package_available("sagemaker"	)



def 			A__		(					)							->	List[str]:

	'''simple docstring'''


	return _is_package_available("tqdm"	)



def 			A__		(					)							->	str:

	'''simple docstring'''


	return _is_package_available("mlflow"	)



def 			A__		(					)							->	Optional[int]:

	'''simple docstring'''


	return is_torch_version(">="				,    "1.12"	) and torch.backends.mps.is_available() and torch.backends.mps.is_built()



def 			A__		(					)							->	int:

	'''simple docstring'''


	def get_major_and_minor_from_version(_UpperCAmelCase    :      Any	):
		return str(version.parse(_UpperCAmelCase	).major	) + "." + str(version.parse(_UpperCAmelCase	).minor	)

	snake_case__  :						Any		=      importlib.metadata.version("torch"	)
	if importlib.util.find_spec("intel_extension_for_pytorch"	) is None:
		return False
	snake_case__  :						Any		=      "N/A"
	try:
		snake_case__  :						Tuple		=      importlib.metadata.version("intel_extension_for_pytorch"	)
	except importlib.metadata.PackageNotFoundError:
		return False
	snake_case__  :						List[Any]		=      get_major_and_minor_from_version(_torch_version	)
	snake_case__  :						Union[str, Any]		=      get_major_and_minor_from_version(_ipex_version	)
	if torch_major_and_minor != ipex_major_and_minor:
		warnings.warn(
		    F"""Intel Extension for PyTorch {ipex_major_and_minor} needs to work with PyTorch {ipex_major_and_minor}.*,"""
		    F""" but PyTorch {_torch_version} is found. Please switch to the matching version and run again."""	)
		return False
	return True



@lru_cache
def 			A__		(					_UpperCAmelCase    :      Optional[Any]=False	)							->	List[str]:

	'''simple docstring'''


	if importlib.util.find_spec("torch"	) is None or importlib.util.find_spec("torch_npu"	) is None:
		return False

	import torch
	import torch_npu  # noqa: F401

	if check_device:
		try:
			# Will raise a RuntimeError if no NPU is found
			snake_case__  :						str		=      torch.npu.device_count()
			return torch.npu.is_available()
		except RuntimeError:
			return False
	return hasattr(_UpperCAmelCase				,    "npu"	) and torch.npu.is_available()



@lru_cache
def 			A__		(					_UpperCAmelCase    :      str=False	)							->	Union[str, Any]:

	'''simple docstring'''


	if not parse_flag_from_env("ACCELERATE_USE_XPU"				,    default=_UpperCAmelCase	):
		return False
	if is_ipex_available():
		import torch

		if is_torch_version("<="				,    "1.12"	):
			return False
	else:
		return False

	import intel_extension_for_pytorch  # noqa: F401

	if check_device:
		try:
			# Will raise a RuntimeError if no XPU  is found
			snake_case__  :						int		=      torch.xpu.device_count()
			return torch.xpu.is_available()
		except RuntimeError:
			return False
	return hasattr(_UpperCAmelCase				,    "xpu"	) and torch.xpu.is_available()






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


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

lowerCAmelCase__	:  str						=  {
    '''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 					__snake_case    (     _lowerCamelCase      ):
					__lowerCamelCase			   =  """bigbird_pegasus"""
					__lowerCamelCase			   =  ["""past_key_values"""]
					__lowerCamelCase			   =  {
					    """num_attention_heads""": """encoder_attention_heads""",
					    """hidden_size""": """d_model""",
					    """attention_probs_dropout_prob""": """attention_dropout""",
					}




					def __init__(  self       ,			__UpperCamelCase=96103       ,			__UpperCamelCase=4096       ,			__UpperCamelCase=16       ,			__UpperCamelCase=4096       ,			__UpperCamelCase=16       ,			__UpperCamelCase=16       ,			__UpperCamelCase=4096       ,			__UpperCamelCase=16       ,			__UpperCamelCase=0.0       ,			__UpperCamelCase=0.0       ,			__UpperCamelCase=True       ,			__UpperCamelCase=True       ,			__UpperCamelCase="gelu_new"       ,			__UpperCamelCase=1024       ,			__UpperCamelCase=0.1       ,			__UpperCamelCase=0.0       ,			__UpperCamelCase=0.0       ,			__UpperCamelCase=0.0_2       ,			__UpperCamelCase=2       ,			__UpperCamelCase=0.0       ,			__UpperCamelCase=True       ,			__UpperCamelCase=0       ,			__UpperCamelCase=2       ,			__UpperCamelCase=1       ,			__UpperCamelCase="block_sparse"       ,			__UpperCamelCase=64       ,			__UpperCamelCase=3       ,			__UpperCamelCase=False       ,			**__UpperCamelCase       ,			)	->    Tuple:



									'''simple docstring'''







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

									# extra config
									snake_case__   :      Optional[Any]   				= attention_type
									snake_case__   :      Optional[int]   				= block_size
									snake_case__   :      Optional[Any]   				= num_random_blocks
									snake_case__   :      List[str]   				= use_bias

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





class 					__snake_case    (     _lowerCamelCase      ):




					@property
					def      __a     (  self						)	->    Mapping[str, Mapping[int, str]]:



									'''simple docstring'''







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

													if self.use_past:
																	snake_case__   :      Optional[int]   				= {0: 'batch'}
																	snake_case__   :      str   				= {0: 'batch', 1: 'past_decoder_sequence + sequence'}
													else:
																	snake_case__   :      Any   				= {0: 'batch', 1: 'decoder_sequence'}
																	snake_case__   :      List[str]   				= {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.
													snake_case__   :      Any   				= OrderedDict(
													    [
													        ('input_ids', {0: 'batch', 1: 'encoder_sequence'}),
													        ('attention_mask', {0: 'batch', 1: 'encoder_sequence'}),
													    ]						)
													if self.use_past:
																	snake_case__				,    snake_case__   :      Optional[Any]   				= self.num_layers
																	for i in range(__UpperCamelCase						):
																					snake_case__   :      str   				= {0: 'batch', 2: 'past_sequence + sequence'}
																					snake_case__   :      Optional[int]   				= {0: 'batch', 2: 'past_sequence + sequence'}
									else:
													snake_case__   :      Union[str, Any]   				= 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      __a     (  self						)	->    Mapping[str, Mapping[int, str]]:



									'''simple docstring'''







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




					def      __a     (  self       ,			__UpperCamelCase       ,			__UpperCamelCase = -1       ,			__UpperCamelCase = -1       ,			__UpperCamelCase = False       ,			__UpperCamelCase = None       ,			)	->    Mapping[str, Any]:



									'''simple docstring'''







									snake_case__   :      List[str]   				= self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
									    __UpperCamelCase       ,			__UpperCamelCase       ,			__UpperCamelCase       ,			__UpperCamelCase       ,			__UpperCamelCase						)

									# Generate decoder inputs
									snake_case__   :      Optional[Any]   				= seq_length if not self.use_past else 1
									snake_case__   :      int   				= self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
									    __UpperCamelCase       ,			__UpperCamelCase       ,			__UpperCamelCase       ,			__UpperCamelCase       ,			__UpperCamelCase						)
									snake_case__   :      str   				= {F"""decoder_{name}""": tensor for name, tensor in decoder_inputs.items()}
									snake_case__   :      Any   				= 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
													snake_case__				,    snake_case__   :      int   				= common_inputs['input_ids'].shape
													snake_case__   :      int   				= common_inputs['decoder_input_ids'].shape[1]
													snake_case__				,    snake_case__   :      List[Any]   				= self.num_attention_heads
													snake_case__   :      Any   				= (
													    batch,
													    num_encoder_attention_heads,
													    encoder_seq_length,
													    self._config.hidden_size // num_encoder_attention_heads,
													)
													snake_case__   :      Optional[int]   				= decoder_seq_length + 3
													snake_case__   :      Union[str, Any]   				= (
													    batch,
													    num_decoder_attention_heads,
													    decoder_past_length,
													    self._config.hidden_size // num_decoder_attention_heads,
													)

													snake_case__   :      Union[str, Any]   				= torch.cat(
													    [common_inputs['decoder_attention_mask'], torch.ones(__UpperCamelCase       ,			__UpperCamelCase						)]       ,			dim=1						)

													snake_case__   :      Optional[Any]   				= []
													# If the number of encoder and decoder layers are present in the model configuration, both are considered
													snake_case__				,    snake_case__   :      str   				= self.num_layers
													snake_case__   :      str   				= min(__UpperCamelCase       ,			__UpperCamelCase						)
													snake_case__   :      Optional[int]   				= max(__UpperCamelCase       ,			__UpperCamelCase						) - min_num_layers
													snake_case__   :      int   				= '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.
													snake_case__   :      Dict   				= 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      __a     (  self       ,			__UpperCamelCase       ,			__UpperCamelCase = -1       ,			__UpperCamelCase = -1       ,			__UpperCamelCase = False       ,			__UpperCamelCase = None       ,			)	->    Mapping[str, Any]:



									'''simple docstring'''







									snake_case__   :      List[str]   				= 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
													snake_case__				,    snake_case__   :      Optional[int]   				= common_inputs['input_ids'].shape
													# Not using the same length for past_key_values
													snake_case__   :      str   				= seqlen + 2
													snake_case__				,    snake_case__   :      List[str]   				= self.num_layers
													snake_case__				,    snake_case__   :      str   				= self.num_attention_heads
													snake_case__   :      Tuple   				= (
													    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(__UpperCamelCase       ,			__UpperCamelCase       ,			dtype=__UpperCamelCase						)]       ,			dim=1						)
													snake_case__   :      int   				= [
													    (torch.zeros(__UpperCamelCase						), torch.zeros(__UpperCamelCase						)) for _ in range(__UpperCamelCase						)
													]
									return common_inputs




					def      __a     (  self       ,			__UpperCamelCase       ,			__UpperCamelCase = -1       ,			__UpperCamelCase = -1       ,			__UpperCamelCase = False       ,			__UpperCamelCase = None       ,			)	->    Mapping[str, Any]:



									'''simple docstring'''







									snake_case__   :      Optional[int]   				= 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
									snake_case__   :      str   				= tokenizer.num_special_tokens_to_add(__UpperCamelCase						)
									snake_case__   :      List[str]   				= 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
									snake_case__   :      Optional[Any]   				= [' '.join([tokenizer.unk_token]						) * seq_length] * batch_size
									snake_case__   :      Optional[Any]   				= dict(tokenizer(__UpperCamelCase       ,			return_tensors=__UpperCamelCase						)						)
									return common_inputs




					def      __a     (  self       ,			__UpperCamelCase       ,			__UpperCamelCase = -1       ,			__UpperCamelCase = -1       ,			__UpperCamelCase = False       ,			__UpperCamelCase = None       ,			)	->    Mapping[str, Any]:



									'''simple docstring'''







									if self.task in ["default", "seq2seq-lm"]:
													snake_case__   :      Dict   				= 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":
													snake_case__   :      Union[str, Any]   				= self._generate_dummy_inputs_for_causal_lm(
													    __UpperCamelCase       ,			batch_size=__UpperCamelCase       ,			seq_length=__UpperCamelCase       ,			is_pair=__UpperCamelCase       ,			framework=__UpperCamelCase						)
									else:
													snake_case__   :      Any   				= 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      __a     (  self       ,			__UpperCamelCase       ,			__UpperCamelCase       ,			__UpperCamelCase       ,			__UpperCamelCase						)	->    Dict:



									'''simple docstring'''







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


