




"""simple docstring"""


import gc
import random
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, CycleDiffusionPipeline, DDIMScheduler, UNetaDConditionModel
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

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


enable_full_determinism()






class 					lowercase(     UpperCAmelCase_       ,			UpperCAmelCase_       ,			unittest.TestCase ):







 '''simple docstring'''





 lowercase__ 					=					CycleDiffusionPipeline
 lowercase__ 					=					TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
     'negative_prompt',
     'height',
     'width',
     'negative_prompt_embeds',
 }
 lowercase__ 					=					PipelineTesterMixin.required_optional_params - {'latents'}
 lowercase__ 					=					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"source_prompt"} )
 lowercase__ 					=					IMAGE_TO_IMAGE_IMAGE_PARAMS
 lowercase__ 					=					IMAGE_TO_IMAGE_IMAGE_PARAMS


 def  UpperCamelCase_   (				self:		str		):




   '''simple docstring'''



   torch.manual_seed(0		)
   _snake_case							:		Optional[int]								=  UNetaDConditionModel(
       block_out_channels=(32, 64),      layers_per_block=2,      sample_size=32,      in_channels=4,      out_channels=4,      down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D"""),      up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D"""),      cross_attention_dim=32,      )
   _snake_case							:		Optional[Any]								=  DDIMScheduler(
       beta_start=0.00_085,      beta_end=0.012,      beta_schedule="""scaled_linear""",      num_train_timesteps=1_000,      clip_sample=__UpperCAmelCase,      set_alpha_to_one=__UpperCAmelCase,      )
   torch.manual_seed(0		)
   _snake_case							:		Any								=  AutoencoderKL(
       block_out_channels=[32, 64],      in_channels=3,      out_channels=3,      down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""],      up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""],      latent_channels=4,      )
   torch.manual_seed(0		)
   _snake_case							:		Tuple								=  CLIPTextConfig(
       bos_token_id=0,      eos_token_id=2,      hidden_size=32,      intermediate_size=37,      layer_norm_eps=1E-05,      num_attention_heads=4,      num_hidden_layers=5,      pad_token_id=1,      vocab_size=1_000,      )
   _snake_case							:		Union[str, Any]								=  CLIPTextModel(__UpperCAmelCase		)
   _snake_case							:		Any								=  CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""		)

   _snake_case							:		Optional[Any]								=  {
       """unet""": unet,
       """scheduler""": scheduler,
       """vae""": vae,
       """text_encoder""": text_encoder,
       """tokenizer""": tokenizer,
       """safety_checker""": None,
       """feature_extractor""": None,
   }
   return components


 def  UpperCamelCase_   (				self:		str,      a_:		Union[str, Any],      a_:		List[Any]=0		):




   '''simple docstring'''



   _snake_case							:		Optional[int]								=  floats_tensor((1, 3, 32, 32),      rng=random.Random(__UpperCAmelCase		)		).to(__UpperCAmelCase		)
   _snake_case							:		Any								=  image / 2 + 0.5
   if str(__UpperCAmelCase		).startswith("""mps"""		):
     _snake_case							:		List[Any]								=  torch.manual_seed(__UpperCAmelCase		)
   else:
     _snake_case							:		Optional[Any]								=  torch.Generator(device=__UpperCAmelCase		).manual_seed(__UpperCAmelCase		)
   _snake_case							:		List[str]								=  {
       """prompt""": """An astronaut riding an elephant""",
       """source_prompt""": """An astronaut riding a horse""",
       """image""": image,
       """generator""": generator,
       """num_inference_steps""": 2,
       """eta""": 0.1,
       """strength""": 0.8,
       """guidance_scale""": 3,
       """source_guidance_scale""": 1,
       """output_type""": """numpy""",
   }
   return inputs


 def  UpperCamelCase_   (				self:		List[Any]		):




   '''simple docstring'''



   _snake_case							:		List[str]								=  """cpu"""  # ensure determinism for the device-dependent torch.Generator

   _snake_case							:		Optional[Any]								=  self.get_dummy_components()
   _snake_case							:		Optional[int]								=  CycleDiffusionPipeline(**__UpperCAmelCase		)
   _snake_case							:		Optional[Any]								=  pipe.to(__UpperCAmelCase		)
   pipe.set_progress_bar_config(disable=__UpperCAmelCase		)

   _snake_case							:		Union[str, Any]								=  self.get_dummy_inputs(__UpperCAmelCase		)
   _snake_case							:		Tuple								=  pipe(**__UpperCAmelCase		)
   _snake_case							:		Optional[int]								=  output.images

   _snake_case							:		Dict								=  images[0, -3:, -3:, -1]

   assert images.shape == (1, 32, 32, 3)
   _snake_case							:		int								=  np.array([0.4_459, 0.4_943, 0.4_544, 0.6_643, 0.5_474, 0.4_327, 0.5_701, 0.5_959, 0.5_179]		)

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


 @unittest.skipIf(torch_device != """cuda""",      """This test requires a GPU"""		)
 def  UpperCamelCase_   (				self:		int		):




   '''simple docstring'''



   _snake_case							:		Optional[Any]								=  self.get_dummy_components()
   for name, module in components.items():
     if hasattr(__UpperCAmelCase,      """half"""		):
       _snake_case							:		Optional[Any]								=  module.half()
   _snake_case							:		str								=  CycleDiffusionPipeline(**__UpperCAmelCase		)
   _snake_case							:		List[Any]								=  pipe.to(__UpperCAmelCase		)
   pipe.set_progress_bar_config(disable=__UpperCAmelCase		)

   _snake_case							:		Tuple								=  self.get_dummy_inputs(__UpperCAmelCase		)
   _snake_case							:		Optional[Any]								=  pipe(**__UpperCAmelCase		)
   _snake_case							:		List[Any]								=  output.images

   _snake_case							:		Optional[int]								=  images[0, -3:, -3:, -1]

   assert images.shape == (1, 32, 32, 3)
   _snake_case							:		Optional[Any]								=  np.array([0.3_506, 0.4_543, 0.446, 0.4_575, 0.5_195, 0.4_155, 0.5_273, 0.518, 0.4_116]		)

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


 @skip_mps
 def  UpperCamelCase_   (				self:		Optional[int]		):




   '''simple docstring'''



   return super().test_save_load_local()


 @unittest.skip("""non-deterministic pipeline"""		)
 def  UpperCamelCase_   (				self:		Any		):




   '''simple docstring'''



   return super().test_inference_batch_single_identical()


 @skip_mps
 def  UpperCamelCase_   (				self:		List[str]		):




   '''simple docstring'''



   return super().test_dict_tuple_outputs_equivalent()


 @skip_mps
 def  UpperCamelCase_   (				self:		Optional[int]		):




   '''simple docstring'''



   return super().test_save_load_optional_components()


 @skip_mps
 def  UpperCamelCase_   (				self:		Union[str, Any]		):




   '''simple docstring'''



   return super().test_attention_slicing_forward_pass()







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







 '''simple docstring'''





 def  UpperCamelCase_   (				self:		Dict		):




   '''simple docstring'''



   super().tearDown()
   gc.collect()
   torch.cuda.empty_cache()


 def  UpperCamelCase_   (				self:		Union[str, Any]		):




   '''simple docstring'''



   _snake_case							:		str								=  load_image(
       """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
       """/cycle-diffusion/black_colored_car.png"""		)
   _snake_case							:		Tuple								=  load_numpy(
       """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy"""		)
   _snake_case							:		List[str]								=  init_image.resize((512, 512)		)

   _snake_case							:		Union[str, Any]								=  """CompVis/stable-diffusion-v1-4"""
   _snake_case							:		Any								=  DDIMScheduler.from_pretrained(__UpperCAmelCase,      subfolder="""scheduler"""		)
   _snake_case							:		int								=  CycleDiffusionPipeline.from_pretrained(
       __UpperCAmelCase,      scheduler=__UpperCAmelCase,      safety_checker=__UpperCAmelCase,      torch_dtype=torch.floataa,      revision="""fp16"""		)

   pipe.to(__UpperCAmelCase		)
   pipe.set_progress_bar_config(disable=__UpperCAmelCase		)
   pipe.enable_attention_slicing()

   _snake_case							:		Dict								=  """A black colored car"""
   _snake_case							:		int								=  """A blue colored car"""

   _snake_case							:		str								=  torch.manual_seed(0		)
   _snake_case							:		Any								=  pipe(
       prompt=__UpperCAmelCase,      source_prompt=__UpperCAmelCase,      image=__UpperCAmelCase,      num_inference_steps=100,      eta=0.1,      strength=0.85,      guidance_scale=3,      source_guidance_scale=1,      generator=__UpperCAmelCase,      output_type="""np""",      )
   _snake_case							:		Tuple								=  output.images

   # the values aren't exactly equal, but the images look the same visually
   assert np.abs(image - expected_image		).max() < 5E-1


 def  UpperCamelCase_   (				self:		Any		):




   '''simple docstring'''



   _snake_case							:		Dict								=  load_image(
       """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
       """/cycle-diffusion/black_colored_car.png"""		)
   _snake_case							:		int								=  load_numpy(
       """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy"""		)
   _snake_case							:		Union[str, Any]								=  init_image.resize((512, 512)		)

   _snake_case							:		Any								=  """CompVis/stable-diffusion-v1-4"""
   _snake_case							:		Optional[Any]								=  DDIMScheduler.from_pretrained(__UpperCAmelCase,      subfolder="""scheduler"""		)
   _snake_case							:		Dict								=  CycleDiffusionPipeline.from_pretrained(__UpperCAmelCase,      scheduler=__UpperCAmelCase,      safety_checker=__UpperCAmelCase		)

   pipe.to(__UpperCAmelCase		)
   pipe.set_progress_bar_config(disable=__UpperCAmelCase		)
   pipe.enable_attention_slicing()

   _snake_case							:		Tuple								=  """A black colored car"""
   _snake_case							:		List[Any]								=  """A blue colored car"""

   _snake_case							:		Tuple								=  torch.manual_seed(0		)
   _snake_case							:		Dict								=  pipe(
       prompt=__UpperCAmelCase,      source_prompt=__UpperCAmelCase,      image=__UpperCAmelCase,      num_inference_steps=100,      eta=0.1,      strength=0.85,      guidance_scale=3,      source_guidance_scale=1,      generator=__UpperCAmelCase,      output_type="""np""",      )
   _snake_case							:		int								=  output.images

   assert np.abs(image - expected_image		).max() < 2E-2



