






'''simple docstring'''




import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
from shutil import copyfile

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

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    PROCESSOR_MAPPING,
    TOKENIZER_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    AutoProcessor,
    AutoTokenizer,
    BertTokenizer,
    ProcessorMixin,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
)
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test
from transformers.tokenization_utils import TOKENIZER_CONFIG_FILE
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_tokenizers_available


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
from test_module.custom_processing import CustomProcessor  # noqa E402
from test_module.custom_tokenization import CustomTokenizer  # noqa E402


_lowerCAmelCase         =						get_tests_dir("fixtures/dummy_feature_extractor_config.json")
_lowerCAmelCase         =						get_tests_dir("fixtures/vocab.json")
_lowerCAmelCase         =						get_tests_dir("fixtures")






class   _SCREAMING_SNAKE_CASE   (   unittest.TestCase     ):
 __SCREAMING_SNAKE_CASE					:Optional[Any]      	=					["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]



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



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



 def 			snake_case__  (   self		:   int      ):
      with tempfile.TemporaryDirectory() as tmpdirname:
           __magic_name__						       =    WavaVecaConfig()
           __magic_name__						       =    AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''      )

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

           __magic_name__						       =    AutoProcessor.from_pretrained(a__      )

      self.assertIsInstance(a__			,				a__      )



 def 			snake_case__  (   self		:   Optional[int]      ):
      with tempfile.TemporaryDirectory() as tmpdirname:
           # copy relevant files
           copyfile(a__			,				os.path.join(a__			,				a__      )      )
           copyfile(a__			,				os.path.join(a__			,				'''vocab.json'''      )      )

           __magic_name__						       =    AutoProcessor.from_pretrained(a__      )

      self.assertIsInstance(a__			,				a__      )



 def 			snake_case__  (   self		:   Any      ):
      with tempfile.TemporaryDirectory() as tmpdirname:
           __magic_name__						       =    WavaVecaFeatureExtractor()
           __magic_name__						       =    AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h'''      )

           __magic_name__						       =    WavaVecaProcessor(a__			,				a__      )

           # save in new folder
           processor.save_pretrained(a__      )

           # drop `processor_class` in tokenizer
           with open(os.path.join(a__			,				a__      )			,				'''r'''      ) as f:
                __magic_name__						       =    json.load(a__      )
                config_dict.pop('''processor_class'''      )

           with open(os.path.join(a__			,				a__      )			,				'''w'''      ) as f:
                f.write(json.dumps(a__      )      )

           __magic_name__						       =    AutoProcessor.from_pretrained(a__      )

      self.assertIsInstance(a__			,				a__      )



 def 			snake_case__  (   self		:   Optional[int]      ):
      with tempfile.TemporaryDirectory() as tmpdirname:
           __magic_name__						       =    WavaVecaFeatureExtractor()
           __magic_name__						       =    AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h'''      )

           __magic_name__						       =    WavaVecaProcessor(a__			,				a__      )

           # save in new folder
           processor.save_pretrained(a__      )

           # drop `processor_class` in feature extractor
           with open(os.path.join(a__			,				a__      )			,				'''r'''      ) as f:
                __magic_name__						       =    json.load(a__      )
                config_dict.pop('''processor_class'''      )

           with open(os.path.join(a__			,				a__      )			,				'''w'''      ) as f:
                f.write(json.dumps(a__      )      )

           __magic_name__						       =    AutoProcessor.from_pretrained(a__      )

      self.assertIsInstance(a__			,				a__      )



 def 			snake_case__  (   self		:   List[str]      ):
      with tempfile.TemporaryDirectory() as tmpdirname:
           __magic_name__						       =    WavaVecaConfig(processor_class='''Wav2Vec2Processor'''      )
           model_config.save_pretrained(a__      )
           # copy relevant files
           copyfile(a__			,				os.path.join(a__			,				'''vocab.json'''      )      )
           # create emtpy sample processor
           with open(os.path.join(a__			,				a__      )			,				'''w'''      ) as f:
                f.write('''{}'''      )

           __magic_name__						       =    AutoProcessor.from_pretrained(a__      )

      self.assertIsInstance(a__			,				a__      )



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

      __magic_name__						       =    AutoProcessor.from_pretrained('''hf-internal-testing/test_dynamic_processor'''			,				trust_remote_code=a__      )
      self.assertTrue(processor.special_attribute_present      )
      self.assertEqual(processor.__class__.__name__			,				'''NewProcessor'''      )

      __magic_name__						       =    processor.feature_extractor
      self.assertTrue(feature_extractor.special_attribute_present      )
      self.assertEqual(feature_extractor.__class__.__name__			,				'''NewFeatureExtractor'''      )

      __magic_name__						       =    processor.tokenizer
      self.assertTrue(tokenizer.special_attribute_present      )
      if is_tokenizers_available():
           self.assertEqual(tokenizer.__class__.__name__			,				'''NewTokenizerFast'''      )

           # Test we can also load the slow version
           __magic_name__						       =    AutoProcessor.from_pretrained(
               '''hf-internal-testing/test_dynamic_processor'''			,				trust_remote_code=a__			,				use_fast=a__      )
           __magic_name__						       =    new_processor.tokenizer
           self.assertTrue(new_tokenizer.special_attribute_present      )
           self.assertEqual(new_tokenizer.__class__.__name__			,				'''NewTokenizer'''      )
      else:
           self.assertEqual(tokenizer.__class__.__name__			,				'''NewTokenizer'''      )



 def 			snake_case__  (   self		:   Tuple      ):
      try:
           AutoConfig.register('''custom'''			,				a__      )
           AutoFeatureExtractor.register(a__			,				a__      )
           AutoTokenizer.register(a__			,				slow_tokenizer_class=a__      )
           AutoProcessor.register(a__			,				a__      )
           # Trying to register something existing in the Transformers library will raise an error
           with self.assertRaises(a__      ):
                AutoProcessor.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:
                __magic_name__						       =    os.path.join(a__			,				'''vocab.txt'''      )
                with open(a__			,				'''w'''			,				encoding='''utf-8'''      ) as vocab_writer:
                     vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]      )      )
                __magic_name__						       =    CustomTokenizer(a__      )

           __magic_name__						       =    CustomProcessor(a__			,				a__      )

           with tempfile.TemporaryDirectory() as tmp_dir:
                processor.save_pretrained(a__      )
                __magic_name__						       =    AutoProcessor.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]
           if CustomConfig in TOKENIZER_MAPPING._extra_content:
                del TOKENIZER_MAPPING._extra_content[CustomConfig]
           if CustomConfig in PROCESSOR_MAPPING._extra_content:
                del PROCESSOR_MAPPING._extra_content[CustomConfig]



 def 			snake_case__  (   self		:   List[Any]      ):






      class   _SCREAMING_SNAKE_CASE   (   SCREAMING_SNAKE_CASE__     ):
       __SCREAMING_SNAKE_CASE					:Union[str, Any]      	=					False






      class   _SCREAMING_SNAKE_CASE   (   SCREAMING_SNAKE_CASE__     ):
       __SCREAMING_SNAKE_CASE					:List[Any]      	=					False






      class   _SCREAMING_SNAKE_CASE   (   SCREAMING_SNAKE_CASE__     ):
       __SCREAMING_SNAKE_CASE					:Tuple      	=					"""AutoFeatureExtractor"""
       __SCREAMING_SNAKE_CASE					:Union[str, Any]      	=					"""AutoTokenizer"""
       __SCREAMING_SNAKE_CASE					:Dict      	=					False

      try:
           AutoConfig.register('''custom'''			,				a__      )
           AutoFeatureExtractor.register(a__			,				a__      )
           AutoTokenizer.register(a__			,				slow_tokenizer_class=a__      )
           AutoProcessor.register(a__			,				a__      )
           # If remote code is not set, the default is to use local classes.
           __magic_name__						       =    AutoProcessor.from_pretrained('''hf-internal-testing/test_dynamic_processor'''      )
           self.assertEqual(processor.__class__.__name__			,				'''NewProcessor'''      )
           self.assertFalse(processor.special_attribute_present      )
           self.assertFalse(processor.feature_extractor.special_attribute_present      )
           self.assertFalse(processor.tokenizer.special_attribute_present      )

           # If remote code is disabled, we load the local ones.
           __magic_name__						       =    AutoProcessor.from_pretrained(
               '''hf-internal-testing/test_dynamic_processor'''			,				trust_remote_code=a__      )
           self.assertEqual(processor.__class__.__name__			,				'''NewProcessor'''      )
           self.assertFalse(processor.special_attribute_present      )
           self.assertFalse(processor.feature_extractor.special_attribute_present      )
           self.assertFalse(processor.tokenizer.special_attribute_present      )

           # If remote is enabled, we load from the Hub.
           __magic_name__						       =    AutoProcessor.from_pretrained(
               '''hf-internal-testing/test_dynamic_processor'''			,				trust_remote_code=a__      )
           self.assertEqual(processor.__class__.__name__			,				'''NewProcessor'''      )
           self.assertTrue(processor.special_attribute_present      )
           self.assertTrue(processor.feature_extractor.special_attribute_present      )
           self.assertTrue(processor.tokenizer.special_attribute_present      )

      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]
           if CustomConfig in TOKENIZER_MAPPING._extra_content:
                del TOKENIZER_MAPPING._extra_content[CustomConfig]
           if CustomConfig in PROCESSOR_MAPPING._extra_content:
                del PROCESSOR_MAPPING._extra_content[CustomConfig]



 def 			snake_case__  (   self		:   str      ):
      __magic_name__						       =    AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-bert'''      )
      self.assertEqual(processor.__class__.__name__			,				'''BertTokenizerFast'''      )



 def 			snake_case__  (   self		:   Tuple      ):
      __magic_name__						       =    AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-convnext'''      )
      self.assertEqual(processor.__class__.__name__			,				'''ConvNextImageProcessor'''      )






