


'''simple docstring'''


import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np
import torch

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


UpperCAmelCase_						:  Union[str, Any]				  =						logging.get_logger(__name__)





class        lowercase__    (		_snake_case   ):





							'''simple docstring'''


							A_				: List[str]       =    ["""input_features""", """is_longer"""]






							def __init__(       self       ,   __snake_case=64       ,   __snake_case=4_8000       ,   __snake_case=480       ,   __snake_case=10       ,   __snake_case=1024       ,   __snake_case=0.0       ,   __snake_case=False       ,   __snake_case = 0       ,   __snake_case = 1_4000       ,   __snake_case = None       ,   __snake_case = "fusion"       ,   __snake_case = "repeatpad"       ,   **__snake_case       ,   ):
											super().__init__(
											    feature_size=__snake_case       ,   sampling_rate=__snake_case       ,   padding_value=__snake_case       ,   return_attention_mask=__snake_case       ,   **__snake_case       ,   )
											_SCREAMING_SNAKE_CASE					:			Dict							= top_db
											_SCREAMING_SNAKE_CASE					:			List[str]							= truncation
											_SCREAMING_SNAKE_CASE					:			Dict							= padding
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= fft_window_size
											_SCREAMING_SNAKE_CASE					:			List[str]							= (fft_window_size >> 1) + 1
											_SCREAMING_SNAKE_CASE					:			Tuple							= hop_length
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= max_length_s
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= max_length_s * sampling_rate
											_SCREAMING_SNAKE_CASE					:			str							= sampling_rate
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= frequency_min
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= frequency_max
											_SCREAMING_SNAKE_CASE					:			str							= mel_filter_bank(
											    num_frequency_bins=self.nb_frequency_bins       ,   num_mel_filters=__snake_case       ,   min_frequency=__snake_case       ,   max_frequency=__snake_case       ,   sampling_rate=__snake_case       ,   norm=__snake_case       ,   mel_scale="""htk"""       ,   )
											_SCREAMING_SNAKE_CASE					:			int							= mel_filter_bank(
											    num_frequency_bins=self.nb_frequency_bins       ,   num_mel_filters=__snake_case       ,   min_frequency=__snake_case       ,   max_frequency=__snake_case       ,   sampling_rate=__snake_case       ,   norm="""slaney"""       ,   mel_scale="""slaney"""       ,   )






							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			Any							= copy.deepcopy(self.__dict__    )
											_SCREAMING_SNAKE_CASE					:			Dict							= self.__class__.__name__
											if "mel_filters" in output:
															del output["mel_filters"]
											if "mel_filters_slaney" in output:
															del output["mel_filters_slaney"]
											return output






							def UpperCAmelCase_      (       self       ,   __snake_case       ,   __snake_case = None    ):
											_SCREAMING_SNAKE_CASE					:			List[Any]							= spectrogram(
											    __snake_case       ,   window_function(self.fft_window_size       ,   """hann"""    )       ,   frame_length=self.fft_window_size       ,   hop_length=self.hop_length       ,   power=2.0       ,   mel_filters=__snake_case       ,   log_mel="""dB"""       ,   )
											return log_mel_spectrogram.T






							def UpperCAmelCase_      (       self       ,   __snake_case       ,   __snake_case       ,   __snake_case    ):
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= np.array_split(list(range(0       ,   total_frames - chunk_frames + 1    )    )       ,   3    )
											if len(ranges[1]    ) == 0:
															# if the audio is too short, we just use the first chunk
															_SCREAMING_SNAKE_CASE					:			List[Any]							= [0]
											if len(ranges[2]    ) == 0:
															# if the audio is too short, we just use the first chunk
															_SCREAMING_SNAKE_CASE					:			Optional[Any]							= [0]
											# randomly choose index for each part
											_SCREAMING_SNAKE_CASE					:			List[str]							= np.random.choice(ranges[0]    )
											_SCREAMING_SNAKE_CASE					:			Dict							= np.random.choice(ranges[1]    )
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= np.random.choice(ranges[2]    )

											_SCREAMING_SNAKE_CASE					:			List[Any]							= mel[idx_front : idx_front + chunk_frames, :]
											_SCREAMING_SNAKE_CASE					:			int							= mel[idx_middle : idx_middle + chunk_frames, :]
											_SCREAMING_SNAKE_CASE					:			List[Any]							= mel[idx_back : idx_back + chunk_frames, :]

											_SCREAMING_SNAKE_CASE					:			Dict							= torch.tensor(mel[None, None, :]    )
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= torch.nn.functional.interpolate(
											    __snake_case       ,   size=[chunk_frames, 64]       ,   mode="""bilinear"""       ,   align_corners=__snake_case    )
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= mel_shrink[0][0].numpy()
											_SCREAMING_SNAKE_CASE					:			str							= np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]       ,   axis=0    )
											return mel_fusion






							def UpperCAmelCase_      (       self       ,   __snake_case       ,   __snake_case       ,   __snake_case       ,   __snake_case    ):
											if waveform.shape[0] > max_length:
															if truncation == "rand_trunc":
																			_SCREAMING_SNAKE_CASE					:			List[str]							= True
																			# random crop to max_length (for compatibility) -> this should be handled by self.pad
																			_SCREAMING_SNAKE_CASE					:			str							= len(__snake_case    ) - max_length
																			_SCREAMING_SNAKE_CASE					:			List[Any]							= np.random.randint(0       ,   overflow + 1    )
																			_SCREAMING_SNAKE_CASE					:			int							= waveform[idx : idx + max_length]
																			_SCREAMING_SNAKE_CASE					:			int							= self._np_extract_fbank_features(__snake_case       ,   self.mel_filters_slaney    )[None, :]
															elif truncation == "fusion":
																			_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= self._np_extract_fbank_features(__snake_case       ,   self.mel_filters    )
																			_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
																			_SCREAMING_SNAKE_CASE					:			str							= mel.shape[0]
																			if chunk_frames == total_frames:
																							# there is a corner case where the audio length is larger than max_length but smaller than max_length+hop_length.
																							# In this case, we just use the whole audio.
																							_SCREAMING_SNAKE_CASE					:			Optional[Any]							= np.stack([mel, mel, mel, mel]       ,   axis=0    )
																							_SCREAMING_SNAKE_CASE					:			List[str]							= False
																			else:
																							_SCREAMING_SNAKE_CASE					:			str							= self._random_mel_fusion(__snake_case       ,   __snake_case       ,   __snake_case    )
																							_SCREAMING_SNAKE_CASE					:			Tuple							= True
															else:
																			raise NotImplementedError(f"""data_truncating {truncation} not implemented"""    )

											else:
															_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= False
															# only use repeat as a new possible value for padding. you repeat the audio before applying the usual max_length padding
															if waveform.shape[0] < max_length:
																			if padding == "repeat":
																							_SCREAMING_SNAKE_CASE					:			Optional[int]							= int(max_length / len(__snake_case    )    )
																							_SCREAMING_SNAKE_CASE					:			List[str]							= np.stack(np.tile(__snake_case       ,   n_repeat + 1    )    )[:max_length]
																			if padding == "repeatpad":
																							_SCREAMING_SNAKE_CASE					:			List[str]							= int(max_length / len(__snake_case    )    )
																							_SCREAMING_SNAKE_CASE					:			Dict							= np.stack(np.tile(__snake_case       ,   __snake_case    )    )
																			_SCREAMING_SNAKE_CASE					:			Any							= np.pad(__snake_case       ,   (0, max_length - waveform.shape[0])       ,   mode="""constant"""       ,   constant_values=0    )

															if truncation == "fusion":
																			_SCREAMING_SNAKE_CASE					:			Optional[Any]							= self._np_extract_fbank_features(__snake_case       ,   self.mel_filters    )
																			_SCREAMING_SNAKE_CASE					:			Any							= np.stack([input_mel, input_mel, input_mel, input_mel]       ,   axis=0    )
															else:
																			_SCREAMING_SNAKE_CASE					:			Optional[int]							= self._np_extract_fbank_features(__snake_case       ,   self.mel_filters_slaney    )[None, :]

											return input_mel, longer






							def __call__(       self       ,   __snake_case       ,   __snake_case = None       ,   __snake_case = None       ,   __snake_case = None       ,   __snake_case = None       ,   __snake_case = None       ,   **__snake_case       ,   ):
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= truncation if truncation is not None else self.truncation
											_SCREAMING_SNAKE_CASE					:			Dict							= padding if padding else self.padding

											if sampling_rate is not None:
															if sampling_rate != self.sampling_rate:
																			raise ValueError(
																			    f"""The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a"""
																			    f""" sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input"""
																			    f""" was sampled 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					:			Optional[Any]							= isinstance(__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}"""    )
											_SCREAMING_SNAKE_CASE					:			List[str]							= is_batched_numpy or (
											    isinstance(__snake_case       ,   (list, tuple)    ) and (isinstance(raw_speech[0]       ,   (np.ndarray, tuple, list)    ))
											)

											if is_batched:
															_SCREAMING_SNAKE_CASE					:			Optional[int]							= [np.asarray(__snake_case       ,   dtype=np.floataa    ) for speech in raw_speech]
											elif not is_batched and not isinstance(__snake_case       ,   np.ndarray    ):
															_SCREAMING_SNAKE_CASE					:			str							= np.asarray(__snake_case       ,   dtype=np.floataa    )
											elif isinstance(__snake_case       ,   np.ndarray    ) and raw_speech.dtype is np.dtype(np.floataa    ):
															_SCREAMING_SNAKE_CASE					:			Dict							= raw_speech.astype(np.floataa    )

											# always return batch
											if not is_batched:
															_SCREAMING_SNAKE_CASE					:			str							= [np.asarray(__snake_case    )]

											# convert to mel spectrogram, truncate and pad if needed.
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= [
											    self._get_input_mel(__snake_case       ,   max_length if max_length else self.nb_max_samples       ,   __snake_case       ,   __snake_case    )
											    for waveform in raw_speech
											]

											_SCREAMING_SNAKE_CASE					:			List[str]							= []
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= []
											for mel, longer in padded_inputs:
															input_mel.append(__snake_case    )
															is_longer.append(__snake_case    )

											if truncation == "fusion" and sum(__snake_case    ) == 0:
															# if no audio is longer than 10s, then randomly select one audio to be longer
															_SCREAMING_SNAKE_CASE					:			List[Any]							= np.random.randint(0       ,   len(__snake_case    )    )
															_SCREAMING_SNAKE_CASE					:			str							= True

											if isinstance(input_mel[0]       ,   __snake_case    ):
															_SCREAMING_SNAKE_CASE					:			int							= [np.asarray(__snake_case       ,   dtype=np.floataa    ) for feature in input_mel]

											# is_longer is a list of bool
											_SCREAMING_SNAKE_CASE					:			List[str]							= [[longer] for longer in is_longer]

											_SCREAMING_SNAKE_CASE					:			Dict							= {"""input_features""": input_mel, """is_longer""": is_longer}
											_SCREAMING_SNAKE_CASE					:			List[str]							= BatchFeature(__snake_case    )

											if return_tensors is not None:
															_SCREAMING_SNAKE_CASE					:			Optional[Any]							= input_features.convert_to_tensors(__snake_case    )

											return input_features








