

import copy
import inspect

from transformers.testing_utils import require_torch, torch_device
from transformers.utils.backbone_utils import BackboneType





@require_torch
class       __UpperCAmelCase  :
   UpperCamelCase       =  ()
   UpperCamelCase       =  True



   def  __magic_name__					(							self							:  Optional[int]		):
     UpperCAmelCase :							Dict								=				self.config_class

     # test default config
     UpperCAmelCase :							Tuple								=				config_class()
     self.assertIsNotNone(__A		)
     UpperCAmelCase :							int								=				['''stem'''] + [F'''stage{idx}''' for idx in range(1,	len(config.depths		) + 1		)]
     self.assertEqual(config.stage_names,	__A		)
     self.assertTrue(set(config.out_features		).issubset(set(config.stage_names		)		)		)

     # Test out_features and out_indices are correctly set
     # out_features and out_indices both None
     UpperCAmelCase :							List[str]								=				config_class(out_features=__A,	out_indices=__A		)
     self.assertEqual(config.out_features,	[config.stage_names[-1]]		)
     self.assertEqual(config.out_indices,	[len(config.stage_names		) - 1]		)

     # out_features and out_indices both set
     UpperCAmelCase :							Any								=				config_class(out_features=['''stem''', '''stage1'''],	out_indices=[0, 1]		)
     self.assertEqual(config.out_features,	['''stem''', '''stage1''']		)
     self.assertEqual(config.out_indices,	[0, 1]		)

     # Only out_features set
     UpperCAmelCase :							str								=				config_class(out_features=['''stage1''', '''stage3''']		)
     self.assertEqual(config.out_features,	['''stage1''', '''stage3''']		)
     self.assertEqual(config.out_indices,	[1, 3]		)

     # Only out_indices set
     UpperCAmelCase :							Any								=				config_class(out_indices=[0, 2]		)
     self.assertEqual(config.out_features,	[config.stage_names[0], config.stage_names[2]]		)
     self.assertEqual(config.out_indices,	[0, 2]		)

     # Error raised when out_indices do not correspond to out_features
     with self.assertRaises(__A		):
       UpperCAmelCase :							int								=				config_class(out_features=['''stage1''', '''stage2'''],	out_indices=[0, 2]		)



   def  __magic_name__					(							self							:  str		):
     UpperCAmelCase				,		UpperCAmelCase :							Optional[Any]								=				self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
       UpperCAmelCase :							Optional[Any]								=				model_class(__A		)
       UpperCAmelCase :							List[str]								=				inspect.signature(model.forward		)
       # signature.parameters is an OrderedDict => so arg_names order is deterministic
       UpperCAmelCase :							List[str]								=				[*signature.parameters.keys()]
       UpperCAmelCase :							Tuple								=				['''pixel_values''']
       self.assertListEqual(arg_names[:1],	__A		)



   def  __magic_name__					(							self							:  Optional[int]		):
     UpperCAmelCase				,		UpperCAmelCase :							Optional[Any]								=				self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
       UpperCAmelCase :							List[Any]								=				model_class(__A		)
       self.assertEqual(len(model.channels		),	len(config.out_features		)		)
       UpperCAmelCase :							Optional[int]								=				model.num_features
       UpperCAmelCase :							Optional[Any]								=				[config.stage_names.index(__A		) for feat in config.out_features]
       UpperCAmelCase :							Optional[Any]								=				[num_features[idx] for idx in out_indices]
       self.assertListEqual(model.channels,	__A		)

       UpperCAmelCase :							List[str]								=				copy.deepcopy(__A		)
       UpperCAmelCase :							int								=				None
       UpperCAmelCase :							str								=				model_class(__A		)
       self.assertEqual(len(model.channels		),	1		)
       self.assertListEqual(model.channels,	[num_features[-1]]		)

       UpperCAmelCase :							Any								=				copy.deepcopy(__A		)
       UpperCAmelCase :							Optional[Any]								=				None
       UpperCAmelCase :							Optional[int]								=				model_class(__A		)
       self.assertEqual(len(model.channels		),	1		)
       self.assertListEqual(model.channels,	[num_features[-1]]		)



   def  __magic_name__					(							self							:  Union[str, Any]		):
     UpperCAmelCase				,		UpperCAmelCase :							Optional[Any]								=				self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
       UpperCAmelCase :							Dict								=				model_class(__A		)
       model.to(__A		)
       model.eval()
       UpperCAmelCase :							List[Any]								=				model(**__A		)

       self.assertEqual(len(result.feature_maps		),	len(config.out_features		)		)
       self.assertEqual(len(model.channels		),	len(config.out_features		)		)
       self.assertEqual(len(result.feature_maps		),	len(config.out_indices		)		)
       self.assertEqual(len(model.channels		),	len(config.out_indices		)		)

       # Check output of last stage is taken if out_features=None, out_indices=None
       UpperCAmelCase :							str								=				copy.deepcopy(__A		)
       UpperCAmelCase :							Dict								=				None
       UpperCAmelCase :							Any								=				model_class(__A		)
       model.to(__A		)
       model.eval()
       UpperCAmelCase :							Union[str, Any]								=				model(**__A		)

       self.assertEqual(len(result.feature_maps		),	1		)
       self.assertEqual(len(model.channels		),	1		)

       UpperCAmelCase :							Union[str, Any]								=				copy.deepcopy(__A		)
       UpperCAmelCase :							Optional[int]								=				None
       UpperCAmelCase :							Any								=				model_class(__A		)
       model.to(__A		)
       model.eval()
       UpperCAmelCase :							Dict								=				model(**__A		)

       self.assertEqual(len(result.feature_maps		),	1		)
       self.assertEqual(len(model.channels		),	1		)

       # Check backbone can be initialized with fresh weights
       UpperCAmelCase :							str								=				copy.deepcopy(__A		)
       UpperCAmelCase :							Union[str, Any]								=				False
       UpperCAmelCase :							Union[str, Any]								=				model_class(__A		)
       model.to(__A		)
       model.eval()
       UpperCAmelCase :							Union[str, Any]								=				model(**__A		)



   def  __magic_name__					(							self							:  List[str]		):
     UpperCAmelCase				,		UpperCAmelCase :							Union[str, Any]								=				self.model_tester.prepare_config_and_inputs_for_common()

     for backbone_class in self.all_model_classes:
       UpperCAmelCase :							Tuple								=				backbone_class(__A		)

       self.assertTrue(hasattr(__A,	'''backbone_type'''		)		)
       self.assertTrue(hasattr(__A,	'''stage_names'''		)		)
       self.assertTrue(hasattr(__A,	'''num_features'''		)		)
       self.assertTrue(hasattr(__A,	'''out_indices'''		)		)
       self.assertTrue(hasattr(__A,	'''out_features'''		)		)
       self.assertTrue(hasattr(__A,	'''out_feature_channels'''		)		)
       self.assertTrue(hasattr(__A,	'''channels'''		)		)

       self.assertIsInstance(backbone.backbone_type,	__A		)
       # Verify num_features has been initialized in the backbone init
       self.assertIsNotNone(backbone.num_features		)
       self.assertTrue(len(backbone.channels		) == len(backbone.out_indices		)		)
       self.assertTrue(len(backbone.stage_names		) == len(backbone.num_features		)		)
       self.assertTrue(len(backbone.channels		) <= len(backbone.num_features		)		)
       self.assertTrue(len(backbone.out_feature_channels		) == len(backbone.stage_names		)		)





   def  __magic_name__					(							self							:  Optional[int]		):
     UpperCAmelCase				,		UpperCAmelCase :							int								=				self.model_tester.prepare_config_and_inputs_for_common()
     UpperCAmelCase :							List[Any]								=				inputs_dict['''pixel_values'''].shape[0]

     for backbone_class in self.all_model_classes:
       UpperCAmelCase :							int								=				backbone_class(__A		)
       backbone.to(__A		)
       backbone.eval()

       UpperCAmelCase :							List[str]								=				backbone(**__A		)

       # Test default outputs and verify feature maps
       self.assertIsInstance(outputs.feature_maps,	__A		)
       self.assertTrue(len(outputs.feature_maps		) == len(backbone.channels		)		)
       for feature_map, n_channels in zip(outputs.feature_maps,	backbone.channels		):
         self.assertTrue(feature_map.shape[:2],	(batch_size, n_channels)		)
       self.assertIsNone(outputs.hidden_states		)
       self.assertIsNone(outputs.attentions		)

       # Test output_hidden_states=True
       UpperCAmelCase :							Tuple								=				backbone(**__A,	output_hidden_states=__A		)
       self.assertIsNotNone(outputs.hidden_states		)
       self.assertTrue(len(outputs.hidden_states		),	len(backbone.stage_names		)		)
       for hidden_state, n_channels in zip(outputs.hidden_states,	backbone.channels		):
         self.assertTrue(hidden_state.shape[:2],	(batch_size, n_channels)		)

       # Test output_attentions=True
       if self.has_attentions:
         UpperCAmelCase :							Optional[Any]								=				backbone(**__A,	output_attentions=__A		)
         self.assertIsNotNone(outputs.attentions		)




