




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        =    logging.get_logger(__name__)


if is_vision_available():
					import PIL







class     __lowerCAmelCase (       __magic_name__ ):

			"""simple docstring"""







			snake_case_   		=       ['''pixel_values''']


			def __init__(      self			,						lowerCamelCase__ = True			,						lowerCamelCase__ = None			,						lowerCamelCase__ = PILImageResampling.BICUBIC			,						lowerCamelCase__ = True			,						lowerCamelCase__ = None			,						lowerCamelCase__ = True			,						lowerCamelCase__ = 1 / 255			,						lowerCamelCase__ = True			,						lowerCamelCase__ = None			,						lowerCamelCase__ = None			,						lowerCamelCase__ = True			,						**lowerCamelCase__			,						)    ->      None:



							'''simple docstring'''
							super().__init__(**lowerCamelCase__							)
							__lowerCamelCase		       =    size if size is not None else {'shortest_edge': 224}
							__lowerCamelCase		       =    get_size_dict(lowerCamelCase__			,						default_to_square=lowerCamelCase__							)
							__lowerCamelCase		       =    crop_size if crop_size is not None else {'height': 224, 'width': 224}
							__lowerCamelCase		       =    get_size_dict(lowerCamelCase__			,						default_to_square=lowerCamelCase__			,						param_name='crop_size'							)

							__lowerCamelCase		       =    do_resize
							__lowerCamelCase		       =    size
							__lowerCamelCase		       =    resample
							__lowerCamelCase		       =    do_center_crop
							__lowerCamelCase		       =    crop_size
							__lowerCamelCase		       =    do_rescale
							__lowerCamelCase		       =    rescale_factor
							__lowerCamelCase		       =    do_normalize
							__lowerCamelCase		       =    image_mean if image_mean is not None else OPENAI_CLIP_MEAN
							__lowerCamelCase		       =    image_std if image_std is not None else OPENAI_CLIP_STD
							__lowerCamelCase		       =    do_convert_rgb


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__ = PILImageResampling.BICUBIC			,						lowerCamelCase__ = None			,						**lowerCamelCase__			,						)    ->      np.ndarray:



							'''simple docstring'''
							__lowerCamelCase		       =    get_size_dict(lowerCamelCase__			,						default_to_square=lowerCamelCase__							)
							if "shortest_edge" not in size:
											raise ValueError(f"""The `size` parameter must contain the key `shortest_edge`. Got {size.keys()}"""							)
							__lowerCamelCase		       =    get_resize_output_image_size(lowerCamelCase__			,						size=size['shortest_edge']			,						default_to_square=lowerCamelCase__							)
							return resize(lowerCamelCase__			,						size=lowerCamelCase__			,						resample=lowerCamelCase__			,						data_format=lowerCamelCase__			,						**lowerCamelCase__							)


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__ = None			,						**lowerCamelCase__			,						)    ->      np.ndarray:



							'''simple docstring'''
							__lowerCamelCase		       =    get_size_dict(lowerCamelCase__							)
							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(lowerCamelCase__			,						size=(size['height'], size['width'])			,						data_format=lowerCamelCase__			,						**lowerCamelCase__							)


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__ = None			,						**lowerCamelCase__			,						)    ->      Any:



							'''simple docstring'''
							return rescale(lowerCamelCase__			,						scale=lowerCamelCase__			,						data_format=lowerCamelCase__			,						**lowerCamelCase__							)


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__ = None			,						**lowerCamelCase__			,						)    ->      np.ndarray:



							'''simple docstring'''
							return normalize(lowerCamelCase__			,						mean=lowerCamelCase__			,						std=lowerCamelCase__			,						data_format=lowerCamelCase__			,						**lowerCamelCase__							)






			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__ = None			,						lowerCamelCase__ = None			,						lowerCamelCase__ = None			,						lowerCamelCase__ = None			,						lowerCamelCase__ = None			,						lowerCamelCase__ = None			,						lowerCamelCase__ = None			,						lowerCamelCase__ = None			,						lowerCamelCase__ = None			,						lowerCamelCase__ = None			,						lowerCamelCase__ = None			,						lowerCamelCase__ = None			,						lowerCamelCase__ = ChannelDimension.FIRST			,						**lowerCamelCase__			,						)    ->      PIL.Image.Image:



							'''simple docstring'''
							__lowerCamelCase		       =    do_resize if do_resize is not None else self.do_resize
							__lowerCamelCase		       =    size if size is not None else self.size
							__lowerCamelCase		       =    get_size_dict(lowerCamelCase__			,						param_name='size'			,						default_to_square=lowerCamelCase__							)
							__lowerCamelCase		       =    resample if resample is not None else self.resample
							__lowerCamelCase		       =    do_center_crop if do_center_crop is not None else self.do_center_crop
							__lowerCamelCase		       =    crop_size if crop_size is not None else self.crop_size
							__lowerCamelCase		       =    get_size_dict(lowerCamelCase__			,						param_name='crop_size'			,						default_to_square=lowerCamelCase__							)
							__lowerCamelCase		       =    do_rescale if do_rescale is not None else self.do_rescale
							__lowerCamelCase		       =    rescale_factor if rescale_factor is not None else self.rescale_factor
							__lowerCamelCase		       =    do_normalize if do_normalize is not None else self.do_normalize
							__lowerCamelCase		       =    image_mean if image_mean is not None else self.image_mean
							__lowerCamelCase		       =    image_std if image_std is not None else self.image_std
							__lowerCamelCase		       =    do_convert_rgb if do_convert_rgb is not None else self.do_convert_rgb

							__lowerCamelCase		       =    make_list_of_images(lowerCamelCase__							)

							if not valid_images(lowerCamelCase__							):
											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:
											__lowerCamelCase		       =    [convert_to_rgb(lowerCamelCase__							) for image in images]

							# All transformations expect numpy arrays.
							__lowerCamelCase		       =    [to_numpy_array(lowerCamelCase__							) for image in images]

							if do_resize:
											__lowerCamelCase		       =    [self.resize(image=lowerCamelCase__			,						size=lowerCamelCase__			,						resample=lowerCamelCase__							) for image in images]

							if do_center_crop:
											__lowerCamelCase		       =    [self.center_crop(image=lowerCamelCase__			,						size=lowerCamelCase__							) for image in images]

							if do_rescale:
											__lowerCamelCase		       =    [self.rescale(image=lowerCamelCase__			,						scale=lowerCamelCase__							) for image in images]

							if do_normalize:
											__lowerCamelCase		       =    [self.normalize(image=lowerCamelCase__			,						mean=lowerCamelCase__			,						std=lowerCamelCase__							) for image in images]

							__lowerCamelCase		       =    [to_channel_dimension_format(lowerCamelCase__			,						lowerCamelCase__							) for image in images]

							__lowerCamelCase		       =    {'pixel_values': images}
							return BatchFeature(data=lowerCamelCase__			,						tensor_type=lowerCamelCase__							)








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


