



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


_lowercase     							=						logging.get_logger(__name__)
class 				__A					( __lowercase      ):
   UpperCamelCase					:List[str]     =			['''input_features''', '''is_longer''']





   def __init__(self							,    __magic_name__=64							,    __magic_name__=48000							,    __magic_name__=480							,    __magic_name__=10							,    __magic_name__=1024							,    __magic_name__=0.0							,    __magic_name__=False							,    __magic_name__ = 0							,    __magic_name__ = 14000							,    __magic_name__ = None							,    __magic_name__ = "fusion"							,    __magic_name__ = "repeatpad"							,    **__magic_name__							,    ):
      super().__init__(
          feature_size=__a							,    sampling_rate=__a							,    padding_value=__a							,    return_attention_mask=__a							,    **__a							,    )
      lowerCamelCase__ :      List[Any]								=		top_db
      lowerCamelCase__ :      Optional[int]								=		truncation
      lowerCamelCase__ :      Any								=		padding
      lowerCamelCase__ :      Dict								=		fft_window_size
      lowerCamelCase__ :      Tuple								=		(fft_window_size >> 1) + 1
      lowerCamelCase__ :      Dict								=		hop_length
      lowerCamelCase__ :      Tuple								=		max_length_s
      lowerCamelCase__ :      Any								=		max_length_s * sampling_rate
      lowerCamelCase__ :      Optional[int]								=		sampling_rate
      lowerCamelCase__ :      str								=		frequency_min
      lowerCamelCase__ :      Optional[int]								=		frequency_max
      lowerCamelCase__ :      str								=		mel_filter_bank(
          num_frequency_bins=self.nb_frequency_bins							,    num_mel_filters=__a							,    min_frequency=__a							,    max_frequency=__a							,    sampling_rate=__a							,    norm=__a							,    mel_scale="""htk"""							,    )
      lowerCamelCase__ :      Optional[Any]								=		mel_filter_bank(
          num_frequency_bins=self.nb_frequency_bins							,    num_mel_filters=__a							,    min_frequency=__a							,    max_frequency=__a							,    sampling_rate=__a							,    norm="""slaney"""							,    mel_scale="""slaney"""							,    )





   def        _snake_case    (self				):
      lowerCamelCase__ :      Union[str, Any]								=		copy.deepcopy(self.__dict__				)
      lowerCamelCase__ :      Optional[Any]								=		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        _snake_case    (self							,    __magic_name__							,    __magic_name__ = None				):
      lowerCamelCase__ :      Optional[Any]								=		spectrogram(
          __a							,    window_function(self.fft_window_size							,    """hann"""				)							,    frame_length=self.fft_window_size							,    hop_length=self.hop_length							,    power=2.0							,    mel_filters=__a							,    log_mel="""dB"""							,    )
      return log_mel_spectrogram.T





   def        _snake_case    (self							,    __magic_name__							,    __magic_name__							,    __magic_name__				):
      lowerCamelCase__ :      List[str]								=		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
         lowerCamelCase__ :      Any								=		[0]
      if len(ranges[2]				) == 0:
         # if the audio is too short, we just use the first chunk
         lowerCamelCase__ :      List[Any]								=		[0]
      # randomly choose index for each part
      lowerCamelCase__ :      int								=		np.random.choice(ranges[0]				)
      lowerCamelCase__ :      List[str]								=		np.random.choice(ranges[1]				)
      lowerCamelCase__ :      str								=		np.random.choice(ranges[2]				)

      lowerCamelCase__ :      Tuple								=		mel[idx_front : idx_front + chunk_frames, :]
      lowerCamelCase__ :      str								=		mel[idx_middle : idx_middle + chunk_frames, :]
      lowerCamelCase__ :      str								=		mel[idx_back : idx_back + chunk_frames, :]

      lowerCamelCase__ :      Union[str, Any]								=		torch.tensor(mel[None, None, :]				)
      lowerCamelCase__ :      Dict								=		torch.nn.functional.interpolate(
          __a							,    size=[chunk_frames, 64]							,    mode="""bilinear"""							,    align_corners=__a				)
      lowerCamelCase__ :      str								=		mel_shrink[0][0].numpy()
      lowerCamelCase__ :      Optional[int]								=		np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]							,    axis=0				)
      return mel_fusion





   def        _snake_case    (self							,    __magic_name__							,    __magic_name__							,    __magic_name__							,    __magic_name__				):
      if waveform.shape[0] > max_length:
         if truncation == "rand_trunc":
            lowerCamelCase__ :      int								=		True
            # random crop to max_length (for compatibility) -> this should be handled by self.pad
            lowerCamelCase__ :      str								=		len(__a				) - max_length
            lowerCamelCase__ :      Optional[Any]								=		np.random.randint(0							,    overflow + 1				)
            lowerCamelCase__ :      Union[str, Any]								=		waveform[idx : idx + max_length]
            lowerCamelCase__ :      List[Any]								=		self._np_extract_fbank_features(__a							,    self.mel_filters_slaney				)[None, :]
         elif truncation == "fusion":
            lowerCamelCase__ :      Optional[Any]								=		self._np_extract_fbank_features(__a							,    self.mel_filters				)
            lowerCamelCase__ :      Tuple								=		max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
            lowerCamelCase__ :      str								=		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.
               lowerCamelCase__ :      Any								=		np.stack([mel, mel, mel, mel]							,    axis=0				)
               lowerCamelCase__ :      List[str]								=		False
            else:
               lowerCamelCase__ :      Any								=		self._random_mel_fusion(__a							,    __a							,    __a				)
               lowerCamelCase__ :      Optional[Any]								=		True
         else:
            raise NotImplementedError(f"data_truncating {truncation} not implemented"				)

      else:
         lowerCamelCase__ :      Dict								=		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":
               lowerCamelCase__ :      Tuple								=		int(max_length / len(__a				)				)
               lowerCamelCase__ :      Dict								=		np.stack(np.tile(__a							,    n_repeat + 1				)				)[:max_length]
            if padding == "repeatpad":
               lowerCamelCase__ :      Optional[Any]								=		int(max_length / len(__a				)				)
               lowerCamelCase__ :      Optional[Any]								=		np.stack(np.tile(__a							,    __a				)				)
            lowerCamelCase__ :      Optional[Any]								=		np.pad(__a							,    (0, max_length - waveform.shape[0])							,    mode="""constant"""							,    constant_values=0				)

         if truncation == "fusion":
            lowerCamelCase__ :      str								=		self._np_extract_fbank_features(__a							,    self.mel_filters				)
            lowerCamelCase__ :      Optional[Any]								=		np.stack([input_mel, input_mel, input_mel, input_mel]							,    axis=0				)
         else:
            lowerCamelCase__ :      Dict								=		self._np_extract_fbank_features(__a							,    self.mel_filters_slaney				)[None, :]

      return input_mel, longer





   def __call__(self							,    __magic_name__							,    __magic_name__ = None							,    __magic_name__ = None							,    __magic_name__ = None							,    __magic_name__ = None							,    __magic_name__ = None							,    **__magic_name__							,    ):
      lowerCamelCase__ :      Optional[Any]								=		truncation if truncation is not None else self.truncation
      lowerCamelCase__ :      str								=		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."""				)

      lowerCamelCase__ :      int								=		isinstance(__a							,    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__ :      str								=		is_batched_numpy or (
          isinstance(__a							,    (list, tuple)				) and (isinstance(raw_speech[0]							,    (np.ndarray, tuple, list)				))
      )

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

      # always return batch
      if not is_batched:
         lowerCamelCase__ :      List[Any]								=		[np.asarray(__a				)]

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

      lowerCamelCase__ :      Optional[int]								=		[]
      lowerCamelCase__ :      Union[str, Any]								=		[]
      for mel, longer in padded_inputs:
         input_mel.append(__a				)
         is_longer.append(__a				)

      if truncation == "fusion" and sum(__a				) == 0:
         # if no audio is longer than 10s, then randomly select one audio to be longer
         lowerCamelCase__ :      Optional[Any]								=		np.random.randint(0							,    len(__a				)				)
         lowerCamelCase__ :      Union[str, Any]								=		True

      if isinstance(input_mel[0]							,    __a				):
         lowerCamelCase__ :      List[str]								=		[np.asarray(__a							,    dtype=np.floataa				) for feature in input_mel]

      # is_longer is a list of bool
      lowerCamelCase__ :      Tuple								=		[[longer] for longer in is_longer]

      lowerCamelCase__ :      Union[str, Any]								=		{"""input_features""": input_mel, """is_longer""": is_longer}
      lowerCamelCase__ :      Any								=		BatchFeature(__a				)

      if return_tensors is not None:
         lowerCamelCase__ :      str								=		input_features.convert_to_tensors(__a				)

      return input_features








import copy
from dataclasses import dataclass, field
from typing import ClassVar, Dict

from ..features import Audio, Features, Value
from .base import TaskTemplate
@dataclass(frozen=A_      )
class 				__A					( A_      ):
			UpperCamelCase					:str     =			field(default='''automatic-speech-recognition'''					, metadata={'''include_in_asdict_even_if_is_default''': True}      )
			UpperCamelCase					:ClassVar[Features]     =			Features({'''audio''': Audio()}      )
			UpperCamelCase					:ClassVar[Features]     =			Features({'''transcription''': Value('''string'''      )}      )
			UpperCamelCase					:str     =			"audio"
			UpperCamelCase					:str     =			"transcription"





			def        _snake_case    (self							,    __magic_name__				):
						if self.audio_column not in features:
									raise ValueError(f"Column {self.audio_column} is not present in features."				)
						if not isinstance(features[self.audio_column]							,    __magic_name__				):
									raise ValueError(f"Column {self.audio_column} is not an Audio type."				)
						lowerCamelCase__ :      int								=		copy.deepcopy(self				)
						lowerCamelCase__ :      Tuple								=		self.input_schema.copy()
						lowerCamelCase__ :      Tuple								=		features[self.audio_column]
						lowerCamelCase__ :      List[str]								=		input_schema
						return task_template





			@property
			def        _snake_case    (self				):
						return {self.audio_column: "audio", self.transcription_column: "transcription"}



