



"""simple docstring"""

import json
import os
from pathlib import Path
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple, Union

import sentencepiece

from ...tokenization_utils import BatchEncoding, PreTrainedTokenizer
from ...utils import logging


lowercase__					:						Any												=			logging.get_logger(__name__)

lowercase__					:						List[str]												=			'''▁'''

lowercase__					:						Any												=			{
    '''vocab_file''': '''vocab.json''',
    '''spm_file''': '''sentencepiece.bpe.model''',
    '''tokenizer_config_file''': '''tokenizer_config.json''',
}

lowercase__					:						Any												=			{
    '''vocab_file''': {
        '''facebook/m2m100_418M''': '''https://huggingface.co/facebook/m2m100_418M/resolve/main/vocab.json''',
        '''facebook/m2m100_1.2B''': '''https://huggingface.co/facebook/m2m100_1.2B/resolve/main/vocab.json''',
    },
    '''spm_file''': {
        '''facebook/m2m100_418M''': '''https://huggingface.co/facebook/m2m100_418M/resolve/main/sentencepiece.bpe.model''',
        '''facebook/m2m100_1.2B''': '''https://huggingface.co/facebook/m2m100_1.2B/resolve/main/sentencepiece.bpe.model''',
    },
    '''tokenizer_config_file''': {
        '''facebook/m2m100_418M''': '''https://huggingface.co/facebook/m2m100_418M/resolve/main/tokenizer_config.json''',
        '''facebook/m2m100_1.2B''': '''https://huggingface.co/facebook/m2m100_1.2B/resolve/main/tokenizer_config.json''',
    },
}

lowercase__					:						Optional[int]												=			{
    '''facebook/m2m100_418M''': 10_24,
}

