




import gc
import random
import unittest

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModel,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import AutoencoderKL, DDIMScheduler, DDPMScheduler, StableUnCLIPImgaImgPipeline, UNetaDConditionModel
from diffusers.pipelines.pipeline_utils import DiffusionPipeline
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    floats_tensor,
    load_image,
    load_numpy,
    require_torch_gpu,
    skip_mps,
    slow,
    torch_device,
)

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()

class  _lowerCamelCase						(			_SCREAMING_SNAKE_CASE							,			_SCREAMING_SNAKE_CASE							,			_SCREAMING_SNAKE_CASE							,			unittest.TestCase   ):

      """simple docstring"""

      lowerCAmelCase__			: int    =					StableUnCLIPImgaImgPipeline
      lowerCAmelCase__			: Tuple    =					TEXT_GUIDED_IMAGE_VARIATION_PARAMS
      lowerCAmelCase__			: List[Any]    =					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
      lowerCAmelCase__			: Optional[Any]    =					frozenset(
          []   )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
      lowerCAmelCase__			: List[Any]    =					frozenset([]   )







      def       snake_case	(     self					: Dict				):
          __UpperCamelCase  		=						32
          __UpperCamelCase  		=						embedder_hidden_size

          # image encoding components

          __UpperCamelCase  		=						CLIPImageProcessor(crop_size=32 ,       size=32				)

          torch.manual_seed(0				)
          __UpperCamelCase  		=						CLIPVisionModelWithProjection(
              CLIPVisionConfig(
                  hidden_size=snake_case ,       projection_dim=snake_case ,       num_hidden_layers=5 ,       num_attention_heads=4 ,       image_size=32 ,       intermediate_size=37 ,       patch_size=1 ,       )				)

          # regular denoising components

          torch.manual_seed(0				)
          __UpperCamelCase  		=						StableUnCLIPImageNormalizer(embedding_dim=snake_case				)
          __UpperCamelCase  		=						DDPMScheduler(beta_schedule='''squaredcos_cap_v2'''				)

          torch.manual_seed(0				)
          __UpperCamelCase  		=						CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''				)

          torch.manual_seed(0				)
          __UpperCamelCase  		=						CLIPTextModel(
              CLIPTextConfig(
                  bos_token_id=0 ,       eos_token_id=2 ,       hidden_size=snake_case ,       projection_dim=32 ,       intermediate_size=37 ,       layer_norm_eps=1E-05 ,       num_attention_heads=4 ,       num_hidden_layers=5 ,       pad_token_id=1 ,       vocab_size=1000 ,       )				)

          torch.manual_seed(0				)
          __UpperCamelCase  		=						UNetaDConditionModel(
              sample_size=32 ,       in_channels=4 ,       out_channels=4 ,       down_block_types=('''CrossAttnDownBlock2D''', '''DownBlock2D''') ,       up_block_types=('''UpBlock2D''', '''CrossAttnUpBlock2D''') ,       block_out_channels=(32, 64) ,       attention_head_dim=(2, 4) ,       class_embed_type='''projection''' ,       projection_class_embeddings_input_dim=embedder_projection_dim * 2 ,       cross_attention_dim=snake_case ,       layers_per_block=1 ,       upcast_attention=snake_case ,       use_linear_projection=snake_case ,       )

          torch.manual_seed(0				)
          __UpperCamelCase  		=						DDIMScheduler(
              beta_schedule='''scaled_linear''' ,       beta_start=0.0_00_85 ,       beta_end=0.0_12 ,       prediction_type='''v_prediction''' ,       set_alpha_to_one=snake_case ,       steps_offset=1 ,       )

          torch.manual_seed(0				)
          __UpperCamelCase  		=						AutoencoderKL()

          __UpperCamelCase  		=						{
              # image encoding components
              '''feature_extractor''': feature_extractor,
              '''image_encoder''': image_encoder.eval(),
              # image noising components
              '''image_normalizer''': image_normalizer.eval(),
              '''image_noising_scheduler''': image_noising_scheduler,
              # regular denoising components
              '''tokenizer''': tokenizer,
              '''text_encoder''': text_encoder.eval(),
              '''unet''': unet.eval(),
              '''scheduler''': scheduler,
              '''vae''': vae.eval(),
          }

          return components







      def       snake_case	(     self					: Dict ,       snake_case					: Optional[int] ,       snake_case					: List[str]=0 ,       snake_case					: List[str]=True				):
          if str(snake_case				).startswith('''mps'''				):
              __UpperCamelCase  		=						torch.manual_seed(snake_case				)
          else:
              __UpperCamelCase  		=						torch.Generator(device=snake_case				).manual_seed(snake_case				)

          __UpperCamelCase  		=						floats_tensor((1, 3, 32, 32) ,       rng=random.Random(snake_case				)				).to(snake_case				)

          if pil_image:
              __UpperCamelCase  		=						input_image * 0.5 + 0.5
              __UpperCamelCase  		=						input_image.clamp(0 ,       1				)
              __UpperCamelCase  		=						input_image.cpu().permute(0 ,       2 ,       3 ,       1				).float().numpy()
              __UpperCamelCase  		=						DiffusionPipeline.numpy_to_pil(snake_case				)[0]

          return {
              "prompt": "An anime racoon running a marathon",
              "image": input_image,
              "generator": generator,
              "num_inference_steps": 2,
              "output_type": "np",
          }







      @skip_mps
      def       snake_case	(     self					: Optional[Any]				):
          __UpperCamelCase  		=						'''cpu'''  # ensure determinism for the device-dependent torch.Generator
          __UpperCamelCase  		=						self.get_dummy_components()
          __UpperCamelCase  		=						StableUnCLIPImgaImgPipeline(**snake_case				)
          __UpperCamelCase  		=						sd_pipe.to(snake_case				)
          sd_pipe.set_progress_bar_config(disable=snake_case				)

          __UpperCamelCase  		=						self.get_dummy_inputs(snake_case				)
          inputs.update({'''image_embeds''': None}				)
          __UpperCamelCase  		=						sd_pipe(**snake_case				).images
          __UpperCamelCase  		=						image[0, -3:, -3:, -1]

          assert image.shape == (1, 32, 32, 3)
          __UpperCamelCase  		=						np.array([0.38_72, 0.72_24, 0.56_01, 0.47_41, 0.68_72, 0.58_14, 0.46_36, 0.38_67, 0.50_78]				)

          assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-3







      def       snake_case	(     self					: List[Any]				):
          __UpperCamelCase  		=						torch_device in ['''cpu''', '''mps''']

          self._test_attention_slicing_forward_pass(test_max_difference=snake_case				)







      def       snake_case	(     self					: List[str]				):
          __UpperCamelCase  		=						torch_device in ['''cpu''', '''mps''']

          self._test_inference_batch_single_identical(test_max_difference=snake_case				)







      @unittest.skipIf(
          torch_device != '''cuda''' or not is_xformers_available() ,       reason='''XFormers attention is only available with CUDA and `xformers` installed''' ,       )
      def       snake_case	(     self					: Dict				):
          self._test_xformers_attention_forwardGenerator_pass(test_max_difference=snake_case				)







