

'''simple docstring'''




import json
import os
import shutil
import tempfile
from unittest import TestCase

from transformers import BartTokenizer, BartTokenizerFast, DPRQuestionEncoderTokenizer, DPRQuestionEncoderTokenizerFast
from transformers.models.bart.configuration_bart import BartConfig
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES as DPR_VOCAB_FILES_NAMES
from transformers.models.dpr.configuration_dpr import DPRConfig
from transformers.models.roberta.tokenization_roberta import VOCAB_FILES_NAMES as BART_VOCAB_FILES_NAMES
from transformers.testing_utils import require_faiss, require_tokenizers, require_torch, slow
from transformers.utils import is_datasets_available, is_faiss_available, is_torch_available


if is_torch_available() and is_datasets_available() and is_faiss_available():
      from transformers.models.rag.configuration_rag import RagConfig
      from transformers.models.rag.tokenization_rag import RagTokenizer






@require_faiss
@require_torch
class   _SCREAMING_SNAKE_CASE   (   __lowercase     ):



 def 			snake_case__  (   self		:   Any      ):
      __magic_name__						       =    tempfile.mkdtemp()
      __magic_name__						       =    8

      # DPR tok
      __magic_name__						       =    [
          """[UNK]""",
          """[CLS]""",
          """[SEP]""",
          """[PAD]""",
          """[MASK]""",
          """want""",
          """##want""",
          """##ed""",
          """wa""",
          """un""",
          """runn""",
          """##ing""",
          """,""",
          """low""",
          """lowest""",
      ]
      __magic_name__						       =    os.path.join(self.tmpdirname			,				'''dpr_tokenizer'''      )
      os.makedirs(a__			,				exist_ok=a__      )
      __magic_name__						       =    os.path.join(a__			,				DPR_VOCAB_FILES_NAMES['''vocab_file''']      )
      with open(self.vocab_file			,				'''w'''			,				encoding='''utf-8'''      ) as vocab_writer:
           vocab_writer.write(''''''.join([x + '''\n''' for x in vocab_tokens]      )      )

      # BART tok
      __magic_name__						       =    [
          """l""",
          """o""",
          """w""",
          """e""",
          """r""",
          """s""",
          """t""",
          """i""",
          """d""",
          """n""",
          """\u0120""",
          """\u0120l""",
          """\u0120n""",
          """\u0120lo""",
          """\u0120low""",
          """er""",
          """\u0120lowest""",
          """\u0120newer""",
          """\u0120wider""",
          """<unk>""",
      ]
      __magic_name__						       =    dict(zip(a__			,				range(len(a__      )      )      )      )
      __magic_name__						       =    ["""#version: 0.2""", """\u0120 l""", """\u0120l o""", """\u0120lo w""", """e r""", """"""]
      __magic_name__						       =    {"""unk_token""": """<unk>"""}

      __magic_name__						       =    os.path.join(self.tmpdirname			,				'''bart_tokenizer'''      )
      os.makedirs(a__			,				exist_ok=a__      )
      __magic_name__						       =    os.path.join(a__			,				BART_VOCAB_FILES_NAMES['''vocab_file''']      )
      __magic_name__						       =    os.path.join(a__			,				BART_VOCAB_FILES_NAMES['''merges_file''']      )
      with open(self.vocab_file			,				'''w'''			,				encoding='''utf-8'''      ) as fp:
           fp.write(json.dumps(a__      ) + '''\n'''      )
      with open(self.merges_file			,				'''w'''			,				encoding='''utf-8'''      ) as fp:
           fp.write('''\n'''.join(a__      )      )



 def 			snake_case__  (   self		:   Tuple      ):
      return DPRQuestionEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname			,				'''dpr_tokenizer'''      )      )



 def 			snake_case__  (   self		:   Optional[int]      ):
      return BartTokenizer.from_pretrained(os.path.join(self.tmpdirname			,				'''bart_tokenizer'''      )      )



 def 			snake_case__  (   self		:   Tuple      ):
      shutil.rmtree(self.tmpdirname      )



 @require_tokenizers
 def 			snake_case__  (   self		:   List[Any]      ):
      __magic_name__						       =    os.path.join(self.tmpdirname			,				'''rag_tokenizer'''      )
      __magic_name__						       =    RagConfig(question_encoder=DPRConfig().to_dict()			,				generator=BartConfig().to_dict()      )
      __magic_name__						       =    RagTokenizer(question_encoder=self.get_dpr_tokenizer()			,				generator=self.get_bart_tokenizer()      )
      rag_config.save_pretrained(a__      )
      rag_tokenizer.save_pretrained(a__      )
      __magic_name__						       =    RagTokenizer.from_pretrained(a__			,				config=a__      )
      self.assertIsInstance(new_rag_tokenizer.question_encoder			,				a__      )
      self.assertEqual(new_rag_tokenizer.question_encoder.get_vocab()			,				rag_tokenizer.question_encoder.get_vocab()      )
      self.assertIsInstance(new_rag_tokenizer.generator			,				a__      )
      self.assertEqual(new_rag_tokenizer.generator.get_vocab()			,				rag_tokenizer.generator.get_vocab()      )



 @slow
 def 			snake_case__  (   self		:   List[str]      ):
      __magic_name__						       =    RagTokenizer.from_pretrained('''facebook/rag-token-nq'''      )
      __magic_name__						       =    [
          """who got the first nobel prize in physics""",
          """when is the next deadpool movie being released""",
          """which mode is used for short wave broadcast service""",
          """who is the owner of reading football club""",
          """when is the next scandal episode coming out""",
          """when is the last time the philadelphia won the superbowl""",
          """what is the most current adobe flash player version""",
          """how many episodes are there in dragon ball z""",
          """what is the first step in the evolution of the eye""",
          """where is gall bladder situated in human body""",
          """what is the main mineral in lithium batteries""",
          """who is the president of usa right now""",
          """where do the greasers live in the outsiders""",
          """panda is a national animal of which country""",
          """what is the name of manchester united stadium""",
      ]
      __magic_name__						       =    tokenizer(a__      )
      self.assertIsNotNone(a__      )



 @slow
 def 			snake_case__  (   self		:   Dict      ):
      __magic_name__						       =    RagTokenizer.from_pretrained('''facebook/rag-sequence-nq'''      )
      __magic_name__						       =    [
          """who got the first nobel prize in physics""",
          """when is the next deadpool movie being released""",
          """which mode is used for short wave broadcast service""",
          """who is the owner of reading football club""",
          """when is the next scandal episode coming out""",
          """when is the last time the philadelphia won the superbowl""",
          """what is the most current adobe flash player version""",
          """how many episodes are there in dragon ball z""",
          """what is the first step in the evolution of the eye""",
          """where is gall bladder situated in human body""",
          """what is the main mineral in lithium batteries""",
          """who is the president of usa right now""",
          """where do the greasers live in the outsiders""",
          """panda is a national animal of which country""",
          """what is the name of manchester united stadium""",
      ]
      __magic_name__						       =    tokenizer(a__      )
      self.assertIsNotNone(a__      )




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


