from typing import Dict

import numpy as np

from ..utils import add_end_docstrings, is_tf_available, is_torch_available, logging
from .base import PIPELINE_INIT_ARGS, GenericTensor, Pipeline, PipelineException


if is_tf_available():
	import tensorflow as tf

	from ..tf_utils import stable_softmax


if is_torch_available():
	import torch


_UpperCAmelCase						=		logging.get_logger(__name__)





@add_end_docstrings(
    __SCREAMING_SNAKE_CASE					,       R'''\n        top_k (`int`, defaults to 5):\n            The number of predictions to return.\n        targets (`str` or `List[str]`, *optional*):\n            When passed, the model will limit the scores to the passed targets instead of looking up in the whole\n            vocab. If the provided targets are not in the model vocab, they will be tokenized and the first resulting\n            token will be used (with a warning, and that might be slower).\n\n    '''					,       )
class UpperCAmelCase						(					__SCREAMING_SNAKE_CASE  ):


				'''simple docstring'''






				def   lowerCAmelCase_   (						self		,	lowercase					):



						"""simple docstring"""





						if self.framework == "tf":
								A_		:				List[Any]     =      tf.where(input_ids == self.tokenizer.mask_token_id					).numpy()
						elif self.framework == "pt":
								A_		:				Dict     =      torch.nonzero(input_ids == self.tokenizer.mask_token_id		,	as_tuple=lowercase					)
						else:
								raise ValueError('Unsupported framework'					)
						return masked_index
				def   lowerCAmelCase_   (						self		,	lowercase					):



						"""simple docstring"""





						A_		:				Tuple     =      self.get_masked_index(lowercase					)
						A_		:				Any     =      np.prod(masked_index.shape					)
						if numel < 1:
								raise PipelineException(
								    'fill-mask'		,	self.model.base_model_prefix		,	F'''No mask_token ({self.tokenizer.mask_token}) found on the input'''		,	)
				def   lowerCAmelCase_   (						self		,	lowercase					):



						"""simple docstring"""





						if isinstance(lowercase		,	lowercase					):
								for model_input in model_inputs:
										self._ensure_exactly_one_mask_token(model_input['input_ids'][0]					)
						else:
								for input_ids in model_inputs["input_ids"]:
										self._ensure_exactly_one_mask_token(lowercase					)
				def   lowerCAmelCase_   (						self		,	lowercase		,	lowercase=None		,	**lowercase					):



						"""simple docstring"""





						if return_tensors is None:
								A_		:				Union[str, Any]     =      self.framework
						A_		:				Tuple     =      self.tokenizer(lowercase		,	return_tensors=lowercase					)
						self.ensure_exactly_one_mask_token(lowercase					)
						return model_inputs
				def   lowerCAmelCase_   (						self		,	lowercase					):



						"""simple docstring"""





						A_		:				str     =      self.model(**lowercase					)
						A_		:				Any     =      model_inputs['input_ids']
						return model_outputs
				def   lowerCAmelCase_   (						self		,	lowercase		,	lowercase=5		,	lowercase=None					):



						"""simple docstring"""





						if target_ids is not None and target_ids.shape[0] < top_k:
								A_		:				Optional[Any]     =      target_ids.shape[0]
						A_		:				Any     =      model_outputs['input_ids'][0]
						A_		:				List[Any]     =      model_outputs['logits']

						if self.framework == "tf":
								A_		:				Union[str, Any]     =      tf.where(input_ids == self.tokenizer.mask_token_id					).numpy()[:, 0]

								A_		:				Union[str, Any]     =      outputs.numpy()

								A_		:				int     =      outputs[0, masked_index, :]
								A_		:				Optional[int]     =      stable_softmax(lowercase		,	axis=-1					)
								if target_ids is not None:
										A_		:				Optional[Any]     =      tf.gather_nd(tf.squeeze(lowercase		,	0					)		,	target_ids.reshape(-1		,	1					)					)
										A_		:				List[str]     =      tf.expand_dims(lowercase		,	0					)

								A_		:				Any     =      tf.math.top_k(lowercase		,	k=lowercase					)
								A_   ,			A_		:				int     =      topk.values.numpy(), topk.indices.numpy()
						else:
								A_		:				Tuple     =      torch.nonzero(input_ids == self.tokenizer.mask_token_id		,	as_tuple=lowercase					).squeeze(-1					)
								# Fill mask pipeline supports only one ${mask_token} per sample

								A_		:				str     =      outputs[0, masked_index, :]
								A_		:				str     =      logits.softmax(dim=-1					)
								if target_ids is not None:
										A_		:				List[Any]     =      probs[..., target_ids]

								A_   ,			A_		:				List[str]     =      probs.topk(lowercase					)

						A_		:				int     =      []
						A_		:				Union[str, Any]     =      values.shape[0] == 1
						for i, (_values, _predictions) in enumerate(zip(values.tolist()		,	predictions.tolist()					)					):
								A_		:				Any     =      []
								for v, p in zip(_values		,	_predictions					):
										# Copy is important since we're going to modify this array in place
										A_		:				int     =      input_ids.numpy().copy()
										if target_ids is not None:
												A_		:				Tuple     =      target_ids[p].tolist()

										A_		:				Any     =      p
										# Filter padding out:
										A_		:				str     =      tokens[np.where(tokens != self.tokenizer.pad_token_id					)]
										# Originally we skip special tokens to give readable output.
										# For multi masks though, the other [MASK] would be removed otherwise
										# making the output look odd, so we add them back
										A_		:				List[str]     =      self.tokenizer.decode(lowercase		,	skip_special_tokens=lowercase					)
										A_		:				Tuple     =      {'score': v, 'token': p, 'token_str': self.tokenizer.decode([p]					), 'sequence': sequence}
										row.append(lowercase					)
								result.append(lowercase					)
						if single_mask:
								return result[0]
						return result
				def   lowerCAmelCase_   (						self		,	lowercase		,	lowercase=None					):



						"""simple docstring"""





						if isinstance(lowercase		,	lowercase					):
								A_		:				str     =      [targets]
						try:
								A_		:				Union[str, Any]     =      self.tokenizer.get_vocab()
						except Exception:
								A_		:				str     =      {}
						A_		:				int     =      []
						for target in targets:
								A_		:				Union[str, Any]     =      vocab.get(lowercase		,	lowercase					)
								if id_ is None:
										A_		:				Dict     =      self.tokenizer(
										    lowercase		,	add_special_tokens=lowercase		,	return_attention_mask=lowercase		,	return_token_type_ids=lowercase		,	max_length=1		,	truncation=lowercase		,	)['input_ids']
										if len(lowercase					) == 0:
												logger.warning(
												    F'''The specified target token `{target}` does not exist in the model vocabulary. '''
												    'We cannot replace it with anything meaningful, ignoring it'					)
												continue
										A_		:				int     =      input_ids[0]
										# XXX: If users encounter this pass
										# it becomes pretty slow, so let's make sure
										# The warning enables them to fix the input to
										# get faster performance.
										logger.warning(
										    F'''The specified target token `{target}` does not exist in the model vocabulary. '''
										    F'''Replacing with `{self.tokenizer.convert_ids_to_tokens(id_					)}`.'''					)
								target_ids.append(id_					)
						A_		:				str     =      list(set(lowercase					)					)
						if len(lowercase					) == 0:
								raise ValueError('At least one target must be provided when passed.'					)
						A_		:				Optional[int]     =      np.array(lowercase					)
						return target_ids
				def   lowerCAmelCase_   (						self		,	lowercase=None		,	lowercase=None					):



						"""simple docstring"""





						A_		:				Union[str, Any]     =      {}

						if targets is not None:
								A_		:				Optional[int]     =      self.get_target_ids(lowercase		,	lowercase					)
								A_		:				List[str]     =      target_ids

						if top_k is not None:
								A_		:				Union[str, Any]     =      top_k

						if self.tokenizer.mask_token_id is None:
								raise PipelineException(
								    'fill-mask'		,	self.model.base_model_prefix		,	'The tokenizer does not define a `mask_token`.'					)
						return {}, {}, postprocess_params







				def __call__(						self		,	lowercase		,	*lowercase		,	**lowercase					):



						"""simple docstring"""





						A_		:				Optional[Any]     =      super().__call__(lowercase		,	**lowercase					)
						if isinstance(lowercase		,	lowercase					) and len(lowercase					) == 1:
								return outputs[0]
						return outputs








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

