"""simple docstring"""






from dataclasses import dataclass
from typing import Dict, Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, apply_forward_hook
from .attention_processor import AttentionProcessor, AttnProcessor
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, DiagonalGaussianDistribution, Encoder
@dataclass
class 				snake_case       ( __UpperCAmelCase				):




					"""simple docstring"""

					snake_case__				    = 42






class 				snake_case       ( __UpperCAmelCase      ,    __UpperCAmelCase				):




					"""simple docstring"""

					snake_case__				    = True







					@register_to_config
					def __init__(				self   :	Dict    ,lowerCamelCase__   :	int = 3    ,lowerCamelCase__   :	int = 3    ,lowerCamelCase__   :	Tuple[str] = ("DownEncoderBlock2D",)    ,lowerCamelCase__   :	Tuple[str] = ("UpDecoderBlock2D",)    ,lowerCamelCase__   :	Tuple[int] = (64,)    ,lowerCamelCase__   :	int = 1    ,lowerCamelCase__   :	str = "silu"    ,lowerCamelCase__   :	int = 4    ,lowerCamelCase__   :	int = 32    ,lowerCamelCase__   :	int = 32    ,lowerCamelCase__   :	float = 0.1_8_2_1_5    ,):
									super().__init__()

									# pass init params to Encoder
									UpperCAmelCase__             =							Encoder(
									    in_channels=lowerCamelCase__    ,out_channels=lowerCamelCase__    ,down_block_types=lowerCamelCase__    ,block_out_channels=lowerCamelCase__    ,layers_per_block=lowerCamelCase__    ,act_fn=lowerCamelCase__    ,norm_num_groups=lowerCamelCase__    ,double_z=lowerCamelCase__    ,)

									# pass init params to Decoder
									UpperCAmelCase__             =							Decoder(
									    in_channels=lowerCamelCase__    ,out_channels=lowerCamelCase__    ,up_block_types=lowerCamelCase__    ,block_out_channels=lowerCamelCase__    ,layers_per_block=lowerCamelCase__    ,norm_num_groups=lowerCamelCase__    ,act_fn=lowerCamelCase__    ,)

									UpperCAmelCase__             =							nn.Convad(2 * latent_channels    ,2 * latent_channels    ,1			)
									UpperCAmelCase__             =							nn.Convad(lowerCamelCase__    ,lowerCamelCase__    ,1			)

									UpperCAmelCase__             =							False
									UpperCAmelCase__             =							False

									# only relevant if vae tiling is enabled
									UpperCAmelCase__             =							self.config.sample_size
									UpperCAmelCase__             =							(
									    self.config.sample_size[0]
									    if isinstance(self.config.sample_size    ,(list, tuple)			)
									    else self.config.sample_size
									)
									UpperCAmelCase__             =							int(sample_size / (2 ** (len(self.config.block_out_channels			) - 1))			)
									UpperCAmelCase__             =							0.2_5







					def 		__lowerCAmelCase						(				self   :	int    ,lowerCamelCase__   :	Optional[Any]    ,lowerCamelCase__   :	int=False			):
									if isinstance(lowerCamelCase__    ,(Encoder, Decoder)			):
													UpperCAmelCase__             =							value







					def 		__lowerCAmelCase						(				self   :	Optional[int]    ,lowerCamelCase__   :	bool = True			):
									UpperCAmelCase__             =							use_tiling







					def 		__lowerCAmelCase						(				self   :	List[Any]			):
									self.enable_tiling(lowerCamelCase__			)







					def 		__lowerCAmelCase						(				self   :	Any			):
									UpperCAmelCase__             =							True







					def 		__lowerCAmelCase						(				self   :	Union[str, Any]			):
									UpperCAmelCase__             =							False







					@property
					# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
					def 		__lowerCAmelCase						(				self   :	int			):
									UpperCAmelCase__             =							{}

									def fn_recursive_add_processors(lowerCamelCase__   :	str    ,lowerCamelCase__   :	torch.nn.Module    ,lowerCamelCase__   :	Dict[str, AttentionProcessor]			):
													if hasattr(lowerCamelCase__    ,'set_processor'			):
																	UpperCAmelCase__             =							module.processor

													for sub_name, child in module.named_children():
																	fn_recursive_add_processors(f'''{name}.{sub_name}'''    ,lowerCamelCase__    ,lowerCamelCase__			)

													return processors

									for name, module in self.named_children():
													fn_recursive_add_processors(lowerCamelCase__    ,lowerCamelCase__    ,lowerCamelCase__			)

									return processors







					def 		__lowerCAmelCase						(				self   :	Union[str, Any]    ,lowerCamelCase__   :	Union[AttentionProcessor, Dict[str, AttentionProcessor]]			):
									UpperCAmelCase__             =							len(self.attn_processors.keys()			)

									if isinstance(lowerCamelCase__    ,lowerCamelCase__			) and len(lowerCamelCase__			) != count:
													raise ValueError(
													    f'''A dict of processors was passed, but the number of processors {len(lowerCamelCase__			)} does not match the'''
													    f''' number of attention layers: {count}. Please make sure to pass {count} processor classes.'''			)

									def fn_recursive_attn_processor(lowerCamelCase__   :	str    ,lowerCamelCase__   :	torch.nn.Module    ,lowerCamelCase__   :	str			):
													if hasattr(lowerCamelCase__    ,'set_processor'			):
																	if not isinstance(lowerCamelCase__    ,lowerCamelCase__			):
																					module.set_processor(lowerCamelCase__			)
																	else:
																					module.set_processor(processor.pop(f'''{name}.processor'''			)			)

													for sub_name, child in module.named_children():
																	fn_recursive_attn_processor(f'''{name}.{sub_name}'''    ,lowerCamelCase__    ,lowerCamelCase__			)

									for name, module in self.named_children():
													fn_recursive_attn_processor(lowerCamelCase__    ,lowerCamelCase__    ,lowerCamelCase__			)







					def 		__lowerCAmelCase						(				self   :	int			):
									self.set_attn_processor(AttnProcessor()			)







					@apply_forward_hook
					def 		__lowerCAmelCase						(				self   :	Dict    ,lowerCamelCase__   :	torch.FloatTensor    ,lowerCamelCase__   :	bool = True			):
									if self.use_tiling and (x.shape[-1] > self.tile_sample_min_size or x.shape[-2] > self.tile_sample_min_size):
													return self.tiled_encode(lowerCamelCase__    ,return_dict=lowerCamelCase__			)

									if self.use_slicing and x.shape[0] > 1:
													UpperCAmelCase__             =							[self.encoder(lowerCamelCase__			) for x_slice in x.split(1			)]
													UpperCAmelCase__             =							torch.cat(lowerCamelCase__			)
									else:
													UpperCAmelCase__             =							self.encoder(lowerCamelCase__			)

									UpperCAmelCase__             =							self.quant_conv(lowerCamelCase__			)
									UpperCAmelCase__             =							DiagonalGaussianDistribution(lowerCamelCase__			)

									if not return_dict:
													return (posterior,)

									return AutoencoderKLOutput(latent_dist=lowerCamelCase__			)







					def 		__lowerCAmelCase						(				self   :	List[Any]    ,lowerCamelCase__   :	torch.FloatTensor    ,lowerCamelCase__   :	bool = True			):
									if self.use_tiling and (z.shape[-1] > self.tile_latent_min_size or z.shape[-2] > self.tile_latent_min_size):
													return self.tiled_decode(lowerCamelCase__    ,return_dict=lowerCamelCase__			)

									UpperCAmelCase__             =							self.post_quant_conv(lowerCamelCase__			)
									UpperCAmelCase__             =							self.decoder(lowerCamelCase__			)

									if not return_dict:
													return (dec,)

									return DecoderOutput(sample=lowerCamelCase__			)







					@apply_forward_hook
					def 		__lowerCAmelCase						(				self   :	Dict    ,lowerCamelCase__   :	torch.FloatTensor    ,lowerCamelCase__   :	bool = True			):
									if self.use_slicing and z.shape[0] > 1:
													UpperCAmelCase__             =							[self._decode(lowerCamelCase__			).sample for z_slice in z.split(1			)]
													UpperCAmelCase__             =							torch.cat(lowerCamelCase__			)
									else:
													UpperCAmelCase__             =							self._decode(lowerCamelCase__			).sample

									if not return_dict:
													return (decoded,)

									return DecoderOutput(sample=lowerCamelCase__			)







					def 		__lowerCAmelCase						(				self   :	Optional[int]    ,lowerCamelCase__   :	Optional[int]    ,lowerCamelCase__   :	List[Any]    ,lowerCamelCase__   :	Optional[int]			):
									UpperCAmelCase__             =							min(a.shape[2]    ,b.shape[2]    ,lowerCamelCase__			)
									for y in range(lowerCamelCase__			):
													UpperCAmelCase__             =							a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
									return b







					def 		__lowerCAmelCase						(				self   :	Any    ,lowerCamelCase__   :	int    ,lowerCamelCase__   :	Union[str, Any]    ,lowerCamelCase__   :	Union[str, Any]			):
									UpperCAmelCase__             =							min(a.shape[3]    ,b.shape[3]    ,lowerCamelCase__			)
									for x in range(lowerCamelCase__			):
													UpperCAmelCase__             =							a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
									return b







					def 		__lowerCAmelCase						(				self   :	str    ,lowerCamelCase__   :	torch.FloatTensor    ,lowerCamelCase__   :	bool = True			):
									UpperCAmelCase__             =							int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)			)
									UpperCAmelCase__             =							int(self.tile_latent_min_size * self.tile_overlap_factor			)
									UpperCAmelCase__             =							self.tile_latent_min_size - blend_extent

									# Split the image into 512x512 tiles and encode them separately.
									UpperCAmelCase__             =							[]
									for i in range(0    ,x.shape[2]    ,lowerCamelCase__			):
													UpperCAmelCase__             =							[]
													for j in range(0    ,x.shape[3]    ,lowerCamelCase__			):
																	UpperCAmelCase__             =							x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
																	UpperCAmelCase__             =							self.encoder(lowerCamelCase__			)
																	UpperCAmelCase__             =							self.quant_conv(lowerCamelCase__			)
																	row.append(lowerCamelCase__			)
													rows.append(lowerCamelCase__			)
									UpperCAmelCase__             =							[]
									for i, row in enumerate(lowerCamelCase__			):
													UpperCAmelCase__             =							[]
													for j, tile in enumerate(lowerCamelCase__			):
																	# blend the above tile and the left tile
																	# to the current tile and add the current tile to the result row
																	if i > 0:
																					UpperCAmelCase__             =							self.blend_v(rows[i - 1][j]    ,lowerCamelCase__    ,lowerCamelCase__			)
																	if j > 0:
																					UpperCAmelCase__             =							self.blend_h(row[j - 1]    ,lowerCamelCase__    ,lowerCamelCase__			)
																	result_row.append(tile[:, :, :row_limit, :row_limit]			)
													result_rows.append(torch.cat(lowerCamelCase__    ,dim=3			)			)

									UpperCAmelCase__             =							torch.cat(lowerCamelCase__    ,dim=2			)
									UpperCAmelCase__             =							DiagonalGaussianDistribution(lowerCamelCase__			)

									if not return_dict:
													return (posterior,)

									return AutoencoderKLOutput(latent_dist=lowerCamelCase__			)







					def 		__lowerCAmelCase						(				self   :	List[Any]    ,lowerCamelCase__   :	torch.FloatTensor    ,lowerCamelCase__   :	bool = True			):
									UpperCAmelCase__             =							int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)			)
									UpperCAmelCase__             =							int(self.tile_sample_min_size * self.tile_overlap_factor			)
									UpperCAmelCase__             =							self.tile_sample_min_size - blend_extent

									# Split z into overlapping 64x64 tiles and decode them separately.
									# The tiles have an overlap to avoid seams between tiles.
									UpperCAmelCase__             =							[]
									for i in range(0    ,z.shape[2]    ,lowerCamelCase__			):
													UpperCAmelCase__             =							[]
													for j in range(0    ,z.shape[3]    ,lowerCamelCase__			):
																	UpperCAmelCase__             =							z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
																	UpperCAmelCase__             =							self.post_quant_conv(lowerCamelCase__			)
																	UpperCAmelCase__             =							self.decoder(lowerCamelCase__			)
																	row.append(lowerCamelCase__			)
													rows.append(lowerCamelCase__			)
									UpperCAmelCase__             =							[]
									for i, row in enumerate(lowerCamelCase__			):
													UpperCAmelCase__             =							[]
													for j, tile in enumerate(lowerCamelCase__			):
																	# blend the above tile and the left tile
																	# to the current tile and add the current tile to the result row
																	if i > 0:
																					UpperCAmelCase__             =							self.blend_v(rows[i - 1][j]    ,lowerCamelCase__    ,lowerCamelCase__			)
																	if j > 0:
																					UpperCAmelCase__             =							self.blend_h(row[j - 1]    ,lowerCamelCase__    ,lowerCamelCase__			)
																	result_row.append(tile[:, :, :row_limit, :row_limit]			)
													result_rows.append(torch.cat(lowerCamelCase__    ,dim=3			)			)

									UpperCAmelCase__             =							torch.cat(lowerCamelCase__    ,dim=2			)
									if not return_dict:
													return (dec,)

									return DecoderOutput(sample=lowerCamelCase__			)







					def 		__lowerCAmelCase						(				self   :	List[Any]    ,lowerCamelCase__   :	torch.FloatTensor    ,lowerCamelCase__   :	bool = False    ,lowerCamelCase__   :	bool = True    ,lowerCamelCase__   :	Optional[torch.Generator] = None    ,):
									UpperCAmelCase__             =							sample
									UpperCAmelCase__             =							self.encode(lowerCamelCase__			).latent_dist
									if sample_posterior:
													UpperCAmelCase__             =							posterior.sample(generator=lowerCamelCase__			)
									else:
													UpperCAmelCase__             =							posterior.mode()
									UpperCAmelCase__             =							self.decode(lowerCamelCase__			).sample

									if not return_dict:
													return (dec,)

									return DecoderOutput(sample=lowerCamelCase__			)




