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


_A     =  get_tests_dir("fixtures")





class 			_lowerCAmelCase      (			unittest.TestCase ):




				def   __a		(      self  )       ->   str:
							# A mock response for an HTTP head request to emulate server down
							lowerCAmelCase_      			=			mock.Mock()
							lowerCAmelCase_      			=			500
							lowerCAmelCase_      			=			{}
							lowerCAmelCase_      			=			HTTPError
							lowerCAmelCase_      			=			{}

							# Download this model to make sure it's in the cache.
							lowerCAmelCase_      			=			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:
										lowerCAmelCase_      			=			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  )       ->   Dict:
							# This test is for deprecated behavior and can be removed in v5
							lowerCAmelCase_      			=			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  )       ->   int:
							lowerCAmelCase_      			=			TOKEN
							HfFolder.save_token(_UpperCamelCase  )




				@classmethod
				def   __a		(      cls  )       ->   int:
							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  )       ->   Optional[int]:
							lowerCAmelCase_      			=			WavaVecaFeatureExtractor.from_pretrained(_UpperCamelCase  )
							feature_extractor.push_to_hub("test-feature-extractor"       ,       use_auth_token=self._token  )

							lowerCAmelCase_      			=			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  )

							lowerCAmelCase_      			=			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  )       ->   Optional[Any]:
							lowerCAmelCase_      			=			WavaVecaFeatureExtractor.from_pretrained(_UpperCamelCase  )
							feature_extractor.push_to_hub("valid_org/test-feature-extractor"       ,       use_auth_token=self._token  )

							lowerCAmelCase_      			=			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  )

							lowerCAmelCase_      			=			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  )       ->   Tuple:
							CustomFeatureExtractor.register_for_auto_class()
							lowerCAmelCase_      			=			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"}       ,       )

							lowerCAmelCase_      			=			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"  )





from ...processing_utils import ProcessorMixin





class 			_lowerCAmelCase      (			__a ):
				_lowercase    	='''SpeechT5FeatureExtractor'''
				_lowercase    	='''SpeechT5Tokenizer'''




				def __init__(      self       ,       _UpperCamelCase       ,       _UpperCamelCase  )       ->   int:
							super().__init__(_UpperCamelCase       ,       _UpperCamelCase  )




				def __call__(      self       ,       *_UpperCamelCase       ,       **_UpperCamelCase  )       ->   Optional[int]:
							lowerCAmelCase_      			=			kwargs.pop("audio"       ,       _UpperCamelCase  )
							lowerCAmelCase_      			=			kwargs.pop("text"       ,       _UpperCamelCase  )
							lowerCAmelCase_      			=			kwargs.pop("text_target"       ,       _UpperCamelCase  )
							lowerCAmelCase_      			=			kwargs.pop("audio_target"       ,       _UpperCamelCase  )
							lowerCAmelCase_      			=			kwargs.pop("sampling_rate"       ,       _UpperCamelCase  )

							if audio is not None and text is not None:
										raise ValueError(
										    "Cannot process both `audio` and `text` inputs. Did you mean `audio_target` or `text_target`?"  )
							if audio_target is not None and text_target is not None:
										raise ValueError(
										    "Cannot process both `audio_target` and `text_target` inputs. Did you mean `audio` or `text`?"  )
							if audio is None and audio_target is None and text is None and text_target is None:
										raise ValueError(
										    "You need to specify either an `audio`, `audio_target`, `text`, or `text_target` input to process."  )

							if audio is not None:
										lowerCAmelCase_      			=			self.feature_extractor(_UpperCamelCase       ,       *_UpperCamelCase       ,       sampling_rate=_UpperCamelCase       ,       **_UpperCamelCase  )
							elif text is not None:
										lowerCAmelCase_      			=			self.tokenizer(_UpperCamelCase       ,       **_UpperCamelCase  )
							else:
										lowerCAmelCase_      			=			None

							if audio_target is not None:
										lowerCAmelCase_      			=			self.feature_extractor(audio_target=_UpperCamelCase       ,       *_UpperCamelCase       ,       sampling_rate=_UpperCamelCase       ,       **_UpperCamelCase  )
										lowerCAmelCase_      			=			targets["input_values"]
							elif text_target is not None:
										lowerCAmelCase_      			=			self.tokenizer(_UpperCamelCase       ,       **_UpperCamelCase  )
										lowerCAmelCase_      			=			targets["input_ids"]
							else:
										lowerCAmelCase_      			=			None

							if inputs is None:
										return targets

							if targets is not None:
										lowerCAmelCase_      			=			labels

										lowerCAmelCase_      			=			targets.get("attention_mask"  )
										if decoder_attention_mask is not None:
													lowerCAmelCase_      			=			decoder_attention_mask

							return inputs




				def   __a		(      self       ,       *_UpperCamelCase       ,       **_UpperCamelCase  )       ->   str:
							lowerCAmelCase_      			=			kwargs.pop("input_values"       ,       _UpperCamelCase  )
							lowerCAmelCase_      			=			kwargs.pop("input_ids"       ,       _UpperCamelCase  )
							lowerCAmelCase_      			=			kwargs.pop("labels"       ,       _UpperCamelCase  )

							if input_values is not None and input_ids is not None:
										raise ValueError("Cannot process both `input_values` and `input_ids` inputs."  )
							if input_values is None and input_ids is None and labels is None:
										raise ValueError(
										    "You need to specify either an `input_values`, `input_ids`, or `labels` input to be padded."  )

							if input_values is not None:
										lowerCAmelCase_      			=			self.feature_extractor.pad(_UpperCamelCase       ,       *_UpperCamelCase       ,       **_UpperCamelCase  )
							elif input_ids is not None:
										lowerCAmelCase_      			=			self.tokenizer.pad(_UpperCamelCase       ,       **_UpperCamelCase  )
							else:
										lowerCAmelCase_      			=			None

							if labels is not None:
										if "input_ids" in labels or (isinstance(_UpperCamelCase       ,       _UpperCamelCase  ) and "input_ids" in labels[0]):
													lowerCAmelCase_      			=			self.tokenizer.pad(_UpperCamelCase       ,       **_UpperCamelCase  )
													lowerCAmelCase_      			=			targets["input_ids"]
										else:
													lowerCAmelCase_      			=			self.feature_extractor.feature_size
													lowerCAmelCase_      			=			self.feature_extractor.num_mel_bins
													lowerCAmelCase_      			=			self.feature_extractor.pad(_UpperCamelCase       ,       *_UpperCamelCase       ,       **_UpperCamelCase  )
													lowerCAmelCase_      			=			feature_size_hack
													lowerCAmelCase_      			=			targets["input_values"]
							else:
										lowerCAmelCase_      			=			None

							if inputs is None:
										return targets

							if targets is not None:
										lowerCAmelCase_      			=			labels

										lowerCAmelCase_      			=			targets.get("attention_mask"  )
										if decoder_attention_mask is not None:
													lowerCAmelCase_      			=			decoder_attention_mask

							return inputs




				def   __a		(      self       ,       *_UpperCamelCase       ,       **_UpperCamelCase  )       ->   List[Any]:
							return self.tokenizer.batch_decode(*_UpperCamelCase       ,       **_UpperCamelCase  )




				def   __a		(      self       ,       *_UpperCamelCase       ,       **_UpperCamelCase  )       ->   Optional[Any]:
							return self.tokenizer.decode(*_UpperCamelCase       ,       **_UpperCamelCase  )



