

"""simple docstring"""




import tempfile

import torch

from diffusers import (
    DEISMultistepScheduler,
    DPMSolverMultistepScheduler,
    DPMSolverSinglestepScheduler,
    UniPCMultistepScheduler,
)

from .test_schedulers import SchedulerCommonTest







class      __lowercase			(    __lowerCamelCase					):
  snake_case_				       =       (DPMSolverSinglestepScheduler,)
  snake_case_				       =       (("""num_inference_steps""", 2_5),)

  def    __lowercase       (  self		:			Union[str, Any]					,**A		:			int    ):






   '''simple docstring'''
   UpperCAmelCase__					:  int					   =					{
       """num_train_timesteps""": 1_000,
       """beta_start""": 0.0_0_0_1,
       """beta_end""": 0.0_2,
       """beta_schedule""": """linear""",
       """solver_order""": 2,
       """prediction_type""": """epsilon""",
       """thresholding""": False,
       """sample_max_value""": 1.0,
       """algorithm_type""": """dpmsolver++""",
       """solver_type""": """midpoint""",
       """lambda_min_clipped""": -float("""inf"""    ),
       """variance_type""": None,
   }

   config.update(**A    )
   return config

  def    __lowercase       (  self		:			List[Any]					,A		:			List[str]=0					,**A		:			Optional[int]    ):






   '''simple docstring'''
   UpperCAmelCase__					:  List[Any]					   =					dict(self.forward_default_kwargs    )
   UpperCAmelCase__					:  Union[str, Any]					   =					kwargs.pop("""num_inference_steps"""					,A    )
   UpperCAmelCase__					:  List[Any]					   =					self.dummy_sample
   UpperCAmelCase__					:  Union[str, Any]					   =					0.1 * sample
   UpperCAmelCase__					:  Dict					   =					[residual + 0.2, residual + 0.1_5, residual + 0.1_0]

   for scheduler_class in self.scheduler_classes:
    UpperCAmelCase__					:  Tuple					   =					self.get_scheduler_config(**A    )
    UpperCAmelCase__					:  Optional[Any]					   =					scheduler_class(**A    )
    scheduler.set_timesteps(A    )
    # copy over dummy past residuals
    UpperCAmelCase__					:  List[Any]					   =					dummy_past_residuals[: scheduler.config.solver_order]

    with tempfile.TemporaryDirectory() as tmpdirname:
     scheduler.save_config(A    )
     UpperCAmelCase__					:  Tuple					   =					scheduler_class.from_pretrained(A    )
     new_scheduler.set_timesteps(A    )
     # copy over dummy past residuals
     UpperCAmelCase__					:  int					   =					dummy_past_residuals[: new_scheduler.config.solver_order]

    UpperCAmelCase__   ,      UpperCAmelCase__					:  List[str]					   =					sample, sample
    for t in range(A					,time_step + scheduler.config.solver_order + 1    ):
     UpperCAmelCase__					:  Optional[int]					   =					scheduler.step(A					,A					,A					,**A    ).prev_sample
     UpperCAmelCase__					:  List[Any]					   =					new_scheduler.step(A					,A					,A					,**A    ).prev_sample

     assert torch.sum(torch.abs(output - new_output    )    ) < 1e-5, "Scheduler outputs are not identical"

  def    __lowercase       (  self		:			List[str]    ):






   '''simple docstring'''
   pass

  def    __lowercase       (  self		:			int					,A		:			Any=0					,**A		:			str    ):






   '''simple docstring'''
   UpperCAmelCase__					:  str					   =					dict(self.forward_default_kwargs    )
   UpperCAmelCase__					:  Optional[int]					   =					kwargs.pop("""num_inference_steps"""					,A    )
   UpperCAmelCase__					:  Dict					   =					self.dummy_sample
   UpperCAmelCase__					:  Dict					   =					0.1 * sample
   UpperCAmelCase__					:  List[str]					   =					[residual + 0.2, residual + 0.1_5, residual + 0.1_0]

   for scheduler_class in self.scheduler_classes:
    UpperCAmelCase__					:  Tuple					   =					self.get_scheduler_config()
    UpperCAmelCase__					:  Union[str, Any]					   =					scheduler_class(**A    )
    scheduler.set_timesteps(A    )

    # copy over dummy past residuals (must be after setting timesteps)
    UpperCAmelCase__					:  Optional[int]					   =					dummy_past_residuals[: scheduler.config.solver_order]

    with tempfile.TemporaryDirectory() as tmpdirname:
     scheduler.save_config(A    )
     UpperCAmelCase__					:  Dict					   =					scheduler_class.from_pretrained(A    )
     # copy over dummy past residuals
     new_scheduler.set_timesteps(A    )

     # copy over dummy past residual (must be after setting timesteps)
     UpperCAmelCase__					:  Dict					   =					dummy_past_residuals[: new_scheduler.config.solver_order]

    UpperCAmelCase__					:  Any					   =					scheduler.step(A					,A					,A					,**A    ).prev_sample
    UpperCAmelCase__					:  Union[str, Any]					   =					new_scheduler.step(A					,A					,A					,**A    ).prev_sample

    assert torch.sum(torch.abs(output - new_output    )    ) < 1e-5, "Scheduler outputs are not identical"

  def    __lowercase       (  self		:			Union[str, Any]					,A		:			Optional[Any]=None					,**A		:			List[str]    ):






   '''simple docstring'''
   if scheduler is None:
    UpperCAmelCase__					:  int					   =					self.scheduler_classes[0]
    UpperCAmelCase__					:  Optional[int]					   =					self.get_scheduler_config(**A    )
    UpperCAmelCase__					:  Any					   =					scheduler_class(**A    )

   UpperCAmelCase__					:  int					   =					self.scheduler_classes[0]
   UpperCAmelCase__					:  str					   =					self.get_scheduler_config(**A    )
   UpperCAmelCase__					:  Tuple					   =					scheduler_class(**A    )

   UpperCAmelCase__					:  int					   =					10
   UpperCAmelCase__					:  Any					   =					self.dummy_model()
   UpperCAmelCase__					:  Any					   =					self.dummy_sample_deter
   scheduler.set_timesteps(A    )

   for i, t in enumerate(scheduler.timesteps    ):
    UpperCAmelCase__					:  Union[str, Any]					   =					model(A					,A    )
    UpperCAmelCase__					:  List[str]					   =					scheduler.step(A					,A					,A    ).prev_sample

   return sample

  def    __lowercase       (  self		:			Any    ):






   '''simple docstring'''
   UpperCAmelCase__					:  Union[str, Any]					   =					DPMSolverSinglestepScheduler(**self.get_scheduler_config()    )
   UpperCAmelCase__					:  Optional[int]					   =					50
   UpperCAmelCase__					:  List[str]					   =					self.dummy_model()
   UpperCAmelCase__					:  str					   =					self.dummy_sample_deter
   scheduler.set_timesteps(A    )

   # make sure that the first t is uneven
   for i, t in enumerate(scheduler.timesteps[3:]    ):
    UpperCAmelCase__					:  int					   =					model(A					,A    )
    UpperCAmelCase__					:  int					   =					scheduler.step(A					,A					,A    ).prev_sample

   UpperCAmelCase__					:  Union[str, Any]					   =					torch.mean(torch.abs(A    )    )

   assert abs(result_mean.item() - 0.2_5_7_4    ) < 1e-3

  def    __lowercase       (  self		:			Union[str, Any]    ):






   '''simple docstring'''
   for timesteps in [25, 50, 100, 999, 1_000]:
    self.check_over_configs(num_train_timesteps=A    )

  def    __lowercase       (  self		:			Union[str, Any]    ):






   '''simple docstring'''
   # make sure that iterating over schedulers with same config names gives same results
   # for defaults
   UpperCAmelCase__					:  int					   =					DPMSolverSinglestepScheduler(**self.get_scheduler_config()    )
   UpperCAmelCase__					:  List[str]					   =					self.full_loop(scheduler=A    )
   UpperCAmelCase__					:  Optional[Any]					   =					torch.mean(torch.abs(A    )    )

   assert abs(result_mean.item() - 0.2_7_9_1    ) < 1e-3

   UpperCAmelCase__					:  Dict					   =					DEISMultistepScheduler.from_config(scheduler.config    )
   UpperCAmelCase__					:  str					   =					DPMSolverMultistepScheduler.from_config(scheduler.config    )
   UpperCAmelCase__					:  Tuple					   =					UniPCMultistepScheduler.from_config(scheduler.config    )
   UpperCAmelCase__					:  Union[str, Any]					   =					DPMSolverSinglestepScheduler.from_config(scheduler.config    )

   UpperCAmelCase__					:  Union[str, Any]					   =					self.full_loop(scheduler=A    )
   UpperCAmelCase__					:  Tuple					   =					torch.mean(torch.abs(A    )    )

   assert abs(result_mean.item() - 0.2_7_9_1    ) < 1e-3

  def    __lowercase       (  self		:			int    ):






   '''simple docstring'''
   self.check_over_configs(thresholding=A    )
   for order in [1, 2, 3]:
    for solver_type in ["midpoint", "heun"]:
     for threshold in [0.5, 1.0, 2.0]:
      for prediction_type in ["epsilon", "sample"]:
       self.check_over_configs(
           thresholding=A					,prediction_type=A					,sample_max_value=A					,algorithm_type="""dpmsolver++"""					,solver_order=A					,solver_type=A					,)

  def    __lowercase       (  self		:			Dict    ):






   '''simple docstring'''
   for prediction_type in ["epsilon", "v_prediction"]:
    self.check_over_configs(prediction_type=A    )

  def    __lowercase       (  self		:			Any    ):






   '''simple docstring'''
   for algorithm_type in ["dpmsolver", "dpmsolver++"]:
    for solver_type in ["midpoint", "heun"]:
     for order in [1, 2, 3]:
      for prediction_type in ["epsilon", "sample"]:
       self.check_over_configs(
           solver_order=A					,solver_type=A					,prediction_type=A					,algorithm_type=A					,)
       UpperCAmelCase__					:  str					   =					self.full_loop(
           solver_order=A					,solver_type=A					,prediction_type=A					,algorithm_type=A					,)
       assert not torch.isnan(A    ).any(), "Samples have nan numbers"

  def    __lowercase       (  self		:			Any    ):






   '''simple docstring'''
   self.check_over_configs(lower_order_final=A    )
   self.check_over_configs(lower_order_final=A    )

  def    __lowercase       (  self		:			Optional[int]    ):






   '''simple docstring'''
   self.check_over_configs(lambda_min_clipped=-float("""inf"""    )    )
   self.check_over_configs(lambda_min_clipped=-5.1    )

  def    __lowercase       (  self		:			List[Any]    ):






   '''simple docstring'''
   self.check_over_configs(variance_type=A    )
   self.check_over_configs(variance_type="""learned_range"""    )

  def    __lowercase       (  self		:			Optional[Any]    ):






   '''simple docstring'''
   for num_inference_steps in [1, 2, 3, 5, 10, 50, 100, 999, 1_000]:
    self.check_over_forward(num_inference_steps=A					,time_step=0    )

  def    __lowercase       (  self		:			List[str]    ):






   '''simple docstring'''
   UpperCAmelCase__					:  List[Any]					   =					self.full_loop()
   UpperCAmelCase__					:  Optional[Any]					   =					torch.mean(torch.abs(A    )    )

   assert abs(result_mean.item() - 0.2_7_9_1    ) < 1e-3

  def    __lowercase       (  self		:			str    ):






   '''simple docstring'''
   UpperCAmelCase__					:  Optional[int]					   =					self.full_loop(use_karras_sigmas=A    )
   UpperCAmelCase__					:  Optional[int]					   =					torch.mean(torch.abs(A    )    )

   assert abs(result_mean.item() - 0.2_2_4_8    ) < 1e-3

  def    __lowercase       (  self		:			str    ):






   '''simple docstring'''
   UpperCAmelCase__					:  int					   =					self.full_loop(prediction_type="""v_prediction"""    )
   UpperCAmelCase__					:  Optional[Any]					   =					torch.mean(torch.abs(A    )    )

   assert abs(result_mean.item() - 0.1_4_5_3    ) < 1e-3

  def    __lowercase       (  self		:			List[Any]    ):






   '''simple docstring'''
   UpperCAmelCase__					:  Optional[int]					   =					self.full_loop(prediction_type="""v_prediction"""					,use_karras_sigmas=A    )
   UpperCAmelCase__					:  str					   =					torch.mean(torch.abs(A    )    )

   assert abs(result_mean.item() - 0.0_6_4_9    ) < 1e-3




  def    __lowercase       (  self		:			str    ):






   '''simple docstring'''
   UpperCAmelCase__					:  Tuple					   =					self.scheduler_classes[0]
   UpperCAmelCase__					:  int					   =					self.get_scheduler_config(thresholding=A					,dynamic_thresholding_ratio=0    )
   UpperCAmelCase__					:  Tuple					   =					scheduler_class(**A    )

   UpperCAmelCase__					:  Tuple					   =					10
   UpperCAmelCase__					:  Union[str, Any]					   =					self.dummy_model()
   UpperCAmelCase__					:  Tuple					   =					self.dummy_sample_deter.half()
   scheduler.set_timesteps(A    )

   for i, t in enumerate(scheduler.timesteps    ):
    UpperCAmelCase__					:  Optional[Any]					   =					model(A					,A    )
    UpperCAmelCase__					:  Any					   =					scheduler.step(A					,A					,A    ).prev_sample

   assert sample.dtype == torch.floataa








