
import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModelWithProjection

from diffusers import (
    AutoencoderKL,
    DPMSolverMultistepScheduler,
    PNDMScheduler,
    StableDiffusionImageVariationPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, nightly, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()
class 			lowercase__(    UpperCAmelCase			,						UpperCAmelCase			,						UpperCAmelCase			,						unittest.TestCase							):



	"""simple docstring"""




	a      :int   					=     StableDiffusionImageVariationPipeline
	a      :Tuple   					=     IMAGE_VARIATION_PARAMS
	a      :str   					=     IMAGE_VARIATION_BATCH_PARAMS
	a      :Tuple   					=     frozenset([]							)
	# TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
	a      :Dict   					=     frozenset([]							)
	def 						_lowercase  (	self  :      Any	)	->					int:
					torch.manual_seed(0	)
					lowercase_					  = UNetaDConditionModel(
					    block_out_channels=(3_2, 6_4)	,							layers_per_block=2	,							sample_size=3_2	,							in_channels=4	,							out_channels=4	,							down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')	,							up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')	,							cross_attention_dim=3_2	,							)
					lowercase_					  = PNDMScheduler(skip_prk_steps=SCREAMING_SNAKE_CASE_	)
					torch.manual_seed(0	)
					lowercase_					  = AutoencoderKL(
					    block_out_channels=[3_2, 6_4]	,							in_channels=3	,							out_channels=3	,							down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']	,							up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']	,							latent_channels=4	,							)
					torch.manual_seed(0	)
					lowercase_					  = CLIPVisionConfig(
					    hidden_size=3_2	,							projection_dim=3_2	,							intermediate_size=3_7	,							layer_norm_eps=1e-05	,							num_attention_heads=4	,							num_hidden_layers=5	,							image_size=3_2	,							patch_size=4	,							)
					lowercase_					  = CLIPVisionModelWithProjection(SCREAMING_SNAKE_CASE_	)
					lowercase_					  = CLIPImageProcessor(crop_size=3_2	,							size=3_2	)

					lowercase_					  = {
					    '''unet''': unet,
					    '''scheduler''': scheduler,
					    '''vae''': vae,
					    '''image_encoder''': image_encoder,
					    '''feature_extractor''': feature_extractor,
					    '''safety_checker''': None,
					}
					return components
	def 						_lowercase  (	self  :      str	,							SCREAMING_SNAKE_CASE_  :      List[Any]	,							SCREAMING_SNAKE_CASE_  :      Dict=0	)	->					List[str]:
					lowercase_					  = floats_tensor((1, 3, 3_2, 3_2)	,							rng=random.Random(SCREAMING_SNAKE_CASE_	)	)
					lowercase_					  = image.cpu().permute(0	,							2	,							3	,							1	)[0]
					lowercase_					  = Image.fromarray(np.uinta(SCREAMING_SNAKE_CASE_	)	).convert('''RGB'''	).resize((3_2, 3_2)	)
					if str(SCREAMING_SNAKE_CASE_	).startswith('''mps'''	):
									lowercase_					  = torch.manual_seed(SCREAMING_SNAKE_CASE_	)
					else:
									lowercase_					  = torch.Generator(device=SCREAMING_SNAKE_CASE_	).manual_seed(SCREAMING_SNAKE_CASE_	)
					lowercase_					  = {
					    '''image''': image,
					    '''generator''': generator,
					    '''num_inference_steps''': 2,
					    '''guidance_scale''': 6.0,
					    '''output_type''': '''numpy''',
					}
					return inputs
	def 						_lowercase  (	self  :      Optional[int]	)	->					Optional[int]:
					lowercase_					  = '''cpu'''  # ensure determinism for the device-dependent torch.Generator
					lowercase_					  = self.get_dummy_components()
					lowercase_					  = StableDiffusionImageVariationPipeline(**SCREAMING_SNAKE_CASE_	)
					lowercase_					  = sd_pipe.to(SCREAMING_SNAKE_CASE_	)
					sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_	)

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

					assert image.shape == (1, 6_4, 6_4, 3)
					lowercase_					  = np.array([0.52_39, 0.57_23, 0.47_96, 0.50_49, 0.55_50, 0.46_85, 0.53_29, 0.48_91, 0.49_21]	)

					assert np.abs(image_slice.flatten() - expected_slice	).max() < 1e-3
	def 						_lowercase  (	self  :      Tuple	)	->					Optional[Any]:
					lowercase_					  = '''cpu'''  # ensure determinism for the device-dependent torch.Generator
					lowercase_					  = self.get_dummy_components()
					lowercase_					  = StableDiffusionImageVariationPipeline(**SCREAMING_SNAKE_CASE_	)
					lowercase_					  = sd_pipe.to(SCREAMING_SNAKE_CASE_	)
					sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_	)

					lowercase_					  = self.get_dummy_inputs(SCREAMING_SNAKE_CASE_	)
					lowercase_					  = 2 * [inputs['''image''']]
					lowercase_					  = sd_pipe(**SCREAMING_SNAKE_CASE_	)

					lowercase_					  = output.images

					lowercase_					  = image[-1, -3:, -3:, -1]

					assert image.shape == (2, 6_4, 6_4, 3)
					lowercase_					  = np.array([0.68_92, 0.56_37, 0.58_36, 0.57_71, 0.62_54, 0.64_09, 0.55_80, 0.55_69, 0.52_89]	)

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




	def 						_lowercase  (	self  :      Dict	)	->					Union[str, Any]:
					super().test_inference_batch_single_identical(expected_max_diff=3e-3	)

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



	"""simple docstring"""




	def 						_lowercase  (	self  :      List[str]	)	->					List[str]:
					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()
	def 						_lowercase  (	self  :      int	,							SCREAMING_SNAKE_CASE_  :      Dict	,							SCREAMING_SNAKE_CASE_  :      Optional[Any]="cpu"	,							SCREAMING_SNAKE_CASE_  :      Optional[int]=torch.floataa	,							SCREAMING_SNAKE_CASE_  :      List[Any]=0	)	->					int:
					lowercase_					  = torch.Generator(device=SCREAMING_SNAKE_CASE_	).manual_seed(SCREAMING_SNAKE_CASE_	)
					lowercase_					  = load_image(
					    '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'''
					    '''/stable_diffusion_imgvar/input_image_vermeer.png'''	)
					lowercase_					  = np.random.RandomState(SCREAMING_SNAKE_CASE_	).standard_normal((1, 4, 6_4, 6_4)	)
					lowercase_					  = torch.from_numpy(SCREAMING_SNAKE_CASE_	).to(device=SCREAMING_SNAKE_CASE_	,							dtype=SCREAMING_SNAKE_CASE_	)
					lowercase_					  = {
					    '''image''': init_image,
					    '''latents''': latents,
					    '''generator''': generator,
					    '''num_inference_steps''': 3,
					    '''guidance_scale''': 7.5,
					    '''output_type''': '''numpy''',
					}
					return inputs
	def 						_lowercase  (	self  :      int	)	->					str:
					lowercase_					  = StableDiffusionImageVariationPipeline.from_pretrained(
					    '''lambdalabs/sd-image-variations-diffusers'''	,							safety_checker=SCREAMING_SNAKE_CASE_	)
					lowercase_					  = sd_pipe.to(SCREAMING_SNAKE_CASE_	)
					sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_	)

					lowercase_					  = self.get_inputs(SCREAMING_SNAKE_CASE_	)
					lowercase_					  = sd_pipe(**SCREAMING_SNAKE_CASE_	).images
					lowercase_					  = image[0, -3:, -3:, -1].flatten()

					assert image.shape == (1, 5_1_2, 5_1_2, 3)
					lowercase_					  = np.array([0.8_44_91, 0.9_07_89, 0.7_57_08, 0.7_87_34, 0.8_34_85, 0.7_00_99, 0.6_69_38, 0.6_87_27, 0.6_13_79]	)
					assert np.abs(image_slice - expected_slice	).max() < 6e-3
	def 						_lowercase  (	self  :      Tuple	)	->					Union[str, Any]:
					lowercase_					  = 0

					def callback_fn(SCREAMING_SNAKE_CASE_  :      int	,							SCREAMING_SNAKE_CASE_  :      int	,							SCREAMING_SNAKE_CASE_  :      torch.FloatTensor	) -> None:
									lowercase_					  = True
									nonlocal number_of_steps
									number_of_steps += 1
									if step == 1:
													lowercase_					  = latents.detach().cpu().numpy()
													assert latents.shape == (1, 4, 6_4, 6_4)
													lowercase_					  = latents[0, -3:, -3:, -1]
													lowercase_					  = np.array(
													    [-0.16_21, 0.28_37, -0.79_79, -0.12_21, -1.30_57, 0.76_81, -2.11_91, 0.04_64, 1.63_09]	)

													assert np.abs(latents_slice.flatten() - expected_slice	).max() < 5e-2
									elif step == 2:
													lowercase_					  = latents.detach().cpu().numpy()
													assert latents.shape == (1, 4, 6_4, 6_4)
													lowercase_					  = latents[0, -3:, -3:, -1]
													lowercase_					  = np.array([0.62_99, 1.75_00, 1.19_92, -2.15_82, -1.89_94, 0.73_34, -0.70_90, 1.01_37, 1.52_73]	)

													assert np.abs(latents_slice.flatten() - expected_slice	).max() < 5e-2

					lowercase_					  = False

					lowercase_					  = StableDiffusionImageVariationPipeline.from_pretrained(
					    '''fusing/sd-image-variations-diffusers'''	,							safety_checker=SCREAMING_SNAKE_CASE_	,							torch_dtype=torch.floataa	,							)
					pipe.to(SCREAMING_SNAKE_CASE_	)
					pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_	)
					pipe.enable_attention_slicing()

					lowercase_					  = self.get_inputs(SCREAMING_SNAKE_CASE_	,							dtype=torch.floataa	)
					pipe(**SCREAMING_SNAKE_CASE_	,							callback=SCREAMING_SNAKE_CASE_	,							callback_steps=1	)
					assert callback_fn.has_been_called
					assert number_of_steps == inputs["num_inference_steps"]




	def 						_lowercase  (	self  :      Union[str, Any]	)	->					List[Any]:
					torch.cuda.empty_cache()
					torch.cuda.reset_max_memory_allocated()
					torch.cuda.reset_peak_memory_stats()

					lowercase_					  = '''fusing/sd-image-variations-diffusers'''
					lowercase_					  = StableDiffusionImageVariationPipeline.from_pretrained(
					    SCREAMING_SNAKE_CASE_	,							safety_checker=SCREAMING_SNAKE_CASE_	,							torch_dtype=torch.floataa	)
					lowercase_					  = pipe.to(SCREAMING_SNAKE_CASE_	)
					pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_	)
					pipe.enable_attention_slicing(1	)
					pipe.enable_sequential_cpu_offload()

					lowercase_					  = self.get_inputs(SCREAMING_SNAKE_CASE_	,							dtype=torch.floataa	)
					lowercase_					  = pipe(**SCREAMING_SNAKE_CASE_	)

					lowercase_					  = torch.cuda.max_memory_allocated()
					# make sure that less than 2.6 GB is allocated
					assert mem_bytes < 2.6 * 1_0**9

