


"""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__  							=logging.get_logger(__name__)





class    lowerCamelCase__						(		_a    ):
      a :     str											=		["""input_features""", """is_longer"""]
      def __init__(		self	:	Any      ,       A_	:	Optional[int]=6_4      ,       A_	:	Optional[Any]=4_8_0_0_0      ,       A_	:	Any=4_8_0      ,       A_	:	Optional[Any]=1_0      ,       A_	:	Optional[Any]=1_0_2_4      ,       A_	:	List[str]=0.0      ,       A_	:	str=False      ,       A_	:	float = 0      ,       A_	:	float = 1_4_0_0_0      ,       A_	:	int = None      ,       A_	:	str = "fusion"      ,       A_	:	str = "repeatpad"      ,       **A_	:	Dict      ,       ):






       '''simple docstring'''


       super().__init__(
           feature_size=A_      ,       sampling_rate=A_      ,       padding_value=A_      ,       return_attention_mask=A_      ,       **A_      ,       )
       __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=A_      ,       min_frequency=A_      ,       max_frequency=A_      ,       sampling_rate=A_      ,       norm=A_      ,       mel_scale="""htk"""      ,       )
       __lowercase									=	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 SCREAMING_SNAKE_CASE_	(		self	:	Any							):






       '''simple docstring'''


       __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 SCREAMING_SNAKE_CASE_	(		self	:	str      ,       A_	:	np.array      ,       A_	:	Optional[np.array] = None							):






       '''simple docstring'''


       __lowercase									=	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 SCREAMING_SNAKE_CASE_	(		self	:	Tuple      ,       A_	:	Any      ,       A_	:	Tuple      ,       A_	:	str							):






       '''simple docstring'''


       __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(
           A_      ,       size=[chunk_frames, 6_4]      ,       mode="""bilinear"""      ,       align_corners=A_							)
       __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 SCREAMING_SNAKE_CASE_	(		self	:	Any      ,       A_	:	np.array      ,       A_	:	List[str]      ,       A_	:	Dict      ,       A_	:	Optional[Any]							):






       '''simple docstring'''


       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(A_							) - max_length
         __lowercase									=	np.random.randint(0      ,       overflow + 1							)
         __lowercase									=	waveform[idx : idx + max_length]
         __lowercase									=	self._np_extract_fbank_features(A_      ,       self.mel_filters_slaney							)[None, :]
        elif truncation == "fusion":
         __lowercase									=	self._np_extract_fbank_features(A_      ,       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(A_      ,       A_      ,       A_							)
          __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(A_							)							)
          __lowercase									=	np.stack(np.tile(A_      ,       n_repeat + 1							)							)[:max_length]
         if padding == "repeatpad":
          __lowercase									=	int(max_length / len(A_							)							)
          __lowercase									=	np.stack(np.tile(A_      ,       A_							)							)
         __lowercase									=	np.pad(A_      ,       (0, max_length - waveform.shape[0])      ,       mode="""constant"""      ,       constant_values=0							)

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

       return input_mel, longer






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






       '''simple docstring'''


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

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

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

       __lowercase									=	[]
       __lowercase									=	[]
       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
        __lowercase									=	np.random.randint(0      ,       len(A_							)							)
        __lowercase									=	True

       if isinstance(input_mel[0]      ,       A_							):
        __lowercase									=	[np.asarray(A_      ,       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(A_							)

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

       return input_features




"""simple docstring"""

import inspect
import unittest





class    lowerCamelCase__						(		unittest.TestCase    ):
      def SCREAMING_SNAKE_CASE_	(		self	:	int							):






       '''simple docstring'''


       try:
        import diffusers  # noqa: F401
       except ImportError:
        assert False






      def SCREAMING_SNAKE_CASE_	(		self	:	Optional[Any]							):






       '''simple docstring'''


       import diffusers
       from diffusers.dependency_versions_table import deps

       __lowercase									=	inspect.getmembers(A_      ,       inspect.isclass							)

       for cls_name, cls_module in all_classes:
        if "dummy_" in cls_module.__module__:
         for backend in cls_module._backends:
          if backend == "k_diffusion":
           __lowercase									=	"""k-diffusion"""
          elif backend == "invisible_watermark":
           __lowercase									=	"""invisible-watermark"""
          assert backend in deps, F'''{backend} is not in the deps table!'''
