

"""simple docstring"""




import gc
import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

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      _UpperCAmelCase   (							_lowerCAmelCase		,    _lowerCAmelCase		,    unittest.TestCase       ):
	a__		: Optional[Any]					=					StableDiffusionXLImgaImgPipeline
	a__		: Any					=					TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"height", "width"}
	a__		: str					=					PipelineTesterMixin.required_optional_params - {"latents"}
	a__		: Optional[int]					=					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
	a__		: str					=					IMAGE_TO_IMAGE_IMAGE_PARAMS
	a__		: str					=					IMAGE_TO_IMAGE_IMAGE_PARAMS



	def        a  (  self     :       Union[str, Any]						):
		torch.manual_seed(0						)
		__UpperCAmelCase      		= 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''')			,    attention_head_dim=(2, 4)			,    use_linear_projection=_lowercase			,    addition_embed_type='''text_time'''			,    addition_time_embed_dim=8			,    transformer_layers_per_block=(1, 2)			,    projection_class_embeddings_input_dim=80			,    cross_attention_dim=64			,    )
		__UpperCAmelCase      		= EulerDiscreteScheduler(
		    beta_start=0.00_085			,    beta_end=0.012			,    steps_offset=1			,    beta_schedule='''scaled_linear'''			,    timestep_spacing='''leading'''			,    )
		torch.manual_seed(0						)
		__UpperCAmelCase      		= 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			,    sample_size=1_28			,    )
		torch.manual_seed(0						)
		__UpperCAmelCase      		= 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=10_00			,    hidden_act='''gelu'''			,    projection_dim=32			,    )
		__UpperCAmelCase      		= CLIPTextModel(_lowercase						)
		__UpperCAmelCase      		= CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''			,    local_files_only=_lowercase						)

		__UpperCAmelCase      		= CLIPTextModelWithProjection(_lowercase						)
		__UpperCAmelCase      		= CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''			,    local_files_only=_lowercase						)

		__UpperCAmelCase      		= {
		    '''unet''': unet,
		    '''scheduler''': scheduler,
		    '''vae''': vae,
		    '''text_encoder''': text_encoder,
		    '''tokenizer''': tokenizer,
		    '''text_encoder_2''': text_encoder_a,
		    '''tokenizer_2''': tokenizer_a,
		    # "safety_checker": None,
		    # "feature_extractor": None,
		}
		return components



	def        a  (  self     :       List[Any]			,    _lowercase     :       List[str]			,    _lowercase     :       Any=0						):
		__UpperCAmelCase      		= floats_tensor((1, 3, 32, 32)			,    rng=random.Random(_lowercase						)						).to(_lowercase						)
		__UpperCAmelCase      		= image / 2 + 0.5
		if str(_lowercase						).startswith('''mps'''						):
			__UpperCAmelCase      		= torch.manual_seed(_lowercase						)
		else:
			__UpperCAmelCase      		= torch.Generator(device=_lowercase						).manual_seed(_lowercase						)
		__UpperCAmelCase      		= {
		    '''prompt''': '''A painting of a squirrel eating a burger''',
		    '''image''': image,
		    '''generator''': generator,
		    '''num_inference_steps''': 2,
		    '''guidance_scale''': 5.0,
		    '''output_type''': '''numpy''',
		    '''strength''': 0.75,
		}
		return inputs



	def        a  (  self     :       str						):
		__UpperCAmelCase      		= '''cpu'''  # ensure determinism for the device-dependent torch.Generator
		__UpperCAmelCase      		= self.get_dummy_components()
		__UpperCAmelCase      		= StableDiffusionXLImgaImgPipeline(**_lowercase						)
		__UpperCAmelCase      		= sd_pipe.to(_lowercase						)
		sd_pipe.set_progress_bar_config(disable=_lowercase						)

		__UpperCAmelCase      		= self.get_dummy_inputs(_lowercase						)
		__UpperCAmelCase      		= sd_pipe(**_lowercase						).images
		__UpperCAmelCase      		= image[0, -3:, -3:, -1]

		assert image.shape == (1, 32, 32, 3)

		__UpperCAmelCase      		= np.array([0.4_656, 0.4_840, 0.4_439, 0.6_698, 0.5_574, 0.4_524, 0.5_799, 0.5_943, 0.5_165]						)

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



	def        a  (  self     :       List[Any]						):
		super().test_attention_slicing_forward_pass(expected_max_diff=3E-3						)



	def        a  (  self     :       int						):
		super().test_inference_batch_single_identical(expected_max_diff=3E-3						)



	def        a  (  self     :       Optional[int]						):
		pass



	def        a  (  self     :       int						):
		__UpperCAmelCase      		= self.get_dummy_components()
		__UpperCAmelCase      		= StableDiffusionXLImgaImgPipeline(**_lowercase						)
		__UpperCAmelCase      		= sd_pipe.to(_lowercase						)
		__UpperCAmelCase      		= sd_pipe.to(_lowercase						)
		sd_pipe.set_progress_bar_config(disable=_lowercase						)

		# forward without prompt embeds
		__UpperCAmelCase      		= self.get_dummy_inputs(_lowercase						)
		__UpperCAmelCase      		= 3 * ['''this is a negative prompt''']
		__UpperCAmelCase      		= negative_prompt
		__UpperCAmelCase      		= 3 * [inputs['''prompt''']]

		__UpperCAmelCase      		= sd_pipe(**_lowercase						)
		__UpperCAmelCase      		= output.images[0, -3:, -3:, -1]

		# forward with prompt embeds
		__UpperCAmelCase      		= self.get_dummy_inputs(_lowercase						)
		__UpperCAmelCase      		= 3 * ['''this is a negative prompt''']
		__UpperCAmelCase      		= 3 * [inputs.pop('''prompt'''						)]

		(
		    (
		    __UpperCAmelCase
		)		,							(
		    __UpperCAmelCase
		)		,							(
		    __UpperCAmelCase
		)		,							(
		    __UpperCAmelCase
		)		,							
		)      		= sd_pipe.encode_prompt(_lowercase			,    negative_prompt=_lowercase						)

		__UpperCAmelCase      		= sd_pipe(
		    **_lowercase			,    prompt_embeds=_lowercase			,    negative_prompt_embeds=_lowercase			,    pooled_prompt_embeds=_lowercase			,    negative_pooled_prompt_embeds=_lowercase			,    )
		__UpperCAmelCase      		= output.images[0, -3:, -3:, -1]

		# make sure that it's equal
		assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()						).max() < 1E-4






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



	def        a  (  self     :       Tuple						):
		super().tearDown()
		gc.collect()
		torch.cuda.empty_cache()



	def        a  (  self     :       Optional[int]			,    _lowercase     :       Optional[Any]			,    _lowercase     :       int="cpu"			,    _lowercase     :       List[str]=torch.floataa			,    _lowercase     :       Any=0						):
		__UpperCAmelCase      		= torch.Generator(device=_lowercase						).manual_seed(_lowercase						)
		__UpperCAmelCase      		= np.random.RandomState(_lowercase						).standard_normal((1, 4, 64, 64)						)
		__UpperCAmelCase      		= torch.from_numpy(_lowercase						).to(device=_lowercase			,    dtype=_lowercase						)
		__UpperCAmelCase      		= {
		    '''prompt''': '''a photograph of an astronaut riding a horse''',
		    '''latents''': latents,
		    '''generator''': generator,
		    '''num_inference_steps''': 3,
		    '''guidance_scale''': 7.5,
		    '''output_type''': '''numpy''',
		}
		return inputs



	def        a  (  self     :       Tuple						):
		__UpperCAmelCase      		= DiffusionPipeline.from_pretrained('''stabilityai/stable-diffusion-2-base'''						)
		pipe.to(_lowercase						)
		pipe.set_progress_bar_config(disable=_lowercase						)

		__UpperCAmelCase      		= self.get_inputs(_lowercase						)
		__UpperCAmelCase      		= pipe(**_lowercase						).images
		__UpperCAmelCase      		= image[0, -3:, -3:, -1].flatten()

		assert image.shape == (1, 5_12, 5_12, 3)
		__UpperCAmelCase      		= np.array([0.49_493, 0.47_896, 0.40_798, 0.54_214, 0.53_212, 0.48_202, 0.47_656, 0.46_329, 0.48_506]						)
		assert np.abs(image_slice - expected_slice						).max() < 7E-3








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


lowercase    				=   logging.get_logger(__name__)




class 					UpperCamelCase_ (     snake_case_							):



	'''simple docstring'''




	lowerCAmelCase						=					['''input_features''', '''is_longer''']




	def __init__(			self			,     a=64			,     a=4_80_00			,     a=4_80			,     a=10			,     a=10_24			,     a=0.0			,     a=False			,     a = 0			,     a = 1_40_00			,     a = None			,     a = "fusion"			,     a = "repeatpad"			,     **a			,     )			->       List[str]:
								super().__init__(
								    feature_size=a			,     sampling_rate=a			,     padding_value=a			,     return_attention_mask=a			,     **a			,     )
								snake_case_           =				top_db
								snake_case_           =				truncation
								snake_case_           =				padding
								snake_case_           =				fft_window_size
								snake_case_           =				(fft_window_size >> 1) + 1
								snake_case_           =				hop_length
								snake_case_           =				max_length_s
								snake_case_           =				max_length_s * sampling_rate
								snake_case_           =				sampling_rate
								snake_case_           =				frequency_min
								snake_case_           =				frequency_max
								snake_case_           =				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'			,     )
								snake_case_           =				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 				_UpperCamelCase     (			self   )			->       Dict[str, Any]:
								snake_case_           =				copy.deepcopy(self.__dict__   )
								snake_case_           =				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			,     a			,     a = None   )			->       np.ndarray:
								snake_case_           =				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 				_UpperCamelCase     (			self			,     a			,     a			,     a   )			->       Tuple:
								snake_case_           =				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
															snake_case_           =				[0]
								if len(ranges[2]   ) == 0:
															# if the audio is too short, we just use the first chunk
															snake_case_           =				[0]
								# randomly choose index for each part
								snake_case_           =				np.random.choice(ranges[0]   )
								snake_case_           =				np.random.choice(ranges[1]   )
								snake_case_           =				np.random.choice(ranges[2]   )

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

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




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

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

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

								return input_mel, longer






	def __call__(			self			,     a			,     a = None			,     a = None			,     a = None			,     a = None			,     a = None			,     **a			,     )			->       BatchFeature:
								snake_case_           =				truncation if truncation is not None else self.truncation
								snake_case_           =				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.'   )

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

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

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

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

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

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

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

								snake_case_           =				{'input_features': input_mel, 'is_longer': is_longer}
								snake_case_           =				BatchFeature(a   )

								if return_tensors is not None:
															snake_case_           =				input_features.convert_to_tensors(a   )

								return input_features