"""simple docstring"""






import gc
import random
import unittest

import numpy as np
import torch
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModel

from diffusers import HeunDiscreteScheduler, PriorTransformer, ShapEImgaImgPipeline
from diffusers.pipelines.shap_e import ShapERenderer
from diffusers.utils import floats_tensor, load_image, load_numpy, slow
from diffusers.utils.testing_utils import require_torch_gpu, torch_device

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference
class 				snake_case       ( __UpperCAmelCase      ,    unittest.TestCase				):




					"""simple docstring"""

					snake_case__				    = ShapEImgaImgPipeline
					snake_case__				    = ["image"]
					snake_case__				    = ["image"]
					snake_case__				    = [
					    "num_images_per_prompt",
					    "num_inference_steps",
					    "generator",
					    "latents",
					    "guidance_scale",
					    "frame_size",
					    "output_type",
					    "return_dict",
					]
					snake_case__				    = False







					@property
					def 		__lowerCAmelCase						(				self   :	List[str]			):
									return 32







					@property
					def 		__lowerCAmelCase						(				self   :	str			):
									return 32







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







					@property
					def 		__lowerCAmelCase						(				self   :	List[Any]			):
									return 8







					@property
					def 		__lowerCAmelCase						(				self   :	Optional[int]			):
									torch.manual_seed(0			)
									UpperCAmelCase__             =							CLIPVisionConfig(
									    hidden_size=self.text_embedder_hidden_size    ,image_size=64    ,projection_dim=self.text_embedder_hidden_size    ,intermediate_size=37    ,num_attention_heads=4    ,num_channels=3    ,num_hidden_layers=5    ,patch_size=1    ,)

									UpperCAmelCase__             =							CLIPVisionModel(lowerCamelCase__			)
									return model







					@property
					def 		__lowerCAmelCase						(				self   :	Optional[Any]			):
									UpperCAmelCase__             =							CLIPImageProcessor(
									    crop_size=224    ,do_center_crop=lowerCamelCase__    ,do_normalize=lowerCamelCase__    ,do_resize=lowerCamelCase__    ,image_mean=[0.4_8_1_4_5_4_6_6, 0.4_5_7_8_2_7_5, 0.4_0_8_2_1_0_7_3]    ,image_std=[0.2_6_8_6_2_9_5_4, 0.2_6_1_3_0_2_5_8, 0.2_7_5_7_7_7_1_1]    ,resample=3    ,size=224    ,)

									return image_processor







					@property
					def 		__lowerCAmelCase						(				self   :	str			):
									torch.manual_seed(0			)

									UpperCAmelCase__             =							{
									    'num_attention_heads': 2,
									    'attention_head_dim': 16,
									    'embedding_dim': self.time_input_dim,
									    'num_embeddings': 32,
									    'embedding_proj_dim': self.text_embedder_hidden_size,
									    'time_embed_dim': self.time_embed_dim,
									    'num_layers': 1,
									    'clip_embed_dim': self.time_input_dim * 2,
									    'additional_embeddings': 0,
									    'time_embed_act_fn': 'gelu',
									    'norm_in_type': 'layer',
									    'embedding_proj_norm_type': 'layer',
									    'encoder_hid_proj_type': None,
									    'added_emb_type': None,
									}

									UpperCAmelCase__             =							PriorTransformer(**lowerCamelCase__			)
									return model







					@property
					def 		__lowerCAmelCase						(				self   :	Tuple			):
									torch.manual_seed(0			)

									UpperCAmelCase__             =							{
									    'param_shapes': (
									        (self.renderer_dim, 93),
									        (self.renderer_dim, 8),
									        (self.renderer_dim, 8),
									        (self.renderer_dim, 8),
									    ),
									    'd_latent': self.time_input_dim,
									    'd_hidden': self.renderer_dim,
									    'n_output': 12,
									    'background': (
									        0.1,
									        0.1,
									        0.1,
									    ),
									}
									UpperCAmelCase__             =							ShapERenderer(**lowerCamelCase__			)
									return model







					def 		__lowerCAmelCase						(				self   :	Any			):
									UpperCAmelCase__             =							self.dummy_prior
									UpperCAmelCase__             =							self.dummy_image_encoder
									UpperCAmelCase__             =							self.dummy_image_processor
									UpperCAmelCase__             =							self.dummy_renderer

									UpperCAmelCase__             =							HeunDiscreteScheduler(
									    beta_schedule='exp'    ,num_train_timesteps=1_024    ,prediction_type='sample'    ,use_karras_sigmas=lowerCamelCase__    ,clip_sample=lowerCamelCase__    ,clip_sample_range=1.0    ,)
									UpperCAmelCase__             =							{
									    'prior': prior,
									    'image_encoder': image_encoder,
									    'image_processor': image_processor,
									    'renderer': renderer,
									    'scheduler': scheduler,
									}

									return components







					def 		__lowerCAmelCase						(				self   :	Optional[int]    ,lowerCamelCase__   :	Any    ,lowerCamelCase__   :	str=0			):
									UpperCAmelCase__             =							floats_tensor((1, 3, 64, 64)    ,rng=random.Random(lowerCamelCase__			)			).to(lowerCamelCase__			)

									if str(lowerCamelCase__			).startswith('mps'			):
													UpperCAmelCase__             =							torch.manual_seed(lowerCamelCase__			)
									else:
													UpperCAmelCase__             =							torch.Generator(device=lowerCamelCase__			).manual_seed(lowerCamelCase__			)
									UpperCAmelCase__             =							{
									    'image': input_image,
									    'generator': generator,
									    'num_inference_steps': 1,
									    'frame_size': 32,
									    'output_type': 'np',
									}
									return inputs







					def 		__lowerCAmelCase						(				self   :	Optional[int]			):
									UpperCAmelCase__             =							'cpu'

									UpperCAmelCase__             =							self.get_dummy_components()

									UpperCAmelCase__             =							self.pipeline_class(**lowerCamelCase__			)
									UpperCAmelCase__             =							pipe.to(lowerCamelCase__			)

									pipe.set_progress_bar_config(disable=lowerCamelCase__			)

									UpperCAmelCase__             =							pipe(**self.get_dummy_inputs(lowerCamelCase__			)			)
									UpperCAmelCase__             =							output.images[0]
									UpperCAmelCase__             =							image[0, -3:, -3:, -1]

									assert image.shape == (20, 32, 32, 3)

									UpperCAmelCase__             =							np.array(
									    [
									        0.0_0_0_3_9_2_1_6,
									        0.0_0_0_3_9_2_1_6,
									        0.0_0_0_3_9_2_1_6,
									        0.0_0_0_3_9_2_1_6,
									        0.0_0_0_3_9_2_1_6,
									        0.0_0_0_3_9_2_1_6,
									        0.0_0_0_3_9_2_1_6,
									        0.0_0_0_3_9_2_1_6,
									        0.0_0_0_3_9_2_1_6,
									    ]			)

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







					def 		__lowerCAmelCase						(				self   :	Tuple			):
									# NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
									self._test_inference_batch_consistent(batch_sizes=[1, 2]			)







					def 		__lowerCAmelCase						(				self   :	Tuple			):
									UpperCAmelCase__             =							torch_device == 'cpu'
									UpperCAmelCase__             =							True
									self._test_inference_batch_single_identical(
									    batch_size=2    ,test_max_difference=lowerCamelCase__    ,relax_max_difference=lowerCamelCase__    ,)







					def 		__lowerCAmelCase						(				self   :	List[Any]			):
									UpperCAmelCase__             =							self.get_dummy_components()
									UpperCAmelCase__             =							self.pipeline_class(**lowerCamelCase__			)
									UpperCAmelCase__             =							pipe.to(lowerCamelCase__			)
									pipe.set_progress_bar_config(disable=lowerCamelCase__			)

									UpperCAmelCase__             =							1
									UpperCAmelCase__             =							2

									UpperCAmelCase__             =							self.get_dummy_inputs(lowerCamelCase__			)

									for key in inputs.keys():
													if key in self.batch_params:
																	UpperCAmelCase__             =							batch_size * [inputs[key]]

									UpperCAmelCase__             =							pipe(**lowerCamelCase__    ,num_images_per_prompt=lowerCamelCase__			)[0]

									assert images.shape[0] == batch_size * num_images_per_prompt






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




					"""simple docstring"""

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







					def 		__lowerCAmelCase						(				self   :	Any			):
									UpperCAmelCase__             =							load_image(
									    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main' '/shap_e/corgi.png'			)
									UpperCAmelCase__             =							load_numpy(
									    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
									    '/shap_e/test_shap_e_img2img_out.npy'			)
									UpperCAmelCase__             =							ShapEImgaImgPipeline.from_pretrained('openai/shap-e-img2img'			)
									UpperCAmelCase__             =							pipe.to(lowerCamelCase__			)
									pipe.set_progress_bar_config(disable=lowerCamelCase__			)

									UpperCAmelCase__             =							torch.Generator(device=lowerCamelCase__			).manual_seed(0			)

									UpperCAmelCase__             =							pipe(
									    lowerCamelCase__    ,generator=lowerCamelCase__    ,guidance_scale=3.0    ,num_inference_steps=64    ,frame_size=64    ,output_type='np'    ,).images[0]

									assert images.shape == (20, 64, 64, 3)

									assert_mean_pixel_difference(lowerCamelCase__    ,lowerCamelCase__			)



