





"""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


__UpperCamelCase     :						Any						       =  logging.get_logger(__name__)







class       a							(   a__  ):
 snake_case__             =       ['''input_features''', '''attention_mask''']



 def __init__(     self				,       _snake_case=80				,       _snake_case=1_60_00				,       _snake_case=80				,       _snake_case=0.0				,       _snake_case=True				,       _snake_case=True				,       _snake_case=True				,       **_snake_case				,       ):







  """simple docstring"""






  super().__init__(feature_size=_snake_case				,       sampling_rate=_snake_case				,       padding_value=_snake_case				,       **_snake_case     )
  lowerCAmelCase    				= num_mel_bins
  lowerCAmelCase    				= do_ceptral_normalize
  lowerCAmelCase    				= normalize_means
  lowerCAmelCase    				= normalize_vars
  lowerCAmelCase    				= True



 def    UpperCamelCase__  (     self				,       _snake_case				,       ):







  """simple docstring"""






  lowerCAmelCase    				= waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
  lowerCAmelCase    				= torch.from_numpy(_snake_case     ).unsqueeze(0     )
  lowerCAmelCase    				= ta_kaldi.fbank(_snake_case				,       num_mel_bins=self.num_mel_bins				,       sample_frequency=self.sampling_rate     )
  return features.numpy()



 @staticmethod
 def    UpperCamelCase__  (     _snake_case				,       _snake_case				,       _snake_case = True				,       _snake_case = True				,       _snake_case = 0.0				,       ):







  """simple docstring"""






  if normalize_means:
   lowerCAmelCase    				= x[:input_length].mean(axis=0     )
   lowerCAmelCase    				= np.subtract(_snake_case				,       _snake_case     )
  if normalize_vars:
   lowerCAmelCase    				= x[:input_length].std(axis=0     )
   lowerCAmelCase    				= np.divide(_snake_case				,       _snake_case     )

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

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

  return x



 def    UpperCamelCase__  (     self				,       _snake_case				,       _snake_case = None     ):







  """simple docstring"""






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



 def __call__(     self				,       _snake_case				,       _snake_case = False				,       _snake_case = None				,       _snake_case = False				,       _snake_case = None				,       _snake_case = None				,       _snake_case = None				,       _snake_case = None				,       **_snake_case				,       ):







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

  lowerCAmelCase    				= isinstance(_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}'     )
  lowerCAmelCase    				= is_batched_numpy or (
      isinstance(_snake_case				,       (list, tuple)     ) and (isinstance(raw_speech[0]				,       (np.ndarray, tuple, list)     ))
  )

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

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

  lowerCAmelCase    				= self.pad(
      _snake_case				,       padding=_snake_case				,       max_length=_snake_case				,       truncation=_snake_case				,       pad_to_multiple_of=_snake_case				,       return_attention_mask=_snake_case				,       **_snake_case				,       )

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

  lowerCAmelCase    				= padded_inputs.get('attention_mask'     )
  if attention_mask is not None:
   lowerCAmelCase    				= [np.asarray(_snake_case				,       dtype=np.intaa     ) for array in attention_mask]

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

  if return_tensors is not None:
   lowerCAmelCase    				= padded_inputs.convert_to_tensors(_snake_case     )

  return padded_inputs












"""simple docstring"""
from dataclasses import dataclass
from typing import Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, apply_forward_hook
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, Encoder, VectorQuantizer







@dataclass
class       a							(   a__  ):
 snake_case__             =       42







class       a							(   a__   , a__  ):



 @register_to_config
 def __init__(     self				,       _snake_case = 3				,       _snake_case = 3				,       _snake_case = ("DownEncoderBlock2D",)				,       _snake_case = ("UpDecoderBlock2D",)				,       _snake_case = (64,)				,       _snake_case = 1				,       _snake_case = "silu"				,       _snake_case = 3				,       _snake_case = 32				,       _snake_case = 2_56				,       _snake_case = 32				,       _snake_case = None				,       _snake_case = 0.18_215				,       _snake_case = "group"				,       ):







  """simple docstring"""






  super().__init__()

  # pass init params to Encoder
  lowerCAmelCase    				= Encoder(
      in_channels=_snake_case				,       out_channels=_snake_case				,       down_block_types=_snake_case				,       block_out_channels=_snake_case				,       layers_per_block=_snake_case				,       act_fn=_snake_case				,       norm_num_groups=_snake_case				,       double_z=_snake_case				,       )

  lowerCAmelCase    				= vq_embed_dim if vq_embed_dim is not None else latent_channels

  lowerCAmelCase    				= nn.Convad(_snake_case				,       _snake_case				,       1     )
  lowerCAmelCase    				= VectorQuantizer(_snake_case				,       _snake_case				,       beta=0.25				,       remap=_snake_case				,       sane_index_shape=_snake_case     )
  lowerCAmelCase    				= nn.Convad(_snake_case				,       _snake_case				,       1     )

  # pass init params to Decoder
  lowerCAmelCase    				= Decoder(
      in_channels=_snake_case				,       out_channels=_snake_case				,       up_block_types=_snake_case				,       block_out_channels=_snake_case				,       layers_per_block=_snake_case				,       act_fn=_snake_case				,       norm_num_groups=_snake_case				,       norm_type=_snake_case				,       )



 @apply_forward_hook
 def    UpperCamelCase__  (     self				,       _snake_case				,       _snake_case = True     ):







  """simple docstring"""






  lowerCAmelCase    				= self.encoder(_snake_case     )
  lowerCAmelCase    				= self.quant_conv(_snake_case     )

  if not return_dict:
   return (h,)

  return VQEncoderOutput(latents=_snake_case     )



 @apply_forward_hook
 def    UpperCamelCase__  (     self				,       _snake_case				,       _snake_case = False				,       _snake_case = True     ):







  """simple docstring"""






  if not force_not_quantize:
   lowerCAmelCase				,lowerCAmelCase				,lowerCAmelCase    				= self.quantize(_snake_case     )
  else:
   lowerCAmelCase    				= h
  lowerCAmelCase    				= self.post_quant_conv(_snake_case     )
  lowerCAmelCase    				= self.decoder(_snake_case				,       quant if self.config.norm_type == 'spatial' else None     )

  if not return_dict:
   return (dec,)

  return DecoderOutput(sample=_snake_case     )



 def    UpperCamelCase__  (     self				,       _snake_case				,       _snake_case = True     ):







  """simple docstring"""






  lowerCAmelCase    				= sample
  lowerCAmelCase    				= self.encode(_snake_case     ).latents
  lowerCAmelCase    				= self.decode(_snake_case     ).sample

  if not return_dict:
   return (dec,)

  return DecoderOutput(sample=_snake_case     )





