




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     A(		UpperCamelCase				, UpperCamelCase				, UpperCamelCase				, unittest.TestCase  ):



 '''simple docstring'''


 UpperCamelCase		  =		StableUnCLIPImgaImgPipeline
 UpperCamelCase		  =		TEXT_GUIDED_IMAGE_VARIATION_PARAMS
 UpperCamelCase		  =		TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
 UpperCamelCase		  =		frozenset(
     []  )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
 UpperCamelCase		  =		frozenset([]  )

 def 							a__					(							self   :    List[Any]    )  ->      List[str]:

       """simple docstring"""






       lowerCamelCase_							=       32
       lowerCamelCase_							=       embedder_hidden_size

       # image encoding components

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

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

       # regular denoising components

       torch.manual_seed(0    )
       lowerCamelCase_							=       StableUnCLIPImageNormalizer(embedding_dim=A_    )
       lowerCamelCase_							=       DDPMScheduler(beta_schedule='squaredcos_cap_v2'    )

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

       torch.manual_seed(0    )
       lowerCamelCase_							=       CLIPTextModel(
           CLIPTextConfig(
               bos_token_id=0		,				eos_token_id=2		,				hidden_size=A_		,				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    )
       lowerCamelCase_							=       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=A_		,				layers_per_block=1		,				upcast_attention=A_		,				use_linear_projection=A_		,				)

       torch.manual_seed(0    )
       lowerCamelCase_							=       DDIMScheduler(
           beta_schedule='scaled_linear'		,				beta_start=0.00085		,				beta_end=0.012		,				prediction_type='v_prediction'		,				set_alpha_to_one=A_		,				steps_offset=1		,				)

       torch.manual_seed(0    )
       lowerCamelCase_							=       AutoencoderKL()

       lowerCamelCase_							=       {
           # 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 							a__					(							self   :    List[Any]		,				A_   :    Any		,				A_   :    List[str]=0		,				A_   :    int=True    )  ->      List[Any]:

       """simple docstring"""






       if str(A_    ).startswith('mps'    ):
             lowerCamelCase_							=       torch.manual_seed(A_    )
       else:
             lowerCamelCase_							=       torch.Generator(device=A_    ).manual_seed(A_    )

       lowerCamelCase_							=       floats_tensor((1, 3, 32, 32)		,				rng=random.Random(A_    )    ).to(A_    )

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

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

 @skip_mps
 def 							a__					(							self   :    Optional[Any]    )  ->      Optional[Any]:

       """simple docstring"""






       lowerCamelCase_							=       'cpu'  # ensure determinism for the device-dependent torch.Generator
       lowerCamelCase_							=       self.get_dummy_components()
       lowerCamelCase_							=       StableUnCLIPImgaImgPipeline(**A_    )
       lowerCamelCase_							=       sd_pipe.to(A_    )
       sd_pipe.set_progress_bar_config(disable=A_    )

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

       assert image.shape == (1, 32, 32, 3)
       lowerCamelCase_							=       np.array([0.3872, 0.7224, 0.5601, 0.4741, 0.6872, 0.5814, 0.4636, 0.3867, 0.5078]    )

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

 def 							a__					(							self   :    Union[str, Any]    )  ->      int:

       """simple docstring"""






       lowerCamelCase_							=       torch_device in ['cpu', 'mps']

       self._test_attention_slicing_forward_pass(test_max_difference=A_    )

 def 							a__					(							self   :    Optional[Any]    )  ->      int:

       """simple docstring"""






       lowerCamelCase_							=       torch_device in ['cpu', 'mps']

       self._test_inference_batch_single_identical(test_max_difference=A_    )





 @unittest.skipIf(
     torch_device != 'cuda' or not is_xformers_available()		,				reason='XFormers attention is only available with CUDA and `xformers` installed'		,				)
 def 							a__					(							self   :    List[Any]    )  ->      Optional[int]:

       """simple docstring"""






       self._test_xformers_attention_forwardGenerator_pass(test_max_difference=A_    )







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



 '''simple docstring'''


 def 							a__					(							self   :    Optional[Any]    )  ->      List[str]:

       """simple docstring"""






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

 def 							a__					(							self   :    Any    )  ->      Any:

       """simple docstring"""






       lowerCamelCase_							=       load_image(
           'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'    )

       lowerCamelCase_							=       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'    )

       lowerCamelCase_							=       StableUnCLIPImgaImgPipeline.from_pretrained(
           'fusing/stable-unclip-2-1-l-img2img'		,				torch_dtype=torch.floataa    )
       pipe.to(A_    )
       pipe.set_progress_bar_config(disable=A_    )
       # 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()

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

       lowerCamelCase_							=       output.images[0]

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

       assert_mean_pixel_difference(A_		,				A_    )

 def 							a__					(							self   :    Optional[Any]    )  ->      Tuple:

       """simple docstring"""






       lowerCamelCase_							=       load_image(
           'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'    )

       lowerCamelCase_							=       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'    )

       lowerCamelCase_							=       StableUnCLIPImgaImgPipeline.from_pretrained(
           'fusing/stable-unclip-2-1-h-img2img'		,				torch_dtype=torch.floataa    )
       pipe.to(A_    )
       pipe.set_progress_bar_config(disable=A_    )
       # 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()

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

       lowerCamelCase_							=       output.images[0]

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

       assert_mean_pixel_difference(A_		,				A_    )





 def 							a__					(							self   :    int    )  ->      Tuple:

       """simple docstring"""






       lowerCamelCase_							=       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()

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

       lowerCamelCase_							=       pipe(
           A_		,				'anime turtle'		,				num_inference_steps=2		,				output_type='np'		,				)

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










from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


__lowerCamelCase			:	List[str]					    =	logging.get_logger(__name__)



class    A__  (			__snake_case     ):
      _UpperCAmelCase      :List[str]		       =  ['input_features', 'attention_mask']





      def __init__(					self			,				A_=80			,				A_=1_6000			,				A_=80			,				A_=0.0			,				A_=True			,				A_=True			,				A_=True			,				**A_			,				):
       '''simple docstring'''




       super().__init__(feature_size=A_			,				sampling_rate=A_			,				padding_value=A_			,				**A_  )
       UpperCamelCase      :   List[Any]			      =  num_mel_bins
       UpperCamelCase      :   Any			      =  do_ceptral_normalize
       UpperCamelCase      :   int			      =  normalize_means
       UpperCamelCase      :   Tuple			      =  normalize_vars
       UpperCamelCase      :   List[Any]			      =  True





      def 			__UpperCamelCase(					self			,				A_			,				):
       '''simple docstring'''




       UpperCamelCase      :   Optional[int]			      =  waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
       UpperCamelCase      :   Optional[Any]			      =  torch.from_numpy(A_  ).unsqueeze(0  )
       UpperCamelCase      :   Dict			      =  ta_kaldi.fbank(A_			,				num_mel_bins=self.num_mel_bins			,				sample_frequency=self.sampling_rate  )
       return features.numpy()





      @staticmethod
      def 			__UpperCamelCase(					A_			,				A_			,				A_ = True			,				A_ = True			,				A_ = 0.0			,				):
       '''simple docstring'''




       if normalize_means:
        UpperCamelCase      :   str			      =  x[:input_length].mean(axis=0  )
        UpperCamelCase      :   Any			      =  np.subtract(A_			,				A_  )
       if normalize_vars:
        UpperCamelCase      :   List[str]			      =  x[:input_length].std(axis=0  )
        UpperCamelCase      :   Optional[int]			      =  np.divide(A_			,				A_  )

       if input_length < x.shape[0]:
        UpperCamelCase      :   Dict			      =  padding_value

       # make sure array is in float32
       UpperCamelCase      :   Union[str, Any]			      =  x.astype(np.floataa  )

       return x





      def 			__UpperCamelCase(					self			,				A_			,				A_ = None  ):
       '''simple docstring'''




       UpperCamelCase      :   Optional[Any]			      =  attention_mask.sum(-1  ) if attention_mask is not None else [x.shape[0] for x in input_features]
       return [
           self.utterance_cmvn(A_			,				A_			,				self.normalize_means			,				self.normalize_vars			,				self.padding_value  )
           for x, n in zip(A_			,				A_  )
       ]





      def __call__(					self			,				A_			,				A_ = False			,				A_ = None			,				A_ = False			,				A_ = None			,				A_ = None			,				A_ = None			,				A_ = None			,				**A_			,				):
       '''simple docstring'''




       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 `raw_speech` 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."  )

       UpperCamelCase      :   str			      =  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}"""  )
       UpperCamelCase      :   List[Any]			      =  is_batched_numpy or (
           isinstance(A_			,				(list, tuple)  ) and (isinstance(raw_speech[0]			,				(np.ndarray, tuple, list)  ))
       )

       if is_batched:
        UpperCamelCase      :   List[str]			      =  [np.asarray(A_			,				dtype=np.floataa  ) for speech in raw_speech]
       elif not is_batched and not isinstance(A_			,				np.ndarray  ):
        UpperCamelCase      :   int			      =  np.asarray(A_			,				dtype=np.floataa  )
       elif isinstance(A_			,				np.ndarray  ) and raw_speech.dtype is np.dtype(np.floataa  ):
        UpperCamelCase      :   Union[str, Any]			      =  raw_speech.astype(np.floataa  )

       # always return batch
       if not is_batched:
        UpperCamelCase      :   Optional[int]			      =  [raw_speech]

       # extract fbank features
       UpperCamelCase      :   Optional[int]			      =  [self._extract_fbank_features(A_  ) for waveform in raw_speech]

       # convert into correct format for padding
       UpperCamelCase      :   List[Any]			      =  BatchFeature({"input_features": features}  )

       UpperCamelCase      :   Any			      =  self.pad(
           A_			,				padding=A_			,				max_length=A_			,				truncation=A_			,				pad_to_multiple_of=A_			,				return_attention_mask=A_			,				**A_			,				)

       # make sure list is in array format
       UpperCamelCase      :   Optional[int]			      =  padded_inputs.get("input_features"  )
       if isinstance(input_features[0]			,				A_  ):
        UpperCamelCase      :   str			      =  [np.asarray(A_			,				dtype=np.floataa  ) for feature in input_features]

       UpperCamelCase      :   Optional[int]			      =  padded_inputs.get("attention_mask"  )
       if attention_mask is not None:
        UpperCamelCase      :   Optional[int]			      =  [np.asarray(A_			,				dtype=np.intaa  ) for array in attention_mask]

       # Utterance-level cepstral mean and variance normalization
       if self.do_ceptral_normalize:
        UpperCamelCase      :   List[Any]			      =  (
            np.array(A_			,				dtype=np.intaa  )
            if self._get_padding_strategies(A_			,				max_length=A_  ) is not PaddingStrategy.DO_NOT_PAD
            else None
        )
        UpperCamelCase      :   int			      =  self.normalize(
            padded_inputs["input_features"]			,				attention_mask=A_  )

       if return_tensors is not None:
        UpperCamelCase      :   List[str]			      =  padded_inputs.convert_to_tensors(A_  )

       return padded_inputs
