

'''simple docstring'''



import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np
import torch

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


UpperCAmelCase :       Union[str, Any]			   =     logging.get_logger(__name__)







class 						lowerCAmelCase__   (      a					):



 """simple docstring"""



 lowerCAmelCase__  					=      ["input_features", "is_longer"]
 def __init__(		self					:       List[Any]					,							__SCREAMING_SNAKE_CASE					:       List[Any]=64					,							__SCREAMING_SNAKE_CASE					:       Optional[Any]=48_000					,							__SCREAMING_SNAKE_CASE					:       Union[str, Any]=480					,							__SCREAMING_SNAKE_CASE					:       str=10					,							__SCREAMING_SNAKE_CASE					:       Dict=1_024					,							__SCREAMING_SNAKE_CASE					:       Tuple=0.0					,							__SCREAMING_SNAKE_CASE					:       Dict=False					,							__SCREAMING_SNAKE_CASE					:       float = 0					,							__SCREAMING_SNAKE_CASE					:       float = 14_000					,							__SCREAMING_SNAKE_CASE					:       int = None					,							__SCREAMING_SNAKE_CASE					:       str = "fusion"					,							__SCREAMING_SNAKE_CASE					:       str = "repeatpad"					,							**__SCREAMING_SNAKE_CASE					:       Any					,							)  -> Union[str, Any]:


      """simple docstring"""




      super().__init__(
          feature_size=__SCREAMING_SNAKE_CASE					,							sampling_rate=__SCREAMING_SNAKE_CASE					,							padding_value=__SCREAMING_SNAKE_CASE					,							return_attention_mask=__SCREAMING_SNAKE_CASE					,							**__SCREAMING_SNAKE_CASE					,							)
      __SCREAMING_SNAKE_CASE     =      top_db
      __SCREAMING_SNAKE_CASE     =      truncation
      __SCREAMING_SNAKE_CASE     =      padding
      __SCREAMING_SNAKE_CASE     =      fft_window_size
      __SCREAMING_SNAKE_CASE     =      (fft_window_size >> 1) + 1
      __SCREAMING_SNAKE_CASE     =      hop_length
      __SCREAMING_SNAKE_CASE     =      max_length_s
      __SCREAMING_SNAKE_CASE     =      max_length_s * sampling_rate
      __SCREAMING_SNAKE_CASE     =      sampling_rate
      __SCREAMING_SNAKE_CASE     =      frequency_min
      __SCREAMING_SNAKE_CASE     =      frequency_max
      __SCREAMING_SNAKE_CASE     =      mel_filter_bank(
          num_frequency_bins=self.nb_frequency_bins					,							num_mel_filters=__SCREAMING_SNAKE_CASE					,							min_frequency=__SCREAMING_SNAKE_CASE					,							max_frequency=__SCREAMING_SNAKE_CASE					,							sampling_rate=__SCREAMING_SNAKE_CASE					,							norm=__SCREAMING_SNAKE_CASE					,							mel_scale="""htk"""					,							)
      __SCREAMING_SNAKE_CASE     =      mel_filter_bank(
          num_frequency_bins=self.nb_frequency_bins					,							num_mel_filters=__SCREAMING_SNAKE_CASE					,							min_frequency=__SCREAMING_SNAKE_CASE					,							max_frequency=__SCREAMING_SNAKE_CASE					,							sampling_rate=__SCREAMING_SNAKE_CASE					,							norm="""slaney"""					,							mel_scale="""slaney"""					,							)
 def       UpperCAmelCase__		(		self					:       Dict			)  -> Dict[str, Any]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      copy.deepcopy(self.__dict__			)
      __SCREAMING_SNAKE_CASE     =      self.__class__.__name__
      if "mel_filters" in output:
           del output["mel_filters"]
      if "mel_filters_slaney" in output:
           del output["mel_filters_slaney"]
      return output
 def       UpperCAmelCase__		(		self					:       Tuple					,							__SCREAMING_SNAKE_CASE					:       np.array					,							__SCREAMING_SNAKE_CASE					:       Optional[np.array] = None			)  -> np.ndarray:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      spectrogram(
          __SCREAMING_SNAKE_CASE					,							window_function(self.fft_window_size					,							"""hann"""			)					,							frame_length=self.fft_window_size					,							hop_length=self.hop_length					,							power=2.0					,							mel_filters=__SCREAMING_SNAKE_CASE					,							log_mel="""dB"""					,							)
      return log_mel_spectrogram.T
 def       UpperCAmelCase__		(		self					:       Union[str, Any]					,							__SCREAMING_SNAKE_CASE					:       int					,							__SCREAMING_SNAKE_CASE					:       Dict					,							__SCREAMING_SNAKE_CASE					:       Optional[Any]			)  -> int:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      np.array_split(list(range(0					,							total_frames - chunk_frames + 1			)			)					,							3			)
      if len(ranges[1]			) == 0:
           # if the audio is too short, we just use the first chunk
           __SCREAMING_SNAKE_CASE     =      [0]
      if len(ranges[2]			) == 0:
           # if the audio is too short, we just use the first chunk
           __SCREAMING_SNAKE_CASE     =      [0]
      # randomly choose index for each part
      __SCREAMING_SNAKE_CASE     =      np.random.choice(ranges[0]			)
      __SCREAMING_SNAKE_CASE     =      np.random.choice(ranges[1]			)
      __SCREAMING_SNAKE_CASE     =      np.random.choice(ranges[2]			)

      __SCREAMING_SNAKE_CASE     =      mel[idx_front : idx_front + chunk_frames, :]
      __SCREAMING_SNAKE_CASE     =      mel[idx_middle : idx_middle + chunk_frames, :]
      __SCREAMING_SNAKE_CASE     =      mel[idx_back : idx_back + chunk_frames, :]

      __SCREAMING_SNAKE_CASE     =      torch.tensor(mel[None, None, :]			)
      __SCREAMING_SNAKE_CASE     =      torch.nn.functional.interpolate(
          __SCREAMING_SNAKE_CASE					,							size=[chunk_frames, 64]					,							mode="""bilinear"""					,							align_corners=__SCREAMING_SNAKE_CASE			)
      __SCREAMING_SNAKE_CASE     =      mel_shrink[0][0].numpy()
      __SCREAMING_SNAKE_CASE     =      np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]					,							axis=0			)
      return mel_fusion
 def       UpperCAmelCase__		(		self					:       List[str]					,							__SCREAMING_SNAKE_CASE					:       np.array					,							__SCREAMING_SNAKE_CASE					:       str					,							__SCREAMING_SNAKE_CASE					:       Dict					,							__SCREAMING_SNAKE_CASE					:       Dict			)  -> np.array:


      """simple docstring"""




      if waveform.shape[0] > max_length:
           if truncation == "rand_trunc":
                __SCREAMING_SNAKE_CASE     =      True
                # random crop to max_length (for compatibility) -> this should be handled by self.pad
                __SCREAMING_SNAKE_CASE     =      len(__SCREAMING_SNAKE_CASE			) - max_length
                __SCREAMING_SNAKE_CASE     =      np.random.randint(0					,							overflow + 1			)
                __SCREAMING_SNAKE_CASE     =      waveform[idx : idx + max_length]
                __SCREAMING_SNAKE_CASE     =      self._np_extract_fbank_features(__SCREAMING_SNAKE_CASE					,							self.mel_filters_slaney			)[None, :]
           elif truncation == "fusion":
                __SCREAMING_SNAKE_CASE     =      self._np_extract_fbank_features(__SCREAMING_SNAKE_CASE					,							self.mel_filters			)
                __SCREAMING_SNAKE_CASE     =      max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
                __SCREAMING_SNAKE_CASE     =      mel.shape[0]
                if chunk_frames == total_frames:
                     # there is a corner case where the audio length is larger than max_length but smaller than max_length+hop_length.
                     # In this case, we just use the whole audio.
                     __SCREAMING_SNAKE_CASE     =      np.stack([mel, mel, mel, mel]					,							axis=0			)
                     __SCREAMING_SNAKE_CASE     =      False
                else:
                     __SCREAMING_SNAKE_CASE     =      self._random_mel_fusion(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)
                     __SCREAMING_SNAKE_CASE     =      True
           else:
                raise NotImplementedError(f'data_truncating {truncation} not implemented'			)

      else:
           __SCREAMING_SNAKE_CASE     =      False
           # only use repeat as a new possible value for padding. you repeat the audio before applying the usual max_length padding
           if waveform.shape[0] < max_length:
                if padding == "repeat":
                     __SCREAMING_SNAKE_CASE     =      int(max_length / len(__SCREAMING_SNAKE_CASE			)			)
                     __SCREAMING_SNAKE_CASE     =      np.stack(np.tile(__SCREAMING_SNAKE_CASE					,							n_repeat + 1			)			)[:max_length]
                if padding == "repeatpad":
                     __SCREAMING_SNAKE_CASE     =      int(max_length / len(__SCREAMING_SNAKE_CASE			)			)
                     __SCREAMING_SNAKE_CASE     =      np.stack(np.tile(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)			)
                __SCREAMING_SNAKE_CASE     =      np.pad(__SCREAMING_SNAKE_CASE					,							(0, max_length - waveform.shape[0])					,							mode="""constant"""					,							constant_values=0			)

           if truncation == "fusion":
                __SCREAMING_SNAKE_CASE     =      self._np_extract_fbank_features(__SCREAMING_SNAKE_CASE					,							self.mel_filters			)
                __SCREAMING_SNAKE_CASE     =      np.stack([input_mel, input_mel, input_mel, input_mel]					,							axis=0			)
           else:
                __SCREAMING_SNAKE_CASE     =      self._np_extract_fbank_features(__SCREAMING_SNAKE_CASE					,							self.mel_filters_slaney			)[None, :]

      return input_mel, longer


 def __call__(		self					:       Tuple					,							__SCREAMING_SNAKE_CASE					:       Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]					,							__SCREAMING_SNAKE_CASE					:       str = None					,							__SCREAMING_SNAKE_CASE					:       Optional[str] = None					,							__SCREAMING_SNAKE_CASE					:       Optional[int] = None					,							__SCREAMING_SNAKE_CASE					:       Optional[int] = None					,							__SCREAMING_SNAKE_CASE					:       Optional[Union[str, TensorType]] = None					,							**__SCREAMING_SNAKE_CASE					:       Dict					,							)  -> BatchFeature:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      truncation if truncation is not None else self.truncation
      __SCREAMING_SNAKE_CASE     =      padding if padding else self.padding

      if sampling_rate is not None:
           if sampling_rate != self.sampling_rate:
                raise ValueError(
                    f'The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a'
                    f' sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input'
                    f' was sampled with {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."""			)

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

      if is_batched:
           __SCREAMING_SNAKE_CASE     =      [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			):
           __SCREAMING_SNAKE_CASE     =      np.asarray(__SCREAMING_SNAKE_CASE					,							dtype=np.floataa			)
      elif isinstance(__SCREAMING_SNAKE_CASE					,							np.ndarray			) and raw_speech.dtype is np.dtype(np.floataa			):
           __SCREAMING_SNAKE_CASE     =      raw_speech.astype(np.floataa			)

      # always return batch
      if not is_batched:
           __SCREAMING_SNAKE_CASE     =      [np.asarray(__SCREAMING_SNAKE_CASE			)]

      # convert to mel spectrogram, truncate and pad if needed.
      __SCREAMING_SNAKE_CASE     =      [
          self._get_input_mel(__SCREAMING_SNAKE_CASE					,							max_length if max_length else self.nb_max_samples					,							__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)
          for waveform in raw_speech
      ]

      __SCREAMING_SNAKE_CASE     =      []
      __SCREAMING_SNAKE_CASE     =      []
      for mel, longer in padded_inputs:
           input_mel.append(__SCREAMING_SNAKE_CASE			)
           is_longer.append(__SCREAMING_SNAKE_CASE			)

      if truncation == "fusion" and sum(__SCREAMING_SNAKE_CASE			) == 0:
           # if no audio is longer than 10s, then randomly select one audio to be longer
           __SCREAMING_SNAKE_CASE     =      np.random.randint(0					,							len(__SCREAMING_SNAKE_CASE			)			)
           __SCREAMING_SNAKE_CASE     =      True

      if isinstance(input_mel[0]					,							__SCREAMING_SNAKE_CASE			):
           __SCREAMING_SNAKE_CASE     =      [np.asarray(__SCREAMING_SNAKE_CASE					,							dtype=np.floataa			) for feature in input_mel]

      # is_longer is a list of bool
      __SCREAMING_SNAKE_CASE     =      [[longer] for longer in is_longer]

      __SCREAMING_SNAKE_CASE     =      {"""input_features""": input_mel, """is_longer""": is_longer}
      __SCREAMING_SNAKE_CASE     =      BatchFeature(__SCREAMING_SNAKE_CASE			)

      if return_tensors is not None:
           __SCREAMING_SNAKE_CASE     =      input_features.convert_to_tensors(__SCREAMING_SNAKE_CASE			)

      return input_features









'''simple docstring'''



from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


UpperCAmelCase :       Union[str, Any]			   =     {'configuration_sew': ['SEW_PRETRAINED_CONFIG_ARCHIVE_MAP', 'SEWConfig']}

try:
  if not is_torch_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  UpperCAmelCase :       Union[str, Any]			   =     [
      'SEW_PRETRAINED_MODEL_ARCHIVE_LIST',
      'SEWForCTC',
      'SEWForSequenceClassification',
      'SEWModel',
      'SEWPreTrainedModel',
  ]

if TYPE_CHECKING:
  from .configuration_sew import SEW_PRETRAINED_CONFIG_ARCHIVE_MAP, SEWConfig

  try:
    if not is_torch_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .modeling_sew import (
        SEW_PRETRAINED_MODEL_ARCHIVE_LIST,
        SEWForCTC,
        SEWForSequenceClassification,
        SEWModel,
        SEWPreTrainedModel,
    )


else:
  import sys

  UpperCAmelCase :       int			   =     _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)