_lowerCAmelCase         =						get_tests_dir("fixtures")
_lowerCAmelCase         =						get_tests_dir("fixtures/dummy_feature_extractor_config.json")
_lowerCAmelCase         =						get_tests_dir("fixtures/dummy-config.json")






class   _SCREAMING_SNAKE_CASE   (   unittest.TestCase     ):



 def 			snake_case__  (   self		:   Union[str, Any]      ):
      __magic_name__						       =    0



 def 			snake_case__  (   self		:   Optional[int]      ):
      __magic_name__						       =    AutoFeatureExtractor.from_pretrained('''facebook/wav2vec2-base-960h'''      )
      self.assertIsInstance(a__			,				a__      )



 def 			snake_case__  (   self		:   Optional[int]      ):
      __magic_name__						       =    AutoFeatureExtractor.from_pretrained(a__      )
      self.assertIsInstance(a__			,				a__      )



 def 			snake_case__  (   self		:   Optional[Any]      ):
      with tempfile.TemporaryDirectory() as tmpdirname:
           __magic_name__						       =    WavaVecaConfig()

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

           config_dict.pop('''feature_extractor_type'''      )
           __magic_name__						       =    WavaVecaFeatureExtractor(**a__      )

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

           __magic_name__						       =    AutoFeatureExtractor.from_pretrained(a__      )

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

      self.assertIsInstance(a__			,				a__      )



 def 			snake_case__  (   self		:   Optional[Any]      ):
      __magic_name__						       =    AutoFeatureExtractor.from_pretrained(a__      )
      self.assertIsInstance(a__			,				a__      )



 def 			snake_case__  (   self		:   str      ):
      with self.assertRaisesRegex(
          a__			,				'''bert-base is not a local folder and is not a valid model identifier'''      ):
           __magic_name__						       =    AutoFeatureExtractor.from_pretrained('''bert-base'''      )



 def 			snake_case__  (   self		:   str      ):
      with self.assertRaisesRegex(
          a__			,				r'''aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)'''      ):
           __magic_name__						       =    AutoFeatureExtractor.from_pretrained(a__			,				revision='''aaaaaa'''      )



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



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

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

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



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

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

      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 			snake_case__  (   self		:   int      ):






      class   _SCREAMING_SNAKE_CASE   (   __a     ):
       __SCREAMING_SNAKE_CASE					:Optional[int]      	=					True

      try:
           AutoConfig.register('''custom'''			,				a__      )
           AutoFeatureExtractor.register(a__			,				a__      )
           # If remote code is not set, the default is to use local
           __magic_name__						       =    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.
           __magic_name__						       =    AutoFeatureExtractor.from_pretrained(
               '''hf-internal-testing/test_dynamic_feature_extractor'''			,				trust_remote_code=a__      )
           self.assertEqual(feature_extractor.__class__.__name__			,				'''NewFeatureExtractor'''      )
           self.assertTrue(feature_extractor.is_local      )

           # If remote is enabled, we load from the Hub
           __magic_name__						       =    AutoFeatureExtractor.from_pretrained(
               '''hf-internal-testing/test_dynamic_feature_extractor'''			,				trust_remote_code=a__      )
           self.assertEqual(feature_extractor.__class__.__name__			,				'''NewFeatureExtractor'''      )
           self.assertTrue(not hasattr(a__			,				'''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]






