




"""simple docstring"""


import os
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


A_      =    logging.get_logger(__name__)

A_      =    '''▁'''

A_      =    {'''vocab_file''': '''sentencepiece.bpe.model''', '''monolingual_vocab_file''': '''dict.txt'''}

A_      =    {
    '''vocab_file''': {
        '''vinai/bartpho-syllable''': '''https://huggingface.co/vinai/bartpho-syllable/resolve/main/sentencepiece.bpe.model''',
    },
    '''monolingual_vocab_file''': {
        '''vinai/bartpho-syllable''': '''https://huggingface.co/vinai/bartpho-syllable/resolve/main/dict.txt''',
    },
}

A_      =    {'''vinai/bartpho-syllable''': 10_24}






class  lowercase(					__a			):







 '''simple docstring'''





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


 def __init__(    self:		int,				a_:		str,				a_:		Optional[int],				a_:		List[str]="<s>",				a_:		List[Any]="</s>",				a_:		Dict="</s>",				a_:		Union[str, Any]="<s>",				a_:		Optional[int]="<unk>",				a_:		Optional[Any]="<pad>",				a_:		str="<mask>",				a_:		Optional[Dict[str, Any]] = None,				**a_:		int,				):




   '''simple docstring'''



   _snake_case							:		Optional[Any]								=  AddedToken(a_,				lstrip=a_,				rstrip=a_		) if isinstance(a_,				a_		) else mask_token

   _snake_case							:		Optional[Any]								=  {} if sp_model_kwargs is None else sp_model_kwargs

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

   _snake_case							:		Any								=  vocab_file
   _snake_case							:		List[Any]								=  monolingual_vocab_file
   _snake_case							:		Union[str, Any]								=  spm.SentencePieceProcessor(**self.sp_model_kwargs		)
   self.sp_model.Load(str(a_		)		)

   # Load the reduced vocab

   # Keep order of special tokens for backward compatibility
   _snake_case							:		Tuple								=  {}
   _snake_case							:		Union[str, Any]								=  0
   for token in [bos_token, pad_token, eos_token, unk_token, sep_token, cls_token]:
     if str(a_		) not in self.fairseq_tokens_to_ids:
       _snake_case							:		Optional[int]								=  cnt
       cnt += 1
   with open(a_,				"""r""",				encoding="""utf-8"""		) as f:
     for line in f.readlines():
       _snake_case							:		Union[str, Any]								=  line.strip().split()[0]
       _snake_case							:		Tuple								=  len(self.fairseq_tokens_to_ids		)
   if str(a_		) not in self.fairseq_tokens_to_ids:
     _snake_case							:		str								=  len(self.fairseq_tokens_to_ids		)

   _snake_case							:		Tuple								=  {v: k for k, v in self.fairseq_tokens_to_ids.items()}


 def __getstate__(    self:		Optional[int]		):




   '''simple docstring'''



   _snake_case							:		Tuple								=  self.__dict__.copy()
   _snake_case							:		List[Any]								=  None
   _snake_case							:		Union[str, Any]								=  self.sp_model.serialized_model_proto()
   return state


 def __setstate__(    self:		Optional[Any],				a_:		List[str]		):




   '''simple docstring'''



   _snake_case							:		Optional[Any]								=  d

   # for backward compatibility
   if not hasattr(self,				"""sp_model_kwargs"""		):
     _snake_case							:		Optional[int]								=  {}

   _snake_case							:		Dict								=  spm.SentencePieceProcessor(**self.sp_model_kwargs		)
   self.sp_model.LoadFromSerializedProto(self.sp_model_proto		)


 def 			UpperCamelCase_       (    self:		Union[str, Any],				a_:		List[int],				a_:		Optional[List[int]] = None		):




   '''simple docstring'''



   if token_ids_a is None:
     return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
   _snake_case							:		str								=  [self.cls_token_id]
   _snake_case							:		Optional[Any]								=  [self.sep_token_id]
   return cls + token_ids_a + sep + sep + token_ids_a + sep


 def 			UpperCamelCase_       (    self:		Dict,				a_:		List[int],				a_:		Optional[List[int]] = None,				a_:		bool = False		):




   '''simple docstring'''



   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, 1] + ([0] * len(a_		)) + [1]


 def 			UpperCamelCase_       (    self:		Union[str, Any],				a_:		List[int],				a_:		Optional[List[int]] = None		):




   '''simple docstring'''



   _snake_case							:		Union[str, Any]								=  [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 			UpperCamelCase_       (    self:		int		):




   '''simple docstring'''



   return len(self.fairseq_ids_to_tokens		)


 def 			UpperCamelCase_       (    self:		Any		):




   '''simple docstring'''



   _snake_case							:		Tuple								=  {self.convert_ids_to_tokens(a_		): i for i in range(self.vocab_size		)}
   vocab.update(self.added_tokens_encoder		)
   return vocab


 def 			UpperCamelCase_       (    self:		Tuple,				a_:		str		):




   '''simple docstring'''



   return self.sp_model.encode(a_,				out_type=a_		)


 def 			UpperCamelCase_       (    self:		Optional[int],				a_:		List[str]		):




   '''simple docstring'''



   if token in self.fairseq_tokens_to_ids:
     return self.fairseq_tokens_to_ids[token]
   else:
     return self.unk_token_id


 def 			UpperCamelCase_       (    self:		List[str],				a_:		List[str]		):




   '''simple docstring'''



   return self.fairseq_ids_to_tokens[index]


 def 			UpperCamelCase_       (    self:		Tuple,				a_:		Tuple		):




   '''simple docstring'''



   _snake_case							:		Optional[int]								=  """""".join(a_		).replace(a_,				""" """		).strip()
   return out_string


 def 			UpperCamelCase_       (    self:		List[str],				a_:		str,				a_:		Optional[str] = None		):




   '''simple docstring'''



   if not os.path.isdir(a_		):
     logger.error(f"Vocabulary path ({save_directory}) should be a directory"		)
     return
   _snake_case							:		Any								=  os.path.join(
       a_,				(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]		)
   _snake_case							:		Any								=  os.path.join(
       a_,				(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""monolingual_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:
       _snake_case							:		Optional[Any]								=  self.sp_model.serialized_model_proto()
       fi.write(a_		)

   if os.path.abspath(self.monolingual_vocab_file		) != os.path.abspath(
       a_		) and os.path.isfile(self.monolingual_vocab_file		):
     copyfile(self.monolingual_vocab_file,				a_		)
   elif not os.path.isfile(self.monolingual_vocab_file		):
     with open(a_,				"""w""",				encoding="""utf-8"""		) as fp:
       for token in self.fairseq_tokens_to_ids:
         if token not in self.all_special_tokens:
           fp.write(f"{str(a_		)} \n"		)

   return out_vocab_file, out_monolingual_vocab_file








"""simple docstring"""


import torch
from torch import nn






class  lowercase(					nn.Module			):







 '''simple docstring'''





 def __init__(    self:		Any,				a_:		List[str],				a_:		Union[str, Any],				a_:		int,				a_:		int,				a_:		List[Any]=1,				a_:		Union[str, Any]=False		):




   '''simple docstring'''



   super().__init__()

   _snake_case							:		int								=  n_token
   _snake_case							:		Tuple								=  d_embed
   _snake_case							:		List[str]								=  d_proj

   _snake_case							:		Optional[int]								=  cutoffs + [n_token]
   _snake_case							:		Any								=  [0] + self.cutoffs
   _snake_case							:		Tuple								=  div_val

   _snake_case							:		Optional[int]								=  self.cutoffs[0]
   _snake_case							:		Union[str, Any]								=  len(self.cutoffs		) - 1
   _snake_case							:		Union[str, Any]								=  self.shortlist_size + self.n_clusters

   if self.n_clusters > 0:
     _snake_case							:		List[Any]								=  nn.Parameter(torch.zeros(self.n_clusters,				self.d_embed		)		)
     _snake_case							:		Tuple								=  nn.Parameter(torch.zeros(self.n_clusters		)		)

   _snake_case							:		Any								=  nn.ModuleList()
   _snake_case							:		Optional[Any]								=  nn.ParameterList()

   if div_val == 1:
     for i in range(len(self.cutoffs		)		):
       if d_proj != d_embed:
         self.out_projs.append(nn.Parameter(torch.FloatTensor(a_,				a_		)		)		)
       else:
         self.out_projs.append(a_		)

     self.out_layers.append(nn.Linear(a_,				a_		)		)
   else:
     for i in range(len(self.cutoffs		)		):
       _snake_case		, _snake_case							:		List[Any]								=  self.cutoff_ends[i], self.cutoff_ends[i + 1]
       _snake_case							:		Union[str, Any]								=  d_embed // (div_val**i)

       self.out_projs.append(nn.Parameter(torch.FloatTensor(a_,				a_		)		)		)

       self.out_layers.append(nn.Linear(a_,				r_idx - l_idx		)		)

   _snake_case							:		Optional[int]								=  keep_order


 def 			UpperCamelCase_       (    self:		str,				a_:		Union[str, Any],				a_:		Dict,				a_:		int,				a_:		Tuple		):




   '''simple docstring'''



   if proj is None:
     _snake_case							:		List[Any]								=  nn.functional.linear(a_,				a_,				bias=a_		)
   else:
     # if CUDA_MAJOR <= 9 and CUDA_MINOR <= 1:
     _snake_case							:		List[Any]								=  nn.functional.linear(a_,				proj.t().contiguous()		)
     _snake_case							:		Tuple								=  nn.functional.linear(a_,				a_,				bias=a_		)
     # else:
     #     logit = torch.einsum('bd,de,ev->bv', (hidden, proj, weight.t()))
     #     if bias is not None:
     #         logit = logit + bias

   return logit


 def 			UpperCamelCase_       (    self:		Dict,				a_:		Dict,				a_:		str=None,				a_:		Union[str, Any]=False		):




   '''simple docstring'''



   if labels is not None:
     # Shift so that tokens < n predict n
     _snake_case							:		int								=  hidden[..., :-1, :].contiguous()
     _snake_case							:		List[Any]								=  labels[..., 1:].contiguous()
     _snake_case							:		str								=  hidden.view(-1,				hidden.size(-1		)		)
     _snake_case							:		Dict								=  labels.view(-1		)
     if hidden.size(0		) != labels.size(0		):
       raise RuntimeError("""Input and labels should have the same size in the batch dimension."""		)
   else:
     _snake_case							:		int								=  hidden.view(-1,				hidden.size(-1		)		)

   if self.n_clusters == 0:
     _snake_case							:		Tuple								=  self._compute_logit(a_,				self.out_layers[0].weight,				self.out_layers[0].bias,				self.out_projs[0]		)
     if labels is not None:
       _snake_case							:		Dict								=  labels != -100
       _snake_case							:		str								=  torch.zeros_like(a_,				dtype=hidden.dtype,				device=hidden.device		)
       _snake_case							:		str								=  (
           -nn.functional.log_softmax(a_,				dim=-1		)[mask].gather(1,				labels[mask].unsqueeze(1		)		).squeeze(1		)
       )
     else:
       _snake_case							:		Optional[int]								=  nn.functional.log_softmax(a_,				dim=-1		)
   else:
     # construct weights and biases
     _snake_case		, _snake_case							:		Optional[int]								=  [], []
     for i in range(len(self.cutoffs		)		):
       if self.div_val == 1:
         _snake_case		, _snake_case							:		Optional[int]								=  self.cutoff_ends[i], self.cutoff_ends[i + 1]
         _snake_case							:		List[Any]								=  self.out_layers[0].weight[l_idx:r_idx]
         _snake_case							:		Tuple								=  self.out_layers[0].bias[l_idx:r_idx]
       else:
         _snake_case							:		Optional[int]								=  self.out_layers[i].weight
         _snake_case							:		int								=  self.out_layers[i].bias

       if i == 0:
         _snake_case							:		List[str]								=  torch.cat([weight_i, self.cluster_weight],				dim=0		)
         _snake_case							:		int								=  torch.cat([bias_i, self.cluster_bias],				dim=0		)

       weights.append(a_		)
       biases.append(a_		)

     _snake_case		, _snake_case		, _snake_case							:		Any								=  weights[0], biases[0], self.out_projs[0]

     _snake_case							:		List[str]								=  self._compute_logit(a_,				a_,				a_,				a_		)
     _snake_case							:		Union[str, Any]								=  nn.functional.log_softmax(a_,				dim=1		)

     if labels is None:
       _snake_case							:		Tuple								=  hidden.new_empty((head_logit.size(0		), self.n_token)		)
     else:
       _snake_case							:		Dict								=  torch.zeros_like(a_,				dtype=hidden.dtype,				device=hidden.device		)

     _snake_case							:		Union[str, Any]								=  0
     _snake_case							:		Optional[Any]								=  [0] + self.cutoffs
     for i in range(len(a_		) - 1		):
       _snake_case		, _snake_case							:		Dict								=  cutoff_values[i], cutoff_values[i + 1]

       if labels is not None:
         _snake_case							:		Dict								=  (labels >= l_idx) & (labels < r_idx)
         _snake_case							:		int								=  mask_i.nonzero().squeeze()

         if indices_i.numel() == 0:
           continue

         _snake_case							:		List[str]								=  labels.index_select(0,				a_		) - l_idx
         _snake_case							:		List[str]								=  head_logprob.index_select(0,				a_		)
         _snake_case							:		List[str]								=  hidden.index_select(0,				a_		)
       else:
         _snake_case							:		List[str]								=  hidden

       if i == 0:
         if labels is not None:
           _snake_case							:		Dict								=  head_logprob_i.gather(1,				target_i[:, None]		).squeeze(1		)
         else:
           _snake_case							:		Optional[Any]								=  head_logprob[:, : self.cutoffs[0]]
       else:
         _snake_case		, _snake_case		, _snake_case							:		Dict								=  weights[i], biases[i], self.out_projs[i]

         _snake_case							:		int								=  self._compute_logit(a_,				a_,				a_,				a_		)
         _snake_case							:		int								=  nn.functional.log_softmax(a_,				dim=1		)
         _snake_case							:		Dict								=  self.cutoffs[0] + i - 1  # No probability for the head cluster
         if labels is not None:
           _snake_case							:		Dict								=  head_logprob_i[:, cluster_prob_idx] + tail_logprob_i.gather(
               1,				target_i[:, None]		).squeeze(1		)
         else:
           _snake_case							:		Optional[Any]								=  head_logprob[:, cluster_prob_idx, None] + tail_logprob_i
           _snake_case							:		Any								=  logprob_i

       if labels is not None:
         if (hasattr(self,				"""keep_order"""		) and self.keep_order) or keep_order:
           out.index_copy_(0,				a_,				-logprob_i		)
         else:
           out[offset : offset + logprob_i.size(0		)].copy_(-logprob_i		)
         offset += logprob_i.size(0		)

   return out


 def 			UpperCamelCase_       (    self:		Union[str, Any],				a_:		Optional[int]		):




   '''simple docstring'''



   if self.n_clusters == 0:
     _snake_case							:		Optional[int]								=  self._compute_logit(a_,				self.out_layers[0].weight,				self.out_layers[0].bias,				self.out_projs[0]		)
     return nn.functional.log_softmax(a_,				dim=-1		)
   else:
     # construct weights and biases
     _snake_case		, _snake_case							:		List[Any]								=  [], []
     for i in range(len(self.cutoffs		)		):
       if self.div_val == 1:
         _snake_case		, _snake_case							:		Optional[Any]								=  self.cutoff_ends[i], self.cutoff_ends[i + 1]
         _snake_case							:		Optional[int]								=  self.out_layers[0].weight[l_idx:r_idx]
         _snake_case							:		str								=  self.out_layers[0].bias[l_idx:r_idx]
       else:
         _snake_case							:		List[Any]								=  self.out_layers[i].weight
         _snake_case							:		Union[str, Any]								=  self.out_layers[i].bias

       if i == 0:
         _snake_case							:		int								=  torch.cat([weight_i, self.cluster_weight],				dim=0		)
         _snake_case							:		Dict								=  torch.cat([bias_i, self.cluster_bias],				dim=0		)

       weights.append(a_		)
       biases.append(a_		)

     _snake_case		, _snake_case		, _snake_case							:		int								=  weights[0], biases[0], self.out_projs[0]
     _snake_case							:		List[Any]								=  self._compute_logit(a_,				a_,				a_,				a_		)

     _snake_case							:		List[Any]								=  hidden.new_empty((head_logit.size(0		), self.n_token)		)
     _snake_case							:		int								=  nn.functional.log_softmax(a_,				dim=1		)

     _snake_case							:		List[Any]								=  [0] + self.cutoffs
     for i in range(len(a_		) - 1		):
       _snake_case		, _snake_case							:		List[str]								=  cutoff_values[i], cutoff_values[i + 1]

       if i == 0:
         _snake_case							:		List[str]								=  head_logprob[:, : self.cutoffs[0]]
       else:
         _snake_case		, _snake_case		, _snake_case							:		List[Any]								=  weights[i], biases[i], self.out_projs[i]

         _snake_case							:		Optional[int]								=  self._compute_logit(a_,				a_,				a_,				a_		)
         _snake_case							:		Any								=  nn.functional.log_softmax(a_,				dim=1		)

         _snake_case							:		Dict								=  head_logprob[:, -i] + tail_logprob_i
         _snake_case							:		Any								=  logprob_i

     return out


