


import tempfile
import unittest

import numpy as np
import torch

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







class  lowerCamelCase__  (			unittest.TestCase):





     def 	__A   (self						,	UpperCAmelCase = False			)	->  int:
      _lowercase				 =1_0

      if only_cross_attention:
       _lowercase				 =1_2
      else:
       # when only cross attention is not set, the cross attention dim must be the same as the query dim
       _lowercase				 =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						,	UpperCAmelCase						,	UpperCAmelCase			)	->  Optional[int]:
      _lowercase				 =2

      _lowercase				 =torch.rand(UpperCAmelCase						,	UpperCAmelCase						,	3						,	2			)
      _lowercase				 =torch.rand(UpperCAmelCase						,	4						,	UpperCAmelCase			)
      _lowercase				 =None

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





     def 	__A   (self			)	->  List[str]:
      # self and cross attention

      torch.manual_seed(0			)

      _lowercase				 =self.get_constructor_arguments(only_cross_attention=UpperCAmelCase			)
      _lowercase				 =Attention(**UpperCAmelCase			)

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

      _lowercase				 =self.get_forward_arguments(
          query_dim=constructor_args['''query_dim''']						,	added_kv_proj_dim=constructor_args['''added_kv_proj_dim''']			)

      _lowercase				 =attn(**UpperCAmelCase			)

      # only self attention

      torch.manual_seed(0			)

      _lowercase				 =self.get_constructor_arguments(only_cross_attention=UpperCAmelCase			)
      _lowercase				 =Attention(**UpperCAmelCase			)

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

      _lowercase				 =self.get_forward_arguments(
          query_dim=constructor_args['''query_dim''']						,	added_kv_proj_dim=constructor_args['''added_kv_proj_dim''']			)

      _lowercase				 =attn(**UpperCAmelCase			)

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







class  lowerCamelCase__  (			unittest.TestCase):





     def 	__A   (self			)	->  List[str]:
      _lowercase				 =DiffusionPipeline.from_pretrained('''hf-internal-testing/tiny-stable-diffusion-pipe'''						,	safety_checker=UpperCAmelCase			)

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

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

      _lowercase				 =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(UpperCAmelCase			)

       # can also load the converted weights
       _lowercase				 =DiffusionPipeline.from_pretrained(UpperCAmelCase						,	device_map='''sequential'''						,	safety_checker=UpperCAmelCase			)

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

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





from typing import List, Optional, Tuple, Union

import torch

from ...models import UNetaDModel
from ...schedulers import PNDMScheduler
from ...utils import randn_tensor
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput







class  lowerCamelCase__  (			lowerCAmelCase):
     SCREAMING_SNAKE_CASE__      =     42
     SCREAMING_SNAKE_CASE__      =     42





     def __init__(self						,	UpperCAmelCase						,	UpperCAmelCase			)	->  Tuple:
      super().__init__()

      _lowercase				 =PNDMScheduler.from_config(scheduler.config			)

      self.register_modules(unet=UpperCAmelCase						,	scheduler=UpperCAmelCase			)





     @torch.no_grad()
     def __call__(self						,	UpperCAmelCase = 1						,	UpperCAmelCase = 5_0						,	UpperCAmelCase = None						,	UpperCAmelCase = "pil"						,	UpperCAmelCase = True						,	**UpperCAmelCase						,	)	->  Union[ImagePipelineOutput, Tuple]:
      _lowercase				 =randn_tensor(
          (batch_size, self.unet.config.in_channels, self.unet.config.sample_size, self.unet.config.sample_size)						,	generator=UpperCAmelCase						,	device=self.device						,	)

      self.scheduler.set_timesteps(UpperCAmelCase			)
      for t in self.progress_bar(self.scheduler.timesteps			):
       _lowercase				 =self.unet(UpperCAmelCase						,	UpperCAmelCase			).sample

       _lowercase				 =self.scheduler.step(UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase			).prev_sample

      _lowercase				 =(image / 2 + 0.5).clamp(0						,	1			)
      _lowercase				 =image.cpu().permute(0						,	2						,	3						,	1			).numpy()
      if output_type == "pil":
       _lowercase				 =self.numpy_to_pil(UpperCAmelCase			)

      if not return_dict:
       return (image,)

      return ImagePipelineOutput(images=UpperCAmelCase			)

