'''simple docstring'''






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     A__							(       UpperCamelCase   ):





   """simple docstring"""





   UpperCamelCase_       :						Any    				=     ['''input_features''', '''attention_mask''']




   def __init__( self     :		Union[str, Any]						,     lowerCAmelCase__     :		Dict=8_0						,     lowerCAmelCase__     :		List[str]=1_6_0_0_0						,     lowerCAmelCase__     :		Optional[int]=8_0						,     lowerCAmelCase__     :		Optional[int]=0.0						,     lowerCAmelCase__     :		Dict=True						,     lowerCAmelCase__     :		Dict=True						,     lowerCAmelCase__     :		Union[str, Any]=True						,     **lowerCAmelCase__     :		int						,     )       ->    Dict:







      """simple docstring"""




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




   def 			_lowerCAmelCase		( self     :		List[str]						,     lowerCAmelCase__     :		str						,     )       ->    np.ndarray:







      """simple docstring"""




      _UpperCAmelCase							:	List[Any]							=   waveform * (2**1_5)  # Kaldi compliance: 16-bit signed integers
      _UpperCAmelCase							:	List[str]							=   torch.from_numpy(lowerCAmelCase__			).unsqueeze(0			)
      _UpperCAmelCase							:	Any							=   ta_kaldi.fbank(lowerCAmelCase__						,     num_mel_bins=self.num_mel_bins						,     sample_frequency=self.sampling_rate			)
      return features.numpy()




   @staticmethod
   def 			_lowerCAmelCase		( lowerCAmelCase__     :		int						,     lowerCAmelCase__     :		Tuple						,     lowerCAmelCase__     :		Any = True						,     lowerCAmelCase__     :		Any = True						,     lowerCAmelCase__     :		Any = 0.0						,     )       ->    np.ndarray:







      """simple docstring"""




      if normalize_means:
         _UpperCAmelCase							:	Union[str, Any]							=   x[:input_length].mean(axis=0			)
         _UpperCAmelCase							:	Optional[int]							=   np.subtract(lowerCAmelCase__						,     lowerCAmelCase__			)
      if normalize_vars:
         _UpperCAmelCase							:	Dict							=   x[:input_length].std(axis=0			)
         _UpperCAmelCase							:	int							=   np.divide(lowerCAmelCase__						,     lowerCAmelCase__			)

      if input_length < x.shape[0]:
         _UpperCAmelCase							:	Optional[int]							=   padding_value

      # make sure array is in float32
      _UpperCAmelCase							:	Any							=   x.astype(np.floataa			)

      return x




   def 			_lowerCAmelCase		( self     :		Optional[Any]						,     lowerCAmelCase__     :		str						,     lowerCAmelCase__     :		List[Any] = None			)       ->    List[np.ndarray]:







      """simple docstring"""




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




   def __call__( self     :		Union[str, Any]						,     lowerCAmelCase__     :		Dict						,     lowerCAmelCase__     :		Union[str, Any] = False						,     lowerCAmelCase__     :		Dict = None						,     lowerCAmelCase__     :		Union[str, Any] = False						,     lowerCAmelCase__     :		Tuple = None						,     lowerCAmelCase__     :		Union[str, Any] = None						,     lowerCAmelCase__     :		int = None						,     lowerCAmelCase__     :		Tuple = None						,     **lowerCAmelCase__     :		int						,     )       ->    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							:	str							=   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}"""			)
      _UpperCAmelCase							:	str							=   is_batched_numpy or (
          isinstance(lowerCAmelCase__						,     (list, tuple)			) and (isinstance(raw_speech[0]						,     (np.ndarray, tuple, list)			))
      )

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

      # always return batch
      if not is_batched:
         _UpperCAmelCase							:	Optional[int]							=   [raw_speech]

      # extract fbank features
      _UpperCAmelCase							:	str							=   [self._extract_fbank_features(lowerCAmelCase__			) for waveform in raw_speech]

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

      _UpperCAmelCase							:	List[Any]							=   self.pad(
          lowerCAmelCase__						,     padding=lowerCAmelCase__						,     max_length=lowerCAmelCase__						,     truncation=lowerCAmelCase__						,     pad_to_multiple_of=lowerCAmelCase__						,     return_attention_mask=lowerCAmelCase__						,     **lowerCAmelCase__						,     )

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

      _UpperCAmelCase							:	Tuple							=   padded_inputs.get("attention_mask"			)
      if attention_mask is not None:
         _UpperCAmelCase							:	Optional[int]							=   [np.asarray(lowerCAmelCase__						,     dtype=np.intaa			) for array in attention_mask]

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

      if return_tensors is not None:
         _UpperCAmelCase							:	Optional[Any]							=   padded_inputs.convert_to_tensors(lowerCAmelCase__			)

      return padded_inputs



import functools
import operator

from ...configuration_utils import PretrainedConfig
from ...utils import logging


UpperCAmelCase__   	=							logging.get_logger(__name__)

UpperCAmelCase__   	=							{
    '''microsoft/wavlm-base''': '''https://huggingface.co/microsoft/wavlm-base/resolve/main/config.json''',
    # See all WavLM models at https://huggingface.co/models?filter=wavlm
}







class       lowerCamelCase__   (       lowerCAmelCase):
     SCREAMING_SNAKE_CASE__      			=			'''wavlm'''





     def __init__(self		,     UpperCAmelCase=3_2		,     UpperCAmelCase=7_6_8		,     UpperCAmelCase=1_2		,     UpperCAmelCase=1_2		,     UpperCAmelCase=3_0_7_2		,     UpperCAmelCase="gelu"		,     UpperCAmelCase=0.1		,     UpperCAmelCase=0.1		,     UpperCAmelCase=0.1		,     UpperCAmelCase=0.0		,     UpperCAmelCase=0.1		,     UpperCAmelCase=0.1		,     UpperCAmelCase=0.02		,     UpperCAmelCase=1e-5		,     UpperCAmelCase="group"		,     UpperCAmelCase="gelu"		,     UpperCAmelCase=(5_1_2, 5_1_2, 5_1_2, 5_1_2, 5_1_2, 5_1_2, 5_1_2)		,     UpperCAmelCase=(5, 2, 2, 2, 2, 2, 2)		,     UpperCAmelCase=(1_0, 3, 3, 3, 3, 2, 2)		,     UpperCAmelCase=False		,     UpperCAmelCase=1_2_8		,     UpperCAmelCase=1_6		,     UpperCAmelCase=3_2_0		,     UpperCAmelCase=8_0_0		,     UpperCAmelCase=False		,     UpperCAmelCase=True		,     UpperCAmelCase=0.05		,     UpperCAmelCase=1_0		,     UpperCAmelCase=2		,     UpperCAmelCase=0.0		,     UpperCAmelCase=1_0		,     UpperCAmelCase=3_2_0		,     UpperCAmelCase=2		,     UpperCAmelCase=0.1		,     UpperCAmelCase=1_0_0		,     UpperCAmelCase=2_5_6		,     UpperCAmelCase=2_5_6		,     UpperCAmelCase=0.1		,     UpperCAmelCase="mean"		,     UpperCAmelCase=False		,     UpperCAmelCase=False		,     UpperCAmelCase=2_5_6		,     UpperCAmelCase=(5_1_2, 5_1_2, 5_1_2, 5_1_2, 1_5_0_0)		,     UpperCAmelCase=(5, 3, 3, 1, 1)		,     UpperCAmelCase=(1, 2, 3, 1, 1)		,     UpperCAmelCase=5_1_2		,     UpperCAmelCase=8_0		,     UpperCAmelCase=0		,     UpperCAmelCase=1		,     UpperCAmelCase=2		,     UpperCAmelCase=False		,     UpperCAmelCase=3		,     UpperCAmelCase=2		,     UpperCAmelCase=3		,     UpperCAmelCase=None		,     **UpperCAmelCase		,     )     ->   Optional[Any]:
      super().__init__(**UpperCAmelCase		,     pad_token_id=UpperCAmelCase		,     bos_token_id=UpperCAmelCase		,     eos_token_id=UpperCAmelCase       )
      _lowercase							      =hidden_size
      _lowercase							      =feat_extract_norm
      _lowercase							      =feat_extract_activation
      _lowercase							      =list(UpperCAmelCase       )
      _lowercase							      =list(UpperCAmelCase       )
      _lowercase							      =list(UpperCAmelCase       )
      _lowercase							      =conv_bias
      _lowercase							      =num_buckets
      _lowercase							      =max_bucket_distance
      _lowercase							      =num_conv_pos_embeddings
      _lowercase							      =num_conv_pos_embedding_groups
      _lowercase							      =len(self.conv_dim       )
      _lowercase							      =num_hidden_layers
      _lowercase							      =intermediate_size
      _lowercase							      =hidden_act
      _lowercase							      =num_attention_heads
      _lowercase							      =hidden_dropout
      _lowercase							      =attention_dropout
      _lowercase							      =activation_dropout
      _lowercase							      =feat_proj_dropout
      _lowercase							      =final_dropout
      _lowercase							      =layerdrop
      _lowercase							      =layer_norm_eps
      _lowercase							      =initializer_range
      _lowercase							      =num_ctc_classes
      _lowercase							      =vocab_size
      _lowercase							      =do_stable_layer_norm
      _lowercase							      =use_weighted_layer_sum
      _lowercase							      =classifier_proj_size

      if (
          (len(self.conv_stride       ) != self.num_feat_extract_layers)
          or (len(self.conv_kernel       ) != self.num_feat_extract_layers)
          or (len(self.conv_dim       ) != self.num_feat_extract_layers)
      ):
       raise ValueError(
           '''Configuration for convolutional layers is incorrect. It is required that `len(config.conv_dim)` =='''
           ''' `len(config.conv_stride)` == `len(config.conv_kernel)`, but is `len(config.conv_dim) ='''
           f" {len(self.conv_dim       )}`, `len(config.conv_stride) = {len(self.conv_stride       )}`,"
           f" `len(config.conv_kernel) = {len(self.conv_kernel       )}`."       )

      # fine-tuning config parameters for SpecAugment: https://arxiv.org/abs/1904.08779
      _lowercase							      =apply_spec_augment
      _lowercase							      =mask_time_prob
      _lowercase							      =mask_time_length
      _lowercase							      =mask_time_min_masks
      _lowercase							      =mask_feature_prob
      _lowercase							      =mask_feature_length

      # parameters for pretraining with codevector quantized representations
      _lowercase							      =num_codevectors_per_group
      _lowercase							      =num_codevector_groups
      _lowercase							      =contrastive_logits_temperature
      _lowercase							      =num_negatives
      _lowercase							      =codevector_dim
      _lowercase							      =proj_codevector_dim
      _lowercase							      =diversity_loss_weight

      # ctc loss
      _lowercase							      =ctc_loss_reduction
      _lowercase							      =ctc_zero_infinity

      # adapter
      _lowercase							      =add_adapter
      _lowercase							      =adapter_kernel_size
      _lowercase							      =adapter_stride
      _lowercase							      =num_adapter_layers
      _lowercase							      =output_hidden_size or hidden_size

      # SequenceClassification-specific parameter. Feel free to ignore for other classes.
      _lowercase							      =classifier_proj_size

      # XVector-specific parameters. Feel free to ignore for other classes.
      _lowercase							      =list(UpperCAmelCase       )
      _lowercase							      =list(UpperCAmelCase       )
      _lowercase							      =list(UpperCAmelCase       )
      _lowercase							      =xvector_output_dim





     @property
     def      __A					(self       )     ->   int:
      return functools.reduce(operator.mul		,     self.conv_stride		,     1       )

