



"""simple docstring"""
import itertools
import os
import random
import tempfile
import unittest

import numpy as np
from datasets import load_dataset

from transformers import is_speech_available
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_speech_available():
					from transformers import WhisperFeatureExtractor

if is_torch_available():
					import torch

SCREAMING_SNAKE_CASE_				= random.Random()



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__=1.0, SCREAMING_SNAKE_CASE__=None, SCREAMING_SNAKE_CASE__=None	)			->      Any:
		if rng is None:
				a_     :      str 						=     global_rng

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

		return values




@require_torch
@require_torchaudio
class 			snake_case_						(						unittest.TestCase				):






							def __init__(   self			,     a_			,     a_=7			,     a_=4_0_0			,     a_=2_0_0_0			,     a_=1_0			,     a_=1_6_0			,     a_=8			,     a_=0.0			,     a_=4_0_0_0			,     a_=False			,     a_=True			,     ):
									a_     :      List[str] 						=     parent
									a_     :      Optional[Any] 						=     batch_size
									a_     :      List[Any] 						=     min_seq_length
									a_     :      Union[str, Any] 						=     max_seq_length
									a_     :      Tuple 						=     (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
									a_     :      Optional[Any] 						=     padding_value
									a_     :      List[str] 						=     sampling_rate
									a_     :      Optional[int] 						=     return_attention_mask
									a_     :      int 						=     do_normalize
									a_     :      Tuple 						=     feature_size
									a_     :      int 						=     chunk_length
									a_     :      Any 						=     hop_length






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






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

									if equal_length:
											a_     :      Any 						=     [floats_list((self.max_seq_length, self.feature_size)      ) for _ in range(self.batch_size      )]
									else:
											# make sure that inputs increase in size
											a_     :      Any 						=     [
											    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_     :      Tuple 						=     [np.asarray(__A      ) for x in speech_inputs]
									return speech_inputs




@require_torch
@require_torchaudio
class 			snake_case_						(						UpperCamelCase_     ,unittest.TestCase				):
							__lowerCAmelCase    					=					WhisperFeatureExtractor if is_speech_available() else None






							def snake_case_  (   self      ):
									a_     :      Optional[Any] 						=     WhisperFeatureExtractionTester(self      )






							def snake_case_  (   self      ):
									a_     :      int 						=     self.feature_extraction_class(**self.feat_extract_dict      )

									with tempfile.TemporaryDirectory() as tmpdirname:
											a_     :      List[str] 						=     feat_extract_first.save_pretrained(__A      )[0]
											check_json_file_has_correct_format(__A      )
											a_     :      List[str] 						=     self.feature_extraction_class.from_pretrained(__A      )

									a_     :      Tuple 						=     feat_extract_first.to_dict()
									a_     :      List[Any] 						=     feat_extract_second.to_dict()
									a_     :      int 						=     feat_extract_first.mel_filters
									a_     :      Any 						=     feat_extract_second.mel_filters
									self.assertTrue(np.allclose(__A			,     __A      )      )
									self.assertEqual(__A			,     __A      )






							def snake_case_  (   self      ):
									a_     :      List[str] 						=     self.feature_extraction_class(**self.feat_extract_dict      )

									with tempfile.TemporaryDirectory() as tmpdirname:
											a_     :      Any 						=     os.path.join(__A			,     "feat_extract.json"      )
											feat_extract_first.to_json_file(__A      )
											a_     :      List[str] 						=     self.feature_extraction_class.from_json_file(__A      )

									a_     :      Union[str, Any] 						=     feat_extract_first.to_dict()
									a_     :      List[Any] 						=     feat_extract_second.to_dict()
									a_     :      Any 						=     feat_extract_first.mel_filters
									a_     :      Tuple 						=     feat_extract_second.mel_filters
									self.assertTrue(np.allclose(__A			,     __A      )      )
									self.assertEqual(__A			,     __A      )






							def snake_case_  (   self      ):
									# Tests that all call wrap to encode_plus and batch_encode_plus
									a_     :      Tuple 						=     self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()      )
									# create three inputs of length 800, 1000, and 1200
									a_     :      int 						=     [floats_list((1, x)      )[0] for x in range(8_0_0			,     1_4_0_0			,     2_0_0      )]
									a_     :      Any 						=     [np.asarray(__A      ) for speech_input in speech_inputs]

									# Test feature size
									a_     :      List[str] 						=     feature_extractor(__A			,     padding="max_length"			,     return_tensors="np"      ).input_features
									self.assertTrue(input_features.ndim == 3      )
									self.assertTrue(input_features.shape[-1] == feature_extractor.nb_max_frames      )
									self.assertTrue(input_features.shape[-2] == feature_extractor.feature_size      )

									# Test not batched input
									a_     :      Any 						=     feature_extractor(speech_inputs[0]			,     return_tensors="np"      ).input_features
									a_     :      int 						=     feature_extractor(np_speech_inputs[0]			,     return_tensors="np"      ).input_features
									self.assertTrue(np.allclose(__A			,     __A			,     atol=1e-3      )      )

									# Test batched
									a_     :      List[str] 						=     feature_extractor(__A			,     return_tensors="np"      ).input_features
									a_     :      Tuple 						=     feature_extractor(__A			,     return_tensors="np"      ).input_features
									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_     :      Optional[Any] 						=     [floats_list((1, x)      )[0] for x in (8_0_0, 8_0_0, 8_0_0)]
									a_     :      List[Any] 						=     np.asarray(__A      )
									a_     :      Tuple 						=     feature_extractor(__A			,     return_tensors="np"      ).input_features
									a_     :      Union[str, Any] 						=     feature_extractor(__A			,     return_tensors="np"      ).input_features
									for enc_seq_a, enc_seq_a in zip(__A			,     __A      ):
											self.assertTrue(np.allclose(__A			,     __A			,     atol=1e-3      )      )

									# Test truncation required
									a_     :      Optional[Any] 						=     [floats_list((1, x)      )[0] for x in range(2_0_0			,     (feature_extractor.n_samples + 5_0_0)			,     2_0_0      )]
									a_     :      Optional[Any] 						=     [np.asarray(__A      ) for speech_input in speech_inputs]

									a_     :      Union[str, Any] 						=     [x[: feature_extractor.n_samples] for x in speech_inputs]
									a_     :      List[Any] 						=     [np.asarray(__A      ) for speech_input in speech_inputs_truncated]

									a_     :      Union[str, Any] 						=     feature_extractor(__A			,     return_tensors="np"      ).input_features
									a_     :      Tuple 						=     feature_extractor(__A			,     return_tensors="np"      ).input_features
									for enc_seq_a, enc_seq_a in zip(__A			,     __A      ):
											self.assertTrue(np.allclose(__A			,     __A			,     atol=1e-3      )      )






							def snake_case_  (   self      ):
									import torch

									a_     :      int 						=     self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()      )
									a_     :      List[Any] 						=     np.random.rand(1_0_0			,     3_2      ).astype(np.floataa      )
									a_     :      Union[str, Any] 						=     np_speech_inputs.tolist()

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






							def snake_case_  (   self			,     a_      ):
									a_     :      Any 						=     load_dataset("hf-internal-testing/librispeech_asr_dummy"			,     "clean"			,     split="validation"      )
									# automatic decoding with librispeech
									a_     :      Optional[int] 						=     ds.sort("id"      ).select(range(__A      )      )[:num_samples]["audio"]

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






							def snake_case_  (   self      ):
									# fmt: off
									a_     :      List[str] 						=     torch.tensor(
									    [
									        0.1_193, -0.0_946, -0.1_098, -0.0_196, 0.0_225, -0.0_690, -0.1_736, 0.0_951,
									        0.0_971, -0.0_817, -0.0_702, 0.0_162, 0.0_260, 0.0_017, -0.0_192, -0.1_678,
									        0.0_709, -0.1_867, -0.0_655, -0.0_274, -0.0_234, -0.1_884, -0.0_516, -0.0_554,
									        -0.0_274, -0.1_425, -0.1_423, 0.0_837, 0.0_377, -0.0_854
									    ]      )
									# fmt: on

									a_     :      Tuple 						=     self._load_datasamples(1      )
									a_     :      Any 						=     WhisperFeatureExtractor()
									a_     :      List[str] 						=     feature_extractor(__A			,     return_tensors="pt"      ).input_features
									self.assertEqual(input_features.shape			,     (1, 8_0, 3_0_0_0)      )
									self.assertTrue(torch.allclose(input_features[0, 0, :3_0]			,     __A			,     atol=1e-4      )      )






							def snake_case_  (   self      ):
									a_     :      Tuple 						=     self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()      )
									a_     :      Dict 						=     self._load_datasamples(1      )[0]
									a_     :      Dict 						=     ((audio - audio.min()) / (audio.max() - audio.min())) * 6_5_5_3_5  # Rescale to [0, 65535] to show issue
									a_     :      List[str] 						=     feat_extract.zero_mean_unit_var_norm([audio]			,     attention_mask=__A      )[0]

									self.assertTrue(np.all(np.mean(__A      ) < 1e-3      )      )
									self.assertTrue(np.all(np.abs(np.var(__A      ) - 1      ) < 1e-3      )      )