"""simple docstring"""




# This model implementation is heavily inspired by https://github.com/haofanwang/ControlNet-for-Diffusers/

import gc
import random
import tempfile
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    ControlNetModel,
    DDIMScheduler,
    StableDiffusionControlNetImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion_controlnet import MultiControlNetModel
from diffusers.utils import floats_tensor, load_image, load_numpy, randn_tensor, slow, torch_device
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
)


enable_full_determinism()







class      __lowercase			(    __lowerCamelCase	,			__lowerCamelCase	,			__lowerCamelCase	,			unittest.TestCase					):
  snake_case_				       =       StableDiffusionControlNetImgaImgPipeline
  snake_case_				       =       TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width"""}
  snake_case_				       =       TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
  snake_case_				       =       IMAGE_TO_IMAGE_IMAGE_PARAMS.union({"""control_image"""}					)
  snake_case_				       =       IMAGE_TO_IMAGE_IMAGE_PARAMS

  def    __lowercase       (  self		:			str    ):






   '''simple docstring'''
   torch.manual_seed(0    )
   UpperCAmelCase__					:  Union[str, 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""")					,cross_attention_dim=32					,)
   torch.manual_seed(0    )
   UpperCAmelCase__					:  Optional[int]					   =					ControlNetModel(
       block_out_channels=(32, 64)					,layers_per_block=2					,in_channels=4					,down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")					,cross_attention_dim=32					,conditioning_embedding_out_channels=(16, 32)					,)
   torch.manual_seed(0    )
   UpperCAmelCase__					:  Tuple					   =					DDIMScheduler(
       beta_start=0.0_0_0_8_5					,beta_end=0.0_1_2					,beta_schedule="""scaled_linear"""					,clip_sample=A					,set_alpha_to_one=A					,)
   torch.manual_seed(0    )
   UpperCAmelCase__					:  Any					   =					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					,)
   torch.manual_seed(0    )
   UpperCAmelCase__					:  int					   =					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=1_000					,)
   UpperCAmelCase__					:  List[str]					   =					CLIPTextModel(A    )
   UpperCAmelCase__					:  Dict					   =					CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""    )

   UpperCAmelCase__					:  Optional[int]					   =					{
       """unet""": unet,
       """controlnet""": controlnet,
       """scheduler""": scheduler,
       """vae""": vae,
       """text_encoder""": text_encoder,
       """tokenizer""": tokenizer,
       """safety_checker""": None,
       """feature_extractor""": None,
   }
   return components

  def    __lowercase       (  self		:			Union[str, Any]					,A		:			Dict					,A		:			Optional[Any]=0    ):






   '''simple docstring'''
   if str(A    ).startswith("""mps"""    ):
    UpperCAmelCase__					:  List[str]					   =					torch.manual_seed(A    )
   else:
    UpperCAmelCase__					:  List[Any]					   =					torch.Generator(device=A    ).manual_seed(A    )

   UpperCAmelCase__					:  Tuple					   =					2
   UpperCAmelCase__					:  Any					   =					randn_tensor(
       (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)					,generator=A					,device=torch.device(A    )					,)
   UpperCAmelCase__					:  str					   =					floats_tensor(control_image.shape					,rng=random.Random(A    )    ).to(A    )
   UpperCAmelCase__					:  Dict					   =					image.cpu().permute(0					,2					,3					,1    )[0]
   UpperCAmelCase__					:  List[str]					   =					Image.fromarray(np.uinta(A    )    ).convert("""RGB"""    ).resize((64, 64)    )
   UpperCAmelCase__					:  Any					   =					{
       """prompt""": """A painting of a squirrel eating a burger""",
       """generator""": generator,
       """num_inference_steps""": 2,
       """guidance_scale""": 6.0,
       """output_type""": """numpy""",
       """image""": image,
       """control_image""": control_image,
   }

   return inputs

  def    __lowercase       (  self		:			Optional[Any]    ):






   '''simple docstring'''
   return self._test_attention_slicing_forward_pass(expected_max_diff=2e-3    )

  @unittest.skipIf(
      torch_device != """cuda""" or not is_xformers_available()					,reason="""XFormers attention is only available with CUDA and `xformers` installed"""					,)
  def    __lowercase       (  self		:			Dict    ):






   '''simple docstring'''
   self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3    )




  def    __lowercase       (  self		:			Any    ):






   '''simple docstring'''
   self._test_inference_batch_single_identical(expected_max_diff=2e-3    )







