



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


a_							=    logging.get_logger(__name__)







class 					A_(_lowerCAmelCase      ):







			"""simple docstring"""



			a_       :			List[Any]												=			["""audio_values""", """audio_mask"""]
			def __init__(	self				,   A=2048				,   A=1				,   A=[16, 16]				,   A=128				,   A=4_4100				,   A=86				,   A=2048				,   A=0.0				,   **A				,   ):
						super().__init__(
						    feature_size=SCREAMING_SNAKE_CASE_				,   sampling_rate=SCREAMING_SNAKE_CASE_				,   padding_value=SCREAMING_SNAKE_CASE_				,   **SCREAMING_SNAKE_CASE_				,   )

						_lowerCamelCase					:					Optional[Any]	       =    spectrogram_length
						_lowerCamelCase					:					Dict	       =    num_channels
						_lowerCamelCase					:					List[str]	       =    patch_size
						_lowerCamelCase					:					Union[str, Any]	       =    feature_size // self.patch_size[1]
						_lowerCamelCase					:					int	       =    n_fft
						_lowerCamelCase					:					Tuple	       =    sampling_rate // hop_length_to_sampling_rate
						_lowerCamelCase					:					Tuple	       =    sampling_rate
						_lowerCamelCase					:					List[str]	       =    padding_value
						_lowerCamelCase					:					Dict	       =    mel_filter_bank(
						    num_frequency_bins=1 + n_fft // 2				,   num_mel_filters=SCREAMING_SNAKE_CASE_				,   min_frequency=0.0				,   max_frequency=2_2050.0				,   sampling_rate=SCREAMING_SNAKE_CASE_				,   norm='slaney'				,   mel_scale='slaney'				,   ).T
			def    _lowerCAmelCase		(	self				,   A							):
						_lowerCamelCase					:					Optional[Any]	       =    spectrogram(
						    SCREAMING_SNAKE_CASE_				,   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				,   )
						_lowerCamelCase					:					Tuple	       =    log_spec[:, :-1]
						_lowerCamelCase					:					Any	       =    log_spec - 20.0
						_lowerCamelCase					:					Dict	       =    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				,   ):

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

						_lowerCamelCase					:					Optional[Any]	       =    isinstance(SCREAMING_SNAKE_CASE_				,   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					:					Tuple	       =    is_batched_numpy or (
						    isinstance(SCREAMING_SNAKE_CASE_				,   (list, tuple)							) and (isinstance(raw_speech[0]				,   (np.ndarray, tuple, list)							))
						)
						if is_batched:
									_lowerCamelCase					:					Union[str, Any]	       =    [np.asarray([speech]				,   dtype=np.floataa							).T for speech in raw_speech]
						elif not is_batched and not isinstance(SCREAMING_SNAKE_CASE_				,   np.ndarray							):
									_lowerCamelCase					:					Optional[int]	       =    np.asarray(SCREAMING_SNAKE_CASE_				,   dtype=np.floataa							)
						elif isinstance(SCREAMING_SNAKE_CASE_				,   np.ndarray							) and raw_speech.dtype is np.dtype(np.floataa							):
									_lowerCamelCase					:					Union[str, Any]	       =    raw_speech.astype(np.floataa							)
						# always return batch
						if not is_batched:
									_lowerCamelCase					:					int	       =    [np.asarray([raw_speech]							).T]

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

						# Create audio attention mask
						_lowerCamelCase					:					Optional[Any]	       =    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:
									_lowerCamelCase					:					List[str]	       =    [
									    (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
									]
									_lowerCamelCase					:					Dict	       =    np.array(SCREAMING_SNAKE_CASE_							).astype(np.floataa							)

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

						# return as BatchFeature
						if return_attention_mask:
									_lowerCamelCase					:					Any	       =    {"""audio_values""": padded_audio_features, """audio_mask""": audio_mask}
						else:
									_lowerCamelCase					:					Union[str, Any]	       =    {"""audio_values""": padded_audio_features}

						_lowerCamelCase					:					Dict	       =    BatchFeature(data=SCREAMING_SNAKE_CASE_				,   tensor_type=SCREAMING_SNAKE_CASE_							)
						return encoded_inputs










from typing import Dict, List, Optional, Union

import numpy as np

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


a      :  List[str]			     =			logging.get_logger(__name__)


if is_vision_available():
	import PIL







class 				_a				(   _lowerCAmelCase					):
			A      		=	['''pixel_values''']

			def __init__(self,   SCREAMING_SNAKE_CASE_ = True,   SCREAMING_SNAKE_CASE_ = None,   SCREAMING_SNAKE_CASE_ = PILImageResampling.BICUBIC,   SCREAMING_SNAKE_CASE_ = True,   SCREAMING_SNAKE_CASE_ = None,   SCREAMING_SNAKE_CASE_ = True,   SCREAMING_SNAKE_CASE_ = 1 / 255,   SCREAMING_SNAKE_CASE_ = True,   SCREAMING_SNAKE_CASE_ = None,   SCREAMING_SNAKE_CASE_ = None,   SCREAMING_SNAKE_CASE_ = True,   **SCREAMING_SNAKE_CASE_,   )       ->   None:
						super().__init__(**SCREAMING_SNAKE_CASE_					)
						UpperCAmelCase_:		Optional[Any]							      =     size if size is not None else {"""shortest_edge""": 224}
						UpperCAmelCase_:		int							      =     get_size_dict(SCREAMING_SNAKE_CASE_,   default_to_square=SCREAMING_SNAKE_CASE_					)
						UpperCAmelCase_:		Dict							      =     crop_size if crop_size is not None else {"""height""": 224, """width""": 224}
						UpperCAmelCase_:		Union[str, Any]							      =     get_size_dict(SCREAMING_SNAKE_CASE_,   default_to_square=SCREAMING_SNAKE_CASE_,   param_name="""crop_size"""					)

						UpperCAmelCase_:		Optional[int]							      =     do_resize
						UpperCAmelCase_:		Optional[int]							      =     size
						UpperCAmelCase_:		Optional[int]							      =     resample
						UpperCAmelCase_:		Tuple							      =     do_center_crop
						UpperCAmelCase_:		Optional[int]							      =     crop_size
						UpperCAmelCase_:		str							      =     do_rescale
						UpperCAmelCase_:		Optional[int]							      =     rescale_factor
						UpperCAmelCase_:		List[str]							      =     do_normalize
						UpperCAmelCase_:		Any							      =     image_mean if image_mean is not None else OPENAI_CLIP_MEAN
						UpperCAmelCase_:		Dict							      =     image_std if image_std is not None else OPENAI_CLIP_STD
						UpperCAmelCase_:		Any							      =     do_convert_rgb

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_ = PILImageResampling.BICUBIC,   SCREAMING_SNAKE_CASE_ = None,   **SCREAMING_SNAKE_CASE_,   )       ->   np.ndarray:
						UpperCAmelCase_:		Optional[Any]							      =     get_size_dict(SCREAMING_SNAKE_CASE_,   default_to_square=SCREAMING_SNAKE_CASE_					)
						if "shortest_edge" not in size:
									raise ValueError(f'The `size` parameter must contain the key `shortest_edge`. Got {size.keys()}'					)
						UpperCAmelCase_:		Optional[Any]							      =     get_resize_output_image_size(SCREAMING_SNAKE_CASE_,   size=size["""shortest_edge"""],   default_to_square=SCREAMING_SNAKE_CASE_					)
						return resize(SCREAMING_SNAKE_CASE_,   size=SCREAMING_SNAKE_CASE_,   resample=SCREAMING_SNAKE_CASE_,   data_format=SCREAMING_SNAKE_CASE_,   **SCREAMING_SNAKE_CASE_					)

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_ = None,   **SCREAMING_SNAKE_CASE_,   )       ->   np.ndarray:
						UpperCAmelCase_:		Any							      =     get_size_dict(SCREAMING_SNAKE_CASE_					)
						if "height" not in size or "width" not in size:
									raise ValueError(f'The `size` parameter must contain the keys (height, width). Got {size.keys()}'					)
						return center_crop(SCREAMING_SNAKE_CASE_,   size=(size["""height"""], size["""width"""]),   data_format=SCREAMING_SNAKE_CASE_,   **SCREAMING_SNAKE_CASE_					)

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_ = None,   **SCREAMING_SNAKE_CASE_,   )       ->   Union[str, Any]:
						return rescale(SCREAMING_SNAKE_CASE_,   scale=SCREAMING_SNAKE_CASE_,   data_format=SCREAMING_SNAKE_CASE_,   **SCREAMING_SNAKE_CASE_					)

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_ = None,   **SCREAMING_SNAKE_CASE_,   )       ->   np.ndarray:
						return normalize(SCREAMING_SNAKE_CASE_,   mean=SCREAMING_SNAKE_CASE_,   std=SCREAMING_SNAKE_CASE_,   data_format=SCREAMING_SNAKE_CASE_,   **SCREAMING_SNAKE_CASE_					)

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_ = None,   SCREAMING_SNAKE_CASE_ = None,   SCREAMING_SNAKE_CASE_ = None,   SCREAMING_SNAKE_CASE_ = None,   SCREAMING_SNAKE_CASE_ = None,   SCREAMING_SNAKE_CASE_ = None,   SCREAMING_SNAKE_CASE_ = None,   SCREAMING_SNAKE_CASE_ = None,   SCREAMING_SNAKE_CASE_ = None,   SCREAMING_SNAKE_CASE_ = None,   SCREAMING_SNAKE_CASE_ = None,   SCREAMING_SNAKE_CASE_ = None,   SCREAMING_SNAKE_CASE_ = ChannelDimension.FIRST,   **SCREAMING_SNAKE_CASE_,   )       ->   PIL.Image.Image:
						UpperCAmelCase_:		Tuple							      =     do_resize if do_resize is not None else self.do_resize
						UpperCAmelCase_:		Tuple							      =     size if size is not None else self.size
						UpperCAmelCase_:		str							      =     get_size_dict(SCREAMING_SNAKE_CASE_,   param_name="""size""",   default_to_square=SCREAMING_SNAKE_CASE_					)
						UpperCAmelCase_:		List[str]							      =     resample if resample is not None else self.resample
						UpperCAmelCase_:		Optional[int]							      =     do_center_crop if do_center_crop is not None else self.do_center_crop
						UpperCAmelCase_:		Tuple							      =     crop_size if crop_size is not None else self.crop_size
						UpperCAmelCase_:		int							      =     get_size_dict(SCREAMING_SNAKE_CASE_,   param_name="""crop_size""",   default_to_square=SCREAMING_SNAKE_CASE_					)
						UpperCAmelCase_:		Union[str, Any]							      =     do_rescale if do_rescale is not None else self.do_rescale
						UpperCAmelCase_:		int							      =     rescale_factor if rescale_factor is not None else self.rescale_factor
						UpperCAmelCase_:		Tuple							      =     do_normalize if do_normalize is not None else self.do_normalize
						UpperCAmelCase_:		Dict							      =     image_mean if image_mean is not None else self.image_mean
						UpperCAmelCase_:		int							      =     image_std if image_std is not None else self.image_std
						UpperCAmelCase_:		Union[str, Any]							      =     do_convert_rgb if do_convert_rgb is not None else self.do_convert_rgb

						UpperCAmelCase_:		List[str]							      =     make_list_of_images(SCREAMING_SNAKE_CASE_					)

						if not valid_images(SCREAMING_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."""					)

						# PIL RGBA images are converted to RGB
						if do_convert_rgb:
									UpperCAmelCase_:		Any							      =     [convert_to_rgb(SCREAMING_SNAKE_CASE_					) for image in images]

						# All transformations expect numpy arrays.
						UpperCAmelCase_:		List[str]							      =     [to_numpy_array(SCREAMING_SNAKE_CASE_					) for image in images]

						if do_resize:
									UpperCAmelCase_:		str							      =     [self.resize(image=SCREAMING_SNAKE_CASE_,   size=SCREAMING_SNAKE_CASE_,   resample=SCREAMING_SNAKE_CASE_					) for image in images]

						if do_center_crop:
									UpperCAmelCase_:		Dict							      =     [self.center_crop(image=SCREAMING_SNAKE_CASE_,   size=SCREAMING_SNAKE_CASE_					) for image in images]

						if do_rescale:
									UpperCAmelCase_:		Optional[Any]							      =     [self.rescale(image=SCREAMING_SNAKE_CASE_,   scale=SCREAMING_SNAKE_CASE_					) for image in images]

						if do_normalize:
									UpperCAmelCase_:		Dict							      =     [self.normalize(image=SCREAMING_SNAKE_CASE_,   mean=SCREAMING_SNAKE_CASE_,   std=SCREAMING_SNAKE_CASE_					) for image in images]

						UpperCAmelCase_:		Tuple							      =     [to_channel_dimension_format(SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_					) for image in images]

						UpperCAmelCase_:		Union[str, Any]							      =     {"""pixel_values""": images}
						return BatchFeature(data=SCREAMING_SNAKE_CASE_,   tensor_type=SCREAMING_SNAKE_CASE_					)






