
'''simple docstring'''





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


__lowerCAmelCase					:			Tuple			      =		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 						__lowerCAmelCase	(	SCREAMING_SNAKE_CASE_      ):



      """simple docstring"""







      def 		snake_case_			(      self		:       Optional[int]				,       _snake_case		:       Dict				):
          if self.framework == "tf":
              __lowercase		:			List[str]	    =       tf.where(input_ids == self.tokenizer.mask_token_id				).numpy()
          elif self.framework == "pt":
              __lowercase		:			Any	    =       torch.nonzero(input_ids == self.tokenizer.mask_token_id				,       as_tuple=UpperCamelCase__				)
          else:
              raise ValueError('''Unsupported framework'''				)
          return masked_index




      def 		snake_case_			(      self		:       Tuple				,       _snake_case		:       int				):
          __lowercase		:			Dict	    =       self.get_masked_index(UpperCamelCase__				)
          __lowercase		:			str	    =       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 		snake_case_			(      self		:       Any				,       _snake_case		:       Dict				):
          if isinstance(UpperCamelCase__				,       UpperCamelCase__				):
              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(UpperCamelCase__				)




      def 		snake_case_			(      self		:       Optional[Any]				,       _snake_case		:       Tuple				,       _snake_case		:       Optional[int]=None				,       **_snake_case		:       Union[str, Any]				):
          if return_tensors is None:
              __lowercase		:			str	    =       self.framework
          __lowercase		:			str	    =       self.tokenizer(UpperCamelCase__				,       return_tensors=UpperCamelCase__				)
          self.ensure_exactly_one_mask_token(UpperCamelCase__				)
          return model_inputs




      def 		snake_case_			(      self		:       Any				,       _snake_case		:       Optional[Any]				):
          __lowercase		:			Tuple	    =       self.model(**UpperCamelCase__				)
          __lowercase		:			List[str]	    =       model_inputs['''input_ids''']
          return model_outputs




      def 		snake_case_			(      self		:       Union[str, Any]				,       _snake_case		:       Any				,       _snake_case		:       Optional[Any]=5				,       _snake_case		:       Dict=None				):
          # Cap top_k if there are targets
          if target_ids is not None and target_ids.shape[0] < top_k:
              __lowercase		:			Union[str, Any]	    =       target_ids.shape[0]
          __lowercase		:			str	    =       model_outputs['''input_ids'''][0]
          __lowercase		:			Tuple	    =       model_outputs['''logits''']

          if self.framework == "tf":
              __lowercase		:			Tuple	    =       tf.where(input_ids == self.tokenizer.mask_token_id				).numpy()[:, 0]

              __lowercase		:			Dict	    =       outputs.numpy()

              __lowercase		:			Tuple	    =       outputs[0, masked_index, :]
              __lowercase		:			str	    =       stable_softmax(UpperCamelCase__				,       axis=-1				)
              if target_ids is not None:
                  __lowercase		:			int	    =       tf.gather_nd(tf.squeeze(UpperCamelCase__				,       0				)				,       target_ids.reshape(-1				,       1				)				)
                  __lowercase		:			str	    =       tf.expand_dims(UpperCamelCase__				,       0				)

              __lowercase		:			str	    =       tf.math.top_k(UpperCamelCase__				,       k=UpperCamelCase__				)
              __lowercase		:			List[str]	    =       topk.values.numpy(), topk.indices.numpy()
          else:
              __lowercase		:			Optional[Any]	    =       torch.nonzero(input_ids == self.tokenizer.mask_token_id				,       as_tuple=UpperCamelCase__				).squeeze(-1				)
              # Fill mask pipeline supports only one ${mask_token} per sample

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

              __lowercase		:			Tuple	    =       probs.topk(UpperCamelCase__				)

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

                  __lowercase		:			List[str]	    =       p
                  # Filter padding out:
                  __lowercase		:			Dict	    =       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
                  __lowercase		:			Any	    =       self.tokenizer.decode(UpperCamelCase__				,       skip_special_tokens=UpperCamelCase__				)
                  __lowercase		:			str	    =       {'''score''': v, '''token''': p, '''token_str''': self.tokenizer.decode([p]				), '''sequence''': sequence}
                  row.append(UpperCamelCase__				)
              result.append(UpperCamelCase__				)
          if single_mask:
              return result[0]
          return result




      def 		snake_case_			(      self		:       Union[str, Any]				,       _snake_case		:       Optional[Any]				,       _snake_case		:       Optional[Any]=None				):
          if isinstance(UpperCamelCase__				,       UpperCamelCase__				):
              __lowercase		:			Dict	    =       [targets]
          try:
              __lowercase		:			Any	    =       self.tokenizer.get_vocab()
          except Exception:
              __lowercase		:			Tuple	    =       {}
          __lowercase		:			str	    =       []
          for target in targets:
              __lowercase		:			int	    =       vocab.get(UpperCamelCase__				,       UpperCamelCase__				)
              if id_ is None:
                  __lowercase		:			Tuple	    =       self.tokenizer(
                      UpperCamelCase__				,       add_special_tokens=UpperCamelCase__				,       return_attention_mask=UpperCamelCase__				,       return_token_type_ids=UpperCamelCase__				,       max_length=1				,       truncation=UpperCamelCase__				,       )['''input_ids''']
                  if len(UpperCamelCase__				) == 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
                  __lowercase		:			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_				)
          __lowercase		:			str	    =       list(set(UpperCamelCase__				)				)
          if len(UpperCamelCase__				) == 0:
              raise ValueError('''At least one target must be provided when passed.'''				)
          __lowercase		:			Optional[int]	    =       np.array(UpperCamelCase__				)
          return target_ids




      def 		snake_case_			(      self		:       int				,       _snake_case		:       Optional[Any]=None				,       _snake_case		:       List[Any]=None				):
          __lowercase		:			Tuple	    =       {}

          if targets is not None:
              __lowercase		:			Any	    =       self.get_target_ids(UpperCamelCase__				,       UpperCamelCase__				)
              __lowercase		:			List[Any]	    =       target_ids

          if top_k is not None:
              __lowercase		:			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		:       Dict				,       _snake_case		:       List[Any]				,       *_snake_case		:       str				,       **_snake_case		:       Optional[int]				):
          __lowercase		:			Any	    =       super().__call__(UpperCamelCase__				,       **UpperCamelCase__				)
          if isinstance(UpperCamelCase__				,       UpperCamelCase__				) and len(UpperCamelCase__				) == 1:
              return outputs[0]
          return outputs




