

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


SCREAMING_SNAKE_CASE							:				List[str]     	=			"▁"

SCREAMING_SNAKE_CASE							:				List[Any]     	=			{"vocab_file": "spiece.model"}

SCREAMING_SNAKE_CASE							:				int     	=			{
    "vocab_file": {"google/pegasus-xsum": "https://huggingface.co/google/pegasus-xsum/resolve/main/spiece.model"}
}

SCREAMING_SNAKE_CASE							:				Optional[int]     	=			{
    "google/pegasus-xsum": 512,
}


SCREAMING_SNAKE_CASE							:				str     	=			logging.get_logger(__name__)






class    snake_case     (      lowercase_		):






				"""simple docstring"""
				_a    	=     VOCAB_FILES_NAMES

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



				def __init__(    self,     _lowercase,     _lowercase="<pad>",     _lowercase="</s>",     _lowercase="<unk>",     _lowercase="<mask_2>",     _lowercase="<mask_1>",     _lowercase=None,     _lowercase=103,     _lowercase = None,     **_lowercase,     )     ->							None:
											SCREAMING_SNAKE_CASE_				=				offset
											if additional_special_tokens is not None:
																		if not isinstance(_lowercase,     _lowercase       ):
																									raise TypeError(
																									    f"""additional_special_tokens should be of type {type(_lowercase       )}, but is"""
																									    f""" {type(_lowercase       )}"""       )

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

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

											SCREAMING_SNAKE_CASE_				=				{} if sp_model_kwargs is None else sp_model_kwargs

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

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

											SCREAMING_SNAKE_CASE_				=				{v: k for k, v in self.encoder.items()}



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



				def   a__					(    self       )     ->							Dict[str, int]:
											SCREAMING_SNAKE_CASE_				=				{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       )     ->							int:
											SCREAMING_SNAKE_CASE_				=				self.__dict__.copy()
											SCREAMING_SNAKE_CASE_				=				None
											return state



				def __setstate__(    self,     _lowercase       )     ->							Tuple:
											SCREAMING_SNAKE_CASE_				=				d

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

											SCREAMING_SNAKE_CASE_				=				spm.SentencePieceProcessor(**self.sp_model_kwargs       )
											self.sp_model.Load(self.vocab_file       )



				def   a__					(    self,     _lowercase       )     ->							List[str]:
											return self.sp_model.encode(_lowercase,     out_type=_lowercase       )



				def   a__					(    self,     _lowercase       )     ->							int:
											if token in self.decoder:
																		return self.decoder[token]
											elif token in self.added_tokens_decoder:
																		return self.added_tokens_decoder[token]
											SCREAMING_SNAKE_CASE_				=				self.sp_model.piece_to_id(_lowercase       )
											return sp_id + self.offset



				def   a__					(    self,     _lowercase       )     ->							str:
											if index in self.encoder:
																		return self.encoder[index]
											elif index in self.added_tokens_encoder:
																		return self.added_tokens_encoder[index]
											else:
																		SCREAMING_SNAKE_CASE_				=				self.sp_model.IdToPiece(index - self.offset       )
											return token



				def   a__					(    self,     _lowercase       )     ->							int:
											SCREAMING_SNAKE_CASE_				=				[]
											SCREAMING_SNAKE_CASE_				=				''
											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
																									SCREAMING_SNAKE_CASE_				=				[]
																		else:
																									current_sub_tokens.append(_lowercase       )
											out_string += self.sp_model.decode(_lowercase       )
											return out_string.strip()



				def   a__					(    self,     _lowercase=False       )     ->							Dict:
											return 1



				def   a__					(    self,     _lowercase       )     ->							List[Any]:
											SCREAMING_SNAKE_CASE_				=				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   a__					(    self,     _lowercase,     _lowercase = None,     _lowercase = False       )     ->							List[int]:
											if already_has_special_tokens:
																		return self._special_token_mask(_lowercase       )
											elif token_ids_a is None:
																		return self._special_token_mask(_lowercase       ) + [1]
											else:
																		return self._special_token_mask(token_ids_a + token_ids_a       ) + [1]



				def   a__					(    self,     _lowercase,     _lowercase=None       )     ->							List[int]:
											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   a__					(    self,     _lowercase,     _lowercase = None       )     ->							Tuple[str]:
											if not os.path.isdir(_lowercase       ):
																		logger.error(f"""Vocabulary path ({save_directory}) should be a directory"""       )
																		return
											SCREAMING_SNAKE_CASE_				=				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:
																									SCREAMING_SNAKE_CASE_				=				self.sp_model.serialized_model_proto()
																									fi.write(_lowercase       )

											return (out_vocab_file,)