__A        =    logging.get_logger(__name__)







class     __lowerCAmelCase (       __magic_name__ ):

			"""simple docstring"""







			snake_case_   		=       ['''input_features''', '''is_longer''']


			def __init__(      self			,						lowerCamelCase__=64			,						lowerCamelCase__=48_000			,						lowerCamelCase__=480			,						lowerCamelCase__=10			,						lowerCamelCase__=1_024			,						lowerCamelCase__=0.0			,						lowerCamelCase__=False			,						lowerCamelCase__ = 0			,						lowerCamelCase__ = 14_000			,						lowerCamelCase__ = None			,						lowerCamelCase__ = "fusion"			,						lowerCamelCase__ = "repeatpad"			,						**lowerCamelCase__			,						)    ->      Tuple:



							'''simple docstring'''
							super().__init__(
							    feature_size=lowerCamelCase__			,						sampling_rate=lowerCamelCase__			,						padding_value=lowerCamelCase__			,						return_attention_mask=lowerCamelCase__			,						**lowerCamelCase__			,						)
							__lowerCamelCase		       =    top_db
							__lowerCamelCase		       =    truncation
							__lowerCamelCase		       =    padding
							__lowerCamelCase		       =    fft_window_size
							__lowerCamelCase		       =    (fft_window_size >> 1) + 1
							__lowerCamelCase		       =    hop_length
							__lowerCamelCase		       =    max_length_s
							__lowerCamelCase		       =    max_length_s * sampling_rate
							__lowerCamelCase		       =    sampling_rate
							__lowerCamelCase		       =    frequency_min
							__lowerCamelCase		       =    frequency_max
							__lowerCamelCase		       =    mel_filter_bank(
							    num_frequency_bins=self.nb_frequency_bins			,						num_mel_filters=lowerCamelCase__			,						min_frequency=lowerCamelCase__			,						max_frequency=lowerCamelCase__			,						sampling_rate=lowerCamelCase__			,						norm=lowerCamelCase__			,						mel_scale='htk'			,						)
							__lowerCamelCase		       =    mel_filter_bank(
							    num_frequency_bins=self.nb_frequency_bins			,						num_mel_filters=lowerCamelCase__			,						min_frequency=lowerCamelCase__			,						max_frequency=lowerCamelCase__			,						sampling_rate=lowerCamelCase__			,						norm='slaney'			,						mel_scale='slaney'			,						)


			def       lowercase_    (      self							)    ->      Dict[str, Any]:



							'''simple docstring'''
							__lowerCamelCase		       =    copy.deepcopy(self.__dict__							)
							__lowerCamelCase		       =    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       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__ = None							)    ->      np.ndarray:



							'''simple docstring'''
							__lowerCamelCase		       =    spectrogram(
							    lowerCamelCase__			,						window_function(self.fft_window_size			,						'hann'							)			,						frame_length=self.fft_window_size			,						hop_length=self.hop_length			,						power=2.0			,						mel_filters=lowerCamelCase__			,						log_mel='dB'			,						)
							return log_mel_spectrogram.T


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__							)    ->      int:



							'''simple docstring'''
							__lowerCamelCase		       =    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
											__lowerCamelCase		       =    [0]
							if len(ranges[2]							) == 0:
											# if the audio is too short, we just use the first chunk
											__lowerCamelCase		       =    [0]
							# randomly choose index for each part
							__lowerCamelCase		       =    np.random.choice(ranges[0]							)
							__lowerCamelCase		       =    np.random.choice(ranges[1]							)
							__lowerCamelCase		       =    np.random.choice(ranges[2]							)

							__lowerCamelCase		       =    mel[idx_front : idx_front + chunk_frames, :]
							__lowerCamelCase		       =    mel[idx_middle : idx_middle + chunk_frames, :]
							__lowerCamelCase		       =    mel[idx_back : idx_back + chunk_frames, :]

							__lowerCamelCase		       =    torch.tensor(mel[None, None, :]							)
							__lowerCamelCase		       =    torch.nn.functional.interpolate(
							    lowerCamelCase__			,						size=[chunk_frames, 64]			,						mode='bilinear'			,						align_corners=lowerCamelCase__							)
							__lowerCamelCase		       =    mel_shrink[0][0].numpy()
							__lowerCamelCase		       =    np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]			,						axis=0							)
							return mel_fusion


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__							)    ->      np.array:



							'''simple docstring'''
							if waveform.shape[0] > max_length:
											if truncation == "rand_trunc":
															__lowerCamelCase		       =    True
															# random crop to max_length (for compatibility) -> this should be handled by self.pad
															__lowerCamelCase		       =    len(lowerCamelCase__							) - max_length
															__lowerCamelCase		       =    np.random.randint(0			,						overflow + 1							)
															__lowerCamelCase		       =    waveform[idx : idx + max_length]
															__lowerCamelCase		       =    self._np_extract_fbank_features(lowerCamelCase__			,						self.mel_filters_slaney							)[None, :]
											elif truncation == "fusion":
															__lowerCamelCase		       =    self._np_extract_fbank_features(lowerCamelCase__			,						self.mel_filters							)
															__lowerCamelCase		       =    max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
															__lowerCamelCase		       =    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.
																			__lowerCamelCase		       =    np.stack([mel, mel, mel, mel]			,						axis=0							)
																			__lowerCamelCase		       =    False
															else:
																			__lowerCamelCase		       =    self._random_mel_fusion(lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__							)
																			__lowerCamelCase		       =    True
											else:
															raise NotImplementedError(f"""data_truncating {truncation} not implemented"""							)

							else:
											__lowerCamelCase		       =    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":
																			__lowerCamelCase		       =    int(max_length / len(lowerCamelCase__							)							)
																			__lowerCamelCase		       =    np.stack(np.tile(lowerCamelCase__			,						n_repeat + 1							)							)[:max_length]
															if padding == "repeatpad":
																			__lowerCamelCase		       =    int(max_length / len(lowerCamelCase__							)							)
																			__lowerCamelCase		       =    np.stack(np.tile(lowerCamelCase__			,						lowerCamelCase__							)							)
															__lowerCamelCase		       =    np.pad(lowerCamelCase__			,						(0, max_length - waveform.shape[0])			,						mode='constant'			,						constant_values=0							)

											if truncation == "fusion":
															__lowerCamelCase		       =    self._np_extract_fbank_features(lowerCamelCase__			,						self.mel_filters							)
															__lowerCamelCase		       =    np.stack([input_mel, input_mel, input_mel, input_mel]			,						axis=0							)
											else:
															__lowerCamelCase		       =    self._np_extract_fbank_features(lowerCamelCase__			,						self.mel_filters_slaney							)[None, :]

							return input_mel, longer






			def __call__(      self			,						lowerCamelCase__			,						lowerCamelCase__ = None			,						lowerCamelCase__ = None			,						lowerCamelCase__ = None			,						lowerCamelCase__ = None			,						lowerCamelCase__ = None			,						**lowerCamelCase__			,						)    ->      BatchFeature:



							'''simple docstring'''
							__lowerCamelCase		       =    truncation if truncation is not None else self.truncation
							__lowerCamelCase		       =    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.'							)

							__lowerCamelCase		       =    isinstance(lowerCamelCase__			,						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(lowerCamelCase__			,						(list, tuple)							) and (isinstance(raw_speech[0]			,						(np.ndarray, tuple, list)							))
							)

							if is_batched:
											__lowerCamelCase		       =    [np.asarray(lowerCamelCase__			,						dtype=np.floataa							) for speech in raw_speech]
							elif not is_batched and not isinstance(lowerCamelCase__			,						np.ndarray							):
											__lowerCamelCase		       =    np.asarray(lowerCamelCase__			,						dtype=np.floataa							)
							elif isinstance(lowerCamelCase__			,						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		       =    [np.asarray(lowerCamelCase__							)]

							# convert to mel spectrogram, truncate and pad if needed.
							__lowerCamelCase		       =    [
							    self._get_input_mel(lowerCamelCase__			,						max_length if max_length else self.nb_max_samples			,						lowerCamelCase__			,						lowerCamelCase__							)
							    for waveform in raw_speech
							]

							__lowerCamelCase		       =    []
							__lowerCamelCase		       =    []
							for mel, longer in padded_inputs:
											input_mel.append(lowerCamelCase__							)
											is_longer.append(lowerCamelCase__							)

							if truncation == "fusion" and sum(lowerCamelCase__							) == 0:
											# if no audio is longer than 10s, then randomly select one audio to be longer
											__lowerCamelCase		       =    np.random.randint(0			,						len(lowerCamelCase__							)							)
											__lowerCamelCase		       =    True

							if isinstance(input_mel[0]			,						lowerCamelCase__							):
											__lowerCamelCase		       =    [np.asarray(lowerCamelCase__			,						dtype=np.floataa							) for feature in input_mel]

							# is_longer is a list of bool
							__lowerCamelCase		       =    [[longer] for longer in is_longer]

							__lowerCamelCase		       =    {'input_features': input_mel, 'is_longer': is_longer}
							__lowerCamelCase		       =    BatchFeature(lowerCamelCase__							)

							if return_tensors is not None:
											__lowerCamelCase		       =    input_features.convert_to_tensors(lowerCamelCase__							)

							return input_features


