

'''simple docstring'''






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

import numpy as np

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


__SCREAMING_SNAKE_CASE     =						logging.get_logger(__name__)






class 			lowerCAmelCase__     (					snake_case_						):




     """simple docstring"""





     __UpperCamelCase									= ["input_values", "attention_mask"]







     def __init__(		self						:	List[Any]      ,						A__						:	Optional[int] = 1      ,						A__						:	Any = 1_6_0_0_0      ,						A__						:	Dict = 0.0      ,						A__						:	str = False      ,						A__						:	Optional[Any] = 8_0      ,						A__						:	Any = 1_6      ,						A__						:	Optional[int] = 6_4      ,						A__						:	Dict = "hann_window"      ,						A__						:	Any = 1.0      ,						A__						:	Any = 8_0      ,						A__						:	Tuple = 7_6_0_0      ,						A__						:	Tuple = 1E-10      ,						A__						:	Union[str, Any] = 2      ,						A__						:	Tuple = True      ,						**A__						:	str      ,						)       ->			int:
          '''simple docstring'''




          super().__init__(feature_size=A__      ,						sampling_rate=A__      ,						padding_value=A__      ,						**A__	)
          a__   :      List[Any]     =						do_normalize
          a__   :      str     =						return_attention_mask

          a__   :      str     =						num_mel_bins
          a__   :      Any     =						hop_length
          a__   :      Optional[int]     =						win_length
          a__   :      Dict     =						win_function
          a__   :      Tuple     =						frame_signal_scale
          a__   :      Tuple     =						fmin
          a__   :      List[Any]     =						fmax
          a__   :      int     =						mel_floor
          a__   :      Optional[Any]     =						reduction_factor

          a__   :      Tuple     =						win_length * sampling_rate // 1_0_0_0
          a__   :      Dict     =						hop_length * sampling_rate // 1_0_0_0
          a__   :      Any     =						optimal_fft_length(self.sample_size	)
          a__   :      str     =						(self.n_fft // 2) + 1

          a__   :      List[str]     =						window_function(window_length=self.sample_size      ,						name=self.win_function      ,						periodic=A__	)

          a__   :      Optional[int]     =						mel_filter_bank(
              num_frequency_bins=self.n_freqs      ,						num_mel_filters=self.num_mel_bins      ,						min_frequency=self.fmin      ,						max_frequency=self.fmax      ,						sampling_rate=self.sampling_rate      ,						norm='''slaney'''      ,						mel_scale='''slaney'''      ,						)

          if frame_signal_scale != 1.0:
               warnings.warn(
                   '''The argument `frame_signal_scale` is deprecated and will be removed in version 4.30.0 of Transformers'''      ,						A__      ,						)
          if reduction_factor != 2.0:
               warnings.warn(
                   '''The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers'''      ,						A__      ,						)







     @staticmethod
     # Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
     def 			__lowerCAmelCase	(		A__						:	Optional[int]      ,						A__						:	int      ,						A__						:	List[Any] = 0.0	)       ->			List[np.ndarray]:
          '''simple docstring'''




          if attention_mask is not None:
               a__   :      Dict     =						np.array(A__      ,						np.intaa	)
               a__   :      Optional[int]     =						[]

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

                    normed_input_values.append(A__	)
          else:
               a__   :      Dict     =						[(x - x.mean()) / np.sqrt(x.var() + 1E-7	) for x in input_values]

          return normed_input_values







     def 			__lowerCAmelCase	(		self						:	Any      ,						A__						:	Tuple      ,						)       ->			np.ndarray:
          '''simple docstring'''




          a__   :      Tuple     =						spectrogram(
              A__      ,						window=self.window      ,						frame_length=self.sample_size      ,						hop_length=self.sample_stride      ,						fft_length=self.n_fft      ,						mel_filters=self.mel_filters      ,						mel_floor=self.mel_floor      ,						log_mel='''log10'''      ,						)
          return log_mel_spec.T







     def __call__(		self						:	Optional[Any]      ,						A__						:	Optional[Any] = None      ,						A__						:	Tuple = None      ,						A__						:	List[Any] = False      ,						A__						:	Optional[int] = None      ,						A__						:	Union[str, Any] = False      ,						A__						:	Optional[int] = None      ,						A__						:	Dict = None      ,						A__						:	str = None      ,						A__						:	Any = None      ,						**A__						:	Tuple      ,						)       ->			BatchFeature:
          '''simple docstring'''




          if audio is None and audio_target is None:
               raise ValueError('''You must provide either `audio` or `audio_target` values.'''	)

          if sampling_rate is not None:
               if sampling_rate != self.sampling_rate:
                    raise ValueError(
                        F'The model corresponding to this feature extractor: {self} was trained using a sampling rate of'
                        F' {self.sampling_rate}. Please make sure that the provided audio input was sampled with'
                        F' {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.'''	)

          if audio is not None:
               a__   :      Dict     =						self._process_audio(
                   A__      ,						A__      ,						A__      ,						A__      ,						A__      ,						A__      ,						A__      ,						A__      ,						**A__      ,						)
          else:
               a__   :      List[str]     =						None

          if audio_target is not None:
               a__   :      Tuple     =						self._process_audio(
                   A__      ,						A__      ,						A__      ,						A__      ,						A__      ,						A__      ,						A__      ,						A__      ,						**A__      ,						)

               if inputs is None:
                    return inputs_target
               else:
                    a__   :      List[str]     =						inputs_target['''input_values''']
                    a__   :      List[str]     =						inputs_target.get('''attention_mask'''	)
                    if decoder_attention_mask is not None:
                         a__   :      Optional[int]     =						decoder_attention_mask

          return inputs







     def 			__lowerCAmelCase	(		self						:	int      ,						A__						:	Any      ,						A__						:	Optional[int] = False      ,						A__						:	List[Any] = False      ,						A__						:	int = None      ,						A__						:	Dict = False      ,						A__						:	Union[str, Any] = None      ,						A__						:	List[str] = None      ,						A__						:	Optional[int] = None      ,						**A__						:	Optional[int]      ,						)       ->			BatchFeature:
          '''simple docstring'''




          a__   :      Tuple     =						isinstance(A__      ,						np.ndarray	) and len(speech.shape	) > 1
          if is_batched_numpy and len(speech.shape	) > 2:
               raise ValueError(F'Only mono-channel audio is supported for input to {self}'	)
          a__   :      List[str]     =						is_batched_numpy or (
              isinstance(A__      ,						(list, tuple)	) and (isinstance(speech[0]      ,						(np.ndarray, tuple, list)	))
          )

          if is_batched:
               a__   :      Union[str, Any]     =						[np.asarray(A__      ,						dtype=np.floataa	) for speech in speech]
          elif not is_batched and not isinstance(A__      ,						np.ndarray	):
               a__   :      List[Any]     =						np.asarray(A__      ,						dtype=np.floataa	)
          elif isinstance(A__      ,						np.ndarray	) and speech.dtype is np.dtype(np.floataa	):
               a__   :      Optional[int]     =						speech.astype(np.floataa	)

          # always return batch
          if not is_batched:
               a__   :      List[Any]     =						[speech]

          # needed to make pad() work on spectrogram inputs
          a__   :      str     =						self.feature_size

          # convert into correct format for padding
          if is_target:
               a__   :      Optional[Any]     =						[self._extract_mel_features(A__	) for waveform in speech]
               a__   :      List[Any]     =						BatchFeature({'''input_values''': features}	)
               a__   :      Any     =						self.num_mel_bins
          else:
               a__   :      Optional[int]     =						BatchFeature({'''input_values''': speech}	)

          a__   :      Optional[Any]     =						self.pad(
              A__      ,						padding=A__      ,						max_length=A__      ,						truncation=A__      ,						pad_to_multiple_of=A__      ,						return_attention_mask=A__      ,						**A__      ,						)

          a__   :      int     =						feature_size_hack

          # convert input values to correct format
          a__   :      List[str]     =						padded_inputs['''input_values''']
          if not isinstance(input_values[0]      ,						np.ndarray	):
               a__   :      int     =						[np.asarray(A__      ,						dtype=np.floataa	) for array in input_values]
          elif (
              not isinstance(A__      ,						np.ndarray	)
              and isinstance(input_values[0]      ,						np.ndarray	)
              and input_values[0].dtype is np.dtype(np.floataa	)
          ):
               a__   :      str     =						[array.astype(np.floataa	) for array in input_values]
          elif isinstance(A__      ,						np.ndarray	) and input_values.dtype is np.dtype(np.floataa	):
               a__   :      str     =						input_values.astype(np.floataa	)

          # convert attention_mask to correct format
          a__   :      str     =						padded_inputs.get('''attention_mask'''	)
          if attention_mask is not None:
               a__   :      List[Any]     =						[np.asarray(A__      ,						dtype=np.intaa	) for array in attention_mask]

          # zero-mean and unit-variance normalization
          if not is_target and self.do_normalize:
               a__   :      str     =						(
                   attention_mask
                   if self._get_padding_strategies(A__      ,						max_length=A__	) is not PaddingStrategy.DO_NOT_PAD
                   else None
               )
               a__   :      Dict     =						self.zero_mean_unit_var_norm(
                   padded_inputs['''input_values''']      ,						attention_mask=A__      ,						padding_value=self.padding_value	)

          if return_tensors is not None:
               a__   :      Union[str, Any]     =						padded_inputs.convert_to_tensors(A__	)

          return padded_inputs







     def 			__lowerCAmelCase	(		self						:	List[str]	)       ->			Dict[str, Any]:
          '''simple docstring'''




          a__   :      List[Any]     =						super().to_dict()

          # Don't serialize these as they are derived from the other properties.
          a__   :      Optional[Any]     =						['''window''', '''mel_filters''', '''sample_size''', '''sample_stride''', '''n_fft''', '''n_freqs''']
          for name in names:
               if name in output:
                    del output[name]

          return output


import warnings
from contextlib import contextmanager

from ....processing_utils import ProcessorMixin




class        _SCREAMING_SNAKE_CASE	( snake_case_		):
	lowerCAmelCase__   =					'MCTCTFeatureExtractor'
	lowerCAmelCase__   =					'AutoTokenizer'



	def __init__(       self		,    lowercase		,    lowercase					)							->			str:
					super().__init__(lowercase		,    lowercase					)
					lowerCamelCase_         =     self.feature_extractor
					lowerCamelCase_         =     False



	def __call__(       self		,    *lowercase		,    **lowercase					)							->			List[str]:
					# For backward compatibility
					if self._in_target_context_manager:
									return self.current_processor(*lowercase		,    **lowercase					)

					if "raw_speech" in kwargs:
									warnings.warn("Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead."					)
									lowerCamelCase_         =     kwargs.pop("raw_speech"					)
					else:
									lowerCamelCase_         =     kwargs.pop("audio"		,    lowercase					)
					lowerCamelCase_         =     kwargs.pop("sampling_rate"		,    lowercase					)
					lowerCamelCase_         =     kwargs.pop("text"		,    lowercase					)
					if len(lowercase					) > 0:
									lowerCamelCase_         =     args[0]
									lowerCamelCase_         =     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:
									lowerCamelCase_         =     self.feature_extractor(lowercase		,    *lowercase		,    sampling_rate=lowercase		,    **lowercase					)
					if text is not None:
									lowerCamelCase_         =     self.tokenizer(lowercase		,    **lowercase					)

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



	def        SCREAMING_SNAKE_CASE_(       self		,    *lowercase		,    **lowercase					)							->			Optional[Any]:
					return self.tokenizer.batch_decode(*lowercase		,    **lowercase					)



	def        SCREAMING_SNAKE_CASE_(       self		,    *lowercase		,    **lowercase					)							->			Optional[Any]:
					# For backward compatibility
					if self._in_target_context_manager:
									return self.current_processor.pad(*lowercase		,    **lowercase					)

					lowerCamelCase_         =     kwargs.pop("input_features"		,    lowercase					)
					lowerCamelCase_         =     kwargs.pop("labels"		,    lowercase					)
					if len(lowercase					) > 0:
									lowerCamelCase_         =     args[0]
									lowerCamelCase_         =     args[1:]

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

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



	def        SCREAMING_SNAKE_CASE_(       self		,    *lowercase		,    **lowercase					)							->			List[str]:
					return self.tokenizer.decode(*lowercase		,    **lowercase					)



	@contextmanager
	def        SCREAMING_SNAKE_CASE_(       self					)							->			Optional[int]:
					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."					)
					lowerCamelCase_         =     True
					lowerCamelCase_         =     self.tokenizer
					yield
					lowerCamelCase_         =     self.feature_extractor
					lowerCamelCase_         =     False


