



'''simple docstring'''


import os
import re
import warnings
from shutil import copyfile
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer


if TYPE_CHECKING:
						from ...tokenization_utils_base import TextInput
from ...utils import logging


lowercase_      :				int   					=    logging.get_logger(__name__)

lowercase_      :				Tuple   					=    {'''vocab_file''': '''spiece.model'''}

lowercase_      :				Any   					=    {
    '''vocab_file''': {
        '''t5-small''': '''https://huggingface.co/t5-small/resolve/main/spiece.model''',
        '''t5-base''': '''https://huggingface.co/t5-base/resolve/main/spiece.model''',
        '''t5-large''': '''https://huggingface.co/t5-large/resolve/main/spiece.model''',
        '''t5-3b''': '''https://huggingface.co/t5-3b/resolve/main/spiece.model''',
        '''t5-11b''': '''https://huggingface.co/t5-11b/resolve/main/spiece.model''',
    }
}


# TODO(PVP) - this should be removed in Transformers v5
lowercase_      :				Optional[Any]   					=    {
    '''t5-small''': 512,
    '''t5-base''': 512,
    '''t5-large''': 512,
    '''t5-3b''': 512,
    '''t5-11b''': 512,
}

lowercase_      :				int   					=    '''▁'''




class      __UpperCamelCase       (_UpperCAmelCase			):
	__A			  =					VOCAB_FILES_NAMES
	__A			  =					PRETRAINED_VOCAB_FILES_MAP
	__A			  =					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	__A			  =					['''input_ids''', '''attention_mask''']







	def __init__(    self ,				_lowerCAmelCase ,				_lowerCAmelCase="</s>" ,				_lowerCAmelCase="<unk>" ,				_lowerCAmelCase="<pad>" ,				_lowerCAmelCase=100 ,				_lowerCAmelCase=None ,				_lowerCAmelCase = None ,				_lowerCAmelCase=True ,				**_lowerCAmelCase ,				)    ->    None:
							'''simple docstring'''






							if extra_ids > 0 and additional_special_tokens is None:
													lowercase        =							[F"""<extra_id_{i}>""" for i in range(_lowerCAmelCase						)]
							elif extra_ids > 0 and additional_special_tokens is not None:
													# Check that we have the right number of extra_id special tokens
													lowercase        =							len(set(filter(lambda _lowerCAmelCase						: bool("""extra_id""" in str(_lowerCAmelCase						)						) ,				_lowerCAmelCase						)						)						)
													if extra_tokens != extra_ids:
																			raise ValueError(
																			    F"""Both extra_ids ({extra_ids}) and additional_special_tokens ({additional_special_tokens}) are"""
																			    """ provided to T5Tokenizer. In this case the additional_special_tokens must include the extra_ids"""
																			    """ tokens"""						)
							if legacy:
													logger.warning_once(
													    F"""You are using the legacy behaviour of the {self.__class__}. This means that tokens that come after special tokens will not be properly handled. We recommend you to"""
													    """ read the related pull request available at https://github.com/huggingface/transformers/pull/24565"""						)

							lowercase        =							legacy
							lowercase        =							{} if sp_model_kwargs is None else sp_model_kwargs

							super().__init__(
							    eos_token=_lowerCAmelCase ,				unk_token=_lowerCAmelCase ,				pad_token=_lowerCAmelCase ,				extra_ids=_lowerCAmelCase ,				additional_special_tokens=_lowerCAmelCase ,				sp_model_kwargs=self.sp_model_kwargs ,				legacy=_lowerCAmelCase ,				**_lowerCAmelCase ,				)

							lowercase        =							vocab_file
							lowercase        =							extra_ids

							lowercase        =							spm.SentencePieceProcessor(**self.sp_model_kwargs						)
							self.sp_model.Load(_lowerCAmelCase						)







	@staticmethod
	def   _a   (    _lowerCAmelCase ,				_lowerCAmelCase ,				_lowerCAmelCase						)    ->    Dict:
							'''simple docstring'''






							if pretrained_model_name_or_path in TaTokenizer.max_model_input_sizes:
													lowercase        =							TaTokenizer.max_model_input_sizes[pretrained_model_name_or_path]
													if init_max_model_length is not None and init_max_model_length != max_model_length:
																			return init_max_model_length
													elif init_max_model_length is None:
																			warnings.warn(
																			    """This tokenizer was incorrectly instantiated with a model max length of"""
																			    F""" {deprecated_max_model_length} which will be corrected in Transformers v5.\nFor now, this"""
																			    """ behavior is kept to avoid breaking backwards compatibility when padding/encoding with"""
																			    """ `truncation is True`.\n- Be aware that you SHOULD NOT rely on"""
																			    F""" {pretrained_model_name_or_path} automatically truncating your input to"""
																			    F""" {deprecated_max_model_length} when padding/encoding.\n- If you want to encode/pad to sequences"""
																			    F""" longer than {deprecated_max_model_length} you can either instantiate this tokenizer with"""
																			    """ `model_max_length` or pass `max_length` when encoding/padding.\n- To avoid this warning, please"""
																			    """ instantiate this tokenizer with `model_max_length` set to your preferred value.""" ,				_lowerCAmelCase ,				)

							return max_model_length







	@property
	def   _a   (    self						)    ->    Union[str, Any]:
							'''simple docstring'''






							return self.sp_model.get_piece_size() + self._extra_ids







	def   _a   (    self						)    ->    Optional[Any]:
							'''simple docstring'''






							lowercase        =							{self.convert_ids_to_tokens(_lowerCAmelCase						): i for i in range(self.vocab_size						)}
							vocab.update(self.added_tokens_encoder						)
							return vocab







	def   _a   (    self ,				_lowerCAmelCase ,				_lowerCAmelCase = None ,				_lowerCAmelCase = False						)    ->    List[int]:
							'''simple docstring'''






							if already_has_special_tokens:
													return super().get_special_tokens_mask(
													    token_ids_a=_lowerCAmelCase ,				token_ids_a=_lowerCAmelCase ,				already_has_special_tokens=_lowerCAmelCase						)

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







	def   _a   (    self						)    ->    List[Any]:
							'''simple docstring'''






							return list(
							    set(filter(lambda _lowerCAmelCase						: bool(re.search(r"""<extra_id_\d+>""" ,				_lowerCAmelCase						)						) is not None ,				self.additional_special_tokens						)						)						)







	def   _a   (    self						)    ->    List[str]:
							'''simple docstring'''






							return [self._convert_token_to_id(_lowerCAmelCase						) for token in self.get_sentinel_tokens()]







	def   _a   (    self ,				_lowerCAmelCase						)    ->    List[int]:
							'''simple docstring'''






							if len(_lowerCAmelCase						) > 0 and token_ids[-1] == self.eos_token_id:
													warnings.warn(
													    F"""This sequence already has {self.eos_token}. In future versions this behavior may lead to duplicated"""
													    """ eos tokens being added."""						)
													return token_ids
							else:
													return token_ids + [self.eos_token_id]







	def   _a   (    self ,				_lowerCAmelCase ,				_lowerCAmelCase = None						)    ->    List[int]:
							'''simple docstring'''






							lowercase        =							[self.eos_token_id]

							if token_ids_a is None:
													return len(token_ids_a + eos						) * [0]
							return len(token_ids_a + eos + token_ids_a + eos						) * [0]







	def   _a   (    self ,				_lowerCAmelCase ,				_lowerCAmelCase = None						)    ->    List[int]:
							'''simple docstring'''






							lowercase        =							self._add_eos_if_not_present(_lowerCAmelCase						)
							if token_ids_a is None:
													return token_ids_a
							else:
													lowercase        =							self._add_eos_if_not_present(_lowerCAmelCase						)
													return token_ids_a + token_ids_a







	def __getstate__(    self						)    ->    Union[str, Any]:
							'''simple docstring'''






							lowercase        =							self.__dict__.copy()
							lowercase        =							None
							return state







	def __setstate__(    self ,				_lowerCAmelCase						)    ->    str:
							'''simple docstring'''






							lowercase        =							d

							# for backward compatibility
							if not hasattr(self ,				"""sp_model_kwargs"""						):
													lowercase        =							{}

							lowercase        =							spm.SentencePieceProcessor(**self.sp_model_kwargs						)
							self.sp_model.Load(self.vocab_file						)







	def   _a   (    self ,				_lowerCAmelCase ,				**_lowerCAmelCase						)    ->    List[str]:
							'''simple docstring'''






							if not self.legacy:
													lowercase        =							SPIECE_UNDERLINE + text.replace(_lowerCAmelCase ,				""" """						)
							return super().tokenize(_lowerCAmelCase ,				**_lowerCAmelCase						)







	def   _a   (    self ,				_lowerCAmelCase ,				**_lowerCAmelCase						)    ->    List[Any]:
							'''simple docstring'''






							if not self.legacy:
													lowercase        =							text.startswith(_lowerCAmelCase						)
													if is_first:
																			lowercase        =							text[1:]

							lowercase        =							self.sp_model.encode(_lowerCAmelCase ,				out_type=_lowerCAmelCase						)

							if not self.legacy and not is_first and not text.startswith(""" """						) and tokens[0].startswith(_lowerCAmelCase						):
													lowercase        =							([tokens[0][1:]] if len(tokens[0]						) > 1 else []) + tokens[1:]
							return tokens







	def   _a   (    self ,				_lowerCAmelCase						)    ->    int:
							'''simple docstring'''






							if token.startswith("""<extra_id_"""						):
													lowercase        =							re.match(r"""<extra_id_(\d+)>""" ,				_lowerCAmelCase						)
													lowercase        =							int(match.group(1						)						)
													return self.vocab_size - num - 1
							return self.sp_model.piece_to_id(_lowerCAmelCase						)







	def   _a   (    self ,				_lowerCAmelCase						)    ->    Optional[int]:
							'''simple docstring'''






							if index < self.sp_model.get_piece_size():
													lowercase        =							self.sp_model.IdToPiece(_lowerCAmelCase						)
							else:
													lowercase        =							F"""<extra_id_{self.vocab_size - 1 - index}>"""
							return token







	def   _a   (    self ,				_lowerCAmelCase						)    ->    List[str]:
							'''simple docstring'''






							lowercase        =							[]
							lowercase        =							""""""
							lowercase        =							False
							for token in tokens:
													# make sure that special tokens are not decoded using sentencepiece model
													if token in self.all_special_tokens:
																			if not prev_is_special:
																									out_string += " "
																			out_string += self.sp_model.decode(_lowerCAmelCase						) + token
																			lowercase        =							True
																			lowercase        =							[]
													else:
																			current_sub_tokens.append(_lowerCAmelCase						)
																			lowercase        =							False
							out_string += self.sp_model.decode(_lowerCAmelCase						)
							return out_string.strip()







	def   _a   (    self ,				_lowerCAmelCase ,				_lowerCAmelCase = None						)    ->    Tuple[str]:
							'''simple docstring'''






							if not os.path.isdir(_lowerCAmelCase						):
													logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""						)
													return
							lowercase        =							os.path.join(
							    _lowerCAmelCase ,				(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]						)

							if os.path.abspath(self.vocab_file						) != os.path.abspath(_lowerCAmelCase						) and os.path.isfile(self.vocab_file						):
													copyfile(self.vocab_file ,				_lowerCAmelCase						)
							elif not os.path.isfile(self.vocab_file						):
													with open(_lowerCAmelCase ,				"""wb"""						) as fi:
																			lowercase        =							self.sp_model.serialized_model_proto()
																			fi.write(_lowerCAmelCase						)

							return (out_vocab_file,)







