
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 			UpperCamelCase_			(						__UpperCAmelCase	,     __UpperCAmelCase	,     __UpperCAmelCase	,     unittest.TestCase   ):




							'''simple docstring'''

							UpperCamelCase_					   =							StableDiffusionImageVariationPipeline
							UpperCamelCase_					   =							IMAGE_VARIATION_PARAMS
							UpperCamelCase_					   =							IMAGE_VARIATION_BATCH_PARAMS
							UpperCamelCase_					   =							frozenset([]   )
							# TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
							UpperCamelCase_					   =							frozenset([]   )
							def        lowerCAmelCase__ (				self) ->	Any:
										torch.manual_seed(0)
										UpperCamelCase__    :		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     ,						)
										UpperCamelCase__    :		Dict     =				PNDMScheduler(skip_prk_steps=lowercase_)
										torch.manual_seed(0)
										UpperCamelCase__    :		Optional[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)
										UpperCamelCase__    :		Tuple     =				CLIPVisionConfig(
										    hidden_size=32     ,						projection_dim=32     ,						intermediate_size=37     ,						layer_norm_eps=1E-05     ,						num_attention_heads=4     ,						num_hidden_layers=5     ,						image_size=32     ,						patch_size=4     ,						)
										UpperCamelCase__    :		List[Any]     =				CLIPVisionModelWithProjection(lowercase_)
										UpperCamelCase__    :		str     =				CLIPImageProcessor(crop_size=32     ,						size=32)

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

										UpperCamelCase__    :		List[str]     =				self.get_dummy_inputs(lowercase_)
										UpperCamelCase__    :		Dict     =				sd_pipe(**lowercase_).images
										UpperCamelCase__    :		Tuple     =				image[0, -3:, -3:, -1]

										assert image.shape == (1, 64, 64, 3)
										UpperCamelCase__    :		int     =				np.array([0.5239, 0.5723, 0.4796, 0.5049, 0.5550, 0.4685, 0.5329, 0.4891, 0.4921])

										assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-3
							def        lowerCAmelCase__ (				self) ->	Optional[int]:
										UpperCamelCase__    :		str     =				'''cpu'''  # ensure determinism for the device-dependent torch.Generator
										UpperCamelCase__    :		Dict     =				self.get_dummy_components()
										UpperCamelCase__    :		Dict     =				StableDiffusionImageVariationPipeline(**lowercase_)
										UpperCamelCase__    :		Optional[int]     =				sd_pipe.to(lowercase_)
										sd_pipe.set_progress_bar_config(disable=lowercase_)

										UpperCamelCase__    :		List[str]     =				self.get_dummy_inputs(lowercase_)
										UpperCamelCase__    :		Any     =				2 * [inputs['''image''']]
										UpperCamelCase__    :		int     =				sd_pipe(**lowercase_)

										UpperCamelCase__    :		Optional[Any]     =				output.images

										UpperCamelCase__    :		Optional[Any]     =				image[-1, -3:, -3:, -1]

										assert image.shape == (2, 64, 64, 3)
										UpperCamelCase__    :		List[Any]     =				np.array([0.6892, 0.5637, 0.5836, 0.5771, 0.6254, 0.6409, 0.5580, 0.5569, 0.5289])

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






							def        lowerCAmelCase__ (				self) ->	Dict:
										super().test_inference_batch_single_identical(expected_max_diff=3E-3)



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




							'''simple docstring'''

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

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

										assert image.shape == (1, 5_12, 5_12, 3)
										UpperCamelCase__    :		List[Any]     =				np.array([0.8_4491, 0.9_0789, 0.7_5708, 0.7_8734, 0.8_3485, 0.7_0099, 0.6_6938, 0.6_8727, 0.6_1379])
										assert np.abs(image_slice - expected_slice).max() < 6E-3
							def        lowerCAmelCase__ (				self) ->	List[Any]:
										UpperCamelCase__    :		Union[str, Any]     =				0

										def callback_fn(UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase) -> None:
													UpperCamelCase__    :		Tuple     =				True
													nonlocal number_of_steps
													number_of_steps += 1
													if step == 1:
																UpperCamelCase__    :		List[Any]     =				latents.detach().cpu().numpy()
																assert latents.shape == (1, 4, 64, 64)
																UpperCamelCase__    :		Union[str, Any]     =				latents[0, -3:, -3:, -1]
																UpperCamelCase__    :		List[str]     =				np.array(
																    [-0.1621, 0.2837, -0.7979, -0.1221, -1.3057, 0.7681, -2.1191, 0.0464, 1.6309])

																assert np.abs(latents_slice.flatten() - expected_slice).max() < 5E-2
													elif step == 2:
																UpperCamelCase__    :		Optional[Any]     =				latents.detach().cpu().numpy()
																assert latents.shape == (1, 4, 64, 64)
																UpperCamelCase__    :		Dict     =				latents[0, -3:, -3:, -1]
																UpperCamelCase__    :		Any     =				np.array([0.6299, 1.7500, 1.1992, -2.1582, -1.8994, 0.7334, -0.7090, 1.0137, 1.5273])

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

										UpperCamelCase__    :		str     =				False

										UpperCamelCase__    :		Optional[Any]     =				StableDiffusionImageVariationPipeline.from_pretrained(
										    'fusing/sd-image-variations-diffusers'     ,						safety_checker=lowercase_     ,						torch_dtype=torch.floataa     ,						)
										pipe.to(lowercase_)
										pipe.set_progress_bar_config(disable=lowercase_)
										pipe.enable_attention_slicing()

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






							def        lowerCAmelCase__ (				self) ->	Tuple:
										torch.cuda.empty_cache()
										torch.cuda.reset_max_memory_allocated()
										torch.cuda.reset_peak_memory_stats()

										UpperCamelCase__    :		Optional[int]     =				'''fusing/sd-image-variations-diffusers'''
										UpperCamelCase__    :		Union[str, Any]     =				StableDiffusionImageVariationPipeline.from_pretrained(
										    lowercase_     ,						safety_checker=lowercase_     ,						torch_dtype=torch.floataa)
										UpperCamelCase__    :		Optional[int]     =				pipe.to(lowercase_)
										pipe.set_progress_bar_config(disable=lowercase_)
										pipe.enable_attention_slicing(1)
										pipe.enable_sequential_cpu_offload()

										UpperCamelCase__    :		List[Any]     =				self.get_inputs(lowercase_     ,						dtype=torch.floataa)
										UpperCamelCase__    :		Dict     =				pipe(**lowercase_)

										UpperCamelCase__    :		Tuple     =				torch.cuda.max_memory_allocated()
										# make sure that less than 2.6 GB is allocated
										assert mem_bytes < 2.6 * 10**9



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




							'''simple docstring'''

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

										UpperCamelCase__    :		Tuple     =				self.get_inputs(lowercase_)
										UpperCamelCase__    :		Optional[Any]     =				sd_pipe(**lowercase_).images[0]

										UpperCamelCase__    :		List[Any]     =				load_numpy(
										    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
										    '/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy')
										UpperCamelCase__    :		List[Any]     =				np.abs(expected_image - image).max()
										assert max_diff < 1E-3






							def        lowerCAmelCase__ (				self) ->	Dict:
										UpperCamelCase__    :		List[Any]     =				StableDiffusionImageVariationPipeline.from_pretrained('fusing/sd-image-variations-diffusers')
										UpperCamelCase__    :		Dict     =				DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config)
										sd_pipe.to(lowercase_)
										sd_pipe.set_progress_bar_config(disable=lowercase_)

										UpperCamelCase__    :		List[Any]     =				self.get_inputs(lowercase_)
										UpperCamelCase__    :		List[Any]     =				25
										UpperCamelCase__    :		Tuple     =				sd_pipe(**lowercase_).images[0]

										UpperCamelCase__    :		List[Any]     =				load_numpy(
										    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
										    '/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy')
										UpperCamelCase__    :		Any     =				np.abs(expected_image - image).max()
										assert max_diff < 1E-3





