



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


lowerCamelCase_							:Any       = get_tests_dir('''fixtures''')







class      _lowerCAmelCase							(      unittest.TestCase			):


	def    _a     (self							):
				# A mock response for an HTTP head request to emulate server down
				A_     :						List[Any]     =					mock.Mock()
				A_     :						str     =					500
				A_     :						Union[str, Any]     =					{}
				A_     :						Tuple     =					HTTPError
				A_     :						str     =					{}

				# Download this model to make sure it's in the cache.
				A_     :						Tuple     =					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=lowerCamelCase_							) as mock_head:
							A_     :						int     =					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							):
				# This test is for deprecated behavior and can be removed in v5
				A_     :						Optional[Any]     =					WavaVecaFeatureExtractor.from_pretrained(
				    """https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json"""							)







@is_staging_test
class      _lowerCAmelCase							(      unittest.TestCase			):


	@classmethod
	def    _a     (cls							):
				A_     :						int     =					TOKEN
				HfFolder.save_token(lowerCamelCase_							)


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

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

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

				A_     :						Optional[Any]     =					WavaVecaFeatureExtractor.from_pretrained(F'{USER}/test-feature-extractor'							)
				for k, v in feature_extractor.__dict__.items():
							self.assertEqual(lowerCamelCase_	,				getattr(lowerCamelCase_	,				lowerCamelCase_							)							)


	def    _a     (self							):
				A_     :						Any     =					WavaVecaFeatureExtractor.from_pretrained(lowerCamelCase_							)
				feature_extractor.push_to_hub("""valid_org/test-feature-extractor"""	,				use_auth_token=self._token							)

				A_     :						List[Any]     =					WavaVecaFeatureExtractor.from_pretrained("""valid_org/test-feature-extractor"""							)
				for k, v in feature_extractor.__dict__.items():
							self.assertEqual(lowerCamelCase_	,				getattr(lowerCamelCase_	,				lowerCamelCase_							)							)

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

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


	def    _a     (self							):
				CustomFeatureExtractor.register_for_auto_class()
				A_     :						Optional[int]     =					CustomFeatureExtractor.from_pretrained(lowerCamelCase_							)

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

				A_     :						Tuple     =					AutoFeatureExtractor.from_pretrained(
				    F'{USER}/test-dynamic-feature-extractor'	,				trust_remote_code=lowerCamelCase_							)
				# 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'''




from typing import Dict

from .base import GenericTensor, Pipeline







class      _lowerCAmelCase							(      __UpperCAmelCase			):


	def    _a     (self	,				lowercase=None	,				lowercase=None	,				lowercase=None	,				**lowercase							):
				if tokenize_kwargs is None:
							A_     :						Optional[Any]     =					{}

				if truncation is not None:
							if "truncation" in tokenize_kwargs:
										raise ValueError(
										    """truncation parameter defined twice (given as keyword argument as well as in tokenize_kwargs)"""							)
							A_     :						str     =					truncation

				A_     :						List[str]     =					tokenize_kwargs

				A_     :						Dict     =					{}
				if return_tensors is not None:
							A_     :						List[Any]     =					return_tensors

				return preprocess_params, {}, postprocess_params


	def    _a     (self	,				lowercase	,				**lowercase							):
				A_     :						Optional[int]     =					self.framework
				A_     :						str     =					self.tokenizer(lowercase	,				return_tensors=lowercase	,				**lowercase							)
				return model_inputs


	def    _a     (self	,				lowercase							):
				A_     :						str     =					self.model(**lowercase							)
				return model_outputs


	def    _a     (self	,				lowercase	,				lowercase=False							):
				# [0] is the first available tensor, logits or last_hidden_state.
				if return_tensors:
							return model_outputs[0]
				if self.framework == "pt":
							return model_outputs[0].tolist()
				elif self.framework == "tf":
							return model_outputs[0].numpy().tolist()


	def __call__(self	,				*lowercase	,				**lowercase							):
				return super().__call__(*lowercase	,				**lowercase							)