

"""simple docstring"""




import gc
import unittest

from diffusers import FlaxDPMSolverMultistepScheduler, FlaxStableDiffusionPipeline
from diffusers.utils import is_flax_available, slow
from diffusers.utils.testing_utils import require_flax


if is_flax_available():
      import jax
      import jax.numpy as jnp
      from flax.jax_utils import replicate
      from flax.training.common_utils import shard







@slow
@require_flax
class   __lowercase      (     unittest.TestCase       ):

  def  __lowercase							(	self		:      str     ):






   '''simple docstring'''
   super().tearDown()
   gc.collect()

  def  __lowercase							(	self		:      int     ):






   '''simple docstring'''
   UpperCAmelCase__       :	int        =       FlaxStableDiffusionPipeline.from_pretrained(
       """stabilityai/stable-diffusion-2"""       ,revision="""bf16"""       ,dtype=jnp.bfloataa       ,)

   UpperCAmelCase__       :	List[str]        =       "A painting of a squirrel eating a burger"
   UpperCAmelCase__       :	Any        =       jax.device_count()
   UpperCAmelCase__       :	Optional[Any]        =       num_samples * [prompt]
   UpperCAmelCase__       :	Dict        =       sd_pipe.prepare_inputs(a__     )

   UpperCAmelCase__       :	Dict        =       replicate(a__     )
   UpperCAmelCase__       :	List[str]        =       shard(a__     )

   UpperCAmelCase__       :	Union[str, Any]        =       jax.random.PRNGKey(0     )
   UpperCAmelCase__       :	Optional[Any]        =       jax.random.split(a__       ,jax.device_count()     )

   UpperCAmelCase__       :	List[Any]        =       sd_pipe(a__       ,a__       ,a__       ,num_inference_steps=25       ,jit=a__     )[0]
   assert images.shape == (jax.device_count(), 1, 768, 768, 3)

   UpperCAmelCase__       :	Dict        =       images.reshape((images.shape[0] * images.shape[1],) + images.shape[-3:]     )
   UpperCAmelCase__       :	Optional[int]        =       images[0, 253:256, 253:256, -1]

   UpperCAmelCase__       :	str        =       jnp.asarray(jax.device_get(image_slice.flatten()     )     )
   UpperCAmelCase__       :	List[Any]        =       jnp.array([0.4_2_3_8, 0.4_4_1_4, 0.4_3_9_5, 0.4_4_5_3, 0.4_6_2_9, 0.4_5_9_0, 0.4_5_3_1, 0.4_5_5_0_8, 0.4_5_1_2]     )
   print(f"output_slice: {output_slice}"     )
   assert jnp.abs(output_slice - expected_slice     ).max() < 1e-2




  def  __lowercase							(	self		:      Optional[Any]     ):






   '''simple docstring'''
   UpperCAmelCase__       :	Optional[int]        =       "stabilityai/stable-diffusion-2"
   UpperCAmelCase__       :	Union[str, Any]        =       FlaxDPMSolverMultistepScheduler.from_pretrained(a__       ,subfolder="""scheduler"""     )
   UpperCAmelCase__       :	Tuple        =       FlaxStableDiffusionPipeline.from_pretrained(
       a__       ,scheduler=a__       ,revision="""bf16"""       ,dtype=jnp.bfloataa       ,)
   UpperCAmelCase__       :	Any        =       scheduler_params

   UpperCAmelCase__       :	Any        =       "A painting of a squirrel eating a burger"
   UpperCAmelCase__       :	Any        =       jax.device_count()
   UpperCAmelCase__       :	Tuple        =       num_samples * [prompt]
   UpperCAmelCase__       :	Any        =       sd_pipe.prepare_inputs(a__     )

   UpperCAmelCase__       :	str        =       replicate(a__     )
   UpperCAmelCase__       :	List[Any]        =       shard(a__     )

   UpperCAmelCase__       :	Any        =       jax.random.PRNGKey(0     )
   UpperCAmelCase__       :	List[str]        =       jax.random.split(a__       ,jax.device_count()     )

   UpperCAmelCase__       :	Any        =       sd_pipe(a__       ,a__       ,a__       ,num_inference_steps=25       ,jit=a__     )[0]
   assert images.shape == (jax.device_count(), 1, 768, 768, 3)

   UpperCAmelCase__       :	Optional[Any]        =       images.reshape((images.shape[0] * images.shape[1],) + images.shape[-3:]     )
   UpperCAmelCase__       :	Dict        =       images[0, 253:256, 253:256, -1]

   UpperCAmelCase__       :	Dict        =       jnp.asarray(jax.device_get(image_slice.flatten()     )     )
   UpperCAmelCase__       :	int        =       jnp.array([0.4_3_3_6, 0.4_2_9_6_9, 0.4_4_5_3, 0.4_1_9_9, 0.4_2_9_7, 0.4_5_3_1, 0.4_4_3_4, 0.4_4_3_4, 0.4_2_9_7]     )
   print(f"output_slice: {output_slice}"     )
   assert jnp.abs(output_slice - expected_slice     ).max() < 1e-2







