'''simple docstring'''

import tempfile
import unittest

import numpy as np
import torch

from diffusers import DiffusionPipeline
from diffusers.models.attention_processor import Attention, AttnAddedKVProcessor







class     _lowercase					(    unittest.TestCase      ):


 '''simple docstring'''






 def a (				self       :     List[str]				,	SCREAMING_SNAKE_CASE__       :     Any = False							)  ->   Any:
  __lowerCAmelCase    							=						10

  if only_cross_attention:
   __lowerCAmelCase    							=						12
  else:
   # when only cross attention is not set, the cross attention dim must be the same as the query dim
   __lowerCAmelCase    							=						query_dim

  return {
      "query_dim": query_dim,
      "cross_attention_dim": cross_attention_dim,
      "heads": 2,
      "dim_head": 4,
      "added_kv_proj_dim": 6,
      "norm_num_groups": 1,
      "only_cross_attention": only_cross_attention,
      "processor": AttnAddedKVProcessor(),
  }
 def a (				self       :     int				,	SCREAMING_SNAKE_CASE__       :     Optional[int]				,	SCREAMING_SNAKE_CASE__       :     Dict							)  ->   Optional[Any]:
  __lowerCAmelCase    							=						2

  __lowerCAmelCase    							=						torch.rand(lowerCAmelCase__				,	lowerCAmelCase__				,	3				,	2							)
  __lowerCAmelCase    							=						torch.rand(lowerCAmelCase__				,	4				,	lowerCAmelCase__							)
  __lowerCAmelCase    							=						None

  return {
      "hidden_states": hidden_states,
      "encoder_hidden_states": encoder_hidden_states,
      "attention_mask": attention_mask,
  }





 def a (				self       :     str							)  ->   List[Any]:
  torch.manual_seed(0							)

  __lowerCAmelCase    							=						self.get_constructor_arguments(only_cross_attention=lowerCAmelCase__							)
  __lowerCAmelCase    							=						Attention(**lowerCAmelCase__							)

  self.assertTrue(attn.to_k is not None							)
  self.assertTrue(attn.to_v is not None							)

  __lowerCAmelCase    							=						self.get_forward_arguments(
      query_dim=constructor_args["""query_dim"""]				,	added_kv_proj_dim=constructor_args["""added_kv_proj_dim"""]							)

  __lowerCAmelCase    							=						attn(**lowerCAmelCase__							)

  # only self attention

  torch.manual_seed(0							)

  __lowerCAmelCase    							=						self.get_constructor_arguments(only_cross_attention=lowerCAmelCase__							)
  __lowerCAmelCase    							=						Attention(**lowerCAmelCase__							)

  self.assertTrue(attn.to_k is None							)
  self.assertTrue(attn.to_v is None							)

  __lowerCAmelCase    							=						self.get_forward_arguments(
      query_dim=constructor_args["""query_dim"""]				,	added_kv_proj_dim=constructor_args["""added_kv_proj_dim"""]							)

  __lowerCAmelCase    							=						attn(**lowerCAmelCase__							)

  self.assertTrue((only_cross_attn_out != self_and_cross_attn_out).all()							)







class     _lowercase					(    unittest.TestCase      ):


 '''simple docstring'''






 def a (				self       :     List[str]							)  ->   Optional[Any]:
  __lowerCAmelCase    							=						DiffusionPipeline.from_pretrained("""hf-internal-testing/tiny-stable-diffusion-pipe"""				,	safety_checker=lowerCAmelCase__							)

  __lowerCAmelCase    							=						pipe(
      """foo"""				,	num_inference_steps=2				,	generator=torch.Generator("""cpu"""							).manual_seed(0							)				,	output_type="""np"""				,	).images

  # the initial conversion succeeds
  __lowerCAmelCase    							=						DiffusionPipeline.from_pretrained(
      """hf-internal-testing/tiny-stable-diffusion-pipe"""				,	device_map="""sequential"""				,	safety_checker=lowerCAmelCase__							)

  __lowerCAmelCase    							=						pipe(
      """foo"""				,	num_inference_steps=2				,	generator=torch.Generator("""cpu"""							).manual_seed(0							)				,	output_type="""np"""				,	).images

  with tempfile.TemporaryDirectory() as tmpdir:
   # save the converted model
   pipe.save_pretrained(lowerCAmelCase__							)

   # can also load the converted weights
   __lowerCAmelCase    							=						DiffusionPipeline.from_pretrained(lowerCAmelCase__				,	device_map="""sequential"""				,	safety_checker=lowerCAmelCase__							)

  __lowerCAmelCase    							=						pipe(
      """foo"""				,	num_inference_steps=2				,	generator=torch.Generator("""cpu"""							).manual_seed(0							)				,	output_type="""np"""				,	).images

  self.assertTrue(np.allclose(lowerCAmelCase__				,	lowerCAmelCase__				,	atol=1e-5							)							)
  self.assertTrue(np.allclose(lowerCAmelCase__				,	lowerCAmelCase__				,	atol=1e-5							)							)





