





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

import sentencepiece as spm

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


__UpperCAmelCase					:     Dict	      =	'▁'

__UpperCAmelCase					:     Optional[Any]	      =	{'vocab_file': 'spiece.model'}

__UpperCAmelCase					:     Optional[int]	      =	{
    'vocab_file': {'google/pegasus-xsum': 'https://huggingface.co/google/pegasus-xsum/resolve/main/spiece.model'}
}

__UpperCAmelCase					:     List[str]	      =	{
    'google/pegasus-xsum': 512,
}


__UpperCAmelCase					:     Any	      =	logging.get_logger(__name__)


class 			lowerCamelCase						(		SCREAMING_SNAKE_CASE__  ):
  UpperCAmelCase		:       Dict										=     VOCAB_FILES_NAMES

  UpperCAmelCase		:       Union[str, Any]										=     VOCAB_FILES_NAMES
  UpperCAmelCase		:       Dict										=     PRETRAINED_VOCAB_FILES_MAP
  UpperCAmelCase		:       List[Any]										=     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  UpperCAmelCase		:       Any										=     ['input_ids', 'attention_mask']







  def __init__(     self						:   str  ,					__snake_case						:   Any  ,					__snake_case						:   Dict="<pad>"  ,					__snake_case						:   Optional[Any]="</s>"  ,					__snake_case						:   Optional[int]="<unk>"  ,					__snake_case						:   int="<mask_2>"  ,					__snake_case						:   Union[str, Any]="<mask_1>"  ,					__snake_case						:   List[str]=None  ,					__snake_case						:   Tuple=103  ,					__snake_case						:   Dict = None  ,					**__snake_case						:   int  ,					)    ->		List[str]:
        _a							:  Dict				= offset
        if additional_special_tokens is not None:
              if not isinstance(__lowerCamelCase  ,					__lowerCamelCase       ):
                    raise TypeError(
                        f"""additional_special_tokens should be of type {type(__lowerCamelCase       )}, but is"""
                        f""" {type(__lowerCamelCase       )}"""       )

              _a							:  Dict				= (
                  ([mask_token_sent] + additional_special_tokens)
                  if mask_token_sent not in additional_special_tokens and mask_token_sent is not None
                  else additional_special_tokens
              )
              # fill additional tokens with ..., <unk_token_102> in case not all additional tokens are already taken
              additional_special_tokens_extended += [
                  f"""<unk_{i}>""" for i in range(len(__lowerCamelCase       )  ,					self.offset - 1       )
              ]

              if len(set(__lowerCamelCase       )       ) != len(__lowerCamelCase       ):
                    raise ValueError(
                        '''Please make sure that the provided additional_special_tokens do not contain an incorrectly'''
                        f""" shifted list of <unk_x> tokens. Found {additional_special_tokens_extended}."""       )
              _a							:  Tuple				= additional_special_tokens_extended
        else:
              _a							:  Any				= [mask_token_sent] if mask_token_sent is not None else []
              additional_special_tokens += [f"""<unk_{i}>""" for i in range(2  ,					self.offset       )]

        _a							:  Dict				= {} if sp_model_kwargs is None else sp_model_kwargs

        super().__init__(
            eos_token=__lowerCamelCase  ,					unk_token=__lowerCamelCase  ,					mask_token=__lowerCamelCase  ,					pad_token=__lowerCamelCase  ,					mask_token_sent=__lowerCamelCase  ,					offset=__lowerCamelCase  ,					additional_special_tokens=__lowerCamelCase  ,					sp_model_kwargs=self.sp_model_kwargs  ,					**__lowerCamelCase  ,					)
        _a							:  List[Any]				= mask_token_sent
        _a							:  Dict				= vocab_file
        _a							:  Dict				= spm.SentencePieceProcessor(**self.sp_model_kwargs       )
        self.sp_model.Load(__lowerCamelCase       )

        # add special tokens to encoder dict
        _a							:  Dict[int, str]				= {
            0: self.pad_token,
            1: self.eos_token,
        }

        if self.mask_token_sent is not None:
              self.encoder.update(
                  {
                      2: self.mask_token_sent,
                      3: self.mask_token,
                  }       )

        if self.offset > 0:
              # entries 2-104 are only used for pretraining and called <mask_1>, <mask_2>, unk_2, ...unk_102
              # mask_token_sent is already added to list -> so start at 1
              self.encoder.update({i + 3: additional_special_tokens[i] for i in range(1  ,					self.offset - 1       )}       )

        _a							:  Dict[str, int]				= {v: k for k, v in self.encoder.items()}







  @property
  def      snake_case_							(     self						:   str       )    ->		int:
        return len(self.sp_model       ) + self.offset







  def      snake_case_							(     self						:   Dict       )    ->		List[str]:
        _a							:  Optional[Any]				= {self.convert_ids_to_tokens(__lowerCamelCase       ): i for i in range(self.vocab_size       )}
        vocab.update(self.added_tokens_encoder       )
        return vocab







  def __getstate__(     self						:   Dict       )    ->		Dict:
        _a							:  int				= self.__dict__.copy()
        _a							:  Union[str, Any]				= None
        return state







  def __setstate__(     self						:   Union[str, Any]  ,					__snake_case						:   int       )    ->		List[str]:
        _a							:  Optional[int]				= d

        # for backward compatibility
        if not hasattr(self  ,					'''sp_model_kwargs'''       ):
              _a							:  Tuple				= {}

        _a							:  List[Any]				= spm.SentencePieceProcessor(**self.sp_model_kwargs       )
        self.sp_model.Load(self.vocab_file       )







  def      snake_case_							(     self						:   List[Any]  ,					__snake_case						:   int       )    ->		str:
        return self.sp_model.encode(__lowerCamelCase  ,					out_type=__lowerCamelCase       )







  def      snake_case_							(     self						:   List[Any]  ,					__snake_case						:   List[Any]       )    ->		Union[str, Any]:




        if token in self.decoder:
              return self.decoder[token]
        elif token in self.added_tokens_decoder:
              return self.added_tokens_decoder[token]
        _a							:  Any				= self.sp_model.piece_to_id(__lowerCamelCase       )
        return sp_id + self.offset







  def      snake_case_							(     self						:   Optional[int]  ,					__snake_case						:   Optional[int]       )    ->		Tuple:




        if index in self.encoder:
              return self.encoder[index]
        elif index in self.added_tokens_encoder:
              return self.added_tokens_encoder[index]
        else:
              _a							:  Optional[Any]				= self.sp_model.IdToPiece(index - self.offset       )
        return token







  def      snake_case_							(     self						:   Tuple  ,					__snake_case						:   Tuple       )    ->		List[str]:
        _a							:  int				= []
        _a							:  Optional[Any]				= ''''''
        for token in tokens:
              # make sure that special tokens are not decoded using sentencepiece model
              if token in self.all_special_tokens:
                    out_string += self.sp_model.decode(__lowerCamelCase       ) + token
                    _a							:  Dict				= []
              else:
                    current_sub_tokens.append(__lowerCamelCase       )
        out_string += self.sp_model.decode(__lowerCamelCase       )
        return out_string.strip()







  def      snake_case_							(     self						:   Tuple  ,					__snake_case						:   Union[str, Any]=False       )    ->		str:
        return 1







  def      snake_case_							(     self						:   Dict  ,					__snake_case						:   Union[str, Any]       )    ->		int:
        _a							:  int				= set(self.all_special_ids       )  # call it once instead of inside list comp
        all_special_ids.remove(self.unk_token_id       )  # <unk> is only sometimes special

        return [1 if x in all_special_ids else 0 for x in seq]







  def      snake_case_							(     self						:   List[str]  ,					__snake_case						:   str  ,					__snake_case						:   str = None  ,					__snake_case						:   Union[str, Any] = False       )    ->		Optional[int]:




        if already_has_special_tokens:
              return self._special_token_mask(__lowerCamelCase       )
        elif token_ids_a is None:
              return self._special_token_mask(__lowerCamelCase       ) + [1]
        else:
              return self._special_token_mask(token_ids_a + token_ids_a       ) + [1]







  def      snake_case_							(     self						:   Any  ,					__snake_case						:   Dict  ,					__snake_case						:   Any=None       )    ->		Tuple:




        if token_ids_a is None:
              return token_ids_a + [self.eos_token_id]
        # We don't expect to process pairs, but leave the pair logic for API consistency
        return token_ids_a + token_ids_a + [self.eos_token_id]







  def      snake_case_							(     self						:   Tuple  ,					__snake_case						:   Dict  ,					__snake_case						:   Optional[Any] = None       )    ->		int:




        if not os.path.isdir(__lowerCamelCase       ):
              logger.error(f"""Vocabulary path ({save_directory}) should be a directory"""       )
              return
        _a							:  Optional[int]				= os.path.join(
            __lowerCamelCase  ,					(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']       )

        if os.path.abspath(self.vocab_file       ) != os.path.abspath(__lowerCamelCase       ) and os.path.isfile(self.vocab_file       ):
              copyfile(self.vocab_file  ,					__lowerCamelCase       )
        elif not os.path.isfile(self.vocab_file       ):
              with open(__lowerCamelCase  ,					'''wb'''       ) as fi:
                    _a							:  List[Any]				= self.sp_model.serialized_model_proto()
                    fi.write(__lowerCamelCase       )

        return (out_vocab_file,)












"""simple docstring"""



import functools
import operator

from ...configuration_utils import PretrainedConfig
from ...utils import logging


a_					=     logging.get_logger(__name__)

a_					=     {
    """microsoft/unispeech-large-1500h-cv""": (
        """https://huggingface.co/microsoft/unispeech-large-1500h-cv/resolve/main/config.json"""
    ),
    # See all UniSpeech models at https://huggingface.co/models?filter=unispeech
}






class 						__snake_case   (						SCREAMING_SNAKE_CASE__	):





 """simple docstring"""


 _lowerCamelCase								=	"""unispeech"""

 def __init__(       self   ,		__lowerCamelCase=32   ,		__lowerCamelCase=768   ,		__lowerCamelCase=12   ,		__lowerCamelCase=12   ,		__lowerCamelCase=3072   ,		__lowerCamelCase="gelu"   ,		__lowerCamelCase=0.1   ,		__lowerCamelCase=0.1   ,		__lowerCamelCase=0.1   ,		__lowerCamelCase=0.0   ,		__lowerCamelCase=0.0   ,		__lowerCamelCase=0.1   ,		__lowerCamelCase=0.1   ,		__lowerCamelCase=0.0_2   ,		__lowerCamelCase=1e-5   ,		__lowerCamelCase="group"   ,		__lowerCamelCase="gelu"   ,		__lowerCamelCase=(512, 512, 512, 512, 512, 512, 512)   ,		__lowerCamelCase=(5, 2, 2, 2, 2, 2, 2)   ,		__lowerCamelCase=(10, 3, 3, 3, 3, 2, 2)   ,		__lowerCamelCase=False   ,		__lowerCamelCase=128   ,		__lowerCamelCase=16   ,		__lowerCamelCase=False   ,		__lowerCamelCase=True   ,		__lowerCamelCase=0.0_5   ,		__lowerCamelCase=10   ,		__lowerCamelCase=2   ,		__lowerCamelCase=0.0   ,		__lowerCamelCase=10   ,		__lowerCamelCase=0   ,		__lowerCamelCase=320   ,		__lowerCamelCase=2   ,		__lowerCamelCase=0.1   ,		__lowerCamelCase=100   ,		__lowerCamelCase=256   ,		__lowerCamelCase=256   ,		__lowerCamelCase=0.1   ,		__lowerCamelCase="mean"   ,		__lowerCamelCase=False   ,		__lowerCamelCase=False   ,		__lowerCamelCase=256   ,		__lowerCamelCase=80   ,		__lowerCamelCase=0   ,		__lowerCamelCase=1   ,		__lowerCamelCase=2   ,		__lowerCamelCase=0.5   ,		**__lowerCamelCase   ,		):



     '''simple docstring'''




     super().__init__(**__lowerCamelCase   ,		pad_token_id=__lowerCamelCase   ,		bos_token_id=__lowerCamelCase   ,		eos_token_id=__lowerCamelCase    )
     __A       :		Dict							=	hidden_size
     __A       :		Dict							=	feat_extract_norm
     __A       :		int							=	feat_extract_activation
     __A       :		Dict							=	list(__lowerCamelCase    )
     __A       :		str							=	list(__lowerCamelCase    )
     __A       :		Dict							=	list(__lowerCamelCase    )
     __A       :		str							=	conv_bias
     __A       :		Optional[int]							=	num_conv_pos_embeddings
     __A       :		int							=	num_conv_pos_embedding_groups
     __A       :		Union[str, Any]							=	len(self.conv_dim    )
     __A       :		List[str]							=	num_hidden_layers
     __A       :		Dict							=	intermediate_size
     __A       :		List[str]							=	hidden_act
     __A       :		int							=	num_attention_heads
     __A       :		str							=	hidden_dropout
     __A       :		str							=	attention_dropout
     __A       :		Optional[Any]							=	activation_dropout
     __A       :		Optional[int]							=	feat_proj_dropout
     __A       :		Optional[Any]							=	final_dropout
     __A       :		List[Any]							=	layerdrop
     __A       :		Any							=	layer_norm_eps
     __A       :		str							=	initializer_range
     __A       :		List[str]							=	num_ctc_classes
     __A       :		Dict							=	vocab_size
     __A       :		Dict							=	do_stable_layer_norm
     __A       :		Union[str, Any]							=	use_weighted_layer_sum
     __A       :		Tuple							=	classifier_proj_size

     if (
         (len(self.conv_stride    ) != self.num_feat_extract_layers)
         or (len(self.conv_kernel    ) != self.num_feat_extract_layers)
         or (len(self.conv_dim    ) != self.num_feat_extract_layers)
     ):
         raise ValueError(
             '''Configuration for convolutional layers is incorrect. It is required that `len(config.conv_dim)` =='''
             ''' `len(config.conv_stride)` == `len(config.conv_kernel)`, but is `len(config.conv_dim) ='''
             F""" {len(self.conv_dim    )}`, `len(config.conv_stride) = {len(self.conv_stride    )}`,"""
             F""" `len(config.conv_kernel) = {len(self.conv_kernel    )}`."""    )

     # fine-tuning config parameters for SpecAugment: https://arxiv.org/abs/1904.08779
     __A       :		List[Any]							=	apply_spec_augment
     __A       :		str							=	mask_time_prob
     __A       :		Dict							=	mask_time_length
     __A       :		List[str]							=	mask_time_min_masks
     __A       :		Union[str, Any]							=	mask_feature_prob
     __A       :		Tuple							=	mask_feature_length
     __A       :		Any							=	mask_feature_min_masks

     # parameters for pretraining with codevector quantized representations
     __A       :		int							=	num_codevectors_per_group
     __A       :		Any							=	num_codevector_groups
     __A       :		Optional[Any]							=	contrastive_logits_temperature
     __A       :		Dict							=	feat_quantizer_dropout
     __A       :		Tuple							=	num_negatives
     __A       :		List[str]							=	codevector_dim
     __A       :		str							=	proj_codevector_dim
     __A       :		Union[str, Any]							=	diversity_loss_weight

     # ctc loss
     __A       :		Tuple							=	ctc_loss_reduction
     __A       :		Optional[int]							=	ctc_zero_infinity

     # pretraining loss
     __A       :		str							=	replace_prob







 @property
 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     return functools.reduce(operator.mul   ,		self.conv_stride   ,		1    )