'''simple docstring'''

import tempfile

import numpy as np
import torch
from transformers import AutoTokenizer, TaEncoderModel

from diffusers import DDPMScheduler, UNetaDConditionModel
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.pipelines.deepfloyd_if import IFWatermarker
from diffusers.utils.testing_utils import torch_device

from ..test_pipelines_common import to_np




class    lowerCAmelCase       :


 def    _A	(    self  : Union[str, Any]    ):
  '''simple docstring'''

  torch.manual_seed(0    )
  lowerCAmelCase__     :   List[str]     		=	TaEncoderModel.from_pretrained("hf-internal-testing/tiny-random-t5"    )

  torch.manual_seed(0    )
  lowerCAmelCase__     :   Optional[int]     		=	AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-t5"    )

  torch.manual_seed(0    )
  lowerCAmelCase__     :   Any     		=	UNetaDConditionModel(
      sample_size=32			,							layers_per_block=1			,							block_out_channels=[32, 64]			,							down_block_types=[
          "ResnetDownsampleBlock2D",
          "SimpleCrossAttnDownBlock2D",
      ]			,							mid_block_type="UNetMidBlock2DSimpleCrossAttn"			,							up_block_types=["SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"]			,							in_channels=3			,							out_channels=6			,							cross_attention_dim=32			,							encoder_hid_dim=32			,							attention_head_dim=8			,							addition_embed_type="text"			,							addition_embed_type_num_heads=2			,							cross_attention_norm="group_norm"			,							resnet_time_scale_shift="scale_shift"			,							act_fn="gelu"			,							)
  unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

  torch.manual_seed(0    )
  lowerCAmelCase__     :   Union[str, Any]     		=	DDPMScheduler(
      num_train_timesteps=1000			,							beta_schedule="squaredcos_cap_v2"			,							beta_start=0.0001			,							beta_end=0.02			,							thresholding=a__			,							dynamic_thresholding_ratio=0.95			,							sample_max_value=1.0			,							prediction_type="epsilon"			,							variance_type="learned_range"			,							)

  torch.manual_seed(0    )
  lowerCAmelCase__     :   Tuple     		=	IFWatermarker()

  return {
      "text_encoder": text_encoder,
      "tokenizer": tokenizer,
      "unet": unet,
      "scheduler": scheduler,
      "watermarker": watermarker,
      "safety_checker": None,
      "feature_extractor": None,
  }


 def    _A	(    self  : Dict    ):
  '''simple docstring'''

  torch.manual_seed(0    )
  lowerCAmelCase__     :   Any     		=	TaEncoderModel.from_pretrained("hf-internal-testing/tiny-random-t5"    )

  torch.manual_seed(0    )
  lowerCAmelCase__     :   List[Any]     		=	AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-t5"    )

  torch.manual_seed(0    )
  lowerCAmelCase__     :   int     		=	UNetaDConditionModel(
      sample_size=32			,							layers_per_block=[1, 2]			,							block_out_channels=[32, 64]			,							down_block_types=[
          "ResnetDownsampleBlock2D",
          "SimpleCrossAttnDownBlock2D",
      ]			,							mid_block_type="UNetMidBlock2DSimpleCrossAttn"			,							up_block_types=["SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"]			,							in_channels=6			,							out_channels=6			,							cross_attention_dim=32			,							encoder_hid_dim=32			,							attention_head_dim=8			,							addition_embed_type="text"			,							addition_embed_type_num_heads=2			,							cross_attention_norm="group_norm"			,							resnet_time_scale_shift="scale_shift"			,							act_fn="gelu"			,							class_embed_type="timestep"			,							mid_block_scale_factor=1.414			,							time_embedding_act_fn="gelu"			,							time_embedding_dim=32			,							)
  unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

  torch.manual_seed(0    )
  lowerCAmelCase__     :   Tuple     		=	DDPMScheduler(
      num_train_timesteps=1000			,							beta_schedule="squaredcos_cap_v2"			,							beta_start=0.0001			,							beta_end=0.02			,							thresholding=a__			,							dynamic_thresholding_ratio=0.95			,							sample_max_value=1.0			,							prediction_type="epsilon"			,							variance_type="learned_range"			,							)

  torch.manual_seed(0    )
  lowerCAmelCase__     :   Tuple     		=	DDPMScheduler(
      num_train_timesteps=1000			,							beta_schedule="squaredcos_cap_v2"			,							beta_start=0.0001			,							beta_end=0.02			,							)

  torch.manual_seed(0    )
  lowerCAmelCase__     :   Dict     		=	IFWatermarker()

  return {
      "text_encoder": text_encoder,
      "tokenizer": tokenizer,
      "unet": unet,
      "scheduler": scheduler,
      "image_noising_scheduler": image_noising_scheduler,
      "watermarker": watermarker,
      "safety_checker": None,
      "feature_extractor": None,
  }


 def    _A	(    self  : Optional[Any]    ):
  '''simple docstring'''

  lowerCAmelCase__     :   Union[str, Any]     		=	self.get_dummy_components()
  lowerCAmelCase__     :   Dict     		=	self.pipeline_class(**a__    )
  pipe.to(a__    )
  pipe.set_progress_bar_config(disable=a__    )

  lowerCAmelCase__     :   Tuple     		=	self.get_dummy_inputs(a__    )

  lowerCAmelCase__     :   str     		=	inputs["prompt"]
  lowerCAmelCase__     :   Optional[Any]     		=	inputs["generator"]
  lowerCAmelCase__     :   str     		=	inputs["num_inference_steps"]
  lowerCAmelCase__     :   int     		=	inputs["output_type"]

  if "image" in inputs:
   lowerCAmelCase__     :   Optional[int]     		=	inputs["image"]
  else:
   lowerCAmelCase__     :   str     		=	None

  if "mask_image" in inputs:
   lowerCAmelCase__     :   List[str]     		=	inputs["mask_image"]
  else:
   lowerCAmelCase__     :   str     		=	None

  if "original_image" in inputs:
   lowerCAmelCase__     :   Optional[int]     		=	inputs["original_image"]
  else:
   lowerCAmelCase__     :   Union[str, Any]     		=	None

  lowerCAmelCase__		,     lowerCAmelCase__     :   Dict     		=	pipe.encode_prompt(a__    )

  # inputs with prompt converted to embeddings
  lowerCAmelCase__     :   Dict     		=	{
      "prompt_embeds": prompt_embeds,
      "negative_prompt_embeds": negative_prompt_embeds,
      "generator": generator,
      "num_inference_steps": num_inference_steps,
      "output_type": output_type,
  }

  if image is not None:
   lowerCAmelCase__     :   Optional[Any]     		=	image

  if mask_image is not None:
   lowerCAmelCase__     :   Optional[int]     		=	mask_image

  if original_image is not None:
   lowerCAmelCase__     :   Optional[Any]     		=	original_image

  # set all optional components to None
  for optional_component in pipe._optional_components:
   setattr(a__			,							a__			,							a__    )

  lowerCAmelCase__     :   Optional[int]     		=	pipe(**a__    )[0]

  with tempfile.TemporaryDirectory() as tmpdir:
   pipe.save_pretrained(a__    )
   lowerCAmelCase__     :   Tuple     		=	self.pipeline_class.from_pretrained(a__    )
   pipe_loaded.to(a__    )
   pipe_loaded.set_progress_bar_config(disable=a__    )

  pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

  for optional_component in pipe._optional_components:
   self.assertTrue(
       getattr(a__			,							a__    ) is None			,							F'''`{optional_component}` did not stay set to None after loading.'''			,							)

  lowerCAmelCase__     :   Union[str, Any]     		=	self.get_dummy_inputs(a__    )

  lowerCAmelCase__     :   Any     		=	inputs["generator"]
  lowerCAmelCase__     :   Union[str, Any]     		=	inputs["num_inference_steps"]
  lowerCAmelCase__     :   str     		=	inputs["output_type"]

  # inputs with prompt converted to embeddings
  lowerCAmelCase__     :   int     		=	{
      "prompt_embeds": prompt_embeds,
      "negative_prompt_embeds": negative_prompt_embeds,
      "generator": generator,
      "num_inference_steps": num_inference_steps,
      "output_type": output_type,
  }

  if image is not None:
   lowerCAmelCase__     :   Tuple     		=	image

  if mask_image is not None:
   lowerCAmelCase__     :   int     		=	mask_image

  if original_image is not None:
   lowerCAmelCase__     :   List[Any]     		=	original_image

  lowerCAmelCase__     :   Dict     		=	pipe_loaded(**a__    )[0]

  lowerCAmelCase__     :   int     		=	np.abs(to_np(a__    ) - to_np(a__    )    ).max()
  self.assertLess(a__			,							1e-4    )


 def    _A	(    self  : Any    ):
  '''simple docstring'''

  lowerCAmelCase__     :   str     		=	self.get_dummy_components()
  lowerCAmelCase__     :   int     		=	self.pipeline_class(**a__    )
  pipe.to(a__    )
  pipe.set_progress_bar_config(disable=a__    )

  lowerCAmelCase__     :   Any     		=	self.get_dummy_inputs(a__    )
  lowerCAmelCase__     :   Union[str, Any]     		=	pipe(**a__    )[0]

  with tempfile.TemporaryDirectory() as tmpdir:
   pipe.save_pretrained(a__    )
   lowerCAmelCase__     :   str     		=	self.pipeline_class.from_pretrained(a__    )
   pipe_loaded.to(a__    )
   pipe_loaded.set_progress_bar_config(disable=a__    )

  pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

  lowerCAmelCase__     :   str     		=	self.get_dummy_inputs(a__    )
  lowerCAmelCase__     :   Union[str, Any]     		=	pipe_loaded(**a__    )[0]

  lowerCAmelCase__     :   Union[str, Any]     		=	np.abs(to_np(a__    ) - to_np(a__    )    ).max()
  self.assertLess(a__			,							1e-4    )