import json
import os
from functools import lru_cache
from typing import Dict, List, Optional, Tuple, Union

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...tokenization_utils_base import BatchEncoding, EncodedInput
from ...utils import PaddingStrategy, logging


__lowerCAmelCase					:			Tuple			      =		logging.get_logger(__name__)


__lowerCAmelCase					:			int			      =		{"vocab_file": "vocab.json", "merges_file": "merges.txt"}

# See all LED models at https://huggingface.co/models?filter=LED
__lowerCAmelCase					:			List[Any]			      =		{
    "vocab_file": {
        "allenai/led-base-16384": "https://huggingface.co/allenai/led-base-16384/resolve/main/vocab.json",
    },
    "merges_file": {
        "allenai/led-base-16384": "https://huggingface.co/allenai/led-base-16384/resolve/main/merges.txt",
    },
    "tokenizer_file": {
        "allenai/led-base-16384": "https://huggingface.co/allenai/led-base-16384/resolve/main/tokenizer.json",
    },
}

__lowerCAmelCase					:			List[Any]			      =		{
    "allenai/led-base-16384": 16_384,
}


@lru_cache()
# Copied from transformers.models.bart.tokenization_bart.bytes_to_unicode
def     UpperCAmelCase_  (						) ->						List[Any]:
    __lowercase		:			Union[str, Any]	    =       (
        list(range(ord('''!'''	)	,					ord('''~'''	) + 1	)	) + list(range(ord('''¡'''	)	,					ord('''¬'''	) + 1	)	) + list(range(ord('''®'''	)	,					ord('''ÿ'''	) + 1	)	)
    )
    __lowercase		:			Any	    =       bs[:]
    __lowercase		:			Union[str, Any]	    =       0
    for b in range(2**8	):
        if b not in bs:
            bs.append(__lowerCAmelCase	)
            cs.append(2**8 + n	)
            n += 1
    __lowercase		:			List[str]	    =       [chr(__lowerCAmelCase	) for n in cs]
    return dict(zip(__lowerCAmelCase	,					__lowerCAmelCase	)	)


