


"""simple docstring"""


import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import PriorTransformer, UnCLIPPipeline, UnCLIPScheduler, UNetaDConditionModel, UNetaDModel
from diffusers.pipelines.unclip.text_proj import UnCLIPTextProjModel
from diffusers.utils import load_numpy, nightly, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()
class snake_case					(     _UpperCamelCase   , unittest.TestCase):
		__UpperCamelCase							=	UnCLIPPipeline
		__UpperCamelCase							=	TEXT_TO_IMAGE_PARAMS - {
		    'negative_prompt',
		    'height',
		    'width',
		    'negative_prompt_embeds',
		    'guidance_scale',
		    'prompt_embeds',
		    'cross_attention_kwargs',
		}
		__UpperCamelCase							=	TEXT_TO_IMAGE_BATCH_PARAMS
		__UpperCamelCase							=	[
		    'generator',
		    'return_dict',
		    'prior_num_inference_steps',
		    'decoder_num_inference_steps',
		    'super_res_num_inference_steps',
		]
		__UpperCamelCase							=	False







		@property
		def        a_       (    self					:				Dict						)				->  List[str]:





						'''simple docstring'''







						return 32







		@property
		def        a_       (    self					:				Dict						)				->  List[Any]:





						'''simple docstring'''







						return 32







		@property
		def        a_       (    self					:				str						)				->  Tuple:





						'''simple docstring'''







						return self.time_input_dim







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





						'''simple docstring'''







						return self.time_input_dim * 4







		@property
		def        a_       (    self					:				List[str]						)				->  int:





						'''simple docstring'''







						return 1_00







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





						'''simple docstring'''







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







		@property
		def        a_       (    self					:				Any						)				->  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-0_5			,      num_attention_heads=4			,      num_hidden_layers=5			,      pad_token_id=1			,      vocab_size=10_00			,      )
						return CLIPTextModelWithProjection(a__						)







		@property
		def        a_       (    self					:				List[str]						)				->  Any:





						'''simple docstring'''







						torch.manual_seed(0						)

						_A         =	{
						    "num_attention_heads": 2,
						    "attention_head_dim": 12,
						    "embedding_dim": self.text_embedder_hidden_size,
						    "num_layers": 1,
						}

						_A         =	PriorTransformer(**a__						)
						return model







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





						'''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(**a__						)
						return model







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





						'''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(**a__						)
						return model







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





						'''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        a_       (    self					:				List[str]						)				->  str:





						'''simple docstring'''







						torch.manual_seed(0						)

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







		@property
		def        a_       (    self					:				Union[str, Any]						)				->  int:





						'''simple docstring'''







						torch.manual_seed(1						)

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







		def        a_       (    self					:				str						)				->  Any:





						'''simple docstring'''







						_A         =	self.dummy_prior
						_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="fixed_small_log"			,      prediction_type="sample"			,      num_train_timesteps=10_00			,      clip_sample_range=5.0			,      )

						_A         =	UnCLIPScheduler(
						    variance_type="learned_range"			,      prediction_type="epsilon"			,      num_train_timesteps=10_00			,      )

						_A         =	UnCLIPScheduler(
						    variance_type="fixed_small_log"			,      prediction_type="epsilon"			,      num_train_timesteps=10_00			,      )

						_A         =	{
						    "prior": prior,
						    "decoder": decoder,
						    "text_proj": text_proj,
						    "text_encoder": text_encoder,
						    "tokenizer": tokenizer,
						    "super_res_first": super_res_first,
						    "super_res_last": super_res_last,
						    "prior_scheduler": prior_scheduler,
						    "decoder_scheduler": decoder_scheduler,
						    "super_res_scheduler": super_res_scheduler,
						}

						return components







		def        a_       (    self					:				List[Any]			,      a__					:				List[Any]			,      a__					:				List[str]=0						)				->  Dict:





						'''simple docstring'''







						if str(a__						).startswith("mps"						):
										_A         =	torch.manual_seed(a__						)
						else:
										_A         =	torch.Generator(device=a__						).manual_seed(a__						)
						_A         =	{
						    "prompt": "horse",
						    "generator": generator,
						    "prior_num_inference_steps": 2,
						    "decoder_num_inference_steps": 2,
						    "super_res_num_inference_steps": 2,
						    "output_type": "numpy",
						}
						return inputs







		def        a_       (    self					:				Any						)				->  Dict:





						'''simple docstring'''







						_A         =	"cpu"

						_A         =	self.get_dummy_components()

						_A         =	self.pipeline_class(**a__						)
						_A         =	pipe.to(a__						)

						pipe.set_progress_bar_config(disable=a__						)

						_A         =	pipe(**self.get_dummy_inputs(a__						)						)
						_A         =	output.images

						_A         =	pipe(
						    **self.get_dummy_inputs(a__						)			,      return_dict=a__			,      )[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.9_9_8_8,
						        0.0_0_2_8,
						        0.9_9_9_7,
						        0.9_9_8_4,
						        0.9_9_6_5,
						        0.0_0_2_9,
						        0.9_9_8_6,
						        0.0_0_2_5,
						    ]						)

						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        a_       (    self					:				Any						)				->  int:





						'''simple docstring'''







						_A         =	torch.device("cpu"						)
						class snake_case					:
								__UpperCamelCase							=	1





						_A         =	self.get_dummy_components()

						_A         =	self.pipeline_class(**a__						)
						_A         =	pipe.to(a__						)

						_A         =	components["prior"]
						_A         =	components["decoder"]
						_A         =	components["super_res_first"]
						_A         =	components["tokenizer"]
						_A         =	components["text_encoder"]

						_A         =	torch.Generator(device=a__						).manual_seed(0						)
						_A         =	prior.dtype
						_A         =	1

						_A         =	(batch_size, prior.config.embedding_dim)
						_A         =	pipe.prepare_latents(
						    a__			,      dtype=a__			,      device=a__			,      generator=a__			,      latents=a__			,      scheduler=DummyScheduler()						)
						_A         =	(batch_size, decoder.config.in_channels, decoder.config.sample_size, decoder.config.sample_size)
						_A         =	pipe.prepare_latents(
						    a__			,      dtype=a__			,      device=a__			,      generator=a__			,      latents=a__			,      scheduler=DummyScheduler()						)

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

						pipe.set_progress_bar_config(disable=a__						)

						_A         =	"this is a prompt example"

						_A         =	torch.Generator(device=a__						).manual_seed(0						)
						_A         =	pipe(
						    [prompt]			,      generator=a__			,      prior_num_inference_steps=2			,      decoder_num_inference_steps=2			,      super_res_num_inference_steps=2			,      prior_latents=a__			,      decoder_latents=a__			,      super_res_latents=a__			,      output_type="np"			,      )
						_A         =	output.images

						_A         =	tokenizer(
						    a__			,      padding="max_length"			,      max_length=tokenizer.model_max_length			,      return_tensors="pt"			,      )
						_A         =	text_encoder(text_inputs.input_ids						)
						_A         =	text_inputs.attention_mask

						_A         =	torch.Generator(device=a__						).manual_seed(0						)
						_A         =	pipe(
						    generator=a__			,      prior_num_inference_steps=2			,      decoder_num_inference_steps=2			,      super_res_num_inference_steps=2			,      prior_latents=a__			,      decoder_latents=a__			,      super_res_latents=a__			,      text_model_output=a__			,      text_attention_mask=a__			,      output_type="np"			,      )[0]

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







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





						'''simple docstring'''







						_A         =	torch_device == "cpu"

						self._test_attention_slicing_forward_pass(test_max_difference=a__			,      expected_max_diff=0.0_1						)







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





						'''simple docstring'''







						_A         =	torch_device == "cpu"
						_A         =	True
						_A         =	[
						    "prior_num_inference_steps",
						    "decoder_num_inference_steps",
						    "super_res_num_inference_steps",
						]

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







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





						'''simple docstring'''







						_A         =	[
						    "prior_num_inference_steps",
						    "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=a__			,      additional_params_copy_to_batched_inputs=a__			,      )
						else:
										self._test_inference_batch_consistent(
										    additional_params_copy_to_batched_inputs=a__						)







		@skip_mps
		def        a_       (    self					:				Dict						)				->  Union[str, Any]:





						'''simple docstring'''







						return super().test_dict_tuple_outputs_equivalent()







		@skip_mps
		def        a_       (    self					:				List[str]						)				->  Optional[int]:





						'''simple docstring'''







						return super().test_save_load_local()







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





						'''simple docstring'''







						return super().test_save_load_optional_components()





@nightly
class snake_case					(     unittest.TestCase):







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





						'''simple docstring'''







						super().tearDown()
						gc.collect()
						torch.cuda.empty_cache()







		def        a_       (    self					:				Any						)				->  Tuple:





						'''simple docstring'''







						_A         =	load_numpy(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
						    "/unclip/karlo_v1_alpha_horse_cpu.npy"						)

						_A         =	UnCLIPPipeline.from_pretrained("kakaobrain/karlo-v1-alpha"						)
						pipeline.set_progress_bar_config(disable=a__						)

						_A         =	torch.manual_seed(0						)
						_A         =	pipeline(
						    "horse"			,      num_images_per_prompt=1			,      generator=a__			,      output_type="np"			,      )

						_A         =	output.images[0]

						assert image.shape == (2_56, 2_56, 3)
						assert np.abs(expected_image - image						).max() < 1E-1





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







		def        a_       (    self					:				Any						)				->  List[Any]:





						'''simple docstring'''







						super().tearDown()
						gc.collect()
						torch.cuda.empty_cache()







		def        a_       (    self					:				Tuple						)				->  Tuple:





						'''simple docstring'''







						_A         =	load_numpy(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
						    "/unclip/karlo_v1_alpha_horse_fp16.npy"						)

						_A         =	UnCLIPPipeline.from_pretrained("kakaobrain/karlo-v1-alpha"			,      torch_dtype=torch.floataa						)
						_A         =	pipeline.to(a__						)
						pipeline.set_progress_bar_config(disable=a__						)

						_A         =	torch.Generator(device="cpu"						).manual_seed(0						)
						_A         =	pipeline(
						    "horse"			,      generator=a__			,      output_type="np"			,      )

						_A         =	output.images[0]

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

						assert_mean_pixel_difference(a__			,      a__						)







		def        a_       (    self					:				str						)				->  List[str]:





						'''simple docstring'''







						torch.cuda.empty_cache()
						torch.cuda.reset_max_memory_allocated()
						torch.cuda.reset_peak_memory_stats()

						_A         =	UnCLIPPipeline.from_pretrained("kakaobrain/karlo-v1-alpha"			,      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(
						    "horse"			,      num_images_per_prompt=1			,      prior_num_inference_steps=2			,      decoder_num_inference_steps=2			,      super_res_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



"""simple docstring"""


import math
from decimal import Decimal, getcontext






def    a__       ( __lowercase = 400_0000      )		->		int:

				try:
								_A         =	int(__lowercase      )
				except (TypeError, ValueError):
								raise TypeError("Parameter n must be int or castable to int."      )
				if n <= 0:
								raise ValueError("Parameter n must be greater than or equal to one."      )
				_A         =	100
				_A         =	(Decimal(5      ) ** Decimal(0.5      ) + 1) / Decimal(2      )

				_A         =	(math.floor(math.log(n * (phi + 2)   ,	__lowercase      ) - 1      ) // 3) * 3 + 2
				_A         =	Decimal(round(phi ** Decimal(index + 1      )      )      ) / (phi + 2)
				_A         =	num // 2
				return int(__lowercase      )


if __name__ == "__main__":
		print(f'''{solution() = }''')