



"""simple docstring"""

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


lowercase__					:						Any												=			logging.get_logger(__name__)

lowercase__					:						List[str]												=			{'''vocab_file''': '''spm_char.model'''}

lowercase__					:						Any												=			{
    '''vocab_file''': {
        '''microsoft/speecht5_asr''': '''https://huggingface.co/microsoft/speecht5_asr/resolve/main/spm_char.model''',
        '''microsoft/speecht5_tts''': '''https://huggingface.co/microsoft/speecht5_tts/resolve/main/spm_char.model''',
        '''microsoft/speecht5_vc''': '''https://huggingface.co/microsoft/speecht5_vc/resolve/main/spm_char.model''',
    }
}

lowercase__					:						Optional[Any]												=			{
    '''microsoft/speecht5_asr''': 10_24,
    '''microsoft/speecht5_tts''': 10_24,
    '''microsoft/speecht5_vc''': 10_24,
}




class  _UpperCAmelCase (    lowerCAmelCase__):
 _lowerCAmelCase     :    str	      =	VOCAB_FILES_NAMES
 _lowerCAmelCase     :    List[Any]	      =	PRETRAINED_VOCAB_FILES_MAP
 _lowerCAmelCase     :    Optional[Any]	      =	PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 _lowerCAmelCase     :    int	      =	["""input_ids""", """attention_mask"""]

 def __init__(  self	: Any						,  lowercase_	: List[Any]						,  lowercase_	: int="<s>"						,  lowercase_	: int="</s>"						,  lowercase_	: Union[str, Any]="<unk>"						,  lowercase_	: List[Any]="<pad>"						,  lowercase_	: Optional[Dict[str, Any]] = None						,  **lowercase_	: Tuple						,  ):
     snake_case_				:  Union[str, Any]       =			{} if sp_model_kwargs is None else sp_model_kwargs

     super().__init__(
         bos_token=lowercase_						,  eos_token=lowercase_						,  unk_token=lowercase_						,  pad_token=lowercase_						,  sp_model_kwargs=self.sp_model_kwargs						,  **lowercase_						,  )

     snake_case_				:  str       =			vocab_file

     snake_case_				:  Any       =			spm.SentencePieceProcessor(**self.sp_model_kwargs       )
     self.sp_model.Load(lowercase_       )

 @property
 def 		_snake_case      (  self	: Union[str, Any]       ):
     return self.sp_model.get_piece_size()

 def 		_snake_case      (  self	: List[str]       ):
     snake_case_				:  Tuple       =			{self.convert_ids_to_tokens(lowercase_       ): i for i in range(self.vocab_size       )}
     vocab.update(self.added_tokens_encoder       )
     return vocab

 def __getstate__(  self	: Any       ):
     snake_case_				:  Optional[Any]       =			self.__dict__.copy()
     snake_case_				:  Tuple       =			None
     return state

 def __setstate__(  self	: Optional[Any]						,  lowercase_	: List[Any]       ):
     snake_case_				:  List[Any]       =			d

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

     snake_case_				:  Optional[int]       =			spm.SentencePieceProcessor(**self.sp_model_kwargs       )
     self.sp_model.Load(self.vocab_file       )

 def 		_snake_case      (  self	: List[Any]						,  lowercase_	: str       ):
     return self.sp_model.encode(lowercase_						,  out_type=lowercase_       )

 def 		_snake_case      (  self	: Any						,  lowercase_	: List[Any]       ):
     return self.sp_model.piece_to_id(lowercase_       )

 def 		_snake_case      (  self	: Any						,  lowercase_	: Dict       ):
     snake_case_				:  Optional[int]       =			self.sp_model.IdToPiece(lowercase_       )
     return token

 def 		_snake_case      (  self	: str						,  lowercase_	: Tuple       ):
     snake_case_				:  List[str]       =			[]
     snake_case_				:  Union[str, 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(lowercase_       ) + token
             snake_case_				:  List[str]       =			[]
         else:
             current_sub_tokens.append(lowercase_       )
     out_string += self.sp_model.decode(lowercase_       )
     return out_string.strip()

 def 		_snake_case      (  self	: Dict						,  lowercase_	: int						,  lowercase_	: Any=None       ):
     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	: 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_       )

     snake_case_				:  int       =			[1]
     if token_ids_a is None:
         return ([0] * len(lowercase_       )) + suffix_ones
     return ([0] * len(lowercase_       )) + ([0] * len(lowercase_       )) + suffix_ones






 def 		_snake_case      (  self	: Optional[int]						,  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_				:  Optional[Any]       =			os.path.join(
         lowercase_						,  (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']       )

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

     return (out_vocab_file,)










"""simple docstring"""

import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoFeatureExtractor, WavaVecaFeatureExtractor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / '''utils'''))

from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


lowercase__					:						Any												=			get_tests_dir('''fixtures''')




class  _UpperCAmelCase (    unittest.TestCase):

 def 		_snake_case      (  self	: Tuple       ):
     # A mock response for an HTTP head request to emulate server down
     snake_case_				:  str       =			mock.Mock()
     snake_case_				:  Optional[Any]       =			500
     snake_case_				:  str       =			{}
     snake_case_				:  Optional[int]       =			HTTPError
     snake_case_				:  Tuple       =			{}

     # Download this model to make sure it's in the cache.
     snake_case_				:  List[str]       =			WavaVecaFeatureExtractor.from_pretrained('''hf-internal-testing/tiny-random-wav2vec2'''       )
     # Under the mock environment we get a 500 error when trying to reach the model.
     with mock.patch('''requests.Session.request'''						,  return_value=lowercase_       ) as mock_head:
         snake_case_				:  int       =			WavaVecaFeatureExtractor.from_pretrained('''hf-internal-testing/tiny-random-wav2vec2'''       )
         # This check we did call the fake head request
         mock_head.assert_called()






 def 		_snake_case      (  self	: Union[str, Any]       ):
     # This test is for deprecated behavior and can be removed in v5
     snake_case_				:  str       =			WavaVecaFeatureExtractor.from_pretrained(
         '''https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json'''       )





@is_staging_test
class  _UpperCAmelCase (    unittest.TestCase):

 @classmethod
 def 		_snake_case      (  cls	: int       ):
     snake_case_				:  Dict       =			TOKEN
     HfFolder.save_token(lowercase_       )

 @classmethod
 def 		_snake_case      (  cls	: str       ):
     try:
         delete_repo(token=cls._token						,  repo_id='''test-feature-extractor'''       )
     except HTTPError:
         pass

     try:
         delete_repo(token=cls._token						,  repo_id='''valid_org/test-feature-extractor-org'''       )
     except HTTPError:
         pass

     try:
         delete_repo(token=cls._token						,  repo_id='''test-dynamic-feature-extractor'''       )
     except HTTPError:
         pass

 def 		_snake_case      (  self	: Any       ):
     snake_case_				:  int       =			WavaVecaFeatureExtractor.from_pretrained(lowercase_       )
     feature_extractor.push_to_hub('''test-feature-extractor'''						,  use_auth_token=self._token       )

     snake_case_				:  Optional[Any]       =			WavaVecaFeatureExtractor.from_pretrained(f"{USER}/test-feature-extractor"       )
     for k, v in feature_extractor.__dict__.items():
         self.assertEqual(lowercase_						,  getattr(lowercase_						,  lowercase_       )       )

     # Reset repo
     delete_repo(token=self._token						,  repo_id='''test-feature-extractor'''       )

     # Push to hub via save_pretrained
     with tempfile.TemporaryDirectory() as tmp_dir:
         feature_extractor.save_pretrained(
             lowercase_						,  repo_id='''test-feature-extractor'''						,  push_to_hub=lowercase_						,  use_auth_token=self._token       )

     snake_case_				:  Optional[int]       =			WavaVecaFeatureExtractor.from_pretrained(f"{USER}/test-feature-extractor"       )
     for k, v in feature_extractor.__dict__.items():
         self.assertEqual(lowercase_						,  getattr(lowercase_						,  lowercase_       )       )

 def 		_snake_case      (  self	: Dict       ):
     snake_case_				:  Union[str, Any]       =			WavaVecaFeatureExtractor.from_pretrained(lowercase_       )
     feature_extractor.push_to_hub('''valid_org/test-feature-extractor'''						,  use_auth_token=self._token       )

     snake_case_				:  Optional[int]       =			WavaVecaFeatureExtractor.from_pretrained('''valid_org/test-feature-extractor'''       )
     for k, v in feature_extractor.__dict__.items():
         self.assertEqual(lowercase_						,  getattr(lowercase_						,  lowercase_       )       )

     # Reset repo
     delete_repo(token=self._token						,  repo_id='''valid_org/test-feature-extractor'''       )

     # Push to hub via save_pretrained
     with tempfile.TemporaryDirectory() as tmp_dir:
         feature_extractor.save_pretrained(
             lowercase_						,  repo_id='''valid_org/test-feature-extractor-org'''						,  push_to_hub=lowercase_						,  use_auth_token=self._token       )

     snake_case_				:  int       =			WavaVecaFeatureExtractor.from_pretrained('''valid_org/test-feature-extractor-org'''       )
     for k, v in feature_extractor.__dict__.items():
         self.assertEqual(lowercase_						,  getattr(lowercase_						,  lowercase_       )       )






 def 		_snake_case      (  self	: str       ):
     CustomFeatureExtractor.register_for_auto_class()
     snake_case_				:  Optional[Any]       =			CustomFeatureExtractor.from_pretrained(lowercase_       )

     feature_extractor.push_to_hub('''test-dynamic-feature-extractor'''						,  use_auth_token=self._token       )

     # This has added the proper auto_map field to the config
     self.assertDictEqual(
         feature_extractor.auto_map						,  {'''AutoFeatureExtractor''': '''custom_feature_extraction.CustomFeatureExtractor'''}						,  )

     snake_case_				:  Any       =			AutoFeatureExtractor.from_pretrained(
         f"{USER}/test-dynamic-feature-extractor"						,  trust_remote_code=lowercase_       )
     # Can't make an isinstance check because the new_feature_extractor is from the CustomFeatureExtractor class of a dynamic module
     self.assertEqual(new_feature_extractor.__class__.__name__						,  '''CustomFeatureExtractor'''       )





