

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


_a							=							logging.get_logger(__name__)




class  _lowerCAmelCase  (     lowercase       ):







 """simple docstring"""







 __UpperCAmelCase     : Optional[Any]       		=     ["input_features", "is_longer"]


 def __init__(      self							:      int, UpperCAmelCase__							:      Optional[Any]=6_4, UpperCAmelCase__							:      List[str]=4_8_0_0_0, UpperCAmelCase__							:      int=4_8_0, UpperCAmelCase__							:      int=1_0, UpperCAmelCase__							:      Any=1_0_2_4, UpperCAmelCase__							:      List[str]=0.0, UpperCAmelCase__							:      Optional[int]=False, UpperCAmelCase__							:      float = 0, UpperCAmelCase__							:      float = 1_4_0_0_0, UpperCAmelCase__							:      int = None, UpperCAmelCase__							:      str = "fusion", UpperCAmelCase__							:      str = "repeatpad", **UpperCAmelCase__							:      Optional[Any], ):
    super().__init__(
        feature_size=UpperCAmelCase__, sampling_rate=UpperCAmelCase__, padding_value=UpperCAmelCase__, return_attention_mask=UpperCAmelCase__, **UpperCAmelCase__, )
    __lowercase								=						top_db
    __lowercase								=						truncation
    __lowercase								=						padding
    __lowercase								=						fft_window_size
    __lowercase								=						(fft_window_size >> 1) + 1
    __lowercase								=						hop_length
    __lowercase								=						max_length_s
    __lowercase								=						max_length_s * sampling_rate
    __lowercase								=						sampling_rate
    __lowercase								=						frequency_min
    __lowercase								=						frequency_max
    __lowercase								=						mel_filter_bank(
        num_frequency_bins=self.nb_frequency_bins, num_mel_filters=UpperCAmelCase__, min_frequency=UpperCAmelCase__, max_frequency=UpperCAmelCase__, sampling_rate=UpperCAmelCase__, norm=UpperCAmelCase__, mel_scale="htk", )
    __lowercase								=						mel_filter_bank(
        num_frequency_bins=self.nb_frequency_bins, num_mel_filters=UpperCAmelCase__, min_frequency=UpperCAmelCase__, max_frequency=UpperCAmelCase__, sampling_rate=UpperCAmelCase__, norm="slaney", mel_scale="slaney", )


 def  _lowercase			(      self							:      int     ):
    __lowercase								=						copy.deepcopy(self.__dict__     )
    __lowercase								=						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  _lowercase			(      self							:      List[str], UpperCAmelCase__							:      np.array, UpperCAmelCase__							:      Optional[np.array] = None     ):
    __lowercase								=						spectrogram(
        UpperCAmelCase__, window_function(self.fft_window_size, "hann"     ), frame_length=self.fft_window_size, hop_length=self.hop_length, power=2.0, mel_filters=UpperCAmelCase__, log_mel="dB", )
    return log_mel_spectrogram.T


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

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

    __lowercase								=						torch.tensor(mel[None, None, :]     )
    __lowercase								=						torch.nn.functional.interpolate(
        UpperCAmelCase__, size=[chunk_frames, 6_4], mode="bilinear", align_corners=UpperCAmelCase__     )
    __lowercase								=						mel_shrink[0][0].numpy()
    __lowercase								=						np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back], axis=0     )
    return mel_fusion


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

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

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

    return input_mel, longer



 def __call__(      self							:      int, UpperCAmelCase__							:      Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]], UpperCAmelCase__							:      str = None, UpperCAmelCase__							:      Optional[str] = None, UpperCAmelCase__							:      Optional[int] = None, UpperCAmelCase__							:      Optional[int] = None, UpperCAmelCase__							:      Optional[Union[str, TensorType]] = None, **UpperCAmelCase__							:      str, ):
    __lowercase								=						truncation if truncation is not None else self.truncation
    __lowercase								=						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."     )

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

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

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

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

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

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

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

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

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

    return input_features







"""simple docstring"""





_a							=							[
    'DownloadConfig',
    'DownloadManager',
    'DownloadMode',
    'StreamingDownloadManager',
]

from .download_config import DownloadConfig
from .download_manager import DownloadManager, DownloadMode
from .streaming_download_manager import StreamingDownloadManager




