





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 							__SCREAMING_SNAKE_CASE       (  _a ,		unittest.TestCase							):
							snake_case  :   int      =							ShapEImgaImgPipeline
							snake_case  :   Optional[Any]      =							["""image"""]
							snake_case  :   List[Any]      =							["""image"""]
							snake_case  :   Optional[Any]      =							[
							    """num_images_per_prompt""",
							    """num_inference_steps""",
							    """generator""",
							    """latents""",
							    """guidance_scale""",
							    """frame_size""",
							    """output_type""",
							    """return_dict""",
							]
							snake_case  :   str      =							False






							@property
							def 					_lowerCamelCase		(    self	):
														return 32






							@property
							def 					_lowerCamelCase		(    self	):
														return 32






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






							@property
							def 					_lowerCamelCase		(    self	):
														return 8






							@property
							def 					_lowerCamelCase		(    self	):
														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	):
														UpperCamelCase__					  =      CLIPImageProcessor(
														    crop_size=224	,			do_center_crop=__lowerCAmelCase	,			do_normalize=__lowerCAmelCase	,			do_resize=__lowerCAmelCase	,			image_mean=[0.4814_5466, 0.457_8275, 0.4082_1073]	,			image_std=[0.2686_2954, 0.2613_0258, 0.2757_7711]	,			resample=3	,			size=224	,			)

														return image_processor






							@property
							def 					_lowerCamelCase		(    self	):
														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	):
														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	):
														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=1024	,			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	,			__lowerCAmelCase	,			__lowerCAmelCase=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	):
														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.0003_9216,
														        0.0003_9216,
														        0.0003_9216,
														        0.0003_9216,
														        0.0003_9216,
														        0.0003_9216,
														        0.0003_9216,
														        0.0003_9216,
														        0.0003_9216,
														    ]	)

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






							def 					_lowerCamelCase		(    self	):
														# 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	):
														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	):
														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 							__SCREAMING_SNAKE_CASE       (  unittest.TestCase							):






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







							def 					_lowerCamelCase		(    self	):
														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	)











# We ignore warnings about stepping the scheduler since we step it ourselves during gradient accumulation

import warnings

from .state import AcceleratorState, GradientState


warnings.filterwarnings("ignore", category=UserWarning, module="torch.optim.lr_scheduler")

class 							__SCREAMING_SNAKE_CASE       :






							def __init__(    self	,			__lowerCAmelCase	,			__lowerCAmelCase	,			__lowerCAmelCase = True	,			__lowerCAmelCase = False	):
														UpperCamelCase__					  =      scheduler
														UpperCamelCase__					  =      optimizers if isinstance(__lowerCAmelCase	,			(list, tuple)	) else [optimizers]
														UpperCamelCase__					  =      split_batches
														UpperCamelCase__					  =      step_with_optimizer
														UpperCamelCase__					  =      GradientState()






							def 					_lowerCamelCase		(    self	,			*__lowerCAmelCase	,			**__lowerCAmelCase	):
														if not self.step_with_optimizer:
																					# No link between scheduler and optimizer -> just step
																					self.scheduler.step(*__lowerCAmelCase	,			**__lowerCAmelCase	)
																					return

														# Otherwise, first make sure the optimizer was stepped.
														if not self.gradient_state.sync_gradients:
																					if self.gradient_state.adjust_scheduler:
																												self.scheduler._step_count += 1
																					return

														for opt in self.optimizers:
																					if opt.step_was_skipped:
																												return
														if self.split_batches:
																					# Split batches -> the training dataloader batch size is not changed so one step per training step
																					self.scheduler.step(*__lowerCAmelCase	,			**__lowerCAmelCase	)
														else:
																					# Otherwise the training dataloader batch size was multiplied by `num_processes`, so we need to do
																					# num_processes steps per training step
																					UpperCamelCase__					  =      AcceleratorState().num_processes
																					for _ in range(__lowerCAmelCase	):
																												# Special case when using OneCycle and `drop_last` was not used
																												if hasattr(self.scheduler	,			"""total_steps"""	):
																																			if self.scheduler._step_count <= self.scheduler.total_steps:
																																										self.scheduler.step(*__lowerCAmelCase	,			**__lowerCAmelCase	)
																												else:
																																			self.scheduler.step(*__lowerCAmelCase	,			**__lowerCAmelCase	)






							def 					_lowerCamelCase		(    self	):
														return self.scheduler.get_last_lr()






							def 					_lowerCamelCase		(    self	):
														return self.scheduler.state_dict()






							def 					_lowerCamelCase		(    self	,			__lowerCAmelCase	):
														self.scheduler.load_state_dict(__lowerCAmelCase	)






							def 					_lowerCamelCase		(    self	):
														return self.scheduler.get_lr()







							def 					_lowerCamelCase		(    self	,			*__lowerCAmelCase	,			**__lowerCAmelCase	):
														return self.scheduler.print_lr(*__lowerCAmelCase	,			**__lowerCAmelCase	)