"""simple docstring"""
import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyImgaImgPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()




class 			snake_case_						(						a_     ,unittest.TestCase				):
							__lowerCAmelCase    					=					KandinskyImgaImgPipeline
							__lowerCAmelCase    					=					["prompt", "image_embeds", "negative_image_embeds", "image"]
							__lowerCAmelCase    					=					[
							    "prompt",
							    "negative_prompt",
							    "image_embeds",
							    "negative_image_embeds",
							    "image",
							]
							__lowerCAmelCase    					=					[
							    "generator",
							    "height",
							    "width",
							    "strength",
							    "guidance_scale",
							    "negative_prompt",
							    "num_inference_steps",
							    "return_dict",
							    "guidance_scale",
							    "num_images_per_prompt",
							    "output_type",
							    "return_dict",
							]
							__lowerCAmelCase    					=					False






							@property
							def snake_case_  (   self      ):
									return 3_2






							@property
							def snake_case_  (   self      ):
									return 3_2






							@property
							def snake_case_  (   self      ):
									return self.time_input_dim






							@property
							def snake_case_  (   self      ):
									return self.time_input_dim * 4






							@property
							def snake_case_  (   self      ):
									return 1_0_0






							@property
							def snake_case_  (   self      ):
									a_     :      List[str] 						=     XLMRobertaTokenizerFast.from_pretrained("YiYiXu/tiny-random-mclip-base"      )
									return tokenizer






							@property
							def snake_case_  (   self      ):
									torch.manual_seed(0      )
									a_     :      List[Any] 						=     MCLIPConfig(
									    numDims=self.cross_attention_dim			,     transformerDimensions=self.text_embedder_hidden_size			,     hidden_size=self.text_embedder_hidden_size			,     intermediate_size=3_7			,     num_attention_heads=4			,     num_hidden_layers=5			,     vocab_size=1_0_0_5			,     )

									a_     :      str 						=     MultilingualCLIP(a_      )
									a_     :      Any 						=     text_encoder.eval()

									return text_encoder






							@property
							def snake_case_  (   self      ):
									torch.manual_seed(0      )

									a_     :      Union[str, Any] 						=     {
									    "in_channels": 4,
									    # Out channels is double in channels because predicts mean and variance
									    "out_channels": 8,
									    "addition_embed_type": "text_image",
									    "down_block_types": ("ResnetDownsampleBlock2D", "SimpleCrossAttnDownBlock2D"),
									    "up_block_types": ("SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"),
									    "mid_block_type": "UNetMidBlock2DSimpleCrossAttn",
									    "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
									    "layers_per_block": 1,
									    "encoder_hid_dim": self.text_embedder_hidden_size,
									    "encoder_hid_dim_type": "text_image_proj",
									    "cross_attention_dim": self.cross_attention_dim,
									    "attention_head_dim": 4,
									    "resnet_time_scale_shift": "scale_shift",
									    "class_embed_type": None,
									}

									a_     :      Dict 						=     UNetaDConditionModel(**a_      )
									return model






							@property
							def snake_case_  (   self      ):
									return {
									    "block_out_channels": [3_2, 6_4],
									    "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
									    "in_channels": 3,
									    "latent_channels": 4,
									    "layers_per_block": 1,
									    "norm_num_groups": 8,
									    "norm_type": "spatial",
									    "num_vq_embeddings": 1_2,
									    "out_channels": 3,
									    "up_block_types": [
									        "AttnUpDecoderBlock2D",
									        "UpDecoderBlock2D",
									    ],
									    "vq_embed_dim": 4,
									}






							@property
							def snake_case_  (   self      ):
									torch.manual_seed(0      )
									a_     :      int 						=     VQModel(**self.dummy_movq_kwargs      )
									return model






							def snake_case_  (   self      ):
									a_     :      Dict 						=     self.dummy_text_encoder
									a_     :      Dict 						=     self.dummy_tokenizer
									a_     :      Optional[int] 						=     self.dummy_unet
									a_     :      Dict 						=     self.dummy_movq

									a_     :      List[str] 						=     {
									    "num_train_timesteps": 1_0_0_0,
									    "beta_schedule": "linear",
									    "beta_start": 0.00_085,
									    "beta_end": 0.012,
									    "clip_sample": False,
									    "set_alpha_to_one": False,
									    "steps_offset": 0,
									    "prediction_type": "epsilon",
									    "thresholding": False,
									}

									a_     :      List[Any] 						=     DDIMScheduler(**a_      )

									a_     :      Union[str, Any] 						=     {
									    "text_encoder": text_encoder,
									    "tokenizer": tokenizer,
									    "unet": unet,
									    "scheduler": scheduler,
									    "movq": movq,
									}

									return components






							def snake_case_  (   self			,     a_			,     a_=0      ):
									a_     :      Optional[int] 						=     floats_tensor((1, self.cross_attention_dim)			,     rng=random.Random(a_      )      ).to(a_      )
									a_     :      Tuple 						=     floats_tensor((1, self.cross_attention_dim)			,     rng=random.Random(seed + 1      )      ).to(a_      )
									# create init_image
									a_     :      Optional[int] 						=     floats_tensor((1, 3, 6_4, 6_4)			,     rng=random.Random(a_      )      ).to(a_      )
									a_     :      Any 						=     image.cpu().permute(0			,     2			,     3			,     1      )[0]
									a_     :      int 						=     Image.fromarray(np.uinta(a_      )      ).convert("RGB"      ).resize((2_5_6, 2_5_6)      )

									if str(a_      ).startswith("mps"      ):
											a_     :      Any 						=     torch.manual_seed(a_      )
									else:
											a_     :      Any 						=     torch.Generator(device=a_      ).manual_seed(a_      )
									a_     :      List[Any] 						=     {
									    "prompt": "horse",
									    "image": init_image,
									    "image_embeds": image_embeds,
									    "negative_image_embeds": negative_image_embeds,
									    "generator": generator,
									    "height": 6_4,
									    "width": 6_4,
									    "num_inference_steps": 1_0,
									    "guidance_scale": 7.0,
									    "strength": 0.2,
									    "output_type": "np",
									}
									return inputs






							def snake_case_  (   self      ):
									a_     :      Optional[Any] 						=     "cpu"

									a_     :      List[Any] 						=     self.get_dummy_components()

									a_     :      Union[str, Any] 						=     self.pipeline_class(**a_      )
									a_     :      Tuple 						=     pipe.to(a_      )

									pipe.set_progress_bar_config(disable=a_      )

									a_     :      Union[str, Any] 						=     pipe(**self.get_dummy_inputs(a_      )      )
									a_     :      Any 						=     output.images

									a_     :      str 						=     pipe(
									    **self.get_dummy_inputs(a_      )			,     return_dict=a_			,     )[0]

									a_     :      List[str] 						=     image[0, -3:, -3:, -1]
									a_     :      Optional[Any] 						=     image_from_tuple[0, -3:, -3:, -1]

									assert image.shape == (1, 6_4, 6_4, 3)

									a_     :      Optional[int] 						=     np.array(
									    [0.61_474_943, 0.6_073_539, 0.43_308_544, 0.5_928_269, 0.47_493_595, 0.46_755_973, 0.4_613_838, 0.45_368_797, 0.50_119_233]      )
									assert (
									    np.abs(image_slice.flatten() - expected_slice      ).max() < 1e-2
									), F""" expected_slice {expected_slice}, but got {image_slice.flatten()}"""
									assert (
									    np.abs(image_from_tuple_slice.flatten() - expected_slice      ).max() < 1e-2
									), F""" expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}"""




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






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






							def snake_case_  (   self      ):
									a_     :      Optional[int] 						=     load_numpy(
									    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
									    "/kandinsky/kandinsky_img2img_frog.npy"      )

									a_     :      Union[str, Any] 						=     load_image(
									    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main" "/kandinsky/cat.png"      )
									a_     :      Optional[Any] 						=     "A red cartoon frog, 4k"

									a_     :      int 						=     KandinskyPriorPipeline.from_pretrained(
									    "kandinsky-community/kandinsky-2-1-prior"			,     torch_dtype=torch.floataa      )
									pipe_prior.to(a_      )

									a_     :      int 						=     KandinskyImgaImgPipeline.from_pretrained(
									    "kandinsky-community/kandinsky-2-1"			,     torch_dtype=torch.floataa      )
									a_     :      List[Any] 						=     pipeline.to(a_      )

									pipeline.set_progress_bar_config(disable=a_      )

									a_     :      int 						=     torch.Generator(device="cpu"      ).manual_seed(0      )
									a_						,				a_     :      Optional[int] 						=     pipe_prior(
									    a_			,     generator=a_			,     num_inference_steps=5			,     negative_prompt=""			,     ).to_tuple()

									a_     :      List[Any] 						=     pipeline(
									    a_			,     image=a_			,     image_embeds=a_			,     negative_image_embeds=a_			,     generator=a_			,     num_inference_steps=1_0_0			,     height=7_6_8			,     width=7_6_8			,     strength=0.2			,     output_type="np"			,     )

									a_     :      int 						=     output.images[0]

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

									assert_mean_pixel_difference(a_			,     a_      )