
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


lowerCamelCase      =logging.get_logger(__name__)




class 				_lowerCamelCase (      UpperCamelCase_ ):





     """simple docstring"""







     SCREAMING_SNAKE_CASE_		   =  ['''input_features''', '''attention_mask''']






     def __init__(							self						,						__SCREAMING_SNAKE_CASE=8_0						,						__SCREAMING_SNAKE_CASE=1_6_0_0_0						,						__SCREAMING_SNAKE_CASE=8_0						,						__SCREAMING_SNAKE_CASE=0.0						,						__SCREAMING_SNAKE_CASE=True						,						__SCREAMING_SNAKE_CASE=True						,						__SCREAMING_SNAKE_CASE=True						,						**__SCREAMING_SNAKE_CASE						,						)  ->   Tuple:




           """simple docstring"""







           super().__init__(feature_size=__SCREAMING_SNAKE_CASE						,						sampling_rate=__SCREAMING_SNAKE_CASE						,						padding_value=__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )
           UpperCamelCase__    :   Optional[Any]					=				num_mel_bins
           UpperCamelCase__    :   str					=				do_ceptral_normalize
           UpperCamelCase__    :   Optional[Any]					=				normalize_means
           UpperCamelCase__    :   Union[str, Any]					=				normalize_vars
           UpperCamelCase__    :   Optional[int]					=				True






     def    __SCREAMING_SNAKE_CASE			(							self						,						__SCREAMING_SNAKE_CASE						,						)  ->   np.ndarray:




           """simple docstring"""







           UpperCamelCase__    :   Dict					=				waveform * (2**1_5)  # Kaldi compliance: 16-bit signed integers
           UpperCamelCase__    :   Tuple					=				torch.from_numpy(__SCREAMING_SNAKE_CASE  ).unsqueeze(0  )
           UpperCamelCase__    :   Any					=				ta_kaldi.fbank(__SCREAMING_SNAKE_CASE						,						num_mel_bins=self.num_mel_bins						,						sample_frequency=self.sampling_rate  )
           return features.numpy()






     @staticmethod
     def    __SCREAMING_SNAKE_CASE			(							__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE = True						,						__SCREAMING_SNAKE_CASE = True						,						__SCREAMING_SNAKE_CASE = 0.0						,						)  ->   np.ndarray:




           """simple docstring"""







           if normalize_means:
                 UpperCamelCase__    :   List[str]					=				x[:input_length].mean(axis=0  )
                 UpperCamelCase__    :   Any					=				np.subtract(__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE  )
           if normalize_vars:
                 UpperCamelCase__    :   List[Any]					=				x[:input_length].std(axis=0  )
                 UpperCamelCase__    :   List[Any]					=				np.divide(__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE  )

           if input_length < x.shape[0]:
                 UpperCamelCase__    :   List[Any]					=				padding_value

           # make sure array is in float32
           UpperCamelCase__    :   Optional[int]					=				x.astype(np.floataa  )

           return x






     def    __SCREAMING_SNAKE_CASE			(							self						,						__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE = None  )  ->   List[np.ndarray]:




           """simple docstring"""







           UpperCamelCase__    :   Dict					=				attention_mask.sum(-1  ) if attention_mask is not None else [x.shape[0] for x in input_features]
           return [
               self.utterance_cmvn(__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE						,						self.normalize_means						,						self.normalize_vars						,						self.padding_value  )
               for x, n in zip(__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE  )
           ]






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




           """simple docstring"""







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

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

           if is_batched:
                 UpperCamelCase__    :   str					=				[np.asarray(__SCREAMING_SNAKE_CASE						,						dtype=np.floataa  ) for speech in raw_speech]
           elif not is_batched and not isinstance(__SCREAMING_SNAKE_CASE						,						np.ndarray  ):
                 UpperCamelCase__    :   Any					=				np.asarray(__SCREAMING_SNAKE_CASE						,						dtype=np.floataa  )
           elif isinstance(__SCREAMING_SNAKE_CASE						,						np.ndarray  ) and raw_speech.dtype is np.dtype(np.floataa  ):
                 UpperCamelCase__    :   Union[str, Any]					=				raw_speech.astype(np.floataa  )

           # always return batch
           if not is_batched:
                 UpperCamelCase__    :   List[str]					=				[raw_speech]

           # extract fbank features
           UpperCamelCase__    :   List[Any]					=				[self._extract_fbank_features(__SCREAMING_SNAKE_CASE  ) for waveform in raw_speech]

           # convert into correct format for padding
           UpperCamelCase__    :   List[str]					=				BatchFeature({'''input_features''': features}  )

           UpperCamelCase__    :   Optional[int]					=				self.pad(
               __SCREAMING_SNAKE_CASE						,						padding=__SCREAMING_SNAKE_CASE						,						max_length=__SCREAMING_SNAKE_CASE						,						truncation=__SCREAMING_SNAKE_CASE						,						pad_to_multiple_of=__SCREAMING_SNAKE_CASE						,						return_attention_mask=__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE						,						)

           # make sure list is in array format
           UpperCamelCase__    :   Tuple					=				padded_inputs.get('''input_features'''  )
           if isinstance(input_features[0]						,						__SCREAMING_SNAKE_CASE  ):
                 UpperCamelCase__    :   Any					=				[np.asarray(__SCREAMING_SNAKE_CASE						,						dtype=np.floataa  ) for feature in input_features]

           UpperCamelCase__    :   List[Any]					=				padded_inputs.get('''attention_mask'''  )
           if attention_mask is not None:
                 UpperCamelCase__    :   List[Any]					=				[np.asarray(__SCREAMING_SNAKE_CASE						,						dtype=np.intaa  ) for array in attention_mask]

           # Utterance-level cepstral mean and variance normalization
           if self.do_ceptral_normalize:
                 UpperCamelCase__    :   Union[str, Any]					=				(
                     np.array(__SCREAMING_SNAKE_CASE						,						dtype=np.intaa  )
                     if self._get_padding_strategies(__SCREAMING_SNAKE_CASE						,						max_length=__SCREAMING_SNAKE_CASE  ) is not PaddingStrategy.DO_NOT_PAD
                     else None
                 )
                 UpperCamelCase__    :   Dict					=				self.normalize(
                     padded_inputs['''input_features''']						,						attention_mask=__SCREAMING_SNAKE_CASE  )

           if return_tensors is not None:
                 UpperCamelCase__    :   int					=				padded_inputs.convert_to_tensors(__SCREAMING_SNAKE_CASE  )

           return padded_inputs




