




'''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

							_SCREAMING_SNAKE_CASE =	True
except ImportError:
							_SCREAMING_SNAKE_CASE =	False


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





def  lowerCamelCase(       SCREAMING_SNAKE_CASE_    )		->		Optional[Any]:
			A_       		=   importlib.util.find_spec(SCREAMING_SNAKE_CASE_    ) is not None
			if package_exists:
						try:
									A_       		=   importlib.metadata.metadata(SCREAMING_SNAKE_CASE_    )
									return True
						except importlib.metadata.PackageNotFoundError:
									return False





def  lowerCamelCase(       )		->		bool:
			return _torch_distributed_available





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







			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  lowerCamelCase(       )		->		Dict:
			return importlib.metadata.version('oneccl_bind_pt'    )





def  lowerCamelCase(       )		->		Optional[int]:
			return _is_package_available('transformer_engine'    )





@lru_cache
def  lowerCamelCase(       SCREAMING_SNAKE_CASE_=True    )		->		Dict:







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





def  lowerCamelCase(       )		->		Optional[Any]:
			return _is_package_available('deepspeed'    )





def  lowerCamelCase(       SCREAMING_SNAKE_CASE_=False    )		->		Optional[Any]:







			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  lowerCamelCase(       )		->		Optional[int]:
			A_       		=   _is_package_available('bitsandbytes'    )
			if package_exists:
						A_       		=   version.parse(importlib.metadata.version('bitsandbytes'    )    )
						return compare_versions(SCREAMING_SNAKE_CASE_					,'>='					,'0.39.0'    )
			return False





def  lowerCamelCase(       )		->		Optional[int]:
			A_       		=   _is_package_available('bitsandbytes'    )
			if package_exists:
						A_       		=   version.parse(importlib.metadata.version('bitsandbytes'    )    )
						return compare_versions(SCREAMING_SNAKE_CASE_					,'>='					,'0.37.2'    )
			return False





def  lowerCamelCase(       )		->		Optional[Any]:
			return _is_package_available('bitsandbytes'    )





def  lowerCamelCase(       )		->		Any:







			if strtobool(os.environ.get('ACCELERATE_USE_MEGATRON_LM'					,'False'    )    ) == 1:
						A_       		=   importlib.util.find_spec('megatron'    ) is not None
						if package_exists:
									try:
												A_       		=   parse(importlib.metadata.version('megatron-lm'    )    )
												return compare_versions(SCREAMING_SNAKE_CASE_					,'>='					,'2.2.0'    )
									except Exception as e:
												warnings.warn(F'Parse Megatron version failed. Exception:{e}'    )
												return False





def  lowerCamelCase(       )		->		Any:
			return _is_package_available('safetensors'    )





def  lowerCamelCase(       )		->		int:
			return _is_package_available('transformers'    )





def  lowerCamelCase(       )		->		str:
			return _is_package_available('datasets'    )





def  lowerCamelCase(       )		->		Any:
			return _is_package_available('aim'    )





def  lowerCamelCase(       )		->		List[Any]:
			return _is_package_available('tensorboard'    ) or _is_package_available('tensorboardX'    )





def  lowerCamelCase(       )		->		List[Any]:
			return _is_package_available('wandb'    )





def  lowerCamelCase(       )		->		List[Any]:
			return _is_package_available('comet_ml'    )





def  lowerCamelCase(       )		->		List[Any]:
			return _is_package_available('boto3'    )





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







			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'					,SCREAMING_SNAKE_CASE_    )
						return parse_flag_from_env('ACCELERATE_ENABLE_RICH'					,SCREAMING_SNAKE_CASE_    )
			return False





def  lowerCamelCase(       )		->		Optional[Any]:
			return _is_package_available('sagemaker'    )





def  lowerCamelCase(       )		->		Dict:
			return _is_package_available('tqdm'    )





def  lowerCamelCase(       )		->		Any:
			return _is_package_available('mlflow'    )





def  lowerCamelCase(       )		->		Any:
			return is_torch_version('>='					,'1.12'    ) and torch.backends.mps.is_available() and torch.backends.mps.is_built()





def  lowerCamelCase(       )		->		Dict:







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

			A_       		=   importlib.metadata.version('torch'    )
			if importlib.util.find_spec('intel_extension_for_pytorch'    ) is None:
						return False
			A_       		=   'N/A'
			try:
						A_       		=   importlib.metadata.version('intel_extension_for_pytorch'    )
			except importlib.metadata.PackageNotFoundError:
						return False
			A_       		=   get_major_and_minor_from_version(_torch_version    )
			A_       		=   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  lowerCamelCase(       SCREAMING_SNAKE_CASE_=False    )		->		Dict:







			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
									A_       		=   torch.npu.device_count()
									return torch.npu.is_available()
						except RuntimeError:
									return False
			return hasattr(SCREAMING_SNAKE_CASE_					,'npu'    ) and torch.npu.is_available()






@lru_cache
def  lowerCamelCase(       SCREAMING_SNAKE_CASE_=False    )		->		str:







			if not parse_flag_from_env('ACCELERATE_USE_XPU'					,default=SCREAMING_SNAKE_CASE_    ):
						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
									A_       		=   torch.xpu.device_count()
									return torch.xpu.is_available()
						except RuntimeError:
									return False
			return hasattr(SCREAMING_SNAKE_CASE_					,'xpu'    ) and torch.xpu.is_available()











'''simple docstring'''
from typing import Dict, List, Optional, Tuple

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


_SCREAMING_SNAKE_CASE =	logging.get_logger(__name__)
class      __UpperCAmelCase							(    lowerCAmelCase							):



					'''simple docstring'''





					_UpperCamelCase  	=		["""input_ids""", """attention_mask"""]



					def __init__(  self    :	str			,					_lowercase    :	List[Any]="[PAD]"			,					_lowercase    :	Any="[BOS]"			,					_lowercase    :	Any="[EOS]"			,					_lowercase    :	str="[MASK]"			,					_lowercase    :	str="[CLS]"			,					_lowercase    :	Optional[Any]="[SEP]"			,					_lowercase    :	List[str]=2_048			,					**_lowercase    :	int			,					)						->       None:
								A_       		=   AddedToken(_lowercase			,					lstrip=_lowercase			,					rstrip=_lowercase) if isinstance(_lowercase			,					_lowercase) else pad_token
								A_       		=   AddedToken(_lowercase			,					lstrip=_lowercase			,					rstrip=_lowercase) if isinstance(_lowercase			,					_lowercase) else bos_token
								A_       		=   AddedToken(_lowercase			,					lstrip=_lowercase			,					rstrip=_lowercase) if isinstance(_lowercase			,					_lowercase) else eos_token
								A_       		=   AddedToken(_lowercase			,					lstrip=_lowercase			,					rstrip=_lowercase) if isinstance(_lowercase			,					_lowercase) else mask_token
								A_       		=   AddedToken(_lowercase			,					lstrip=_lowercase			,					rstrip=_lowercase) if isinstance(_lowercase			,					_lowercase) else cls_token
								A_       		=   AddedToken(_lowercase			,					lstrip=_lowercase			,					rstrip=_lowercase) if isinstance(_lowercase			,					_lowercase) else sep_token

								super().__init__(
								    pad_token=_lowercase			,					bos_token=_lowercase			,					eos_token=_lowercase			,					mask_token=_lowercase			,					cls_token=_lowercase			,					sep_token=_lowercase			,					model_max_length=_lowercase			,					**_lowercase			,					)

								A_       		=   2**8  # utf is 8 bits

								# define special tokens dict
								A_       		=   {
								    self.pad_token: 0,
								    self.bos_token: 1,
								    self.eos_token: 2,
								    self.mask_token: 3,
								    self.cls_token: 4,
								    self.sep_token: 5,
								}
								A_       		=   len(self.special_tokens_encoder)
								A_       		=   {v: k for k, v in self.special_tokens_encoder.items()}



					def 						__snake_case	(  self    :	List[str])						->       Dict[str, int]:
								A_       		=   self.special_tokens_encoder.copy()
								vocab.update(self.added_tokens_encoder)
								for i in range(self._utf_vocab_size):
											A_       		=   chr(_lowercase)
											A_       		=   i + len(self.special_tokens_encoder)
								return vocab



					@property
					def 						__snake_case	(  self    :	int)						->       Dict:
								return self._utf_vocab_size + self._num_special_tokens



					def 						__snake_case	(  self    :	Union[str, Any]			,					_lowercase    :	List[int]			,					_lowercase    :	Optional[List[int]] = None			,					_lowercase    :	bool = False)						->       List[int]:
								if already_has_special_tokens:
											return super().get_special_tokens_mask(
											    token_ids_a=_lowercase			,					token_ids_a=_lowercase			,					already_has_special_tokens=_lowercase)

								# normal case: some special tokens
								if token_ids_a is None:
											return [1] + [0] * len(_lowercase) + [1]
								return [1] + ([0] * len(_lowercase)) + [1] + ([0] * len(_lowercase)) + [1]



					def 						__snake_case	(  self    :	int			,					_lowercase    :	List[int]			,					_lowercase    :	Optional[List[int]] = None)						->       List[int]:
								if token_ids_a is None:
											return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
								else:
											return [self.cls_token_id] + token_ids_a + [self.sep_token_id] + token_ids_a + [self.sep_token_id]



					def 						__snake_case	(  self    :	str			,					_lowercase    :	str)						->       List[str]:
								A_       		=   [chr(_lowercase) for i in text.encode('utf-8')]
								return tokens



					def 						__snake_case	(  self    :	Union[str, Any]			,					_lowercase    :	Optional[Any])						->       Tuple:
								if token in self.special_tokens_encoder:
											A_       		=   self.special_tokens_encoder[token]
								elif token in self.added_tokens_encoder:
											A_       		=   self.added_tokens_encoder[token]
								elif len(_lowercase) != 1:
											A_       		=   self.unk_token_id
								else:
											A_       		=   ord(_lowercase) + self._num_special_tokens
								return token_id



					def 						__snake_case	(  self    :	str			,					_lowercase    :	List[str])						->       Dict:
								if index in self.special_tokens_decoder:
											A_       		=   self.special_tokens_decoder[index]
								elif index in self.added_tokens_decoder:
											A_       		=   self.added_tokens_decoder[index]
								else:
											A_       		=   chr(index - self._num_special_tokens)
								return token



					def 						__snake_case	(  self    :	str			,					_lowercase    :	Dict)						->       int:
								A_       		=   B''
								for token in tokens:
											if token in self.special_tokens_decoder:
														A_       		=   self.special_tokens_decoder[token].encode('utf-8')
											elif token in self.added_tokens_decoder:
														A_       		=   self.special_tokens_decoder[token].encode('utf-8')
											elif token in self.special_tokens_encoder:
														A_       		=   token.encode('utf-8')
											elif token in self.added_tokens_encoder:
														A_       		=   token.encode('utf-8')
											else:
														A_       		=   bytes([ord(_lowercase)])
											bstring += tok_string
								A_       		=   bstring.decode('utf-8'			,					errors='replace')
								return string



					def 						__snake_case	(  self    :	Optional[int]			,					_lowercase    :	str			,					_lowercase    :	Optional[str] = None)						->       Tuple[str]:
								return ()




