



from typing import Dict, Iterable, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    IMAGENET_DEFAULT_MEAN,
    IMAGENET_DEFAULT_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, logging


a_  :Dict     = logging.get_logger(__name__)




class    snake_case__		(			lowerCAmelCase_	):






     """simple docstring"""


     _SCREAMING_SNAKE_CASE          =      ['''pixel_values''']






     def __init__(			self    :    Tuple,					_snake_case    :    Any = True,					_snake_case    :    Tuple = None,					_snake_case    :    Optional[Any] = PILImageResampling.BICUBIC,					_snake_case    :    Optional[Any] = True,					_snake_case    :    Union[str, Any] = None,					_snake_case    :    Optional[int] = True,					_snake_case    :    Union[str, Any] = 1 / 2_5_5,					_snake_case    :    Optional[Any] = True,					_snake_case    :    int = IMAGENET_DEFAULT_MEAN,					_snake_case    :    Union[str, Any] = IMAGENET_DEFAULT_STD,					**_snake_case    :    List[Any],					)  ->Optional[int]:
            super().__init__(**_snake_case   )
            snake_case__				:      Any        =      size if size is not None else {'shortest_edge': 2_2_4}
            snake_case__				:      List[str]        =      get_size_dict(_snake_case,					default_to_square=_snake_case   )
            snake_case__				:      Tuple        =      crop_size if crop_size is not None else {'height': 2_2_4, 'width': 2_2_4}
            snake_case__				:      List[str]        =      get_size_dict(_snake_case,					param_name='crop_size'   )

            snake_case__				:      Any        =      do_resize
            snake_case__				:      List[str]        =      size
            snake_case__				:      Any        =      resample
            snake_case__				:      Optional[Any]        =      do_center_crop
            snake_case__				:      Any        =      crop_size
            snake_case__				:      Tuple        =      do_rescale
            snake_case__				:      Optional[Any]        =      rescale_factor
            snake_case__				:      List[Any]        =      do_normalize
            snake_case__				:      int        =      image_mean if image_mean is not None else IMAGENET_DEFAULT_MEAN
            snake_case__				:      List[str]        =      image_std if image_std is not None else IMAGENET_DEFAULT_STD






     def   lowercase_  (			self    :    Any,					_snake_case    :    List[Any],					_snake_case    :    List[Any],					_snake_case    :    Any = PILImageResampling.BICUBIC,					_snake_case    :    int = None,					**_snake_case    :    List[str],					)  ->Dict:
            snake_case__				:      int        =      get_size_dict(_snake_case,					default_to_square=_snake_case   )
            # size_dict is a dict with either keys "height" and "width" or "shortest_edge"
            if "shortest_edge" in size:
                   snake_case__				:      str        =      int((2_5_6 / 2_2_4) * size['shortest_edge']   )
                   snake_case__				:      str        =      get_resize_output_image_size(_snake_case,					size=_snake_case,					default_to_square=_snake_case   )
                   snake_case__				:      Dict        =      {'height': output_size[0], 'width': output_size[1]}
            if "height" not in size_dict or "width" not in size_dict:
                   raise ValueError(
                       F'''Size dict must have keys \'height\' and \'width\' or \'shortest_edge\'. Got {size_dict.keys()}'''   )
            return resize(
                _snake_case,					size=(size_dict['height'], size_dict['width']),					resample=_snake_case,					data_format=_snake_case,					**_snake_case   )






     def   lowercase_  (			self    :    Optional[int],					_snake_case    :    Tuple,					_snake_case    :    Any,					_snake_case    :    Any = None,					**_snake_case    :    Any,					)  ->Optional[Any]:
            snake_case__				:      Optional[int]        =      get_size_dict(_snake_case   )
            if "height" not in size or "width" not in size:
                   raise ValueError(F'''Size dict must have keys \'height\' and \'width\'. Got {size.keys()}'''   )
            return center_crop(_snake_case,					size=(size['height'], size['width']),					data_format=_snake_case,					**_snake_case   )






     def   lowercase_  (			self    :    List[Any],					_snake_case    :    List[Any],					_snake_case    :    Optional[Any],					_snake_case    :    List[Any] = None,					**_snake_case    :    int,					)  ->Any:
            return rescale(_snake_case,					scale=_snake_case,					data_format=_snake_case,					**_snake_case   )






     def   lowercase_  (			self    :    Optional[int],					_snake_case    :    Dict,					_snake_case    :    Union[str, Any],					_snake_case    :    Union[str, Any],					_snake_case    :    Optional[Any] = None,					**_snake_case    :    Any,					)  ->int:
            return normalize(_snake_case,					mean=_snake_case,					std=_snake_case,					data_format=_snake_case,					**_snake_case   )






     def   lowercase_  (			self    :    Tuple,					_snake_case    :    Optional[Any],					_snake_case    :    List[Any] = None,					_snake_case    :    Tuple = None,					_snake_case    :    List[str] = None,					_snake_case    :    List[str] = None,					_snake_case    :    List[str] = None,					_snake_case    :    Dict = None,					_snake_case    :    Any = None,					_snake_case    :    Optional[Any] = None,					_snake_case    :    List[Any] = None,					_snake_case    :    Optional[Any] = None,					_snake_case    :    List[str] = None,					_snake_case    :    List[str] = ChannelDimension.FIRST,					**_snake_case    :    int,					)  ->Union[str, Any]:
            snake_case__				:      Union[str, Any]        =      do_resize if do_resize is not None else self.do_resize
            snake_case__				:      Optional[Any]        =      resample if resample is not None else self.resample
            snake_case__				:      Optional[int]        =      do_center_crop if do_center_crop is not None else self.do_center_crop
            snake_case__				:      Union[str, Any]        =      do_rescale if do_rescale is not None else self.do_rescale
            snake_case__				:      Optional[int]        =      rescale_factor if rescale_factor is not None else self.rescale_factor
            snake_case__				:      List[str]        =      do_normalize if do_normalize is not None else self.do_normalize
            snake_case__				:      Optional[Any]        =      image_mean if image_mean is not None else self.image_mean
            snake_case__				:      Optional[Any]        =      image_std if image_std is not None else self.image_std

            snake_case__				:      Dict        =      size if size is not None else self.size
            snake_case__				:      List[str]        =      get_size_dict(_snake_case,					default_to_square=_snake_case   )
            snake_case__				:      Tuple        =      crop_size if crop_size is not None else self.crop_size
            snake_case__				:      Union[str, Any]        =      get_size_dict(_snake_case,					param_name='crop_size'   )

            snake_case__				:      int        =      make_list_of_images(_snake_case   )

            if not valid_images(_snake_case   ):
                   raise ValueError(
                       'Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '
                       'torch.Tensor, tf.Tensor or jax.ndarray.'   )

            if do_resize and size is None:
                   raise ValueError('Size must be specified if do_resize is True.'   )

            if do_center_crop and crop_size is None:
                   raise ValueError('Crop size must be specified if do_center_crop is True.'   )

            if do_rescale and rescale_factor is None:
                   raise ValueError('Rescale factor must be specified if do_rescale is True.'   )

            if do_normalize and (image_mean is None or image_std is None):
                   raise ValueError('Image mean and std must be specified if do_normalize is True.'   )

            # All transformations expect numpy arrays.
            snake_case__				:      Tuple        =      [to_numpy_array(_snake_case   ) for image in images]

            if do_resize:
                   snake_case__				:      List[str]        =      [self.resize(_snake_case,					_snake_case,					_snake_case   ) for image in images]

            if do_center_crop:
                   snake_case__				:      int        =      [self.center_crop(_snake_case,					_snake_case   ) for image in images]

            if do_rescale:
                   snake_case__				:      int        =      [self.rescale(_snake_case,					_snake_case   ) for image in images]

            if do_normalize:
                   snake_case__				:      List[str]        =      [self.normalize(_snake_case,					_snake_case,					_snake_case   ) for image in images]

            snake_case__				:      Optional[int]        =      [to_channel_dimension_format(_snake_case,					_snake_case   ) for image in images]

            snake_case__				:      Tuple        =      {'pixel_values': images}
            return BatchFeature(data=_snake_case,					tensor_type=_snake_case   )