'''simple docstring'''

import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionDualGuidedPipeline
from diffusers.utils.testing_utils import load_image, nightly, require_torch_gpu, torch_device


_A      :				Any 				=    False







@nightly
@require_torch_gpu
class     _lowercase					(    unittest.TestCase      ):


 '''simple docstring'''






 def a (				self       :     Tuple							)  ->   Dict:
  # clean up the VRAM after each test
  super().tearDown()
  gc.collect()
  torch.cuda.empty_cache()
 def a (				self       :     Optional[int]							)  ->   Optional[Any]:
  __lowerCAmelCase    							=						VersatileDiffusionDualGuidedPipeline.from_pretrained("""shi-labs/versatile-diffusion"""							)
  # remove text_unet
  pipe.remove_unused_weights()
  pipe.to(SCREAMING_SNAKE_CASE__							)
  pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__							)

  __lowerCAmelCase    							=						load_image(
      """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg"""							)

  __lowerCAmelCase    							=						torch.manual_seed(0							)
  __lowerCAmelCase    							=						pipe(
      prompt="""first prompt"""				,	image=SCREAMING_SNAKE_CASE__				,	text_to_image_strength=0.7_5				,	generator=SCREAMING_SNAKE_CASE__				,	guidance_scale=7.5				,	num_inference_steps=2				,	output_type="""numpy"""				,	).images

  with tempfile.TemporaryDirectory() as tmpdirname:
   pipe.save_pretrained(SCREAMING_SNAKE_CASE__							)
   __lowerCAmelCase    							=						VersatileDiffusionDualGuidedPipeline.from_pretrained(SCREAMING_SNAKE_CASE__							)

  pipe.to(SCREAMING_SNAKE_CASE__							)
  pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__							)

  __lowerCAmelCase    							=						generator.manual_seed(0							)
  __lowerCAmelCase    							=						pipe(
      prompt="""first prompt"""				,	image=SCREAMING_SNAKE_CASE__				,	text_to_image_strength=0.7_5				,	generator=SCREAMING_SNAKE_CASE__				,	guidance_scale=7.5				,	num_inference_steps=2				,	output_type="""numpy"""				,	).images

  assert np.abs(image - new_image							).sum() < 1e-5, "Models don't have the same forward pass"





 def a (				self       :     str							)  ->   Tuple:
  __lowerCAmelCase    							=						VersatileDiffusionDualGuidedPipeline.from_pretrained("""shi-labs/versatile-diffusion"""							)
  pipe.remove_unused_weights()
  pipe.to(SCREAMING_SNAKE_CASE__							)
  pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__							)

  __lowerCAmelCase    							=						"""cyberpunk 2077"""
  __lowerCAmelCase    							=						load_image(
      """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg"""							)
  __lowerCAmelCase    							=						torch.manual_seed(0							)
  __lowerCAmelCase    							=						pipe(
      prompt=SCREAMING_SNAKE_CASE__				,	image=SCREAMING_SNAKE_CASE__				,	text_to_image_strength=0.7_5				,	generator=SCREAMING_SNAKE_CASE__				,	guidance_scale=7.5				,	num_inference_steps=50				,	output_type="""numpy"""				,	).images

  __lowerCAmelCase    							=						image[0, 2_53:2_56, 2_53:2_56, -1]

  assert image.shape == (1, 5_12, 5_12, 3)
  __lowerCAmelCase    							=						np.array([0.0_7_8_7, 0.0_8_4_9, 0.0_8_2_6, 0.0_8_1_2, 0.0_8_0_7, 0.0_7_9_5, 0.0_8_1_8, 0.0_7_9_8, 0.0_7_7_9]							)

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






