



from typing import List, Optional, Union

import numpy as np

from ....audio_utils import mel_filter_bank, optimal_fft_length, spectrogram, window_function
from ....feature_extraction_sequence_utils import SequenceFeatureExtractor
from ....feature_extraction_utils import BatchFeature
from ....file_utils import PaddingStrategy, TensorType
from ....utils import logging


UpperCAmelCase_									=     logging.get_logger(__name__)





class 		UpperCAmelCase						(						UpperCamelCase__				):
   SCREAMING_SNAKE_CASE__							=     ['''input_features''', '''attention_mask''']





   def __init__(	self       ,     _lowerCAmelCase=80       ,     _lowerCAmelCase=16_000       ,     _lowerCAmelCase=0.0       ,     _lowerCAmelCase=10       ,     _lowerCAmelCase=25       ,     _lowerCAmelCase="hamming_window"       ,     _lowerCAmelCase=32_768.0       ,     _lowerCAmelCase=0.97       ,     _lowerCAmelCase=1.0       ,     _lowerCAmelCase=True       ,     _lowerCAmelCase=True       ,     _lowerCAmelCase=False       ,     **_lowerCAmelCase       ,     ):
         super().__init__(feature_size=__A       ,     sampling_rate=__A       ,     padding_value=__A       ,     **__A			)

         _lowerCAmelCase 			=  feature_size
         _lowerCAmelCase 			=  sampling_rate
         _lowerCAmelCase 			=  padding_value
         _lowerCAmelCase 			=  hop_length
         _lowerCAmelCase 			=  win_length
         _lowerCAmelCase 			=  frame_signal_scale
         _lowerCAmelCase 			=  preemphasis_coeff
         _lowerCAmelCase 			=  mel_floor
         _lowerCAmelCase 			=  normalize_means
         _lowerCAmelCase 			=  normalize_vars
         _lowerCAmelCase 			=  win_function
         _lowerCAmelCase 			=  return_attention_mask

         _lowerCAmelCase 			=  win_length * sampling_rate // 1_000
         _lowerCAmelCase 			=  hop_length * sampling_rate // 1_000

         _lowerCAmelCase 			=  optimal_fft_length(self.sample_size			)
         _lowerCAmelCase 			=  (self.n_fft // 2) + 1





   def 	__lowerCAmelCase							(	self       ,     _lowerCAmelCase			):



         if self.win_function == "hamming_window":
               _lowerCAmelCase 			=  window_function(window_length=self.sample_size       ,     name=self.win_function       ,     periodic=__A			)
         else:
               _lowerCAmelCase 			=  window_function(window_length=self.sample_size       ,     name=self.win_function			)

         _lowerCAmelCase 			=  mel_filter_bank(
             num_frequency_bins=self.n_freqs       ,     num_mel_filters=self.feature_size       ,     min_frequency=0.0       ,     max_frequency=self.sampling_rate / 2.0       ,     sampling_rate=self.sampling_rate       ,     )

         _lowerCAmelCase 			=  spectrogram(
             one_waveform * self.frame_signal_scale       ,     window=__A       ,     frame_length=self.sample_size       ,     hop_length=self.sample_stride       ,     fft_length=self.n_fft       ,     center=__A       ,     preemphasis=self.preemphasis_coeff       ,     mel_filters=__A       ,     mel_floor=self.mel_floor       ,     log_mel='''log'''       ,     )
         return msfc_features.T





   def 	__lowerCAmelCase							(	self       ,     _lowerCAmelCase       ,     _lowerCAmelCase       ,     _lowerCAmelCase			):



         if self.normalize_means:
               _lowerCAmelCase 			=  x[:input_length].mean(axis=0			)
               _lowerCAmelCase 			=  np.subtract(__A       ,     __A			)
         if self.normalize_vars:
               _lowerCAmelCase 			=  x[:input_length].std(axis=0			)
               _lowerCAmelCase 			=  np.divide(__A       ,     __A			)

         if input_length < x.shape[0]:
               _lowerCAmelCase 			=  padding_value

         # make sure array is in float32
         _lowerCAmelCase 			=  x.astype(np.floataa			)

         return x





   def 	__lowerCAmelCase							(	self       ,     _lowerCAmelCase       ,     _lowerCAmelCase = None			):
         _lowerCAmelCase 			=  attention_mask.sum(-1			) if attention_mask is not None else [x.shape[0] for x in input_features]
         return [self._normalize_one(__A       ,     __A       ,     self.padding_value			) for x, n in zip(__A       ,     __A			)]





   def __call__(	self       ,     _lowerCAmelCase       ,     _lowerCAmelCase = False       ,     _lowerCAmelCase = None       ,     _lowerCAmelCase = False       ,     _lowerCAmelCase = None       ,     _lowerCAmelCase = None       ,     _lowerCAmelCase = None       ,     _lowerCAmelCase = None       ,     **_lowerCAmelCase       ,     ):



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

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

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

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

         # extract fbank features
         _lowerCAmelCase 			=  [self._extract_mfsc_features(__A			) for one_waveform in raw_speech]

         # convert into correct format for padding
         _lowerCAmelCase 			=  BatchFeature({'''input_features''': features}			)

         _lowerCAmelCase 			=  self.pad(
             __A       ,     padding=__A       ,     max_length=__A       ,     truncation=__A       ,     pad_to_multiple_of=__A       ,     return_attention_mask=__A       ,     **__A       ,     )
         # make sure list is in array format
         _lowerCAmelCase 			=  padded_inputs.get('''input_features'''			)
         if isinstance(input_features[0]       ,     __A			):
               _lowerCAmelCase 			=  [np.asarray(__A       ,     dtype=np.floataa			) for feature in input_features]

         _lowerCAmelCase 			=  padded_inputs.get('''attention_mask'''			)
         if attention_mask is not None:
               _lowerCAmelCase 			=  [np.asarray(__A       ,     dtype=np.intaa			) for array in attention_mask]

         if self.normalize_means or self.normalize_vars:
               _lowerCAmelCase 			=  (
                   np.array(__A       ,     dtype=np.intaa			)
                   if self._get_padding_strategies(__A       ,     max_length=__A			) is not PaddingStrategy.DO_NOT_PAD
                   and padding
                   else None
               )
               _lowerCAmelCase 			=  self.normalize(
                   padded_inputs['''input_features''']       ,     attention_mask=__A			)

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

         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_video_inputs


if is_torch_available():
    import torch

if is_vision_available():
    from PIL import Image

    from transformers import VivitImageProcessor





class 		UpperCAmelCase						(						unittest.TestCase				):





   def __init__(	self       ,     _lowerCAmelCase       ,     _lowerCAmelCase=7       ,     _lowerCAmelCase=3       ,     _lowerCAmelCase=10       ,     _lowerCAmelCase=18       ,     _lowerCAmelCase=30       ,     _lowerCAmelCase=400       ,     _lowerCAmelCase=True       ,     _lowerCAmelCase=None       ,     _lowerCAmelCase=True       ,     _lowerCAmelCase=[0.5, 0.5, 0.5]       ,     _lowerCAmelCase=[0.5, 0.5, 0.5]       ,     _lowerCAmelCase=None       ,     ):
         _lowerCAmelCase 			=  size if size is not None else {'''shortest_edge''': 18}
         _lowerCAmelCase 			=  crop_size if crop_size is not None else {'''height''': 18, '''width''': 18}

         _lowerCAmelCase 			=  parent
         _lowerCAmelCase 			=  batch_size
         _lowerCAmelCase 			=  num_channels
         _lowerCAmelCase 			=  num_frames
         _lowerCAmelCase 			=  image_size
         _lowerCAmelCase 			=  min_resolution
         _lowerCAmelCase 			=  max_resolution
         _lowerCAmelCase 			=  do_resize
         _lowerCAmelCase 			=  size
         _lowerCAmelCase 			=  do_normalize
         _lowerCAmelCase 			=  image_mean
         _lowerCAmelCase 			=  image_std
         _lowerCAmelCase 			=  crop_size





   def 	__lowerCAmelCase							(	self			):
         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_size": self.crop_size,
         }






