



"""simple docstring"""

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

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


lowercase__					:						Tuple												=			logging.get_logger(__name__)

lowercase__					:						Optional[int]												=			{
    '''vocab_file''': '''vocab.txt''',
    '''merges_file''': '''bpe.codes''',
}

lowercase__					:						Union[str, Any]												=			{
    '''vocab_file''': {
        '''vinai/phobert-base''': '''https://huggingface.co/vinai/phobert-base/resolve/main/vocab.txt''',
        '''vinai/phobert-large''': '''https://huggingface.co/vinai/phobert-large/resolve/main/vocab.txt''',
    },
    '''merges_file''': {
        '''vinai/phobert-base''': '''https://huggingface.co/vinai/phobert-base/resolve/main/bpe.codes''',
        '''vinai/phobert-large''': '''https://huggingface.co/vinai/phobert-large/resolve/main/bpe.codes''',
    },
}

lowercase__					:						List[str]												=			{
    '''vinai/phobert-base''': 2_56,
    '''vinai/phobert-large''': 2_56,
}




def __lowercase    (							_a					):
    snake_case_				:  List[str]       =			set()
    snake_case_				:  Any       =			word[0]
    for char in word[1:]:
        pairs.add((prev_char, char)					)
        snake_case_				:  List[str]       =			char

    snake_case_				:  Any       =			set(_a					)
    return pairs






class  _UpperCAmelCase (    lowerCAmelCase__):
 _lowerCAmelCase     :    Dict	      =	VOCAB_FILES_NAMES
 _lowerCAmelCase     :    int	      =	PRETRAINED_VOCAB_FILES_MAP
 _lowerCAmelCase     :    int	      =	PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES

 def __init__(  self	: str						,  lowercase_	: str						,  lowercase_	: int						,  lowercase_	: List[str]="<s>"						,  lowercase_	: List[str]="</s>"						,  lowercase_	: Any="</s>"						,  lowercase_	: str="<s>"						,  lowercase_	: int="<unk>"						,  lowercase_	: Tuple="<pad>"						,  lowercase_	: Dict="<mask>"						,  **lowercase_	: Any						,  ):
     super().__init__(
         bos_token=lowercase_						,  eos_token=lowercase_						,  unk_token=lowercase_						,  sep_token=lowercase_						,  cls_token=lowercase_						,  pad_token=lowercase_						,  mask_token=lowercase_						,  **lowercase_						,  )

     snake_case_				:  Optional[int]       =			vocab_file
     snake_case_				:  int       =			merges_file

     snake_case_				:  Tuple       =			{}
     snake_case_				:  Tuple       =			0
     snake_case_				:  Dict       =			1
     snake_case_				:  int       =			2
     snake_case_				:  Union[str, Any]       =			3

     self.add_from_file(lowercase_       )

     snake_case_				:  List[Any]       =			{v: k for k, v in self.encoder.items()}

     with open(lowercase_						,  encoding='''utf-8'''       ) as merges_handle:
         snake_case_				:  List[Any]       =			merges_handle.read().split('''\n'''       )[:-1]
     snake_case_				:  str       =			[tuple(merge.split()[:-1]       ) for merge in merges]
     snake_case_				:  Optional[Any]       =			dict(zip(lowercase_						,  range(len(lowercase_       )       )       )       )
     snake_case_				:  List[Any]       =			{}

 def 		_snake_case      (  self	: Any						,  lowercase_	: List[int]						,  lowercase_	: Optional[List[int]] = None       ):

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

 def 		_snake_case      (  self	: Optional[Any]						,  lowercase_	: List[int]						,  lowercase_	: Optional[List[int]] = None						,  lowercase_	: bool = False       ):

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

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

 def 		_snake_case      (  self	: str						,  lowercase_	: List[int]						,  lowercase_	: Optional[List[int]] = None       ):
     snake_case_				:  str       =			[self.sep_token_id]
     snake_case_				:  Optional[int]       =			[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]

 @property
 def 		_snake_case      (  self	: Any       ):
     return len(self.encoder       )

 def 		_snake_case      (  self	: str       ):
     return dict(self.encoder						,  **self.added_tokens_encoder       )

 def 		_snake_case      (  self	: Any						,  lowercase_	: Dict       ):
     if token in self.cache:
         return self.cache[token]
     snake_case_				:  Dict       =			tuple(lowercase_       )
     snake_case_				:  int       =			tuple(list(word[:-1]       ) + [word[-1] + '''</w>''']       )
     snake_case_				:  Dict       =			get_pairs(lowercase_       )

     if not pairs:
         return token

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

             if word[i] == first and i < len(lowercase_       ) - 1 and word[i + 1] == second:
                 new_word.append(first + second       )
                 i += 2
             else:
                 new_word.append(word[i]       )
                 i += 1
         snake_case_				:  Optional[int]       =			tuple(lowercase_       )
         snake_case_				:  Tuple       =			new_word
         if len(lowercase_       ) == 1:
             break
         else:
             snake_case_				:  Optional[Any]       =			get_pairs(lowercase_       )
     snake_case_				:  Dict       =			'''@@ '''.join(lowercase_       )
     snake_case_				:  Tuple       =			word[:-4]
     snake_case_				:  Optional[int]       =			word
     return word

 def 		_snake_case      (  self	: Union[str, Any]						,  lowercase_	: Union[str, Any]       ):
     snake_case_				:  List[str]       =			[]

     snake_case_				:  List[Any]       =			re.findall(r'''\S+\n?'''						,  lowercase_       )

     for token in words:
         split_tokens.extend(list(self.bpe(lowercase_       ).split(''' '''       )       )       )
     return split_tokens

 def 		_snake_case      (  self	: int						,  lowercase_	: Optional[int]       ):
     return self.encoder.get(lowercase_						,  self.encoder.get(self.unk_token       )       )

 def 		_snake_case      (  self	: Any						,  lowercase_	: Optional[int]       ):
     return self.decoder.get(lowercase_						,  self.unk_token       )

 def 		_snake_case      (  self	: str						,  lowercase_	: Any       ):
     snake_case_				:  Optional[int]       =			''' '''.join(lowercase_       ).replace('''@@ '''						,  ''''''       ).strip()
     return out_string

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

     if os.path.abspath(self.vocab_file       ) != os.path.abspath(lowercase_       ):
         copyfile(self.vocab_file						,  lowercase_       )

     if os.path.abspath(self.merges_file       ) != os.path.abspath(lowercase_       ):
         copyfile(self.merges_file						,  lowercase_       )

     return out_vocab_file, out_merge_file






 def 		_snake_case      (  self	: Any						,  lowercase_	: List[Any]       ):
     if isinstance(lowercase_						,  lowercase_       ):
         try:
             with open(lowercase_						,  '''r'''						,  encoding='''utf-8'''       ) as fd:
                 self.add_from_file(lowercase_       )
         except FileNotFoundError as fnfe:
             raise fnfe
         except UnicodeError:
             raise Exception(f"Incorrect encoding detected in {f}, please rebuild the dataset"       )
         return

     snake_case_				:  List[Any]       =			f.readlines()
     for lineTmp in lines:
         snake_case_				:  Dict       =			lineTmp.strip()
         snake_case_				:  Tuple       =			line.rfind(''' '''       )
         if idx == -1:
             raise ValueError('''Incorrect dictionary format, expected \'<token> <cnt>\''''       )
         snake_case_				:  int       =			line[:idx]
         snake_case_				:  str       =			len(self.encoder       )










