





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


_snake_case           =			logging.get_logger(__name__)





class  lowerCAmelCase							(       lowercase_ ):
  __lowerCamelCase		=      ['input_features', 'attention_mask']







  def __init__(      self  :Union[str, Any]   ,							_lowercase  :List[str]=80   ,							_lowercase  :str=1_60_00   ,							_lowercase  :Tuple=80   ,							_lowercase  :int=0.0   ,							_lowercase  :Union[str, Any]=True   ,							_lowercase  :Optional[Any]=True   ,							_lowercase  :Optional[Any]=True   ,							**_lowercase  :Tuple   ,							):






     '''simple docstring'''





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







  def   UpperCAmelCase					(      self  :int   ,							_lowercase  :np.ndarray   ,							):






     '''simple docstring'''





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







  @staticmethod
  def   UpperCAmelCase					(      _lowercase  :np.ndarray   ,							_lowercase  :int   ,							_lowercase  :Optional[bool] = True   ,							_lowercase  :Optional[bool] = True   ,							_lowercase  :float = 0.0   ,							):






     '''simple docstring'''





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

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

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

     return x







  def   UpperCAmelCase					(      self  :Any   ,							_lowercase  :List[np.ndarray]   ,							_lowercase  :Optional[np.ndarray] = None ):






     '''simple docstring'''





     lowercase__			       =  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  :Union[str, Any]   ,							_lowercase  :Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]   ,							_lowercase  :Union[bool, str, PaddingStrategy] = False   ,							_lowercase  :Optional[int] = None   ,							_lowercase  :bool = False   ,							_lowercase  :Optional[int] = None   ,							_lowercase  :Optional[Union[str, TensorType]] = None   ,							_lowercase  :Optional[int] = None   ,							_lowercase  :Optional[bool] = None   ,							**_lowercase  :int   ,							):






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

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

     if is_batched:
        lowercase__			       =  [np.asarray(_lowercase   ,							dtype=np.floataa ) for speech in raw_speech]
     elif not is_batched and not isinstance(_lowercase   ,							np.ndarray ):
        lowercase__			       =  np.asarray(_lowercase   ,							dtype=np.floataa )
     elif isinstance(_lowercase   ,							np.ndarray ) and raw_speech.dtype is np.dtype(np.floataa ):
        lowercase__			       =  raw_speech.astype(np.floataa )

     # always return batch
     if not is_batched:
        lowercase__			       =  [raw_speech]

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

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

     lowercase__			       =  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
     lowercase__			       =  padded_inputs.get("input_features" )
     if isinstance(input_features[0]   ,							_lowercase ):
        lowercase__			       =  [np.asarray(_lowercase   ,							dtype=np.floataa ) for feature in input_features]

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

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

     if return_tensors is not None:
        lowercase__			       =  padded_inputs.convert_to_tensors(_lowercase )

     return padded_inputs













import os
import socket
from contextlib import contextmanager

import torch

from ..commands.config.default import write_basic_config  # noqa: F401
from ..state import PartialState
from .dataclasses import DistributedType
from .imports import is_deepspeed_available, is_tpu_available
from .transformer_engine import convert_model
from .versions import is_torch_version


if is_deepspeed_available():
  from deepspeed import DeepSpeedEngine

if is_tpu_available(check_device=False):
  import torch_xla.core.xla_model as xm


def _A       (   __magic_name__	):
   if is_torch_version("<"  , "2.0.0"	) or not hasattr(__magic_name__  , "_dynamo"	):
      return False
   return isinstance(__magic_name__  , torch._dynamo.eval_frame.OptimizedModule	)


def _A       (   __magic_name__  , __magic_name__ = True	):
   lowercase__			       =  (torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

   lowercase__			       =  is_compiled_module(__magic_name__	)
   if is_compiled:
      lowercase__			       =  model
      lowercase__			       =  model._orig_mod

   if is_deepspeed_available():
      options += (DeepSpeedEngine,)

   while isinstance(__magic_name__  , __magic_name__	):
      lowercase__			       =  model.module

   if not keep_fpaa_wrapper:
      lowercase__			       =  getattr(__magic_name__  , "forward"	)
      lowercase__			       =  model.__dict__.pop("_original_forward"  , __magic_name__	)
      if original_forward is not None:
         while hasattr(__magic_name__  , "__wrapped__"	):
            lowercase__			       =  forward.__wrapped__
            if forward == original_forward:
               break
         lowercase__			       =  forward
      if getattr(__magic_name__  , "_converted_to_transformer_engine"  , __magic_name__	):
         convert_model(__magic_name__  , to_transformer_engine=__magic_name__	)

   if is_compiled:
      lowercase__			       =  model
      lowercase__			       =  compiled_model

   return model


def _A       (   ):
   PartialState().wait_for_everyone()


def _A       (   __magic_name__  , __magic_name__	):
   if PartialState().distributed_type == DistributedType.TPU:
      xm.save(__magic_name__  , __magic_name__	)
   elif PartialState().local_process_index == 0:
      torch.save(__magic_name__  , __magic_name__	)


@contextmanager
def _A       (   **__magic_name__	):
   for key, value in kwargs.items():
      lowercase__			       =  str(__magic_name__	)

   yield

   for key in kwargs:
      if key.upper() in os.environ:
         del os.environ[key.upper()]


def _A       (   __magic_name__	):
   if not hasattr(__magic_name__  , "__qualname__"	) and not hasattr(__magic_name__  , "__name__"	):
      lowercase__			       =  getattr(__magic_name__  , "__class__"  , __magic_name__	)
   if hasattr(__magic_name__  , "__qualname__"	):
      return obj.__qualname__
   if hasattr(__magic_name__  , "__name__"	):
      return obj.__name__
   return str(__magic_name__	)


def _A       (   __magic_name__  , __magic_name__	):
   for key, value in source.items():
      if isinstance(__magic_name__  , __magic_name__	):
         lowercase__			       =  destination.setdefault(__magic_name__  , {}	)
         merge_dicts(__magic_name__  , __magic_name__	)
      else:
         lowercase__			       =  value

   return destination







def _A       (   __magic_name__ = None	):
   if port is None:
      lowercase__			       =  2_9500
   with socket.socket(socket.AF_INET  , socket.SOCK_STREAM	) as s:
      return s.connect_ex(("localhost", port)	) == 0






