
from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


__A							=logging.get_logger(__name__)




class        _SCREAMING_SNAKE_CASE	( snake_case_		):
	lowerCAmelCase__   =					['input_features', 'attention_mask']



	def __init__(       self		,    lowercase=80		,    lowercase=16000		,    lowercase=80		,    lowercase=0.0		,    lowercase=True		,    lowercase=True		,    lowercase=True		,    **lowercase		,    )							->			Any:
					super().__init__(feature_size=lowercase		,    sampling_rate=lowercase		,    padding_value=lowercase		,    **lowercase					)
					lowerCamelCase_         =     num_mel_bins
					lowerCamelCase_         =     do_ceptral_normalize
					lowerCamelCase_         =     normalize_means
					lowerCamelCase_         =     normalize_vars
					lowerCamelCase_         =     True



	def        SCREAMING_SNAKE_CASE_(       self		,    lowercase		,    )							->			np.ndarray:
					lowerCamelCase_         =     waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
					lowerCamelCase_         =     torch.from_numpy(lowercase					).unsqueeze(0					)
					lowerCamelCase_         =     ta_kaldi.fbank(lowercase		,    num_mel_bins=self.num_mel_bins		,    sample_frequency=self.sampling_rate					)
					return features.numpy()



	@staticmethod
	def        SCREAMING_SNAKE_CASE_(       lowercase		,    lowercase		,    lowercase = True		,    lowercase = True		,    lowercase = 0.0		,    )							->			np.ndarray:
					# make sure we normalize float32 arrays
					if normalize_means:
									lowerCamelCase_         =     x[:input_length].mean(axis=0					)
									lowerCamelCase_         =     np.subtract(lowercase		,    lowercase					)
					if normalize_vars:
									lowerCamelCase_         =     x[:input_length].std(axis=0					)
									lowerCamelCase_         =     np.divide(lowercase		,    lowercase					)

					if input_length < x.shape[0]:
									lowerCamelCase_         =     padding_value

					# make sure array is in float32
					lowerCamelCase_         =     x.astype(np.floataa					)

					return x



	def        SCREAMING_SNAKE_CASE_(       self		,    lowercase		,    lowercase = None					)							->			List[np.ndarray]:
					lowerCamelCase_         =     attention_mask.sum(-1					) if attention_mask is not None else [x.shape[0] for x in input_features]
					return [
					    self.utterance_cmvn(lowercase		,    lowercase		,    self.normalize_means		,    self.normalize_vars		,    self.padding_value					)
					    for x, n in zip(lowercase		,    lowercase					)
					]



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

					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 `raw_speech` 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."					)

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

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

					# extract fbank features
					lowerCamelCase_         =     [self._extract_fbank_features(lowercase					) for waveform in raw_speech]

					# convert into correct format for padding
					lowerCamelCase_         =     BatchFeature({"input_features": features}					)

					lowerCamelCase_         =     self.pad(
					    lowercase		,    padding=lowercase		,    max_length=lowercase		,    truncation=lowercase		,    pad_to_multiple_of=lowercase		,    return_attention_mask=lowercase		,    **lowercase		,    )

					# make sure list is in array format
					lowerCamelCase_         =     padded_inputs.get("input_features"					)
					if isinstance(input_features[0]		,    lowercase					):
									lowerCamelCase_         =     [np.asarray(lowercase		,    dtype=np.floataa					) for feature in input_features]

					lowerCamelCase_         =     padded_inputs.get("attention_mask"					)
					if attention_mask is not None:
									lowerCamelCase_         =     [np.asarray(lowercase		,    dtype=np.intaa					) for array in attention_mask]

					# Utterance-level cepstral mean and variance normalization
					if self.do_ceptral_normalize:
									lowerCamelCase_         =     (
									    np.array(lowercase		,    dtype=np.intaa					)
									    if self._get_padding_strategies(lowercase		,    max_length=lowercase					) is not PaddingStrategy.DO_NOT_PAD
									    else None
									)
									lowerCamelCase_         =     self.normalize(
									    padded_inputs["input_features"]		,    attention_mask=lowercase					)

					if return_tensors is not None:
									lowerCamelCase_         =     padded_inputs.convert_to_tensors(lowercase					)

					return padded_inputs




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