@is_staging_test
class   _SCREAMING_SNAKE_CASE   (   unittest.TestCase     ):
 __SCREAMING_SNAKE_CASE					:Union[str, Any]      	=					["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]



 @classmethod
 def 			snake_case__  (   cls		:   Optional[int]      ):
      __magic_name__						       =    TOKEN
      HfFolder.save_token(a__      )



 @classmethod
 def 			snake_case__  (   cls		:   Optional[int]      ):
      try:
           delete_repo(token=cls._token			,				repo_id='''test-processor'''      )
      except HTTPError:
           pass

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

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



 def 			snake_case__  (   self		:   Union[str, Any]      ):
      __magic_name__						       =    WavaVecaProcessor.from_pretrained(a__      )
      with tempfile.TemporaryDirectory() as tmp_dir:
           processor.save_pretrained(
               os.path.join(a__			,				'''test-processor'''      )			,				push_to_hub=a__			,				use_auth_token=self._token      )

           __magic_name__						       =    WavaVecaProcessor.from_pretrained(F'''{USER}/test-processor'''      )
           for k, v in processor.feature_extractor.__dict__.items():
                self.assertEqual(a__			,				getattr(new_processor.feature_extractor			,				a__      )      )
           self.assertDictEqual(new_processor.tokenizer.get_vocab()			,				processor.tokenizer.get_vocab()      )



 def 			snake_case__  (   self		:   Optional[Any]      ):
      __magic_name__						       =    WavaVecaProcessor.from_pretrained(a__      )

      with tempfile.TemporaryDirectory() as tmp_dir:
           processor.save_pretrained(
               os.path.join(a__			,				'''test-processor-org'''      )			,				push_to_hub=a__			,				use_auth_token=self._token			,				organization='''valid_org'''			,				)

           __magic_name__						       =    WavaVecaProcessor.from_pretrained('''valid_org/test-processor-org'''      )
           for k, v in processor.feature_extractor.__dict__.items():
                self.assertEqual(a__			,				getattr(new_processor.feature_extractor			,				a__      )      )
           self.assertDictEqual(new_processor.tokenizer.get_vocab()			,				processor.tokenizer.get_vocab()      )



 def 			snake_case__  (   self		:   Dict      ):
      CustomFeatureExtractor.register_for_auto_class()
      CustomTokenizer.register_for_auto_class()
      CustomProcessor.register_for_auto_class()

      __magic_name__						       =    CustomFeatureExtractor.from_pretrained(a__      )

      with tempfile.TemporaryDirectory() as tmp_dir:
           __magic_name__						       =    os.path.join(a__			,				'''vocab.txt'''      )
           with open(a__			,				'''w'''			,				encoding='''utf-8'''      ) as vocab_writer:
                vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]      )      )
           __magic_name__						       =    CustomTokenizer(a__      )

      __magic_name__						       =    CustomProcessor(a__			,				a__      )

      with tempfile.TemporaryDirectory() as tmp_dir:
           create_repo(F'''{USER}/test-dynamic-processor'''			,				token=self._token      )
           __magic_name__						       =    Repository(a__			,				clone_from=F'''{USER}/test-dynamic-processor'''			,				token=self._token      )
           processor.save_pretrained(a__      )

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

           # This has added the proper auto_map field to the tokenizer config
           with open(os.path.join(a__			,				'''tokenizer_config.json'''      )      ) as f:
                __magic_name__						       =    json.load(a__      )
           self.assertDictEqual(
               tokenizer_config['''auto_map''']			,				{
                   '''AutoTokenizer''': ['''custom_tokenization.CustomTokenizer''', None],
                   '''AutoProcessor''': '''custom_processing.CustomProcessor''',
               }			,				)

           # The code has been copied from fixtures
           self.assertTrue(os.path.isfile(os.path.join(a__			,				'''custom_feature_extraction.py'''      )      )      )
           self.assertTrue(os.path.isfile(os.path.join(a__			,				'''custom_tokenization.py'''      )      )      )
           self.assertTrue(os.path.isfile(os.path.join(a__			,				'''custom_processing.py'''      )      )      )

           repo.push_to_hub()

      __magic_name__						       =    AutoProcessor.from_pretrained(F'''{USER}/test-dynamic-processor'''			,				trust_remote_code=a__      )
      # Can't make an isinstance check because the new_processor is from the CustomProcessor class of a dynamic module
      self.assertEqual(new_processor.__class__.__name__			,				'''CustomProcessor'''      )