'''simple docstring'''



from math import ceil
from typing import List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import BatchFeature, SequenceFeatureExtractor
from ...utils import TensorType, logging


UpperCamelCase_    	=				logging.get_logger(__name__)






class 							_a				(			SCREAMING_SNAKE_CASE ):





       '''simple docstring'''






       A						:				str        =						['''audio_values''', '''audio_mask''']







       def __init__(      self,							A=2_048,							A=1,							A=[16, 16],							A=128,							A=44_100,							A=86,							A=2_048,							A=0.0,							**A,							):







             '''simple docstring'''







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

             SCREAMING_SNAKE_CASE   :      str    					=						spectrogram_length
             SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						num_channels
             SCREAMING_SNAKE_CASE   :      List[str]    					=						patch_size
             SCREAMING_SNAKE_CASE   :      Optional[int]    					=						feature_size // self.patch_size[1]
             SCREAMING_SNAKE_CASE   :      Dict    					=						n_fft
             SCREAMING_SNAKE_CASE   :      Tuple    					=						sampling_rate // hop_length_to_sampling_rate
             SCREAMING_SNAKE_CASE   :      str    					=						sampling_rate
             SCREAMING_SNAKE_CASE   :      int    					=						padding_value
             SCREAMING_SNAKE_CASE   :      Any    					=						mel_filter_bank(
                 num_frequency_bins=1 + n_fft // 2,							num_mel_filters=A,							min_frequency=0.0,							max_frequency=2_20_50.0,							sampling_rate=A,							norm='slaney',							mel_scale='slaney',							).T







       def        UpperCamelCase_							(      self,							A  ):







             '''simple docstring'''







             SCREAMING_SNAKE_CASE   :      Optional[int]    					=						spectrogram(
                 A,							window_function(self.n_fft,							'hann'  ),							frame_length=self.n_fft,							hop_length=self.hop_length,							power=2.0,							mel_filters=self.mel_filters.T,							log_mel='dB',							db_range=80.0,							)
             SCREAMING_SNAKE_CASE   :      Union[str, Any]    					=						log_spec[:, :-1]
             SCREAMING_SNAKE_CASE   :      List[Any]    					=						log_spec - 20.0
             SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						np.clip(log_spec / 40.0,							-2.0,							0.0  ) + 1.0
             return log_spec







       def __call__(      self,							A,							A = None,							A = True,							A = None,							A = False,							A = False,							**A,							):







             '''simple docstring'''







             if sampling_rate is not None:
                   if sampling_rate != self.sampling_rate:
                         raise ValueError(
                             'This feature extractor is set to support sampling rate'
                             F" of {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled"
                             F" with {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   :      List[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}"  )
             SCREAMING_SNAKE_CASE   :      int    					=						is_batched_numpy or (
                 isinstance(A,							(list, tuple)  ) and (isinstance(raw_speech[0],							(np.ndarray, tuple, list)  ))
             )
             if is_batched:
                   SCREAMING_SNAKE_CASE   :      Union[str, Any]    					=						[np.asarray([speech],							dtype=np.floataa  ).T for speech in raw_speech]
             elif not is_batched and not isinstance(A,							np.ndarray  ):
                   SCREAMING_SNAKE_CASE   :      Any    					=						np.asarray(A,							dtype=np.floataa  )
             elif isinstance(A,							np.ndarray  ) and raw_speech.dtype is np.dtype(np.floataa  ):
                   SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						raw_speech.astype(np.floataa  )
             # always return batch
             if not is_batched:
                   SCREAMING_SNAKE_CASE   :      Union[str, Any]    					=						[np.asarray([raw_speech]  ).T]

             # Convert audio signals to log mel spectrograms, truncate by time axis
             SCREAMING_SNAKE_CASE   :      int    					=						[
                 self._np_extract_fbank_features(waveform.squeeze()  ).T[: self.spectrogram_length] for waveform in raw_speech
             ]
             if isinstance(audio_features[0],							A  ):
                   SCREAMING_SNAKE_CASE   :      Union[str, Any]    					=						[np.asarray(A,							dtype=np.floataa  ) for feature in audio_features]

             # Create audio attention mask
             SCREAMING_SNAKE_CASE   :      Tuple    					=						max(
                 [ceil(feature.shape[0] / self.patch_size[0]  ) * self.freq_len for feature in audio_features]  )  # The maximum number of audio patches in a batch
             if return_attention_mask:
                   SCREAMING_SNAKE_CASE   :      List[Any]    					=						[
                       (ceil(feature.shape[0] / self.patch_size[0]  ) * self.freq_len) * [1]
                       + (max_patch_len - ceil(feature.shape[0] / self.patch_size[0]  ) * self.freq_len) * [0]
                       for feature in audio_features
                   ]
                   SCREAMING_SNAKE_CASE   :      Tuple    					=						np.array(A  ).astype(np.floataa  )

             # convert into correct format for padding
             SCREAMING_SNAKE_CASE   :      Tuple    					=						max_patch_len // self.freq_len * self.patch_size[0]  # The maximum audio size in a batch
             SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						np.ones([len(A  ), 1, max_time_len, self.feature_size]  ).astype(np.floataa  )
             SCREAMING_SNAKE_CASE   :      Optional[int]    					=						padded_audio_features * self.padding_value
             for i in range(len(A  )  ):
                   SCREAMING_SNAKE_CASE   :      Optional[int]    					=						audio_features[i]
                   SCREAMING_SNAKE_CASE   :      Union[str, Any]    					=						feature

             # return as BatchFeature
             if return_attention_mask:
                   SCREAMING_SNAKE_CASE   :      Any    					=						{'audio_values': padded_audio_features, 'audio_mask': audio_mask}
             else:
                   SCREAMING_SNAKE_CASE   :      Dict    					=						{'audio_values': padded_audio_features}

             SCREAMING_SNAKE_CASE   :      str    					=						BatchFeature(data=A,							tensor_type=A  )
             return encoded_inputs


