




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


a       :						str								=		logging.get_logger(__name__)



class 		SCREAMING_SNAKE_CASE__  (			_UpperCamelCase	):
			__SCREAMING_SNAKE_CASE    =			["""input_features""", """is_longer"""]





			def __init__(      self						:    Tuple      ,		a_						:    Union[str, Any]=64      ,		a_						:    Dict=48_000      ,		a_						:    List[str]=480      ,		a_						:    Dict=10      ,		a_						:    List[str]=1_024      ,		a_						:    int=0.0      ,		a_						:    str=False      ,		a_						:    float = 0      ,		a_						:    float = 14_000      ,		a_						:    int = None      ,		a_						:    str = "fusion"      ,		a_						:    str = "repeatpad"      ,		**a_						:    str      ,		):



						"""simple docstring"""




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





			def 							A (      self						:    List[Any]							):



						"""simple docstring"""




						__snake_case			=		copy.deepcopy(self.__dict__							)
						__snake_case			=		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						:    str      ,		a_						:    np.array      ,		a_						:    Optional[np.array] = None							):



						"""simple docstring"""




						__snake_case			=		spectrogram(
						    a_      ,		window_function(self.fft_window_size      ,		"hann"							)      ,		frame_length=self.fft_window_size      ,		hop_length=self.hop_length      ,		power=2.0      ,		mel_filters=a_      ,		log_mel="dB"      ,		)
						return log_mel_spectrogram.T





			def 							A (      self						:    Tuple      ,		a_						:    int      ,		a_						:    Tuple      ,		a_						:    List[Any]							):



						"""simple docstring"""




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

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

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





			def 							A (      self						:    List[Any]      ,		a_						:    np.array      ,		a_						:    Tuple      ,		a_						:    List[Any]      ,		a_						:    str							):



						"""simple docstring"""




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

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

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

						return input_mel, longer






			def __call__(      self						:    Any      ,		a_						:    Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]      ,		a_						:    str = None      ,		a_						:    Optional[str] = None      ,		a_						:    Optional[int] = None      ,		a_						:    Optional[int] = None      ,		a_						:    Optional[Union[str, TensorType]] = None      ,		**a_						:    Union[str, Any]      ,		):



						"""simple docstring"""




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

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

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

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

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

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

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

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

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

						__snake_case			=		{"input_features": input_mel, "is_longer": is_longer}
						__snake_case			=		BatchFeature(a_							)

						if return_tensors is not None:
									__snake_case			=		input_features.convert_to_tensors(a_							)

						return input_features







from typing import Dict, List, Optional, Tuple, 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_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_torch_available, is_torch_tensor, logging


if is_torch_available():
							import torch


UpperCamelCase_						:     Tuple	       =							logging.get_logger(__name__)