'''simple docstring'''




import unittest

from transformers import (
    MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
    TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
    TextaTextGenerationPipeline,
    pipeline,
)
from transformers.testing_utils import is_pipeline_test, require_tf, require_torch
from transformers.utils import is_torch_available

from .test_pipelines_common import ANY


if is_torch_available():
      import torch






@is_pipeline_test
class   _SCREAMING_SNAKE_CASE   (   unittest.TestCase     ):
 __SCREAMING_SNAKE_CASE					:List[Any]      	=					MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING
 __SCREAMING_SNAKE_CASE					:Any      	=					TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING



 def 			snake_case__  (   self		:   Tuple			,				a__		:   Tuple			,				a__		:   int			,				a__		:   int      ):
      __magic_name__						       =    TextaTextGenerationPipeline(model=a__			,				tokenizer=a__      )
      return generator, ["Something to write", "Something else"]



 def 			snake_case__  (   self		:   List[str]			,				a__		:   List[Any]			,				a__		:   List[str]      ):
      __magic_name__						       =    generator('''Something there'''      )
      self.assertEqual(a__			,				[{'''generated_text''': ANY(a__      )}]      )
      # These are encoder decoder, they don't just append to incoming string
      self.assertFalse(outputs[0]['''generated_text'''].startswith('''Something there'''      )      )

      __magic_name__						       =    generator(['''This is great !''', '''Something else''']			,				num_return_sequences=2			,				do_sample=a__      )
      self.assertEqual(
          a__			,				[
              [{'''generated_text''': ANY(a__      )}, {'''generated_text''': ANY(a__      )}],
              [{'''generated_text''': ANY(a__      )}, {'''generated_text''': ANY(a__      )}],
          ]			,				)

      __magic_name__						       =    generator(
          ['''This is great !''', '''Something else''']			,				num_return_sequences=2			,				batch_size=2			,				do_sample=a__      )
      self.assertEqual(
          a__			,				[
              [{'''generated_text''': ANY(a__      )}, {'''generated_text''': ANY(a__      )}],
              [{'''generated_text''': ANY(a__      )}, {'''generated_text''': ANY(a__      )}],
          ]			,				)

      with self.assertRaises(a__      ):
           generator(4      )



 @require_torch
 def 			snake_case__  (   self		:   Any      ):
      __magic_name__						       =    pipeline('''text2text-generation'''			,				model='''patrickvonplaten/t5-tiny-random'''			,				framework='''pt'''      )
      # do_sample=False necessary for reproducibility
      __magic_name__						       =    generator('''Something there'''			,				do_sample=a__      )
      self.assertEqual(a__			,				[{'''generated_text''': ''''''}]      )

      __magic_name__						       =    3
      __magic_name__						       =    generator(
          '''Something there'''			,				num_return_sequences=a__			,				num_beams=a__			,				)
      __magic_name__						       =    [
          {'''generated_text''': '''Beide Beide Beide Beide Beide Beide Beide Beide Beide'''},
          {'''generated_text''': '''Beide Beide Beide Beide Beide Beide Beide Beide'''},
          {'''generated_text''': ''''''},
      ]
      self.assertEqual(a__			,				a__      )

      __magic_name__						       =    generator('''This is a test'''			,				do_sample=a__			,				num_return_sequences=2			,				return_tensors=a__      )
      self.assertEqual(
          a__			,				[
              {'''generated_token_ids''': ANY(torch.Tensor      )},
              {'''generated_token_ids''': ANY(torch.Tensor      )},
          ]			,				)
      __magic_name__						       =    generator.model.config.eos_token_id
      __magic_name__						       =    '''<pad>'''
      __magic_name__						       =    generator(
          ['''This is a test''', '''This is a second test''']			,				do_sample=a__			,				num_return_sequences=2			,				batch_size=2			,				return_tensors=a__			,				)
      self.assertEqual(
          a__			,				[
              [
                  {'''generated_token_ids''': ANY(torch.Tensor      )},
                  {'''generated_token_ids''': ANY(torch.Tensor      )},
              ],
              [
                  {'''generated_token_ids''': ANY(torch.Tensor      )},
                  {'''generated_token_ids''': ANY(torch.Tensor      )},
              ],
          ]			,				)



 @require_tf
 def 			snake_case__  (   self		:   int      ):
      __magic_name__						       =    pipeline('''text2text-generation'''			,				model='''patrickvonplaten/t5-tiny-random'''			,				framework='''tf'''      )
      # do_sample=False necessary for reproducibility
      __magic_name__						       =    generator('''Something there'''			,				do_sample=a__      )
      self.assertEqual(a__			,				[{'''generated_text''': ''''''}]      )