from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
)


lowerCamelCase      ={
    "configuration_mobilebert": [
        "MOBILEBERT_PRETRAINED_CONFIG_ARCHIVE_MAP",
        "MobileBertConfig",
        "MobileBertOnnxConfig",
    ],
    "tokenization_mobilebert": ["MobileBertTokenizer"],
}

try:
    if not is_tokenizers_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    lowerCamelCase      =["MobileBertTokenizerFast"]

try:
    if not is_torch_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    lowerCamelCase      =[
        "MOBILEBERT_PRETRAINED_MODEL_ARCHIVE_LIST",
        "MobileBertForMaskedLM",
        "MobileBertForMultipleChoice",
        "MobileBertForNextSentencePrediction",
        "MobileBertForPreTraining",
        "MobileBertForQuestionAnswering",
        "MobileBertForSequenceClassification",
        "MobileBertForTokenClassification",
        "MobileBertLayer",
        "MobileBertModel",
        "MobileBertPreTrainedModel",
        "load_tf_weights_in_mobilebert",
    ]

try:
    if not is_tf_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    lowerCamelCase      =[
        "TF_MOBILEBERT_PRETRAINED_MODEL_ARCHIVE_LIST",
        "TFMobileBertForMaskedLM",
        "TFMobileBertForMultipleChoice",
        "TFMobileBertForNextSentencePrediction",
        "TFMobileBertForPreTraining",
        "TFMobileBertForQuestionAnswering",
        "TFMobileBertForSequenceClassification",
        "TFMobileBertForTokenClassification",
        "TFMobileBertMainLayer",
        "TFMobileBertModel",
        "TFMobileBertPreTrainedModel",
    ]


if TYPE_CHECKING:
    from .configuration_mobilebert import (
        MOBILEBERT_PRETRAINED_CONFIG_ARCHIVE_MAP,
        MobileBertConfig,
        MobileBertOnnxConfig,
    )
    from .tokenization_mobilebert import MobileBertTokenizer

    try:
        if not is_tokenizers_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .tokenization_mobilebert_fast import MobileBertTokenizerFast

    try:
        if not is_torch_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_mobilebert import (
            MOBILEBERT_PRETRAINED_MODEL_ARCHIVE_LIST,
            MobileBertForMaskedLM,
            MobileBertForMultipleChoice,
            MobileBertForNextSentencePrediction,
            MobileBertForPreTraining,
            MobileBertForQuestionAnswering,
            MobileBertForSequenceClassification,
            MobileBertForTokenClassification,
            MobileBertLayer,
            MobileBertModel,
            MobileBertPreTrainedModel,
            load_tf_weights_in_mobilebert,
        )

    try:
        if not is_tf_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_tf_mobilebert import (
            TF_MOBILEBERT_PRETRAINED_MODEL_ARCHIVE_LIST,
            TFMobileBertForMaskedLM,
            TFMobileBertForMultipleChoice,
            TFMobileBertForNextSentencePrediction,
            TFMobileBertForPreTraining,
            TFMobileBertForQuestionAnswering,
            TFMobileBertForSequenceClassification,
            TFMobileBertForTokenClassification,
            TFMobileBertMainLayer,
            TFMobileBertModel,
            TFMobileBertPreTrainedModel,
        )

else:
    import sys

    lowerCamelCase      =_LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)