class   __lowercase						(						__snake_case       ):
	_A       =   ["pixel_values"]






	def __init__(self				: List[str]      ,	snake_case				: bool = True      ,	snake_case				: Optional[Dict[str, int]] = None      ,	snake_case				: PILImageResampling = PILImageResampling.BILINEAR      ,	snake_case				: bool = True      ,	snake_case				: Dict[str, int] = None      ,	snake_case				: bool = True      ,	snake_case				: Union[int, float] = 1 / 255      ,	snake_case				: bool = True      ,	snake_case				: Optional[Union[float, List[float]]] = None      ,	snake_case				: Optional[Union[float, List[float]]] = None      ,	**snake_case				: List[Any]      ,	) ->			None:
							super().__init__(**snake_case    )
							_lowercase       :	List[str]           =   size if size is not None else {"shortest_edge": 256}
							_lowercase       :	Union[str, Any]           =   get_size_dict(snake_case      ,	default_to_square=snake_case    )
							_lowercase       :	List[str]           =   crop_size if crop_size is not None else {"height": 224, "width": 224}
							_lowercase       :	Dict           =   get_size_dict(snake_case      ,	param_name="crop_size"    )
							_lowercase       :	List[Any]           =   do_resize
							_lowercase       :	Optional[Any]           =   size
							_lowercase       :	Tuple           =   resample
							_lowercase       :	Tuple           =   do_center_crop
							_lowercase       :	Any           =   crop_size
							_lowercase       :	str           =   do_rescale
							_lowercase       :	int           =   rescale_factor
							_lowercase       :	List[Any]           =   do_normalize
							_lowercase       :	List[str]           =   image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
							_lowercase       :	Optional[Any]           =   image_std if image_std is not None else IMAGENET_STANDARD_STD






	def       _a(self				: Tuple      ,	snake_case				: np.ndarray      ,	snake_case				: Dict[str, int]      ,	snake_case				: PILImageResampling = PILImageResampling.BICUBIC      ,	snake_case				: Optional[Union[str, ChannelDimension]] = None      ,	**snake_case				: int      ,	) ->			np.ndarray:
							_lowercase       :	Union[str, Any]           =   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()}"""    )
							_lowercase       :	int           =   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       _a(self				: str      ,	snake_case				: np.ndarray      ,	snake_case				: Dict[str, int]      ,	snake_case				: Optional[Union[str, ChannelDimension]] = None      ,	**snake_case				: Union[str, Any]      ,	) ->			np.ndarray:
							_lowercase       :	Union[str, Any]           =   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` and `width`. Got {size.keys()}"""    )
							return center_crop(snake_case      ,	size=(size["height"], size["width"])      ,	data_format=snake_case      ,	**snake_case    )






	def       _a(self				: Union[str, Any]      ,	snake_case				: np.ndarray      ,	snake_case				: float      ,	snake_case				: Optional[Union[str, ChannelDimension]] = None      ,	**snake_case				: List[Any]    ) ->			np.ndarray:
							return rescale(snake_case      ,	scale=snake_case      ,	data_format=snake_case      ,	**snake_case    )






	def       _a(self				: int      ,	snake_case				: np.ndarray      ,	snake_case				: Union[float, List[float]]      ,	snake_case				: Union[float, List[float]]      ,	snake_case				: Optional[Union[str, ChannelDimension]] = None      ,	**snake_case				: Tuple      ,	) ->			np.ndarray:
							return normalize(snake_case      ,	mean=snake_case      ,	std=snake_case      ,	data_format=snake_case      ,	**snake_case    )






	def       _a(self				: Optional[int]      ,	snake_case				: ImageInput      ,	snake_case				: Optional[bool] = None      ,	snake_case				: Dict[str, int] = None      ,	snake_case				: PILImageResampling = None      ,	snake_case				: bool = None      ,	snake_case				: Dict[str, int] = None      ,	snake_case				: Optional[bool] = None      ,	snake_case				: Optional[float] = None      ,	snake_case				: Optional[bool] = None      ,	snake_case				: Optional[Union[float, List[float]]] = None      ,	snake_case				: Optional[Union[float, List[float]]] = None      ,	snake_case				: Optional[Union[str, TensorType]] = None      ,	snake_case				: Union[str, ChannelDimension] = ChannelDimension.FIRST      ,	**snake_case				: Tuple      ,	) ->			Union[str, Any]:
							_lowercase       :	Any           =   do_resize if do_resize is not None else self.do_resize
							_lowercase       :	List[str]           =   size if size is not None else self.size
							_lowercase       :	Optional[int]           =   get_size_dict(snake_case      ,	default_to_square=snake_case    )
							_lowercase       :	Tuple           =   resample if resample is not None else self.resample
							_lowercase       :	List[Any]           =   do_center_crop if do_center_crop is not None else self.do_center_crop
							_lowercase       :	int           =   crop_size if crop_size is not None else self.crop_size
							_lowercase       :	Dict           =   get_size_dict(snake_case      ,	param_name="crop_size"    )
							_lowercase       :	Dict           =   do_rescale if do_rescale is not None else self.do_rescale
							_lowercase       :	List[Any]           =   rescale_factor if rescale_factor is not None else self.rescale_factor
							_lowercase       :	Union[str, Any]           =   do_normalize if do_normalize is not None else self.do_normalize
							_lowercase       :	Dict           =   image_mean if image_mean is not None else self.image_mean
							_lowercase       :	Optional[Any]           =   image_std if image_std is not None else self.image_std

							_lowercase       :	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.
							_lowercase       :	Tuple           =   [to_numpy_array(snake_case    ) for image in images]

							if do_resize:
													_lowercase       :	List[str]           =   [self.resize(image=snake_case      ,	size=snake_case      ,	resample=snake_case    ) for image in images]

							if do_center_crop:
													_lowercase       :	List[str]           =   [self.center_crop(image=snake_case      ,	size=snake_case    ) for image in images]

							if do_rescale:
													_lowercase       :	Any           =   [self.rescale(image=snake_case      ,	scale=snake_case    ) for image in images]

							if do_normalize:
													_lowercase       :	Optional[Any]           =   [self.normalize(image=snake_case      ,	mean=snake_case      ,	std=snake_case    ) for image in images]

							_lowercase       :	Any           =   [to_channel_dimension_format(snake_case      ,	snake_case    ) for image in images]

							_lowercase       :	Optional[Any]           =   {"pixel_values": images}
							return BatchFeature(data=snake_case      ,	tensor_type=snake_case    )






	def       _a(self				: Dict      ,	snake_case				: List[str]      ,	snake_case				: List[Tuple] = None    ) ->			Optional[Any]:
							_lowercase       :	Dict           =   outputs.logits

							# Resize logits and compute semantic segmentation maps
							if target_sizes is not None:
													if len(snake_case    ) != len(snake_case    ):
																			raise ValueError(
																			    "Make sure that you pass in as many target sizes as the batch dimension of the logits"    )

													if is_torch_tensor(snake_case    ):
																			_lowercase       :	Dict           =   target_sizes.numpy()

													_lowercase       :	Tuple           =   []

													for idx in range(len(snake_case    )    ):
																			_lowercase       :	List[str]           =   torch.nn.functional.interpolate(
																			    logits[idx].unsqueeze(dim=0    )      ,	size=target_sizes[idx]      ,	mode="bilinear"      ,	align_corners=snake_case    )
																			_lowercase       :	int           =   resized_logits[0].argmax(dim=0    )
																			semantic_segmentation.append(snake_case    )
							else:
													_lowercase       :	Optional[int]           =   logits.argmax(dim=1    )
													_lowercase       :	int           =   [semantic_segmentation[i] for i in range(semantic_segmentation.shape[0]    )]

							return semantic_segmentation




