


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

import numpy as np

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


_snake_case	:						Tuple         =							logging.get_logger(__name__)

class 							a						(_lowerCAmelCase ):




							"""simple docstring"""


							__UpperCAmelCase       :				Optional[int]	    =    ["input_features"]

							def __init__(				self			:    int       ,      lowerCamelCase			:    str=80       ,      lowerCamelCase			:    List[Any]=16000       ,      lowerCamelCase			:    Dict=160       ,      lowerCamelCase			:    int=30       ,      lowerCamelCase			:    Any=400       ,      lowerCamelCase			:    Optional[Any]=0.0       ,      lowerCamelCase			:    Any=False       ,      **lowerCamelCase			:    Optional[Any]       ,      )      ->       Dict:
														super().__init__(
														    feature_size=lowerCamelCase       ,      sampling_rate=lowerCamelCase       ,      padding_value=lowerCamelCase       ,      return_attention_mask=lowerCamelCase       ,      **lowerCamelCase       ,      )
														__snake_case						:     Union[str, Any]									=  n_fft
														__snake_case						:     List[Any]									=  hop_length
														__snake_case						:     Optional[Any]									=  chunk_length
														__snake_case						:     List[str]									=  chunk_length * sampling_rate
														__snake_case						:     str									=  self.n_samples // hop_length
														__snake_case						:     Any									=  sampling_rate
														__snake_case						:     Dict									=  mel_filter_bank(
														    num_frequency_bins=1 + n_fft // 2       ,      num_mel_filters=lowerCamelCase       ,      min_frequency=0.0       ,      max_frequency=80_00.0       ,      sampling_rate=lowerCamelCase       ,      norm="slaney"       ,      mel_scale="slaney"       ,      )

							def        __snake_case   (				self			:    List[Any]       ,      lowerCamelCase			:    np.array							)      ->       np.ndarray:
														__snake_case						:     Union[str, Any]									=  spectrogram(
														    lowerCamelCase       ,      window_function(self.n_fft       ,      "hann"							)       ,      frame_length=self.n_fft       ,      hop_length=self.hop_length       ,      power=2.0       ,      mel_filters=self.mel_filters       ,      log_mel="log10"       ,      )
														__snake_case						:     Dict									=  log_spec[:, :-1]
														__snake_case						:     Any									=  np.maximum(lowerCamelCase       ,      log_spec.max() - 8.0							)
														__snake_case						:     Union[str, Any]									=  (log_spec + 4.0) / 4.0
														return log_spec

							@staticmethod
							# Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
							def        __snake_case   (				lowerCamelCase			:    List[np.ndarray]       ,      lowerCamelCase			:    List[np.ndarray]       ,      lowerCamelCase			:    float = 0.0							)      ->       List[np.ndarray]:
														if attention_mask is not None:
																					__snake_case						:     List[str]									=  np.array(lowerCamelCase       ,      np.intaa							)
																					__snake_case						:     int									=  []

																					for vector, length in zip(lowerCamelCase       ,      attention_mask.sum(-1							)							):
																												__snake_case						:     int									=  (vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7							)
																												if length < normed_slice.shape[0]:
																																			__snake_case						:     int									=  padding_value

																												normed_input_values.append(lowerCamelCase							)
														else:
																					__snake_case						:     List[str]									=  [(x - x.mean()) / np.sqrt(x.var() + 1E-7							) for x in input_values]

														return normed_input_values

							def __call__(				self			:    Any       ,      lowerCamelCase			:    Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]       ,      lowerCamelCase			:    bool = True       ,      lowerCamelCase			:    Optional[int] = None       ,      lowerCamelCase			:    Optional[Union[str, TensorType]] = None       ,      lowerCamelCase			:    Optional[bool] = None       ,      lowerCamelCase			:    Optional[str] = "max_length"       ,      lowerCamelCase			:    Optional[int] = None       ,      lowerCamelCase			:    Optional[int] = None       ,      lowerCamelCase			:    Optional[bool] = None       ,      **lowerCamelCase			:    Optional[Any]       ,      )      ->       BatchFeature:

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

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

														# always return batch
														if not is_batched:
																					__snake_case						:     List[Any]									=  [np.asarray([raw_speech]							).T]

														__snake_case						:     List[str]									=  BatchFeature({"input_features": raw_speech}							)

														# convert into correct format for padding

														__snake_case						:     Union[str, Any]									=  self.pad(
														    lowerCamelCase       ,      padding=lowerCamelCase       ,      max_length=max_length if max_length else self.n_samples       ,      truncation=lowerCamelCase       ,      pad_to_multiple_of=lowerCamelCase       ,      return_attention_mask=return_attention_mask or do_normalize       ,      )

														# zero-mean and unit-variance normalization
														if do_normalize:
																					__snake_case						:     List[str]									=  self.zero_mean_unit_var_norm(
																					    padded_inputs["input_features"]       ,      attention_mask=padded_inputs["attention_mask"]       ,      padding_value=self.padding_value       ,      )
																					__snake_case						:     Optional[int]									=  np.stack(padded_inputs["input_features"]       ,      axis=0							)

														# make sure list is in array format
														__snake_case						:     Optional[int]									=  padded_inputs.get("input_features"							).transpose(2       ,      0       ,      1							)

														__snake_case						:     Tuple									=  [self._np_extract_fbank_features(lowerCamelCase							) for waveform in input_features[0]]

														if isinstance(input_features[0]       ,      lowerCamelCase							):
																					__snake_case						:     Dict									=  [np.asarray(lowerCamelCase       ,      dtype=np.floataa							) for feature in input_features]
														else:
																					__snake_case						:     Any									=  input_features

														if return_attention_mask:
																					# rescale from sample (48000) to feature (3000)
																					__snake_case						:     Tuple									=  padded_inputs["attention_mask"][:, :: self.hop_length]

														if return_tensors is not None:
																					__snake_case						:     Dict									=  padded_inputs.convert_to_tensors(lowerCamelCase							)

														return padded_inputs






							def        __snake_case   (				self			:    str							)      ->       Dict[str, Any]:
														__snake_case						:     Tuple									=  copy.deepcopy(self.__dict__							)
														__snake_case						:     str									=  self.__class__.__name__
														if "mel_filters" in output:
																					del output["mel_filters"]
														return output