@slow
@require_torch_gpu
class  _lowerCamelCase						(			unittest.TestCase   ):

      """simple docstring"""

      def       snake_case	(     self					: Union[str, Any]				):
          # clean up the VRAM after each test
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()







      def       snake_case	(     self					: Tuple				):
          __UpperCamelCase  		=						load_image(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'''				)

          __UpperCamelCase  		=						load_numpy(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_img2img_anime_turtle_fp16.npy'''				)

          __UpperCamelCase  		=						StableUnCLIPImgaImgPipeline.from_pretrained(
              '''fusing/stable-unclip-2-1-l-img2img''' ,       torch_dtype=torch.floataa				)
          pipe.to(snake_case				)
          pipe.set_progress_bar_config(disable=snake_case				)
          # stable unclip will oom when integration tests are run on a V100,
          # so turn on memory savings
          pipe.enable_attention_slicing()
          pipe.enable_sequential_cpu_offload()

          __UpperCamelCase  		=						torch.Generator(device='''cpu'''				).manual_seed(0				)
          __UpperCamelCase  		=						pipe(snake_case ,       '''anime turle''' ,       generator=snake_case ,       output_type='''np'''				)

          __UpperCamelCase  		=						output.images[0]

          assert image.shape == (768, 768, 3)

          assert_mean_pixel_difference(snake_case ,       snake_case				)







      def       snake_case	(     self					: Any				):
          __UpperCamelCase  		=						load_image(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'''				)

          __UpperCamelCase  		=						load_numpy(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_h_img2img_anime_turtle_fp16.npy'''				)

          __UpperCamelCase  		=						StableUnCLIPImgaImgPipeline.from_pretrained(
              '''fusing/stable-unclip-2-1-h-img2img''' ,       torch_dtype=torch.floataa				)
          pipe.to(snake_case				)
          pipe.set_progress_bar_config(disable=snake_case				)
          # stable unclip will oom when integration tests are run on a V100,
          # so turn on memory savings
          pipe.enable_attention_slicing()
          pipe.enable_sequential_cpu_offload()

          __UpperCamelCase  		=						torch.Generator(device='''cpu'''				).manual_seed(0				)
          __UpperCamelCase  		=						pipe(snake_case ,       '''anime turle''' ,       generator=snake_case ,       output_type='''np'''				)

          __UpperCamelCase  		=						output.images[0]

          assert image.shape == (768, 768, 3)

          assert_mean_pixel_difference(snake_case ,       snake_case				)







      def       snake_case	(     self					: List[str]				):
          __UpperCamelCase  		=						load_image(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'''				)

          torch.cuda.empty_cache()
          torch.cuda.reset_max_memory_allocated()
          torch.cuda.reset_peak_memory_stats()

          __UpperCamelCase  		=						StableUnCLIPImgaImgPipeline.from_pretrained(
              '''fusing/stable-unclip-2-1-h-img2img''' ,       torch_dtype=torch.floataa				)
          __UpperCamelCase  		=						pipe.to(snake_case				)
          pipe.set_progress_bar_config(disable=snake_case				)
          pipe.enable_attention_slicing()
          pipe.enable_sequential_cpu_offload()

          __UpperCamelCase  		=						pipe(
              snake_case ,       '''anime turtle''' ,       num_inference_steps=2 ,       output_type='''np''' ,       )

          __UpperCamelCase  		=						torch.cuda.max_memory_allocated()
          # make sure that less than 7 GB is allocated
          assert mem_bytes < 7 * 10**9












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						(			_SCREAMING_SNAKE_CASE   ):

      """simple docstring"""

      lowerCAmelCase__			: Dict    =					["input_features", "is_longer"]







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







      def       snake_case	(     self					: Union[str, Any]				):
          __UpperCamelCase  		=						copy.deepcopy(self.__dict__				)
          __UpperCamelCase  		=						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					: Tuple ,       snake_case					: np.array ,       snake_case					: Optional[np.array] = None				):
          __UpperCamelCase  		=						spectrogram(
              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=snake_case ,       log_mel='''dB''' ,       )
          return log_mel_spectrogram.T







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

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

          __UpperCamelCase  		=						torch.tensor(mel[None, None, :]				)
          __UpperCamelCase  		=						torch.nn.functional.interpolate(
              snake_case ,       size=[chunk_frames, 64] ,       mode='''bilinear''' ,       align_corners=snake_case				)
          __UpperCamelCase  		=						mel_shrink[0][0].numpy()
          __UpperCamelCase  		=						np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back] ,       axis=0				)
          return mel_fusion







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

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

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

          return input_mel, longer







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

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

          if is_batched:
              __UpperCamelCase  		=						[np.asarray(snake_case ,       dtype=np.floataa				) for speech in raw_speech]
          elif not is_batched and not isinstance(snake_case ,       np.ndarray				):
              __UpperCamelCase  		=						np.asarray(snake_case ,       dtype=np.floataa				)
          elif isinstance(snake_case ,       np.ndarray				) and raw_speech.dtype is np.dtype(np.floataa				):
              __UpperCamelCase  		=						raw_speech.astype(np.floataa				)

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

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

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

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

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

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

          __UpperCamelCase  		=						{'''input_features''': input_mel, '''is_longer''': is_longer}
          __UpperCamelCase  		=						BatchFeature(snake_case				)

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

          return input_features






