




'''simple docstring'''



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

import regex as re

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


UpperCamelCase_    	=				logging.get_logger(__name__)


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

# See all BART models at https://huggingface.co/models?filter=bart
UpperCamelCase_    	=				{
    "vocab_file": {
        "facebook/bart-base": "https://huggingface.co/facebook/bart-base/resolve/main/vocab.json",
        "facebook/bart-large": "https://huggingface.co/facebook/bart-large/resolve/main/vocab.json",
        "facebook/bart-large-mnli": "https://huggingface.co/facebook/bart-large-mnli/resolve/main/vocab.json",
        "facebook/bart-large-cnn": "https://huggingface.co/facebook/bart-large-cnn/resolve/main/vocab.json",
        "facebook/bart-large-xsum": "https://huggingface.co/facebook/bart-large-xsum/resolve/main/vocab.json",
        "yjernite/bart_eli5": "https://huggingface.co/yjernite/bart_eli5/resolve/main/vocab.json",
    },
    "merges_file": {
        "facebook/bart-base": "https://huggingface.co/facebook/bart-base/resolve/main/merges.txt",
        "facebook/bart-large": "https://huggingface.co/facebook/bart-large/resolve/main/merges.txt",
        "facebook/bart-large-mnli": "https://huggingface.co/facebook/bart-large-mnli/resolve/main/merges.txt",
        "facebook/bart-large-cnn": "https://huggingface.co/facebook/bart-large-cnn/resolve/main/merges.txt",
        "facebook/bart-large-xsum": "https://huggingface.co/facebook/bart-large-xsum/resolve/main/merges.txt",
        "yjernite/bart_eli5": "https://huggingface.co/yjernite/bart_eli5/resolve/main/merges.txt",
    },
}

UpperCamelCase_    	=				{
    "facebook/bart-base": 1_0_2_4,
    "facebook/bart-large": 1_0_2_4,
    "facebook/bart-large-mnli": 1_0_2_4,
    "facebook/bart-large-cnn": 1_0_2_4,
    "facebook/bart-large-xsum": 1_0_2_4,
    "yjernite/bart_eli5": 1_0_2_4,
}




@lru_cache()
def   lowercase__(			):

      """simple docstring"""






      SCREAMING_SNAKE_CASE   :      int    					=						(
          list(range(ord('!'    )      ,ord('~'    ) + 1    )    ) + list(range(ord('¡'    )      ,ord('¬'    ) + 1    )    ) + list(range(ord('®'    )      ,ord('ÿ'    ) + 1    )    )
      )
      SCREAMING_SNAKE_CASE   :      int    					=						bs[:]
      SCREAMING_SNAKE_CASE   :      Union[str, Any]    					=						0
      for b in range(2**8    ):
            if b not in bs:
                  bs.append(__UpperCamelCase    )
                  cs.append(2**8 + n    )
                  n += 1
      SCREAMING_SNAKE_CASE   :      Optional[int]    					=						[chr(__UpperCamelCase    ) for n in cs]
      return dict(zip(__UpperCamelCase      ,__UpperCamelCase    )    )




def   lowercase__(			__UpperCamelCase:					Union[str, Any]    ):

      """simple docstring"""






      SCREAMING_SNAKE_CASE   :      List[Any]    					=						set()
      SCREAMING_SNAKE_CASE   :      Dict    					=						word[0]
      for char in word[1:]:
            pairs.add((prev_char, char)    )
            SCREAMING_SNAKE_CASE   :      Optional[int]    					=						char
      return pairs






