


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


_lowercase    :    List[Any]			     =  logging.get_logger(__name__)






class  UpperCamelCase__(     lowerCAmelCase ):
				__magic_name__       :   str							= ["input_features", "is_longer"]




				def __init__(	self    :				Union[str, Any]   ,					lowerCAmelCase    :				int=64   ,					lowerCAmelCase    :				int=48000   ,					lowerCAmelCase    :				int=480   ,					lowerCAmelCase    :				List[Any]=10   ,					lowerCAmelCase    :				Dict=1024   ,					lowerCAmelCase    :				Tuple=0.0   ,					lowerCAmelCase    :				Dict=False   ,					lowerCAmelCase    :				float = 0   ,					lowerCAmelCase    :				float = 14000   ,					lowerCAmelCase    :				int = None   ,					lowerCAmelCase    :				str = "fusion"   ,					lowerCAmelCase    :				str = "repeatpad"   ,					**lowerCAmelCase    :				str   ,					)->  int:

										"""simple docstring"""




										super().__init__(
										    feature_size=lowerCAmelCase   ,					sampling_rate=lowerCAmelCase   ,					padding_value=lowerCAmelCase   ,					return_attention_mask=lowerCAmelCase   ,					**lowerCAmelCase   ,					)
										UpperCAmelCase         =			top_db
										UpperCAmelCase         =			truncation
										UpperCAmelCase         =			padding
										UpperCAmelCase         =			fft_window_size
										UpperCAmelCase         =			(fft_window_size >> 1) + 1
										UpperCAmelCase         =			hop_length
										UpperCAmelCase         =			max_length_s
										UpperCAmelCase         =			max_length_s * sampling_rate
										UpperCAmelCase         =			sampling_rate
										UpperCAmelCase         =			frequency_min
										UpperCAmelCase         =			frequency_max
										UpperCAmelCase         =			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'''   ,					)
										UpperCAmelCase         =			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   a__(	self    :				Optional[int]   )->  Dict[str, Any]:

										"""simple docstring"""




										UpperCAmelCase         =			copy.deepcopy(self.__dict__   )
										UpperCAmelCase         =			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   a__(	self    :				List[str]   ,					lowerCAmelCase    :				np.array   ,					lowerCAmelCase    :				Optional[np.array] = None   )->  np.ndarray:

										"""simple docstring"""




										UpperCAmelCase         =			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   a__(	self    :				Union[str, Any]   ,					lowerCAmelCase    :				Tuple   ,					lowerCAmelCase    :				List[str]   ,					lowerCAmelCase    :				Optional[Any]   )->  List[Any]:

										"""simple docstring"""




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

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

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




				def   a__(	self    :				List[Any]   ,					lowerCAmelCase    :				np.array   ,					lowerCAmelCase    :				List[str]   ,					lowerCAmelCase    :				Optional[Any]   ,					lowerCAmelCase    :				Any   )->  np.array:

										"""simple docstring"""




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

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

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

										return input_mel, longer





				def __call__(	self    :				List[str]   ,					lowerCAmelCase    :				Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]   ,					lowerCAmelCase    :				str = None   ,					lowerCAmelCase    :				Optional[str] = None   ,					lowerCAmelCase    :				Optional[int] = None   ,					lowerCAmelCase    :				Optional[int] = None   ,					lowerCAmelCase    :				Optional[Union[str, TensorType]] = None   ,					**lowerCAmelCase    :				str   ,					)->  BatchFeature:

										"""simple docstring"""




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

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

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

										# always return batch
										if not is_batched:
																UpperCAmelCase         =			[np.asarray(lowerCAmelCase   )]

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

										UpperCAmelCase         =			[]
										UpperCAmelCase         =			[]
										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
																UpperCAmelCase         =			np.random.randint(0   ,					len(lowerCAmelCase   )   )
																UpperCAmelCase         =			True

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

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

										UpperCAmelCase         =			{'''input_features''': input_mel, '''is_longer''': is_longer}
										UpperCAmelCase         =			BatchFeature(lowerCAmelCase   )

										if return_tensors is not None:
																UpperCAmelCase         =			input_features.convert_to_tensors(lowerCAmelCase   )

										return input_features









'''simple docstring'''



import warnings

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding






class  UpperCamelCase__(     lowerCAmelCase ):
				__magic_name__       :   Tuple							= ["image_processor", "tokenizer"]
				__magic_name__       :   Any							= "ViTImageProcessor"
				__magic_name__       :   str							= ("CLIPTokenizer", "CLIPTokenizerFast")




				def __init__(	self    :				List[str]   ,					lowerCAmelCase    :				List[Any]=None   ,					lowerCAmelCase    :				List[str]=None   ,					**lowerCAmelCase    :				Optional[int]   )->  List[Any]:

										"""simple docstring"""




										UpperCAmelCase         =			None
										if "feature_extractor" in kwargs:
																warnings.warn(
																    '''The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'''
																    ''' instead.'''   ,					lowerCAmelCase   ,					)
																UpperCAmelCase         =			kwargs.pop('''feature_extractor'''   )

										UpperCAmelCase         =			image_processor if image_processor is not None else feature_extractor
										if image_processor is None:
																raise ValueError('''You need to specify an `image_processor`.'''   )
										if tokenizer is None:
																raise ValueError('''You need to specify a `tokenizer`.'''   )

										super().__init__(lowerCAmelCase   ,					lowerCAmelCase   )




				def __call__(	self    :				str   ,					lowerCAmelCase    :				Optional[Any]=None   ,					lowerCAmelCase    :				Optional[Any]=None   ,					lowerCAmelCase    :				Dict=None   ,					lowerCAmelCase    :				str=None   ,					**lowerCAmelCase    :				List[str]   )->  List[Any]:

										"""simple docstring"""




										if text is None and visual_prompt is None and images is None:
																raise ValueError('''You have to specify either text, visual prompt or images.'''   )

										if text is not None and visual_prompt is not None:
																raise ValueError('''You have to specify exactly one type of prompt. Either text or visual prompt.'''   )

										if text is not None:
																UpperCAmelCase         =			self.tokenizer(lowerCAmelCase   ,					return_tensors=lowerCAmelCase   ,					**lowerCAmelCase   )

										if visual_prompt is not None:
																UpperCAmelCase         =			self.image_processor(lowerCAmelCase   ,					return_tensors=lowerCAmelCase   ,					**lowerCAmelCase   )

										if images is not None:
																UpperCAmelCase         =			self.image_processor(lowerCAmelCase   ,					return_tensors=lowerCAmelCase   ,					**lowerCAmelCase   )

										if visual_prompt is not None and images is not None:
																UpperCAmelCase         =			{
																    '''pixel_values''': image_features.pixel_values,
																    '''conditional_pixel_values''': prompt_features.pixel_values,
																}
																return encoding
										elif text is not None and images is not None:
																UpperCAmelCase         =			image_features.pixel_values
																return encoding
										elif text is not None:
																return encoding
										elif visual_prompt is not None:
																UpperCAmelCase         =			{
																    '''conditional_pixel_values''': prompt_features.pixel_values,
																}
																return encoding
										else:
																return BatchEncoding(data=dict(**lowerCAmelCase   )   ,					tensor_type=lowerCAmelCase   )




				def   a__(	self    :				List[str]   ,					*lowerCAmelCase    :				str   ,					**lowerCAmelCase    :				Optional[int]   )->  Union[str, Any]:

										"""simple docstring"""




										return self.tokenizer.batch_decode(*lowerCAmelCase   ,					**lowerCAmelCase   )




				def   a__(	self    :				Dict   ,					*lowerCAmelCase    :				Tuple   ,					**lowerCAmelCase    :				List[str]   )->  Optional[Any]:

										"""simple docstring"""




										return self.tokenizer.decode(*lowerCAmelCase   ,					**lowerCAmelCase   )




				@property
				def   a__(	self    :				List[Any]   )->  Optional[Any]:

										"""simple docstring"""




										warnings.warn(
										    '''`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.'''   ,					lowerCAmelCase   ,					)
										return self.image_processor_class





				@property
				def   a__(	self    :				Any   )->  List[Any]:

										"""simple docstring"""




										warnings.warn(
										    '''`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.'''   ,					lowerCAmelCase   ,					)
										return self.image_processor





