



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


UpperCAmelCase							       =    logging.get_logger(__name__)

class        __snake_case					(      __lowerCamelCase):







							'''simple docstring'''






							UpperCamelCase__ :					Any			   =    ["input_values", "attention_mask"]


							def __init__(							self     ,     a_=1     ,     a_=16_000     ,     a_=128     ,     a_=1_024     ,     a_=0.0     ,     a_=True     ,     a_=-4.2_677_393     ,     a_=4.5_689_974     ,     a_=False     ,     **a_     ,     ):
											super().__init__(feature_size=_UpperCamelCase     ,     sampling_rate=_UpperCamelCase     ,     padding_value=_UpperCamelCase     ,     **_UpperCamelCase						)
											a__ 					=						num_mel_bins
											a__ 					=						max_length
											a__ 					=						do_normalize
											a__ 					=						mean
											a__ 					=						std
											a__ 					=						return_attention_mask


							def _a					(							self     ,     a_     ,     a_     ,     ):
											a__ 					=						torch.from_numpy(_UpperCamelCase						).unsqueeze(0						)
											a__ 					=						ta_kaldi.fbank(
											    _UpperCamelCase     ,     htk_compat=_UpperCamelCase     ,     sample_frequency=self.sampling_rate     ,     use_energy=_UpperCamelCase     ,     window_type="""hanning"""     ,     num_mel_bins=self.num_mel_bins     ,     dither=0.0     ,     frame_shift=10     ,     )

											a__ 					=						fbank.shape[0]
											a__ 					=						max_length - n_frames

											# pad or truncate, depending on difference
											if difference > 0:
															a__ 					=						torch.nn.ZeroPadad((0, 0, 0, difference)						)
															a__ 					=						pad_module(_UpperCamelCase						)
											elif difference < 0:
															a__ 					=						fbank[0:max_length, :]

											a__ 					=						fbank.numpy()

											return fbank


							def _a					(							self     ,     a_						):
											return (input_values - (self.mean)) / (self.std * 2)







							def __call__(							self     ,     a_     ,     a_ = None     ,     a_ = None     ,     **a_     ,     ):





											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."""						)

											a__ 					=						isinstance(_UpperCamelCase     ,     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}'''						)
											a__ 					=						is_batched_numpy or (
											    isinstance(_UpperCamelCase     ,     (list, tuple)						) and (isinstance(raw_speech[0]     ,     (np.ndarray, tuple, list)						))
											)

											if is_batched:
															a__ 					=						[np.asarray(_UpperCamelCase     ,     dtype=np.floataa						) for speech in raw_speech]
											elif not is_batched and not isinstance(_UpperCamelCase     ,     np.ndarray						):
															a__ 					=						np.asarray(_UpperCamelCase     ,     dtype=np.floataa						)
											elif isinstance(_UpperCamelCase     ,     np.ndarray						) and raw_speech.dtype is np.dtype(np.floataa						):
															a__ 					=						raw_speech.astype(np.floataa						)

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

											# extract fbank features and pad/truncate to max_length
											a__ 					=						[self._extract_fbank_features(_UpperCamelCase     ,     max_length=self.max_length						) for waveform in raw_speech]

											# convert into BatchFeature
											a__ 					=						BatchFeature({"""input_values""": features}						)

											# make sure list is in array format
											a__ 					=						padded_inputs.get("""input_values"""						)
											if isinstance(input_values[0]     ,     _UpperCamelCase						):
															a__ 					=						[np.asarray(_UpperCamelCase     ,     dtype=np.floataa						) for feature in input_values]

											# normalization
											if self.do_normalize:
															a__ 					=						[self.normalize(_UpperCamelCase						) for feature in input_values]

											if return_tensors is not None:
															a__ 					=						padded_inputs.convert_to_tensors(_UpperCamelCase						)

											return padded_inputs







import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
    import torch

if is_vision_available():
    from PIL import Image

    from transformers import ConvNextImageProcessor
class       __a					( unittest.TestCase     ):


     """simple docstring"""


     def __init__(	self			: List[str]				,_UpperCamelCase			: int				,_UpperCamelCase			: int=7				,_UpperCamelCase			: int=3				,_UpperCamelCase			: List[str]=1_8				,_UpperCamelCase			: Dict=3_0				,_UpperCamelCase			: Tuple=4_0_0				,_UpperCamelCase			: Optional[int]=True				,_UpperCamelCase			: int=None				,_UpperCamelCase			: str=0.875				,_UpperCamelCase			: Tuple=True				,_UpperCamelCase			: Dict=[0.5, 0.5, 0.5]				,_UpperCamelCase			: str=[0.5, 0.5, 0.5]				,)		->   int:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__								=size if size is not None else {"""shortest_edge""": 2_0}
         SCREAMING_SNAKE_CASE__								=parent
         SCREAMING_SNAKE_CASE__								=batch_size
         SCREAMING_SNAKE_CASE__								=num_channels
         SCREAMING_SNAKE_CASE__								=image_size
         SCREAMING_SNAKE_CASE__								=min_resolution
         SCREAMING_SNAKE_CASE__								=max_resolution
         SCREAMING_SNAKE_CASE__								=do_resize
         SCREAMING_SNAKE_CASE__								=size
         SCREAMING_SNAKE_CASE__								=crop_pct
         SCREAMING_SNAKE_CASE__								=do_normalize
         SCREAMING_SNAKE_CASE__								=image_mean
         SCREAMING_SNAKE_CASE__								=image_std




     def   __A				(	self			: Dict						)		->   str:


         '''simple docstring'''





         return {
             "image_mean": self.image_mean,
             "image_std": self.image_std,
             "do_normalize": self.do_normalize,
             "do_resize": self.do_resize,
             "size": self.size,
             "crop_pct": self.crop_pct,
         }






@require_torch
@require_vision
class       __a					( __lowerCamelCase							,     unittest.TestCase     ):


     """simple docstring"""


     _A				:    Dict									=  ConvNextImageProcessor if is_vision_available() else None




     def   __A				(	self			: List[str]						)		->   Dict:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__								=ConvNextImageProcessingTester(self						)




     @property
     def   __A				(	self			: str						)		->   Optional[int]:


         '''simple docstring'''





         return self.image_processor_tester.prepare_image_processor_dict()




     def   __A				(	self			: Optional[int]						)		->   Any:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__								=self.image_processing_class(**self.image_processor_dict						)
         self.assertTrue(hasattr(_UpperCamelCase				,"""do_resize"""						)						)
         self.assertTrue(hasattr(_UpperCamelCase				,"""size"""						)						)
         self.assertTrue(hasattr(_UpperCamelCase				,"""crop_pct"""						)						)
         self.assertTrue(hasattr(_UpperCamelCase				,"""do_normalize"""						)						)
         self.assertTrue(hasattr(_UpperCamelCase				,"""image_mean"""						)						)
         self.assertTrue(hasattr(_UpperCamelCase				,"""image_std"""						)						)




     def   __A				(	self			: Tuple						)		->   Optional[Any]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__								=self.image_processing_class.from_dict(self.image_processor_dict						)
         self.assertEqual(image_processor.size				,{"""shortest_edge""": 2_0}						)

         SCREAMING_SNAKE_CASE__								=self.image_processing_class.from_dict(self.image_processor_dict				,size=4_2						)
         self.assertEqual(image_processor.size				,{"""shortest_edge""": 4_2}						)




     def   __A				(	self			: Any						)		->   List[str]:


         '''simple docstring'''





         pass




     def   __A				(	self			: Optional[int]						)		->   int:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__								=self.image_processing_class(**self.image_processor_dict						)
         # create random PIL images
         SCREAMING_SNAKE_CASE__								=prepare_image_inputs(self.image_processor_tester				,equal_resolution=_UpperCamelCase						)
         for image in image_inputs:
             self.assertIsInstance(_UpperCamelCase				,Image.Image						)

         # Test not batched input
         SCREAMING_SNAKE_CASE__								=image_processing(image_inputs[0]				,return_tensors="""pt"""						).pixel_values
         self.assertEqual(
             encoded_images.shape				,(
                 1,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.size["""shortest_edge"""],
                 self.image_processor_tester.size["""shortest_edge"""],
             )				,)

         # Test batched
         SCREAMING_SNAKE_CASE__								=image_processing(_UpperCamelCase				,return_tensors="""pt"""						).pixel_values
         self.assertEqual(
             encoded_images.shape				,(
                 self.image_processor_tester.batch_size,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.size["""shortest_edge"""],
                 self.image_processor_tester.size["""shortest_edge"""],
             )				,)




     def   __A				(	self			: str						)		->   str:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__								=self.image_processing_class(**self.image_processor_dict						)
         # create random numpy tensors
         SCREAMING_SNAKE_CASE__								=prepare_image_inputs(self.image_processor_tester				,equal_resolution=_UpperCamelCase				,numpify=_UpperCamelCase						)
         for image in image_inputs:
             self.assertIsInstance(_UpperCamelCase				,np.ndarray						)

         # Test not batched input
         SCREAMING_SNAKE_CASE__								=image_processing(image_inputs[0]				,return_tensors="""pt"""						).pixel_values
         self.assertEqual(
             encoded_images.shape				,(
                 1,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.size["""shortest_edge"""],
                 self.image_processor_tester.size["""shortest_edge"""],
             )				,)

         # Test batched
         SCREAMING_SNAKE_CASE__								=image_processing(_UpperCamelCase				,return_tensors="""pt"""						).pixel_values
         self.assertEqual(
             encoded_images.shape				,(
                 self.image_processor_tester.batch_size,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.size["""shortest_edge"""],
                 self.image_processor_tester.size["""shortest_edge"""],
             )				,)




     def   __A				(	self			: List[Any]						)		->   Union[str, Any]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__								=self.image_processing_class(**self.image_processor_dict						)
         # create random PyTorch tensors
         SCREAMING_SNAKE_CASE__								=prepare_image_inputs(self.image_processor_tester				,equal_resolution=_UpperCamelCase				,torchify=_UpperCamelCase						)
         for image in image_inputs:
             self.assertIsInstance(_UpperCamelCase				,torch.Tensor						)

         # Test not batched input
         SCREAMING_SNAKE_CASE__								=image_processing(image_inputs[0]				,return_tensors="""pt"""						).pixel_values
         self.assertEqual(
             encoded_images.shape				,(
                 1,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.size["""shortest_edge"""],
                 self.image_processor_tester.size["""shortest_edge"""],
             )				,)

         # Test batched
         SCREAMING_SNAKE_CASE__								=image_processing(_UpperCamelCase				,return_tensors="""pt"""						).pixel_values
         self.assertEqual(
             encoded_images.shape				,(
                 self.image_processor_tester.batch_size,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.size["""shortest_edge"""],
                 self.image_processor_tester.size["""shortest_edge"""],
             )				,)






