import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()



class 					snake_case__       (				_UpperCamelCase					,						_UpperCamelCase					,						unittest.TestCase							):
   _SCREAMING_SNAKE_CASE	: List[Any]						    =      StableDiffusionXLPipeline
   _SCREAMING_SNAKE_CASE	: Dict						    =      TEXT_TO_IMAGE_PARAMS
   _SCREAMING_SNAKE_CASE	: Union[str, Any]						    =      TEXT_TO_IMAGE_BATCH_PARAMS
   _SCREAMING_SNAKE_CASE	: List[str]						    =      TEXT_TO_IMAGE_IMAGE_PARAMS
   _SCREAMING_SNAKE_CASE	: str						    =      TEXT_TO_IMAGE_IMAGE_PARAMS



   def 					UpperCAmelCase__       (				self   :   Dict   )							->     Any:




         '''simple docstring'''




         torch.manual_seed(0   )
         snake_case_     :			Optional[Any]       =       UNetaDConditionModel(
             block_out_channels=(32, 64)			,	layers_per_block=2			,	sample_size=32			,	in_channels=4			,	out_channels=4			,	down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")			,	up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")			,	attention_head_dim=(2, 4)			,	use_linear_projection=A__			,	addition_embed_type="text_time"			,	addition_time_embed_dim=8			,	transformer_layers_per_block=(1, 2)			,	projection_class_embeddings_input_dim=80			,	cross_attention_dim=64			,	)
         snake_case_     :			str       =       EulerDiscreteScheduler(
             beta_start=0.0_0085			,	beta_end=0.012			,	steps_offset=1			,	beta_schedule="scaled_linear"			,	timestep_spacing="leading"			,	)
         torch.manual_seed(0   )
         snake_case_     :			int       =       AutoencoderKL(
             block_out_channels=[32, 64]			,	in_channels=3			,	out_channels=3			,	down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]			,	up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]			,	latent_channels=4			,	sample_size=1_28			,	)
         torch.manual_seed(0   )
         snake_case_     :			str       =       CLIPTextConfig(
             bos_token_id=0			,	eos_token_id=2			,	hidden_size=32			,	intermediate_size=37			,	layer_norm_eps=1E-05			,	num_attention_heads=4			,	num_hidden_layers=5			,	pad_token_id=1			,	vocab_size=10_00			,	hidden_act="gelu"			,	projection_dim=32			,	)
         snake_case_     :			List[Any]       =       CLIPTextModel(A__   )
         snake_case_     :			Any       =       CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"			,	local_files_only=A__   )

         snake_case_     :			int       =       CLIPTextModelWithProjection(A__   )
         snake_case_     :			Tuple       =       CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"			,	local_files_only=A__   )

         snake_case_     :			Dict       =       {
             "unet": unet,
             "scheduler": scheduler,
             "vae": vae,
             "text_encoder": text_encoder,
             "tokenizer": tokenizer,
             "text_encoder_2": text_encoder_a,
             "tokenizer_2": tokenizer_a,
             # "safety_checker": None,
             # "feature_extractor": None,
         }
         return components



   def 					UpperCAmelCase__       (				self   :   Any			,	A__   :   List[Any]			,	A__   :   Any=0   )							->     Optional[int]:




         '''simple docstring'''




         if str(A__   ).startswith("mps"   ):
               snake_case_     :			str       =       torch.manual_seed(A__   )
         else:
               snake_case_     :			Union[str, Any]       =       torch.Generator(device=A__   ).manual_seed(A__   )
         snake_case_     :			str       =       {
             "prompt": "A painting of a squirrel eating a burger",
             "generator": generator,
             "num_inference_steps": 2,
             "guidance_scale": 5.0,
             "output_type": "numpy",
         }
         return inputs



   def 					UpperCAmelCase__       (				self   :   Tuple   )							->     Any:




         '''simple docstring'''




         snake_case_     :			List[Any]       =       "cpu"  # ensure determinism for the device-dependent torch.Generator
         snake_case_     :			int       =       self.get_dummy_components()
         snake_case_     :			Any       =       StableDiffusionXLPipeline(**A__   )
         snake_case_     :			Tuple       =       sd_pipe.to(A__   )
         sd_pipe.set_progress_bar_config(disable=A__   )

         snake_case_     :			int       =       self.get_dummy_inputs(A__   )
         snake_case_     :			str       =       sd_pipe(**A__   ).images
         snake_case_     :			Optional[Any]       =       image[0, -3:, -3:, -1]

         assert image.shape == (1, 64, 64, 3)
         snake_case_     :			Optional[Any]       =       np.array([0.5873, 0.6128, 0.4797, 0.5122, 0.5674, 0.4639, 0.5227, 0.5149, 0.4747]   )

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



   def 					UpperCAmelCase__       (				self   :   List[str]   )							->     Dict:




         '''simple docstring'''




         snake_case_     :			List[Any]       =       self.get_dummy_components()
         snake_case_     :			Optional[int]       =       StableDiffusionXLPipeline(**A__   )
         snake_case_     :			str       =       sd_pipe.to(A__   )
         snake_case_     :			str       =       sd_pipe.to(A__   )
         sd_pipe.set_progress_bar_config(disable=A__   )

         # forward without prompt embeds
         snake_case_     :			str       =       self.get_dummy_inputs(A__   )
         snake_case_     :			Union[str, Any]       =       3 * ["this is a negative prompt"]
         snake_case_     :			List[Any]       =       negative_prompt
         snake_case_     :			Tuple       =       3 * [inputs["prompt"]]

         snake_case_     :			Union[str, Any]       =       sd_pipe(**A__   )
         snake_case_     :			Union[str, Any]       =       output.images[0, -3:, -3:, -1]

         # forward with prompt embeds
         snake_case_     :			str       =       self.get_dummy_inputs(A__   )
         snake_case_     :			str       =       3 * ["this is a negative prompt"]
         snake_case_     :			int       =       3 * [inputs.pop("prompt"   )]

         (
             (
             snake_case_
         )  ,(
             snake_case_
         )  ,(
             snake_case_
         )  ,(
             snake_case_
         )  ,
         )     :			Optional[Any]       =       sd_pipe.encode_prompt(A__			,	negative_prompt=A__   )

         snake_case_     :			Any       =       sd_pipe(
             **A__			,	prompt_embeds=A__			,	negative_prompt_embeds=A__			,	pooled_prompt_embeds=A__			,	negative_pooled_prompt_embeds=A__			,	)
         snake_case_     :			str       =       output.images[0, -3:, -3:, -1]

         # make sure that it's equal
         assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()   ).max() < 1E-4



   def 					UpperCAmelCase__       (				self   :   str   )							->     int:




         '''simple docstring'''




         super().test_attention_slicing_forward_pass(expected_max_diff=3E-3   )



   def 					UpperCAmelCase__       (				self   :   str   )							->     Any:




         '''simple docstring'''




         super().test_inference_batch_single_identical(expected_max_diff=3E-3   )



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



   def 					UpperCAmelCase__       (				self   :   Dict   )							->     Optional[int]:




         '''simple docstring'''




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



   def 					UpperCAmelCase__       (				self   :   Optional[int]			,	A__   :   Optional[int]			,	A__   :   List[Any]="cpu"			,	A__   :   Optional[Any]=torch.floataa			,	A__   :   Optional[int]=0   )							->     Tuple:




         '''simple docstring'''




         snake_case_     :			Any       =       torch.Generator(device=A__   ).manual_seed(A__   )
         snake_case_     :			Union[str, Any]       =       np.random.RandomState(A__   ).standard_normal((1, 4, 64, 64)   )
         snake_case_     :			str       =       torch.from_numpy(A__   ).to(device=A__			,	dtype=A__   )
         snake_case_     :			List[Any]       =       {
             "prompt": "a photograph of an astronaut riding a horse",
             "latents": latents,
             "generator": generator,
             "num_inference_steps": 3,
             "guidance_scale": 7.5,
             "output_type": "numpy",
         }
         return inputs



   def 					UpperCAmelCase__       (				self   :   int   )							->     str:




         '''simple docstring'''




         snake_case_     :			List[Any]       =       DiffusionPipeline.from_pretrained("stabilityai/stable-diffusion-2-base"   )
         pipe.to(A__   )
         pipe.set_progress_bar_config(disable=A__   )

         snake_case_     :			Optional[Any]       =       self.get_inputs(A__   )
         snake_case_     :			str       =       pipe(**A__   ).images
         snake_case_     :			Optional[int]       =       image[0, -3:, -3:, -1].flatten()

         assert image.shape == (1, 5_12, 5_12, 3)
         snake_case_     :			Optional[int]       =       np.array([0.4_9493, 0.4_7896, 0.4_0798, 0.5_4214, 0.5_3212, 0.4_8202, 0.4_7656, 0.4_6329, 0.4_8506]   )
         assert np.abs(image_slice - expected_slice   ).max() < 7E-3




