



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


SCREAMING_SNAKE_CASE__      					=     get_tests_dir("fixtures")


class 				snake_case (unittest.TestCase   ):



       def 							_a      (				self      )  ->							Optional[int]:
           # A mock response for an HTTP head request to emulate server down
           lowercase__		       =						mock.Mock()
           lowercase__		       =						500
           lowercase__		       =						{}
           lowercase__		       =						HTTPError
           lowercase__		       =						{}

           # Download this model to make sure it's in the cache.
           lowercase__		       =						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=UpperCAmelCase_      ) as mock_head:
               lowercase__		       =						WavaVecaFeatureExtractor.from_pretrained("hf-internal-testing/tiny-random-wav2vec2"      )
               # This check we did call the fake head request
               mock_head.assert_called()







       def 							_a      (				self      )  ->							Tuple:
           # This test is for deprecated behavior and can be removed in v5
           lowercase__		       =						WavaVecaFeatureExtractor.from_pretrained(
               "https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json"      )





@is_staging_test
class 				snake_case (unittest.TestCase   ):



       @classmethod
       def 							_a      (				cls      )  ->							Optional[Any]:
           lowercase__		       =						TOKEN
           HfFolder.save_token(UpperCAmelCase_      )



       @classmethod
       def 							_a      (				cls      )  ->							List[Any]:
           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 							_a      (				self      )  ->							List[str]:
           lowercase__		       =						WavaVecaFeatureExtractor.from_pretrained(UpperCAmelCase_      )
           feature_extractor.push_to_hub("test-feature-extractor"     ,use_auth_token=self._token      )

           lowercase__		       =						WavaVecaFeatureExtractor.from_pretrained(F'''{USER}/test-feature-extractor'''      )
           for k, v in feature_extractor.__dict__.items():
               self.assertEqual(UpperCAmelCase_     ,getattr(UpperCAmelCase_     ,UpperCAmelCase_      )      )

           # 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(
                   UpperCAmelCase_     ,repo_id="test-feature-extractor"     ,push_to_hub=UpperCAmelCase_     ,use_auth_token=self._token      )

           lowercase__		       =						WavaVecaFeatureExtractor.from_pretrained(F'''{USER}/test-feature-extractor'''      )
           for k, v in feature_extractor.__dict__.items():
               self.assertEqual(UpperCAmelCase_     ,getattr(UpperCAmelCase_     ,UpperCAmelCase_      )      )



       def 							_a      (				self      )  ->							Tuple:
           lowercase__		       =						WavaVecaFeatureExtractor.from_pretrained(UpperCAmelCase_      )
           feature_extractor.push_to_hub("valid_org/test-feature-extractor"     ,use_auth_token=self._token      )

           lowercase__		       =						WavaVecaFeatureExtractor.from_pretrained("valid_org/test-feature-extractor"      )
           for k, v in feature_extractor.__dict__.items():
               self.assertEqual(UpperCAmelCase_     ,getattr(UpperCAmelCase_     ,UpperCAmelCase_      )      )

           # 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(
                   UpperCAmelCase_     ,repo_id="valid_org/test-feature-extractor-org"     ,push_to_hub=UpperCAmelCase_     ,use_auth_token=self._token      )

           lowercase__		       =						WavaVecaFeatureExtractor.from_pretrained("valid_org/test-feature-extractor-org"      )
           for k, v in feature_extractor.__dict__.items():
               self.assertEqual(UpperCAmelCase_     ,getattr(UpperCAmelCase_     ,UpperCAmelCase_      )      )







       def 							_a      (				self      )  ->							List[str]:
           CustomFeatureExtractor.register_for_auto_class()
           lowercase__		       =						CustomFeatureExtractor.from_pretrained(UpperCAmelCase_      )

           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"}     ,)

           lowercase__		       =						AutoFeatureExtractor.from_pretrained(
               F'''{USER}/test-dynamic-feature-extractor'''     ,trust_remote_code=UpperCAmelCase_      )
           # 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"      )









'''simple docstring'''





import os
from typing import Dict, List, Union

import tensorflow as tf
from keras_nlp.tokenizers import BytePairTokenizer
from tensorflow_text import pad_model_inputs

from .tokenization_gpta import GPTaTokenizer


class 				snake_case (tf.keras.layers.Layer   ):



       def __init__(				self     ,UpperCAmelCase_     ,UpperCAmelCase_     ,UpperCAmelCase_ = None     ,UpperCAmelCase_ = None      )  ->							List[Any]:
           super().__init__()
           lowercase__		       =						pad_token_id
           lowercase__		       =						max_length
           lowercase__		       =						vocab
           lowercase__		       =						merges
           lowercase__		       =						BytePairTokenizer(UpperCAmelCase_     ,UpperCAmelCase_     ,sequence_length=UpperCAmelCase_      )



       @classmethod
       def 							_a      (				cls     ,UpperCAmelCase_     ,*UpperCAmelCase_     ,**UpperCAmelCase_      )  ->							List[str]:
           lowercase__		       =						[" ".join(UpperCAmelCase_      ) for m in tokenizer.bpe_ranks.keys()]
           lowercase__		       =						tokenizer.get_vocab()
           return cls(UpperCAmelCase_     ,UpperCAmelCase_     ,*UpperCAmelCase_     ,**UpperCAmelCase_      )



       @classmethod
       def 							_a      (				cls     ,UpperCAmelCase_     ,*UpperCAmelCase_     ,**UpperCAmelCase_      )  ->							Optional[Any]:
           lowercase__		       =						GPTaTokenizer.from_pretrained(UpperCAmelCase_     ,*UpperCAmelCase_     ,**UpperCAmelCase_      )
           return cls.from_tokenizer(UpperCAmelCase_     ,*UpperCAmelCase_     ,**UpperCAmelCase_      )



       @classmethod
       def 							_a      (				cls     ,UpperCAmelCase_      )  ->							Optional[Any]:
           return cls(**UpperCAmelCase_      )



       def 							_a      (				self      )  ->							Tuple:
           return {
               "vocab": self.vocab,
               "merges": self.merges,
               "max_length": self.max_length,
               "pad_token_id": self.pad_token_id,
           }







       def 							_a      (				self     ,UpperCAmelCase_     ,UpperCAmelCase_ = None      )  ->							Tuple:
           lowercase__		       =						self.tf_tokenizer(UpperCAmelCase_      )
           lowercase__		       =						tf.ones_like(UpperCAmelCase_      )

           if self.pad_token_id is not None:
               # pad the tokens up to max length
               lowercase__		       =						max_length if max_length is not None else self.max_length

               if max_length is not None:
                   lowercase__       ,       lowercase__		       =						pad_model_inputs(
                       UpperCAmelCase_     ,max_seq_length=UpperCAmelCase_     ,pad_value=self.pad_token_id      )

           return {"attention_mask": attention_mask, "input_ids": input_ids}