import sentencepiece as spm

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


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

SCREAMING_SNAKE_CASE							:int							 =				{'vocab_file': 'spiece.model'}

SCREAMING_SNAKE_CASE							:Union[str, Any]							 =				{
    'vocab_file': {
        'google/bigbird-roberta-base': 'https://huggingface.co/google/bigbird-roberta-base/resolve/main/spiece.model',
        'google/bigbird-roberta-large': (
            'https://huggingface.co/google/bigbird-roberta-large/resolve/main/spiece.model'
        ),
        'google/bigbird-base-trivia-itc': (
            'https://huggingface.co/google/bigbird-base-trivia-itc/resolve/main/spiece.model'
        ),
    }
}

SCREAMING_SNAKE_CASE							:int							 =				{
    'google/bigbird-roberta-base': 4096,
    'google/bigbird-roberta-large': 4096,
    'google/bigbird-base-trivia-itc': 4096,
}





class 							UpperCAmelCase      (  __SCREAMING_SNAKE_CASE  ):




	'''simple docstring'''





	snake_case_					    =     VOCAB_FILES_NAMES
	snake_case_					    =     PRETRAINED_VOCAB_FILES_MAP
	snake_case_					    =     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	snake_case_					    =     ["input_ids", "attention_mask"]
	snake_case_					    =     []






	def __init__(       self					: Any							,A					: List[str]							,A					: str="<unk>"							,A					: int="<s>"							,A					: Union[str, Any]="</s>"							,A					: List[str]="<pad>"							,A					: int="[SEP]"							,A					: Optional[Any]="[MASK]"							,A					: Tuple="[CLS]"							,A					: Optional[Dict[str, Any]] = None							,**A					: Any							,):
					__A   			= AddedToken(A							,lstrip=A							,rstrip=A ) if isinstance(A							,A ) else bos_token
					__A   			= AddedToken(A							,lstrip=A							,rstrip=A ) if isinstance(A							,A ) else eos_token
					__A   			= AddedToken(A							,lstrip=A							,rstrip=A ) if isinstance(A							,A ) else unk_token
					__A   			= AddedToken(A							,lstrip=A							,rstrip=A ) if isinstance(A							,A ) else pad_token
					__A   			= AddedToken(A							,lstrip=A							,rstrip=A ) if isinstance(A							,A ) else cls_token
					__A   			= AddedToken(A							,lstrip=A							,rstrip=A ) if isinstance(A							,A ) else sep_token

					# Mask token behave like a normal word, i.e. include the space before it
					__A   			= AddedToken(A							,lstrip=A							,rstrip=A ) if isinstance(A							,A ) else mask_token

					__A   			= {} if sp_model_kwargs is None else sp_model_kwargs

					super().__init__(
					    bos_token=A							,eos_token=A							,unk_token=A							,pad_token=A							,sep_token=A							,mask_token=A							,cls_token=A							,sp_model_kwargs=self.sp_model_kwargs							,**A							,)

					__A   			= vocab_file

					__A   			= spm.SentencePieceProcessor(**self.sp_model_kwargs )
					self.sp_model.Load(A )






	@property
	def 				UpperCamelCase_  (       self					: List[str] ):
					return self.sp_model.get_piece_size()






	def 				UpperCamelCase_  (       self					: Optional[Any] ):
					__A   			= {self.convert_ids_to_tokens(A ): i for i in range(self.vocab_size )}
					vocab.update(self.added_tokens_encoder )
					return vocab






	def __getstate__(       self					: Optional[int] ):
					__A   			= self.__dict__.copy()
					__A   			= None
					return state






	def __setstate__(       self					: str							,A					: Optional[Any] ):
					__A   			= d

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

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






	def 				UpperCamelCase_  (       self					: Any							,A					: str ):
					return self.sp_model.encode(A							,out_type=A )






	def 				UpperCamelCase_  (       self					: List[str]							,A					: Tuple ):
					return self.sp_model.piece_to_id(A )






	def 				UpperCamelCase_  (       self					: List[Any]							,A					: Tuple ):
					__A   			= self.sp_model.IdToPiece(A )
					return token






	def 				UpperCamelCase_  (       self					: List[Any]							,A					: int ):
					__A   			= []
					__A   			= ""
					__A   			= 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(A ) + token
													__A   			= True
													__A   			= []
									else:
													current_sub_tokens.append(A )
													__A   			= False
					out_string += self.sp_model.decode(A )
					return out_string.strip()






	def 				UpperCamelCase_  (       self					: Tuple							,A					: List[int]							,A					: bool = False							,A					: bool = None							,A					: bool = True							,**A					: Union[str, Any]							,):
					__A   			= kwargs.pop("use_source_tokenizer"							,A )

					__A   			= self.convert_ids_to_tokens(A							,skip_special_tokens=A )

					# To avoid mixing byte-level and unicode for byte-level BPT
					# we need to build string separately for added tokens and byte-level tokens
					# cf. https://github.com/huggingface/transformers/issues/1133
					__A   			= []
					__A   			= []
					for token in filtered_tokens:
									if skip_special_tokens and token in self.all_special_ids:
													continue
									if token in self.added_tokens_encoder:
													if current_sub_text:
																	sub_texts.append(self.convert_tokens_to_string(A ) )
																	__A   			= []
													sub_texts.append(A )
									else:
													current_sub_text.append(A )
					if current_sub_text:
									sub_texts.append(self.convert_tokens_to_string(A ) )

					# Mimic the behavior of the Rust tokenizer:
					# No space before [MASK] and [SEP]
					if spaces_between_special_tokens:
									__A   			= re.sub(R" (\[(MASK|SEP)\])"							,R"\1"							," ".join(A ) )
					else:
									__A   			= "".join(A )

					__A   			= (
					    clean_up_tokenization_spaces
					    if clean_up_tokenization_spaces is not None
					    else self.clean_up_tokenization_spaces
					)
					if clean_up_tokenization_spaces:
									__A   			= self.clean_up_tokenization(A )
									return clean_text
					else:
									return text






	def 				UpperCamelCase_  (       self					: str							,A					: str							,A					: Optional[str] = None ):
					if not os.path.isdir(A ):
									logger.error(f'''Vocabulary path ({save_directory}) should be a directory''' )
									return
					__A   			= os.path.join(
					    A							,(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"] )

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

					return (out_vocab_file,)






	def 				UpperCamelCase_  (       self					: Dict							,A					: List[int]							,A					: Optional[List[int]] = None ):
					if token_ids_a is None:
									return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
					__A   			= [self.cls_token_id]
					__A   			= [self.sep_token_id]
					return cls + token_ids_a + sep + token_ids_a + sep






	def 				UpperCamelCase_  (       self					: Optional[int]							,A					: List[int]							,A					: Optional[List[int]] = None							,A					: bool = False ):
					if already_has_special_tokens:
									return super().get_special_tokens_mask(
									    token_ids_a=A							,token_ids_a=A							,already_has_special_tokens=A )

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







	def 				UpperCamelCase_  (       self					: Any							,A					: List[int]							,A					: Optional[List[int]] = None ):
					__A   			= [self.sep_token_id]
					__A   			= [self.cls_token_id]
					if token_ids_a is None:
									return len(cls + token_ids_a + sep ) * [0]
					return len(cls + token_ids_a + sep ) * [0] + len(token_ids_a + sep ) * [1]
