

'''simple docstring'''

import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionImageVariationPipeline
from diffusers.utils.testing_utils import load_image, require_torch_gpu, slow, torch_device


A_       						=							False





class 							_snake_case (       unittest.TestCase						):
   pass





@slow
@require_torch_gpu
class 							_snake_case (       unittest.TestCase						):
   def     __UpperCamelCase						( self			:      str							):
     SCREAMING_SNAKE_CASE:List[str]						      =					VersatileDiffusionImageVariationPipeline.from_pretrained("shi-labs/versatile-diffusion"							)
     pipe.to(SCREAMING_SNAKE_CASE__							)
     pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__							)

     SCREAMING_SNAKE_CASE:Tuple						      =					load_image(
         "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg"							)
     SCREAMING_SNAKE_CASE:Optional[Any]						      =					torch.manual_seed(0							)
     SCREAMING_SNAKE_CASE:int						      =					pipe(
         image=SCREAMING_SNAKE_CASE__     ,generator=SCREAMING_SNAKE_CASE__     ,guidance_scale=7.5     ,num_inference_steps=50     ,output_type="numpy"     ,).images

     SCREAMING_SNAKE_CASE:str						      =					image[0, 253:256, 253:256, -1]

     assert image.shape == (1, 512, 512, 3)
     SCREAMING_SNAKE_CASE:Optional[Any]						      =					np.array([0.0_441, 0.0_469, 0.0_507, 0.0_575, 0.0_632, 0.0_650, 0.0_865, 0.0_909, 0.0_945]							)

     assert np.abs(image_slice.flatten() - expected_slice							).max() < 1e-2