'''simple docstring'''

import json
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
)
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER, get_tests_dir


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

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


SCREAMING_SNAKE_CASE							:				int     	=			get_tests_dir("fixtures")
SCREAMING_SNAKE_CASE							:				str     	=			get_tests_dir("fixtures/dummy_feature_extractor_config.json")
SCREAMING_SNAKE_CASE							:				str     	=			get_tests_dir("fixtures/dummy-config.json")






class    snake_case     (      unittest.TestCase		):






				"""simple docstring"""



				def   a__					(    self       )     ->							str:
											SCREAMING_SNAKE_CASE_				=				0



				def   a__					(    self       )     ->							Tuple:
											SCREAMING_SNAKE_CASE_				=				AutoFeatureExtractor.from_pretrained('facebook/wav2vec2-base-960h'       )
											self.assertIsInstance(_lowercase,     _lowercase       )



				def   a__					(    self       )     ->							List[str]:
											SCREAMING_SNAKE_CASE_				=				AutoFeatureExtractor.from_pretrained(_lowercase       )
											self.assertIsInstance(_lowercase,     _lowercase       )



				def   a__					(    self       )     ->							str:
											with tempfile.TemporaryDirectory() as tmpdirname:
																		SCREAMING_SNAKE_CASE_				=				WavaVecaConfig()

																		# remove feature_extractor_type to make sure config.json alone is enough to load feature processor locally
																		SCREAMING_SNAKE_CASE_				=				AutoFeatureExtractor.from_pretrained(_lowercase       ).to_dict()

																		config_dict.pop('feature_extractor_type'       )
																		SCREAMING_SNAKE_CASE_				=				WavaVecaFeatureExtractor(**_lowercase       )

																		# save in new folder
																		model_config.save_pretrained(_lowercase       )
																		config.save_pretrained(_lowercase       )

																		SCREAMING_SNAKE_CASE_				=				AutoFeatureExtractor.from_pretrained(_lowercase       )

																		# make sure private variable is not incorrectly saved
																		SCREAMING_SNAKE_CASE_				=				json.loads(config.to_json_string()       )
																		self.assertTrue('_processor_class' not in dict_as_saved       )

											self.assertIsInstance(_lowercase,     _lowercase       )



				def   a__					(    self       )     ->							int:
											SCREAMING_SNAKE_CASE_				=				AutoFeatureExtractor.from_pretrained(_lowercase       )
											self.assertIsInstance(_lowercase,     _lowercase       )



				def   a__					(    self       )     ->							Any:
											with self.assertRaisesRegex(
											    _lowercase,     'bert-base is not a local folder and is not a valid model identifier'       ):
																		SCREAMING_SNAKE_CASE_				=				AutoFeatureExtractor.from_pretrained('bert-base'       )



				def   a__					(    self       )     ->							List[Any]:
											with self.assertRaisesRegex(
											    _lowercase,     R'aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)'       ):
																		SCREAMING_SNAKE_CASE_				=				AutoFeatureExtractor.from_pretrained(_lowercase,     revision='aaaaaa'       )



				def   a__					(    self       )     ->							List[Any]:
											with self.assertRaisesRegex(
											    _lowercase,     'hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json.',     ):
																		SCREAMING_SNAKE_CASE_				=				AutoFeatureExtractor.from_pretrained('hf-internal-testing/config-no-model'       )



				def   a__					(    self       )     ->							List[str]:
											# If remote code is not set, we will time out when asking whether to load the model.
											with self.assertRaises(_lowercase       ):
																		SCREAMING_SNAKE_CASE_				=				AutoFeatureExtractor.from_pretrained(
																		    'hf-internal-testing/test_dynamic_feature_extractor'       )
											# If remote code is disabled, we can't load this config.
											with self.assertRaises(_lowercase       ):
																		SCREAMING_SNAKE_CASE_				=				AutoFeatureExtractor.from_pretrained(
																		    'hf-internal-testing/test_dynamic_feature_extractor',     trust_remote_code=_lowercase       )

											SCREAMING_SNAKE_CASE_				=				AutoFeatureExtractor.from_pretrained(
											    'hf-internal-testing/test_dynamic_feature_extractor',     trust_remote_code=_lowercase       )
											self.assertEqual(feature_extractor.__class__.__name__,     'NewFeatureExtractor'       )

											# Test feature extractor can be reloaded.
											with tempfile.TemporaryDirectory() as tmp_dir:
																		feature_extractor.save_pretrained(_lowercase       )
																		SCREAMING_SNAKE_CASE_				=				AutoFeatureExtractor.from_pretrained(_lowercase,     trust_remote_code=_lowercase       )
											self.assertEqual(reloaded_feature_extractor.__class__.__name__,     'NewFeatureExtractor'       )



				def   a__					(    self       )     ->							int:
											try:
																		AutoConfig.register('custom',     _lowercase       )
																		AutoFeatureExtractor.register(_lowercase,     _lowercase       )
																		# Trying to register something existing in the Transformers library will raise an error
																		with self.assertRaises(_lowercase       ):
																									AutoFeatureExtractor.register(_lowercase,     _lowercase       )

																		# Now that the config is registered, it can be used as any other config with the auto-API
																		SCREAMING_SNAKE_CASE_				=				CustomFeatureExtractor.from_pretrained(_lowercase       )
																		with tempfile.TemporaryDirectory() as tmp_dir:
																									feature_extractor.save_pretrained(_lowercase       )
																									SCREAMING_SNAKE_CASE_				=				AutoFeatureExtractor.from_pretrained(_lowercase       )
																									self.assertIsInstance(_lowercase,     _lowercase       )

											finally:
																		if "custom" in CONFIG_MAPPING._extra_content:
																									del CONFIG_MAPPING._extra_content["custom"]
																		if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
																									del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]





				def   a__					(    self       )     ->							str:






											class    snake_case     (      lowercase_		):






															"""simple docstring"""
															_a    	=     True

											try:
																		AutoConfig.register('custom',     _lowercase       )
																		AutoFeatureExtractor.register(_lowercase,     _lowercase       )
																		# If remote code is not set, the default is to use local
																		SCREAMING_SNAKE_CASE_				=				AutoFeatureExtractor.from_pretrained(
																		    'hf-internal-testing/test_dynamic_feature_extractor'       )
																		self.assertEqual(feature_extractor.__class__.__name__,     'NewFeatureExtractor'       )
																		self.assertTrue(feature_extractor.is_local       )

																		# If remote code is disabled, we load the local one.
																		SCREAMING_SNAKE_CASE_				=				AutoFeatureExtractor.from_pretrained(
																		    'hf-internal-testing/test_dynamic_feature_extractor',     trust_remote_code=_lowercase       )
																		self.assertEqual(feature_extractor.__class__.__name__,     'NewFeatureExtractor'       )
																		self.assertTrue(feature_extractor.is_local       )

																		# If remote is enabled, we load from the Hub
																		SCREAMING_SNAKE_CASE_				=				AutoFeatureExtractor.from_pretrained(
																		    'hf-internal-testing/test_dynamic_feature_extractor',     trust_remote_code=_lowercase       )
																		self.assertEqual(feature_extractor.__class__.__name__,     'NewFeatureExtractor'       )
																		self.assertTrue(not hasattr(_lowercase,     'is_local'       )       )

											finally:
																		if "custom" in CONFIG_MAPPING._extra_content:
																									del CONFIG_MAPPING._extra_content["custom"]
																		if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
																									del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]