class 							_a				(			lowerCamelCase__ ):





       '''simple docstring'''






       A						:				Tuple        =						VOCAB_FILES_NAMES
       A						:				Optional[int]        =						PRETRAINED_VOCAB_FILES_MAP
       A						:				int        =						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
       A						:				str        =						["""input_ids""", """attention_mask"""]







       def __init__(      self,							A,							A,							A="replace",							A="<s>",							A="</s>",							A="</s>",							A="<s>",							A="<unk>",							A="<pad>",							A="<mask>",							A=False,							**A,							):







             '''simple docstring'''







             SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						AddedToken(snake_case__,							lstrip=snake_case__,							rstrip=snake_case__  ) if isinstance(snake_case__,							snake_case__  ) else bos_token
             SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						AddedToken(snake_case__,							lstrip=snake_case__,							rstrip=snake_case__  ) if isinstance(snake_case__,							snake_case__  ) else eos_token
             SCREAMING_SNAKE_CASE   :      List[Any]    					=						AddedToken(snake_case__,							lstrip=snake_case__,							rstrip=snake_case__  ) if isinstance(snake_case__,							snake_case__  ) else sep_token
             SCREAMING_SNAKE_CASE   :      Optional[int]    					=						AddedToken(snake_case__,							lstrip=snake_case__,							rstrip=snake_case__  ) if isinstance(snake_case__,							snake_case__  ) else cls_token
             SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						AddedToken(snake_case__,							lstrip=snake_case__,							rstrip=snake_case__  ) if isinstance(snake_case__,							snake_case__  ) else unk_token
             SCREAMING_SNAKE_CASE   :      Tuple    					=						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
             SCREAMING_SNAKE_CASE   :      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:
                   SCREAMING_SNAKE_CASE   :      Union[str, Any]    					=						json.load(snake_case__  )
             SCREAMING_SNAKE_CASE   :      Optional[int]    					=						{v: k for k, v in self.encoder.items()}
             SCREAMING_SNAKE_CASE   :      int    					=						errors  # how to handle errors in decoding
             SCREAMING_SNAKE_CASE   :      List[str]    					=						bytes_to_unicode()
             SCREAMING_SNAKE_CASE   :      str    					=						{v: k for k, v in self.byte_encoder.items()}
             with open(snake_case__,							encoding='utf-8'  ) as merges_handle:
                   SCREAMING_SNAKE_CASE   :      Any    					=						merges_handle.read().split('\n'  )[1:-1]
             SCREAMING_SNAKE_CASE   :      Any    					=						[tuple(merge.split()  ) for merge in bpe_merges]
             SCREAMING_SNAKE_CASE   :      Tuple    					=						dict(zip(snake_case__,							range(len(snake_case__  )  )  )  )
             SCREAMING_SNAKE_CASE   :      Dict    					=						{}
             SCREAMING_SNAKE_CASE   :      Any    					=						add_prefix_space

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







       @property
       def        UpperCamelCase_							(      self  ):







             '''simple docstring'''







             return len(self.encoder  )







       def        UpperCamelCase_							(      self  ):







             '''simple docstring'''







             return dict(self.encoder,							**self.added_tokens_encoder  )







       def        UpperCamelCase_							(      self,							A  ):







             '''simple docstring'''







             if token in self.cache:
                   return self.cache[token]
             SCREAMING_SNAKE_CASE   :      Optional[int]    					=						tuple(snake_case__  )
             SCREAMING_SNAKE_CASE   :      List[str]    					=						get_pairs(snake_case__  )

             if not pairs:
                   return token

             while True:
                   SCREAMING_SNAKE_CASE   :      Dict    					=						min(snake_case__,							key=lambda A  : self.bpe_ranks.get(snake_case__,							float('inf'  )  )  )
                   if bigram not in self.bpe_ranks:
                         break
                   SCREAMING_SNAKE_CASE   :      int    					=						bigram
                   SCREAMING_SNAKE_CASE   :      Dict    					=						[]
                   SCREAMING_SNAKE_CASE   :      str    					=						0
                   while i < len(snake_case__  ):
                         try:
                               SCREAMING_SNAKE_CASE   :      List[Any]    					=						word.index(snake_case__,							snake_case__  )
                         except ValueError:
                               new_word.extend(word[i:]  )
                               break
                         else:
                               new_word.extend(word[i:j]  )
                               SCREAMING_SNAKE_CASE   :      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
                   SCREAMING_SNAKE_CASE   :      Dict    					=						tuple(snake_case__  )
                   SCREAMING_SNAKE_CASE   :      List[Any]    					=						new_word
                   if len(snake_case__  ) == 1:
                         break
                   else:
                         SCREAMING_SNAKE_CASE   :      Tuple    					=						get_pairs(snake_case__  )
             SCREAMING_SNAKE_CASE   :      Optional[int]    					=						""" """.join(snake_case__  )
             SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						word
             return word







       def        UpperCamelCase_							(      self,							A  ):







             '''simple docstring'''







             SCREAMING_SNAKE_CASE   :      List[str]    					=						[]
             for token in re.findall(self.pat,							snake_case__  ):
                   SCREAMING_SNAKE_CASE   :      Tuple    					=						"""""".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        UpperCamelCase_							(      self,							A  ):







             '''simple docstring'''







             return self.encoder.get(snake_case__,							self.encoder.get(self.unk_token  )  )







       def        UpperCamelCase_							(      self,							A  ):







             '''simple docstring'''







             return self.decoder.get(snake_case__  )







       def        UpperCamelCase_							(      self,							A  ):







             '''simple docstring'''







             SCREAMING_SNAKE_CASE   :      List[str]    					=						"""""".join(snake_case__  )
             SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						bytearray([self.byte_decoder[c] for c in text]  ).decode('utf-8',							errors=self.errors  )
             return text







       def        UpperCamelCase_							(      self,							A,							A = None  ):







             '''simple docstring'''







             if not os.path.isdir(snake_case__  ):
                   logger.error(F"Vocabulary path ({save_directory}) should be a directory"  )
                   return
             SCREAMING_SNAKE_CASE   :      Optional[int]    					=						os.path.join(
                 snake_case__,							(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']  )
             SCREAMING_SNAKE_CASE   :      int    					=						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'  )

             SCREAMING_SNAKE_CASE   :      List[Any]    					=						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 A  : 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!'  )
                               SCREAMING_SNAKE_CASE   :      List[Any]    					=						token_index
                         writer.write(' '.join(snake_case__  ) + '\n'  )
                         index += 1

             return vocab_file, merge_file







       def        UpperCamelCase_							(      self,							A,							A = None  ):







             '''simple docstring'''







             if token_ids_a is None:
                   return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
             SCREAMING_SNAKE_CASE   :      Optional[int]    					=						[self.cls_token_id]
             SCREAMING_SNAKE_CASE   :      List[Any]    					=						[self.sep_token_id]
             return cls + token_ids_a + sep + sep + token_ids_a + sep







       def        UpperCamelCase_							(      self,							A,							A = None,							A = False  ):







             '''simple docstring'''







             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        UpperCamelCase_							(      self,							A,							A = None  ):







             '''simple docstring'''







             SCREAMING_SNAKE_CASE   :      List[str]    					=						[self.sep_token_id]
             SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						[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        UpperCamelCase_							(      self,							A,							A=False,							**A  ):







             '''simple docstring'''







             SCREAMING_SNAKE_CASE   :      Tuple    					=						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()):
                   SCREAMING_SNAKE_CASE   :      Tuple    					=						""" """ + text
             return (text, kwargs)









import tensorflow as tf

from ...tf_utils import shape_list
class     UpperCamelCase__					(     tf.keras.layers.Layer				):






		'''simple docstring'''






		def __init__(							self,   snake_case__,   snake_case__,   snake_case__,   snake_case__,   snake_case__=1,   snake_case__=False,   **snake_case__  )       -> Optional[int]:






									"""simple docstring"""
									super().__init__(**snake_case__  )

									lowercase_	:   Tuple  				=     vocab_size
									lowercase_	:   Union[str, Any]  				=     d_embed
									lowercase_	:   Optional[Any]  				=     d_proj

									lowercase_	:   Optional[Any]  				=     cutoffs + [vocab_size]
									lowercase_	:   Optional[int]  				=     [0] + self.cutoffs
									lowercase_	:   Union[str, Any]  				=     div_val

									lowercase_	:   Union[str, Any]  				=     self.cutoffs[0]
									lowercase_	:   List[str]  				=     len(self.cutoffs  ) - 1
									lowercase_	:   List[str]  				=     self.shortlist_size + self.n_clusters
									lowercase_	:   Tuple  				=     keep_order

									lowercase_	:   Dict  				=     []
									lowercase_	:   int  				=     []






		def 						snake_case__				(							self,   snake_case__  )       -> Optional[int]:






									"""simple docstring"""
									if self.n_clusters > 0:
																lowercase_	:   List[str]  				=     self.add_weight(
																    shape=(self.n_clusters, self.d_embed),   initializer="""zeros""",   trainable=snake_case__,   name="""cluster_weight"""  )
																lowercase_	:   List[Any]  				=     self.add_weight(
																    shape=(self.n_clusters,),   initializer="""zeros""",   trainable=snake_case__,   name="""cluster_bias"""  )

									if self.div_val == 1:
																for i in range(len(self.cutoffs  )  ):
																							if self.d_proj != self.d_embed:
																														lowercase_	:   Any  				=     self.add_weight(
																														    shape=(self.d_embed, self.d_proj),   initializer="""zeros""",   trainable=snake_case__,   name=f"""out_projs_._{i}""",   )
																														self.out_projs.append(snake_case__  )
																							else:
																														self.out_projs.append(snake_case__  )
																							lowercase_	:   List[str]  				=     self.add_weight(
																							    shape=(self.vocab_size, self.d_embed),   initializer="""zeros""",   trainable=snake_case__,   name=f"""out_layers_._{i}_._weight""",   )
																							lowercase_	:   List[str]  				=     self.add_weight(
																							    shape=(self.vocab_size,),   initializer="""zeros""",   trainable=snake_case__,   name=f"""out_layers_._{i}_._bias""",   )
																							self.out_layers.append((weight, bias)  )
									else:
																for i in range(len(self.cutoffs  )  ):
																							lowercase_				,		lowercase_	:   Optional[Any]  				=     self.cutoff_ends[i], self.cutoff_ends[i + 1]
																							lowercase_	:   Union[str, Any]  				=     self.d_embed // (self.div_val**i)

																							lowercase_	:   int  				=     self.add_weight(
																							    shape=(d_emb_i, self.d_proj),   initializer="""zeros""",   trainable=snake_case__,   name=f"""out_projs_._{i}"""  )
																							self.out_projs.append(snake_case__  )
																							lowercase_	:   Any  				=     self.add_weight(
																							    shape=(r_idx - l_idx, d_emb_i),   initializer="""zeros""",   trainable=snake_case__,   name=f"""out_layers_._{i}_._weight""",   )
																							lowercase_	:   Tuple  				=     self.add_weight(
																							    shape=(r_idx - l_idx,),   initializer="""zeros""",   trainable=snake_case__,   name=f"""out_layers_._{i}_._bias""",   )
																							self.out_layers.append((weight, bias)  )
									super().build(snake_case__  )






		@staticmethod
		def 						snake_case__				(							snake_case__,   snake_case__,   snake_case__,   snake_case__=None  )       -> Optional[Any]:






									"""simple docstring"""
									lowercase_	:   Dict  				=     x
									if proj is not None:
																lowercase_	:   List[Any]  				=     tf.einsum("""ibd,ed->ibe""",   snake_case__,   snake_case__  )
									return tf.einsum("""ibd,nd->ibn""",   snake_case__,   snake_case__  ) + b






		@staticmethod
		def 						snake_case__				(							snake_case__,   snake_case__  )       -> List[str]:






									"""simple docstring"""
									lowercase_	:   Optional[int]  				=     shape_list(snake_case__  )
									lowercase_	:   Optional[Any]  				=     tf.range(lp_size[0],   dtype=target.dtype  )
									lowercase_	:   str  				=     tf.stack([r, target],   1  )
									return tf.gather_nd(snake_case__,   snake_case__  )






		def 						snake_case__				(							self,   snake_case__,   snake_case__,   snake_case__=True,   snake_case__=False  )       -> Dict:






									"""simple docstring"""
									lowercase_	:   Any  				=     0
									if self.n_clusters == 0:
																lowercase_	:   Union[str, Any]  				=     self._logit(snake_case__,   self.out_layers[0][0],   self.out_layers[0][1],   self.out_projs[0]  )
																if target is not None:
																							lowercase_	:   str  				=     tf.nn.sparse_softmax_cross_entropy_with_logits(labels=snake_case__,   logits=snake_case__  )
																lowercase_	:   Optional[int]  				=     tf.nn.log_softmax(snake_case__,   axis=-1  )
									else:
																lowercase_	:   int  				=     shape_list(snake_case__  )
																lowercase_	:   List[Any]  				=     []
																lowercase_	:   Tuple  				=     tf.zeros(hidden_sizes[:2]  )
																for i in range(len(self.cutoffs  )  ):
																							lowercase_				,		lowercase_	:   Optional[int]  				=     self.cutoff_ends[i], self.cutoff_ends[i + 1]
																							if target is not None:
																														lowercase_	:   Tuple  				=     (target >= l_idx) & (target < r_idx)
																														lowercase_	:   Union[str, Any]  				=     tf.where(snake_case__  )
																														lowercase_	:   Optional[Any]  				=     tf.boolean_mask(snake_case__,   snake_case__  ) - l_idx

																							if self.div_val == 1:
																														lowercase_	:   Optional[Any]  				=     self.out_layers[0][0][l_idx:r_idx]
																														lowercase_	:   Union[str, Any]  				=     self.out_layers[0][1][l_idx:r_idx]
																							else:
																														lowercase_	:   Dict  				=     self.out_layers[i][0]
																														lowercase_	:   int  				=     self.out_layers[i][1]

																							if i == 0:
																														lowercase_	:   Optional[int]  				=     tf.concat([cur_W, self.cluster_weight],   0  )
																														lowercase_	:   Optional[Any]  				=     tf.concat([cur_b, self.cluster_bias],   0  )

																														lowercase_	:   List[str]  				=     self._logit(snake_case__,   snake_case__,   snake_case__,   self.out_projs[0]  )
																														lowercase_	:   List[str]  				=     tf.nn.log_softmax(snake_case__  )
																														out.append(head_logprob[..., : self.cutoffs[0]]  )
																														if target is not None:
																																					lowercase_	:   int  				=     tf.boolean_mask(snake_case__,   snake_case__  )
																																					lowercase_	:   Optional[int]  				=     self._gather_logprob(snake_case__,   snake_case__  )
																							else:
																														lowercase_	:   List[str]  				=     self._logit(snake_case__,   snake_case__,   snake_case__,   self.out_projs[i]  )
																														lowercase_	:   Dict  				=     tf.nn.log_softmax(snake_case__  )
																														lowercase_	:   Optional[int]  				=     self.cutoffs[0] + i - 1  # No probability for the head cluster
																														lowercase_	:   int  				=     head_logprob[..., cluster_prob_idx, None] + tail_logprob
																														out.append(snake_case__  )
																														if target is not None:
																																					lowercase_	:   Optional[Any]  				=     tf.boolean_mask(snake_case__,   snake_case__  )
																																					lowercase_	:   Optional[int]  				=     tf.boolean_mask(snake_case__,   snake_case__  )
																																					lowercase_	:   Union[str, Any]  				=     self._gather_logprob(snake_case__,   snake_case__  )
																																					cur_logprob += cur_head_logprob[:, self.cutoff_ends[1] + i - 1]
																							if target is not None:
																														loss += tf.scatter_nd(snake_case__,   -cur_logprob,   shape_list(snake_case__  )  )
																lowercase_	:   List[str]  				=     tf.concat(snake_case__,   axis=-1  )

									if target is not None:
																if return_mean:
																							lowercase_	:   Tuple  				=     tf.reduce_mean(snake_case__  )
																# Add the training-time loss value to the layer using `self.add_loss()`.
																self.add_loss(snake_case__  )

																# Log the loss as a metric (we could log arbitrary metrics,
																# including different metrics for training and inference.
																self.add_metric(snake_case__,   name=self.name,   aggregation="""mean""" if return_mean else """"""  )

									return out