'''simple docstring'''


import json
import logging
import os
import socket

import git
import numpy as np
import torch


logging.basicConfig(
    format='''%(asctime)s - %(levelname)s - %(name)s - PID: %(process)d -  %(message)s''',
    datefmt='''%m/%d/%Y %H:%M:%S''',
    level=logging.INFO,
)
lowercase_      :				Tuple   					=    logging.getLogger(__name__)







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      str						):
						lowercase        =							git.Repo(search_parent_directories=lowercase_						)
						lowercase        =							{
						    """repo_id""": str(lowercase_						),
						    """repo_sha""": str(repo.head.object.hexsha						),
						    """repo_branch""": str(repo.active_branch						),
						}

						with open(os.path.join(lowercase_ ,  """git_log.json"""						) ,  """w"""						) as f:
												json.dump(lowercase_ ,  lowercase_ ,  indent=4						)







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      str						):
						if params.n_gpu <= 0:
												lowercase        =							0
												lowercase        =							-1
												lowercase        =							True
												lowercase        =							False
												return

						assert torch.cuda.is_available()

						logger.info("""Initializing GPUs"""						)
						if params.n_gpu > 1:
												assert params.local_rank != -1

												lowercase        =							int(os.environ["""WORLD_SIZE"""]						)
												lowercase        =							int(os.environ["""N_GPU_NODE"""]						)
												lowercase        =							int(os.environ["""RANK"""]						)

												# number of nodes / node ID
												lowercase        =							params.world_size // params.n_gpu_per_node
												lowercase        =							params.global_rank // params.n_gpu_per_node
												lowercase        =							True

												assert params.n_nodes == int(os.environ["""N_NODES"""]						)
												assert params.node_id == int(os.environ["""NODE_RANK"""]						)

						# local job (single GPU)
						else:
												assert params.local_rank == -1

												lowercase        =							1
												lowercase        =							0
												lowercase        =							0
												lowercase        =							0
												lowercase        =							1
												lowercase        =							1
												lowercase        =							False

						# sanity checks
						assert params.n_nodes >= 1
						assert 0 <= params.node_id < params.n_nodes
						assert 0 <= params.local_rank <= params.global_rank < params.world_size
						assert params.world_size == params.n_nodes * params.n_gpu_per_node

						# define whether this is the master process / if we are in multi-node distributed mode
						lowercase        =							params.node_id == 0 and params.local_rank == 0
						lowercase        =							params.n_nodes > 1

						# summary
						lowercase        =							F"""--- Global rank: {params.global_rank} - """
						logger.info(PREFIX + """Number of nodes: %i""" % params.n_nodes						)
						logger.info(PREFIX + """Node ID        : %i""" % params.node_id						)
						logger.info(PREFIX + """Local rank     : %i""" % params.local_rank						)
						logger.info(PREFIX + """World size     : %i""" % params.world_size						)
						logger.info(PREFIX + """GPUs per node  : %i""" % params.n_gpu_per_node						)
						logger.info(PREFIX + """Master         : %s""" % str(params.is_master						)						)
						logger.info(PREFIX + """Multi-node     : %s""" % str(params.multi_node						)						)
						logger.info(PREFIX + """Multi-GPU      : %s""" % str(params.multi_gpu						)						)
						logger.info(PREFIX + """Hostname       : %s""" % socket.gethostname()						)

						# set GPU device
						torch.cuda.set_device(params.local_rank						)

						# initialize multi-GPU
						if params.multi_gpu:
												logger.info("""Initializing PyTorch distributed"""						)
												torch.distributed.init_process_group(
												    init_method="""env://""" ,  backend="""nccl""" ,  )







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      Optional[Any]						):
						np.random.seed(args.seed						)
						torch.manual_seed(args.seed						)
						if args.n_gpu > 0:
												torch.cuda.manual_seed_all(args.seed						)