from typing import Optional

from ..utils.logging import get_logger
from .audio_classificiation import AudioClassification
from .automatic_speech_recognition import AutomaticSpeechRecognition
from .base import TaskTemplate
from .image_classification import ImageClassification
from .language_modeling import LanguageModeling
from .question_answering import QuestionAnsweringExtractive
from .summarization import Summarization
from .text_classification import TextClassification


UpperCAmelCase__				:  Tuple       					=				[
    '''AutomaticSpeechRecognition''',
    '''AudioClassification''',
    '''ImageClassification''',
    '''LanguageModeling''',
    '''QuestionAnsweringExtractive''',
    '''Summarization''',
    '''TaskTemplate''',
    '''TextClassification''',
]

UpperCAmelCase__				:  Dict       					=				get_logger(__name__)


UpperCAmelCase__				:  Dict       					=				{
    AutomaticSpeechRecognition.task: AutomaticSpeechRecognition,
    AudioClassification.task: AudioClassification,
    ImageClassification.task: ImageClassification,
    LanguageModeling.task: LanguageModeling,
    QuestionAnsweringExtractive.task: QuestionAnsweringExtractive,
    Summarization.task: Summarization,
    TextClassification.task: TextClassification,
}



def       _lowercase					(      __SCREAMING_SNAKE_CASE )	->     Optional[TaskTemplate]:
			UpperCamelCase__    :		Any     =				task_template_dict.get('task' )
			if task_name is None:
						logger.warning(F"""Couldn't find template for task '{task_name}'. Available templates: {list(__SCREAMING_SNAKE_CASE )}""" )
						return None
			UpperCamelCase__    :		Any     =				NAMEaTEMPLATE.get(__SCREAMING_SNAKE_CASE )
			return template.from_dict(__SCREAMING_SNAKE_CASE )