@require_torch
@require_vision
class 		UpperCAmelCase						(						snake_case_       ,unittest.TestCase				):
   SCREAMING_SNAKE_CASE__							=     VivitImageProcessor if is_vision_available() else None





   def 	__lowerCAmelCase							(	self			):
         _lowerCAmelCase 			=  VivitImageProcessingTester(self			)





   @property
   def 	__lowerCAmelCase							(	self			):
         return self.image_processor_tester.prepare_image_processor_dict()





   def 	__lowerCAmelCase							(	self			):
         _lowerCAmelCase 			=  self.image_processing_class(**self.image_processor_dict			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''image_mean'''			)			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''image_std'''			)			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''do_normalize'''			)			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''do_resize'''			)			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''do_center_crop'''			)			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''size'''			)			)





   def 	__lowerCAmelCase							(	self			):
         _lowerCAmelCase 			=  self.image_processing_class.from_dict(self.image_processor_dict			)
         self.assertEqual(image_processor.size       ,     {'''shortest_edge''': 18}			)
         self.assertEqual(image_processor.crop_size       ,     {'''height''': 18, '''width''': 18}			)

         _lowerCAmelCase 			=  self.image_processing_class.from_dict(self.image_processor_dict       ,     size=42       ,     crop_size=84			)
         self.assertEqual(image_processor.size       ,     {'''shortest_edge''': 42}			)
         self.assertEqual(image_processor.crop_size       ,     {'''height''': 84, '''width''': 84}			)





   def 	__lowerCAmelCase							(	self			):
         # Initialize image_processing
         _lowerCAmelCase 			=  self.image_processing_class(**self.image_processor_dict			)
         # create random PIL videos
         _lowerCAmelCase 			=  prepare_video_inputs(self.image_processor_tester       ,     equal_resolution=_lowerCAmelCase			)
         for video in video_inputs:
               self.assertIsInstance(_lowerCAmelCase       ,     _lowerCAmelCase			)
               self.assertIsInstance(video[0]       ,     Image.Image			)

         # Test not batched input
         _lowerCAmelCase 			=  image_processing(video_inputs[0]       ,     return_tensors='''pt'''			).pixel_values
         self.assertEqual(
             encoded_videos.shape       ,     (
                 1,
                 self.image_processor_tester.num_frames,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             )       ,     )

         # Test batched
         _lowerCAmelCase 			=  image_processing(_lowerCAmelCase       ,     return_tensors='''pt'''			).pixel_values
         self.assertEqual(
             encoded_videos.shape       ,     (
                 self.image_processor_tester.batch_size,
                 self.image_processor_tester.num_frames,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             )       ,     )





   def 	__lowerCAmelCase							(	self			):
         # Initialize image_processing
         _lowerCAmelCase 			=  self.image_processing_class(**self.image_processor_dict			)
         # create random numpy tensors
         _lowerCAmelCase 			=  prepare_video_inputs(self.image_processor_tester       ,     equal_resolution=_lowerCAmelCase       ,     numpify=_lowerCAmelCase			)
         for video in video_inputs:
               self.assertIsInstance(_lowerCAmelCase       ,     _lowerCAmelCase			)
               self.assertIsInstance(video[0]       ,     np.ndarray			)

         # Test not batched input
         _lowerCAmelCase 			=  image_processing(video_inputs[0]       ,     return_tensors='''pt'''			).pixel_values
         self.assertEqual(
             encoded_videos.shape       ,     (
                 1,
                 self.image_processor_tester.num_frames,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             )       ,     )

         # Test batched
         _lowerCAmelCase 			=  image_processing(_lowerCAmelCase       ,     return_tensors='''pt'''			).pixel_values
         self.assertEqual(
             encoded_videos.shape       ,     (
                 self.image_processor_tester.batch_size,
                 self.image_processor_tester.num_frames,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             )       ,     )





   def 	__lowerCAmelCase							(	self			):
         # Initialize image_processing
         _lowerCAmelCase 			=  self.image_processing_class(**self.image_processor_dict			)
         # create random PyTorch tensors
         _lowerCAmelCase 			=  prepare_video_inputs(self.image_processor_tester       ,     equal_resolution=_lowerCAmelCase       ,     torchify=_lowerCAmelCase			)
         for video in video_inputs:
               self.assertIsInstance(_lowerCAmelCase       ,     _lowerCAmelCase			)
               self.assertIsInstance(video[0]       ,     torch.Tensor			)

         # Test not batched input
         _lowerCAmelCase 			=  image_processing(video_inputs[0]       ,     return_tensors='''pt'''			).pixel_values
         self.assertEqual(
             encoded_videos.shape       ,     (
                 1,
                 self.image_processor_tester.num_frames,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             )       ,     )

         # Test batched
         _lowerCAmelCase 			=  image_processing(_lowerCAmelCase       ,     return_tensors='''pt'''			).pixel_values
         self.assertEqual(
             encoded_videos.shape       ,     (
                 self.image_processor_tester.batch_size,
                 self.image_processor_tester.num_frames,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             )       ,     )