
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 							UpperCAmelCase				(  __SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      unittest.TestCase  ):




	'''simple docstring'''





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






	def     UpperCamelCase_  (	self :			List[Any]			):
					__A             =			32
					__A             =			embedder_hidden_size

					# image encoding components

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

					torch.manual_seed(0			)
					__A             =			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			)
					__A             =			StableUnCLIPImageNormalizer(embedding_dim=A			)
					__A             =			DDPMScheduler(beta_schedule="squaredcos_cap_v2"			)

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

					torch.manual_seed(0			)
					__A             =			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=10_00							,)			)

					torch.manual_seed(0			)
					__A             =			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			)
					__A             =			DDIMScheduler(
					    beta_schedule="scaled_linear"							,beta_start=0.0_00_85							,beta_end=0.0_12							,prediction_type="v_prediction"							,set_alpha_to_one=A							,steps_offset=1							,)

					torch.manual_seed(0			)
					__A             =			AutoencoderKL()

					__A             =			{
					    # 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     UpperCamelCase_  (	self :			str							,A :			Optional[int]							,A :			List[Any]=0							,A :			Any=True			):
					if str(A			).startswith("mps"			):
									__A             =			torch.manual_seed(A			)
					else:
									__A             =			torch.Generator(device=A			).manual_seed(A			)

					__A             =			floats_tensor((1, 3, 32, 32)							,rng=random.Random(A			)			).to(A			)

					if pil_image:
									__A             =			input_image * 0.5 + 0.5
									__A             =			input_image.clamp(0							,1			)
									__A             =			input_image.cpu().permute(0							,2							,3							,1			).float().numpy()
									__A             =			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     UpperCamelCase_  (	self :			Any			):
					__A             =			"cpu"  # ensure determinism for the device-dependent torch.Generator
					__A             =			self.get_dummy_components()
					__A             =			StableUnCLIPImgaImgPipeline(**A			)
					__A             =			sd_pipe.to(A			)
					sd_pipe.set_progress_bar_config(disable=A			)

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

					assert image.shape == (1, 32, 32, 3)
					__A             =			np.array([0.38_72, 0.72_24, 0.56_01, 0.47_41, 0.68_72, 0.58_14, 0.46_36, 0.38_67, 0.50_78]			)

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






	def     UpperCamelCase_  (	self :			List[Any]			):
					__A             =			torch_device in ["cpu", "mps"]

					self._test_attention_slicing_forward_pass(test_max_difference=A			)






	def     UpperCamelCase_  (	self :			Dict			):
					__A             =			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     UpperCamelCase_  (	self :			Optional[int]			):
					self._test_xformers_attention_forwardGenerator_pass(test_max_difference=A			)





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




	'''simple docstring'''





	def     UpperCamelCase_  (	self :			str			):
					# clean up the VRAM after each test
					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()






	def     UpperCamelCase_  (	self :			Dict			):
					__A             =			load_image(
					    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"			)

					__A             =			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"			)

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

					__A             =			torch.Generator(device="cpu"			).manual_seed(0			)
					__A             =			pipe(A							,"anime turle"							,generator=A							,output_type="np"			)

					__A             =			output.images[0]

					assert image.shape == (7_68, 7_68, 3)

					assert_mean_pixel_difference(A							,A			)






	def     UpperCamelCase_  (	self :			Union[str, Any]			):
					__A             =			load_image(
					    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"			)

					__A             =			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"			)

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

					__A             =			torch.Generator(device="cpu"			).manual_seed(0			)
					__A             =			pipe(A							,"anime turle"							,generator=A							,output_type="np"			)

					__A             =			output.images[0]

					assert image.shape == (7_68, 7_68, 3)

					assert_mean_pixel_difference(A							,A			)







	def     UpperCamelCase_  (	self :			Dict			):
					__A             =			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()

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

					__A             =			pipe(
					    A							,"anime turtle"							,num_inference_steps=2							,output_type="np"							,)

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


import copy
from dataclasses import dataclass, field
from typing import ClassVar, Dict

from ..features import Audio, Features, Value
from .base import TaskTemplate





@dataclass(frozen=__SCREAMING_SNAKE_CASE  )
class 							UpperCAmelCase				(  __SCREAMING_SNAKE_CASE  ):




	'''simple docstring'''





	snake_case_    =					field(default="automatic-speech-recognition"      ,      metadata={"include_in_asdict_even_if_is_default": True}  )
	snake_case_    =					Features({"audio": Audio()}  )
	snake_case_    =					Features({"transcription": Value("string"  )}  )
	snake_case_    =					"audio"
	snake_case_    =					"transcription"






	def     UpperCamelCase_  (	self :			Optional[int]							,A :			Union[str, Any]			):
					if self.audio_column not in features:
									raise ValueError(f'''Column {self.audio_column} is not present in features.'''			)
					if not isinstance(features[self.audio_column]							,A			):
									raise ValueError(f'''Column {self.audio_column} is not an Audio type.'''			)
					__A             =			copy.deepcopy(self			)
					__A             =			self.input_schema.copy()
					__A             =			features[self.audio_column]
					__A             =			input_schema
					return task_template







	@property
	def     UpperCamelCase_  (	self :			int			):
					return {self.audio_column: "audio", self.transcription_column: "transcription"}