@nightly
@require_torch_gpu
class 			lowercase__(    unittest.TestCase							):



	"""simple docstring"""




	def 						_lowercase  (	self  :      Tuple	)	->					Any:
					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()
	def 						_lowercase  (	self  :      Dict	,							SCREAMING_SNAKE_CASE_  :      Union[str, Any]	,							SCREAMING_SNAKE_CASE_  :      Union[str, Any]="cpu"	,							SCREAMING_SNAKE_CASE_  :      Dict=torch.floataa	,							SCREAMING_SNAKE_CASE_  :      List[str]=0	)	->					Optional[Any]:
					lowercase_					  = torch.Generator(device=SCREAMING_SNAKE_CASE_	).manual_seed(SCREAMING_SNAKE_CASE_	)
					lowercase_					  = load_image(
					    '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'''
					    '''/stable_diffusion_imgvar/input_image_vermeer.png'''	)
					lowercase_					  = np.random.RandomState(SCREAMING_SNAKE_CASE_	).standard_normal((1, 4, 6_4, 6_4)	)
					lowercase_					  = torch.from_numpy(SCREAMING_SNAKE_CASE_	).to(device=SCREAMING_SNAKE_CASE_	,							dtype=SCREAMING_SNAKE_CASE_	)
					lowercase_					  = {
					    '''image''': init_image,
					    '''latents''': latents,
					    '''generator''': generator,
					    '''num_inference_steps''': 5_0,
					    '''guidance_scale''': 7.5,
					    '''output_type''': '''numpy''',
					}
					return inputs
	def 						_lowercase  (	self  :      Any	)	->					Union[str, Any]:
					lowercase_					  = StableDiffusionImageVariationPipeline.from_pretrained('''fusing/sd-image-variations-diffusers'''	)
					sd_pipe.to(SCREAMING_SNAKE_CASE_	)
					sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_	)

					lowercase_					  = self.get_inputs(SCREAMING_SNAKE_CASE_	)
					lowercase_					  = sd_pipe(**SCREAMING_SNAKE_CASE_	).images[0]

					lowercase_					  = load_numpy(
					    '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'''
					    '''/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy'''	)
					lowercase_					  = np.abs(expected_image - image	).max()
					assert max_diff < 1e-3




	def 						_lowercase  (	self  :      Union[str, Any]	)	->					Union[str, Any]:
					lowercase_					  = StableDiffusionImageVariationPipeline.from_pretrained('''fusing/sd-image-variations-diffusers'''	)
					lowercase_					  = DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config	)
					sd_pipe.to(SCREAMING_SNAKE_CASE_	)
					sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_	)

					lowercase_					  = self.get_inputs(SCREAMING_SNAKE_CASE_	)
					lowercase_					  = 2_5
					lowercase_					  = sd_pipe(**SCREAMING_SNAKE_CASE_	).images[0]

					lowercase_					  = load_numpy(
					    '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'''
					    '''/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy'''	)
					lowercase_					  = np.abs(expected_image - image	).max()
					assert max_diff < 1e-3






