






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




   def __init__(   self      :    str	,     _lowercase      :    Any	,     _lowercase      :    List[Any]=7	,     _lowercase      :    Optional[Any]=3	,     _lowercase      :    int=10	,     _lowercase      :    Tuple=18	,     _lowercase      :    Union[str, Any]=30	,     _lowercase      :    Any=4_00	,     _lowercase      :    List[str]=True	,     _lowercase      :    Union[str, Any]=None	,     _lowercase      :    str=True	,     _lowercase      :    Union[str, Any]=[0.5, 0.5, 0.5]	,     _lowercase      :    Any=[0.5, 0.5, 0.5]	,     _lowercase      :    Dict=None	,     ):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__      							=				size if size is not None else {"shortest_edge": 18}
     SCREAMING_SNAKE_CASE__      							=				crop_size if crop_size is not None else {"height": 18, "width": 18}

     SCREAMING_SNAKE_CASE__      							=				parent
     SCREAMING_SNAKE_CASE__      							=				batch_size
     SCREAMING_SNAKE_CASE__      							=				num_channels
     SCREAMING_SNAKE_CASE__      							=				num_frames
     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__      							=				do_normalize
     SCREAMING_SNAKE_CASE__      							=				image_mean
     SCREAMING_SNAKE_CASE__      							=				image_std
     SCREAMING_SNAKE_CASE__      							=				crop_size




   def      __a						(   self      :    int				):



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







@require_torch
@require_vision
class __snake_case    (						__SCREAMING_SNAKE_CASE       ,		unittest.TestCase		):
   lowerCAmelCase_														=							VivitImageProcessor if is_vision_available() else None




   def      __a						(   self      :    int				):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__      							=				VivitImageProcessingTester(self				)




   @property
   def      __a						(   self      :    int				):



     """simple docstring"""






     return self.image_processor_tester.prepare_image_processor_dict()




   def      __a						(   self      :    List[Any]				):



     """simple docstring"""






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




   def      __a						(   self      :    Any				):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__      							=				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}				)

     SCREAMING_SNAKE_CASE__      							=				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      __a						(   self      :    int				):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__      							=				self.image_processing_class(**self.image_processor_dict				)
     # create random PIL videos
     SCREAMING_SNAKE_CASE__      							=				prepare_video_inputs(self.image_processor_tester	,     equal_resolution=_SCREAMING_SNAKE_CASE				)
     for video in video_inputs:
       self.assertIsInstance(_SCREAMING_SNAKE_CASE	,     _SCREAMING_SNAKE_CASE				)
       self.assertIsInstance(video[0]	,     Image.Image				)

     # Test not batched input
     SCREAMING_SNAKE_CASE__      							=				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
     SCREAMING_SNAKE_CASE__      							=				image_processing(_SCREAMING_SNAKE_CASE	,     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      __a						(   self      :    List[str]				):



     """simple docstring"""






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

     # Test not batched input
     SCREAMING_SNAKE_CASE__      							=				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
     SCREAMING_SNAKE_CASE__      							=				image_processing(_SCREAMING_SNAKE_CASE	,     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      __a						(   self      :    Dict				):



     """simple docstring"""






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

     # Test not batched input
     SCREAMING_SNAKE_CASE__      							=				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
     SCREAMING_SNAKE_CASE__      							=				image_processing(_SCREAMING_SNAKE_CASE	,     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"""],
         )	,     )

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


__lowerCamelCase      :			Any							       =		logging.get_logger(__name__)







class __snake_case    (						lowerCamelCase_		):
   lowerCAmelCase_														=							["input_features", "attention_mask"]




   def __init__(   self      :    int	,     _lowercase      :    Dict=80	,     _lowercase      :    Union[str, Any]=1_60_00	,     _lowercase      :    str=0.0	,     _lowercase      :    int=10	,     _lowercase      :    str=25	,     _lowercase      :    Union[str, Any]="hamming_window"	,     _lowercase      :    Dict=3_27_68.0	,     _lowercase      :    Optional[int]=0.97	,     _lowercase      :    Optional[int]=1.0	,     _lowercase      :    int=True	,     _lowercase      :    Union[str, Any]=True	,     _lowercase      :    Optional[int]=False	,     **_lowercase      :    Union[str, Any]	,     ):



     """simple docstring"""






     super().__init__(feature_size=_lowercase	,     sampling_rate=_lowercase	,     padding_value=_lowercase	,     **_lowercase				)

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

     SCREAMING_SNAKE_CASE__      							=				win_length * sampling_rate // 10_00
     SCREAMING_SNAKE_CASE__      							=				hop_length * sampling_rate // 10_00

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




   def      __a						(   self      :    str	,     _lowercase      :    np.array				):



     """simple docstring"""






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

     SCREAMING_SNAKE_CASE__      							=				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	,     )

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




   def      __a						(   self      :    Tuple	,     _lowercase      :    Union[str, Any]	,     _lowercase      :    Optional[int]	,     _lowercase      :    int				):



     """simple docstring"""






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

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

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

     return x




   def      __a						(   self      :    int	,     _lowercase      :    List[np.ndarray]	,     _lowercase      :    Optional[np.ndarray] = None				):



     """simple docstring"""






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




   def __call__(   self      :    Dict	,     _lowercase      :    Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]	,     _lowercase      :    Union[bool, str, PaddingStrategy] = False	,     _lowercase      :    Optional[int] = None	,     _lowercase      :    bool = False	,     _lowercase      :    Optional[int] = None	,     _lowercase      :    Optional[bool] = None	,     _lowercase      :    Optional[Union[str, TensorType]] = None	,     _lowercase      :    Optional[int] = None	,     **_lowercase      :    List[str]	,     ):



     """simple docstring"""






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

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

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

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

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

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

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

     SCREAMING_SNAKE_CASE__      							=				padded_inputs.get("""attention_mask"""				)
     if attention_mask is not None:
       SCREAMING_SNAKE_CASE__      							=				[np.asarray(_lowercase	,     dtype=np.intaa				) for array in attention_mask]

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

     if return_tensors is not None:
       SCREAMING_SNAKE_CASE__      							=				padded_inputs.convert_to_tensors(_lowercase				)

     return padded_inputs
