


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


lowerCAmelCase__  :str             =							'''▁'''

lowerCAmelCase__  :int             =							{'''vocab_file''': '''spiece.model'''}

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

lowerCAmelCase__  :List[str]             =							{
    '''google/pegasus-xsum''': 5_1_2,
}


lowerCAmelCase__  :List[str]             =							logging.get_logger(__name__)

class 		__a				(    UpperCAmelCase      ):
		_a     :  Any           =     VOCAB_FILES_NAMES

		_a     :  Tuple           =     VOCAB_FILES_NAMES
		_a     :  str           =     PRETRAINED_VOCAB_FILES_MAP
		_a     :  Optional[Any]           =     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
		_a     :  List[Any]           =     ['input_ids', 'attention_mask']
		def __init__(      self      ,   _SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE="<pad>"      ,   _SCREAMING_SNAKE_CASE="</s>"      ,   _SCREAMING_SNAKE_CASE="<unk>"      ,   _SCREAMING_SNAKE_CASE="<mask_2>"      ,   _SCREAMING_SNAKE_CASE="<mask_1>"      ,   _SCREAMING_SNAKE_CASE=None      ,   _SCREAMING_SNAKE_CASE=103      ,   _SCREAMING_SNAKE_CASE = None      ,   **_SCREAMING_SNAKE_CASE      ,   )			->	None:




									"""simple docstring"""





									_UpperCAmelCase					=      offset
									if additional_special_tokens is not None:
																if not isinstance(_SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE						):
																							raise TypeError(
																							    f'''additional_special_tokens should be of type {type(_SCREAMING_SNAKE_CASE						)}, but is'''
																							    f''' {type(_SCREAMING_SNAKE_CASE						)}'''						)

																_UpperCAmelCase					=      (
																    ([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(_SCREAMING_SNAKE_CASE						)      ,   self.offset - 1						)
																]

																if len(set(_SCREAMING_SNAKE_CASE						)						) != len(_SCREAMING_SNAKE_CASE						):
																							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}.'''						)
																_UpperCAmelCase					=      additional_special_tokens_extended
									else:
																_UpperCAmelCase					=      [mask_token_sent] if mask_token_sent is not None else []
																additional_special_tokens += [f'''<unk_{i}>''' for i in range(2      ,   self.offset						)]

									_UpperCAmelCase					=      {} if sp_model_kwargs is None else sp_model_kwargs

									super().__init__(
									    eos_token=_SCREAMING_SNAKE_CASE      ,   unk_token=_SCREAMING_SNAKE_CASE      ,   mask_token=_SCREAMING_SNAKE_CASE      ,   pad_token=_SCREAMING_SNAKE_CASE      ,   mask_token_sent=_SCREAMING_SNAKE_CASE      ,   offset=_SCREAMING_SNAKE_CASE      ,   additional_special_tokens=_SCREAMING_SNAKE_CASE      ,   sp_model_kwargs=self.sp_model_kwargs      ,   **_SCREAMING_SNAKE_CASE      ,   )
									_UpperCAmelCase					=      mask_token_sent
									_UpperCAmelCase					=      vocab_file
									_UpperCAmelCase					=      spm.SentencePieceProcessor(**self.sp_model_kwargs						)
									self.sp_model.Load(_SCREAMING_SNAKE_CASE						)

									# add special tokens to encoder dict
									_UpperCAmelCase					=      {
									    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						)}						)

									_UpperCAmelCase					=      {v: k for k, v in self.encoder.items()}
		@property
		def      UpperCAmelCase__    (      self						)			->	int:




									"""simple docstring"""





									return len(self.sp_model						) + self.offset
		def      UpperCAmelCase__    (      self						)			->	Dict[str, int]:




									"""simple docstring"""





									_UpperCAmelCase					=      {self.convert_ids_to_tokens(_SCREAMING_SNAKE_CASE						): i for i in range(self.vocab_size						)}
									vocab.update(self.added_tokens_encoder						)
									return vocab
		def __getstate__(      self						)			->	List[str]:




									"""simple docstring"""





									_UpperCAmelCase					=      self.__dict__.copy()
									_UpperCAmelCase					=      None
									return state
		def __setstate__(      self      ,   _SCREAMING_SNAKE_CASE						)			->	Optional[int]:




									"""simple docstring"""





									_UpperCAmelCase					=      d

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

									_UpperCAmelCase					=      spm.SentencePieceProcessor(**self.sp_model_kwargs						)
									self.sp_model.Load(self.vocab_file						)
		def      UpperCAmelCase__    (      self      ,   _SCREAMING_SNAKE_CASE						)			->	List[str]:




									"""simple docstring"""





									return self.sp_model.encode(_SCREAMING_SNAKE_CASE      ,   out_type=_SCREAMING_SNAKE_CASE						)
		def      UpperCAmelCase__    (      self      ,   _SCREAMING_SNAKE_CASE						)			->	int:




									"""simple docstring"""





									if token in self.decoder:
																return self.decoder[token]
									elif token in self.added_tokens_decoder:
																return self.added_tokens_decoder[token]
									_UpperCAmelCase					=      self.sp_model.piece_to_id(_SCREAMING_SNAKE_CASE						)
									return sp_id + self.offset
		def      UpperCAmelCase__    (      self      ,   _SCREAMING_SNAKE_CASE						)			->	str:




									"""simple docstring"""





									if index in self.encoder:
																return self.encoder[index]
									elif index in self.added_tokens_encoder:
																return self.added_tokens_encoder[index]
									else:
																_UpperCAmelCase					=      self.sp_model.IdToPiece(index - self.offset						)
									return token
		def      UpperCAmelCase__    (      self      ,   _SCREAMING_SNAKE_CASE						)			->	List[Any]:




									"""simple docstring"""





									_UpperCAmelCase					=      []
									_UpperCAmelCase					=      ''
									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(_SCREAMING_SNAKE_CASE						) + token
																							_UpperCAmelCase					=      []
																else:
																							current_sub_tokens.append(_SCREAMING_SNAKE_CASE						)
									out_string += self.sp_model.decode(_SCREAMING_SNAKE_CASE						)
									return out_string.strip()
		def      UpperCAmelCase__    (      self      ,   _SCREAMING_SNAKE_CASE=False						)			->	List[str]:




									"""simple docstring"""





									return 1
		def      UpperCAmelCase__    (      self      ,   _SCREAMING_SNAKE_CASE						)			->	Dict:




									"""simple docstring"""





									_UpperCAmelCase					=      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      UpperCAmelCase__    (      self      ,   _SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE = None      ,   _SCREAMING_SNAKE_CASE = False						)			->	List[int]:




									"""simple docstring"""





									if already_has_special_tokens:
																return self._special_token_mask(_SCREAMING_SNAKE_CASE						)
									elif token_ids_a is None:
																return self._special_token_mask(_SCREAMING_SNAKE_CASE						) + [1]
									else:
																return self._special_token_mask(token_ids_a + token_ids_a						) + [1]
		def      UpperCAmelCase__    (      self      ,   _SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE=None						)			->	List[int]:




									"""simple docstring"""





									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      UpperCAmelCase__    (      self      ,   _SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE = None						)			->	Tuple[str]:




									"""simple docstring"""





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

									if os.path.abspath(self.vocab_file						) != os.path.abspath(_SCREAMING_SNAKE_CASE						) and os.path.isfile(self.vocab_file						):
																copyfile(self.vocab_file      ,   _SCREAMING_SNAKE_CASE						)
									elif not os.path.isfile(self.vocab_file						):
																with open(_SCREAMING_SNAKE_CASE      ,   'wb'						) as fi:
																							_UpperCAmelCase					=      self.sp_model.serialized_model_proto()
																							fi.write(_SCREAMING_SNAKE_CASE						)

									return (out_vocab_file,)




import warnings
from contextlib import contextmanager

from ....processing_utils import ProcessorMixin

class 		__a				(    UpperCAmelCase      ):
		_a     :  Optional[int]           =     'MCTCTFeatureExtractor'
		_a     :  int           =     'AutoTokenizer'
		def __init__(      self      ,   _SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE						)			->	List[str]:




									"""simple docstring"""





									super().__init__(_SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE						)
									_UpperCAmelCase					=      self.feature_extractor
									_UpperCAmelCase					=      False
		def __call__(      self      ,   *_SCREAMING_SNAKE_CASE      ,   **_SCREAMING_SNAKE_CASE						)			->	Tuple:




									"""simple docstring"""





									if self._in_target_context_manager:
																return self.current_processor(*_SCREAMING_SNAKE_CASE      ,   **_SCREAMING_SNAKE_CASE						)

									if "raw_speech" in kwargs:
																warnings.warn('Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead.'						)
																_UpperCAmelCase					=      kwargs.pop('raw_speech'						)
									else:
																_UpperCAmelCase					=      kwargs.pop('audio'      ,   _SCREAMING_SNAKE_CASE						)
									_UpperCAmelCase					=      kwargs.pop('sampling_rate'      ,   _SCREAMING_SNAKE_CASE						)
									_UpperCAmelCase					=      kwargs.pop('text'      ,   _SCREAMING_SNAKE_CASE						)
									if len(_SCREAMING_SNAKE_CASE						) > 0:
																_UpperCAmelCase					=      args[0]
																_UpperCAmelCase					=      args[1:]

									if audio is None and text is None:
																raise ValueError('You need to specify either an `audio` or `text` input to process.'						)

									if audio is not None:
																_UpperCAmelCase					=      self.feature_extractor(_SCREAMING_SNAKE_CASE      ,   *_SCREAMING_SNAKE_CASE      ,   sampling_rate=_SCREAMING_SNAKE_CASE      ,   **_SCREAMING_SNAKE_CASE						)
									if text is not None:
																_UpperCAmelCase					=      self.tokenizer(_SCREAMING_SNAKE_CASE      ,   **_SCREAMING_SNAKE_CASE						)

									if text is None:
																return inputs
									elif audio is None:
																return encodings
									else:
																_UpperCAmelCase					=      encodings['input_ids']
																return inputs
		def      UpperCAmelCase__    (      self      ,   *_SCREAMING_SNAKE_CASE      ,   **_SCREAMING_SNAKE_CASE						)			->	Optional[int]:




									"""simple docstring"""





									return self.tokenizer.batch_decode(*_SCREAMING_SNAKE_CASE      ,   **_SCREAMING_SNAKE_CASE						)
		def      UpperCAmelCase__    (      self      ,   *_SCREAMING_SNAKE_CASE      ,   **_SCREAMING_SNAKE_CASE						)			->	str:




									"""simple docstring"""





									if self._in_target_context_manager:
																return self.current_processor.pad(*_SCREAMING_SNAKE_CASE      ,   **_SCREAMING_SNAKE_CASE						)

									_UpperCAmelCase					=      kwargs.pop('input_features'      ,   _SCREAMING_SNAKE_CASE						)
									_UpperCAmelCase					=      kwargs.pop('labels'      ,   _SCREAMING_SNAKE_CASE						)
									if len(_SCREAMING_SNAKE_CASE						) > 0:
																_UpperCAmelCase					=      args[0]
																_UpperCAmelCase					=      args[1:]

									if input_features is not None:
																_UpperCAmelCase					=      self.feature_extractor.pad(_SCREAMING_SNAKE_CASE      ,   *_SCREAMING_SNAKE_CASE      ,   **_SCREAMING_SNAKE_CASE						)
									if labels is not None:
																_UpperCAmelCase					=      self.tokenizer.pad(_SCREAMING_SNAKE_CASE      ,   **_SCREAMING_SNAKE_CASE						)

									if labels is None:
																return input_features
									elif input_features is None:
																return labels
									else:
																_UpperCAmelCase					=      labels['input_ids']
																return input_features
		def      UpperCAmelCase__    (      self      ,   *_SCREAMING_SNAKE_CASE      ,   **_SCREAMING_SNAKE_CASE						)			->	int:




									"""simple docstring"""





									return self.tokenizer.decode(*_SCREAMING_SNAKE_CASE      ,   **_SCREAMING_SNAKE_CASE						)






		@contextmanager
		def      UpperCAmelCase__    (      self						)			->	Optional[Any]:




									"""simple docstring"""





									warnings.warn(
									    '`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your '
									    'labels by using the argument `text` of the regular `__call__` method (either in the same call as '
									    'your audio inputs, or in a separate call.'						)
									_UpperCAmelCase					=      True
									_UpperCAmelCase					=      self.tokenizer
									yield
									_UpperCAmelCase					=      self.feature_extractor
									_UpperCAmelCase					=      False