import random
from typing import List, Optional, Union

import numpy as np
import torch

from ..state import AcceleratorState
from .constants import CUDA_DISTRIBUTED_TYPES
from .dataclasses import DistributedType, RNGType
from .imports import is_npu_available, is_tpu_available, is_xpu_available


if is_tpu_available(check_device=False):
     import torch_xla.core.xla_model as xm






def 					SCREAMING_SNAKE_CASE_			(		lowerCAmelCase_: int ,	lowerCAmelCase_: bool = False     ):
      if device_specific:
            seed += AcceleratorState().process_index
      random.seed(lowerCAmelCase_     )
      np.random.seed(lowerCAmelCase_     )
      torch.manual_seed(lowerCAmelCase_     )
      if is_xpu_available():
            torch.xpu.manual_seed_all(lowerCAmelCase_     )
      elif is_npu_available():
            torch.npu.manual_seed_all(lowerCAmelCase_     )
      else:
            torch.cuda.manual_seed_all(lowerCAmelCase_     )
      # ^^ safe to call this function even if cuda is not available
      if is_tpu_available():
            xm.set_rng_state(lowerCAmelCase_     )






def 					SCREAMING_SNAKE_CASE_			(		lowerCAmelCase_: Optional[RNGType] = None ,	lowerCAmelCase_: Optional[torch.Generator] = None     ):
      # Get the proper rng state
      if rng_type == RNGType.TORCH:
            snake_case_     :			List[Any]       =       torch.get_rng_state()
      elif rng_type == RNGType.CUDA:
            snake_case_     :			Dict       =       torch.cuda.get_rng_state()
      elif rng_type == RNGType.XLA:
            assert is_tpu_available(), "Can't synchronize XLA seeds on an environment without TPUs."
            snake_case_     :			Tuple       =       torch.tensor(xm.get_rng_state()     )
      elif rng_type == RNGType.NPU:
            assert is_npu_available(), "Can't synchronize NPU seeds on an environment without NPUs."
            snake_case_     :			List[str]       =       torch.npu.get_rng_state()
      elif rng_type == RNGType.XPU:
            assert is_xpu_available(), "Can't synchronize XPU seeds on an environment without XPUs."
            snake_case_     :			List[str]       =       torch.xpu.get_rng_state()
      elif rng_type == RNGType.GENERATOR:
            assert generator is not None, "Need a generator to synchronize its seed."
            snake_case_     :			Optional[Any]       =       generator.get_state()

      # Broadcast the rng state from device 0 to other devices
      snake_case_     :			List[Any]       =       AcceleratorState()
      if state.distributed_type == DistributedType.TPU:
            snake_case_     :			int       =       rng_state.to(xm.xla_device()     )
            xm.collective_broadcast([rng_state]     )
            xm.mark_step()
            snake_case_     :			List[str]       =       rng_state.cpu()
      elif (
          state.distributed_type in CUDA_DISTRIBUTED_TYPES
          or state.distributed_type == DistributedType.MULTI_NPU
          or state.distributed_type == DistributedType.MULTI_XPU
      ):
            snake_case_     :			Dict       =       rng_state.to(state.device     )
            torch.distributed.broadcast(lowerCAmelCase_ ,	0     )
            snake_case_     :			Dict       =       rng_state.cpu()
      elif state.distributed_type == DistributedType.MULTI_CPU:
            torch.distributed.broadcast(lowerCAmelCase_ ,	0     )

      # Set the broadcast rng state
      if rng_type == RNGType.TORCH:
            torch.set_rng_state(lowerCAmelCase_     )
      elif rng_type == RNGType.CUDA:
            torch.cuda.set_rng_state(lowerCAmelCase_     )
      elif rng_type == RNGType.NPU:
            torch.npu.set_rng_state(lowerCAmelCase_     )
      elif rng_type == RNGType.XPU:
            torch.xpu.set_rng_state(lowerCAmelCase_     )
      elif rng_type == RNGType.XLA:
            xm.set_rng_state(rng_state.item()     )
      elif rng_type == RNGType.GENERATOR:
            generator.set_state(lowerCAmelCase_     )






def 					SCREAMING_SNAKE_CASE_			(		lowerCAmelCase_: List[Union[str, RNGType]] ,	lowerCAmelCase_: Optional[torch.Generator] = None     ):
      for rng_type in rng_types:
            synchronize_rng_state(RNGType(lowerCAmelCase_     ) ,	generator=lowerCAmelCase_     )