from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


_lowerCamelCase       :     str           =							logging.get_logger(__name__)





class       __UpperCAmelCase  ( lowerCamelCase__		):
   UpperCamelCase       =  ["""input_values""", """attention_mask"""]



   def __init__(							self							:  Union[str, Any],	__A							:  Dict=1,	__A							:  Any=1_6_0_0_0,	__A							:  List[str]=1_2_8,	__A							:  str=1_0_2_4,	__A							:  int=0.0,	__A							:  str=True,	__A							:  Any=-4.2_6_7_7_3_9_3,	__A							:  Optional[int]=4.5_6_8_9_9_7_4,	__A							:  Optional[int]=False,	**__A							:  str,	):
     super().__init__(feature_size=__A,	sampling_rate=__A,	padding_value=__A,	**__A		)
     UpperCAmelCase :							int								=				num_mel_bins
     UpperCAmelCase :							Tuple								=				max_length
     UpperCAmelCase :							str								=				do_normalize
     UpperCAmelCase :							List[Any]								=				mean
     UpperCAmelCase :							Union[str, Any]								=				std
     UpperCAmelCase :							Any								=				return_attention_mask



   def  __magic_name__					(							self							:  Dict,	__A							:  np.ndarray,	__A							:  int,	):
     UpperCAmelCase :							Union[str, Any]								=				torch.from_numpy(__A		).unsqueeze(0		)
     UpperCAmelCase :							List[str]								=				ta_kaldi.fbank(
         __A,	htk_compat=__A,	sample_frequency=self.sampling_rate,	use_energy=__A,	window_type='''hanning''',	num_mel_bins=self.num_mel_bins,	dither=0.0,	frame_shift=1_0,	)

     UpperCAmelCase :							int								=				fbank.shape[0]
     UpperCAmelCase :							int								=				max_length - n_frames

     # pad or truncate, depending on difference
     if difference > 0:
       UpperCAmelCase :							List[Any]								=				torch.nn.ZeroPadad((0, 0, 0, difference)		)
       UpperCAmelCase :							int								=				pad_module(__A		)
     elif difference < 0:
       UpperCAmelCase :							Dict								=				fbank[0:max_length, :]

     UpperCAmelCase :							int								=				fbank.numpy()

     return fbank



   def  __magic_name__					(							self							:  Optional[Any],	__A							:  np.ndarray		):
     return (input_values - (self.mean)) / (self.std * 2)





   def __call__(							self							:  Union[str, Any],	__A							:  Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]],	__A							:  Optional[int] = None,	__A							:  Optional[Union[str, TensorType]] = None,	**__A							:  Optional[Any],	):

     if sampling_rate is not None:
       if sampling_rate != self.sampling_rate:
         raise ValueError(
             F'''The model corresponding to this feature extractor: {self} was trained using a sampling rate of'''
             F''' {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with'''
             F''' {self.sampling_rate} and not {sampling_rate}.'''		)
     else:
       logger.warning(
           '''It is strongly recommended to pass the `sampling_rate` argument to this function. '''
           '''Failing to do so can result in silent errors that might be hard to debug.'''		)

     UpperCAmelCase :							Any								=				isinstance(__A,	np.ndarray		) and len(raw_speech.shape		) > 1
     if is_batched_numpy and len(raw_speech.shape		) > 2:
       raise ValueError(F'''Only mono-channel audio is supported for input to {self}'''		)
     UpperCAmelCase :							Dict								=				is_batched_numpy or (
         isinstance(__A,	(list, tuple)		) and (isinstance(raw_speech[0],	(np.ndarray, tuple, list)		))
     )

     if is_batched:
       UpperCAmelCase :							List[Any]								=				[np.asarray(__A,	dtype=np.floataa		) for speech in raw_speech]
     elif not is_batched and not isinstance(__A,	np.ndarray		):
       UpperCAmelCase :							Union[str, Any]								=				np.asarray(__A,	dtype=np.floataa		)
     elif isinstance(__A,	np.ndarray		) and raw_speech.dtype is np.dtype(np.floataa		):
       UpperCAmelCase :							int								=				raw_speech.astype(np.floataa		)

     # always return batch
     if not is_batched:
       UpperCAmelCase :							str								=				[raw_speech]

     # extract fbank features and pad/truncate to max_length
     UpperCAmelCase :							Union[str, Any]								=				[self._extract_fbank_features(__A,	max_length=self.max_length		) for waveform in raw_speech]

     # convert into BatchFeature
     UpperCAmelCase :							Optional[Any]								=				BatchFeature({'''input_values''': features}		)

     # make sure list is in array format
     UpperCAmelCase :							Optional[int]								=				padded_inputs.get('''input_values'''		)
     if isinstance(input_values[0],	__A		):
       UpperCAmelCase :							Dict								=				[np.asarray(__A,	dtype=np.floataa		) for feature in input_values]

     # normalization
     if self.do_normalize:
       UpperCAmelCase :							str								=				[self.normalize(__A		) for feature in input_values]

     if return_tensors is not None:
       UpperCAmelCase :							Tuple								=				padded_inputs.convert_to_tensors(__A		)

     return padded_inputs





