
"""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        _UpperCAmelCase(       unittest.TestCase      ):



		def 		UpperCAmelCase				(   self)			->			str:






							'''simple docstring'''
							# A mock response for an HTTP head request to emulate server down
							_UpperCamelCase									=    mock.Mock()
							_UpperCamelCase									=    5_00
							_UpperCamelCase									=    {}
							_UpperCamelCase									=    HTTPError
							_UpperCamelCase									=    {}

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




		def 		UpperCAmelCase				(   self)			->			Tuple:






							'''simple docstring'''
							# This test is for deprecated behavior and can be removed in v5
							_UpperCamelCase									=    WavaVecaFeatureExtractor.from_pretrained(
							    '''https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json''')






@is_staging_test
class        _UpperCAmelCase(       unittest.TestCase      ):



		@classmethod
		def 		UpperCAmelCase				(   cls)			->			Dict:






							'''simple docstring'''
							_UpperCamelCase									=    TOKEN
							HfFolder.save_token(A__)



		@classmethod
		def 		UpperCAmelCase				(   cls)			->			Optional[int]:






							'''simple docstring'''
							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 		UpperCAmelCase				(   self)			->			str:






							'''simple docstring'''
							_UpperCamelCase									=    WavaVecaFeatureExtractor.from_pretrained(A__)
							feature_extractor.push_to_hub('''test-feature-extractor'''							,						use_auth_token=self._token)

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

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

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



		def 		UpperCAmelCase				(   self)			->			int:






							'''simple docstring'''
							_UpperCamelCase									=    WavaVecaFeatureExtractor.from_pretrained(A__)
							feature_extractor.push_to_hub('''valid_org/test-feature-extractor'''							,						use_auth_token=self._token)

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

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

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




		def 		UpperCAmelCase				(   self)			->			List[Any]:






							'''simple docstring'''
							CustomFeatureExtractor.register_for_auto_class()
							_UpperCamelCase									=    CustomFeatureExtractor.from_pretrained(A__)

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

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














import flax.linen as nn
import jax.numpy as jnp

from .attention_flax import FlaxTransformeraDModel
from .resnet_flax import FlaxDownsampleaD, FlaxResnetBlockaD, FlaxUpsampleaD






class 						UpperCAmelCase__ (							nn.Module							):
		__snake_case      :       int
		__snake_case      :       int
		__snake_case      :       float     =   0.0
		__snake_case      :       int     =   1
		__snake_case      :       int     =   1
		__snake_case      :       bool     =   True
		__snake_case      :       bool     =   False
		__snake_case      :       bool     =   False
		__snake_case      :       bool     =   False
		__snake_case      :       jnp.dtype     =   jnp.floataa







		def  A__      (					self		):
						_A	:				Optional[Any]   =      []
						_A	:				str   =      []

						for i in range(self.num_layers		):
										_A	:				Union[str, Any]   =      self.in_channels if i == 0 else self.out_channels

										_A	:				str   =      FlaxResnetBlockaD(
										    in_channels=A__       ,out_channels=self.out_channels       ,dropout_prob=self.dropout       ,dtype=self.dtype       ,)
										resnets.append(A__		)

										_A	:				List[str]   =      FlaxTransformeraDModel(
										    in_channels=self.out_channels       ,n_heads=self.num_attention_heads       ,d_head=self.out_channels // self.num_attention_heads       ,depth=1       ,use_linear_projection=self.use_linear_projection       ,only_cross_attention=self.only_cross_attention       ,use_memory_efficient_attention=self.use_memory_efficient_attention       ,dtype=self.dtype       ,)
										attentions.append(A__		)

						_A	:				List[str]   =      resnets
						_A	:				Any   =      attentions

						if self.add_downsample:
										_A	:				Any   =      FlaxDownsampleaD(self.out_channels       ,dtype=self.dtype		)







		def __call__(					self       ,A__       ,A__       ,A__       ,A__=True		):
						_A	:				List[str]   =      ()

						for resnet, attn in zip(self.resnets       ,self.attentions		):
										_A	:				Optional[Any]   =      resnet(A__       ,A__       ,deterministic=A__		)
										_A	:				Tuple   =      attn(A__       ,A__       ,deterministic=A__		)
										output_states += (hidden_states,)

						if self.add_downsample:
										_A	:				Optional[int]   =      self.downsamplers_a(A__		)
										output_states += (hidden_states,)

						return hidden_states, output_states






class 						UpperCAmelCase__ (							nn.Module							):
		__snake_case      :       int
		__snake_case      :       int
		__snake_case      :       float     =   0.0
		__snake_case      :       int     =   1
		__snake_case      :       bool     =   True
		__snake_case      :       jnp.dtype     =   jnp.floataa







		def  A__      (					self		):
						_A	:				List[Any]   =      []

						for i in range(self.num_layers		):
										_A	:				int   =      self.in_channels if i == 0 else self.out_channels

										_A	:				List[Any]   =      FlaxResnetBlockaD(
										    in_channels=A__       ,out_channels=self.out_channels       ,dropout_prob=self.dropout       ,dtype=self.dtype       ,)
										resnets.append(A__		)
						_A	:				Tuple   =      resnets

						if self.add_downsample:
										_A	:				Optional[Any]   =      FlaxDownsampleaD(self.out_channels       ,dtype=self.dtype		)







		def __call__(					self       ,A__       ,A__       ,A__=True		):
						_A	:				List[Any]   =      ()

						for resnet in self.resnets:
										_A	:				Optional[int]   =      resnet(A__       ,A__       ,deterministic=A__		)
										output_states += (hidden_states,)

						if self.add_downsample:
										_A	:				List[Any]   =      self.downsamplers_a(A__		)
										output_states += (hidden_states,)

						return hidden_states, output_states






class 						UpperCAmelCase__ (							nn.Module							):
		__snake_case      :       int
		__snake_case      :       int
		__snake_case      :       int
		__snake_case      :       float     =   0.0
		__snake_case      :       int     =   1
		__snake_case      :       int     =   1
		__snake_case      :       bool     =   True
		__snake_case      :       bool     =   False
		__snake_case      :       bool     =   False
		__snake_case      :       bool     =   False
		__snake_case      :       jnp.dtype     =   jnp.floataa







		def  A__      (					self		):
						_A	:				List[str]   =      []
						_A	:				Optional[Any]   =      []

						for i in range(self.num_layers		):
										_A	:				Dict   =      self.in_channels if (i == self.num_layers - 1) else self.out_channels
										_A	:				List[str]   =      self.prev_output_channel if i == 0 else self.out_channels

										_A	:				Optional[int]   =      FlaxResnetBlockaD(
										    in_channels=resnet_in_channels + res_skip_channels       ,out_channels=self.out_channels       ,dropout_prob=self.dropout       ,dtype=self.dtype       ,)
										resnets.append(A__		)

										_A	:				Union[str, Any]   =      FlaxTransformeraDModel(
										    in_channels=self.out_channels       ,n_heads=self.num_attention_heads       ,d_head=self.out_channels // self.num_attention_heads       ,depth=1       ,use_linear_projection=self.use_linear_projection       ,only_cross_attention=self.only_cross_attention       ,use_memory_efficient_attention=self.use_memory_efficient_attention       ,dtype=self.dtype       ,)
										attentions.append(A__		)

						_A	:				Dict   =      resnets
						_A	:				int   =      attentions

						if self.add_upsample:
										_A	:				str   =      FlaxUpsampleaD(self.out_channels       ,dtype=self.dtype		)







		def __call__(					self       ,A__       ,A__       ,A__       ,A__       ,A__=True		):
						for resnet, attn in zip(self.resnets       ,self.attentions		):
										# pop res hidden states
										_A	:				List[Any]   =      res_hidden_states_tuple[-1]
										_A	:				int   =      res_hidden_states_tuple[:-1]
										_A	:				List[Any]   =      jnp.concatenate((hidden_states, res_hidden_states)       ,axis=-1		)

										_A	:				str   =      resnet(A__       ,A__       ,deterministic=A__		)
										_A	:				Optional[int]   =      attn(A__       ,A__       ,deterministic=A__		)

						if self.add_upsample:
										_A	:				Union[str, Any]   =      self.upsamplers_a(A__		)

						return hidden_states






class 						UpperCAmelCase__ (							nn.Module							):
		__snake_case      :       int
		__snake_case      :       int
		__snake_case      :       int
		__snake_case      :       float     =   0.0
		__snake_case      :       int     =   1
		__snake_case      :       bool     =   True
		__snake_case      :       jnp.dtype     =   jnp.floataa







		def  A__      (					self		):
						_A	:				Optional[Any]   =      []

						for i in range(self.num_layers		):
										_A	:				Any   =      self.in_channels if (i == self.num_layers - 1) else self.out_channels
										_A	:				Optional[Any]   =      self.prev_output_channel if i == 0 else self.out_channels

										_A	:				List[Any]   =      FlaxResnetBlockaD(
										    in_channels=resnet_in_channels + res_skip_channels       ,out_channels=self.out_channels       ,dropout_prob=self.dropout       ,dtype=self.dtype       ,)
										resnets.append(A__		)

						_A	:				int   =      resnets

						if self.add_upsample:
										_A	:				List[str]   =      FlaxUpsampleaD(self.out_channels       ,dtype=self.dtype		)







		def __call__(					self       ,A__       ,A__       ,A__       ,A__=True		):
						for resnet in self.resnets:
										# pop res hidden states
										_A	:				Tuple   =      res_hidden_states_tuple[-1]
										_A	:				Optional[int]   =      res_hidden_states_tuple[:-1]
										_A	:				Any   =      jnp.concatenate((hidden_states, res_hidden_states)       ,axis=-1		)

										_A	:				Optional[Any]   =      resnet(A__       ,A__       ,deterministic=A__		)

						if self.add_upsample:
										_A	:				Tuple   =      self.upsamplers_a(A__		)

						return hidden_states






class 						UpperCAmelCase__ (							nn.Module							):
		__snake_case      :       int
		__snake_case      :       float     =   0.0
		__snake_case      :       int     =   1
		__snake_case      :       int     =   1
		__snake_case      :       bool     =   False
		__snake_case      :       bool     =   False
		__snake_case      :       jnp.dtype     =   jnp.floataa







		def  A__      (					self		):
						# there is always at least one resnet
						_A	:				List[str]   =      [
						    FlaxResnetBlockaD(
						        in_channels=self.in_channels       ,out_channels=self.in_channels       ,dropout_prob=self.dropout       ,dtype=self.dtype       ,)
						]

						_A	:				Union[str, Any]   =      []

						for _ in range(self.num_layers		):
										_A	:				Any   =      FlaxTransformeraDModel(
										    in_channels=self.in_channels       ,n_heads=self.num_attention_heads       ,d_head=self.in_channels // self.num_attention_heads       ,depth=1       ,use_linear_projection=self.use_linear_projection       ,use_memory_efficient_attention=self.use_memory_efficient_attention       ,dtype=self.dtype       ,)
										attentions.append(A__		)

										_A	:				Any   =      FlaxResnetBlockaD(
										    in_channels=self.in_channels       ,out_channels=self.in_channels       ,dropout_prob=self.dropout       ,dtype=self.dtype       ,)
										resnets.append(A__		)

						_A	:				List[str]   =      resnets
						_A	:				str   =      attentions







		def __call__(					self       ,A__       ,A__       ,A__       ,A__=True		):
						_A	:				Optional[int]   =      self.resnets[0](A__       ,A__		)
						for attn, resnet in zip(self.attentions       ,self.resnets[1:]		):
										_A	:				Dict   =      attn(A__       ,A__       ,deterministic=A__		)
										_A	:				Any   =      resnet(A__       ,A__       ,deterministic=A__		)

						return hidden_states



