

import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()



class 					_a(__magic_name__	,     __magic_name__	,     __magic_name__	,     unittest.TestCase       ):


					'''simple docstring'''






					UpperCAmelCase__:     Dict													= StableUnCLIPPipeline
					UpperCAmelCase__:     Any													= TEXT_TO_IMAGE_PARAMS
					UpperCAmelCase__:     List[str]													= TEXT_TO_IMAGE_BATCH_PARAMS
					UpperCAmelCase__:     Dict													= TEXT_TO_IMAGE_IMAGE_PARAMS
					UpperCAmelCase__:     str													= TEXT_TO_IMAGE_IMAGE_PARAMS

					# TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
					UpperCAmelCase__:     Union[str, Any]													= False






					def 						__A    (				self    ):
						A__							:					Dict       =	32
						A__							:					Any       =	embedder_hidden_size

						# prior components

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

						torch.manual_seed(0    )
						A__							:					int       =	CLIPTextModelWithProjection(
						    CLIPTextConfig(
						        bos_token_id=0					, eos_token_id=2					, hidden_size=A__					, projection_dim=A__					, 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    )
						A__							:					List[Any]       =	PriorTransformer(
						    num_attention_heads=2					, attention_head_dim=12					, embedding_dim=A__					, num_layers=1					, )

						torch.manual_seed(0    )
						A__							:					Optional[int]       =	DDPMScheduler(
						    variance_type="""fixed_small_log"""					, prediction_type="""sample"""					, num_train_timesteps=1000					, clip_sample=A__					, clip_sample_range=5.0					, beta_schedule="""squaredcos_cap_v2"""					, )

						# regular denoising components

						torch.manual_seed(0    )
						A__							:					Dict       =	StableUnCLIPImageNormalizer(embedding_dim=A__    )
						A__							:					int       =	DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""    )

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

						torch.manual_seed(0    )
						A__							:					List[str]       =	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    )
						A__							:					Union[str, Any]       =	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__							:					Union[str, Any]       =	DDIMScheduler(
						    beta_schedule="""scaled_linear"""					, beta_start=0.0_0_0_8_5					, beta_end=0.0_1_2					, prediction_type="""v_prediction"""					, set_alpha_to_one=A__					, steps_offset=1					, )

						torch.manual_seed(0    )
						A__							:					List[Any]       =	AutoencoderKL()

						A__							:					Union[str, Any]       =	{
						    # prior components
						    """prior_tokenizer""": prior_tokenizer,
						    """prior_text_encoder""": prior_text_encoder,
						    """prior""": prior,
						    """prior_scheduler""": prior_scheduler,
						    # image noising components
						    """image_normalizer""": image_normalizer,
						    """image_noising_scheduler""": image_noising_scheduler,
						    # regular denoising components
						    """tokenizer""": tokenizer,
						    """text_encoder""": text_encoder,
						    """unet""": unet,
						    """scheduler""": scheduler,
						    """vae""": vae,
						}

						return components






					def 						__A    (				self					, A__					, A__=0    ):
						if str(A__    ).startswith("""mps"""    ):
							A__							:					Optional[int]       =	torch.manual_seed(A__    )
						else:
							A__							:					int       =	torch.Generator(device=A__    ).manual_seed(A__    )
						A__							:					Dict       =	{
						    """prompt""": """A painting of a squirrel eating a burger""",
						    """generator""": generator,
						    """num_inference_steps""": 2,
						    """prior_num_inference_steps""": 2,
						    """output_type""": """numpy""",
						}
						return inputs






					def 						__A    (				self    ):
						A__							:					int       =	torch_device == """cpu"""

						self._test_attention_slicing_forward_pass(test_max_difference=A__    )






					def 						__A    (				self    ):
						A__							:					str       =	torch_device in ["""cpu""", """mps"""]

						self._test_inference_batch_single_identical(test_max_difference=A__    )



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


					'''simple docstring'''






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






					def 						__A    (				self    ):
						A__							:					Any       =	load_numpy(
						    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy"""    )

						A__							:					Optional[Any]       =	StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""					, 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__							:					Optional[int]       =	torch.Generator(device="""cpu"""    ).manual_seed(0    )
						A__							:					List[str]       =	pipe("""anime turle"""					, generator=A__					, output_type="""np"""    )

						A__							:					Tuple       =	output.images[0]

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

						assert_mean_pixel_difference(A__					, A__    )






					def 						__A    (				self    ):
						torch.cuda.empty_cache()
						torch.cuda.reset_max_memory_allocated()
						torch.cuda.reset_peak_memory_stats()

						A__							:					Optional[int]       =	StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""					, torch_dtype=torch.floataa    )
						A__							:					List[Any]       =	pipe.to(A__    )
						pipe.set_progress_bar_config(disable=A__    )
						pipe.enable_attention_slicing()
						pipe.enable_sequential_cpu_offload()

						A__							:					Any       =	pipe(
						    """anime turtle"""					, prior_num_inference_steps=2					, num_inference_steps=2					, output_type="""np"""					, )

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






import itertools
import random
import unittest

import numpy as np

from transformers import ASTFeatureExtractor
from transformers.testing_utils import require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


A_ :				Dict						 =    random.Random()

if is_torch_available():
	import torch

def 		UpperCamelCase      (lowercase_:				Tuple					,   lowercase_:				Tuple=1.0					,   lowercase_:				Dict=None					,   lowercase_:				int=None      )  ->   str:
	if rng is None:
		A__							:					Optional[Any]       =	global_rng

	A__							:					List[str]       =	[]
	for batch_idx in range(shape[0]      ):
		values.append([]      )
		for _ in range(shape[1]      ):
			values[-1].append(rng.random() * scale      )

	return values



class 					_a		(unittest.TestCase       ):


					'''simple docstring'''






					def __init__(				self					, A__					, A__=7					, A__=400					, A__=2000					, A__=1					, A__=0.0					, A__=1_6000					, A__=True					, A__=True					, ):
						A__							:					Any       =	parent
						A__							:					Optional[int]       =	batch_size
						A__							:					Union[str, Any]       =	min_seq_length
						A__							:					Dict       =	max_seq_length
						A__							:					Tuple       =	(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
						A__							:					str       =	feature_size
						A__							:					Optional[int]       =	padding_value
						A__							:					List[str]       =	sampling_rate
						A__							:					List[str]       =	return_attention_mask
						A__							:					int       =	do_normalize






					def 						__A    (				self    ):
						return {
						    "feature_size": self.feature_size,
						    "padding_value": self.padding_value,
						    "sampling_rate": self.sampling_rate,
						    "return_attention_mask": self.return_attention_mask,
						    "do_normalize": self.do_normalize,
						}






					def 						__A    (				self					, A__=False					, A__=False    ):
						def _flatten(A__    ):
							return list(itertools.chain(*A__    )    )

						if equal_length:
							A__							:					Dict       =	floats_list((self.batch_size, self.max_seq_length)    )
						else:
							# make sure that inputs increase in size
							A__							:					Union[str, Any]       =	[
							    _flatten(floats_list((x, self.feature_size)    )    )
							    for x in range(self.min_seq_length					, self.max_seq_length					, self.seq_length_diff    )
							]

						if numpify:
							A__							:					Optional[int]       =	[np.asarray(A__    ) for x in speech_inputs]

						return speech_inputs



@require_torch
@require_torchaudio
class 					_a		(__magic_name__	,     unittest.TestCase       ):


					'''simple docstring'''






					UpperCAmelCase__:     int													= ASTFeatureExtractor






					def 						__A    (				self    ):
						A__							:					Optional[Any]       =	ASTFeatureExtractionTester(self    )






					def 						__A    (				self    ):
						# Tests that all call wrap to encode_plus and batch_encode_plus
						A__							:					Optional[int]       =	self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()    )
						# create three inputs of length 800, 1000, and 1200
						A__							:					List[Any]       =	[floats_list((1, x)    )[0] for x in range(800					, 1400					, 200    )]
						A__							:					Optional[Any]       =	[np.asarray(A__    ) for speech_input in speech_inputs]

						# Test not batched input
						A__							:					Tuple       =	feat_extract(speech_inputs[0]					, return_tensors="""np"""    ).input_values
						A__							:					Tuple       =	feat_extract(np_speech_inputs[0]					, return_tensors="""np"""    ).input_values
						self.assertTrue(np.allclose(A__					, A__					, atol=1e-3    )    )

						# Test batched
						A__							:					Tuple       =	feat_extract(A__					, padding=A__					, return_tensors="""np"""    ).input_values
						A__							:					Tuple       =	feat_extract(A__					, padding=A__					, return_tensors="""np"""    ).input_values
						for enc_seq_a, enc_seq_a in zip(A__					, A__    ):
							self.assertTrue(np.allclose(A__					, A__					, atol=1e-3    )    )

						# Test 2-D numpy arrays are batched.
						A__							:					int       =	[floats_list((1, x)    )[0] for x in (800, 800, 800)]
						A__							:					List[str]       =	np.asarray(A__    )
						A__							:					Union[str, Any]       =	feat_extract(A__					, return_tensors="""np"""    ).input_values
						A__							:					Optional[Any]       =	feat_extract(A__					, return_tensors="""np"""    ).input_values
						for enc_seq_a, enc_seq_a in zip(A__					, A__    ):
							self.assertTrue(np.allclose(A__					, A__					, atol=1e-3    )    )






					@require_torch
					def 						__A    (				self    ):
						import torch

						A__							:					Union[str, Any]       =	self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()    )
						A__							:					Tuple       =	np.random.rand(100    ).astype(np.floataa    )
						A__							:					Tuple       =	np_speech_inputs.tolist()

						for inputs in [py_speech_inputs, np_speech_inputs]:
							A__							:					List[str]       =	feature_extractor.pad([{"""input_values""": inputs}]					, return_tensors="""np"""    )
							self.assertTrue(np_processed.input_values.dtype == np.floataa    )
							A__							:					Any       =	feature_extractor.pad([{"""input_values""": inputs}]					, return_tensors="""pt"""    )
							self.assertTrue(pt_processed.input_values.dtype == torch.floataa    )






					def 						__A    (				self					, A__    ):
						from datasets import load_dataset

						A__							:					str       =	load_dataset("""hf-internal-testing/librispeech_asr_dummy"""					, """clean"""					, split="""validation"""    )
						# automatic decoding with librispeech
						A__							:					str       =	ds.sort("""id"""    ).select(range(A__    )    )[:num_samples]["""audio"""]

						return [x["array"] for x in speech_samples]






					@require_torch
					def 						__A    (				self    ):
						# fmt: off
						A__							:					Optional[Any]       =	torch.tensor(
						    [-0.9_8_9_4, -1.2_7_7_6, -0.9_0_6_6, -1.2_7_7_6, -0.9_3_4_9, -1.2_6_0_9, -1.0_3_8_6, -1.2_7_7_6,
						     -1.1_5_6_1, -1.2_7_7_6, -1.2_0_5_2, -1.2_7_2_3, -1.2_1_9_0, -1.2_1_3_2, -1.2_7_7_6, -1.1_1_3_3,
						     -1.1_9_5_3, -1.1_3_4_3, -1.1_5_8_4, -1.2_2_0_3, -1.1_7_7_0, -1.2_4_7_4, -1.2_3_8_1, -1.1_9_3_6,
						     -0.9_2_7_0, -0.8_3_1_7, -0.8_0_4_9, -0.7_7_0_6, -0.7_5_6_5, -0.7_8_6_9]    )
						# fmt: on

						A__							:					Any       =	self._load_datasamples(1    )
						A__							:					Tuple       =	ASTFeatureExtractor()
						A__							:					Dict       =	feature_extractor(A__					, return_tensors="""pt"""    ).input_values
						self.assertEquals(input_values.shape					, (1, 1024, 128)    )
						self.assertTrue(torch.allclose(input_values[0, 0, :30]					, A__					, atol=1e-4    )    )