# fmt: off
lowercase__					:						Optional[int]												=			{
    '''m2m100''': ['''af''', '''am''', '''ar''', '''ast''', '''az''', '''ba''', '''be''', '''bg''', '''bn''', '''br''', '''bs''', '''ca''', '''ceb''', '''cs''', '''cy''', '''da''', '''de''', '''el''', '''en''', '''es''', '''et''', '''fa''', '''ff''', '''fi''', '''fr''', '''fy''', '''ga''', '''gd''', '''gl''', '''gu''', '''ha''', '''he''', '''hi''', '''hr''', '''ht''', '''hu''', '''hy''', '''id''', '''ig''', '''ilo''', '''is''', '''it''', '''ja''', '''jv''', '''ka''', '''kk''', '''km''', '''kn''', '''ko''', '''lb''', '''lg''', '''ln''', '''lo''', '''lt''', '''lv''', '''mg''', '''mk''', '''ml''', '''mn''', '''mr''', '''ms''', '''my''', '''ne''', '''nl''', '''no''', '''ns''', '''oc''', '''or''', '''pa''', '''pl''', '''ps''', '''pt''', '''ro''', '''ru''', '''sd''', '''si''', '''sk''', '''sl''', '''so''', '''sq''', '''sr''', '''ss''', '''su''', '''sv''', '''sw''', '''ta''', '''th''', '''tl''', '''tn''', '''tr''', '''uk''', '''ur''', '''uz''', '''vi''', '''wo''', '''xh''', '''yi''', '''yo''', '''zh''', '''zu'''],
    '''wmt21''': ['''en''', '''ha''', '''is''', '''ja''', '''cs''', '''ru''', '''zh''', '''de''']
}




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

 _lowerCAmelCase     :    List[int]	      =	[]
 _lowerCAmelCase     :    List[int]	      =	[]

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

     snake_case_				:  List[Any]       =			language_codes
     snake_case_				:  List[Any]       =			FAIRSEQ_LANGUAGE_CODES[language_codes]
     snake_case_				:  Tuple       =			{lang_code: f"__{lang_code}__" for lang_code in fairseq_language_code}

     snake_case_				:  Any       =			kwargs.get('''additional_special_tokens'''						,  []       )
     kwargs["additional_special_tokens"] += [
         self.get_lang_token(lowercase_       )
         for lang_code in fairseq_language_code
         if self.get_lang_token(lowercase_       ) not in kwargs["additional_special_tokens"]
     ]

     super().__init__(
         src_lang=lowercase_						,  tgt_lang=lowercase_						,  bos_token=lowercase_						,  eos_token=lowercase_						,  sep_token=lowercase_						,  unk_token=lowercase_						,  pad_token=lowercase_						,  language_codes=lowercase_						,  sp_model_kwargs=self.sp_model_kwargs						,  num_madeup_words=lowercase_						,  **lowercase_						,  )

     snake_case_				:  Dict       =			vocab_file
     snake_case_				:  List[str]       =			load_json(lowercase_       )
     snake_case_				:  Optional[Any]       =			{v: k for k, v in self.encoder.items()}
     snake_case_				:  int       =			spm_file
     snake_case_				:  Union[str, Any]       =			load_spm(lowercase_						,  self.sp_model_kwargs       )

     snake_case_				:  Dict       =			len(self.encoder       )

     snake_case_				:  Optional[int]       =			{
         self.get_lang_token(lowercase_       ): self.encoder_size + i for i, lang_code in enumerate(lowercase_       )
     }
     snake_case_				:  Union[str, Any]       =			{lang_code: self.encoder_size + i for i, lang_code in enumerate(lowercase_       )}
     snake_case_				:  str       =			{v: k for k, v in self.lang_token_to_id.items()}

     snake_case_				:  Optional[Any]       =			src_lang if src_lang is not None else '''en'''
     snake_case_				:  List[Any]       =			tgt_lang
     snake_case_				:  Dict       =			self.get_lang_id(self._src_lang       )
     self.set_src_lang_special_tokens(self._src_lang       )

     snake_case_				:  Any       =			num_madeup_words

 @property
 def 		_snake_case      (  self	: Tuple       ):
     return len(self.encoder       ) + len(self.lang_token_to_id       )

 @property
 def 		_snake_case      (  self	: List[Any]       ):
     return self._src_lang

 @src_lang.setter
 def 		_snake_case      (  self	: Union[str, Any]						,  lowercase_	: str       ):
     snake_case_				:  str       =			new_src_lang
     self.set_src_lang_special_tokens(self._src_lang       )

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

 def 		_snake_case      (  self	: str						,  lowercase_	: Optional[int]       ):
     if token in self.lang_token_to_id:
         return self.lang_token_to_id[token]
     return self.encoder.get(lowercase_						,  self.encoder[self.unk_token]       )

 def 		_snake_case      (  self	: List[str]						,  lowercase_	: int       ):
     if index in self.id_to_lang_token:
         return self.id_to_lang_token[index]
     return self.decoder.get(lowercase_						,  self.unk_token       )

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

 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_				:  Tuple       =			[1] * len(self.prefix_tokens       )
     snake_case_				:  List[str]       =			[1] * len(self.suffix_tokens       )
     if token_ids_a is None:
         return prefix_ones + ([0] * len(lowercase_       )) + suffix_ones
     return prefix_ones + ([0] * len(lowercase_       )) + ([0] * len(lowercase_       )) + suffix_ones

 def 		_snake_case      (  self	: Union[str, Any]						,  lowercase_	: List[int]						,  lowercase_	: Optional[List[int]] = None       ):
     if token_ids_a is None:
         return self.prefix_tokens + token_ids_a + self.suffix_tokens
     # We don't expect to process pairs, but leave the pair logic for API consistency
     return self.prefix_tokens + token_ids_a + token_ids_a + self.suffix_tokens

 def 		_snake_case      (  self	: Dict       ):
     snake_case_				:  Optional[Any]       =			{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	: Tuple       ):
     snake_case_				:  Dict       =			self.__dict__.copy()
     snake_case_				:  str       =			None
     return state

 def __setstate__(  self	: Union[str, Any]						,  lowercase_	: Dict       ):
     snake_case_				:  Dict       =			d

     # for backward compatibility
     if not hasattr(self						,  '''sp_model_kwargs'''       ):
         snake_case_				:  Union[str, Any]       =			{}

     snake_case_				:  List[Any]       =			load_spm(self.spm_file						,  self.sp_model_kwargs       )

 def 		_snake_case      (  self	: List[str]						,  lowercase_	: str						,  lowercase_	: Optional[str] = None       ):
     snake_case_				:  str       =			Path(lowercase_       )
     if not save_dir.is_dir():
         raise OSError(f"{save_directory} should be a directory"       )
     snake_case_				:  Optional[Any]       =			save_dir / (
         (filename_prefix + '''-''' if filename_prefix else '''''') + self.vocab_files_names['''vocab_file''']
     )
     snake_case_				:  Optional[int]       =			save_dir / (
         (filename_prefix + '''-''' if filename_prefix else '''''') + self.vocab_files_names['''spm_file''']
     )

     save_json(self.encoder						,  lowercase_       )

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

     return (str(lowercase_       ), str(lowercase_       ))

 def 		_snake_case      (  self	: Any						,  lowercase_	: List[str]						,  lowercase_	: str = "en"						,  lowercase_	: Optional[List[str]] = None						,  lowercase_	: str = "ro"						,  **lowercase_	: Optional[Any]						,  ):
     snake_case_				:  List[str]       =			src_lang
     snake_case_				:  Dict       =			tgt_lang
     self.set_src_lang_special_tokens(self.src_lang       )
     return super().prepare_seqaseq_batch(lowercase_						,  lowercase_						,  **lowercase_       )

 def 		_snake_case      (  self	: Optional[int]						,  lowercase_	: Tuple						,  lowercase_	: Optional[str]						,  lowercase_	: Optional[str]						,  **lowercase_	: Optional[Any]       ):
     if src_lang is None or tgt_lang is None:
         raise ValueError('''Translation requires a `src_lang` and a `tgt_lang` for this model'''       )
     snake_case_				:  Dict       =			src_lang
     snake_case_				:  Optional[int]       =			self(lowercase_						,  add_special_tokens=lowercase_						,  **lowercase_       )
     snake_case_				:  str       =			self.get_lang_id(lowercase_       )
     snake_case_				:  str       =			tgt_lang_id
     return inputs

 def 		_snake_case      (  self	: Dict       ):
     self.set_src_lang_special_tokens(self.src_lang       )

 def 		_snake_case      (  self	: List[str]       ):
     self.set_tgt_lang_special_tokens(self.tgt_lang       )

 def 		_snake_case      (  self	: int						,  lowercase_	: str       ):
     snake_case_				:  Dict       =			self.get_lang_token(lowercase_       )
     snake_case_				:  str       =			self.lang_token_to_id[lang_token]
     snake_case_				:  Optional[Any]       =			[self.cur_lang_id]
     snake_case_				:  Union[str, Any]       =			[self.eos_token_id]

 def 		_snake_case      (  self	: Dict						,  lowercase_	: str       ):
     snake_case_				:  Tuple       =			self.get_lang_token(lowercase_       )
     snake_case_				:  Optional[Any]       =			self.lang_token_to_id[lang_token]
     snake_case_				:  Any       =			[self.cur_lang_id]
     snake_case_				:  Optional[int]       =			[self.eos_token_id]

 def 		_snake_case      (  self	: Optional[int]						,  lowercase_	: str       ):
     return self.lang_code_to_token[lang]






 def 		_snake_case      (  self	: Any						,  lowercase_	: str       ):
     snake_case_				:  List[str]       =			self.get_lang_token(lowercase_       )
     return self.lang_token_to_id[lang_token]





def __lowercase    (							_a	,    _a					):
    snake_case_				:  List[Any]       =			sentencepiece.SentencePieceProcessor(**_a					)
    spm.Load(str(_a					)					)
    return spm




def __lowercase    (							_a					):
    with open(_a	,    '''r'''					) as f:
        return json.load(_a					)






def __lowercase    (							_a	,    _a					):
    with open(_a	,    '''w'''					) as f:
        json.dump(_a	,    _a	,    indent=2					)










"""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'''       )