from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_sentencepiece_available,
    is_speech_available,
    is_torch_available,
)


__a        = {
    'configuration_speech_to_text_2': ['SPEECH_TO_TEXT_2_PRETRAINED_CONFIG_ARCHIVE_MAP', 'Speech2Text2Config'],
    'processing_speech_to_text_2': ['Speech2Text2Processor'],
    'tokenization_speech_to_text_2': ['Speech2Text2Tokenizer'],
}


try:
						if not is_torch_available():
												raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
						pass
else:
						__a        = [
						    'SPEECH_TO_TEXT_2_PRETRAINED_MODEL_ARCHIVE_LIST',
						    'Speech2Text2ForCausalLM',
						    'Speech2Text2PreTrainedModel',
						]


if TYPE_CHECKING:
						from .configuration_speech_to_text_a import SPEECH_TO_TEXT_2_PRETRAINED_CONFIG_ARCHIVE_MAP, SpeechaTextaConfig
						from .processing_speech_to_text_a import SpeechaTextaProcessor
						from .tokenization_speech_to_text_a import SpeechaTextaTokenizer

						try:
												if not is_torch_available():
																		raise OptionalDependencyNotAvailable()
						except OptionalDependencyNotAvailable:
												pass
						else:
												from .modeling_speech_to_text_a import (
												    SPEECH_TO_TEXT_2_PRETRAINED_MODEL_ARCHIVE_LIST,
												    SpeechaTextaForCausalLM,
												    SpeechaTextaPreTrainedModel,
												)

else:
						import sys

						__a        = _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)




