
"""simple docstring"""





import gc
import random
import unittest

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import (
    DiffusionPipeline,
    UnCLIPImageVariationPipeline,
    UnCLIPScheduler,
    UNetaDConditionModel,
    UNetaDModel,
)
from diffusers.pipelines.unclip.text_proj import UnCLIPTextProjModel
from diffusers.utils import floats_tensor, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, load_image, require_torch_gpu, skip_mps

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()



class __lowerCAmelCase					(       _lowercase  ,					unittest.TestCase   ):




			"""simple docstring"""







			snake_case						   =					UnCLIPImageVariationPipeline
			snake_case						   =					IMAGE_VARIATION_PARAMS - {"height", "width", "guidance_scale"}
			snake_case						   =					IMAGE_VARIATION_BATCH_PARAMS

			snake_case						   =					[
			    "generator",
			    "return_dict",
			    "decoder_num_inference_steps",
			    "super_res_num_inference_steps",
			]
			snake_case						   =					False




			@property
			def   lowerCamelCase__  (					self    :       Optional[Any]						)   ->			Union[str, Any]:







								"""simple docstring"""
								return 32




			@property
			def   lowerCamelCase__  (					self    :       int						)   ->			Dict:







								"""simple docstring"""
								return 32




			@property
			def   lowerCamelCase__  (					self    :       int						)   ->			Tuple:







								"""simple docstring"""
								return self.time_input_dim




			@property
			def   lowerCamelCase__  (					self    :       Optional[int]						)   ->			Tuple:







								"""simple docstring"""
								return self.time_input_dim * 4




			@property
			def   lowerCamelCase__  (					self    :       int						)   ->			str:







								"""simple docstring"""
								return 100




			@property
			def   lowerCamelCase__  (					self    :       Optional[Any]						)   ->			Dict:







								"""simple docstring"""
								A_								=    CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"						)
								return tokenizer




			@property
			def   lowerCamelCase__  (					self    :       Optional[int]						)   ->			int:







								"""simple docstring"""
								torch.manual_seed(0						)
								A_								=    CLIPTextConfig(
								    bos_token_id=0							, eos_token_id=2							, hidden_size=self.text_embedder_hidden_size							, projection_dim=self.text_embedder_hidden_size							, intermediate_size=37							, layer_norm_eps=1e-05							, num_attention_heads=4							, num_hidden_layers=5							, pad_token_id=1							, vocab_size=1_000							, )
								return CLIPTextModelWithProjection(_snake_case						)




			@property
			def   lowerCamelCase__  (					self    :       Union[str, Any]						)   ->			List[Any]:







								"""simple docstring"""
								torch.manual_seed(0						)
								A_								=    CLIPVisionConfig(
								    hidden_size=self.text_embedder_hidden_size							, projection_dim=self.text_embedder_hidden_size							, num_hidden_layers=5							, num_attention_heads=4							, image_size=32							, intermediate_size=37							, patch_size=1							, )
								return CLIPVisionModelWithProjection(_snake_case						)




			@property
			def   lowerCamelCase__  (					self    :       Dict						)   ->			Tuple:







								"""simple docstring"""
								torch.manual_seed(0						)

								A_								=    {
								    "clip_embeddings_dim": self.text_embedder_hidden_size,
								    "time_embed_dim": self.time_embed_dim,
								    "cross_attention_dim": self.cross_attention_dim,
								}

								A_								=    UnCLIPTextProjModel(**_snake_case						)
								return model




			@property
			def   lowerCamelCase__  (					self    :       Optional[int]						)   ->			List[Any]:







								"""simple docstring"""
								torch.manual_seed(0						)

								A_								=    {
								    "sample_size": 32,
								    # RGB in channels
								    "in_channels": 3,
								    # Out channels is double in channels because predicts mean and variance
								    "out_channels": 6,
								    "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,
								    "cross_attention_dim": self.cross_attention_dim,
								    "attention_head_dim": 4,
								    "resnet_time_scale_shift": "scale_shift",
								    "class_embed_type": "identity",
								}

								A_								=    UNetaDConditionModel(**_snake_case						)
								return model




			@property
			def   lowerCamelCase__  (					self    :       str						)   ->			Optional[int]:







								"""simple docstring"""
								return {
								    "sample_size": 64,
								    "layers_per_block": 1,
								    "down_block_types": ("ResnetDownsampleBlock2D", "ResnetDownsampleBlock2D"),
								    "up_block_types": ("ResnetUpsampleBlock2D", "ResnetUpsampleBlock2D"),
								    "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
								    "in_channels": 6,
								    "out_channels": 3,
								}




			@property
			def   lowerCamelCase__  (					self    :       Dict						)   ->			Tuple:







								"""simple docstring"""
								torch.manual_seed(0						)

								A_								=    UNetaDModel(**self.dummy_super_res_kwargs						)
								return model




			@property
			def   lowerCamelCase__  (					self    :       Tuple						)   ->			Optional[int]:







								"""simple docstring"""
								# seeded differently to get different unet than `self.dummy_super_res_first`
								torch.manual_seed(1						)

								A_								=    UNetaDModel(**self.dummy_super_res_kwargs						)
								return model




			def   lowerCamelCase__  (					self    :       Union[str, Any]						)   ->			Optional[int]:







								"""simple docstring"""
								A_								=    self.dummy_decoder
								A_								=    self.dummy_text_proj
								A_								=    self.dummy_text_encoder
								A_								=    self.dummy_tokenizer
								A_								=    self.dummy_super_res_first
								A_								=    self.dummy_super_res_last

								A_								=    UnCLIPScheduler(
								    variance_type="learned_range"							, prediction_type="epsilon"							, num_train_timesteps=1_000							, )

								A_								=    UnCLIPScheduler(
								    variance_type="fixed_small_log"							, prediction_type="epsilon"							, num_train_timesteps=1_000							, )

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

								A_								=    self.dummy_image_encoder

								return {
								    "decoder": decoder,
								    "text_encoder": text_encoder,
								    "tokenizer": tokenizer,
								    "text_proj": text_proj,
								    "feature_extractor": feature_extractor,
								    "image_encoder": image_encoder,
								    "super_res_first": super_res_first,
								    "super_res_last": super_res_last,
								    "decoder_scheduler": decoder_scheduler,
								    "super_res_scheduler": super_res_scheduler,
								}




			def   lowerCamelCase__  (					self    :       Optional[int]							, _snake_case    :       Union[str, Any]							, _snake_case    :       str=0							, _snake_case    :       List[str]=True						)   ->			Union[str, Any]:







								"""simple docstring"""
								A_								=    floats_tensor((1, 3, 32, 32)							, rng=random.Random(_snake_case						)						).to(_snake_case						)
								if str(_snake_case						).startswith("mps"						):
													A_								=    torch.manual_seed(_snake_case						)
								else:
													A_								=    torch.Generator(device=_snake_case						).manual_seed(_snake_case						)

								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(_snake_case						)[0]

								return {
								    "image": input_image,
								    "generator": generator,
								    "decoder_num_inference_steps": 2,
								    "super_res_num_inference_steps": 2,
								    "output_type": "np",
								}




			def   lowerCamelCase__  (					self    :       int						)   ->			Optional[int]:







								"""simple docstring"""
								A_								=    "cpu"

								A_								=    self.get_dummy_components()

								A_								=    self.pipeline_class(**_snake_case						)
								A_								=    pipe.to(_snake_case						)

								pipe.set_progress_bar_config(disable=_snake_case						)

								A_								=    self.get_dummy_inputs(_snake_case							, pil_image=_snake_case						)

								A_								=    pipe(**_snake_case						)
								A_								=    output.images

								A_								=    self.get_dummy_inputs(_snake_case							, pil_image=_snake_case						)

								A_								=    pipe(
								    **_snake_case							, return_dict=_snake_case							, )[0]

								A_								=    image[0, -3:, -3:, -1]
								A_								=    image_from_tuple[0, -3:, -3:, -1]

								assert image.shape == (1, 64, 64, 3)

								A_								=    np.array(
								    [
								        0.9_9_9_7,
								        0.0_0_0_2,
								        0.9_9_9_7,
								        0.9_9_9_7,
								        0.9_9_6_9,
								        0.0_0_2_3,
								        0.9_9_9_7,
								        0.9_9_6_9,
								        0.9_9_7_0,
								    ]						)

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




			def   lowerCamelCase__  (					self    :       List[Any]						)   ->			str:







								"""simple docstring"""
								A_								=    "cpu"

								A_								=    self.get_dummy_components()

								A_								=    self.pipeline_class(**_snake_case						)
								A_								=    pipe.to(_snake_case						)

								pipe.set_progress_bar_config(disable=_snake_case						)

								A_								=    self.get_dummy_inputs(_snake_case							, pil_image=_snake_case						)

								A_								=    pipe(**_snake_case						)
								A_								=    output.images

								A_								=    self.get_dummy_inputs(_snake_case							, pil_image=_snake_case						)

								A_								=    pipe(
								    **_snake_case							, return_dict=_snake_case							, )[0]

								A_								=    image[0, -3:, -3:, -1]
								A_								=    image_from_tuple[0, -3:, -3:, -1]

								assert image.shape == (1, 64, 64, 3)

								A_								=    np.array([0.9_9_9_7, 0.0_0_0_3, 0.9_9_9_7, 0.9_9_9_7, 0.9_9_7_0, 0.0_0_2_4, 0.9_9_9_7, 0.9_9_7_1, 0.9_9_7_1]						)

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




			def   lowerCamelCase__  (					self    :       Optional[Any]						)   ->			Union[str, Any]:







								"""simple docstring"""
								A_								=    "cpu"

								A_								=    self.get_dummy_components()

								A_								=    self.pipeline_class(**_snake_case						)
								A_								=    pipe.to(_snake_case						)

								pipe.set_progress_bar_config(disable=_snake_case						)

								A_								=    self.get_dummy_inputs(_snake_case							, pil_image=_snake_case						)
								A_								=    [
								    pipeline_inputs["image"],
								    pipeline_inputs["image"],
								]

								A_								=    pipe(**_snake_case						)
								A_								=    output.images

								A_								=    self.get_dummy_inputs(_snake_case							, pil_image=_snake_case						)
								A_								=    [
								    tuple_pipeline_inputs["image"],
								    tuple_pipeline_inputs["image"],
								]

								A_								=    pipe(
								    **_snake_case							, return_dict=_snake_case							, )[0]

								A_								=    image[0, -3:, -3:, -1]
								A_								=    image_from_tuple[0, -3:, -3:, -1]

								assert image.shape == (2, 64, 64, 3)

								A_								=    np.array(
								    [
								        0.9_9_9_7,
								        0.9_9_8_9,
								        0.0_0_0_8,
								        0.0_0_2_1,
								        0.9_9_6_0,
								        0.0_0_1_8,
								        0.0_0_1_4,
								        0.0_0_0_2,
								        0.9_9_3_3,
								    ]						)

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




			def   lowerCamelCase__  (					self    :       str						)   ->			Any:







								"""simple docstring"""
								A_								=    torch.device("cpu"						)



								class __lowerCAmelCase					:




											"""simple docstring"""







											snake_case						   =					1





								A_								=    self.get_dummy_components()

								A_								=    self.pipeline_class(**_snake_case						)
								A_								=    pipe.to(_snake_case						)

								pipe.set_progress_bar_config(disable=_snake_case						)

								A_								=    torch.Generator(device=_snake_case						).manual_seed(0						)
								A_								=    pipe.decoder.dtype
								A_								=    1

								A_								=    (
								    batch_size,
								    pipe.decoder.config.in_channels,
								    pipe.decoder.config.sample_size,
								    pipe.decoder.config.sample_size,
								)
								A_								=    pipe.prepare_latents(
								    _snake_case							, dtype=_snake_case							, device=_snake_case							, generator=_snake_case							, latents=_snake_case							, scheduler=DummyScheduler()						)

								A_								=    (
								    batch_size,
								    pipe.super_res_first.config.in_channels // 2,
								    pipe.super_res_first.config.sample_size,
								    pipe.super_res_first.config.sample_size,
								)
								A_								=    pipe.prepare_latents(
								    _snake_case							, dtype=_snake_case							, device=_snake_case							, generator=_snake_case							, latents=_snake_case							, scheduler=DummyScheduler()						)

								A_								=    self.get_dummy_inputs(_snake_case							, pil_image=_snake_case						)

								A_								=    pipe(
								    **_snake_case							, decoder_latents=_snake_case							, super_res_latents=_snake_case						).images

								A_								=    self.get_dummy_inputs(_snake_case							, pil_image=_snake_case						)
								# Don't pass image, instead pass embedding
								A_								=    pipeline_inputs.pop("image"						)
								A_								=    pipe.image_encoder(_snake_case						).image_embeds

								A_								=    pipe(
								    **_snake_case							, decoder_latents=_snake_case							, super_res_latents=_snake_case							, image_embeddings=_snake_case							, ).images

								# make sure passing text embeddings manually is identical
								assert np.abs(img_out_a - img_out_a						).max() < 1e-4




			@skip_mps
			def   lowerCamelCase__  (					self    :       Tuple						)   ->			str:







								"""simple docstring"""
								A_								=    torch_device == "cpu"

								# Check is relaxed because there is not a torch 2.0 sliced attention added kv processor
								A_								=    1e-2

								self._test_attention_slicing_forward_pass(
								    test_max_difference=_snake_case							, expected_max_diff=_snake_case						)




			@skip_mps
			def   lowerCamelCase__  (					self    :       str						)   ->			List[Any]:







								"""simple docstring"""
								A_								=    torch_device == "cpu"
								A_								=    True
								A_								=    [
								    "decoder_num_inference_steps",
								    "super_res_num_inference_steps",
								]

								self._test_inference_batch_single_identical(
								    test_max_difference=_snake_case							, relax_max_difference=_snake_case							, additional_params_copy_to_batched_inputs=_snake_case							, )




			def   lowerCamelCase__  (					self    :       List[Any]						)   ->			Optional[Any]:







								"""simple docstring"""
								A_								=    [
								    "decoder_num_inference_steps",
								    "super_res_num_inference_steps",
								]

								if torch_device == "mps":
													# TODO: MPS errors with larger batch sizes
													A_								=    [2, 3]
													self._test_inference_batch_consistent(
													    batch_sizes=_snake_case							, additional_params_copy_to_batched_inputs=_snake_case							, )
								else:
													self._test_inference_batch_consistent(
													    additional_params_copy_to_batched_inputs=_snake_case						)




			@skip_mps
			def   lowerCamelCase__  (					self    :       Optional[Any]						)   ->			int:







								"""simple docstring"""
								return super().test_dict_tuple_outputs_equivalent()




			@skip_mps
			def   lowerCamelCase__  (					self    :       Tuple						)   ->			Union[str, Any]:







								"""simple docstring"""
								return super().test_save_load_local()






			@skip_mps
			def   lowerCamelCase__  (					self    :       Any						)   ->			List[Any]:







								"""simple docstring"""
								return super().test_save_load_optional_components()





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




			"""simple docstring"""







			def   lowerCamelCase__  (					self    :       Union[str, Any]						)   ->			Union[str, Any]:







								"""simple docstring"""
								# clean up the VRAM after each test
								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()






			def   lowerCamelCase__  (					self    :       Union[str, Any]						)   ->			Dict:







								"""simple docstring"""
								A_								=    load_image(
								    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/unclip/cat.png"						)
								A_								=    load_numpy(
								    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
								    "/unclip/karlo_v1_alpha_cat_variation_fp16.npy"						)

								A_								=    UnCLIPImageVariationPipeline.from_pretrained(
								    "kakaobrain/karlo-v1-alpha-image-variations"							, torch_dtype=torch.floataa						)
								A_								=    pipeline.to(_snake_case						)
								pipeline.set_progress_bar_config(disable=_snake_case						)

								A_								=    torch.Generator(device="cpu"						).manual_seed(0						)
								A_								=    pipeline(
								    _snake_case							, generator=_snake_case							, output_type="np"							, )

								A_								=    output.images[0]

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

								assert_mean_pixel_difference(_snake_case							, _snake_case							, 15						)






"""simple docstring"""





import operator
def 				A_  (__a					, __a = False					, __a = None					):

					'''simple docstring'''





					A_								=    operator.lt if reverse else operator.gt
					A_								=    solution or []

					if not arr:
										return solution

					A_								=    [arr.pop(0					)]
					for i, item in enumerate(__a					):
										if _operator(__a					, sublist[-1]					):
															sublist.append(__a					)
															arr.pop(__a					)

    #  merging sublist into solution list
					if not solution:
										solution.extend(__a					)
					else:
										while sublist:
															A_								=    sublist.pop(0					)
															for i, xx in enumerate(__a					):
																				if not _operator(__a					, __a					):
																									solution.insert(__a					, __a					)
																									break
															else:
																				solution.append(__a					)

					strand_sort(__a					, __a					, __a					)
					return solution


if __name__ == "__main__":
		assert strand_sort([4, 3, 5, 1, 2]) == [1, 2, 3, 4, 5]
		assert strand_sort([4, 3, 5, 1, 2], reverse=True) == [5, 4, 3, 2, 1]




