


"""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 TensorType, logging


__lowercase          = logging.get_logger(__name__)






class 		_A    (    _a    ):







    """simple docstring"""




    UpperCAmelCase			:				List[str]      =			["""input_values""", """attention_mask"""]






    def __init__(   self							:       Union[str, Any]							,					__UpperCAmelCase							:       int=1							,					__UpperCAmelCase							:       Optional[Any]=16000							,					__UpperCAmelCase							:       Dict=128							,					__UpperCAmelCase							:       Optional[Any]=1024							,					__UpperCAmelCase							:       Tuple=0.0							,					__UpperCAmelCase							:       List[str]=True							,					__UpperCAmelCase							:       List[str]=-4.2_677_393							,					__UpperCAmelCase							:       str=4.5_689_974							,					__UpperCAmelCase							:       Tuple=False							,					**__UpperCAmelCase							:       int							,					):
      super().__init__(feature_size=__UpperCAmelCase							,					sampling_rate=__UpperCAmelCase							,					padding_value=__UpperCAmelCase							,					**__UpperCAmelCase)
      a		:						Union[str, Any] 				=     num_mel_bins
      a		:						Optional[Any] 				=     max_length
      a		:						Dict 				=     do_normalize
      a		:						int 				=     mean
      a		:						List[str] 				=     std
      a		:						Union[str, Any] 				=     return_attention_mask






    def       __snake_case				(   self							:       Optional[Any]							,					__UpperCAmelCase							:       np.ndarray							,					__UpperCAmelCase							:       int							,					):
      a		:						Dict 				=     torch.from_numpy(__UpperCAmelCase).unsqueeze(0)
      a		:						int 				=     ta_kaldi.fbank(
          __UpperCAmelCase							,					htk_compat=__UpperCAmelCase							,					sample_frequency=self.sampling_rate							,					use_energy=__UpperCAmelCase							,					window_type="hanning"							,					num_mel_bins=self.num_mel_bins							,					dither=0.0							,					frame_shift=10							,					)

      a		:						Optional[Any] 				=     fbank.shape[0]
      a		:						Optional[int] 				=     max_length - n_frames

      # pad or truncate, depending on difference
      if difference > 0:
        a		:						List[str] 				=     torch.nn.ZeroPadad((0, 0, 0, difference))
        a		:						Dict 				=     pad_module(__UpperCAmelCase)
      elif difference < 0:
        a		:						List[str] 				=     fbank[0:max_length, :]

      a		:						Optional[int] 				=     fbank.numpy()

      return fbank






    def       __snake_case				(   self							:       Optional[Any]							,					__UpperCAmelCase							:       np.ndarray):
      return (input_values - (self.mean)) / (self.std * 2)






    def __call__(   self							:       str							,					__UpperCAmelCase							:       Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]							,					__UpperCAmelCase							:       Optional[int] = None							,					__UpperCAmelCase							:       Optional[Union[str, TensorType]] = None							,					**__UpperCAmelCase							:       str							,					):

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

      a		:						int 				=     isinstance(__UpperCAmelCase							,					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}''')
      a		:						Optional[int] 				=     is_batched_numpy or (
          isinstance(__UpperCAmelCase							,					(list, tuple)) and (isinstance(raw_speech[0]							,					(np.ndarray, tuple, list)))
      )

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

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

      # extract fbank features and pad/truncate to max_length
      a		:						Dict 				=     [self._extract_fbank_features(__UpperCAmelCase							,					max_length=self.max_length) for waveform in raw_speech]

      # convert into BatchFeature
      a		:						Tuple 				=     BatchFeature({"input_values": features})

      # make sure list is in array format
      a		:						List[Any] 				=     padded_inputs.get("input_values")
      if isinstance(input_values[0]							,					__UpperCAmelCase):
        a		:						Tuple 				=     [np.asarray(__UpperCAmelCase							,					dtype=np.floataa) for feature in input_values]

      # normalization
      if self.do_normalize:
        a		:						List[str] 				=     [self.normalize(__UpperCAmelCase) for feature in input_values]

      if return_tensors is not None:
        a		:						Dict 				=     padded_inputs.convert_to_tensors(__UpperCAmelCase)

      return padded_inputs










"""simple docstring"""







from ..utils import DummyObject, requires_backends






class 		_A    (    metaclass=_a    ):







    """simple docstring"""




    UpperCAmelCase			:				int      =			["""torch""", """transformers""", """invisible_watermark"""]






    def __init__(   self							:       int							,					*__UpperCAmelCase							:       List[Any]							,					**__UpperCAmelCase							:       Optional[int]):
      requires_backends(self							,					["torch", "transformers", "invisible_watermark"])






    @classmethod
    def       __snake_case				(   cls							:       Optional[Any]							,					*__UpperCAmelCase							:       Any							,					**__UpperCAmelCase							:       int):
      requires_backends(cls							,					["torch", "transformers", "invisible_watermark"])






    @classmethod
    def       __snake_case				(   cls							:       List[Any]							,					*__UpperCAmelCase							:       Optional[Any]							,					**__UpperCAmelCase							:       Tuple):
      requires_backends(cls							,					["torch", "transformers", "invisible_watermark"])






class 		_A    (    metaclass=_a    ):







    """simple docstring"""




    UpperCAmelCase			:				Optional[int]      =			["""torch""", """transformers""", """invisible_watermark"""]






    def __init__(   self							:       int							,					*__UpperCAmelCase							:       Tuple							,					**__UpperCAmelCase							:       Tuple):
      requires_backends(self							,					["torch", "transformers", "invisible_watermark"])






    @classmethod
    def       __snake_case				(   cls							:       int							,					*__UpperCAmelCase							:       str							,					**__UpperCAmelCase							:       Optional[int]):
      requires_backends(cls							,					["torch", "transformers", "invisible_watermark"])






    @classmethod
    def       __snake_case				(   cls							:       Any							,					*__UpperCAmelCase							:       Optional[int]							,					**__UpperCAmelCase							:       Union[str, Any]):
      requires_backends(cls							,					["torch", "transformers", "invisible_watermark"])