def     UpperCAmelCase_  (						__lowerCAmelCase	) ->						List[str]:
    __lowercase		:			Optional[int]	    =       set()
    __lowercase		:			Dict	    =       word[0]
    for char in word[1:]:
        pairs.add((prev_char, char)	)
        __lowercase		:			List[Any]	    =       char
    return pairs





class 						__lowerCAmelCase	(	lowerCAmelCase_      ):



      """simple docstring"""







      A__	:			str        =    VOCAB_FILES_NAMES
      A__	:			Any        =    PRETRAINED_VOCAB_FILES_MAP
      A__	:			int        =    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      A__	:			Union[str, Any]        =    ['''input_ids''', '''attention_mask''']




      def __init__(      self		:       Dict				,       _snake_case		:       List[Any]				,       _snake_case		:       Dict				,       _snake_case		:       int="replace"				,       _snake_case		:       Optional[int]="<s>"				,       _snake_case		:       List[Any]="</s>"				,       _snake_case		:       str="</s>"				,       _snake_case		:       Optional[int]="<s>"				,       _snake_case		:       List[Any]="<unk>"				,       _snake_case		:       Optional[Any]="<pad>"				,       _snake_case		:       List[Any]="<mask>"				,       _snake_case		:       Optional[Any]=False				,       **_snake_case		:       Dict				,       ):
          __lowercase		:			str	    =       AddedToken(_snake_case				,       lstrip=_snake_case				,       rstrip=_snake_case				) if isinstance(_snake_case				,       _snake_case				) else bos_token
          __lowercase		:			List[Any]	    =       AddedToken(_snake_case				,       lstrip=_snake_case				,       rstrip=_snake_case				) if isinstance(_snake_case				,       _snake_case				) else eos_token
          __lowercase		:			Union[str, Any]	    =       AddedToken(_snake_case				,       lstrip=_snake_case				,       rstrip=_snake_case				) if isinstance(_snake_case				,       _snake_case				) else sep_token
          __lowercase		:			List[str]	    =       AddedToken(_snake_case				,       lstrip=_snake_case				,       rstrip=_snake_case				) if isinstance(_snake_case				,       _snake_case				) else cls_token
          __lowercase		:			Any	    =       AddedToken(_snake_case				,       lstrip=_snake_case				,       rstrip=_snake_case				) if isinstance(_snake_case				,       _snake_case				) else unk_token
          __lowercase		:			Optional[Any]	    =       AddedToken(_snake_case				,       lstrip=_snake_case				,       rstrip=_snake_case				) if isinstance(_snake_case				,       _snake_case				) else pad_token

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

          super().__init__(
              errors=_snake_case				,       bos_token=_snake_case				,       eos_token=_snake_case				,       unk_token=_snake_case				,       sep_token=_snake_case				,       cls_token=_snake_case				,       pad_token=_snake_case				,       mask_token=_snake_case				,       add_prefix_space=_snake_case				,       **_snake_case				,       )

          with open(_snake_case				,       encoding='''utf-8'''				) as vocab_handle:
              __lowercase		:			List[str]	    =       json.load(_snake_case				)
          __lowercase		:			Optional[Any]	    =       {v: k for k, v in self.encoder.items()}
          __lowercase		:			Dict	    =       errors  # how to handle errors in decoding
          __lowercase		:			str	    =       bytes_to_unicode()
          __lowercase		:			Dict	    =       {v: k for k, v in self.byte_encoder.items()}
          with open(_snake_case				,       encoding='''utf-8'''				) as merges_handle:
              __lowercase		:			Any	    =       merges_handle.read().split('''\n'''				)[1:-1]
          __lowercase		:			Tuple	    =       [tuple(merge.split()				) for merge in bpe_merges]
          __lowercase		:			Dict	    =       dict(zip(_snake_case				,       range(len(_snake_case				)				)				)				)
          __lowercase		:			Dict	    =       {}
          __lowercase		:			str	    =       add_prefix_space

          # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
          __lowercase		:			str	    =       re.compile(r'''\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'''				)




      @property
      # Copied from transformers.models.bart.tokenization_bart.BartTokenizer.vocab_size
      def 		snake_case_			(      self		:       Dict				):
          return len(self.encoder				)




      def 		snake_case_			(      self		:       List[str]				):
          return dict(self.encoder				,       **self.added_tokens_encoder				)




      def 		snake_case_			(      self		:       Tuple				,       _snake_case		:       str				):
          if token in self.cache:
              return self.cache[token]
          __lowercase		:			Optional[Any]	    =       tuple(_snake_case				)
          __lowercase		:			str	    =       get_pairs(_snake_case				)

          if not pairs:
              return token

          while True:
              __lowercase		:			Dict	    =       min(_snake_case				,       key=lambda _snake_case				: self.bpe_ranks.get(_snake_case				,       float('''inf'''				)				)				)
              if bigram not in self.bpe_ranks:
                  break
              __lowercase    ,      __lowercase		:			Tuple	    =       bigram
              __lowercase		:			Dict	    =       []
              __lowercase		:			int	    =       0
              while i < len(_snake_case				):
                  try:
                      __lowercase		:			List[Any]	    =       word.index(_snake_case				,       _snake_case				)
                  except ValueError:
                      new_word.extend(word[i:]				)
                      break
                  else:
                      new_word.extend(word[i:j]				)
                      __lowercase		:			Optional[int]	    =       j

                  if word[i] == first and i < len(_snake_case				) - 1 and word[i + 1] == second:
                      new_word.append(first + second				)
                      i += 2
                  else:
                      new_word.append(word[i]				)
                      i += 1
              __lowercase		:			str	    =       tuple(_snake_case				)
              __lowercase		:			Tuple	    =       new_word
              if len(_snake_case				) == 1:
                  break
              else:
                  __lowercase		:			Any	    =       get_pairs(_snake_case				)
          __lowercase		:			Union[str, Any]	    =       ''' '''.join(_snake_case				)
          __lowercase		:			Union[str, Any]	    =       word
          return word




      def 		snake_case_			(      self		:       Tuple				,       _snake_case		:       List[Any]				):
          __lowercase		:			Optional[int]	    =       []
          for token in re.findall(self.pat				,       _snake_case				):
              __lowercase		:			Dict	    =       ''''''.join(
                  self.byte_encoder[b] for b in token.encode('''utf-8'''				)				)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
              bpe_tokens.extend(bpe_token for bpe_token in self.bpe(_snake_case				).split(''' '''				)				)
          return bpe_tokens




      def 		snake_case_			(      self		:       Union[str, Any]				,       _snake_case		:       List[Any]				):
          return self.encoder.get(_snake_case				,       self.encoder.get(self.unk_token				)				)




      def 		snake_case_			(      self		:       Dict				,       _snake_case		:       List[Any]				):
          return self.decoder.get(_snake_case				)




      def 		snake_case_			(      self		:       int				,       _snake_case		:       Dict				):
          __lowercase		:			List[Any]	    =       ''''''.join(_snake_case				)
          __lowercase		:			Any	    =       bytearray([self.byte_decoder[c] for c in text]				).decode('''utf-8'''				,       errors=self.errors				)
          return text




      def 		snake_case_			(      self		:       Optional[Any]				,       _snake_case		:       str				,       _snake_case		:       Optional[str] = None				):
          if not os.path.isdir(_snake_case				):
              logger.error(F'Vocabulary path ({save_directory}) should be a directory'				)
              return
          __lowercase		:			Union[str, Any]	    =       os.path.join(
              _snake_case				,       (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']				)
          __lowercase		:			str	    =       os.path.join(
              _snake_case				,       (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']				)

          with open(_snake_case				,       '''w'''				,       encoding='''utf-8'''				) as f:
              f.write(json.dumps(self.encoder				,       indent=2				,       sort_keys=_snake_case				,       ensure_ascii=_snake_case				) + '''\n'''				)

          __lowercase		:			List[str]	    =       0
          with open(_snake_case				,       '''w'''				,       encoding='''utf-8'''				) as writer:
              writer.write('''#version: 0.2\n'''				)
              for bpe_tokens, token_index in sorted(self.bpe_ranks.items()				,       key=lambda _snake_case				: kv[1]				):
                  if index != token_index:
                      logger.warning(
                          F'Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'
                          ''' Please check that the tokenizer is not corrupted!'''				)
                      __lowercase		:			List[str]	    =       token_index
                  writer.write(''' '''.join(_snake_case				) + '''\n'''				)
                  index += 1

          return vocab_file, merge_file




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




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

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




      def 		snake_case_			(      self		:       Union[str, Any]				,       _snake_case		:       List[int]				,       _snake_case		:       Optional[List[int]] = None				):
          __lowercase		:			Dict	    =       [self.sep_token_id]
          __lowercase		:			List[str]	    =       [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 + sep + token_ids_a + sep				) * [0]




      def 		snake_case_			(      self		:       List[str]				,       _snake_case		:       Union[str, Any]				,       _snake_case		:       Optional[int]=False				,       **_snake_case		:       List[Any]				):
          __lowercase		:			int	    =       kwargs.pop('''add_prefix_space'''				,       self.add_prefix_space				)
          if (is_split_into_words or add_prefix_space) and (len(_snake_case				) > 0 and not text[0].isspace()):
              __lowercase		:			Optional[Any]	    =       ''' ''' + text
          return (text, kwargs)




      def 		snake_case_			(      self		:       Tuple				,       _snake_case		:       Union[Dict[str, EncodedInput], BatchEncoding]				,       _snake_case		:       Optional[int] = None				,       _snake_case		:       PaddingStrategy = PaddingStrategy.DO_NOT_PAD				,       _snake_case		:       Optional[int] = None				,       _snake_case		:       Optional[bool] = None				,       ):
          __lowercase		:			Optional[Any]	    =       super()._pad(
              encoded_inputs=_snake_case				,       max_length=_snake_case				,       padding_strategy=_snake_case				,       pad_to_multiple_of=_snake_case				,       return_attention_mask=_snake_case				,       )

          # Load from model defaults
          if return_attention_mask is None:
              __lowercase		:			List[str]	    =       '''attention_mask''' in self.model_input_names

          if return_attention_mask and "global_attention_mask" in encoded_inputs:
              __lowercase		:			Dict	    =       encoded_inputs[self.model_input_names[0]]
              # `global_attention_mask` need to have the same length as other (sequential) inputs.
              __lowercase		:			Tuple	    =       len(encoded_inputs['''global_attention_mask''']				) != len(_snake_case				)

              if needs_to_be_padded:
                  __lowercase		:			List[Any]	    =       len(_snake_case				) - len(encoded_inputs['''global_attention_mask''']				)

                  if self.padding_side == "right":
                      # Use `-1` since `0` in `global_attention_mask` means `local attention` instead of `not to attend`
                      __lowercase		:			Union[str, Any]	    =       (
                          encoded_inputs['''global_attention_mask'''] + [-1] * difference
                      )
                  elif self.padding_side == "left":
                      __lowercase		:			Union[str, Any]	    =       [-1] * difference + encoded_inputs[
                          '''global_attention_mask'''
                      ]
                  else:
                      raise ValueError('''Invalid padding strategy:''' + str(self.padding_side				)				)

          return encoded_inputs