'''simple docstring'''




import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin
from .feature_extraction_wavaveca import WavaVecaFeatureExtractor
from .tokenization_wavaveca import WavaVecaCTCTokenizer




class      UpperCamelCase__					(					lowercase_			):




					"""simple docstring"""
					SCREAMING_SNAKE_CASE__				=	'''Wav2Vec2FeatureExtractor'''
					SCREAMING_SNAKE_CASE__				=	'''AutoTokenizer'''


					def __init__(						self				:			Optional[int]      ,						lowerCamelCase_				:			str      ,						lowerCamelCase_				:			Dict						):

							'''simple docstring'''



							super().__init__(lowerCamelCase_      ,						lowerCamelCase_						)
							SCREAMING_SNAKE_CASE :				int		=			self.feature_extractor
							SCREAMING_SNAKE_CASE :				List[Any]		=			False


					@classmethod
					def lowerCamelCase_			(						cls				:			int      ,						lowerCamelCase_				:			List[str]      ,						**lowerCamelCase_				:			Any						):

							'''simple docstring'''



							try:
									return super().from_pretrained(lowerCamelCase_      ,						**lowerCamelCase_						)
							except OSError:
									warnings.warn(
									    f'''Loading a tokenizer inside {cls.__name__} from a config that does not'''
									    """ include a `tokenizer_class` attribute is deprecated and will be """
									    """removed in v5. Please add `'tokenizer_class': 'Wav2Vec2CTCTokenizer'`"""
									    """ attribute to either your `config.json` or `tokenizer_config.json` """
									    """file to suppress this warning: """      ,						lowerCamelCase_      ,						)

									SCREAMING_SNAKE_CASE :				Optional[Any]		=			WavaVecaFeatureExtractor.from_pretrained(lowerCamelCase_      ,						**lowerCamelCase_						)
									SCREAMING_SNAKE_CASE :				Dict		=			WavaVecaCTCTokenizer.from_pretrained(lowerCamelCase_      ,						**lowerCamelCase_						)

									return cls(feature_extractor=lowerCamelCase_      ,						tokenizer=lowerCamelCase_						)


					def __call__(						self				:			List[Any]      ,						*lowerCamelCase_				:			str      ,						**lowerCamelCase_				:			int						):

							'''simple docstring'''



							if self._in_target_context_manager:
									return self.current_processor(*lowerCamelCase_      ,						**lowerCamelCase_						)

							if "raw_speech" in kwargs:
									warnings.warn("""Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead."""						)
									SCREAMING_SNAKE_CASE :				Dict		=			kwargs.pop("""raw_speech"""						)
							else:
									SCREAMING_SNAKE_CASE :				Any		=			kwargs.pop("""audio"""      ,						lowerCamelCase_						)
							SCREAMING_SNAKE_CASE :				Tuple		=			kwargs.pop("""sampling_rate"""      ,						lowerCamelCase_						)
							SCREAMING_SNAKE_CASE :				int		=			kwargs.pop("""text"""      ,						lowerCamelCase_						)
							if len(lowerCamelCase_						) > 0:
									SCREAMING_SNAKE_CASE :				Union[str, Any]		=			args[0]
									SCREAMING_SNAKE_CASE :				Dict		=			args[1:]

							if audio is None and text is None:
									raise ValueError("""You need to specify either an `audio` or `text` input to process."""						)

							if audio is not None:
									SCREAMING_SNAKE_CASE :				str		=			self.feature_extractor(lowerCamelCase_      ,						*lowerCamelCase_      ,						sampling_rate=lowerCamelCase_      ,						**lowerCamelCase_						)
							if text is not None:
									SCREAMING_SNAKE_CASE :				List[Any]		=			self.tokenizer(lowerCamelCase_      ,						**lowerCamelCase_						)

							if text is None:
									return inputs
							elif audio is None:
									return encodings
							else:
									SCREAMING_SNAKE_CASE :				List[str]		=			encodings["""input_ids"""]
									return inputs


					def lowerCamelCase_			(						self				:			Optional[Any]      ,						*lowerCamelCase_				:			Tuple      ,						**lowerCamelCase_				:			Dict						):

							'''simple docstring'''



							if self._in_target_context_manager:
									return self.current_processor.pad(*lowerCamelCase_      ,						**lowerCamelCase_						)

							SCREAMING_SNAKE_CASE :				str		=			kwargs.pop("""input_features"""      ,						lowerCamelCase_						)
							SCREAMING_SNAKE_CASE :				Dict		=			kwargs.pop("""labels"""      ,						lowerCamelCase_						)
							if len(lowerCamelCase_						) > 0:
									SCREAMING_SNAKE_CASE :				Optional[int]		=			args[0]
									SCREAMING_SNAKE_CASE :				Optional[Any]		=			args[1:]

							if input_features is not None:
									SCREAMING_SNAKE_CASE :				Any		=			self.feature_extractor.pad(lowerCamelCase_      ,						*lowerCamelCase_      ,						**lowerCamelCase_						)
							if labels is not None:
									SCREAMING_SNAKE_CASE :				int		=			self.tokenizer.pad(lowerCamelCase_      ,						**lowerCamelCase_						)

							if labels is None:
									return input_features
							elif input_features is None:
									return labels
							else:
									SCREAMING_SNAKE_CASE :				Dict		=			labels["""input_ids"""]
									return input_features


					def lowerCamelCase_			(						self				:			Optional[int]      ,						*lowerCamelCase_				:			List[str]      ,						**lowerCamelCase_				:			Optional[Any]						):

							'''simple docstring'''



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


					def lowerCamelCase_			(						self				:			List[str]      ,						*lowerCamelCase_				:			Optional[int]      ,						**lowerCamelCase_				:			Optional[int]						):

							'''simple docstring'''



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



					@contextmanager
					def lowerCamelCase_			(						self				:			Tuple						):

							'''simple docstring'''



							warnings.warn(
							    """`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your """
							    """labels by using the argument `text` of the regular `__call__` method (either in the same call as """
							    """your audio inputs, or in a separate call."""						)
							SCREAMING_SNAKE_CASE :				int		=			True
							SCREAMING_SNAKE_CASE :				Optional[Any]		=			self.tokenizer
							yield
							SCREAMING_SNAKE_CASE :				int		=			self.feature_extractor
							SCREAMING_SNAKE_CASE :				Optional[Any]		=			False