class      __lowercase			(    __lowerCamelCase	,			__lowerCamelCase	,			unittest.TestCase					):
  snake_case_				       =       StableDiffusionControlNetImgaImgPipeline
  snake_case_				       =       TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width"""}
  snake_case_				       =       TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
  snake_case_				       =       frozenset([]					)  # TO_DO: add image_params once refactored VaeImageProcessor.preprocess

  def    __lowercase       (  self		:			str    ):






   '''simple docstring'''
   torch.manual_seed(0    )
   UpperCAmelCase__					:  Tuple					   =					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""")					,cross_attention_dim=32					,)
   torch.manual_seed(0    )

   def init_weights(A		:			Union[str, Any]    ):
    if isinstance(A					,torch.nn.Convad    ):
     torch.nn.init.normal(m.weight    )
     m.bias.data.fill_(1.0    )

   UpperCAmelCase__					:  int					   =					ControlNetModel(
       block_out_channels=(32, 64)					,layers_per_block=2					,in_channels=4					,down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")					,cross_attention_dim=32					,conditioning_embedding_out_channels=(16, 32)					,)
   controlneta.controlnet_down_blocks.apply(A    )

   torch.manual_seed(0    )
   UpperCAmelCase__					:  Dict					   =					ControlNetModel(
       block_out_channels=(32, 64)					,layers_per_block=2					,in_channels=4					,down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")					,cross_attention_dim=32					,conditioning_embedding_out_channels=(16, 32)					,)
   controlneta.controlnet_down_blocks.apply(A    )

   torch.manual_seed(0    )
   UpperCAmelCase__					:  str					   =					DDIMScheduler(
       beta_start=0.0_0_0_8_5					,beta_end=0.0_1_2					,beta_schedule="""scaled_linear"""					,clip_sample=A					,set_alpha_to_one=A					,)
   torch.manual_seed(0    )
   UpperCAmelCase__					:  Dict					   =					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					,)
   torch.manual_seed(0    )
   UpperCAmelCase__					:  Tuple					   =					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=1_000					,)
   UpperCAmelCase__					:  Optional[Any]					   =					CLIPTextModel(A    )
   UpperCAmelCase__					:  List[str]					   =					CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""    )

   UpperCAmelCase__					:  Dict					   =					MultiControlNetModel([controlneta, controlneta]    )

   UpperCAmelCase__					:  Union[str, Any]					   =					{
       """unet""": unet,
       """controlnet""": controlnet,
       """scheduler""": scheduler,
       """vae""": vae,
       """text_encoder""": text_encoder,
       """tokenizer""": tokenizer,
       """safety_checker""": None,
       """feature_extractor""": None,
   }
   return components

  def    __lowercase       (  self		:			Optional[int]					,A		:			List[Any]					,A		:			List[Any]=0    ):






   '''simple docstring'''
   if str(A    ).startswith("""mps"""    ):
    UpperCAmelCase__					:  List[str]					   =					torch.manual_seed(A    )
   else:
    UpperCAmelCase__					:  int					   =					torch.Generator(device=A    ).manual_seed(A    )

   UpperCAmelCase__					:  Optional[int]					   =					2

   UpperCAmelCase__					:  List[Any]					   =					[
       randn_tensor(
           (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)					,generator=A					,device=torch.device(A    )					,),
       randn_tensor(
           (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)					,generator=A					,device=torch.device(A    )					,),
   ]

   UpperCAmelCase__					:  List[Any]					   =					floats_tensor(control_image[0].shape					,rng=random.Random(A    )    ).to(A    )
   UpperCAmelCase__					:  str					   =					image.cpu().permute(0					,2					,3					,1    )[0]
   UpperCAmelCase__					:  Union[str, Any]					   =					Image.fromarray(np.uinta(A    )    ).convert("""RGB"""    ).resize((64, 64)    )
   UpperCAmelCase__					:  Any					   =					{
       """prompt""": """A painting of a squirrel eating a burger""",
       """generator""": generator,
       """num_inference_steps""": 2,
       """guidance_scale""": 6.0,
       """output_type""": """numpy""",
       """image""": image,
       """control_image""": control_image,
   }

   return inputs

  def    __lowercase       (  self		:			Dict    ):






   '''simple docstring'''
   UpperCAmelCase__					:  Optional[int]					   =					self.get_dummy_components()
   UpperCAmelCase__					:  List[str]					   =					self.pipeline_class(**A    )
   pipe.to(A    )

   UpperCAmelCase__					:  Any					   =					1_0.0
   UpperCAmelCase__					:  Any					   =					4

   UpperCAmelCase__					:  Optional[int]					   =					self.get_dummy_inputs(A    )
   UpperCAmelCase__					:  Optional[Any]					   =					steps
   UpperCAmelCase__					:  List[str]					   =					scale
   UpperCAmelCase__					:  Optional[Any]					   =					pipe(**A    )[0]

   UpperCAmelCase__					:  Any					   =					self.get_dummy_inputs(A    )
   UpperCAmelCase__					:  Optional[int]					   =					steps
   UpperCAmelCase__					:  Tuple					   =					scale
   UpperCAmelCase__					:  List[Any]					   =					pipe(**A					,control_guidance_start=0.1					,control_guidance_end=0.2    )[0]

   UpperCAmelCase__					:  List[str]					   =					self.get_dummy_inputs(A    )
   UpperCAmelCase__					:  Union[str, Any]					   =					steps
   UpperCAmelCase__					:  Any					   =					scale
   UpperCAmelCase__					:  str					   =					pipe(**A					,control_guidance_start=[0.1, 0.3]					,control_guidance_end=[0.2, 0.7]    )[0]

   UpperCAmelCase__					:  Any					   =					self.get_dummy_inputs(A    )
   UpperCAmelCase__					:  List[str]					   =					steps
   UpperCAmelCase__					:  Union[str, Any]					   =					scale
   UpperCAmelCase__					:  Union[str, Any]					   =					pipe(**A					,control_guidance_start=0.4					,control_guidance_end=[0.5, 0.8]    )[0]

   # make sure that all outputs are different
   assert np.sum(np.abs(output_a - output_a    )    ) > 1e-3
   assert np.sum(np.abs(output_a - output_a    )    ) > 1e-3
   assert np.sum(np.abs(output_a - output_a    )    ) > 1e-3

  def    __lowercase       (  self		:			Any    ):






   '''simple docstring'''
   return self._test_attention_slicing_forward_pass(expected_max_diff=2e-3    )

  @unittest.skipIf(
      torch_device != """cuda""" or not is_xformers_available()					,reason="""XFormers attention is only available with CUDA and `xformers` installed"""					,)
  def    __lowercase       (  self		:			Optional[int]    ):






   '''simple docstring'''
   self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3    )

  def    __lowercase       (  self		:			Union[str, Any]    ):






   '''simple docstring'''
   self._test_inference_batch_single_identical(expected_max_diff=2e-3    )




  def    __lowercase       (  self		:			Dict    ):






   '''simple docstring'''
   UpperCAmelCase__					:  Dict					   =					self.get_dummy_components()
   UpperCAmelCase__					:  str					   =					self.pipeline_class(**A    )
   pipe.to(A    )
   pipe.set_progress_bar_config(disable=A    )
   with tempfile.TemporaryDirectory() as tmpdir:
    try:
     # save_pretrained is not implemented for Multi-ControlNet
     pipe.save_pretrained(A    )
    except NotImplementedError:
     pass







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

  def    __lowercase       (  self		:			List[str]    ):






   '''simple docstring'''
   super().tearDown()
   gc.collect()
   torch.cuda.empty_cache()




  def    __lowercase       (  self		:			Union[str, Any]    ):






   '''simple docstring'''
   UpperCAmelCase__					:  str					   =					ControlNetModel.from_pretrained("""lllyasviel/sd-controlnet-canny"""    )

   UpperCAmelCase__					:  Tuple					   =					StableDiffusionControlNetImgaImgPipeline.from_pretrained(
       """runwayml/stable-diffusion-v1-5"""					,safety_checker=A					,controlnet=A    )
   pipe.enable_model_cpu_offload()
   pipe.set_progress_bar_config(disable=A    )

   UpperCAmelCase__					:  Optional[int]					   =					torch.Generator(device="""cpu"""    ).manual_seed(0    )
   UpperCAmelCase__					:  Dict					   =					"""evil space-punk bird"""
   UpperCAmelCase__					:  Optional[int]					   =					load_image(
       """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/bird_canny.png"""    ).resize((512, 512)    )
   UpperCAmelCase__					:  Optional[Any]					   =					load_image(
       """https://huggingface.co/lllyasviel/sd-controlnet-canny/resolve/main/images/bird.png"""    ).resize((512, 512)    )

   UpperCAmelCase__					:  List[str]					   =					pipe(
       A					,A					,control_image=A					,generator=A					,output_type="""np"""					,num_inference_steps=50					,strength=0.6					,)

   UpperCAmelCase__					:  Optional[Any]					   =					output.images[0]

   assert image.shape == (512, 512, 3)

   UpperCAmelCase__					:  Any					   =					load_numpy(
       """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/img2img.npy"""    )

   assert np.abs(expected_image - image    ).max() < 9e-2