"""simple docstring"""

import tensorflow as tf

from ...tf_utils import shape_list




class  _UpperCAmelCase (    tf.keras.layers.Layer):

 def __init__(  self	: int						,  lowercase_	: Tuple						,  lowercase_	: Union[str, Any]						,  lowercase_	: Tuple						,  lowercase_	: List[Any]						,  lowercase_	: Tuple=1						,  lowercase_	: List[str]=False						,  **lowercase_	: Optional[Any]       ):
     super().__init__(**lowercase_       )

     snake_case_				:  int       =			vocab_size
     snake_case_				:  Union[str, Any]       =			d_embed
     snake_case_				:  Optional[int]       =			d_proj

     snake_case_				:  int       =			cutoffs + [vocab_size]
     snake_case_				:  Optional[int]       =			[0] + self.cutoffs
     snake_case_				:  List[str]       =			div_val

     snake_case_				:  int       =			self.cutoffs[0]
     snake_case_				:  Optional[Any]       =			len(self.cutoffs       ) - 1
     snake_case_				:  Any       =			self.shortlist_size + self.n_clusters
     snake_case_				:  Dict       =			keep_order

     snake_case_				:  Tuple       =			[]
     snake_case_				:  Optional[Any]       =			[]

 def 		_snake_case      (  self	: Dict						,  lowercase_	: int       ):
     if self.n_clusters > 0:
         snake_case_				:  Tuple       =			self.add_weight(
             shape=(self.n_clusters, self.d_embed)						,  initializer='''zeros'''						,  trainable=lowercase_						,  name='''cluster_weight'''       )
         snake_case_				:  List[str]       =			self.add_weight(
             shape=(self.n_clusters,)						,  initializer='''zeros'''						,  trainable=lowercase_						,  name='''cluster_bias'''       )

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

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

 @staticmethod
 def 		_snake_case      (  lowercase_	: Dict						,  lowercase_	: Union[str, Any]						,  lowercase_	: Optional[int]						,  lowercase_	: str=None       ):
     snake_case_				:  str       =			x
     if proj is not None:
         snake_case_				:  str       =			tf.einsum('''ibd,ed->ibe'''						,  lowercase_						,  lowercase_       )
     return tf.einsum('''ibd,nd->ibn'''						,  lowercase_						,  lowercase_       ) + b

 @staticmethod
 def 		_snake_case      (  lowercase_	: Tuple						,  lowercase_	: Union[str, Any]       ):
     snake_case_				:  Optional[int]       =			shape_list(lowercase_       )
     snake_case_				:  Any       =			tf.range(lp_size[0]						,  dtype=target.dtype       )
     snake_case_				:  Optional[int]       =			tf.stack([r, target]						,  1       )
     return tf.gather_nd(lowercase_						,  lowercase_       )






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

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

             if i == 0:
                 snake_case_				:  Tuple       =			tf.concat([cur_W, self.cluster_weight]						,  0       )
                 snake_case_				:  List[str]       =			tf.concat([cur_b, self.cluster_bias]						,  0       )

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

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

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

     return out