'''simple docstring'''


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


UpperCAmelCase_						:  int				  =						logging.get_logger(__name__)


if is_vision_available():
				import PIL





class        lowercase__    (		_snake_case   ):





							'''simple docstring'''


							A_				: int       =    ["""pixel_values"""]






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

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






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






							def UpperCAmelCase_      (       self       ,   __snake_case       ,   __snake_case       ,   __snake_case = None       ,   **__snake_case       ,   ):
											_SCREAMING_SNAKE_CASE					:			List[str]							= get_size_dict(__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(__snake_case       ,   size=(size["""height"""], size["""width"""])       ,   data_format=__snake_case       ,   **__snake_case    )






							def UpperCAmelCase_      (       self       ,   __snake_case       ,   __snake_case       ,   __snake_case = None       ,   **__snake_case       ,   ):
											return rescale(__snake_case       ,   scale=__snake_case       ,   data_format=__snake_case       ,   **__snake_case    )






							def UpperCAmelCase_      (       self       ,   __snake_case       ,   __snake_case       ,   __snake_case       ,   __snake_case = None       ,   **__snake_case       ,   ):
											return normalize(__snake_case       ,   mean=__snake_case       ,   std=__snake_case       ,   data_format=__snake_case       ,   **__snake_case    )






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

											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= 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."""    )

											# PIL RGBA images are converted to RGB
											if do_convert_rgb:
															_SCREAMING_SNAKE_CASE					:			Optional[int]							= [convert_to_rgb(__snake_case    ) for image in images]

											# All transformations expect numpy arrays.
											_SCREAMING_SNAKE_CASE					:			str							= [to_numpy_array(__snake_case    ) for image in images]

											if do_resize:
															_SCREAMING_SNAKE_CASE					:			Optional[Any]							= [self.resize(image=__snake_case       ,   size=__snake_case       ,   resample=__snake_case    ) for image in images]

											if do_center_crop:
															_SCREAMING_SNAKE_CASE					:			Tuple							= [self.center_crop(image=__snake_case       ,   size=__snake_case    ) for image in images]

											if do_rescale:
															_SCREAMING_SNAKE_CASE					:			List[Any]							= [self.rescale(image=__snake_case       ,   scale=__snake_case    ) for image in images]

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

											_SCREAMING_SNAKE_CASE					:			str							= [to_channel_dimension_format(__snake_case       ,   __snake_case    ) for image in images]

											_SCREAMING_SNAKE_CASE					:			int							= {"""pixel_values""": images}
											return BatchFeature(data=__snake_case       ,   tensor_type=__snake_case    )