import warnings
from typing import Any, Dict, List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, optimal_fft_length, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


UpperCamelCase__											=					logging.get_logger(__name__)





class      A							(				UpperCAmelCase_     ):
   __UpperCAmelCase : int      =			['input_values', 'attention_mask']


   def __init__(self     :						Any       ,  __UpperCAmelCase     :						int = 1       ,  __UpperCAmelCase     :						int = 1_6_0_0_0       ,  __UpperCAmelCase     :						float = 0.0       ,  __UpperCAmelCase     :						bool = False       ,  __UpperCAmelCase     :						int = 8_0       ,  __UpperCAmelCase     :						int = 1_6       ,  __UpperCAmelCase     :						int = 6_4       ,  __UpperCAmelCase     :						str = "hann_window"       ,  __UpperCAmelCase     :						float = 1.0       ,  __UpperCAmelCase     :						float = 8_0       ,  __UpperCAmelCase     :						float = 7_6_0_0       ,  __UpperCAmelCase     :						float = 1E-10       ,  __UpperCAmelCase     :						int = 2       ,  __UpperCAmelCase     :						bool = True       ,  **__UpperCAmelCase     :						Any       ,  )						->				str:
         """simple docstring"""



         super().__init__(feature_size=__UpperCAmelCase       ,  sampling_rate=__UpperCAmelCase       ,  padding_value=__UpperCAmelCase       ,  **__UpperCAmelCase		)
         UpperCAmelCase__								=  do_normalize
         UpperCAmelCase__								=  return_attention_mask

         UpperCAmelCase__								=  num_mel_bins
         UpperCAmelCase__								=  hop_length
         UpperCAmelCase__								=  win_length
         UpperCAmelCase__								=  win_function
         UpperCAmelCase__								=  frame_signal_scale
         UpperCAmelCase__								=  fmin
         UpperCAmelCase__								=  fmax
         UpperCAmelCase__								=  mel_floor
         UpperCAmelCase__								=  reduction_factor

         UpperCAmelCase__								=  win_length * sampling_rate // 1_0_0_0
         UpperCAmelCase__								=  hop_length * sampling_rate // 1_0_0_0
         UpperCAmelCase__								=  optimal_fft_length(self.sample_size		)
         UpperCAmelCase__								=  (self.n_fft // 2) + 1

         UpperCAmelCase__								=  window_function(window_length=self.sample_size       ,  name=self.win_function       ,  periodic=__UpperCAmelCase		)

         UpperCAmelCase__								=  mel_filter_bank(
             num_frequency_bins=self.n_freqs       ,  num_mel_filters=self.num_mel_bins       ,  min_frequency=self.fmin       ,  max_frequency=self.fmax       ,  sampling_rate=self.sampling_rate       ,  norm="slaney"       ,  mel_scale="slaney"       ,  )

         if frame_signal_scale != 1.0:
               warnings.warn(
                   "The argument `frame_signal_scale` is deprecated and will be removed in version 4.30.0 of Transformers"       ,  __UpperCAmelCase       ,  )
         if reduction_factor != 2.0:
               warnings.warn(
                   "The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers"       ,  __UpperCAmelCase       ,  )


   @staticmethod
   # Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
   def 			lowercase_      (__UpperCAmelCase     :						List[np.ndarray]       ,  __UpperCAmelCase     :						List[np.ndarray]       ,  __UpperCAmelCase     :						float = 0.0		)						->				List[np.ndarray]:
         """simple docstring"""



         if attention_mask is not None:
               UpperCAmelCase__								=  np.array(__UpperCAmelCase       ,  np.intaa		)
               UpperCAmelCase__								=  []

               for vector, length in zip(__UpperCAmelCase       ,  attention_mask.sum(-1		)		):
                     UpperCAmelCase__								=  (vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7		)
                     if length < normed_slice.shape[0]:
                           UpperCAmelCase__								=  padding_value

                     normed_input_values.append(__UpperCAmelCase		)
         else:
               UpperCAmelCase__								=  [(x - x.mean()) / np.sqrt(x.var() + 1E-7		) for x in input_values]

         return normed_input_values


   def 			lowercase_      (self     :						Optional[int]       ,  __UpperCAmelCase     :						np.ndarray       ,  )						->				np.ndarray:
         """simple docstring"""



         UpperCAmelCase__								=  spectrogram(
             __UpperCAmelCase       ,  window=self.window       ,  frame_length=self.sample_size       ,  hop_length=self.sample_stride       ,  fft_length=self.n_fft       ,  mel_filters=self.mel_filters       ,  mel_floor=self.mel_floor       ,  log_mel="log10"       ,  )
         return log_mel_spec.T


   def __call__(self     :						Any       ,  __UpperCAmelCase     :						Optional[Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]] = None       ,  __UpperCAmelCase     :						Optional[Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]] = None       ,  __UpperCAmelCase     :						Union[bool, str, PaddingStrategy] = False       ,  __UpperCAmelCase     :						Optional[int] = None       ,  __UpperCAmelCase     :						bool = False       ,  __UpperCAmelCase     :						Optional[int] = None       ,  __UpperCAmelCase     :						Optional[bool] = None       ,  __UpperCAmelCase     :						Optional[Union[str, TensorType]] = None       ,  __UpperCAmelCase     :						Optional[int] = None       ,  **__UpperCAmelCase     :						str       ,  )						->				BatchFeature:
         """simple docstring"""



         if audio is None and audio_target is None:
               raise ValueError("You must provide either `audio` or `audio_target` values."		)

         if sampling_rate is not None:
               if sampling_rate != self.sampling_rate:
                     raise ValueError(
                         f"""The model corresponding to this feature extractor: {self} was trained using a sampling rate of"""
                         f""" {self.sampling_rate}. Please make sure that the provided audio input was sampled with"""
                         f""" {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."		)

         if audio is not None:
               UpperCAmelCase__								=  self._process_audio(
                   __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  **__UpperCAmelCase       ,  )
         else:
               UpperCAmelCase__								=  None

         if audio_target is not None:
               UpperCAmelCase__								=  self._process_audio(
                   __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  **__UpperCAmelCase       ,  )

               if inputs is None:
                     return inputs_target
               else:
                     UpperCAmelCase__								=  inputs_target["input_values"]
                     UpperCAmelCase__								=  inputs_target.get("attention_mask"		)
                     if decoder_attention_mask is not None:
                           UpperCAmelCase__								=  decoder_attention_mask

         return inputs


   def 			lowercase_      (self     :						Optional[int]       ,  __UpperCAmelCase     :						Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]       ,  __UpperCAmelCase     :						bool = False       ,  __UpperCAmelCase     :						Union[bool, str, PaddingStrategy] = False       ,  __UpperCAmelCase     :						Optional[int] = None       ,  __UpperCAmelCase     :						bool = False       ,  __UpperCAmelCase     :						Optional[int] = None       ,  __UpperCAmelCase     :						Optional[bool] = None       ,  __UpperCAmelCase     :						Optional[Union[str, TensorType]] = None       ,  **__UpperCAmelCase     :						Any       ,  )						->				BatchFeature:
         """simple docstring"""



         UpperCAmelCase__								=  isinstance(__UpperCAmelCase       ,  np.ndarray		) and len(speech.shape		) > 1
         if is_batched_numpy and len(speech.shape		) > 2:
               raise ValueError(f"""Only mono-channel audio is supported for input to {self}"""		)
         UpperCAmelCase__								=  is_batched_numpy or (
             isinstance(__UpperCAmelCase       ,  (list, tuple)		) and (isinstance(speech[0]       ,  (np.ndarray, tuple, list)		))
         )

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

         # always return batch
         if not is_batched:
               UpperCAmelCase__								=  [speech]

         # needed to make pad() work on spectrogram inputs
         UpperCAmelCase__								=  self.feature_size

         # convert into correct format for padding
         if is_target:
               UpperCAmelCase__								=  [self._extract_mel_features(__UpperCAmelCase		) for waveform in speech]
               UpperCAmelCase__								=  BatchFeature({"input_values": features}		)
               UpperCAmelCase__								=  self.num_mel_bins
         else:
               UpperCAmelCase__								=  BatchFeature({"input_values": speech}		)

         UpperCAmelCase__								=  self.pad(
             __UpperCAmelCase       ,  padding=__UpperCAmelCase       ,  max_length=__UpperCAmelCase       ,  truncation=__UpperCAmelCase       ,  pad_to_multiple_of=__UpperCAmelCase       ,  return_attention_mask=__UpperCAmelCase       ,  **__UpperCAmelCase       ,  )

         UpperCAmelCase__								=  feature_size_hack

         # convert input values to correct format
         UpperCAmelCase__								=  padded_inputs["input_values"]
         if not isinstance(input_values[0]       ,  np.ndarray		):
               UpperCAmelCase__								=  [np.asarray(__UpperCAmelCase       ,  dtype=np.floataa		) for array in input_values]
         elif (
             not isinstance(__UpperCAmelCase       ,  np.ndarray		)
             and isinstance(input_values[0]       ,  np.ndarray		)
             and input_values[0].dtype is np.dtype(np.floataa		)
         ):
               UpperCAmelCase__								=  [array.astype(np.floataa		) for array in input_values]
         elif isinstance(__UpperCAmelCase       ,  np.ndarray		) and input_values.dtype is np.dtype(np.floataa		):
               UpperCAmelCase__								=  input_values.astype(np.floataa		)

         # convert attention_mask to correct format
         UpperCAmelCase__								=  padded_inputs.get("attention_mask"		)
         if attention_mask is not None:
               UpperCAmelCase__								=  [np.asarray(__UpperCAmelCase       ,  dtype=np.intaa		) for array in attention_mask]

         # zero-mean and unit-variance normalization
         if not is_target and self.do_normalize:
               UpperCAmelCase__								=  (
                   attention_mask
                   if self._get_padding_strategies(__UpperCAmelCase       ,  max_length=__UpperCAmelCase		) is not PaddingStrategy.DO_NOT_PAD
                   else None
               )
               UpperCAmelCase__								=  self.zero_mean_unit_var_norm(
                   padded_inputs["input_values"]       ,  attention_mask=__UpperCAmelCase       ,  padding_value=self.padding_value		)

         if return_tensors is not None:
               UpperCAmelCase__								=  padded_inputs.convert_to_tensors(__UpperCAmelCase		)

         return padded_inputs





   def 			lowercase_      (self     :						Tuple		)						->				Dict[str, Any]:
         """simple docstring"""



         UpperCAmelCase__								=  super().to_dict()

         # Don't serialize these as they are derived from the other properties.
         UpperCAmelCase__								=  ["window", "mel_filters", "sample_size", "sample_stride", "n_fft", "n_freqs"]
         for name in names:
               if name in output:
                     del output[name]

         return output