'''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 							_snake_case (       _a						):
   _A       :     Any       =					['''input_features''', '''is_longer''']
   def __init__( self			:      str     ,SCREAMING_SNAKE_CASE__			:      str=64     ,SCREAMING_SNAKE_CASE__			:      Optional[Any]=48_000     ,SCREAMING_SNAKE_CASE__			:      List[str]=480     ,SCREAMING_SNAKE_CASE__			:      Optional[int]=10     ,SCREAMING_SNAKE_CASE__			:      List[Any]=1_024     ,SCREAMING_SNAKE_CASE__			:      List[str]=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__			:      int     ,):
     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:Dict						      =					top_db
     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					truncation
     SCREAMING_SNAKE_CASE:Optional[Any]						      =					padding
     SCREAMING_SNAKE_CASE:int						      =					fft_window_size
     SCREAMING_SNAKE_CASE:Tuple						      =					(fft_window_size >> 1) + 1
     SCREAMING_SNAKE_CASE:Dict						      =					hop_length
     SCREAMING_SNAKE_CASE:List[Any]						      =					max_length_s
     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					max_length_s * sampling_rate
     SCREAMING_SNAKE_CASE:int						      =					sampling_rate
     SCREAMING_SNAKE_CASE:Optional[int]						      =					frequency_min
     SCREAMING_SNAKE_CASE:List[str]						      =					frequency_max
     SCREAMING_SNAKE_CASE:int						      =					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:Optional[Any]						      =					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			:      str							):
     SCREAMING_SNAKE_CASE:List[str]						      =					copy.deepcopy(self.__dict__							)
     SCREAMING_SNAKE_CASE:Union[str, 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     __UpperCamelCase						( self			:      Dict     ,SCREAMING_SNAKE_CASE__			:      np.array     ,SCREAMING_SNAKE_CASE__			:      Optional[np.array] = None							):
     SCREAMING_SNAKE_CASE:Dict						      =					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			:      str     ,SCREAMING_SNAKE_CASE__			:      Optional[Any]     ,SCREAMING_SNAKE_CASE__			:      Union[str, Any]     ,SCREAMING_SNAKE_CASE__			:      Optional[int]							):
     SCREAMING_SNAKE_CASE:Tuple						      =					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:Tuple						      =					[0]
     if len(ranges[2]							) == 0:
       # if the audio is too short, we just use the first chunk
       SCREAMING_SNAKE_CASE:Optional[Any]						      =					[0]
     # randomly choose index for each part
     SCREAMING_SNAKE_CASE:Optional[Any]						      =					np.random.choice(ranges[0]							)
     SCREAMING_SNAKE_CASE:Optional[Any]						      =					np.random.choice(ranges[1]							)
     SCREAMING_SNAKE_CASE:Any						      =					np.random.choice(ranges[2]							)

     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					mel[idx_front : idx_front + chunk_frames, :]
     SCREAMING_SNAKE_CASE:Any						      =					mel[idx_middle : idx_middle + chunk_frames, :]
     SCREAMING_SNAKE_CASE:Optional[int]						      =					mel[idx_back : idx_back + chunk_frames, :]

     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					torch.tensor(mel[None, None, :]							)
     SCREAMING_SNAKE_CASE:List[str]						      =					torch.nn.functional.interpolate(
         SCREAMING_SNAKE_CASE__     ,size=[chunk_frames, 64]     ,mode="bilinear"     ,align_corners=SCREAMING_SNAKE_CASE__							)
     SCREAMING_SNAKE_CASE:Optional[Any]						      =					mel_shrink[0][0].numpy()
     SCREAMING_SNAKE_CASE:Any						      =					np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]     ,axis=0							)
     return mel_fusion
   def     __UpperCamelCase						( self			:      Any     ,SCREAMING_SNAKE_CASE__			:      np.array     ,SCREAMING_SNAKE_CASE__			:      str     ,SCREAMING_SNAKE_CASE__			:      Optional[Any]     ,SCREAMING_SNAKE_CASE__			:      Optional[int]							):
     if waveform.shape[0] > max_length:
       if truncation == "rand_trunc":
         SCREAMING_SNAKE_CASE:Optional[Any]						      =					True
         # random crop to max_length (for compatibility) -> this should be handled by self.pad
         SCREAMING_SNAKE_CASE:Dict						      =					len(SCREAMING_SNAKE_CASE__							) - max_length
         SCREAMING_SNAKE_CASE:List[str]						      =					np.random.randint(0     ,overflow + 1							)
         SCREAMING_SNAKE_CASE:int						      =					waveform[idx : idx + max_length]
         SCREAMING_SNAKE_CASE:Optional[int]						      =					self._np_extract_fbank_features(SCREAMING_SNAKE_CASE__     ,self.mel_filters_slaney							)[None, :]
       elif truncation == "fusion":
         SCREAMING_SNAKE_CASE:Union[str, Any]						      =					self._np_extract_fbank_features(SCREAMING_SNAKE_CASE__     ,self.mel_filters							)
         SCREAMING_SNAKE_CASE:Optional[int]						      =					max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
         SCREAMING_SNAKE_CASE:Dict						      =					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:Union[str, Any]						      =					np.stack([mel, mel, mel, mel]     ,axis=0							)
           SCREAMING_SNAKE_CASE:Dict						      =					False
         else:
           SCREAMING_SNAKE_CASE:Optional[Any]						      =					self._random_mel_fusion(SCREAMING_SNAKE_CASE__     ,SCREAMING_SNAKE_CASE__     ,SCREAMING_SNAKE_CASE__							)
           SCREAMING_SNAKE_CASE:Optional[Any]						      =					True
       else:
         raise NotImplementedError(F'''data_truncating {truncation} not implemented'''							)

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

       if truncation == "fusion":
         SCREAMING_SNAKE_CASE:List[Any]						      =					self._np_extract_fbank_features(SCREAMING_SNAKE_CASE__     ,self.mel_filters							)
         SCREAMING_SNAKE_CASE:Dict						      =					np.stack([input_mel, input_mel, input_mel, input_mel]     ,axis=0							)
       else:
         SCREAMING_SNAKE_CASE:Any						      =					self._np_extract_fbank_features(SCREAMING_SNAKE_CASE__     ,self.mel_filters_slaney							)[None, :]

     return input_mel, longer







   def __call__( self			:      str     ,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__			:      List[str]     ,):
     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					truncation if truncation is not None else self.truncation
     SCREAMING_SNAKE_CASE:List[Any]						      =					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:Optional[Any]						      =					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:Dict						      =					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:Any						      =					[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:str						      =					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:Union[str, Any]						      =					raw_speech.astype(np.floataa							)

     # always return batch
     if not is_batched:
       SCREAMING_SNAKE_CASE:List[str]						      =					[np.asarray(SCREAMING_SNAKE_CASE__							)]

     # convert to mel spectrogram, truncate and pad if needed.
     SCREAMING_SNAKE_CASE:List[str]						      =					[
         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:Union[str, Any]						      =					[]
     SCREAMING_SNAKE_CASE:Dict						      =					[]
     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:str						      =					np.random.randint(0     ,len(SCREAMING_SNAKE_CASE__							)							)
       SCREAMING_SNAKE_CASE:str						      =					True

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

     # is_longer is a list of bool
     SCREAMING_SNAKE_CASE:int						      =					[[longer] for longer in is_longer]

     SCREAMING_SNAKE_CASE:Tuple						      =					{"input_features": input_mel, "is_longer": is_longer}
     SCREAMING_SNAKE_CASE:str						      =					BatchFeature(SCREAMING_SNAKE_CASE__							)

     if return_tensors is not None:
       SCREAMING_SNAKE_CASE:List[str]						      =					input_features.convert_to_tensors(SCREAMING_SNAKE_CASE__							)

     return input_features






